diff --git a/go/cstx_ffi.h b/go/cstx_ffi.h index 4af148a..6c1e802 100644 --- a/go/cstx_ffi.h +++ b/go/cstx_ffi.h @@ -131,11 +131,34 @@ CstxStatusCode cstx_graph_add_nodes(struct CstxHandle *handle, uint64_t *affected, struct CstxBuffer *error); +/** + * Write each node as its current state, replacing the stored record. + * + * The merge path (`cstx_graph_add_nodes`) owns bulk ingest and keeps its JSON + * fast path. A replace batch is a caller restating records it already holds — + * a task's oracles, a document's current revision — so it goes through the + * shared `Value` path rather than earning a second parser. + */ +CstxStatusCode cstx_graph_replace_nodes(struct CstxHandle *handle, + struct CstxSlice data, + uint64_t *affected, + struct CstxBuffer *error); + CstxStatusCode cstx_graph_add_edges(struct CstxHandle *handle, struct CstxSlice data, uint64_t *affected, struct CstxBuffer *error); +CstxStatusCode cstx_graph_delete_nodes(struct CstxHandle *handle, + struct CstxSlice node_ids_json, + uint64_t *output, + struct CstxBuffer *error); + +CstxStatusCode cstx_graph_delete_edges(struct CstxHandle *handle, + struct CstxSlice edge_ids_json, + uint64_t *output, + struct CstxBuffer *error); + CstxStatusCode cstx_graph_ingest(struct CstxHandle *handle, struct CstxSlice source, struct CstxSlice data, @@ -362,20 +385,97 @@ CstxStatusCode cstx_repo_commit(struct CstxHandle *handle, struct CstxBuffer *output, struct CstxBuffer *error); +CstxStatusCode cstx_repo_prepare(struct CstxHandle *handle, + struct CstxSlice message, + struct CstxSlice ref_name, + struct CstxSlice expected_head, + struct CstxSlice metadata_json, + int64_t timestamp, + uint8_t has_timestamp, + struct CstxBuffer *output, + struct CstxBuffer *error); + +CstxStatusCode cstx_repo_accept(struct CstxHandle *handle, + struct CstxSlice commit, + struct CstxBuffer *error); + +CstxStatusCode cstx_repo_discard(struct CstxHandle *handle, struct CstxBuffer *error); + +CstxStatusCode cstx_repo_synchronize(struct CstxHandle *handle, + struct CstxSlice payload_json, + struct CstxBuffer *error); + +CstxStatusCode cstx_repo_contains(struct CstxHandle *handle, + struct CstxSlice object, + uint8_t *output, + struct CstxBuffer *error); + +CstxStatusCode cstx_repo_missing_tree(struct CstxHandle *handle, + struct CstxSlice commit, + struct CstxBuffer *output, + struct CstxBuffer *error); + +CstxStatusCode cstx_repo_object_closure(struct CstxHandle *handle, + struct CstxSlice commit, + struct CstxBuffer *output, + struct CstxBuffer *error); + +CstxStatusCode cstx_repo_missing_prepare(struct CstxHandle *handle, + struct CstxSlice commit, + struct CstxBuffer *output, + struct CstxBuffer *error); + +CstxStatusCode cstx_repo_missing_history(struct CstxHandle *handle, + struct CstxSlice commit, + struct CstxSlice entity_id, + struct CstxBuffer *output, + struct CstxBuffer *error); + +CstxStatusCode cstx_repo_missing_stat(struct CstxHandle *handle, + struct CstxSlice commit, + struct CstxBuffer *output, + struct CstxBuffer *error); + +CstxStatusCode cstx_repo_missing_commits(struct CstxHandle *handle, + struct CstxSlice commit, + size_t limit, + struct CstxBuffer *output, + struct CstxBuffer *error); + +CstxStatusCode cstx_repo_missing_diff(struct CstxHandle *handle, + struct CstxSlice base, + struct CstxSlice head, + struct CstxSlice detail, + struct CstxBuffer *output, + struct CstxBuffer *error); + +CstxStatusCode cstx_repo_missing_delta(struct CstxHandle *handle, + struct CstxSlice commit, + int64_t start_timestamp, + uint8_t has_start, + int64_t end_timestamp, + uint8_t has_end, + struct CstxBuffer *output, + struct CstxBuffer *error); + +CstxStatusCode cstx_repo_missing_merge(struct CstxHandle *handle, + struct CstxSlice source, + struct CstxSlice target, + struct CstxBuffer *output, + struct CstxBuffer *error); + +CstxStatusCode cstx_repo_release_transient_objects(struct CstxHandle *handle, + struct CstxBuffer *error); + CstxStatusCode cstx_repo_diff(struct CstxHandle *handle, struct CstxSlice base_ref, struct CstxSlice head_ref, size_t limit, uint8_t has_limit, + struct CstxSlice detail, struct CstxBuffer *output, struct CstxBuffer *error); -CstxStatusCode cstx_repo_diff_stat(struct CstxHandle *handle, - struct CstxSlice base_ref, - struct CstxSlice head_ref, - struct CstxBuffer *output, - struct CstxBuffer *error); - CstxStatusCode cstx_repo_head(struct CstxHandle *handle, struct CstxSlice ref_name, struct CstxBuffer *output, diff --git a/go/cstx_native_test.go b/go/cstx_native_test.go index 8bd2fe1..7844080 100644 --- a/go/cstx_native_test.go +++ b/go/cstx_native_test.go @@ -65,6 +65,151 @@ func addDomain(t *testing.T, rt *CSTX, value string) uint64 { return affected } +func TestRepositoryExternalPersistenceRoundTrip(t *testing.T) { + writer := openRuntime(t) + addDomain(t, writer, "persisted.example") + + prepared, err := writer.Repo.Prepare( + testContext, + "external persistence", + "main", + nil, + map[string]any{"source": "go-test"}, + nil, + ) + if err != nil { + t.Fatalf("prepare: %v", err) + } + if prepared.Commit.ID == "" || prepared.IndexRoot == "" || len(prepared.Objects) == 0 { + t.Fatalf("incomplete prepared payload: %+v", prepared) + } + + objects := make(map[string]RepositoryObject, len(prepared.Objects)) + var commitObject, indexObject RepositoryObject + for _, object := range prepared.Objects { + stored := RepositoryObject{ID: object.ID, Envelope: append([]byte(nil), object.Envelope...)} + objects[object.ID] = stored + if object.Kind == "commit" && object.ID == prepared.Commit.ID { + commitObject = stored + } + if object.ID == prepared.IndexRoot { + indexObject = stored + } + } + if commitObject.ID == "" || indexObject.ID == "" { + t.Fatal("prepared payload does not contain commit and index-root envelopes") + } + if err := writer.Repo.Accept(testContext, prepared.Commit.ID); err != nil { + t.Fatalf("accept: %v", err) + } + + reader := openRuntime(t) + head := prepared.Commit.ID + if err := reader.Repo.Synchronize(testContext, RepositorySync{ + Objects: []RepositoryObject{commitObject, indexObject}, + }); err != nil { + t.Fatalf("synchronize commit objects: %v", err) + } + if err := reader.Repo.Synchronize(testContext, RepositorySync{ + Refs: []RepositoryRef{{Name: "main", Commit: &head}}, + Indexes: []RepositoryIndex{{Commit: head, IndexRoot: prepared.IndexRoot}}, + }); err != nil { + t.Fatalf("synchronize commit frontier: %v", err) + } + + for { + missing, err := reader.Repo.MissingTree(testContext, head) + if err != nil { + t.Fatalf("plan missing tree: %v", err) + } + if len(missing) == 0 { + break + } + batch := make([]RepositoryObject, 0, len(missing)) + for _, id := range missing { + object, ok := objects[id] + if !ok { + t.Fatalf("planner requested unknown object %s", id) + } + batch = append(batch, object) + } + if err := reader.Repo.Synchronize(testContext, RepositorySync{Objects: batch}); err != nil { + t.Fatalf("hydrate tree: %v", err) + } + } + if _, err := reader.Repo.Checkout(testContext, "main", true); err != nil { + t.Fatalf("checkout hydrated main: %v", err) + } + node, err := reader.Graph.Node(testContext, "domain:persisted.example") + if err != nil || node.Value != "persisted.example" { + t.Fatalf("restored node: %+v err=%v", node, err) + } + if err := reader.Repo.ReleaseTransientObjects(testContext); err != nil { + t.Fatalf("release transient objects: %v", err) + } +} + +func TestGraphDeleteNodesCascadesAndCommits(t *testing.T) { + rt := openRuntime(t) + addDomain(t, rt, "delete-a.example") + addDomain(t, rt, "delete-b.example") + addDomain(t, rt, "keep.example") + edges := []Edge{ + relatedEdge("domain:delete-a.example", "domain:delete-b.example"), + relatedEdge("domain:delete-b.example", "domain:keep.example"), + } + if _, err := rt.Graph.AddEdges(testContext, edges); err != nil { + t.Fatalf("add edges: %v", err) + } + base, err := rt.Repo.Commit(testContext, "base", "main", nil, nil) + if err != nil { + t.Fatalf("commit base: %v", err) + } + cursor, err := rt.Graph.Nodes(testContext, NodeFilter{}, CollectionOptions{}) + if err != nil { + t.Fatalf("open cursor: %v", err) + } + defer cursor.Close() + + affected, err := rt.Graph.DeleteNodes(testContext, []string{"domain:delete-b.example"}) + if err != nil || affected != 3 { + t.Fatalf("delete node: affected=%d err=%v", affected, err) + } + if _, err := cursor.Page(testContext, 10, 1); !IsCode(err, CodeCursorInvalidated) { + t.Fatalf("expected cursor invalidation, got %v", err) + } + if count, _ := rt.Graph.NodeCount(testContext); count != 2 { + t.Fatalf("node count after delete=%d", count) + } + if count, _ := rt.Graph.EdgeCount(testContext); count != 0 { + t.Fatalf("edge count after cascade=%d", count) + } + change, err := rt.LastChange(testContext) + if err != nil || !reflect.DeepEqual(change.RemovedNodeIDs, []string{"domain:delete-b.example"}) || len(change.RemovedEdgeIDs) != 2 { + t.Fatalf("delete change=%+v err=%v", change, err) + } + head, err := rt.Repo.Commit(testContext, "delete", "main", &base.ID, nil) + if err != nil { + t.Fatalf("commit delete: %v", err) + } + diff, err := rt.Repo.Diff(testContext, base.ID, head.ID, DiffOptions{}) + if err != nil || !reflect.DeepEqual(diff.Removed["domain"], []string{"domain:delete-b.example"}) || len(diff.Removed["edge:related"]) != 2 { + t.Fatalf("delete diff=%+v err=%v", diff, err) + } +} + +func TestGraphDeleteIsAtomicOnMissingID(t *testing.T) { + rt := openRuntime(t) + addDomain(t, rt, "present.example") + affected, err := rt.Graph.DeleteNodes(testContext, []string{"domain:present.example", "domain:missing.example"}) + if err == nil || affected != 0 { + t.Fatalf("expected atomic validation failure: affected=%d err=%v", affected, err) + } + if count, _ := rt.Graph.NodeCount(testContext); count != 1 { + t.Fatalf("failed delete changed graph: count=%d", count) + } +} + func TestSchemas(t *testing.T) { rt := openRuntime(t) valueField := "domain" @@ -500,13 +645,13 @@ func TestRepositoryRoundTrip(t *testing.T) { if err != nil { t.Fatalf("second commit: %v", err) } - diff, err := rt.Repo.Diff(testContext, commit.ID, second.ID, nil) - if err != nil || len(diff.Added["domain"]) != 1 { + diff, err := rt.Repo.Diff(testContext, commit.ID, second.ID, DiffOptions{}) + if err != nil || len(diff.Added["domain"]) != 1 || diff.Stats.AddedNodes != 1 { t.Fatalf("diff: %+v %v", diff, err) } - diffStat, err := rt.Repo.DiffStat(testContext, commit.ID, second.ID) - if err != nil || diffStat.AddedNodes != 1 { - t.Fatalf("diff stat: %+v %v", diffStat, err) + counted, err := rt.Repo.Diff(testContext, commit.ID, second.ID, DiffOptions{Detail: DiffCounts}) + if err != nil || counted.Stats.AddedNodes != 1 || len(counted.Added) != 0 { + t.Fatalf("counted diff: %+v %v", counted, err) } log, err := rt.Repo.Log(testContext, "main", 10) if err != nil || len(log) != 2 { diff --git a/go/engine.go b/go/engine.go index a8e1f61..0e13433 100644 --- a/go/engine.go +++ b/go/engine.go @@ -25,7 +25,10 @@ type engine interface { schemaAnchorConcepts(context.Context) ([]AnchorConcept, error) graphAddNodes(context.Context, []Node) (uint64, error) + graphReplaceNodes(context.Context, []Node) (uint64, error) graphAddEdges(context.Context, []Edge) (uint64, error) + graphDeleteNodes(context.Context, []string) (uint64, error) + graphDeleteEdges(context.Context, []string) (uint64, error) graphIngest(context.Context, string, []byte) (uint64, error) graphNode(context.Context, string) (Node, error) graphContains(context.Context, string) (bool, error) @@ -43,8 +46,22 @@ type engine interface { repoHead(context.Context, string) (*string, error) repoCheckout(context.Context, string, bool) (Commit, error) repoCommit(context.Context, string, string, *string, any) (Commit, error) - repoDiff(context.Context, string, string, *int) (GraphDiff, error) - repoDiffStat(context.Context, string, string) (Delta, error) + repoPrepare(context.Context, string, string, *string, any, *int64) (PreparedCommit, error) + repoAccept(context.Context, string) error + repoDiscard(context.Context) error + repoSynchronize(context.Context, RepositorySync) error + repoContains(context.Context, string) (bool, error) + repoMissingTree(context.Context, string) ([]string, error) + repoObjectClosure(context.Context, string) ([]string, error) + repoMissingPrepare(context.Context, string) ([]string, error) + repoMissingHistory(context.Context, string, string) ([]string, error) + repoMissingStat(context.Context, string) ([]string, error) + repoMissingCommits(context.Context, string, int) ([]string, error) + repoMissingDiff(context.Context, string, string, DiffDetail) ([]string, error) + repoMissingDelta(context.Context, string, *int64, *int64) ([]string, error) + repoMissingMerge(context.Context, string, string) ([]string, error) + repoReleaseTransientObjects(context.Context) error + repoDiff(context.Context, string, string, DiffOptions) (GraphDiff, error) repoLog(context.Context, string, int) ([]map[string]any, error) repoHistory(context.Context, string, string, *int) ([]map[string]any, error) repoBranch(context.Context, string, string) (string, error) diff --git a/go/engine_native.go b/go/engine_native.go index 575b6af..c435763 100644 --- a/go/engine_native.go +++ b/go/engine_native.go @@ -13,6 +13,7 @@ import "C" import ( "context" + "encoding/hex" "encoding/json" "runtime" "unsafe" @@ -74,6 +75,30 @@ func (e *nativeEngine) graphSubgraph(_ context.Context, seedIDs []string, depth return derived, nil } +func (e *nativeEngine) graphDeleteNodes(_ context.Context, nodeIDs []string) (uint64, error) { + payload, err := marshalInput("graph.delete_nodes", nodeIDs) + if err != nil { + return 0, err + } + return countResult("graph.delete_nodes", func(out *C.uint64_t, errBuf *C.CstxBuffer) C.CstxStatusCode { + rc := C.cstx_graph_delete_nodes(e.handle, byteSlice(payload), out, errBuf) + runtime.KeepAlive(payload) + return rc + }) +} + +func (e *nativeEngine) graphDeleteEdges(_ context.Context, edgeIDs []string) (uint64, error) { + payload, err := marshalInput("graph.delete_edges", edgeIDs) + if err != nil { + return 0, err + } + return countResult("graph.delete_edges", func(out *C.uint64_t, errBuf *C.CstxBuffer) C.CstxStatusCode { + rc := C.cstx_graph_delete_edges(e.handle, byteSlice(payload), out, errBuf) + runtime.KeepAlive(payload) + return rc + }) +} + // --- C transport helpers ------------------------------------------------- var emptySliceByte byte @@ -365,6 +390,15 @@ func (e *nativeEngine) graphAddNodes(_ context.Context, nodes []Node) (uint64, e }) } +func (e *nativeEngine) graphReplaceNodes(_ context.Context, nodes []Node) (uint64, error) { + payload := marshal(nodes) + return countResult("graph.replace_nodes", func(out *C.uint64_t, errBuf *C.CstxBuffer) C.CstxStatusCode { + rc := C.cstx_graph_replace_nodes(e.handle, byteSlice(payload), out, errBuf) + runtime.KeepAlive(payload) + return rc + }) +} + func (e *nativeEngine) graphAddEdges(_ context.Context, edges []Edge) (uint64, error) { payload := marshal(edges) return countResult("graph.add_edges", func(out *C.uint64_t, errBuf *C.CstxBuffer) C.CstxStatusCode { @@ -576,34 +610,259 @@ func (e *nativeEngine) repoCommit( return commit, err } -func (e *nativeEngine) repoDiff(_ context.Context, baseRef, headRef string, limit *int) (GraphDiff, error) { +type preparedObjectWire struct { + ID string `json:"id"` + Kind string `json:"kind"` + Envelope string `json:"envelope"` +} + +type preparedCommitWire struct { + Commit Commit `json:"commit"` + IndexRoot string `json:"index_root"` + Objects []preparedObjectWire `json:"objects"` +} + +func (e *nativeEngine) repoPrepare( + _ context.Context, + message string, + refName string, + expectedHead *string, + metadata any, + timestamp *int64, +) (PreparedCommit, error) { + metadataJSON, err := marshalInput("repo.prepare", metadata) + if err != nil { + return PreparedCommit{}, err + } + var wire preparedCommitWire + var nativeTimestamp C.int64_t + var hasTimestamp C.uint8_t + if timestamp != nil { + nativeTimestamp = C.int64_t(*timestamp) + hasTimestamp = 1 + } + err = jsonResult("repo.prepare", &wire, func(out, errBuf *C.CstxBuffer) C.CstxStatusCode { + var expected C.CstxSlice + if expectedHead != nil { + expected = stringSlice(*expectedHead) + } + rc := C.cstx_repo_prepare(e.handle, stringSlice(message), stringSlice(refName), expected, byteSlice(metadataJSON), nativeTimestamp, hasTimestamp, out, errBuf) + runtime.KeepAlive(message) + runtime.KeepAlive(refName) + runtime.KeepAlive(expectedHead) + runtime.KeepAlive(metadataJSON) + return rc + }) + if err != nil { + return PreparedCommit{}, err + } + prepared := PreparedCommit{Commit: wire.Commit, IndexRoot: wire.IndexRoot, Objects: make([]PreparedObject, len(wire.Objects))} + for i, object := range wire.Objects { + envelope, err := hex.DecodeString(object.Envelope) + if err != nil { + return PreparedCommit{}, &Error{Code: CodeCorruptData, Operation: "repo.prepare", Message: "invalid object envelope: " + err.Error()} + } + prepared.Objects[i] = PreparedObject{ID: object.ID, Kind: object.Kind, Envelope: envelope} + } + return prepared, nil +} + +func (e *nativeEngine) repoAccept(_ context.Context, commit string) error { + return statusCall("repo.accept", func(errBuf *C.CstxBuffer) C.CstxStatusCode { + rc := C.cstx_repo_accept(e.handle, stringSlice(commit), errBuf) + runtime.KeepAlive(commit) + return rc + }) +} + +func (e *nativeEngine) repoDiscard(_ context.Context) error { + return statusCall("repo.discard", func(errBuf *C.CstxBuffer) C.CstxStatusCode { + return C.cstx_repo_discard(e.handle, errBuf) + }) +} + +func (e *nativeEngine) repoSynchronize(_ context.Context, state RepositorySync) error { + type objectWire struct { + ID string `json:"id"` + Envelope string `json:"envelope"` + } + type refWire struct { + Name string `json:"name"` + Commit *string `json:"commit"` + } + type indexWire struct { + Commit string `json:"commit"` + IndexRoot string `json:"index_root"` + } + payload := struct { + Objects []objectWire `json:"objects"` + Refs []refWire `json:"refs"` + Indexes []indexWire `json:"indexes"` + }{ + Objects: make([]objectWire, len(state.Objects)), + Refs: make([]refWire, len(state.Refs)), + Indexes: make([]indexWire, len(state.Indexes)), + } + for i, object := range state.Objects { + payload.Objects[i] = objectWire{ID: object.ID, Envelope: hex.EncodeToString(object.Envelope)} + } + for i, ref := range state.Refs { + payload.Refs[i] = refWire{Name: ref.Name, Commit: ref.Commit} + } + for i, index := range state.Indexes { + payload.Indexes[i] = indexWire{Commit: index.Commit, IndexRoot: index.IndexRoot} + } + data, err := marshalInput("repo.synchronize", payload) + if err != nil { + return err + } + return statusCall("repo.synchronize", func(errBuf *C.CstxBuffer) C.CstxStatusCode { + rc := C.cstx_repo_synchronize(e.handle, byteSlice(data), errBuf) + runtime.KeepAlive(data) + return rc + }) +} + +func (e *nativeEngine) repoContains(_ context.Context, object string) (bool, error) { + return boolResult("repo.contains", func(out *C.uint8_t, errBuf *C.CstxBuffer) C.CstxStatusCode { + rc := C.cstx_repo_contains(e.handle, stringSlice(object), out, errBuf) + runtime.KeepAlive(object) + return rc + }) +} + +func (e *nativeEngine) repoMissingTree(_ context.Context, commit string) ([]string, error) { + var ids []string + err := jsonResult("repo.missing_tree", &ids, func(out, errBuf *C.CstxBuffer) C.CstxStatusCode { + rc := C.cstx_repo_missing_tree(e.handle, stringSlice(commit), out, errBuf) + runtime.KeepAlive(commit) + return rc + }) + return ids, err +} + +func (e *nativeEngine) repoObjectClosure(_ context.Context, commit string) ([]string, error) { + var ids []string + err := jsonResult("repo.object_closure", &ids, func(out, errBuf *C.CstxBuffer) C.CstxStatusCode { + rc := C.cstx_repo_object_closure(e.handle, stringSlice(commit), out, errBuf) + runtime.KeepAlive(commit) + return rc + }) + return ids, err +} + +func (e *nativeEngine) repoMissingPrepare(_ context.Context, commit string) ([]string, error) { + var ids []string + err := jsonResult("repo.missing_prepare", &ids, func(out, errBuf *C.CstxBuffer) C.CstxStatusCode { + rc := C.cstx_repo_missing_prepare(e.handle, stringSlice(commit), out, errBuf) + runtime.KeepAlive(commit) + return rc + }) + return ids, err +} + +func (e *nativeEngine) repoMissingHistory(_ context.Context, commit, entity string) ([]string, error) { + var ids []string + err := jsonResult("repo.missing_history", &ids, func(out, errBuf *C.CstxBuffer) C.CstxStatusCode { + rc := C.cstx_repo_missing_history(e.handle, stringSlice(commit), stringSlice(entity), out, errBuf) + runtime.KeepAlive(commit) + runtime.KeepAlive(entity) + return rc + }) + return ids, err +} + +func (e *nativeEngine) repoMissingStat(_ context.Context, commit string) ([]string, error) { + var ids []string + err := jsonResult("repo.missing_stat", &ids, func(out, errBuf *C.CstxBuffer) C.CstxStatusCode { + rc := C.cstx_repo_missing_stat(e.handle, stringSlice(commit), out, errBuf) + runtime.KeepAlive(commit) + return rc + }) + return ids, err +} + +func (e *nativeEngine) repoMissingCommits(_ context.Context, commit string, limit int) ([]string, error) { + var ids []string + err := jsonResult("repo.missing_commits", &ids, func(out, errBuf *C.CstxBuffer) C.CstxStatusCode { + rc := C.cstx_repo_missing_commits(e.handle, stringSlice(commit), C.size_t(limit), out, errBuf) + runtime.KeepAlive(commit) + return rc + }) + return ids, err +} + +func (e *nativeEngine) repoMissingDiff(_ context.Context, base, head string, detail DiffDetail) ([]string, error) { + var ids []string + nativeDetail := string(detail) + err := jsonResult("repo.missing_diff", &ids, func(out, errBuf *C.CstxBuffer) C.CstxStatusCode { + rc := C.cstx_repo_missing_diff(e.handle, stringSlice(base), stringSlice(head), stringSlice(nativeDetail), out, errBuf) + runtime.KeepAlive(base) + runtime.KeepAlive(head) + runtime.KeepAlive(nativeDetail) + return rc + }) + return ids, err +} + +func (e *nativeEngine) repoMissingDelta(_ context.Context, commit string, start, end *int64) ([]string, error) { + var ids []string + var nativeStart, nativeEnd C.int64_t + var hasStart, hasEnd C.uint8_t + if start != nil { + nativeStart = C.int64_t(*start) + hasStart = 1 + } + if end != nil { + nativeEnd = C.int64_t(*end) + hasEnd = 1 + } + err := jsonResult("repo.missing_delta", &ids, func(out, errBuf *C.CstxBuffer) C.CstxStatusCode { + rc := C.cstx_repo_missing_delta(e.handle, stringSlice(commit), nativeStart, hasStart, nativeEnd, hasEnd, out, errBuf) + runtime.KeepAlive(commit) + return rc + }) + return ids, err +} + +// repoMissingMerge takes an empty target to mean "merge into the current head", +// which optionalStringSlice turns into the runtime's None. +func (e *nativeEngine) repoMissingMerge(_ context.Context, source, target string) ([]string, error) { + var ids []string + err := jsonResult("repo.missing_merge", &ids, func(out, errBuf *C.CstxBuffer) C.CstxStatusCode { + rc := C.cstx_repo_missing_merge(e.handle, stringSlice(source), optionalStringSlice(target), out, errBuf) + runtime.KeepAlive(source) + runtime.KeepAlive(target) + return rc + }) + return ids, err +} + +func (e *nativeEngine) repoReleaseTransientObjects(_ context.Context) error { + return statusCall("repo.release_transient_objects", func(errBuf *C.CstxBuffer) C.CstxStatusCode { + return C.cstx_repo_release_transient_objects(e.handle, errBuf) + }) +} + +func (e *nativeEngine) repoDiff(_ context.Context, baseRef, headRef string, options DiffOptions) (GraphDiff, error) { var diff GraphDiff var nativeLimit C.size_t var hasLimit C.uint8_t - if limit != nil { - nativeLimit = C.size_t(*limit) + if options.Limit != nil { + nativeLimit = C.size_t(*options.Limit) hasLimit = 1 } + detail := string(options.detail()) err := jsonResult("repo.diff", &diff, func(out, errBuf *C.CstxBuffer) C.CstxStatusCode { - rc := C.cstx_repo_diff(e.handle, stringSlice(baseRef), stringSlice(headRef), nativeLimit, hasLimit, out, errBuf) + rc := C.cstx_repo_diff(e.handle, stringSlice(baseRef), stringSlice(headRef), nativeLimit, hasLimit, stringSlice(detail), out, errBuf) runtime.KeepAlive(baseRef) runtime.KeepAlive(headRef) + runtime.KeepAlive(detail) return rc }) return diff, err } -func (e *nativeEngine) repoDiffStat(_ context.Context, baseRef, headRef string) (Delta, error) { - var value Delta - err := jsonResult("repo.diff_stat", &value, func(out, errBuf *C.CstxBuffer) C.CstxStatusCode { - rc := C.cstx_repo_diff_stat(e.handle, stringSlice(baseRef), stringSlice(headRef), out, errBuf) - runtime.KeepAlive(baseRef) - runtime.KeepAlive(headRef) - return rc - }) - return value, err -} - func (e *nativeEngine) repoHead(_ context.Context, refName string) (*string, error) { var head *string err := jsonResult("repo.head", &head, func(out, errBuf *C.CstxBuffer) C.CstxStatusCode { diff --git a/go/graph.go b/go/graph.go index fd3646e..24c03fd 100644 --- a/go/graph.go +++ b/go/graph.go @@ -16,6 +16,22 @@ func (g *Graph) AddNodes(ctx context.Context, nodes []Node) (uint64, error) { return g.eng.graphAddNodes(ctx, nodes) } +// ReplaceNodes atomically writes each node as its current state and returns the +// number of elements actually changed. +// +// AddNodes merges: fields fill in, sources accumulate, and two different values +// under one extras key are kept as both. That is what aggregating sightings of +// one entity needs. ReplaceNodes is for records that have a current value — an +// oracle that moved from "future" to "intent" has one status — where merging +// would silently keep the old value alongside the new one. Restating an +// unchanged record still reports zero and writes no history. +func (g *Graph) ReplaceNodes(ctx context.Context, nodes []Node) (uint64, error) { + if err := contextError(ctx); err != nil { + return 0, err + } + return g.eng.graphReplaceNodes(ctx, nodes) +} + // AddEdges atomically adds or merges relationships. func (g *Graph) AddEdges(ctx context.Context, edges []Edge) (uint64, error) { if err := contextError(ctx); err != nil { @@ -24,6 +40,22 @@ func (g *Graph) AddEdges(ctx context.Context, edges []Edge) (uint64, error) { return g.eng.graphAddEdges(ctx, edges) } +// DeleteNodes atomically removes nodes and all incident relationships. +func (g *Graph) DeleteNodes(ctx context.Context, nodeIDs []string) (uint64, error) { + if err := contextError(ctx); err != nil { + return 0, err + } + return g.eng.graphDeleteNodes(ctx, nodeIDs) +} + +// DeleteEdges atomically removes relationships by stable CSTX ID. +func (g *Graph) DeleteEdges(ctx context.Context, edgeIDs []string) (uint64, error) { + if err := contextError(ctx); err != nil { + return 0, err + } + return g.eng.graphDeleteEdges(ctx, edgeIDs) +} + // Ingest feeds one linked native-plugin payload into the shared graph. func (g *Graph) Ingest(ctx context.Context, source string, data []byte) (uint64, error) { if err := contextError(ctx); err != nil { diff --git a/go/lib/darwin_amd64/libcstx_ffi.a b/go/lib/darwin_amd64/libcstx_ffi.a index 9cbeaad..8597bd2 100644 Binary files a/go/lib/darwin_amd64/libcstx_ffi.a and b/go/lib/darwin_amd64/libcstx_ffi.a differ diff --git a/go/lib/darwin_arm64/libcstx_ffi.a b/go/lib/darwin_arm64/libcstx_ffi.a index 4a61e88..33186b3 100644 Binary files a/go/lib/darwin_arm64/libcstx_ffi.a and b/go/lib/darwin_arm64/libcstx_ffi.a differ diff --git a/go/lib/linux_amd64/libcstx_ffi.a b/go/lib/linux_amd64/libcstx_ffi.a index ce46fe4..addff3c 100644 Binary files a/go/lib/linux_amd64/libcstx_ffi.a and b/go/lib/linux_amd64/libcstx_ffi.a differ diff --git a/go/lib/linux_arm64/libcstx_ffi.a b/go/lib/linux_arm64/libcstx_ffi.a index afad0e5..7a19a6f 100644 Binary files a/go/lib/linux_arm64/libcstx_ffi.a and b/go/lib/linux_arm64/libcstx_ffi.a differ diff --git a/go/lib/windows_amd64/libcstx_ffi.a b/go/lib/windows_amd64/libcstx_ffi.a index 762e855..6dfffbd 100644 Binary files a/go/lib/windows_amd64/libcstx_ffi.a and b/go/lib/windows_amd64/libcstx_ffi.a differ diff --git a/go/repository.go b/go/repository.go index c17a08b..9a171ca 100644 --- a/go/repository.go +++ b/go/repository.go @@ -39,23 +39,161 @@ func (r *Repository) Commit( return r.eng.repoCommit(ctx, message, refName, expectedHead, metadata) } +// Prepare computes one commit without publishing its ref. The caller must +// durably persist the returned payload and atomically advance the external ref, +// then call Accept. Call Discard when the external transaction fails. +func (r *Repository) Prepare( + ctx context.Context, + message string, + refName string, + expectedHead *string, + metadata any, + timestamp *int64, +) (PreparedCommit, error) { + if err := contextError(ctx); err != nil { + return PreparedCommit{}, err + } + return r.eng.repoPrepare(ctx, message, refName, expectedHead, metadata, timestamp) +} + +// Accept finalizes a prepared commit after its external transaction commits. +func (r *Repository) Accept(ctx context.Context, commit string) error { + if err := contextError(ctx); err != nil { + return err + } + return r.eng.repoAccept(ctx, commit) +} + +// Discard abandons a prepared commit while preserving the working journal. +func (r *Repository) Discard(ctx context.Context) error { + if err := contextError(ctx); err != nil { + return err + } + return r.eng.repoDiscard(ctx) +} + +// Synchronize loads externally persisted objects, refs, and index roots into +// this computation session. +func (r *Repository) Synchronize(ctx context.Context, state RepositorySync) error { + if err := contextError(ctx); err != nil { + return err + } + return r.eng.repoSynchronize(ctx, state) +} + +func (r *Repository) Contains(ctx context.Context, object string) (bool, error) { + if err := contextError(ctx); err != nil { + return false, err + } + return r.eng.repoContains(ctx, object) +} + +// MissingTree plans immutable object reads required to materialize a commit. +func (r *Repository) MissingTree(ctx context.Context, commit string) ([]string, error) { + if err := contextError(ctx); err != nil { + return nil, err + } + return r.eng.repoMissingTree(ctx, commit) +} + +// ObjectClosure returns every object one commit and its ancestry are built +// from. Deleting whatever the union of this set over every ref does not name +// reclaims space without breaking any supported operation on those refs. +// +// It answers from stored bytes, so unlike the Missing* planners the result does +// not depend on what this process has already loaded. +func (r *Repository) ObjectClosure(ctx context.Context, commit string) ([]string, error) { + if err := contextError(ctx); err != nil { + return nil, err + } + return r.eng.repoObjectClosure(ctx, commit) +} + +// MissingPrepare plans index reads required before preparing a child commit. +func (r *Repository) MissingPrepare(ctx context.Context, commit string) ([]string, error) { + if err := contextError(ctx); err != nil { + return nil, err + } + return r.eng.repoMissingPrepare(ctx, commit) +} + +// MissingHistory plans the index reads required to answer History for one +// entity. A host that keeps objects outside the runtime resolves this to empty +// before calling History; the index only pages in the postings for that entity, +// so the walk costs what the entity changed, not what the range contains. +func (r *Repository) MissingHistory(ctx context.Context, commit, entity string) ([]string, error) { + if err := contextError(ctx); err != nil { + return nil, err + } + return r.eng.repoMissingHistory(ctx, commit, entity) +} + +// MissingStat plans the reads required to summarize a commit. +func (r *Repository) MissingStat(ctx context.Context, commit string) ([]string, error) { + if err := contextError(ctx); err != nil { + return nil, err + } + return r.eng.repoMissingStat(ctx, commit) +} + +// MissingCommits plans the reads required to walk a commit's ancestry. +func (r *Repository) MissingCommits(ctx context.Context, commit string, limit int) ([]string, error) { + if err := contextError(ctx); err != nil { + return nil, err + } + return r.eng.repoMissingCommits(ctx, commit, limit) +} + +// MissingDiff plans the reads required to diff two revisions at one detail +// level. A limit never narrows the plan, so it is not part of the request. +func (r *Repository) MissingDiff(ctx context.Context, base, head string, detail DiffDetail) ([]string, error) { + if err := contextError(ctx); err != nil { + return nil, err + } + if detail == "" { + detail = DiffEntities + } + return r.eng.repoMissingDiff(ctx, base, head, detail) +} + +// MissingDelta plans the reads required to count changes in a time range. +func (r *Repository) MissingDelta(ctx context.Context, commit string, start, end *int64) ([]string, error) { + if err := contextError(ctx); err != nil { + return nil, err + } + return r.eng.repoMissingDelta(ctx, commit, start, end) +} + +// MissingMerge plans the reads required to merge source into target. An empty +// target means the current head. +func (r *Repository) MissingMerge(ctx context.Context, source, target string) ([]string, error) { + if err := contextError(ctx); err != nil { + return nil, err + } + return r.eng.repoMissingMerge(ctx, source, target) +} + +// ReleaseTransientObjects drops objects hydrated for one external operation. +func (r *Repository) ReleaseTransientObjects(ctx context.Context) error { + if err := contextError(ctx); err != nil { + return err + } + return r.eng.repoReleaseTransientObjects(ctx) +} + +// Diff compares two revisions. Options select a limit on the reported entity +// IDs and whether they are reported at all; the counts in GraphDiff.Stats are +// exact either way. func (r *Repository) Diff( ctx context.Context, base string, head string, - limit *int, + options DiffOptions, ) (GraphDiff, error) { if err := contextError(ctx); err != nil { return GraphDiff{}, err } - return r.eng.repoDiff(ctx, base, head, limit) -} - -func (r *Repository) DiffStat(ctx context.Context, base, head string) (Delta, error) { - if err := contextError(ctx); err != nil { - return Delta{}, err - } - return r.eng.repoDiffStat(ctx, base, head) + return r.eng.repoDiff(ctx, base, head, options) } func (r *Repository) Log( diff --git a/go/repository_history_test.go b/go/repository_history_test.go new file mode 100644 index 0000000..9c35c14 --- /dev/null +++ b/go/repository_history_test.go @@ -0,0 +1,165 @@ +package cstx + +import ( + "fmt" + "testing" +) + +// A per-entity history is cheap because the index pages in the postings for that +// one entity, not every object the range's snapshots contain. That property is +// only reachable from Go once MissingHistory exists: a host that keeps objects +// outside the runtime has no other way to learn which index pages to hand over, +// and would have to fall back to materializing each snapshot and comparing +// content hashes — the very cost the index exists to avoid. +// +// This exercises the whole external-storage round trip and then measures both +// plans against the same stored objects. +func TestRepositoryHistoryHydratesOnlyEntityPostings(t *testing.T) { + const rounds = 12 + const width = 20 + const tracked = "domain:tracked.example" + + writer := openRuntime(t) + objects := map[string]RepositoryObject{} + var head, indexRoot string + var commits []string + var commitObject RepositoryObject + + for round := range rounds { + // The tracked node changes every round... + if _, err := writer.Graph.AddNodes(testContext, []Node{{ + ID: tracked, Type: "domain", Value: "tracked.example", + Model: map[string]any{"domain": "tracked.example", "cstx_flags": 0}, + Sources: []string{"test"}, + Extras: map[string]any{"round": round}, + }}); err != nil { + t.Fatalf("round %d tracked node: %v", round, err) + } + // ...surrounded by nodes that do not, so the snapshot is wide while the + // entity's own history stays short. + filler := make([]Node, 0, width) + for i := range width { + filler = append(filler, domainNode(fmt.Sprintf("filler-%d-%d.example", round, i))) + } + if _, err := writer.Graph.AddNodes(testContext, filler); err != nil { + t.Fatalf("round %d filler: %v", round, err) + } + + var expected *string + if head != "" { + expected = &head + } + prepared, err := writer.Repo.Prepare( + testContext, fmt.Sprintf("round %d", round), "main", expected, nil, nil, + ) + if err != nil { + t.Fatalf("prepare round %d: %v", round, err) + } + for _, object := range prepared.Objects { + stored := RepositoryObject{ID: object.ID, Envelope: append([]byte(nil), object.Envelope...)} + objects[object.ID] = stored + if object.Kind == "commit" && object.ID == prepared.Commit.ID { + commitObject = stored + } + } + if err := writer.Repo.Accept(testContext, prepared.Commit.ID); err != nil { + t.Fatalf("accept round %d: %v", round, err) + } + head = prepared.Commit.ID + commits = append(commits, head) + indexRoot = prepared.IndexRoot + } + + rootObject, ok := objects[indexRoot] + if !ok { + t.Fatalf("index root %s was never published as an object", indexRoot) + } + + // seed gives a fresh runtime only the commit frontier: it owns no tree and no + // postings, so everything a plan needs has to arrive through synchronize. + seed := func() *CSTX { + reader := openRuntime(t) + if err := reader.Repo.Synchronize(testContext, RepositorySync{ + Objects: []RepositoryObject{commitObject, rootObject}, + }); err != nil { + t.Fatalf("synchronize frontier objects: %v", err) + } + if err := reader.Repo.Synchronize(testContext, RepositorySync{ + Refs: []RepositoryRef{{Name: "main", Commit: &head}}, + Indexes: []RepositoryIndex{{Commit: head, IndexRoot: indexRoot}}, + }); err != nil { + t.Fatalf("synchronize frontier refs: %v", err) + } + return reader + } + + hydrate := func(reader *CSTX, plan func() ([]string, error)) int { + read := 0 + for { + missing, err := plan() + if err != nil { + t.Fatalf("plan: %v", err) + } + if len(missing) == 0 { + return read + } + batch := make([]RepositoryObject, 0, len(missing)) + for _, id := range missing { + object, ok := objects[id] + if !ok { + t.Fatalf("planner requested an object that was never stored: %s", id) + } + batch = append(batch, object) + } + read += len(batch) + if err := reader.Repo.Synchronize(testContext, RepositorySync{Objects: batch}); err != nil { + t.Fatalf("synchronize: %v", err) + } + } + } + + historyReader := seed() + historyObjects := hydrate(historyReader, func() ([]string, error) { + return historyReader.Repo.MissingHistory(testContext, head, tracked) + }) + entries, err := historyReader.Repo.History(testContext, tracked, head, nil) + if err != nil { + t.Fatalf("history: %v", err) + } + if len(entries.Entries) != rounds { + t.Fatalf("history returned %d entries, want %d (one per round)", len(entries.Entries), rounds) + } + + // The fallback a host without MissingHistory is stuck with: materialize every + // snapshot in the range and compare the entity's content hash across them. + // One snapshot is cheap; the range is not, and it grows with history depth + // while the entity's own change count does not. + walkObjects := 0 + for _, commit := range commits { + reader := openRuntime(t) + commitEnvelope, ok := objects[commit] + if !ok { + t.Fatalf("commit object %s was never published", commit) + } + if err := reader.Repo.Synchronize(testContext, RepositorySync{ + Objects: []RepositoryObject{commitEnvelope}, + }); err != nil { + t.Fatalf("synchronize commit %s: %v", commit, err) + } + walkObjects += 1 + hydrate(reader, func() ([]string, error) { + return reader.Repo.MissingTree(testContext, commit) + }) + } + + if historyObjects >= walkObjects { + t.Fatalf( + "per-entity history hydrated %d objects and the snapshot walk hydrated %d "+ + "across %d commits; the index exists so the walk is not needed", + historyObjects, walkObjects, len(commits), + ) + } + t.Logf( + "per-entity history: %d objects for %d changes; snapshot walk over %d commits: %d objects", + historyObjects, len(entries.Entries), len(commits), walkObjects, + ) +} diff --git a/go/types.go b/go/types.go index 74a682d..21025bb 100644 --- a/go/types.go +++ b/go/types.go @@ -128,6 +128,50 @@ type Commit struct { CreatedAt int64 `json:"created_at"` } +// PreparedObject is one immutable CSTX object ready for external persistence. +// Envelope is the canonical encoded object and must be stored without changes. +type PreparedObject struct { + ID string + Kind string + Envelope []byte +} + +// PreparedCommit is the complete immutable portion of one external publish +// transaction. The ref must only be advanced after all Objects and IndexRoot +// are durably stored. +type PreparedCommit struct { + Commit Commit + IndexRoot string + Objects []PreparedObject +} + +// RepositoryObject hydrates one immutable object into a CSTX computation +// session. The ID is verified against Envelope by the runtime. +type RepositoryObject struct { + ID string + Envelope []byte +} + +// RepositoryRef synchronizes one mutable named reference. A nil Commit deletes +// the reference from the computation session. +type RepositoryRef struct { + Name string + Commit *string +} + +// RepositoryIndex binds a commit to its immutable history index root. +type RepositoryIndex struct { + Commit string + IndexRoot string +} + +// RepositorySync is one batch of externally persisted repository state. +type RepositorySync struct { + Objects []RepositoryObject + Refs []RepositoryRef + Indexes []RepositoryIndex +} + // GraphDiff groups added, removed, and modified element IDs by element type. type GraphDiff struct { Added map[string][]string `json:"added"` @@ -137,6 +181,36 @@ type GraphDiff struct { // change. An empty group is otherwise ambiguous between "nothing of that // type changed" and "the limit ran out first". Truncated bool `json:"truncated"` + // Stats counts the whole compared range, whatever a limit left out of the + // maps above. + Stats Delta `json:"stats"` +} + +// DiffDetail selects how much of a diff the caller needs back. +type DiffDetail string + +const ( + // DiffEntities lists every changed entity, and counts them. + DiffEntities DiffDetail = "entities" + // DiffCounts returns counts alone, which page summaries can often answer + // without reading the pages themselves. + DiffCounts DiffDetail = "counts" +) + +// DiffOptions is one diff request. The zero value lists entities without a +// limit. +type DiffOptions struct { + // Limit caps reported entity IDs. Counts stay exact whatever it drops. + Limit *int + // Detail selects the entity lists or counts alone. + Detail DiffDetail +} + +func (o DiffOptions) detail() DiffDetail { + if o.Detail == "" { + return DiffEntities + } + return o.Detail } // JoinRuleSpec is the portable native-linker rule shared by all bindings. diff --git a/include/cstx_ffi.h b/include/cstx_ffi.h index 4af148a..6c1e802 100644 --- a/include/cstx_ffi.h +++ b/include/cstx_ffi.h @@ -131,11 +131,34 @@ CstxStatusCode cstx_graph_add_nodes(struct CstxHandle *handle, uint64_t *affected, struct CstxBuffer *error); +/** + * Write each node as its current state, replacing the stored record. + * + * The merge path (`cstx_graph_add_nodes`) owns bulk ingest and keeps its JSON + * fast path. A replace batch is a caller restating records it already holds — + * a task's oracles, a document's current revision — so it goes through the + * shared `Value` path rather than earning a second parser. + */ +CstxStatusCode cstx_graph_replace_nodes(struct CstxHandle *handle, + struct CstxSlice data, + uint64_t *affected, + struct CstxBuffer *error); + CstxStatusCode cstx_graph_add_edges(struct CstxHandle *handle, struct CstxSlice data, uint64_t *affected, struct CstxBuffer *error); +CstxStatusCode cstx_graph_delete_nodes(struct CstxHandle *handle, + struct CstxSlice node_ids_json, + uint64_t *output, + struct CstxBuffer *error); + +CstxStatusCode cstx_graph_delete_edges(struct CstxHandle *handle, + struct CstxSlice edge_ids_json, + uint64_t *output, + struct CstxBuffer *error); + CstxStatusCode cstx_graph_ingest(struct CstxHandle *handle, struct CstxSlice source, struct CstxSlice data, @@ -362,20 +385,97 @@ CstxStatusCode cstx_repo_commit(struct CstxHandle *handle, struct CstxBuffer *output, struct CstxBuffer *error); +CstxStatusCode cstx_repo_prepare(struct CstxHandle *handle, + struct CstxSlice message, + struct CstxSlice ref_name, + struct CstxSlice expected_head, + struct CstxSlice metadata_json, + int64_t timestamp, + uint8_t has_timestamp, + struct CstxBuffer *output, + struct CstxBuffer *error); + +CstxStatusCode cstx_repo_accept(struct CstxHandle *handle, + struct CstxSlice commit, + struct CstxBuffer *error); + +CstxStatusCode cstx_repo_discard(struct CstxHandle *handle, struct CstxBuffer *error); + +CstxStatusCode cstx_repo_synchronize(struct CstxHandle *handle, + struct CstxSlice payload_json, + struct CstxBuffer *error); + +CstxStatusCode cstx_repo_contains(struct CstxHandle *handle, + struct CstxSlice object, + uint8_t *output, + struct CstxBuffer *error); + +CstxStatusCode cstx_repo_missing_tree(struct CstxHandle *handle, + struct CstxSlice commit, + struct CstxBuffer *output, + struct CstxBuffer *error); + +CstxStatusCode cstx_repo_object_closure(struct CstxHandle *handle, + struct CstxSlice commit, + struct CstxBuffer *output, + struct CstxBuffer *error); + +CstxStatusCode cstx_repo_missing_prepare(struct CstxHandle *handle, + struct CstxSlice commit, + struct CstxBuffer *output, + struct CstxBuffer *error); + +CstxStatusCode cstx_repo_missing_history(struct CstxHandle *handle, + struct CstxSlice commit, + struct CstxSlice entity_id, + struct CstxBuffer *output, + struct CstxBuffer *error); + +CstxStatusCode cstx_repo_missing_stat(struct CstxHandle *handle, + struct CstxSlice commit, + struct CstxBuffer *output, + struct CstxBuffer *error); + +CstxStatusCode cstx_repo_missing_commits(struct CstxHandle *handle, + struct CstxSlice commit, + size_t limit, + struct CstxBuffer *output, + struct CstxBuffer *error); + +CstxStatusCode cstx_repo_missing_diff(struct CstxHandle *handle, + struct CstxSlice base, + struct CstxSlice head, + struct CstxSlice detail, + struct CstxBuffer *output, + struct CstxBuffer *error); + +CstxStatusCode cstx_repo_missing_delta(struct CstxHandle *handle, + struct CstxSlice commit, + int64_t start_timestamp, + uint8_t has_start, + int64_t end_timestamp, + uint8_t has_end, + struct CstxBuffer *output, + struct CstxBuffer *error); + +CstxStatusCode cstx_repo_missing_merge(struct CstxHandle *handle, + struct CstxSlice source, + struct CstxSlice target, + struct CstxBuffer *output, + struct CstxBuffer *error); + +CstxStatusCode cstx_repo_release_transient_objects(struct CstxHandle *handle, + struct CstxBuffer *error); + CstxStatusCode cstx_repo_diff(struct CstxHandle *handle, struct CstxSlice base_ref, struct CstxSlice head_ref, size_t limit, uint8_t has_limit, + struct CstxSlice detail, struct CstxBuffer *output, struct CstxBuffer *error); -CstxStatusCode cstx_repo_diff_stat(struct CstxHandle *handle, - struct CstxSlice base_ref, - struct CstxSlice head_ref, - struct CstxBuffer *output, - struct CstxBuffer *error); - CstxStatusCode cstx_repo_head(struct CstxHandle *handle, struct CstxSlice ref_name, struct CstxBuffer *output, diff --git a/python/python/cstxpy/_cstxpy.pyi b/python/python/cstxpy/_cstxpy.pyi index 8f4cef3..ea49289 100644 --- a/python/python/cstxpy/_cstxpy.pyi +++ b/python/python/cstxpy/_cstxpy.pyi @@ -232,10 +232,22 @@ class CSTXGraph: """Atomically mutate native dictionaries without a JSON round trip.""" ... + def replace_nodes(self, nodes: list[dict[str, Any]]) -> int: + """Atomically overwrite native dictionaries instead of merging them.""" + ... + def add_edges(self, edges: list[dict[str, Any]]) -> int: """Atomically mutate native relationship dictionaries.""" ... + def delete_nodes(self, node_ids: list[str]) -> int: + """Atomically remove nodes and their incident relationships.""" + ... + + def delete_edges(self, edge_ids: list[str]) -> int: + """Atomically remove relationships by stable CSTX ID.""" + ... + def node(self, node_id: str) -> dict[str, Any]: """Return one node dictionary or raise ``CSTXError(NOT_FOUND)``.""" ... @@ -432,6 +444,10 @@ class Repository: """Return graph-tree objects missing from the native object set.""" ... + def _object_closure(self, commit: bytes) -> list[bytes]: + """Return every object this commit and its ancestry are built from.""" + ... + def _missing_stat(self, commit: bytes) -> list[bytes]: """Return the graph root needed for persisted statistics.""" ... @@ -444,10 +460,6 @@ class Repository: """Return the commit frontier or graph objects needed by merge.""" ... - def _missing_diff_stat(self, base: bytes, head: bytes) -> list[bytes]: - """Return index objects needed for an exact count-only diff.""" - ... - def _missing_delta( self, commit: bytes, @@ -473,9 +485,13 @@ class Repository: self, base: bytes, head: bytes, - limit: int | None = None, + detail: str = "entities", ) -> list[bytes]: - """Return index or graph objects needed for a revision diff.""" + """Return index or graph objects needed for a revision diff. + + A limit never narrows the plan, so the request carries only the detail + level: ``"counts"`` skips the pages a page summary already answers for. + """ ... def _commits(self, commit: bytes, limit: int) -> list[bytes]: @@ -514,12 +530,13 @@ class Repository: base: str, head: str, limit: int | None = None, + detail: str = "entities", ) -> dict[str, Any]: - """Compare two revisions with an optional result limit.""" - ... + """Compare two revisions. - def diff_stat(self, base: str, head: str) -> dict[str, Any]: - """Count an exact diff without materializing entity IDs.""" + ``limit`` bounds the reported entity IDs; ``detail="counts"`` drops them + entirely. ``stats`` counts the whole range either way. + """ ... def log( diff --git a/python/tests/test_graph.py b/python/tests/test_graph.py index ceaeb4b..ef1e602 100644 --- a/python/tests/test_graph.py +++ b/python/tests/test_graph.py @@ -527,6 +527,9 @@ def test_every_supported_api_has_runtime_documentation(): "analyze", "add_nodes", "add_edges", + "replace_nodes", + "delete_nodes", + "delete_edges", "node", "edge", "create_relationship", @@ -564,7 +567,6 @@ def test_every_supported_api_has_runtime_documentation(): "checkout", "commit", "diff", - "diff_stat", "log", "history", "branch", diff --git a/ts/wasm/cstx_wasm.d.ts b/ts/wasm/cstx_wasm.d.ts index bdfdae1..414f00c 100644 --- a/ts/wasm/cstx_wasm.d.ts +++ b/ts/wasm/cstx_wasm.d.ts @@ -79,8 +79,7 @@ export class Repository { checkout(revision?: string | null, force?: boolean | null): any; commit(message: string, ref_name?: string | null, expected_head?: string | null, metadata?: any | null, timestamp?: bigint | null): any; delta(revision?: string | null, start_timestamp?: bigint | null, end_timestamp?: bigint | null): any; - diff(base: string, head: string, limit?: number | null): any; - diffStat(base: string, head: string): any; + diff(base: string, head: string, limit?: number | null, detail?: string | null): any; head(ref_name?: string | null): any; history(entity_id: string, revision?: string | null, limit?: number | null): any; log(revision?: string | null, limit?: number | null): any; @@ -172,8 +171,7 @@ export interface InitOutput { readonly repository_resolve: (a: number, b: number, c: number, d: number) => void; readonly repository_checkout: (a: number, b: number, c: number, d: number, e: number) => void; readonly repository_commit: (a: number, b: number, c: number, d: number, e: number, f: number, g: number, h: number, i: number, j: number, k: bigint) => void; - readonly repository_diff: (a: number, b: number, c: number, d: number, e: number, f: number, g: number) => void; - readonly repository_diffStat: (a: number, b: number, c: number, d: number, e: number, f: number) => void; + readonly repository_diff: (a: number, b: number, c: number, d: number, e: number, f: number, g: number, h: number, i: number) => void; readonly repository_head: (a: number, b: number, c: number, d: number) => void; readonly repository_branch: (a: number, b: number, c: number, d: number, e: number, f: number) => void; readonly repository_history: (a: number, b: number, c: number, d: number, e: number, f: number, g: number) => void; @@ -196,8 +194,8 @@ export interface InitOutput { readonly rust_zstd_wasm_shim_memcpy: (a: number, b: number, c: number) => number; readonly rust_zstd_wasm_shim_memmove: (a: number, b: number, c: number) => number; readonly rust_zstd_wasm_shim_memset: (a: number, b: number, c: number) => number; - readonly __wbg_schemas_free: (a: number, b: number) => void; readonly __wbg_repository_free: (a: number, b: number) => void; + readonly __wbg_schemas_free: (a: number, b: number) => void; readonly cstx_schemas: (a: number) => number; readonly cstx_repository: (a: number) => number; readonly __wbindgen_export: (a: number, b: number) => number; diff --git a/ts/wasm/cstx_wasm.js b/ts/wasm/cstx_wasm.js index 8347797..b72729d 100644 --- a/ts/wasm/cstx_wasm.js +++ b/ts/wasm/cstx_wasm.js @@ -1100,40 +1100,19 @@ export class Repository { * @param {string} base * @param {string} head * @param {number | null} [limit] + * @param {string | null} [detail] * @returns {any} */ - diff(base, head, limit) { + diff(base, head, limit, detail) { try { const retptr = wasm.__wbindgen_add_to_stack_pointer(-16); const ptr0 = passStringToWasm0(base, wasm.__wbindgen_export, wasm.__wbindgen_export2); const len0 = WASM_VECTOR_LEN; const ptr1 = passStringToWasm0(head, wasm.__wbindgen_export, wasm.__wbindgen_export2); const len1 = WASM_VECTOR_LEN; - wasm.repository_diff(retptr, this.__wbg_ptr, ptr0, len0, ptr1, len1, isLikeNone(limit) ? Number.MAX_SAFE_INTEGER : (limit) >>> 0); - var r0 = getDataViewMemory0().getInt32(retptr + 4 * 0, true); - var r1 = getDataViewMemory0().getInt32(retptr + 4 * 1, true); - var r2 = getDataViewMemory0().getInt32(retptr + 4 * 2, true); - if (r2) { - throw takeObject(r1); - } - return takeObject(r0); - } finally { - wasm.__wbindgen_add_to_stack_pointer(16); - } - } - /** - * @param {string} base - * @param {string} head - * @returns {any} - */ - diffStat(base, head) { - try { - const retptr = wasm.__wbindgen_add_to_stack_pointer(-16); - const ptr0 = passStringToWasm0(base, wasm.__wbindgen_export, wasm.__wbindgen_export2); - const len0 = WASM_VECTOR_LEN; - const ptr1 = passStringToWasm0(head, wasm.__wbindgen_export, wasm.__wbindgen_export2); - const len1 = WASM_VECTOR_LEN; - wasm.repository_diffStat(retptr, this.__wbg_ptr, ptr0, len0, ptr1, len1); + var ptr2 = isLikeNone(detail) ? 0 : passStringToWasm0(detail, wasm.__wbindgen_export, wasm.__wbindgen_export2); + var len2 = WASM_VECTOR_LEN; + wasm.repository_diff(retptr, this.__wbg_ptr, ptr0, len0, ptr1, len1, isLikeNone(limit) ? Number.MAX_SAFE_INTEGER : (limit) >>> 0, ptr2, len2); var r0 = getDataViewMemory0().getInt32(retptr + 4 * 0, true); var r1 = getDataViewMemory0().getInt32(retptr + 4 * 1, true); var r2 = getDataViewMemory0().getInt32(retptr + 4 * 2, true); diff --git a/ts/wasm/cstx_wasm_bg.wasm b/ts/wasm/cstx_wasm_bg.wasm index 504c753..1a31bbc 100644 Binary files a/ts/wasm/cstx_wasm_bg.wasm and b/ts/wasm/cstx_wasm_bg.wasm differ diff --git a/ts/wasm/cstx_wasm_bg.wasm.d.ts b/ts/wasm/cstx_wasm_bg.wasm.d.ts index 9f3a34b..642ed87 100644 --- a/ts/wasm/cstx_wasm_bg.wasm.d.ts +++ b/ts/wasm/cstx_wasm_bg.wasm.d.ts @@ -59,8 +59,7 @@ export const graph_elevate: (a: number, b: number, c: number, d: number) => void export const repository_resolve: (a: number, b: number, c: number, d: number) => void; export const repository_checkout: (a: number, b: number, c: number, d: number, e: number) => void; export const repository_commit: (a: number, b: number, c: number, d: number, e: number, f: number, g: number, h: number, i: number, j: number, k: bigint) => void; -export const repository_diff: (a: number, b: number, c: number, d: number, e: number, f: number, g: number) => void; -export const repository_diffStat: (a: number, b: number, c: number, d: number, e: number, f: number) => void; +export const repository_diff: (a: number, b: number, c: number, d: number, e: number, f: number, g: number, h: number, i: number) => void; export const repository_head: (a: number, b: number, c: number, d: number) => void; export const repository_branch: (a: number, b: number, c: number, d: number, e: number, f: number) => void; export const repository_history: (a: number, b: number, c: number, d: number, e: number, f: number, g: number) => void; @@ -83,8 +82,8 @@ export const rust_zstd_wasm_shim_free: (a: number) => void; export const rust_zstd_wasm_shim_memcpy: (a: number, b: number, c: number) => number; export const rust_zstd_wasm_shim_memmove: (a: number, b: number, c: number) => number; export const rust_zstd_wasm_shim_memset: (a: number, b: number, c: number) => number; -export const __wbg_schemas_free: (a: number, b: number) => void; export const __wbg_repository_free: (a: number, b: number) => void; +export const __wbg_schemas_free: (a: number, b: number) => void; export const cstx_schemas: (a: number) => number; export const cstx_repository: (a: number) => number; export const __wbindgen_export: (a: number, b: number) => number;