diff --git a/packages/agent/src/dkg-agent-lifecycle.ts b/packages/agent/src/dkg-agent-lifecycle.ts index 61e19012f3..e3a20f3c7e 100644 --- a/packages/agent/src/dkg-agent-lifecycle.ts +++ b/packages/agent/src/dkg-agent-lifecycle.ts @@ -131,7 +131,6 @@ import { type WorkspaceAgentRecipientResolverInput, type WorkspaceSenderKeyEncryptInput, type SharedMemoryPublicSnapshotStorageConfig, type WorkspacePublicSnapshotStore, - withKeyedLocks, swmKaWriteLockKey, } from '@origintrail-official/dkg-publisher'; import { ethers } from 'ethers'; import { join } from 'node:path'; @@ -239,6 +238,7 @@ import { getSyncCheckpointKey } from './sync/checkpoint/state.js'; import { runDurableSync, type VerifiedFullSnapshot } from './sync/requester/durable-sync.js'; import { resolveSyncAgentsMeta, shouldWithholdAgentsDurableMeta } from './sync/agents-meta-policy.js'; import { runSharedMemorySync, sharedMemoryOwnershipKeyFromGraph } from './sync/requester/shared-memory-sync.js'; +import { createSharedMemorySnapshotMaterializer } from './sync/requester/swm-snapshot-materializer.js'; import { runOrderedContextGraphSyncs, type ContextGraphSyncWork, @@ -4853,69 +4853,18 @@ export class LifecycleSyncMethods extends DKGAgentBase { // data quads for them by design — their content arrives as // immutable snapshots, and without this the catch-up lane cached // every verified snapshot and never wrote one to the store. - snapshotMaterializer: { - // The SAME lock the live-gossip write path takes: this.writeLocks - // is the map injected into SharedMemoryHandler, and the key comes - // from the shared helper so the two sites cannot drift. This is - // what closes the check-then-replace race with gossip. - withKaWriteLock: (contextGraphId, subGraphName, kaUal, fn) => - withKeyedLocks(this.writeLocks, [swmKaWriteLockKey(contextGraphId, subGraphName, kaUal)], fn), - // MUST prove the CONTENT is present, not merely that the metadata - // pointer is. The pre-fix bug inserted the head->assertionGraph - // marker while never writing the graph — that IS the observed - // "0 data + N meta" state. A marker-only predicate reports every - // already-broken node as materialized and skips the cached - // snapshot, so the repair would never reach the nodes that need - // it most, and a partially-fetched metadata round could strand an - // asset forever behind its own marker. - // - // Count the assertion graph itself and require it to match the - // descriptor's public quad count: exact-IRI scope, so bounded. - isGraphAssetMaterialized: async (asset) => { - const expected = Number(asset.publicQuadsCount); - if (!Number.isFinite(expected) || expected <= 0) return false; - const result = await this.store.query( - `SELECT (COUNT(*) AS ?n) WHERE { GRAPH <${assertSafeIri(asset.assertionGraph)}> { ?s ?p ?o } }`, - { priority: 'background', source: 'agent.sharedMemorySync.isGraphAssetMaterialized' }, - ); - if (result.type !== 'bindings' || result.bindings.length === 0) return false; - const raw = String(result.bindings[0]?.['n'] ?? '0').replace(/^"|"[^"]*$/g, ''); - const present = Number.parseInt(raw, 10); - // Strictly equal: a short graph is a partial write and must be - // replaced, not treated as already materialized. - return Number.isFinite(present) && present === expected; - }, - // Read INSIDE the lock by the caller: a lock prevents - // interleaving but not overwriting-with-older, and gossip may - // have advanced this KA while catch-up waited on the lock. - readStoredAssertionVersion: async (asset) => { - const result = await this.store.query( - `SELECT (MAX(?v) AS ?v) WHERE { GRAPH <${assertSafeIri(asset.metaGraph)}> { ` - + `<${assertSafeIri(asset.headSubject)}> ` - + ` ?v } }`, - { priority: 'background', source: 'agent.sharedMemorySync.readStoredAssertionVersion' }, - ); - if (result.type !== 'bindings' || result.bindings.length === 0) return null; - const raw = String(result.bindings[0]?.['v'] ?? ''); - if (raw.length === 0) return null; - const literal = /^"([^"]*)"/.exec(raw); - return literal ? literal[1] : raw; - }, - // Deliberately NOT routed through storeInsert below: that is a - // union insert with an oversize guard, whereas a KA graph is - // all-or-nothing and digest-verified. Insert would risk partial - // or duplicated graph state across retries. - replaceGraph: async (graphUri, quads) => { - if (typeof this.store.replaceGraph !== 'function') { - throw new Error('triple store does not support atomic graph replace'); - } - await this.store.replaceGraph(graphUri, quads, { - priority: 'background', - source: 'agent.sharedMemorySync.materializeSnapshot', - }); - this.invalidateListContextGraphsCache(); - }, - }, + // Thin wiring only: the materialization policy (content-digest + // guard, MAX head read + duplicate repair, atomic replace, head + // metadata swap) lives in `swm-snapshot-materializer.ts`. What + // the agent contributes here is its own resources — the store, + // the SAME lock map injected into SharedMemoryHandler (sharing + // the map + key helper is what closes the check-then-replace + // race with gossip), and list-cache invalidation. + snapshotMaterializer: createSharedMemorySnapshotMaterializer({ + store: this.store, + writeLocks: this.writeLocks, + invalidateListContextGraphsCache: () => this.invalidateListContextGraphsCache(), + }), storeInsert: async (quads) => { // Oversize guard (OT-RFC-56): drop+tombstone protocol-violating // literals BEFORE insert so the SWM page cursor advances instead diff --git a/packages/agent/src/sync/requester/shared-memory-sync.ts b/packages/agent/src/sync/requester/shared-memory-sync.ts index 4768a67894..9e0edf6d0d 100644 --- a/packages/agent/src/sync/requester/shared-memory-sync.ts +++ b/packages/agent/src/sync/requester/shared-memory-sync.ts @@ -11,6 +11,7 @@ import { parseGraphScopedSwmRecoveryDescriptors, type GraphScopedSwmRecoveryDescriptor, } from '../graph-scoped-swm-recovery.js'; +import type { SharedMemorySnapshotMaterializer } from './swm-snapshot-materializer.js'; const DKG = 'http://dkg.io/ontology/'; @@ -69,55 +70,17 @@ interface SharedMemorySyncContext { storeInsert: (quads: Quad[]) => Promise; /** * Everything needed to MATERIALIZE verified public SWM snapshots into the - * triple store, as ONE cohesive dependency. - * - * Why one object: these capabilities are only meaningful together. An - * earlier revision exposed them as independent optionals, which allowed a - * silent half-configured mode — a caller supplying the snapshot store but - * not the guard would compile fine and quietly skip materialization. - * Absent entirely => materialization is skipped (never half-applied). + * triple store, as ONE cohesive dependency — the contract (and the + * production implementation) live in `swm-snapshot-materializer.ts`. * * Why it exists at all: contentScopeVersion-2 KAs carry no dkg:rootEntity, * so the aggregate data phase legitimately returns 0 data quads for them — * their content travels as immutable snapshots. The catch-up lane fetched * and VERIFIED those snapshots and then never wrote them, so a node that * missed the live gossip stayed empty forever ("0 data + N meta triples"). + * Absent entirely => materialization is skipped (never half-applied). */ - snapshotMaterializer?: { - /** - * Serialize against the live-gossip write path for one KA. MUST take the - * same key on the same lock map SharedMemoryHandler uses (the agent owns - * the map; derive the key with swmKaWriteLockKey). Without it this - * interleaving destroys data: catch-up observes the graph absent → gossip - * commits a richer version → catch-up replaces it with the older snapshot. - */ - withKaWriteLock: ( - contextGraphId: string, - subGraphName: string | undefined, - kaUal: string, - fn: () => Promise, - ) => Promise; - /** - * True only when the KA's assertion graph CONTENT is present and matches - * the descriptor's public quad count. A marker-only predicate re-reports - * the pre-fix broken state (head metadata written, graph never written) as - * materialized, so the repair would skip exactly the nodes that need it. - */ - isGraphAssetMaterialized: (descriptor: GraphScopedSwmRecoveryDescriptor) => Promise; - /** - * The assertionVersion currently recorded on the local head for this KA, - * or null when no head exists. Read INSIDE the lock: a lock prevents - * interleaving but not overwriting-with-older, and gossip may have - * committed a newer version while catch-up waited. - */ - readStoredAssertionVersion: (descriptor: GraphScopedSwmRecoveryDescriptor) => Promise; - /** - * Atomic whole-graph replace. Replace, not insert: a KA graph is - * all-or-nothing and digest-verified; union-insert risks partial or - * duplicated state across retries. - */ - replaceGraph: (graphUri: string, quads: Quad[]) => Promise; - }; + snapshotMaterializer?: SharedMemorySnapshotMaterializer; publicSnapshotStore?: WorkspacePublicSnapshotStore; getRegisteredSubGraphNames?: (contextGraphId: string) => Promise; getExcludedSubGraphNames?: (contextGraphId: string) => Promise; @@ -356,19 +319,35 @@ export async function runSharedMemorySync(context: SharedMemorySyncContext): Pro // would be overwrite-with-older, byte-for-byte the regression // this path once shipped (peer at 76 quads clobbered to 27). // Unparseable versions count as newer: when we cannot reason - // about ordering we must not destroy. - const stored = await snapshotMaterializer.readStoredAssertionVersion(descriptor); - if (stored !== null && storedVersionOutranksDescriptor(stored, descriptor.assertionVersion)) { + // about ordering we must not destroy. Nor may we "repair" the + // head rows here — gossip owns a newer head and its + // delete-then-insert already wrote it unambiguously. + const storedHead = await snapshotMaterializer.readStoredHead(descriptor); + if ( + storedHead.version !== null + && storedVersionOutranksDescriptor(storedHead.version, descriptor.assertionVersion) + ) { materializedKeys.add(graphKey); logDebug(ctx, `SWM sync for "${pid}": snapshot ${snapshotRef} superseded by ` - + `stored version ${stored} (descriptor ${descriptor.assertionVersion}); skipping`); + + `stored version ${storedHead.version} (descriptor ${descriptor.assertionVersion}); skipping`); return; } - // (b) Exact content already present (same version, complete - // graph). Equal-version-but-short means a partial write or the - // pre-fix marker-only state — those must be REPAIRED, which is - // why this check is content-count-based, not marker-based. + // (b) Exact content already present. Count AND digest: a + // marker-only or short graph is the pre-fix broken state and + // must be REPAIRED; an equal-count graph with a different + // digest is an OLDER version of the same size and must be + // replaced, not skipped. if (await snapshotMaterializer.isGraphAssetMaterialized(descriptor)) { + if (storedHead.needsRepair) { + // Content is already this descriptor's, but the head + // subject still carries union-insert residue (several + // version/operation rows) — e.g. a prior round replaced + // the graph and then failed before finishing the metadata + // swap. Collapse the head now; the fresh verified meta for + // this descriptor is re-inserted after the snapshot phase, + // exactly like the replace path below. + await snapshotMaterializer.replaceHeadMetadata(pid, descriptor); + } materializedKeys.add(graphKey); return; } @@ -379,6 +358,14 @@ export async function runSharedMemorySync(context: SharedMemorySyncContext): Pro }); await ensureContextGraph(pid); await snapshotMaterializer.replaceGraph(asset.assertionGraph, [...asset.quads]); + // Graph first, THEN the head swap — a crash between the two + // leaves content newer than the head, which the next round + // repairs (digest matches → head collapsed above). The swap + // deletes the old head + its operations so the append-style + // `storeInsert(processed.verifiedMeta)` below lands on a clean + // subject instead of stacking a second version onto it + // (LIMIT-1 head readers would otherwise see an arbitrary mix). + await snapshotMaterializer.replaceHeadMetadata(pid, descriptor); materializedKeys.add(graphKey); materializedGraphs += 1; materializedQuads += asset.quads.length; diff --git a/packages/agent/src/sync/requester/swm-snapshot-materializer.ts b/packages/agent/src/sync/requester/swm-snapshot-materializer.ts new file mode 100644 index 0000000000..08874827fd --- /dev/null +++ b/packages/agent/src/sync/requester/swm-snapshot-materializer.ts @@ -0,0 +1,247 @@ +/** + * The store adapter behind public SWM catch-up snapshot materialization. + * + * This module OWNS the persistence policy for turning a verified graph-scoped + * snapshot into durable store state: what "already materialized" means, how + * the stored head version is read, how stale head metadata is replaced, and + * which lock serializes it all against live gossip. `runSharedMemorySync` + * consumes it as one cohesive dependency (see `SharedMemorySnapshotMaterializer`); + * `dkg-agent-lifecycle` is reduced to wiring agent-owned resources into + * `createSharedMemorySnapshotMaterializer`. + * + * Every SPARQL read/write here is scoped to an exact per-KA IRI (the head + * subject, the operation subject, or the KA's own assertion graph), so each + * query is bounded by one KA's size — never by context-graph or fleet growth. + */ +import { assertSafeIri } from '@origintrail-official/dkg-core'; +import { + swmKaWriteLockKey, + withKeyedLocks, + workspacePublicQuadsDigest, +} from '@origintrail-official/dkg-publisher'; +import type { Quad, TripleStore } from '@origintrail-official/dkg-storage'; +import type { GraphScopedSwmRecoveryDescriptor } from '../graph-scoped-swm-recovery.js'; + +const DKG = 'http://dkg.io/ontology/'; + +/** What the local store currently records on one KA's SWM head subject. */ +export interface StoredWorkspaceHeadState { + /** + * The NEWEST assertionVersion on the head subject (MAX, not an arbitrary + * binding), or null when no version/operation row pair exists. MAX matters + * because the append-style meta insert can leave several version rows on one + * subject; reading an arbitrary one would let an older row veto — or worse, + * authorize — a replace decision. + */ + version: string | null; + /** + * True when the head subject carries rows from more than one assertion + * version or share operation — the residue a union-style meta insert leaves + * behind. Such a head is ambiguous for LIMIT-1 readers + * (`resolveKnowledgeAssetWorkspaceHead`) and must be collapsed back to + * exactly one version's rows via `replaceHeadMetadata`. + */ + needsRepair: boolean; +} + +/** + * Everything `runSharedMemorySync` needs to MATERIALIZE verified public SWM + * snapshots into the triple store, as ONE cohesive dependency. + * + * Why one object: these capabilities are only meaningful together. An earlier + * revision exposed them as independent optionals, which allowed a silent + * half-configured mode — a caller supplying the snapshot store but not the + * guard would compile fine and quietly skip materialization. Absent entirely + * => materialization is skipped (never half-applied). + */ +export interface SharedMemorySnapshotMaterializer { + /** + * Serialize against the live-gossip write path for one KA. MUST take the + * same key on the same lock map SharedMemoryHandler uses (the agent owns + * the map; the key comes from the shared `swmKaWriteLockKey`). Without it + * this interleaving destroys data: catch-up observes the graph absent → + * gossip commits a richer version → catch-up replaces it with the older + * snapshot. + */ + withKaWriteLock( + contextGraphId: string, + subGraphName: string | undefined, + kaUal: string, + fn: () => Promise, + ): Promise; + /** + * Read the KA's stored head state (newest version + ambiguity flag). Read + * INSIDE the lock: a lock prevents interleaving but not overwriting-with- + * older, and gossip may have committed a newer version while catch-up + * waited. + */ + readStoredHead(descriptor: GraphScopedSwmRecoveryDescriptor): Promise; + /** + * True only when the KA's assertion graph CONTENT equals the descriptor's: + * same quad count AND same public-quads digest. A marker-only predicate + * re-reports the pre-fix broken state (head metadata written, graph never + * written) as materialized; a count-only predicate cannot tell two versions + * of equal size apart and would skip a verified newer snapshot. + */ + isGraphAssetMaterialized(descriptor: GraphScopedSwmRecoveryDescriptor): Promise; + /** + * Atomic whole-graph replace. Replace, not insert: a KA graph is + * all-or-nothing and digest-verified; union-insert risks partial or + * duplicated state across retries. + */ + replaceGraph(graphUri: string, quads: Quad[]): Promise; + /** + * Delete the KA's head rows and every share-operation subject its head + * references (including the descriptor's own, which the caller re-inserts + * from fresh verified metadata). This is the catch-up lane's counterpart of + * gossip's delete-then-insert (`storeKnowledgeAssetWorkspaceHead`) and the + * private recovery lane's `replaceMetaForGraphAssets`: without it the + * append-style meta insert stacks old and new head rows on one subject and + * the durable current head becomes ambiguous. + */ + replaceHeadMetadata( + contextGraphId: string, + descriptor: GraphScopedSwmRecoveryDescriptor, + ): Promise; +} + +/** + * Build the production materializer over the agent's own store, lock map and + * list-cache invalidation hook. + */ +export function createSharedMemorySnapshotMaterializer(deps: { + store: TripleStore; + /** + * The SAME map injected into SharedMemoryHandler — sharing the map (and the + * key helper) is what closes the check-then-replace race with gossip. + */ + writeLocks: Map>; + invalidateListContextGraphsCache: () => void; +}): SharedMemorySnapshotMaterializer { + return { + withKaWriteLock: (contextGraphId, subGraphName, kaUal, fn) => + withKeyedLocks(deps.writeLocks, [swmKaWriteLockKey(contextGraphId, subGraphName, kaUal)], fn), + + readStoredHead: async (descriptor) => { + // Aggregates over ONE bound subject in the KA's meta graph: bounded by + // that subject's row count. COUNT(DISTINCT …) doubles as the duplicate + // detector — more than one version or operation value on the head is the + // union-insert residue that must be repaired. + const result = await deps.store.query( + `SELECT (MAX(?v) AS ?maxVersion) (COUNT(DISTINCT ?v) AS ?versions) ` + + `(COUNT(DISTINCT ?op) AS ?operations) WHERE { ` + + `GRAPH <${assertSafeIri(descriptor.metaGraph)}> { ` + + `<${assertSafeIri(descriptor.headSubject)}> ` + + `<${DKG}assertionVersion> ?v ; ` + + `<${DKG}shareOperationId> ?op } }`, + { priority: 'background', source: 'agent.sharedMemorySync.snapshotMaterializer.readStoredHead' }, + ); + if (result.type !== 'bindings' || result.bindings.length === 0) { + return { version: null, needsRepair: false }; + } + const row = result.bindings[0]; + const version = literalValue(row?.['maxVersion']); + const versions = parseCount(row?.['versions']); + const operations = parseCount(row?.['operations']); + return { + version: version && version.length > 0 ? version : null, + needsRepair: versions > 1 || operations > 1, + }; + }, + + isGraphAssetMaterialized: async (descriptor) => { + const expected = descriptor.publicQuadsCount; + if (!Number.isSafeInteger(expected) || expected <= 0) return false; + // 1) Count gate: exact-IRI scope, so bounded — and cheap enough to run + // every round. Strictly equal: a short graph is a partial write and must + // be replaced, not treated as already materialized. + const countResult = await deps.store.query( + `SELECT (COUNT(*) AS ?n) WHERE { GRAPH <${assertSafeIri(descriptor.assertionGraph)}> { ?s ?p ?o } }`, + { priority: 'background', source: 'agent.sharedMemorySync.snapshotMaterializer.countGraph' }, + ); + if (countResult.type !== 'bindings' || countResult.bindings.length === 0) return false; + const present = Number.parseInt(literalValue(countResult.bindings[0]?.['n']) ?? '0', 10); + if (!Number.isFinite(present) || present !== expected) return false; + // 2) Content binding: a matching count does not prove the stored graph + // is THIS descriptor's content — all versions of a graph-scoped KA share + // one graph URI, so an older version of equal size would otherwise pass + // and the verified newer snapshot would be skipped forever. Reading the + // graph back only runs when the count already matches, so it is bounded + // by exactly the snapshot size we would otherwise write; the digest is + // the same store-roundtrip check `resolveWorkspaceOperation` relies on. + const contentResult = await deps.store.query( + `CONSTRUCT { ?s ?p ?o } WHERE { GRAPH <${assertSafeIri(descriptor.assertionGraph)}> { ?s ?p ?o } }`, + { priority: 'background', source: 'agent.sharedMemorySync.snapshotMaterializer.readGraph' }, + ); + if (contentResult.type !== 'quads') return false; + const stored = contentResult.quads.map((quad) => ({ ...quad, graph: '' })); + return workspacePublicQuadsDigest(stored) === descriptor.publicQuadsDigest; + }, + + replaceGraph: async (graphUri, quads) => { + // Deliberately NOT routed through the sync lane's guarded union insert: + // a KA graph is all-or-nothing and digest-verified, so it must land via + // the atomic replace or not at all. + if (typeof deps.store.replaceGraph !== 'function') { + throw new Error('triple store does not support atomic graph replace'); + } + await deps.store.replaceGraph(graphUri, quads, { + priority: 'background', + source: 'agent.sharedMemorySync.materializeSnapshot', + }); + deps.invalidateListContextGraphsCache(); + }, + + replaceHeadMetadata: async (contextGraphId, descriptor) => { + // Collect every share operation the head currently references — via the + // BOUND head subject, then per-candidate bound-subject ASKs, so no query + // scans the meta bucket. The kaUal guard mirrors the recovery lane's + // `replaceMetaForGraphAssets` join: a head row pointing at another KA's + // operation must not delete that KA's metadata. + const shareIds = await deps.store.query( + `SELECT DISTINCT ?op WHERE { GRAPH <${assertSafeIri(descriptor.metaGraph)}> { ` + + `<${assertSafeIri(descriptor.headSubject)}> <${DKG}shareOperationId> ?op } }`, + { priority: 'background', source: 'agent.sharedMemorySync.snapshotMaterializer.findOperations' }, + ); + const operationSubjects = new Set([descriptor.operationSubject]); + if (shareIds.type === 'bindings') { + for (const row of shareIds.bindings) { + const shareId = literalValue(row?.['op']); + if (!shareId) continue; + const candidate = `urn:dkg:share:${contextGraphId}:${shareId}`; + if (operationSubjects.has(candidate)) continue; + const ownedByThisKa = await deps.store.query( + `ASK { GRAPH <${assertSafeIri(descriptor.metaGraph)}> { ` + + `<${assertSafeIri(candidate)}> <${DKG}kaUal> <${assertSafeIri(descriptor.kaUal)}> } }`, + { priority: 'background', source: 'agent.sharedMemorySync.snapshotMaterializer.checkOperation' }, + ); + if (ownedByThisKa.type === 'boolean' && ownedByThisKa.value) { + operationSubjects.add(candidate); + } + } + } + await deps.store.deleteByPattern( + { graph: descriptor.metaGraph, subject: descriptor.headSubject }, + { priority: 'background', source: 'agent.sharedMemorySync.snapshotMaterializer.deleteHead' }, + ); + for (const operationSubject of operationSubjects) { + await deps.store.deleteByPattern( + { graph: descriptor.metaGraph, subject: operationSubject }, + { priority: 'background', source: 'agent.sharedMemorySync.snapshotMaterializer.deleteOperation' }, + ); + } + }, + }; +} + +/** Strip the lexical value out of an N-Triples-style literal binding. */ +function literalValue(binding: string | undefined): string | undefined { + if (binding === undefined) return undefined; + const literal = /^"([^"]*)"/.exec(binding); + return literal ? literal[1] : binding; +} + +function parseCount(binding: string | undefined): number { + const parsed = Number.parseInt(literalValue(binding) ?? '0', 10); + return Number.isFinite(parsed) ? parsed : 0; +} diff --git a/packages/agent/src/sync/responder/graph-plan.ts b/packages/agent/src/sync/responder/graph-plan.ts index 30df4110d1..c5b20d7598 100644 --- a/packages/agent/src/sync/responder/graph-plan.ts +++ b/packages/agent/src/sync/responder/graph-plan.ts @@ -23,7 +23,12 @@ import { SYNC_RESPONDER_SNAPSHOT_BUILD_MAX_ROWS, SYNC_RESPONDER_SNAPSHOT_BUILD_PAGE_ROWS, } from './snapshot-cache.js'; -import { SyncRowSnapshotBudgetError } from './snapshot-budget.js'; +import { bytesToHex } from '@noble/hashes/utils.js'; +import { sha256 } from '@noble/hashes/sha2.js'; +import { + SyncRowSnapshotBudgetError, + type SyncResponderSnapshotBudget, +} from './snapshot-budget.js'; import { estimateStringRowHeapBytes } from '../memory-telemetry.js'; import type { ChangelogSyncResponse, ChangelogDeltaRecord } from '../changelog/wire.js'; import { durableMetaDelegationSubjectAdmissionExpression } from './durable-meta-admission.js'; @@ -100,6 +105,78 @@ export interface FreshSwmDataGraphPlanMemo { ): Promise; } +interface FreshSwmMetaSubjectEntry { + readonly subject: string; + readonly rowCount: number; +} + +interface FreshSwmMetaGraphPlanEntry { + readonly graph: string; + /** TTL-admitted subjects, compareCodePoint-sorted; row counts are exact at plan build. */ + readonly subjects: readonly FreshSwmMetaSubjectEntry[]; + readonly rowCount: number; +} + +/** + * Session plan for the TTL-filtered SWM meta phase (#1847). Holds only + * graph/subject/count scalars — never payload rows — and is bounded at + * CONSTRUCTION: discovery queries carry LIMIT/response-byte caps, the admitted + * subject cardinality is capped by {@link FRESH_SWM_META_PLAN_MAX_SUBJECTS}, + * and the retained scalar estimate is capped by the fixed snapshot build byte + * cap, so plan building can never materialize an unbounded store result. The + * retained estimate is additionally charged to the process-wide responder + * snapshot budget by the memo (see createResponderFreshSwmMetaPlanMemo). + * + * The plan is IMMUTABLE once built — every reader treats it as a frozen + * pagination description. The mutable per-session content-digest bindings that + * used to live on subject entries are held in a sidecar keyed by plan instance + * (see {@link sessionDigestBindingsFor}), so nothing that "reads a plan" can + * change it. + */ +interface FreshSwmMetaPlan { + readonly entries: readonly FreshSwmMetaGraphPlanEntry[]; + readonly totalRows: number; + /** Estimated retained heap bytes of the plan's subject/count scalars. */ + readonly bytesEstimate: number; +} + +/** + * Sidecar for the mutable per-session digest state of a TTL meta plan (#1868 + * review): content bindings for whole subject row-groups, established on a + * subject's FIRST window read of the session and verified on every REREAD. Row + * counts alone pass on same-count replacements, and a reread sliced at the + * plan's prefix sums could then combine rows of two different versions of one + * subject across response pages; the digest makes any content or ordering + * change of an already-served subject fail the session instead (the requester + * restarts with a fresh plan). A subject read exactly once needs no binding: + * its row-group is served whole from a single query. + * + * Keyed WEAKLY by plan object identity, which is exactly the binding's + * intended lifetime: the memoized plan IS the session (offset>0 requires the + * existing plan; refresh/rebuild produces a NEW plan object and therefore a + * fresh, empty binding map), and evicting or expiring the plan releases its + * digests with it. Map keys are `graph U+0000 subject` (NUL cannot appear + * in an IRI, so the composite key cannot collide). + */ +const freshSwmMetaSessionDigests = new WeakMap>(); + +function sessionDigestBindingsFor(plan: FreshSwmMetaPlan): Map { + let bindings = freshSwmMetaSessionDigests.get(plan); + if (!bindings) { + bindings = new Map(); + freshSwmMetaSessionDigests.set(plan, bindings); + } + return bindings; +} + +export interface FreshSwmMetaPlanMemo { + get( + key: string, + load: () => Promise, + options?: { refresh?: boolean; requireExisting?: boolean; signal?: AbortSignal }, + ): Promise; +} + interface ExactGraphPagePlanEntry { graph: string; rowCount: number; @@ -251,11 +328,66 @@ export function createResponderFreshSwmDataGraphPlanMemo( ttlMs = 10 * 60_000, maxEntries = 32, ): FreshSwmDataGraphPlanMemo { - const cached = new Map(); - const inflight = new Map>(); + return createSessionPlanMemo(ttlMs, maxEntries); +} + +/** + * Session-scoped plan cache for the TTL-filtered SWM META phase (#1847). Same + * lifetime/refresh contract as {@link createResponderFreshSwmDataGraphPlanMemo}: + * touched on every page, offset>0 requires the existing plan so a rebuilt plan + * against a moved TTL cutoff can never make a numeric offset skip or duplicate. + * + * When a responder snapshot budget is supplied, every retained plan's scalar + * estimate is charged to the GLOBAL budget as a control-plane entry: peers + * cannot stack up to maxEntries uncharged plans, admission under global memory + * pressure fails as the quiet retryable limit, and an idle plan is LRU-evicted + * exactly like a retained row snapshot (the session then expires and the + * requester restarts it). + */ +export function createResponderFreshSwmMetaPlanMemo( + ttlMs = 10 * 60_000, + maxEntries = 32, + budget?: SyncResponderSnapshotBudget, +): FreshSwmMetaPlanMemo { + return createSessionPlanMemo( + ttlMs, + maxEntries, + budget && { + budget, + phase: 'shared_memory', + bytesEstimate: (plan) => plan.bytesEstimate, + }, + ); +} + +interface SessionPlanBudgetAccounting { + budget: SyncResponderSnapshotBudget; + phase: 'shared_memory' | 'durable_meta' | 'durable_data'; + bytesEstimate: (value: T) => number; +} + +function createSessionPlanMemo( + ttlMs: number, + maxEntries: number, + accounting?: SessionPlanBudgetAccounting, +): { + get( + key: string, + load: () => Promise, + options?: { refresh?: boolean; requireExisting?: boolean; signal?: AbortSignal }, + ): Promise; +} { + const cached = new Map(); + const inflight = new Map>(); + const deleteEntry = (key: string, reason: 'expired' | 'released' | 'replaced') => { + const entry = cached.get(key); + if (!entry) return; + cached.delete(key); + if (entry.budgetEntryId) accounting?.budget.remove(entry.budgetEntryId, reason); + }; const prune = (now = Date.now()) => { for (const [key, entry] of cached) { - if (now - entry.cachedAt >= ttlMs) cached.delete(key); + if (now - entry.cachedAt >= ttlMs) deleteEntry(key, 'expired'); } }; return { @@ -268,14 +400,44 @@ export function createResponderFreshSwmDataGraphPlanMemo( const existing = cached.get(key); if (!options?.refresh && existing) { cached.delete(key); - cached.set(key, { value: existing.value, cachedAt: now }); + cached.set(key, { ...existing, cachedAt: now }); + if (existing.budgetEntryId) { + // Refresh the global-budget LRU position, then stay evictable: an + // entry pinned forever would let idle plans exempt themselves from + // memory-pressure eviction. + accounting?.budget.touch(existing.budgetEntryId); + accounting?.budget.release(existing.budgetEntryId); + } return existing.value; } if (options?.requireExisting) return null; - if (!existing && cached.size >= maxEntries) cached.delete(cached.keys().next().value!); + if (!existing && cached.size >= maxEntries) { + deleteEntry(cached.keys().next().value!, 'released'); + } const pendingLoad = load() .then((value) => { - cached.set(key, { value, cachedAt: Date.now() }); + const replaced = cached.get(key); + let budgetEntryId: symbol | undefined; + if (accounting) { + budgetEntryId = Symbol(key); + // Throws the typed global budget error when the process-wide + // responder budget cannot admit the plan; the failed refresh leaves + // any previously-admitted plan in place (memo entry untouched). + accounting.budget.admit({ + id: budgetEntryId, + key, + phase: accounting.phase, + rows: 0, + bytesEstimate: accounting.bytesEstimate(value), + controlPlane: true, + replaceId: replaced?.budgetEntryId, + onEvict: () => { + if (cached.get(key)?.budgetEntryId === budgetEntryId) cached.delete(key); + }, + }); + accounting.budget.release(budgetEntryId); + } + cached.set(key, { value, cachedAt: Date.now(), budgetEntryId }); return value; }) .finally(() => inflight.delete(key)); @@ -296,38 +458,7 @@ export function createResponderExactGraphPagePlanMemo( ttlMs = 10 * 60_000, maxEntries = 32, ): ExactGraphPagePlanMemo { - const cached = new Map(); - const inflight = new Map>(); - const prune = (now = Date.now()) => { - for (const [key, entry] of cached) { - if (now - entry.cachedAt >= ttlMs) cached.delete(key); - } - }; - return { - async get(key, load, options) { - throwIfAborted(options?.signal); - const now = Date.now(); - prune(now); - const pending = inflight.get(key); - if (pending) return raceAgainstAbort(pending, options?.signal); - const existing = cached.get(key); - if (!options?.refresh && existing) { - cached.delete(key); - cached.set(key, { value: existing.value, cachedAt: now }); - return existing.value; - } - if (options?.requireExisting) return null; - if (!existing && cached.size >= maxEntries) cached.delete(cached.keys().next().value!); - const pendingLoad = load() - .then((value) => { - cached.set(key, { value, cachedAt: Date.now() }); - return value; - }) - .finally(() => inflight.delete(key)); - inflight.set(key, pendingLoad); - return raceAgainstAbort(pendingLoad, options?.signal); - }, - }; + return createSessionPlanMemo(ttlMs, maxEntries); } function createSubGraphNameMemo( @@ -466,6 +597,7 @@ export async function readSwmMetaPage(params: { rowListCacheKey?: string; refreshRowList?: boolean; refreshGeneration?: string; + freshMetaPlanMemo?: FreshSwmMetaPlanMemo; }): Promise { const graphs = swmGraphsForRegisteredSubGraphs(params.contextGraphId, params.registeredSubGraphNames, true); const graphSet = new Set(params.graphList); @@ -479,32 +611,99 @@ export async function readSwmMetaPage(params: { expiredMessage: 'Shared-memory meta sync session snapshot expired before page completion', } : undefined; - return readResponderRowsPage( - cache, - (offset, limit, signal) => readSwmMetaRowsPage( + + if (params.cutoffIso == null) { + // Legacy unfiltered sessions: unchanged bounded raw-graph snapshot with the + // existing store-paged compatibility fallback. + return readResponderRowsPage( + cache, + (offset, limit, signal) => readSwmMetaRowsPage( + params.store, + candidateGraphs, + offset, + limit, + signal, + ), + params.offset, + params.limit, + params.signal, + cache + ? { + loadSnapshot: () => readBoundedSwmMetaSnapshot( + params.store, + candidateGraphs, + cache, + ), + } + : undefined, + ); + } + + // #1847: the TTL-filtered lane. Two invariants shape it: + // + // 1. The old bounded snapshot loaded the RAW meta graph and applied the + // row/byte budget BEFORE the TTL filter, so a long-lived CG whose `_meta` + // crossed 64,000 raw rows was refused even when its fresh subset was a + // few hundred rows — and with the fallback gated off for TTL sessions the + // refusal was permanent (10/15 mainnet cores, fifa-world-cup-2026). + // 2. The old TTL fallback query (DISTINCT + UNION join + global + // `ORDER BY ?g ?s ?p ?o` + growing OFFSET over a mutable graph family) + // was gated off DELIBERATELY: it can pin cores and gigabytes on large + // stores (#1597 class). Re-enabling the flag alone would trade a bounded + // refusal for a store-melting query; that query is deleted, not revived. + // + // The fix mirrors buildFreshSwmDataGraphPlan: tiny discovery queries find the + // TTL-admitted subjects (small results, no payload sort), the session plan + // caches only graph/subject/count scalars, the snapshot materializes only the + // ADMITTED rows (so the budget now binds on what is actually served), and an + // intrinsically-oversized fresh set degrades to bounded whole-subject window + // pages from the same plan instead of failing permanently. + const cutoffIso = params.cutoffIso; + const budgetKey = cache?.key ?? `swm-meta:${params.contextGraphId}`; + const getPlan = createSessionPlanGetter( + params.freshMetaPlanMemo, + params.rowListCacheKey, + params.refreshRowList === true, + (signal) => buildFreshSwmMetaPlan( params.store, candidateGraphs, - params.cutoffIso, + cutoffIso, + budgetKey, + signal, + ), + 'Shared-memory meta sync session graph plan expired before page completion', + ); + const loadStoreBoundedPage: StorePageLoader = async (offset, limit, signal) => + readFreshSwmMetaRowsPageFromPlan( + params.store, + await getPlan(offset, signal), offset, limit, + budgetKey, signal, - ), + ); + return readResponderRowsPage( + cache, + loadStoreBoundedPage, params.offset, params.limit, params.signal, - cache - ? () => readBoundedSwmMetaSnapshot( - params.store, - candidateGraphs, - params.cutoffIso, - cache, - ) - : undefined, - // The TTL-filtered SPARQL fallback joins and globally sorts a mutable meta - // graph. On large stores that query is worse than a bounded refusal: it can - // consume multiple cores and gigabytes until the HTTP timeout. Unfiltered - // legacy sessions retain the existing store-paged compatibility path. - params.cutoffIso == null, + { + loadSnapshot: cache + ? async () => readBoundedFreshSwmMetaSnapshot( + params.store, + await getPlan(0, undefined), + cutoffIso, + cache, + ) + : undefined, + // The per-snapshot budget fallback MUST stay enabled here (#1847): it + // degrades to the bounded plan-paged reader above, never to the deleted + // global-sort query. This policy used to be a positional boolean, and + // passing `params.cutoffIso == null` in that position is the exact defect + // that made every 64,000-row `_meta` CG permanently unsyncable on mainnet. + fallbackOnPerSnapshotBudget: true, + }, ); } @@ -645,12 +844,14 @@ export async function readDurableMetaPage(params: { params.limit, params.signal, cache - ? () => readBoundedDurableMetaSnapshot( - params.store, - params.contextGraphId, - params.registeredSubGraphNames, - cache, - ) + ? { + loadSnapshot: () => readBoundedDurableMetaSnapshot( + params.store, + params.contextGraphId, + params.registeredSubGraphNames, + cache, + ), + } : undefined, ); } @@ -697,11 +898,7 @@ async function readGraphScopedVmManifest( key: `durable-v2-manifest:${contextGraphId}`, reason: 'snapshot_bytes', rows: 0, - bytesEstimate: typeof error.actualBytes === 'bigint' - ? Number(error.actualBytes > BigInt(Number.MAX_SAFE_INTEGER) - ? BigInt(Number.MAX_SAFE_INTEGER) - : error.actualBytes) - : error.actualBytes, + bytesEstimate: storeResponseActualBytes(error), limit: SYNC_RESPONDER_SNAPSHOT_BUILD_MAX_BYTES_ESTIMATE, }); } @@ -1326,36 +1523,18 @@ async function readPagedRowsFromExactGraphPlanLoader( planMemo: ExactGraphPagePlanMemo | undefined, loadExactGraphPlan: (signal?: AbortSignal) => Promise, ): Promise { - // A small snapshot is first assembled into the row cache. If that build - // crosses its cap, readResponderRowsPage immediately retries page zero via - // the store-bounded path. Consume the explicit session refresh only once so - // that fallback reuses the exact graph/count plan instead of counting every - // graph twice. - let planRefreshPending = cache?.refresh === true; const rowSnapshotLimits = cache?.memo.snapshotLoadLimits ?? { maxRows: SYNC_RESPONDER_SNAPSHOT_BUILD_MAX_ROWS, maxBytesEstimate: SYNC_RESPONDER_SNAPSHOT_BUILD_MAX_BYTES_ESTIMATE, pageRows: SYNC_RESPONDER_SNAPSHOT_BUILD_PAGE_ROWS, }; - const getPlan = async ( - pageOffset: number, - pageSignal: AbortSignal | undefined, - ): Promise => { - const loadPlan = () => loadExactGraphPlan(pageSignal); - const refreshPlan = pageOffset === 0 && planRefreshPending; - if (refreshPlan) planRefreshPending = false; - const plan = planMemo && cache - ? await planMemo.get(cache.key, loadPlan, { - refresh: refreshPlan, - requireExisting: pageOffset > 0, - signal: pageSignal, - }) - : await loadPlan(); - if (!plan) { - throw new Error('Sync session exact-graph plan expired before page completion'); - } - return plan; - }; + const getPlan = createSessionPlanGetter( + planMemo, + cache?.key, + cache?.refresh === true, + (planSignal) => loadExactGraphPlan(planSignal), + 'Sync session exact-graph plan expired before page completion', + ); const loadPage: StorePageLoader = async (pageOffset, pageLimit, pageSignal) => { const plan = await getPlan(pageOffset, pageSignal); return readRowsPageFromExactGraphPlan( @@ -1377,12 +1556,14 @@ async function readPagedRowsFromExactGraphPlanLoader( limit, signal, cache - ? async () => readExactGraphPlanSnapshot( - store, - await getPlan(0, undefined), - cache, - rowSnapshotLimits, - ) + ? { + loadSnapshot: async () => readExactGraphPlanSnapshot( + store, + await getPlan(0, undefined), + cache, + rowSnapshotLimits, + ), + } : undefined, ); } @@ -1427,6 +1608,15 @@ function snapshotResponseByteLimit(maxBytesEstimate: number): number { ); } +/** Clamp a store response-cap overshoot (possibly bigint) into a safe number. */ +function storeResponseActualBytes(error: StoreResponseTooLargeError): number { + return typeof error.actualBytes === 'bigint' + ? Number(error.actualBytes > BigInt(Number.MAX_SAFE_INTEGER) + ? BigInt(Number.MAX_SAFE_INTEGER) + : error.actualBytes) + : error.actualBytes; +} + function snapshotBudgetError(params: { key: string; reason: 'snapshot_rows' | 'snapshot_bytes'; @@ -1644,6 +1834,61 @@ function isPerSnapshotBudgetError(error: unknown): error is SyncRowSnapshotBudge (error.reason === 'snapshot_rows' || error.reason === 'snapshot_bytes'); } +/** + * Session-plan getter shared by the plan-backed lanes (exact-graph and TTL SWM + * meta), owning the one lifecycle both must agree on: + * + * - the explicit session refresh is consumed exactly ONCE, so when a snapshot + * build crosses its budget, the immediate page-zero fallback reuses the + * just-built plan instead of rebuilding (and re-counting) it against a + * moving store; + * - offset>0 REQUIRES the existing plan — silently rebuilding against moved + * data would make the numeric offset skip or duplicate rows; + * - memo expiry becomes the lane's session-expired error. + * + * The SWM data lane intentionally does not use this helper: it has no snapshot + * lane, so a single plan access per page means per-call refresh semantics are + * equivalent and simpler there. + */ +/** + * Sessionless callers (no `syncSessionId` => no memo cache key) rebuild the + * plan on EVERY page: per-page discovery + chunked GROUP BY cost (bounded, and + * still far cheaper than the deleted global-sort query), no `requireExisting` + * protection, and a fresh digest sidecar per plan object. Offset>0 pages + * against a mutating store can therefore skip or duplicate rows for such + * requesters — sessionless TTL paging is BEST-EFFORT, and the per-subject + * digest guard does NOT cover it. This matches the exposure of the old OFFSET + * lane (not a regression); requester-side verification still gates admission. + */ +function createSessionPlanGetter( + memo: { + get( + key: string, + load: () => Promise, + options?: { refresh?: boolean; requireExisting?: boolean; signal?: AbortSignal }, + ): Promise; + } | undefined, + cacheKey: string | undefined, + initialRefreshPending: boolean, + loadPlan: (signal?: AbortSignal) => Promise, + expiredMessage: string, +): (pageOffset: number, pageSignal: AbortSignal | undefined) => Promise { + let planRefreshPending = initialRefreshPending; + return async (pageOffset, pageSignal) => { + const refreshPlan = pageOffset === 0 && planRefreshPending; + if (refreshPlan) planRefreshPending = false; + const plan = memo && cacheKey + ? await memo.get(cacheKey, () => loadPlan(pageSignal), { + refresh: refreshPlan, + requireExisting: pageOffset > 0, + signal: pageSignal, + }) + : await loadPlan(pageSignal); + if (!plan) throw new Error(expiredMessage); + return plan; + }; +} + /** * Serve one responder page, owning the single budget-fallback policy for every * memoized phase. It tries the stable-snapshot cache first, but an @@ -1729,15 +1974,33 @@ async function loadStorePagedSnapshot( } } +/** + * Optional behavior of {@link readResponderRowsPage}, named instead of + * positional: a bare boolean in this helper's signature is how the #1847 + * production defect happened (`params.cutoffIso == null` read as the fallback + * policy), so call sites must now spell the policy out. + */ +interface ResponderRowsPageOptions { + /** Session snapshot loader; omitted phases build via the store-paged loader. */ + loadSnapshot?: () => Promise; + /** + * Whether a PER-snapshot rows/bytes budget refusal degrades to the + * store-bounded page loader for this and every later page of the session + * (defaults to true; global budget pressure always propagates). + */ + fallbackOnPerSnapshotBudget?: boolean; +} + async function readResponderRowsPage( cache: RowListCache | undefined, loadStoreBoundedPage: StorePageLoader, offset: number, limit: number, signal?: AbortSignal, - loadSnapshot?: () => Promise, - fallbackOnPerSnapshotBudget = true, + options?: ResponderRowsPageOptions, ): Promise { + const loadSnapshot = options?.loadSnapshot; + const fallbackOnPerSnapshotBudget = options?.fallbackOnPerSnapshotBudget ?? true; const safeOffset = Math.max(0, Math.floor(offset)); const safeLimit = Math.max(0, Math.floor(limit)); if (safeLimit === 0) return []; @@ -2004,38 +2267,15 @@ async function readRowsAcrossGraphsExcludingSubjectPrefix( .sort(compareRows); } -async function readSwmMetaRows( - store: TripleStore, - swmMetaGraphs: readonly string[], - cutoffIso: string | null, - signal?: AbortSignal, -): Promise { - const swmMetaValues = graphValues(swmMetaGraphs); - if (!swmMetaValues) return []; - const res = await store.query(` - SELECT DISTINCT ?g ?s ?p ?o WHERE { - VALUES ?g { ${swmMetaValues} } - GRAPH ?g { - ?s ?p ?o . - ${cutoffIso - ? ` - ?s <${DKG_PUBLISHED_AT}> ?ts . - FILTER(?ts >= ${sparqlString(cutoffIso)}^^)` - : ''} - } - } - `, syncResponderStoreOptions(signal, 'sync.responder.readSwmMetaRows')); - if (res.type !== 'bindings') return []; - return res.bindings - .map((row) => ({ s: row['s'], p: row['p'], o: row['o'], g: row['g'] })) - .filter((row) => row.s && row.p && row.o && row.g) - .sort(compareRows); -} - +/** + * Legacy (cutoffIso == null) bounded snapshot: reads the raw candidate meta + * graphs under the per-snapshot budget. TTL-filtered sessions use + * {@link readBoundedFreshSwmMetaSnapshot}, whose budget binds on the admitted + * fresh subset instead of the raw graph size (#1847). + */ async function readBoundedSwmMetaSnapshot( store: TripleStore, swmMetaGraphs: readonly string[], - cutoffIso: string | null, cache: RowListCache, ): Promise { const limits = cache.memo.snapshotLoadLimits ?? { @@ -2072,16 +2312,11 @@ async function readBoundedSwmMetaSnapshot( }); } catch (error) { if (!(error instanceof StoreResponseTooLargeError)) throw error; - const actualBytes = typeof error.actualBytes === 'bigint' - ? Number(error.actualBytes > BigInt(Number.MAX_SAFE_INTEGER) - ? BigInt(Number.MAX_SAFE_INTEGER) - : error.actualBytes) - : error.actualBytes; throw snapshotBudgetError({ key: cache.key, reason: 'snapshot_bytes', rows: rows.length, - bytesEstimate: bytesEstimate + actualBytes, + bytesEstimate: bytesEstimate + storeResponseActualBytes(error), limit: limits.maxBytesEstimate, }); } @@ -2116,7 +2351,7 @@ async function readBoundedSwmMetaSnapshot( } } - return filterSwmMetaSnapshotRows(rows, cutoffIso); + return filterSwmMetaSnapshotRows(rows, null); } function filterSwmMetaSnapshotRows( @@ -2177,10 +2412,17 @@ function filterSwmMetaSnapshotRows( return rows.filter((row) => admitted.has(row.s)).sort(compareRows); } +/** + * Legacy UNFILTERED store-paged compatibility path (cutoffIso == null sessions + * only). The former TTL variant of this query — DISTINCT + a six-predicate + * UNION join + global `ORDER BY ?g ?s ?p ?o` re-evaluated with a growing + * OFFSET per page over a mutable graph family — was the #1847 store-melter and + * is deliberately DELETED, not gated: TTL-filtered sessions page from the + * session plan via {@link readFreshSwmMetaRowsPageFromPlan} instead. + */ async function readSwmMetaRowsPage( store: TripleStore, swmMetaGraphs: readonly string[], - cutoffIso: string | null, offset: number, limit: number, signal?: AbortSignal, @@ -2189,40 +2431,15 @@ async function readSwmMetaRowsPage( const safeLimit = Math.max(0, Math.floor(limit)); if (safeLimit === 0) return []; const swmMetaValues = graphValues(swmMetaGraphs); - const swmMetaClause = swmMetaValues - ? ` - VALUES ?g { ${swmMetaValues} } - GRAPH ?g { - ?s ?p ?o . - ${cutoffIso - ? ` - { - ?s <${DKG_PUBLISHED_AT}> ?ts . - } UNION { - # Graph-scoped SWM heads are current-state pointers and therefore - # intentionally have no independent publishedAt row. Bind them to - # the timestamped WorkspaceOperation they select so TTL recovery - # receives the head plus its immutable commitment atomically. - ?s <${DKG_CONTENT_SCOPE_VERSION}> ${GRAPH_KA_CONTENT_SCOPE_VERSION} ; - <${DKG_KA_UAL}> ?headUal ; - <${DKG_ASSERTION_VERSION}> ?headVersion ; - <${DKG_SHARE_OPERATION_ID}> ?shareId . - ?headOperation <${DKG_ONTOLOGY.RDF_TYPE}> <${DKG_WORKSPACE_OPERATION}> ; - <${DKG_CONTENT_SCOPE_VERSION}> ${GRAPH_KA_CONTENT_SCOPE_VERSION} ; - <${DKG_KA_UAL}> ?headUal ; - <${DKG_ASSERTION_VERSION}> ?headVersion ; - <${DKG_SHARE_OPERATION_ID}> ?shareId ; - <${DKG_PUBLISHED_AT}> ?ts . - } - FILTER(?ts >= ${sparqlString(cutoffIso)}^^)` - : ''} - } - ` - : ''; - if (!swmMetaClause) return []; + if (!swmMetaValues) return []; + // sparql-scan-allow: R2 -- ?g is bound by a finite VALUES list of pre-admitted SWM meta graph IRIs + // sparql-scan-allow: R3 -- pre-existing legacy (cutoff-less) compatibility lane, unchanged behavior; TTL sessions page from the session plan instead (#1847) const res = await store.query(` SELECT DISTINCT ?g ?s ?p ?o WHERE { - ${swmMetaClause} + VALUES ?g { ${swmMetaValues} } + GRAPH ?g { + ?s ?p ?o . + } } ORDER BY ?g ?s ?p ?o OFFSET ${safeOffset} @@ -2234,6 +2451,480 @@ async function readSwmMetaRowsPage( .filter((row) => row.s && row.p && row.o && row.g); } +const FRESH_SWM_META_PLAN_SUBJECT_CHUNK = 100; + +/** + * Hard cardinality cap for a TTL meta session plan's admitted subjects, across + * all candidate graphs of the phase. The discovery queries are LIMIT-bounded to + * this cap (plus one sentinel row), so plan construction can never materialize + * an unbounded subject set no matter how large the fresh window is: a fresh set + * beyond the cap is a typed bounded refusal, never an unbounded control-plane + * plan. Sizing: every admitted subject serves at least one row, so this cap + * alone admits sessions far past the point where they run plan-paged, while + * the retained plan stays a few megabytes at worst (also capped by the fixed + * build byte estimate below, which bounds pathological IRI lengths). + */ +export const FRESH_SWM_META_PLAN_MAX_SUBJECTS = 32_000; + +/** + * Discover the TTL-admitted subjects of one SWM meta graph with two + * small-result queries (no payload rows, no sort, no OFFSET), each bounded by + * construction: LIMIT (remaining subject allowance + 1 sentinel) and the fixed + * snapshot-build response byte cap. Crossing either bound is a typed + * per-snapshot budget refusal — the plan lane's one remaining bounded refusal + * besides the single-oversized-subject case. + * + * 1. subjects carrying their own fresh `publishedAt` — the + * {@link readFreshSwmRoots} shape, an indexed predicate probe whose result + * is the fresh subset, not the graph; + * 2. graph-scoped SWM heads. Heads are current-state pointers and + * intentionally have no independent publishedAt row; they are admitted via + * the timestamped WorkspaceOperation they select (same six-predicate join + * the TTL lane has always used), so TTL recovery receives the head plus its + * immutable commitment atomically. + * + * SWM meta subjects are IRIs by contract (workspace writers skolemize blank + * nodes before storage); non-IRI subjects cannot appear in a VALUES clause and + * are skipped. + */ +async function readFreshSwmMetaSubjects( + store: TripleStore, + graph: string, + cutoffIso: string, + maxSubjects: number, + budgetKey: string, + signal?: AbortSignal, +): Promise> { + const cutoffFilter = + `FILTER(?ts >= ${sparqlString(cutoffIso)}^^)`; + const discoveryLimit = Math.max(1, Math.floor(maxSubjects)) + 1; + const subjects = new Set(); + const runDiscovery = async (sparql: string, operation: string): Promise => { + let res; + try { + res = await store.query(sparql, { + ...syncResponderStoreOptions(signal, operation), + maxResponseBytes: snapshotResponseByteLimit( + SYNC_RESPONDER_SNAPSHOT_BUILD_MAX_BYTES_ESTIMATE, + ), + }); + } catch (error) { + if (!(error instanceof StoreResponseTooLargeError)) throw error; + throw snapshotBudgetError({ + key: budgetKey, + reason: 'snapshot_bytes', + rows: subjects.size, + bytesEstimate: storeResponseActualBytes(error), + limit: SYNC_RESPONDER_SNAPSHOT_BUILD_MAX_BYTES_ESTIMATE, + }); + } + if (res.type !== 'bindings') return; + for (const row of res.bindings) { + const subject = row['s']; + if (subject && isIriTerm(subject)) subjects.add(subject); + } + if (subjects.size > maxSubjects) { + throw snapshotBudgetError({ + key: budgetKey, + reason: 'snapshot_rows', + rows: subjects.size, + bytesEstimate: 0, + limit: FRESH_SWM_META_PLAN_MAX_SUBJECTS, + }); + } + }; + await runDiscovery(` + SELECT DISTINCT ?s WHERE { + GRAPH <${assertSafeIri(graph)}> { + ?s <${DKG_PUBLISHED_AT}> ?ts . + ${cutoffFilter} + } + } + LIMIT ${discoveryLimit} + `, 'sync.responder.readFreshSwmMetaSubjects'); + await runDiscovery(` + SELECT DISTINCT ?s WHERE { + GRAPH <${assertSafeIri(graph)}> { + ?s <${DKG_CONTENT_SCOPE_VERSION}> ${GRAPH_KA_CONTENT_SCOPE_VERSION} ; + <${DKG_KA_UAL}> ?headUal ; + <${DKG_ASSERTION_VERSION}> ?headVersion ; + <${DKG_SHARE_OPERATION_ID}> ?shareId . + ?headOperation <${DKG_ONTOLOGY.RDF_TYPE}> <${DKG_WORKSPACE_OPERATION}> ; + <${DKG_CONTENT_SCOPE_VERSION}> ${GRAPH_KA_CONTENT_SCOPE_VERSION} ; + <${DKG_KA_UAL}> ?headUal ; + <${DKG_ASSERTION_VERSION}> ?headVersion ; + <${DKG_SHARE_OPERATION_ID}> ?shareId ; + <${DKG_PUBLISHED_AT}> ?ts . + ${cutoffFilter} + } + } + LIMIT ${discoveryLimit} + `, 'sync.responder.readFreshSwmMetaHeadSubjects'); + return subjects; +} + +function subjectValues(subjects: readonly string[]): string { + return subjects.map((subject) => `<${assertSafeIri(subject)}>`).join(' '); +} + +async function countFreshSwmMetaSubjectRows( + store: TripleStore, + graph: string, + subjects: readonly string[], + budgetKey: string, + signal?: AbortSignal, +): Promise { + const countsBySubject = new Map(); + for (const chunk of chunkValues(subjects, FRESH_SWM_META_PLAN_SUBJECT_CHUNK)) { + let res; + try { + res = await store.query(` + SELECT ?s (COUNT(*) AS ?count) WHERE { + VALUES ?s { ${subjectValues(chunk)} } + GRAPH <${assertSafeIri(graph)}> { ?s ?p ?o } + } + GROUP BY ?s + `, { + ...syncResponderStoreOptions(signal, 'sync.responder.countFreshSwmMetaSubjectRows'), + maxResponseBytes: snapshotResponseByteLimit( + SYNC_RESPONDER_SNAPSHOT_BUILD_MAX_BYTES_ESTIMATE, + ), + }); + } catch (error) { + if (!(error instanceof StoreResponseTooLargeError)) throw error; + throw snapshotBudgetError({ + key: budgetKey, + reason: 'snapshot_bytes', + rows: chunk.length, + bytesEstimate: storeResponseActualBytes(error), + limit: SYNC_RESPONDER_SNAPSHOT_BUILD_MAX_BYTES_ESTIMATE, + }); + } + if (res.type !== 'bindings') continue; + for (const row of res.bindings) { + const subject = row['s']; + if (subject) countsBySubject.set(subject, parseSparqlInteger(row['count'])); + } + } + return subjects + .map((subject) => ({ subject, rowCount: countsBySubject.get(subject) ?? 0 })) + .filter((entry) => entry.rowCount > 0); +} + +/** + * Build the tiny, stable pagination plan for a TTL-filtered SWM meta phase. + * Only graph/subject/count scalars are computed and cached; the payload rows + * stay in the store until a page (or the bounded snapshot) addresses its own + * subject window. Subjects are compareCodePoint-sorted so the plan's prefix + * sums agree with the compareRows order used when window rows are sorted + * in-process — no store-side ORDER BY or OFFSET is ever needed. + * + * The plan itself is bounded by construction: subject cardinality by + * {@link FRESH_SWM_META_PLAN_MAX_SUBJECTS} (enforced inside the LIMIT-bounded + * discovery), and the retained scalar estimate by the FIXED snapshot build + * byte cap — deliberately the constant, not the test/operator-shrinkable + * session budget, so shrinking the session budget forces plan-paged mode + * without ever refusing the plan that paged mode needs (#1847 class). + */ +async function buildFreshSwmMetaPlan( + store: TripleStore, + swmMetaGraphs: readonly string[], + cutoffIso: string, + budgetKey: string, + signal?: AbortSignal, +): Promise { + const entries: FreshSwmMetaGraphPlanEntry[] = []; + let subjectAllowance = FRESH_SWM_META_PLAN_MAX_SUBJECTS; + let bytesEstimate = 0; + for (const graph of dedupeStrings(swmMetaGraphs).sort(compareCodePoint)) { + throwIfAborted(signal); + const admitted = await readFreshSwmMetaSubjects( + store, + graph, + cutoffIso, + subjectAllowance, + budgetKey, + signal, + ); + if (admitted.size === 0) continue; + subjectAllowance -= admitted.size; + const subjects = await countFreshSwmMetaSubjectRows( + store, + graph, + [...admitted].sort(compareCodePoint), + budgetKey, + signal, + ); + if (subjects.length === 0) continue; + for (const entry of subjects) { + bytesEstimate += estimateStringRowHeapBytes(entry.subject, '', '', graph); + } + if (bytesEstimate > SYNC_RESPONDER_SNAPSHOT_BUILD_MAX_BYTES_ESTIMATE) { + throw snapshotBudgetError({ + key: budgetKey, + reason: 'snapshot_bytes', + rows: subjects.length, + bytesEstimate, + limit: SYNC_RESPONDER_SNAPSHOT_BUILD_MAX_BYTES_ESTIMATE, + }); + } + entries.push({ + graph, + subjects, + rowCount: subjects.reduce((sum, entry) => sum + entry.rowCount, 0), + }); + } + return { + entries, + totalRows: entries.reduce((sum, entry) => sum + entry.rowCount, 0), + bytesEstimate, + }; +} + +/** Order/content digest of one subject's compareRows-sorted row-group. */ +function digestSubjectRows(rows: readonly SyncRow[]): string { + const hash = sha256.create(); + const encoder = new TextEncoder(); + for (const row of rows) { + // Length-prefixed fields: literals may contain any delimiter character. + hash.update(encoder.encode(`${row.p.length}:${row.p}${row.o.length}:${row.o}`)); + } + return bytesToHex(hash.digest()); +} + +/** + * Read ALL rows of a whole-subject window in bounded VALUES chunks, verifying + * each subject's row-group against the plan two ways. The plan's prefix sums + * are the pagination cursor, so a mutated subject must fail the session (the + * requester restarts with a fresh plan) rather than silently skip, duplicate, + * or tear rows; a seal/head subject is always read atomically within one chunk + * query, so its row-group can never be torn by a chunk boundary. + * + * 1. PER-SUBJECT row count vs the plan. An aggregate count would pass when + * two subjects in one window mutate by compensating amounts, and the + * prefix-sum slice would then duplicate or skip rows at the page seam. + * 2. Content digest, bound on the subject's first window read of this + * session and verified on every reread. Counts alone pass on a same-count + * replacement, and a reread sliced at the stale prefix sums could combine + * rows of two versions of one subject across response pages. A subject + * that is never reread needs no digest: its group is served whole from a + * single query, so a same-count change before its only read serves the + * NEWER coherent group (bounded freshness skew, like any keyset pager), + * never a hybrid. + * + * `digestBindings` is the plan's session sidecar (see + * {@link sessionDigestBindingsFor}); this reader is the only writer to it, and + * the plan itself is never mutated. + */ +async function readFreshSwmMetaSubjectWindowRows( + store: TripleStore, + graph: string, + subjects: readonly FreshSwmMetaSubjectEntry[], + digestBindings: Map, + signal?: AbortSignal, +): Promise { + const rows: SyncRow[] = []; + for (const chunk of chunkValues(subjects, FRESH_SWM_META_PLAN_SUBJECT_CHUNK)) { + const res = await store.query(` + SELECT ?s ?p ?o WHERE { + VALUES ?s { ${subjectValues(chunk.map((entry) => entry.subject))} } + GRAPH <${assertSafeIri(graph)}> { ?s ?p ?o } + } + `, { + ...syncResponderStoreOptions(signal, 'sync.responder.readFreshSwmMetaSubjectRows'), + maxResponseBytes: snapshotResponseByteLimit( + SYNC_RESPONDER_SNAPSHOT_BUILD_MAX_BYTES_ESTIMATE, + ), + }); + const rowsBySubject = new Map(); + if (res.type === 'bindings') { + for (const row of res.bindings) { + const s = row['s']; + const p = row['p']; + const o = row['o']; + if (!s || !p || !o) continue; + const bucket = rowsBySubject.get(s) ?? []; + bucket.push({ s, p, o, g: graph }); + rowsBySubject.set(s, bucket); + } + } + for (const entry of chunk) { + const subjectRows = (rowsBySubject.get(entry.subject) ?? []).sort(compareRows); + if (subjectRows.length !== entry.rowCount) { + throw new Error( + `Shared-memory meta sync plan changed while reading ${graph}: ` + + `expected ${entry.rowCount} rows for subject ${entry.subject}, found ${subjectRows.length}`, + ); + } + const digest = digestSubjectRows(subjectRows); + const digestKey = `${graph}\u0000${entry.subject}`; + const boundDigest = digestBindings.get(digestKey); + if (boundDigest === undefined) { + digestBindings.set(digestKey, digest); + } else if (boundDigest !== digest) { + throw new Error( + `Shared-memory meta sync plan changed while reading ${graph}: ` + + `subject ${entry.subject} content changed within an active session`, + ); + } + for (const row of subjectRows) rows.push(row); + } + } + return rows.sort(compareRows); +} + +/** + * Store-bounded page reader for an intrinsically-oversized TTL-filtered SWM + * meta phase. Pages advance across the plan's prefix sums; each page reads + * whole subjects (bounded by the page limit plus at most one subject's rows) + * and slices precisely. A single SUBJECT larger than the HARD build cap + * (SYNC_RESPONDER_SNAPSHOT_BUILD_MAX_ROWS — deliberately the fixed constant, + * not the test/operator-shrinkable session budget, so a shrunken budget forces + * paged mode without refusing ordinary multi-row subjects) is the one + * remaining bounded refusal: it cannot be served as a coherent row-group + * within any budget, and unlike the graph-level cap it can only be a + * pathological writer, never organic operation history. + */ +async function readFreshSwmMetaRowsPageFromPlan( + store: TripleStore, + plan: FreshSwmMetaPlan, + offset: number, + limit: number, + budgetKey: string, + signal?: AbortSignal, +): Promise { + let skip = Math.max(0, Math.floor(offset)); + let remaining = Math.max(0, Math.floor(limit)); + if (remaining === 0 || skip >= plan.totalRows) return []; + const digestBindings = sessionDigestBindingsFor(plan); + const rows: SyncRow[] = []; + for (const entry of plan.entries) { + if (skip >= entry.rowCount) { + skip -= entry.rowCount; + continue; + } + // Select the whole-subject window covering [skip, skip + remaining). + const window: FreshSwmMetaSubjectEntry[] = []; + let windowStart = 0; + let windowRows = 0; + let beforeWindow = 0; + for (const subject of entry.subjects) { + if (beforeWindow + subject.rowCount <= skip && window.length === 0) { + beforeWindow += subject.rowCount; + continue; + } + if (window.length === 0) windowStart = beforeWindow; + if (subject.rowCount > SYNC_RESPONDER_SNAPSHOT_BUILD_MAX_ROWS) { + throw snapshotBudgetError({ + key: budgetKey, + reason: 'snapshot_rows', + rows: subject.rowCount, + bytesEstimate: 0, + limit: SYNC_RESPONDER_SNAPSHOT_BUILD_MAX_ROWS, + }); + } + window.push(subject); + windowRows += subject.rowCount; + if (windowStart + windowRows >= skip + remaining) break; + } + if (window.length === 0) { + skip = 0; + continue; + } + const windowRowsRead = await readFreshSwmMetaSubjectWindowRows( + store, + entry.graph, + window, + digestBindings, + signal, + ); + const page = windowRowsRead.slice(skip - windowStart, skip - windowStart + remaining); + for (const row of page) rows.push(row); + remaining -= page.length; + if (remaining <= 0) break; + skip = 0; + } + return rows; +} + +/** + * TTL-filtered bounded snapshot (#1847). The per-snapshot budget binds on the + * plan's ADMITTED row total — what will actually be served — instead of the + * raw graph size, so a 64,000-row `_meta` history with a small fresh subset + * takes the ordinary memoized-snapshot path. The collected rows then pass + * through {@link filterSwmMetaSnapshotRows}, the canonical in-process + * admission filter, exactly as the raw-graph snapshot always has; the plan's + * SPARQL discovery is a candidate superset of that filter for the canonical + * typed-literal meta writes, so both stages agree in production. + */ +async function readBoundedFreshSwmMetaSnapshot( + store: TripleStore, + plan: FreshSwmMetaPlan, + cutoffIso: string, + cache: RowListCache, +): Promise { + const limits = cache.memo.snapshotLoadLimits ?? { + maxRows: SYNC_RESPONDER_SNAPSHOT_BUILD_MAX_ROWS, + maxBytesEstimate: SYNC_RESPONDER_SNAPSHOT_BUILD_MAX_BYTES_ESTIMATE, + pageRows: SYNC_RESPONDER_SNAPSHOT_BUILD_PAGE_ROWS, + }; + if (plan.totalRows > limits.maxRows) { + throw snapshotBudgetError({ + key: cache.key, + reason: 'snapshot_rows', + rows: plan.totalRows, + bytesEstimate: 0, + limit: limits.maxRows, + }); + } + const rows: SyncRow[] = []; + let bytesEstimate = 0; + const digestBindings = sessionDigestBindingsFor(plan); + for (const entry of plan.entries) { + let graphRows; + try { + graphRows = await readFreshSwmMetaSubjectWindowRows( + store, + entry.graph, + entry.subjects, + digestBindings, + ); + } catch (error) { + // The store's response byte cap firing during SNAPSHOT materialization is + // a per-snapshot byte overflow in disguise: the admitted set is + // intrinsically too large to hold at once, so it must degrade to the + // plan-paged reader exactly like the in-process estimate crossing the + // budget — not escape untyped and fail a syncable phase outright. The + // plan-paged reader's own bounded window reads keep the store cap + // un-translated there, so a genuinely oversized single page still + // surfaces as a hard error rather than being masked. + if (!(error instanceof StoreResponseTooLargeError)) throw error; + throw snapshotBudgetError({ + key: cache.key, + reason: 'snapshot_bytes', + rows: rows.length, + bytesEstimate: bytesEstimate + storeResponseActualBytes(error), + limit: limits.maxBytesEstimate, + }); + } + for (const row of graphRows) { + const nextBytes = bytesEstimate + estimateStringRowHeapBytes(row.s, row.p, row.o, row.g); + if (nextBytes > limits.maxBytesEstimate) { + throw snapshotBudgetError({ + key: cache.key, + reason: 'snapshot_bytes', + rows: rows.length + 1, + bytesEstimate: nextBytes, + limit: limits.maxBytesEstimate, + }); + } + rows.push(row); + bytesEstimate = nextBytes; + } + } + return filterSwmMetaSnapshotRows(rows, cutoffIso); +} + // NOTE: keep in sync with its page-safe twin {@link readFreshSwmDataRowsPage} — // both MUST return the same SET of rows (see readDurableMetaRows note). async function readFreshSwmDataRows( diff --git a/packages/agent/src/sync/responder/snapshot-budget.ts b/packages/agent/src/sync/responder/snapshot-budget.ts index c9f21e6e58..e992ae9791 100644 --- a/packages/agent/src/sync/responder/snapshot-budget.ts +++ b/packages/agent/src/sync/responder/snapshot-budget.ts @@ -28,6 +28,15 @@ type SnapshotBudgetAdmission = Omit & { key: string; /** Existing entry replaced atomically after the new entry passes admission. */ replaceId?: symbol; + /** + * Control-plane entries (session pagination plans) are bounded at build time + * by their own FIXED construction caps, deliberately not by the + * operator/test-shrinkable per-snapshot limits: shrinking the per-snapshot + * budget is how a session is forced into plan-paged mode, and rejecting the + * plan itself there would turn that degradation into a refusal (#1847 + * class). Only the GLOBAL rows/bytes budget applies at admission. + */ + controlPlane?: boolean; }; export class SyncRowSnapshotBudgetError extends Error { @@ -137,11 +146,13 @@ export function createSyncResponderSnapshotBudget( return { admit(params) { - if (params.rows > limits.maxSnapshotRows) { - reject(params, 'snapshot_rows', limits.maxSnapshotRows); - } - if (params.bytesEstimate > limits.maxSnapshotBytesEstimate) { - reject(params, 'snapshot_bytes', limits.maxSnapshotBytesEstimate); + if (!params.controlPlane) { + if (params.rows > limits.maxSnapshotRows) { + reject(params, 'snapshot_rows', limits.maxSnapshotRows); + } + if (params.bytesEstimate > limits.maxSnapshotBytesEstimate) { + reject(params, 'snapshot_bytes', limits.maxSnapshotBytesEstimate); + } } const replaced = params.replaceId ? entries.get(params.replaceId) : undefined; diff --git a/packages/agent/src/sync/responder/sync-handler.ts b/packages/agent/src/sync/responder/sync-handler.ts index ed5b09d930..b0520f5ecf 100644 --- a/packages/agent/src/sync/responder/sync-handler.ts +++ b/packages/agent/src/sync/responder/sync-handler.ts @@ -21,6 +21,7 @@ import { createResponderGraphListMemo, createResponderExactGraphPagePlanMemo, createResponderFreshSwmDataGraphPlanMemo, + createResponderFreshSwmMetaPlanMemo, createResponderSyncRowListMemo, createResponderSubGraphRegistrationMemo, createResponderSwmAdmissionMemo, @@ -446,6 +447,14 @@ export function registerSyncHandler(params: RegisterSyncHandlerParams): void { DURABLE_DATA_SYNC_SESSION_TTL_MS, SYNC_RESPONDER_SHARED_MEMORY_SNAPSHOT_LIMIT, ); + const freshSwmMetaPlanMemo = createResponderFreshSwmMetaPlanMemo( + DURABLE_DATA_SYNC_SESSION_TTL_MS, + SYNC_RESPONDER_SHARED_MEMORY_SNAPSHOT_LIMIT, + // #1847 review: retained TTL meta session plans are control-plane state and + // must be charged to the same process-wide budget as retained snapshots — + // peers cannot stack uncharged plans, and global pressure evicts idle ones. + responderSnapshotBudget, + ); const durableDataExactGraphPlanMemo = createResponderExactGraphPagePlanMemo( DURABLE_DATA_SYNC_SESSION_TTL_MS, SYNC_RESPONDER_DURABLE_DATA_SNAPSHOT_LIMIT, @@ -672,6 +681,7 @@ export function registerSyncHandler(params: RegisterSyncHandlerParams): void { rowListCacheKey: session?.rowListCacheKey, refreshRowList: session?.refreshRowList, refreshGeneration: session?.refreshGeneration, + freshMetaPlanMemo: freshSwmMetaPlanMemo, }); const queryDurationMs = Date.now() - queryStartedAt; const serializeStartedAt = Date.now(); diff --git a/packages/agent/test/swm-public-snapshot-materialization.test.ts b/packages/agent/test/swm-public-snapshot-materialization.test.ts index 7e313a24d6..276e279440 100644 --- a/packages/agent/test/swm-public-snapshot-materialization.test.ts +++ b/packages/agent/test/swm-public-snapshot-materialization.test.ts @@ -1,6 +1,6 @@ /** * Public SWM catch-up snapshot MATERIALIZATION — the behavior that turns a - * verified, cached immutable snapshot into a stored per-KA assertion graph. + * verified immutable snapshot into a stored per-KA assertion graph. * * Drives the real `runSharedMemorySync` with crafted graph-scoped meta (the * verifier is injected, so `processSharedMemoryBatch` returns it as verified) @@ -16,9 +16,20 @@ * 4. an already-materialized asset is left alone * 5. a failed replace keeps the phase incomplete and WITHHOLDS the meta * insert, so a marker can never certify a graph that was not written + * 6. after a replace, the stale head metadata is swapped out BEFORE the + * fresh verified meta is appended (graph → head swap → meta ordering) + * 7. a KA under a REGISTERED subgraph is materialized too — dropping the + * subgraph admission pass-through must fail these tests + * 8. a node with NO cached snapshot fetches it over the network + * (phase === 'snapshot') and still materializes it * * Case 2 is deterministic without sleeps: the test holds the real lock, which * IS the pause; catch-up's own lock acquisition is the sync point. + * + * These are decision tests: the materializer is injected so each guard answer + * is scripted. The REAL store-backed materializer implementation (SPARQL + * count/digest guard, MAX head read, head-metadata swap) is covered against a + * real OxigraphStore in `swm-snapshot-materializer.test.ts`. */ import { describe, expect, it } from 'vitest'; import { @@ -40,6 +51,7 @@ import { import type { Quad } from '@origintrail-official/dkg-storage'; import type { SyncPageResult } from '../src/sync/requester/page-fetch.js'; import { runSharedMemorySync } from '../src/sync/requester/shared-memory-sync.js'; +import type { StoredWorkspaceHeadState } from '../src/sync/requester/swm-snapshot-materializer.js'; const CG = 'ws00-snapshot-materialization'; const WS = contextGraphWorkspaceGraphUri(CG); @@ -61,13 +73,16 @@ class MemorySnapshotStore implements WorkspacePublicSnapshotStore { } function page(quads: Quad[], completed = true): SyncPageResult { - return { quads, bytesReceived: 0, resumedFromOffset: 0, nextOffset: quads.length, checkpointKey: 'k', completed }; + return { quads, bytesReceived: 0, resumedFromOffset: 0, nextOffset: quads.length, checkpointKey: 'k', completed, timedOut: false }; } /** One graph-scoped KA share: payload + the meta the strict parser demands. */ -function fixture() { +function fixture(subGraphName?: string) { const scope = createGraphKnowledgeAssetScope(UAL, 1); - const assertionGraph = knowledgeAssetLayerGraphUri(CG, MemoryLayer.SharedWorkingMemory, scope); + const metaGraph = subGraphName + ? `did:dkg:context-graph:${CG}/${subGraphName}/_shared_memory_meta` + : WS_META; + const assertionGraph = knowledgeAssetLayerGraphUri(CG, MemoryLayer.SharedWorkingMemory, scope, subGraphName); const operationId = 'snapshot-materialization-op'; const operationSubject = `urn:dkg:share:${CG}:${operationId}`; const headSubject = `${UAL}#dkg-swm-head`; @@ -86,46 +101,63 @@ function fixture() { privateTripleCount: 0, publisherPeerId: 'peer-source', timestamp: new Date(0), - }, WS_META), - { subject: operationSubject, predicate: `${DKG}publicQuadsDigest`, object: `"${digest}"`, graph: WS_META }, - { subject: operationSubject, predicate: `${DKG}publicSnapshotRef`, object: `"${digest}"`, graph: WS_META }, - { subject: headSubject, predicate: `${DKG}contentScopeVersion`, object: `"${GRAPH_KA_CONTENT_SCOPE_VERSION}"^^<${XSD_INTEGER}>`, graph: WS_META }, - { subject: headSubject, predicate: `${DKG}kaUal`, object: UAL, graph: WS_META }, - { subject: headSubject, predicate: `${DKG}assertionVersion`, object: `"1"^^<${XSD_INTEGER}>`, graph: WS_META }, - { subject: headSubject, predicate: `${DKG}assertionGraph`, object: assertionGraph, graph: WS_META }, - { subject: headSubject, predicate: `${DKG}shareOperationId`, object: `"${operationId}"`, graph: WS_META }, + ...(subGraphName ? { subGraphName } : {}), + }, metaGraph), + { subject: operationSubject, predicate: `${DKG}publicQuadsDigest`, object: `"${digest}"`, graph: metaGraph }, + { subject: operationSubject, predicate: `${DKG}publicSnapshotRef`, object: `"${digest}"`, graph: metaGraph }, + { subject: headSubject, predicate: `${DKG}contentScopeVersion`, object: `"${GRAPH_KA_CONTENT_SCOPE_VERSION}"^^<${XSD_INTEGER}>`, graph: metaGraph }, + { subject: headSubject, predicate: `${DKG}kaUal`, object: UAL, graph: metaGraph }, + { subject: headSubject, predicate: `${DKG}assertionVersion`, object: `"1"^^<${XSD_INTEGER}>`, graph: metaGraph }, + { subject: headSubject, predicate: `${DKG}assertionGraph`, object: assertionGraph, graph: metaGraph }, + { subject: headSubject, predicate: `${DKG}shareOperationId`, object: `"${operationId}"`, graph: metaGraph }, ]; - return { payload, digest, meta, assertionGraph }; + return { payload, digest, meta, metaGraph, assertionGraph }; } interface HarnessOverrides { - storedVersion?: () => string | null; + storedHead?: () => StoredWorkspaceHeadState; contentPresent?: () => boolean; replaceImpl?: (graphUri: string, quads: Quad[]) => Promise; onLockRequested?: () => void; lockMap?: Map>; + subGraphName?: string; + /** Skip the snapshot-store preseed to force the network (phase='snapshot') fetch. */ + preseedSnapshot?: boolean; } function harness(overrides: HarnessOverrides = {}) { - const fx = fixture(); + const fx = fixture(overrides.subGraphName); const snapshotStore = new MemorySnapshotStore(); const events: string[] = []; const replaced: Array<{ graphUri: string; quads: Quad[] }> = []; + const headSwaps: Array<{ contextGraphId: string; headSubject: string }> = []; const inserted: Quad[][] = []; + const snapshotFetches: string[] = []; const lockMap = overrides.lockMap ?? new Map>(); const run = async () => { - // Snapshot pre-seeded: the CACHE path fires onSnapshotReady without any - // network fetch — the same shape as a node whose earlier broken runs - // already cached the blobs (`swm-public-snapshots/`) without writing them. - await snapshotStore.putSnapshot({ digest: fx.digest, quads: fx.payload }); + // Snapshot pre-seeded (default): the CACHE path fires onSnapshotReady + // without any network fetch — the same shape as a node whose earlier + // broken runs already cached the blobs (`swm-public-snapshots/`) without + // writing them. `preseedSnapshot: false` starts cold instead, so the + // snapshot must travel through the phase === 'snapshot' network fetch. + if (overrides.preseedSnapshot !== false) { + await snapshotStore.putSnapshot({ digest: fx.digest, quads: fx.payload }); + } return runSharedMemorySync({ ctx, remotePeerId: 'peer-source', contextGraphIds: [CG], createContextGraphSyncDeadline: () => Number.MAX_SAFE_INTEGER, - fetchSyncPages: async (_c, _p, _cg, _inc, phase): Promise => - phase === 'meta' ? page(fx.meta) : page([]), + fetchSyncPages: async (_c, _p, _cg, _inc, phase, _g, _dl, snapshotRef): Promise => { + if (phase === 'meta') return page(fx.meta); + if (phase === 'snapshot') { + events.push('snapshot-fetched'); + snapshotFetches.push(String(snapshotRef)); + return page(fx.payload.map((quad) => ({ ...quad }))); + } + return page([]); + }, processSharedMemoryBatch: async (wsDataQuads, wsMetaQuads) => ({ verifiedData: wsDataQuads, verifiedMeta: wsMetaQuads, @@ -135,8 +167,17 @@ function harness(overrides: HarnessOverrides = {}) { emptyResponses: 0, entityCreators: [], }), + ...(overrides.subGraphName + ? { + getRegisteredSubGraphNames: async () => [overrides.subGraphName!], + getExcludedSubGraphNames: async () => [], + } + : {}), ensureContextGraph: async () => {}, - storeInsert: async (quads) => { inserted.push(quads); }, + storeInsert: async (quads) => { + events.push('meta-inserted'); + inserted.push(quads); + }, snapshotMaterializer: { withKaWriteLock: (contextGraphId, subGraphName, kaUal, fn) => { events.push('lock-requested'); @@ -147,15 +188,19 @@ function harness(overrides: HarnessOverrides = {}) { events.push('content-checked'); return overrides.contentPresent?.() ?? false; }, - readStoredAssertionVersion: async () => { + readStoredHead: async () => { events.push('version-read'); - return overrides.storedVersion?.() ?? null; + return overrides.storedHead?.() ?? { version: null, needsRepair: false }; }, replaceGraph: async (graphUri, quads) => { events.push('replaced'); if (overrides.replaceImpl) return overrides.replaceImpl(graphUri, quads); replaced.push({ graphUri, quads }); }, + replaceHeadMetadata: async (contextGraphId, descriptor) => { + events.push('head-swapped'); + headSwaps.push({ contextGraphId, headSubject: descriptor.headSubject }); + }, }, publicSnapshotStore: snapshotStore, deleteCheckpoint: () => {}, @@ -166,7 +211,7 @@ function harness(overrides: HarnessOverrides = {}) { logDebug: () => {}, }); }; - return { fx, run, events, replaced, inserted, lockMap }; + return { fx, run, events, replaced, headSwaps, inserted, snapshotFetches, lockMap }; } describe('public SWM snapshot materialization', () => { @@ -184,6 +229,21 @@ describe('public SWM snapshot materialization', () => { expect(h.inserted.some((batch) => batch.some((q) => q.graph === WS_META))).toBe(true); }); + it('swaps the stale head metadata after the replace and BEFORE the meta append', async () => { + // The meta insert below is append/union-style: without the head swap the + // old and new version rows stack on one subject and LIMIT-1 readers + // (`resolveKnowledgeAssetWorkspaceHead`) can return either — the stale + // head bug. Ordering matters both ways: graph before swap (a crash leaves + // repairable content, never a head without content), swap before append + // (the fresh rows land on a clean subject). + const h = harness({ storedHead: () => ({ version: '1', needsRepair: false }), contentPresent: () => false }); + await h.run(); + expect(h.events.indexOf('replaced')).toBeGreaterThan(-1); + expect(h.events.indexOf('head-swapped')).toBeGreaterThan(h.events.indexOf('replaced')); + expect(h.events.indexOf('meta-inserted')).toBeGreaterThan(h.events.indexOf('head-swapped')); + expect(h.headSwaps).toEqual([{ contextGraphId: CG, headSubject: `${UAL}#dkg-swm-head` }]); + }); + it('closes the gossip race: in-lock version re-check skips a superseded snapshot', async () => { // "Gossip" = an external holder of the REAL lock, same map, same key // derivation. Holding it is the deterministic pause; no timing involved. @@ -196,7 +256,7 @@ describe('public SWM snapshot materialization', () => { const h = harness({ lockMap, - storedVersion: () => storedVersion, + storedHead: () => ({ version: storedVersion, needsRepair: false }), onLockRequested: () => sawLockRequest(), }); @@ -220,28 +280,43 @@ describe('public SWM snapshot materialization', () => { await gossipHold; // Catch-up proceeded only after gossip, saw the newer stored version, and - // never touched the graph. A skip is not a failure. + // never touched the graph or the head. A skip is not a failure. expect(h.events.indexOf('gossip-committed')).toBeGreaterThan(h.events.indexOf('lock-requested')); expect(h.events.indexOf('version-read')).toBeGreaterThan(h.events.indexOf('gossip-committed')); expect(h.events).not.toContain('replaced'); + expect(h.events).not.toContain('head-swapped'); expect(summary.failedPhases).toBe(0); }); it('heals the pre-fix broken state: marker present, graph never written', async () => { - // storedVersion equals the descriptor (the marker exists) but the content + // storedHead equals the descriptor (the marker exists) but the content // check reports absent — a marker-based guard would skip forever; the // content-based guard repairs. - const h = harness({ storedVersion: () => '1', contentPresent: () => false }); + const h = harness({ storedHead: () => ({ version: '1', needsRepair: false }), contentPresent: () => false }); await h.run(); expect(h.replaced).toHaveLength(1); expect(h.replaced[0]!.graphUri).toBe(h.fx.assertionGraph); }); it('leaves an already-materialized asset alone', async () => { - const h = harness({ storedVersion: () => '1', contentPresent: () => true }); + const h = harness({ storedHead: () => ({ version: '1', needsRepair: false }), contentPresent: () => true }); const summary = await h.run(); expect(h.events).toContain('content-checked'); expect(h.events).not.toContain('replaced'); + expect(h.events).not.toContain('head-swapped'); + expect(summary.failedPhases).toBe(0); + }); + + it('collapses union-insert residue on the skip path when the head needs repair', async () => { + // Content already matches the descriptor, but the head subject carries + // several version/operation rows (e.g. a prior round failed between the + // replace and the head swap). The skip must still swap the head, or the + // ambiguity becomes permanent — every later round skips on content. + const h = harness({ storedHead: () => ({ version: '1', needsRepair: true }), contentPresent: () => true }); + const summary = await h.run(); + expect(h.events).not.toContain('replaced'); + expect(h.events).toContain('head-swapped'); + expect(h.events.indexOf('meta-inserted')).toBeGreaterThan(h.events.indexOf('head-swapped')); expect(summary.failedPhases).toBe(0); }); @@ -254,4 +329,44 @@ describe('public SWM snapshot materialization', () => { // the asset as materialized and strand it permanently. expect(h.inserted.every((batch) => batch.every((q) => q.graph !== WS_META))).toBe(true); }); + + it('materializes a KA under a REGISTERED subgraph into its subgraph assertion graph', async () => { + // The parser rejects heads in unregistered metadata graphs; without the + // registered/excluded pass-through in the parse call, this fixture throws, + // the catch clears ALL descriptors, and the snapshot is cached but never + // written — silently, for the whole context graph. + const h = harness({ subGraphName: 'notes' }); + const summary = await h.run(); + expect(h.fx.metaGraph).toBe(`did:dkg:context-graph:${CG}/notes/_shared_memory_meta`); + expect(h.replaced).toHaveLength(1); + expect(h.replaced[0]!.graphUri).toBe(h.fx.assertionGraph); + expect(h.fx.assertionGraph).toContain('/notes/'); + expect(summary.failedPhases).toBe(0); + // The subgraph's meta landed too. + expect(h.inserted.some((batch) => batch.some((q) => q.graph === h.fx.metaGraph))).toBe(true); + }); + + it('fetches an uncached snapshot over the network and materializes it', async () => { + // Cold node: nothing in the snapshot store, so onSnapshotReady must fire + // from the phase === 'snapshot' NETWORK branch after digest verification. + const h = harness({ preseedSnapshot: false }); + const summary = await h.run(); + expect(h.snapshotFetches).toEqual([h.fx.digest]); + expect(h.events.indexOf('replaced')).toBeGreaterThan(h.events.indexOf('snapshot-fetched')); + expect(h.replaced).toHaveLength(1); + expect(h.replaced[0]!.graphUri).toBe(h.fx.assertionGraph); + expect(summary.failedPhases).toBe(0); + expect(h.inserted.some((batch) => batch.some((q) => q.graph === WS_META))).toBe(true); + }); + + it('withholds the meta insert when the replace fails AFTER a network fetch', async () => { + const h = harness({ + preseedSnapshot: false, + replaceImpl: async () => { throw new Error('store unavailable'); }, + }); + const summary = await h.run(); + expect(h.events).toContain('snapshot-fetched'); + expect(summary.failedPhases).toBe(1); + expect(h.inserted.every((batch) => batch.every((q) => q.graph !== WS_META))).toBe(true); + }); }); diff --git a/packages/agent/test/swm-snapshot-materializer.test.ts b/packages/agent/test/swm-snapshot-materializer.test.ts new file mode 100644 index 0000000000..381a1a432f --- /dev/null +++ b/packages/agent/test/swm-snapshot-materializer.test.ts @@ -0,0 +1,366 @@ +/** + * The REAL store-backed snapshot materializer, against a REAL OxigraphStore — + * no injected guard answers. This is what proves the production lifecycle + * wiring, not just `runSharedMemorySync`'s decisions around it: + * + * - `isGraphAssetMaterialized` is CONTENT-based: the pre-fix broken state + * (head marker written, assertion graph never written) reads as NOT + * materialized; a short graph reads as NOT materialized; and — the + * count-only trap — an equal-count graph holding an OLDER version's + * content reads as NOT materialized because the digest differs. + * - `readStoredHead` returns the NEWEST version (MAX) when append-style + * meta inserts left several version rows on one head subject, and flags + * that residue for repair. + * - `replaceHeadMetadata` collapses the head to a clean subject: old head + * rows and every operation the head referenced are deleted, other + * subjects (and other KAs' operations) are untouched. + * - end-to-end: a node holding version 1 (same quad COUNT as version 2) + * catches up to version 2 — the graph is replaced, exactly one head + * version remains, and the LIMIT-1 production reader + * (`resolveKnowledgeAssetWorkspaceHead`) resolves version 2 instead of an + * ambiguous mix. A second round is a pure no-op, which also proves the + * digest survives the store round-trip (no churn). + */ +import { describe, expect, it } from 'vitest'; +import { + GRAPH_KA_CONTENT_SCOPE_VERSION, + MemoryLayer, + createGraphKnowledgeAssetScope, + contextGraphWorkspaceMetaGraphUri, + knowledgeAssetLayerGraphUri, + type OperationContext, +} from '@origintrail-official/dkg-core'; +import { + generateKnowledgeAssetShareMetadata, + resolveKnowledgeAssetWorkspaceHead, + workspacePublicQuadsDigest, + type WorkspacePublicSnapshotStore, +} from '@origintrail-official/dkg-publisher'; +import { GraphManager, OxigraphStore, type Quad, type TripleStore } from '@origintrail-official/dkg-storage'; +import { parseGraphScopedSwmRecoveryDescriptors } from '../src/sync/graph-scoped-swm-recovery.js'; +import { createSharedMemorySnapshotMaterializer } from '../src/sync/requester/swm-snapshot-materializer.js'; +import { runSharedMemorySync } from '../src/sync/requester/shared-memory-sync.js'; +import type { SyncPageResult } from '../src/sync/requester/page-fetch.js'; + +const CG = 'ws00-materializer-real-store'; +const WS_META = contextGraphWorkspaceMetaGraphUri(CG); +const DKG = 'http://dkg.io/ontology/'; +const XSD_INTEGER = 'http://www.w3.org/2001/XMLSchema#integer'; +const UAL = 'did:dkg:hardhat:31337/0xbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb/9'; +const ctx: OperationContext = { operationId: 'test', operationName: 'sync' } as never; + +class MemorySnapshotStore implements WorkspacePublicSnapshotStore { + readonly snapshots = new Map(); + async putSnapshot(input: { readonly digest: string; readonly quads: readonly Quad[] }) { + this.snapshots.set(input.digest, input.quads.map((quad) => ({ ...quad }))); + return { ref: input.digest, byteLength: 0 }; + } + async getSnapshot(ref: string): Promise { + return this.snapshots.get(ref)?.map((quad) => ({ ...quad })) ?? null; + } +} + +/** + * One complete graph-scoped share (head + operation meta, payload, digest) + * for `UAL` at `version`. v1 and v2 payloads deliberately have the SAME quad + * count with different content: only a digest-binding guard can tell them + * apart. + */ +function share(version: number, operationId: string, marker: string) { + const scope = createGraphKnowledgeAssetScope(UAL, version); + const assertionGraph = knowledgeAssetLayerGraphUri(CG, MemoryLayer.SharedWorkingMemory, scope); + const operationSubject = `urn:dkg:share:${CG}:${operationId}`; + const headSubject = `${UAL}#dkg-swm-head`; + const payload: Quad[] = [ + { subject: 'urn:snap:a', predicate: 'http://schema.org/status', object: `"${marker}"`, graph: '' }, + { subject: 'urn:snap:b', predicate: 'http://schema.org/status', object: `"${marker}"`, graph: '' }, + ]; + const digest = workspacePublicQuadsDigest(payload); + const meta: Quad[] = [ + ...generateKnowledgeAssetShareMetadata({ + shareOperationId: operationId, + contextGraphId: CG, + kaUal: UAL, + assertionVersion: version, + publicTripleCount: payload.length, + privateTripleCount: 0, + publisherPeerId: 'peer-source', + timestamp: new Date(0), + }, WS_META), + { subject: operationSubject, predicate: `${DKG}publicQuadsDigest`, object: `"${digest}"`, graph: WS_META }, + { subject: operationSubject, predicate: `${DKG}publicSnapshotRef`, object: `"${digest}"`, graph: WS_META }, + { subject: headSubject, predicate: `${DKG}contentScopeVersion`, object: `"${GRAPH_KA_CONTENT_SCOPE_VERSION}"^^<${XSD_INTEGER}>`, graph: WS_META }, + { subject: headSubject, predicate: `${DKG}kaUal`, object: UAL, graph: WS_META }, + { subject: headSubject, predicate: `${DKG}assertionVersion`, object: `"${version}"^^<${XSD_INTEGER}>`, graph: WS_META }, + { subject: headSubject, predicate: `${DKG}assertionGraph`, object: assertionGraph, graph: WS_META }, + { subject: headSubject, predicate: `${DKG}shareOperationId`, object: `"${operationId}"`, graph: WS_META }, + ]; + return { version, operationId, operationSubject, headSubject, assertionGraph, payload, digest, meta }; +} + +const v1 = share(1, 'op-v1', 'version-one'); +const v2 = share(2, 'op-v2', 'version-two'); + +function descriptorFor(fixture: typeof v1) { + const descriptors = parseGraphScopedSwmRecoveryDescriptors({ + contextGraphId: CG, + metaQuads: fixture.meta, + }); + expect(descriptors).toHaveLength(1); + return descriptors[0]!; +} + +function materializerFor(store: TripleStore) { + let invalidations = 0; + const materializer = createSharedMemorySnapshotMaterializer({ + store, + writeLocks: new Map>(), + invalidateListContextGraphsCache: () => { invalidations += 1; }, + }); + return { materializer, invalidations: () => invalidations }; +} + +function inGraph(quads: readonly Quad[], graph: string): Quad[] { + return quads.map((quad) => ({ ...quad, graph })); +} + +async function distinctObjects(store: TripleStore, graph: string, subject: string, predicate: string): Promise { + const result = await store.query( + `SELECT DISTINCT ?o WHERE { GRAPH <${graph}> { <${subject}> <${predicate}> ?o } }`, + ); + if (result.type !== 'bindings') throw new Error(`unexpected ${result.type}`); + return result.bindings.map((row) => String(row['o'])).sort(); +} + +describe('createSharedMemorySnapshotMaterializer against a real OxigraphStore', () => { + it('shares one assertion graph across versions (the premise of the digest guard)', () => { + expect(v1.assertionGraph).toBe(v2.assertionGraph); + expect(v1.payload).toHaveLength(v2.payload.length); + expect(v1.digest).not.toBe(v2.digest); + }); + + describe('isGraphAssetMaterialized', () => { + it('is false for the pre-fix broken state: marker metadata without content', async () => { + const store = new OxigraphStore(); + await store.insert([...v1.meta]); + const { materializer } = materializerFor(store); + expect(await materializer.isGraphAssetMaterialized(descriptorFor(v1))).toBe(false); + }); + + it('is false for a short (partially written) graph', async () => { + const store = new OxigraphStore(); + await store.insert(inGraph(v1.payload.slice(0, 1), v1.assertionGraph)); + const { materializer } = materializerFor(store); + expect(await materializer.isGraphAssetMaterialized(descriptorFor(v1))).toBe(false); + }); + + it('is true for the exact descriptor content (digest survives the store round-trip)', async () => { + const store = new OxigraphStore(); + await store.insert(inGraph(v1.payload, v1.assertionGraph)); + const { materializer } = materializerFor(store); + expect(await materializer.isGraphAssetMaterialized(descriptorFor(v1))).toBe(true); + }); + + it('is false when an EQUAL-COUNT graph holds a different version\'s content', async () => { + // The count-only trap: v1 and v2 have the same quad count and share the + // assertion graph URI. A count-based guard would report v2 as already + // materialized and strand the verified newer snapshot forever. + const store = new OxigraphStore(); + await store.insert(inGraph(v1.payload, v1.assertionGraph)); + const { materializer } = materializerFor(store); + expect(await materializer.isGraphAssetMaterialized(descriptorFor(v2))).toBe(false); + }); + }); + + describe('readStoredHead', () => { + it('is null/clean when no head exists', async () => { + const store = new OxigraphStore(); + const { materializer } = materializerFor(store); + expect(await materializer.readStoredHead(descriptorFor(v1))).toEqual({ version: null, needsRepair: false }); + }); + + it('reads a single-version head without flagging repair', async () => { + const store = new OxigraphStore(); + await store.insert([...v1.meta]); + const { materializer } = materializerFor(store); + expect(await materializer.readStoredHead(descriptorFor(v1))).toEqual({ version: '1', needsRepair: false }); + }); + + it('returns the NEWEST version (MAX) for union-insert residue and flags repair', async () => { + // Append-style meta inserts stacked v1 and v2 rows on one head subject. + // An arbitrary binding (or MIN) could report "1" and authorize an + // overwrite-with-older; MAX must win, and the residue must be flagged. + const store = new OxigraphStore(); + await store.insert([...v1.meta]); + await store.insert([...v2.meta]); + const { materializer } = materializerFor(store); + expect(await materializer.readStoredHead(descriptorFor(v2))).toEqual({ version: '2', needsRepair: true }); + }); + }); + + describe('replaceHeadMetadata', () => { + it('deletes the head and every referenced operation, sparing unrelated subjects', async () => { + const store = new OxigraphStore(); + await store.insert([...v1.meta]); + await store.insert([...v2.meta]); + const unrelated: Quad = { + subject: 'urn:dkg:share:other', + predicate: `${DKG}shareOperationId`, + object: '"unrelated"', + graph: WS_META, + }; + await store.insert([unrelated]); + const { materializer } = materializerFor(store); + + await materializer.replaceHeadMetadata(CG, descriptorFor(v2)); + + expect(await distinctObjects(store, WS_META, v2.headSubject, `${DKG}assertionVersion`)).toEqual([]); + expect(await distinctObjects(store, WS_META, v1.operationSubject, `${DKG}shareOperationId`)).toEqual([]); + expect(await distinctObjects(store, WS_META, v2.operationSubject, `${DKG}shareOperationId`)).toEqual([]); + expect(await distinctObjects(store, WS_META, unrelated.subject, `${DKG}shareOperationId`)).toEqual(['"unrelated"']); + }); + + it('never deletes an operation owned by ANOTHER KA, even if the head references it', async () => { + // A (corrupt) head row pointing at a foreign share operation must not + // let this KA's cleanup destroy the other KA's metadata — same kaUal + // guard the recovery lane's replaceMetaForGraphAssets applies. + const store = new OxigraphStore(); + const otherUal = 'did:dkg:hardhat:31337/0xcccccccccccccccccccccccccccccccccccccccc/3'; + const foreignOp = `urn:dkg:share:${CG}:foreign-op`; + await store.insert([...v1.meta]); + await store.insert([ + { subject: v1.headSubject, predicate: `${DKG}shareOperationId`, object: '"foreign-op"', graph: WS_META }, + { subject: foreignOp, predicate: `${DKG}shareOperationId`, object: '"foreign-op"', graph: WS_META }, + { subject: foreignOp, predicate: `${DKG}kaUal`, object: otherUal, graph: WS_META }, + ]); + const { materializer } = materializerFor(store); + + await materializer.replaceHeadMetadata(CG, descriptorFor(v1)); + + expect(await distinctObjects(store, WS_META, v1.headSubject, `${DKG}shareOperationId`)).toEqual([]); + expect(await distinctObjects(store, WS_META, v1.operationSubject, `${DKG}shareOperationId`)).toEqual([]); + expect(await distinctObjects(store, WS_META, foreignOp, `${DKG}shareOperationId`)).toEqual(['"foreign-op"']); + }); + }); + + it('replaceGraph writes atomically and invalidates the list cache', async () => { + const store = new OxigraphStore(); + const { materializer, invalidations } = materializerFor(store); + await materializer.replaceGraph(v1.assertionGraph, inGraph(v1.payload, v1.assertionGraph)); + expect(invalidations()).toBe(1); + const { materializer: checker } = materializerFor(store); + expect(await checker.isGraphAssetMaterialized(descriptorFor(v1))).toBe(true); + }); + + describe('end-to-end catch-up with the real materializer', () => { + function realHarness(store: TripleStore, served: typeof v1) { + const snapshotStore = new MemorySnapshotStore(); + const { materializer } = materializerFor(store); + let replaceCalls = 0; + const run = async () => { + await snapshotStore.putSnapshot({ digest: served.digest, quads: served.payload }); + return runSharedMemorySync({ + ctx, + remotePeerId: 'peer-source', + contextGraphIds: [CG], + createContextGraphSyncDeadline: () => Number.MAX_SAFE_INTEGER, + fetchSyncPages: async (_c, _p, _cg, _inc, phase): Promise => ({ + quads: phase === 'meta' ? [...served.meta] : [], + bytesReceived: 0, + resumedFromOffset: 0, + nextOffset: phase === 'meta' ? served.meta.length : 0, + checkpointKey: 'k', + completed: true, + timedOut: false, + }), + processSharedMemoryBatch: async (wsDataQuads, wsMetaQuads) => ({ + verifiedData: wsDataQuads, + verifiedMeta: wsMetaQuads, + totalFetchedDataQuads: wsDataQuads.length, + totalFetchedMetaQuads: wsMetaQuads.length, + droppedDataTriples: 0, + emptyResponses: 0, + entityCreators: [], + }), + ensureContextGraph: async () => {}, + storeInsert: async (quads) => { await store.insert(quads); }, + snapshotMaterializer: { + ...materializer, + replaceGraph: async (graphUri, quads) => { + replaceCalls += 1; + return materializer.replaceGraph(graphUri, quads); + }, + }, + publicSnapshotStore: snapshotStore, + deleteCheckpoint: () => {}, + setCheckpoint: () => {}, + ensureOwnedMap: () => new Map(), + logInfo: () => {}, + logWarn: () => {}, + logDebug: () => {}, + }); + }; + return { run, replaceCalls: () => replaceCalls }; + } + + it('heals the pre-fix broken state through the REAL content guard', async () => { + // Marker-only store: head + operation rows exist, the assertion graph + // was never written. The real SPARQL guard must answer "not + // materialized" so the cached snapshot is finally written. + const store = new OxigraphStore(); + await store.insert([...v1.meta]); + const h = realHarness(store, v1); + const summary = await h.run(); + expect(summary.failedPhases).toBe(0); + expect(h.replaceCalls()).toBe(1); + const { materializer } = materializerFor(store); + expect(await materializer.isGraphAssetMaterialized(descriptorFor(v1))).toBe(true); + }); + + it('replaces an equal-count older version and leaves ONE unambiguous head', async () => { + // Start as a node that fully holds version 1 — content AND metadata. + // Version 2 has the SAME quad count. The catch-up must (a) see through + // the equal count via the digest, (b) replace the graph, and (c) swap + // the head so the LIMIT-1 production reader resolves version 2 — not an + // arbitrary row from a v1+v2 union pile-up. + const store = new OxigraphStore(); + await store.insert([...v1.meta]); + await store.insert(inGraph(v1.payload, v1.assertionGraph)); + const h = realHarness(store, v2); + + const summary = await h.run(); + expect(summary.failedPhases).toBe(0); + expect(h.replaceCalls()).toBe(1); + + // Graph content is now v2's. + const { materializer } = materializerFor(store); + expect(await materializer.isGraphAssetMaterialized(descriptorFor(v2))).toBe(true); + expect(await materializer.isGraphAssetMaterialized(descriptorFor(v1))).toBe(false); + + // Exactly one head version / operation reference remains. + expect(await distinctObjects(store, WS_META, v2.headSubject, `${DKG}assertionVersion`)) + .toEqual([`"2"^^<${XSD_INTEGER}>`]); + expect(await distinctObjects(store, WS_META, v2.headSubject, `${DKG}shareOperationId`)) + .toEqual(['"op-v2"']); + // The old operation's rows are gone, the new one's are present. + expect(await distinctObjects(store, WS_META, v1.operationSubject, `${DKG}shareOperationId`)).toEqual([]); + expect(await distinctObjects(store, WS_META, v2.operationSubject, `${DKG}shareOperationId`)).toEqual(['"op-v2"']); + + // The LIMIT-1 production reader resolves version 2 unambiguously. + const head = await resolveKnowledgeAssetWorkspaceHead({ + store, + graphManager: new GraphManager(store), + contextGraphId: CG, + kaUal: UAL, + }); + expect(head?.assertionVersion).toBe('2'); + expect(head?.shareOperationId).toBe('op-v2'); + + // A second round is a pure no-op: the real digest guard skips (which + // also proves the digest survives the store round-trip — no churn). + const again = await h.run(); + expect(again.failedPhases).toBe(0); + expect(h.replaceCalls()).toBe(1); + }); + }); +}); diff --git a/packages/agent/test/sync-responder-swm-meta-ceiling.test.ts b/packages/agent/test/sync-responder-swm-meta-ceiling.test.ts new file mode 100644 index 0000000000..5f0aebf3e9 --- /dev/null +++ b/packages/agent/test/sync-responder-swm-meta-ceiling.test.ts @@ -0,0 +1,996 @@ +import { describe, it, expect, vi } from 'vitest'; +import type { OperationContext } from '@origintrail-official/dkg-core'; +import { + OxigraphStore, + StoreResponseTooLargeError, + type Quad, +} from '@origintrail-official/dkg-storage'; +import { createSyncResponderSnapshotBudget } from '../src/sync/responder/snapshot-budget.js'; +import { + createResponderFreshSwmMetaPlanMemo, + FRESH_SWM_META_PLAN_MAX_SUBJECTS, +} from '../src/sync/responder/graph-plan.js'; +import { + DKG_NS, + RDF_TYPE, + linesFromNquads, + registerTestSyncHandler, + subGraphRegistrationQuads, + workspaceOpQuads, + type CapturedSyncHandler, +} from './_helpers/sync-responder.js'; +import { MemorySyncCheckpointStore } from '../src/sync/checkpoint/state.js'; +import { fetchSyncPages } from '../src/sync/requester/page-fetch.js'; +import type { SyncRequestEnvelope } from '../src/sync/auth/request-build.js'; + +/** + * #1847 — SWM meta lane ceiling. A CG whose `_meta` crossed + * SYNC_RESPONDER_SNAPSHOT_BUILD_MAX_ROWS (64,000) raw rows became permanently + * unsyncable for TTL-filtered sessions: the bounded snapshot applied its budget + * to the RAW graph before the TTL filter, and `readSwmMetaPage` passed + * `params.cutoffIso == null` POSITIONALLY as `fallbackOnPerSnapshotBudget`, so + * the refusal had no fallback. These tests seed real >64,000-row stores and + * prove the lane now serves them completely, page by page, within the DEFAULT + * production budgets — and that the deleted global-sort TTL query never runs. + */ + +const XSD_DT = 'http://www.w3.org/2001/XMLSchema#dateTime'; +const XSD_INT = 'http://www.w3.org/2001/XMLSchema#integer'; +const TTL_MS = 60_000; + +const TINY_SNAPSHOT_BUDGET = { + maxRows: 1_000_000, + maxBytesEstimate: Number.MAX_SAFE_INTEGER, + maxSnapshotRows: 1, + maxSnapshotBytesEstimate: Number.MAX_SAFE_INTEGER, +} as const; + +function freshIso(): string { + return new Date(Date.now() - 1_000).toISOString(); +} + +function staleIso(): string { + return new Date(Date.now() - 10 * TTL_MS).toISOString(); +} + +/** Graph-scoped head + selected WorkspaceOperation (11 rows), per swm-recovery shape. */ +function graphScopedHeadQuads( + cgId: string, + metaGraph: string, + ual: string, + opId: string, + timestamp: string, +): Quad[] { + const op = `urn:dkg:share:${cgId}:${opId}`; + const head = `${ual}#dkg-swm-head`; + return [ + { graph: metaGraph, subject: op, predicate: RDF_TYPE, object: `${DKG_NS}WorkspaceOperation` }, + { graph: metaGraph, subject: op, predicate: `${DKG_NS}publishedAt`, object: `"${timestamp}"^^<${XSD_DT}>` }, + { graph: metaGraph, subject: op, predicate: `${DKG_NS}shareOperationId`, object: `"${opId}"` }, + { graph: metaGraph, subject: op, predicate: `${DKG_NS}contentScopeVersion`, object: `"2"^^<${XSD_INT}>` }, + { graph: metaGraph, subject: op, predicate: `${DKG_NS}kaUal`, object: ual }, + { graph: metaGraph, subject: op, predicate: `${DKG_NS}assertionVersion`, object: `"1"^^<${XSD_INT}>` }, + { graph: metaGraph, subject: head, predicate: `${DKG_NS}contentScopeVersion`, object: `"2"^^<${XSD_INT}>` }, + { graph: metaGraph, subject: head, predicate: `${DKG_NS}kaUal`, object: ual }, + { graph: metaGraph, subject: head, predicate: `${DKG_NS}assertionVersion`, object: `"1"^^<${XSD_INT}>` }, + { graph: metaGraph, subject: head, predicate: `${DKG_NS}shareOperationId`, object: `"${opId}"` }, + { graph: metaGraph, subject: head, predicate: `${DKG_NS}assertionGraph`, object: `${metaGraph.replace(/_meta$/, '')}/0x00000000000000000000000000000000000000ab/1` }, + ]; +} + +async function insertChunked(store: OxigraphStore, quads: Quad[]): Promise { + for (let offset = 0; offset < quads.length; offset += 8_000) { + await store.insert(quads.slice(offset, offset + 8_000)); + } +} + +async function collectAllPages( + cap: CapturedSyncHandler, + base: Omit, + pageSize: number, + maxPages = 300, +): Promise<{ lines: Set; pages: number }> { + const lines = new Set(); + let pages = 0; + for (let offset = 0, page = 0; page < maxPages; page += 1, offset += pageSize) { + const out = await cap.invoke({ ...base, offset }); + const pageLines = linesFromNquads(out); + pages += 1; + for (const line of pageLines) lines.add(line); + if (pageLines.length < pageSize) break; + } + return { lines, pages }; +} + +/** Fails the test if the deleted TTL global-sort shape — or ANY OFFSET/ORDER BY + * query over an SWM meta graph — reaches the store during a TTL session. */ +function forbidSwmMetaSortOrOffsetQueries(store: OxigraphStore) { + const originalQuery = store.query.bind(store); + let windowQueries = 0; + store.query = (async (sparql: string, options?: unknown) => { + const normalized = sparql.replace(/\s+/g, ' ').trim(); + if (normalized.includes('_shared_memory_meta')) { + expect(normalized).not.toMatch(/OFFSET \d/); + expect(normalized).not.toContain('ORDER BY'); + if (normalized.includes('VALUES ?s')) windowQueries += 1; + } + return originalQuery(sparql, options as never); + }) as OxigraphStore['query']; + return { + assertWindowQueriesObserved: () => expect(windowQueries).toBeGreaterThan(0), + }; +} + +describe('SWM meta lane above the 64,000-row snapshot ceiling (#1847)', () => { + it('serves a 64,000+-row _meta with a small fresh subset completely at DEFAULT budgets (the fifa-world-cup-2026 shape)', async () => { + const cgId = 'meta-ceiling-fifa'; + const metaGraph = `did:dkg:context-graph:${cgId}/_shared_memory_meta`; + const stale = staleIso(); + const fresh = freshIso(); + + const quads: Quad[] = []; + // 12,800 stale operations x 5 rows = 64,000 raw rows: over the build cap. + for (let index = 0; index < 12_800; index += 1) { + quads.push(...workspaceOpQuads(cgId, `stale-${index}`, `urn:stale:root:${index}`, metaGraph, stale)); + } + // The small fresh subset that TTL sessions actually need. + const freshOpIds = ['fresh-a', 'fresh-b', 'fresh-c']; + for (const opId of freshOpIds) { + quads.push(...workspaceOpQuads(cgId, opId, `urn:fresh:root:${opId}`, metaGraph, fresh)); + } + const freshUal = 'did:dkg:hardhat:31337/0x00000000000000000000000000000000000000ab/1'; + quads.push(...graphScopedHeadQuads(cgId, metaGraph, freshUal, 'fresh-head-op', fresh)); + expect(quads.length).toBeGreaterThan(64_000); + + const store = new OxigraphStore(); + const seedStartedAt = Date.now(); + await insertChunked(store, quads); + const seedDurationMs = Date.now() - seedStartedAt; + + // DEFAULT production budgets: no snapshotBudget override. + const cap = registerTestSyncHandler(store, { sharedMemoryTtlMs: TTL_MS, syncPageSize: 7 }); + const watch = forbidSwmMetaSortOrOffsetQueries(store); + + const serveStartedAt = Date.now(); + const { lines, pages } = await collectAllPages( + cap, + { contextGraphId: cgId, includeSharedMemory: true, phase: 'meta', limit: 7, syncSessionId: 'fifa-session' }, + 7, + ); + const serveDurationMs = Date.now() - serveStartedAt; + + // Every fresh row is served — the lane is no longer refused. + // 3 ops x 5 rows + head group 11 rows = 26. + expect(lines.size).toBe(26); + const joined = [...lines].join('\n'); + for (const opId of freshOpIds) { + expect(joined).toContain(`urn:dkg:share:${cgId}:${opId}`); + } + expect(joined).toContain(`${freshUal}#dkg-swm-head`); + expect(joined).toContain('assertionVersion'); + expect(joined).not.toContain('urn:stale:root'); + watch.assertWindowQueriesObserved(); + + // eslint-disable-next-line no-console + console.info( + `#1847 fifa-shape: raw=${quads.length} rows, fresh=26 rows, pages=${pages}, ` + + `seed=${seedDurationMs}ms, serve=${serveDurationMs}ms`, + ); + await store.close(); + }, 120_000); + + it('serves an INTRINSICALLY oversized fresh set (>64,000 admitted rows) via bounded plan paging at DEFAULT budgets', async () => { + const cgId = 'meta-ceiling-allfresh'; + const metaGraph = `did:dkg:context-graph:${cgId}/_shared_memory_meta`; + const fresh = freshIso(); + + const quads: Quad[] = []; + // 13,000 FRESH operations x 5 rows = 65,000 admitted rows: even the + // filtered set exceeds the per-snapshot cap, so the session must degrade + // to plan-paged serving instead of refusing. This test is the direct + // mutation-kill for the positional `params.cutoffIso == null` defect: + // reintroduce it and this session throws the per-snapshot budget error. + for (let index = 0; index < 13_000; index += 1) { + quads.push(...workspaceOpQuads(cgId, `f${index}`, `urn:fresh:root:${index}`, metaGraph, fresh)); + } + expect(quads.length).toBe(65_000); + + const store = new OxigraphStore(); + const seedStartedAt = Date.now(); + await insertChunked(store, quads); + const seedDurationMs = Date.now() - seedStartedAt; + + const cap = registerTestSyncHandler(store, { sharedMemoryTtlMs: TTL_MS, syncPageSize: 5000 }); + const watch = forbidSwmMetaSortOrOffsetQueries(store); + + const serveStartedAt = Date.now(); + const { lines, pages } = await collectAllPages( + cap, + { contextGraphId: cgId, includeSharedMemory: true, phase: 'meta', limit: 5000, syncSessionId: 'allfresh-session' }, + 5000, + ); + const serveDurationMs = Date.now() - serveStartedAt; + + // The complete oversized fresh set is served, page by page, no refusal. + expect(lines.size).toBe(65_000); + watch.assertWindowQueriesObserved(); + + // eslint-disable-next-line no-console + console.info( + `#1847 oversized-fresh: rows=65000, pages=${pages}, seed=${seedDurationMs}ms, serve=${serveDurationMs}ms`, + ); + await store.close(); + }, 120_000); + + it('plan-paged serving is set-equivalent to the snapshot lane across buckets, heads and stale exclusion', async () => { + const cgId = 'meta-ceiling-equiv'; + const cgPrefix = `did:dkg:context-graph:${cgId}`; + const rootMeta = `${cgPrefix}/_shared_memory_meta`; + const subMeta = `${cgPrefix}/subx/_shared_memory_meta`; + const fresh = freshIso(); + const stale = staleIso(); + + const quads: Quad[] = [ + ...subGraphRegistrationQuads(cgId, 'subx'), + ...workspaceOpQuads(cgId, 'root-fresh', 'urn:r:fresh', rootMeta, fresh), + ...workspaceOpQuads(cgId, 'root-stale', 'urn:r:stale', rootMeta, stale), + ...workspaceOpQuads(cgId, 'sub-fresh', 'urn:s:fresh', subMeta, fresh), + ...workspaceOpQuads(cgId, 'sub-stale', 'urn:s:stale', subMeta, stale), + ...graphScopedHeadQuads(cgId, rootMeta, 'did:dkg:hardhat:31337/0x00000000000000000000000000000000000000ab/7', 'head-fresh', fresh), + ...graphScopedHeadQuads(cgId, rootMeta, 'did:dkg:hardhat:31337/0x00000000000000000000000000000000000000ab/8', 'head-stale', stale), + ]; + + const canonicalStore = new OxigraphStore(); + await canonicalStore.insert(quads); + const pagedStore = new OxigraphStore(); + await pagedStore.insert(quads); + + const canonicalCap = registerTestSyncHandler(canonicalStore, { sharedMemoryTtlMs: TTL_MS, syncPageSize: 5000 }); + const pagedCap = registerTestSyncHandler(pagedStore, { + sharedMemoryTtlMs: TTL_MS, + syncPageSize: 3, + snapshotBudget: TINY_SNAPSHOT_BUDGET, + }); + const base = { contextGraphId: cgId, includeSharedMemory: true, phase: 'meta' as const }; + + const canonical = (await collectAllPages( + canonicalCap, { ...base, limit: 5000, syncSessionId: 'canon' }, 5000, + )).lines; + const paged = (await collectAllPages( + pagedCap, { ...base, limit: 3, syncSessionId: 'paged' }, 3, + )).lines; + + expect(paged).toEqual(canonical); + const joined = [...canonical].join('\n'); + expect(joined).toContain('urn:dkg:share:meta-ceiling-equiv:root-fresh'); + expect(joined).toContain('urn:dkg:share:meta-ceiling-equiv:sub-fresh'); + expect(joined).toContain('#dkg-swm-head'); + expect(joined).toContain('/7#dkg-swm-head'); + expect(joined).not.toContain('/8#dkg-swm-head'); + expect(joined).not.toContain('root-stale'); + expect(joined).not.toContain('sub-stale'); + // Both buckets appear in the graph position. + expect(joined).toContain(`<${rootMeta}> .`); + expect(joined).toContain(`<${subMeta}> .`); + await canonicalStore.close(); + await pagedStore.close(); + }); + + it('fails the session (not silently skips/duplicates) when an admitted subject mutates between plan pages, and a fresh session recovers', async () => { + const cgId = 'meta-ceiling-mutate'; + const metaGraph = `did:dkg:context-graph:${cgId}/_shared_memory_meta`; + const fresh = freshIso(); + const store = new OxigraphStore(); + // Deterministic subject order: op ids sort a < b < c... + const opIds = ['a', 'b', 'c', 'd', 'e', 'f']; + for (const opId of opIds) { + await store.insert(workspaceOpQuads(cgId, opId, `urn:m:${opId}`, metaGraph, fresh)); + } + + const cap = registerTestSyncHandler(store, { + sharedMemoryTtlMs: TTL_MS, + syncPageSize: 5, + snapshotBudget: TINY_SNAPSHOT_BUDGET, // forces plan-paged mode + }); + const base = { contextGraphId: cgId, includeSharedMemory: true, phase: 'meta' as const, limit: 5 }; + + const page0 = linesFromNquads(await cap.invoke({ ...base, offset: 0, syncSessionId: 'M1' })); + expect(page0).toHaveLength(5); + + // Grow a subject the NEXT page's window must read. + const grownSubject = `urn:dkg:share:${cgId}:${opIds[1]}`; + await store.insert([{ graph: metaGraph, subject: grownSubject, predicate: `${DKG_NS}note`, object: '"grown"' }]); + + await expect(cap.invoke({ ...base, offset: 5, syncSessionId: 'M1' })) + .rejects.toThrow(/Shared-memory meta sync plan changed while reading/); + + // A fresh session rebuilds the plan and serves the grown store completely. + const recovered = await collectAllPages( + cap, { ...base, syncSessionId: 'M2' }, 5, + ); + expect(recovered.lines.size).toBe(6 * 5 + 1); + expect([...recovered.lines].join('\n')).toContain('"grown"'); + await store.close(); + }); + + it('fails the session when an already-served subject is replaced with the SAME row count (content binding, not just cardinality)', async () => { + const cgId = 'meta-ceiling-samecount'; + const metaGraph = `did:dkg:context-graph:${cgId}/_shared_memory_meta`; + const fresh = freshIso(); + const store = new OxigraphStore(); + const opIds = ['a', 'b', 'c', 'd', 'e', 'f']; + for (const opId of opIds) { + await store.insert(workspaceOpQuads(cgId, opId, `urn:m:${opId}`, metaGraph, fresh)); + } + + const cap = registerTestSyncHandler(store, { + sharedMemoryTtlMs: TTL_MS, + // Page size 2 splits the 5-row subject `a` across pages, so page 1 must + // REREAD `a` and slice it at the plan's prefix sums — the exact shape + // that used to accept a same-count replacement and serve a hybrid + // row-group assembled from two versions of one subject. + syncPageSize: 2, + snapshotBudget: TINY_SNAPSHOT_BUDGET, // forces plan-paged mode + }); + const base = { contextGraphId: cgId, includeSharedMemory: true, phase: 'meta' as const, limit: 2 }; + + const page0 = linesFromNquads(await cap.invoke({ ...base, offset: 0, syncSessionId: 'SC1' })); + expect(page0).toHaveLength(2); + + // Same-count replacement of the split subject: 5 rows before, 5 rows after. + const splitSubject = `urn:dkg:share:${cgId}:a`; + await store.delete([ + { graph: metaGraph, subject: splitSubject, predicate: `${DKG_NS}rootEntity`, object: 'urn:m:a' }, + ]); + await store.insert([ + { graph: metaGraph, subject: splitSubject, predicate: `${DKG_NS}note`, object: '"swapped"' }, + ]); + + await expect(cap.invoke({ ...base, offset: 2, syncSessionId: 'SC1' })) + .rejects.toThrow(/Shared-memory meta sync plan changed while reading/); + + // A fresh session rebuilds the plan and serves the replaced content whole. + const recovered = await collectAllPages(cap, { ...base, syncSessionId: 'SC2' }, 2); + expect(recovered.lines.size).toBe(6 * 5); + const joined = [...recovered.lines].join('\n'); + expect(joined).toContain('"swapped"'); + expect(joined).not.toContain(`<${splitSubject}> <${DKG_NS}rootEntity>`); + await store.close(); + }); + + it('serves a coherent NEW row-group when a NOT-yet-read subject mutates same-count (bounded freshness skew, never a tear)', async () => { + // Guarantee boundary, made explicit per review: whole-subject row-groups + // are the consistency unit. A subject read exactly once is served whole + // from a single query, so a same-count change BEFORE its only read serves + // the newer coherent group — the bounded skew any keyset pager has. Only a + // REREAD of a split subject binds (and verifies) content. + const cgId = 'meta-ceiling-skew'; + const metaGraph = `did:dkg:context-graph:${cgId}/_shared_memory_meta`; + const fresh = freshIso(); + const store = new OxigraphStore(); + const opIds = ['a', 'b', 'c', 'd', 'e', 'f']; + for (const opId of opIds) { + await store.insert(workspaceOpQuads(cgId, opId, `urn:m:${opId}`, metaGraph, fresh)); + } + + const cap = registerTestSyncHandler(store, { + sharedMemoryTtlMs: TTL_MS, + syncPageSize: 5, // window = exactly one whole 5-row subject + snapshotBudget: TINY_SNAPSHOT_BUDGET, + }); + const base = { contextGraphId: cgId, includeSharedMemory: true, phase: 'meta' as const, limit: 5 }; + + const page0 = linesFromNquads(await cap.invoke({ ...base, offset: 0, syncSessionId: 'SK1' })); + expect(page0).toHaveLength(5); + + // Same-count mutation of subject `b`, which page 1 will read for the FIRST time. + const nextSubject = `urn:dkg:share:${cgId}:b`; + await store.delete([ + { graph: metaGraph, subject: nextSubject, predicate: `${DKG_NS}rootEntity`, object: 'urn:m:b' }, + ]); + await store.insert([ + { graph: metaGraph, subject: nextSubject, predicate: `${DKG_NS}note`, object: '"swapped-whole"' }, + ]); + + const page1 = linesFromNquads(await cap.invoke({ ...base, offset: 5, syncSessionId: 'SK1' })); + expect(page1).toHaveLength(5); + const joined = page1.join('\n'); + // The NEW group, whole: replacement present, replaced row absent — no hybrid. + expect(joined).toContain('"swapped-whole"'); + expect(joined).not.toContain(`<${nextSubject}> <${DKG_NS}rootEntity>`); + expect(page1.every((line) => line.startsWith(`<${nextSubject}>`))).toBe(true); + await store.close(); + }); + + it('fails the session on a compensating cross-subject count mutation within one window (per-subject counts, not the window aggregate)', async () => { + const cgId = 'meta-ceiling-compensate'; + const metaGraph = `did:dkg:context-graph:${cgId}/_shared_memory_meta`; + const fresh = freshIso(); + const store = new OxigraphStore(); + for (const opId of ['a', 'b', 'c', 'd']) { + await store.insert(workspaceOpQuads(cgId, opId, `urn:m:${opId}`, metaGraph, fresh)); + } + + const cap = registerTestSyncHandler(store, { + sharedMemoryTtlMs: TTL_MS, + syncPageSize: 10, + snapshotBudget: TINY_SNAPSHOT_BUDGET, + }); + const base = { contextGraphId: cgId, includeSharedMemory: true, phase: 'meta' as const, limit: 10 }; + + // Page 0 = subjects a+b whole. Page 1's window will be subjects c+d. + const page0 = linesFromNquads(await cap.invoke({ ...base, offset: 0, syncSessionId: 'CP1' })); + expect(page0).toHaveLength(10); + + // c loses a row, d gains one: the WINDOW aggregate still totals 10, but the + // plan's prefix sums for c/d are now both wrong — an aggregate-count guard + // passes and misaligns every later slice (duplicate/skip at page seams). + await store.delete([ + { graph: metaGraph, subject: `urn:dkg:share:${cgId}:c`, predicate: `${DKG_NS}rootEntity`, object: 'urn:m:c' }, + ]); + await store.insert([ + { graph: metaGraph, subject: `urn:dkg:share:${cgId}:d`, predicate: `${DKG_NS}note`, object: '"extra"' }, + ]); + + await expect(cap.invoke({ ...base, offset: 10, syncSessionId: 'CP1' })) + .rejects.toThrow(/Shared-memory meta sync plan changed while reading/); + await store.close(); + }); + + it('degrades to plan paging when the STORE response byte cap fires during snapshot materialization (#1868 review: untyped escape)', async () => { + const cgId = 'meta-ceiling-storecap'; + const metaGraph = `did:dkg:context-graph:${cgId}/_shared_memory_meta`; + const fresh = freshIso(); + const store = new OxigraphStore(); + const quads: Quad[] = []; + for (let index = 0; index < 40; index += 1) { + quads.push(...workspaceOpQuads(cgId, `op-${String(index).padStart(2, '0')}`, `urn:sc:${index}`, metaGraph, fresh)); + } + await store.insert(quads); + + // Emulate the storage layer's 32 MiB response cap: any whole-subject + // window query addressing MANY subjects at once (the snapshot + // materialization) throws StoreResponseTooLargeError, while the paged + // lane's small windows stay under the cap. Before the fix this error + // escaped untyped past the per-snapshot budget accounting and failed the + // phase outright instead of falling back. + let capThrows = 0; + const originalQuery = store.query.bind(store); + store.query = (async (sparql: string, options?: unknown) => { + const normalized = sparql.replace(/\s+/g, ' '); + if (normalized.includes('VALUES ?s') && !normalized.includes('COUNT(')) { + const subjectCount = (normalized.match(/ 10) { + capThrows += 1; + throw new StoreResponseTooLargeError(1024, 2048); + } + } + return originalQuery(sparql, options as never); + }) as OxigraphStore['query']; + + // DEFAULT budgets: the snapshot lane is attempted first and must degrade. + const cap = registerTestSyncHandler(store, { sharedMemoryTtlMs: TTL_MS, syncPageSize: 7 }); + const { lines } = await collectAllPages( + cap, + { contextGraphId: cgId, includeSharedMemory: true, phase: 'meta', limit: 7, syncSessionId: 'storecap-session' }, + 7, + ); + expect(lines.size).toBe(200); + expect(capThrows).toBeGreaterThan(0); + await store.close(); + }); + + it('degrades to plan paging when the fresh snapshot crosses only the per-snapshot BYTE estimate budget', async () => { + const cgId = 'meta-ceiling-bytebudget'; + const metaGraph = `did:dkg:context-graph:${cgId}/_shared_memory_meta`; + const fresh = freshIso(); + const store = new OxigraphStore(); + for (const opId of ['a', 'b', 'c', 'd', 'e', 'f']) { + await store.insert(workspaceOpQuads(cgId, opId, `urn:m:${opId}`, metaGraph, fresh)); + } + + const cap = registerTestSyncHandler(store, { + sharedMemoryTtlMs: TTL_MS, + syncPageSize: 4, + snapshotBudget: { + maxRows: 1_000_000, + maxBytesEstimate: Number.MAX_SAFE_INTEGER, + maxSnapshotRows: 1_000_000, + // Well below one row's ~200-byte heap estimate: the snapshot path must + // throw the per-snapshot BYTES error (row budget never binds) and the + // session must still complete through the plan-paged reader. + maxSnapshotBytesEstimate: 64, + }, + }); + const watch = forbidSwmMetaSortOrOffsetQueries(store); + const { lines } = await collectAllPages( + cap, + { contextGraphId: cgId, includeSharedMemory: true, phase: 'meta', limit: 4, syncSessionId: 'bytebudget-session' }, + 4, + ); + expect(lines.size).toBe(30); + watch.assertWindowQueriesObserved(); + await store.close(); + }); + + it('refuses a fresh subject set beyond the plan cardinality cap as a TYPED bounded refusal, via LIMIT-bounded discovery', async () => { + const cgId = 'meta-ceiling-cardinality'; + const metaGraph = `did:dkg:context-graph:${cgId}/_shared_memory_meta`; + const fresh = freshIso(); + const store = new OxigraphStore(); + // One row per subject: cap + 1 admitted subjects. The plan would retain a + // subject entry for every one of them — this is the reviewed unbounded + // control-plane growth (#1868), so it must refuse, bounded and typed, + // BEFORE materializing an unbounded discovery result. + const quads: Quad[] = []; + for (let index = 0; index <= FRESH_SWM_META_PLAN_MAX_SUBJECTS; index += 1) { + quads.push({ + graph: metaGraph, + subject: `urn:card:${String(index).padStart(6, '0')}`, + predicate: `${DKG_NS}publishedAt`, + object: `"${fresh}"^^<${XSD_DT}>`, + }); + } + await insertChunked(store, quads); + + // Bounded-by-construction: every TTL discovery query over the meta graph + // must carry the cap-derived LIMIT so the store can never stream an + // unbounded subject set into the plan builder. + let discoveryLimitQueries = 0; + const originalQuery = store.query.bind(store); + store.query = (async (sparql: string, options?: unknown) => { + const normalized = sparql.replace(/\s+/g, ' ').trim(); + if (normalized.includes('SELECT DISTINCT ?s') && normalized.includes('_shared_memory_meta')) { + expect(normalized).toMatch(/LIMIT \d+$/); + if (normalized.endsWith(`LIMIT ${FRESH_SWM_META_PLAN_MAX_SUBJECTS + 1}`)) { + discoveryLimitQueries += 1; + } + } + return originalQuery(sparql, options as never); + }) as OxigraphStore['query']; + + const cap = registerTestSyncHandler(store, { sharedMemoryTtlMs: TTL_MS, syncPageSize: 500 }); + await expect(cap.invoke({ + contextGraphId: cgId, + includeSharedMemory: true, + phase: 'meta', + offset: 0, + limit: 500, + syncSessionId: 'cardinality-session', + })).rejects.toThrow(/per-snapshot rows budget/); + expect(discoveryLimitQueries).toBeGreaterThan(0); + await store.close(); + }, 120_000); + + it('applies the plan cardinality cap in AGGREGATE across root and subgraph meta graphs, not per graph (#1868 review)', async () => { + const cgId = 'meta-ceiling-aggregate'; + const cgPrefix = `did:dkg:context-graph:${cgId}`; + const rootMeta = `${cgPrefix}/_shared_memory_meta`; + const subMeta = `${cgPrefix}/subagg/_shared_memory_meta`; + const fresh = freshIso(); + const store = new OxigraphStore(); + // Exactly the cap in the ROOT bucket plus ONE more fresh subject in a + // registered subgraph bucket. The subject allowance is cumulative across + // the phase's candidate graphs; a regression that reset it per graph would + // happily admit both buckets (retaining up to #graphs x cap plan entries) + // and serve this session — so it must fail this test, which demands the + // same typed bounded refusal as the single-graph overflow. + const quads: Quad[] = [...subGraphRegistrationQuads(cgId, 'subagg')]; + for (let index = 0; index < FRESH_SWM_META_PLAN_MAX_SUBJECTS; index += 1) { + quads.push({ + graph: rootMeta, + subject: `urn:agg:${String(index).padStart(6, '0')}`, + predicate: `${DKG_NS}publishedAt`, + object: `"${fresh}"^^<${XSD_DT}>`, + }); + } + quads.push({ + graph: subMeta, + subject: 'urn:agg:one-over-in-the-subgraph', + predicate: `${DKG_NS}publishedAt`, + object: `"${fresh}"^^<${XSD_DT}>`, + }); + await insertChunked(store, quads); + + const cap = registerTestSyncHandler(store, { sharedMemoryTtlMs: TTL_MS, syncPageSize: 500 }); + await expect(cap.invoke({ + contextGraphId: cgId, + includeSharedMemory: true, + phase: 'meta', + offset: 0, + limit: 500, + syncSessionId: 'aggregate-cap-session', + })).rejects.toThrow(/per-snapshot rows budget/); + await store.close(); + }, 120_000); + + it('keeps a bounded refusal ONLY for a single pathological subject exceeding the hard 64,000-row build cap', async () => { + const cgId = 'meta-ceiling-monster'; + const metaGraph = `did:dkg:context-graph:${cgId}/_shared_memory_meta`; + const fresh = freshIso(); + const store = new OxigraphStore(); + // ONE fresh subject carrying 64,001 rows. Whole-subject windows are the + // consistency unit of the plan lane (they are what keeps a seal/head + // row-group atomic per #1788), so this single row-group can never be + // served coherently within the hard build cap — a bounded refusal, at + // DEFAULT budgets, is the correct answer. Ordinary multi-row subjects + // under a shrunken session budget are covered by the paged tests above. + const subject = 'urn:monster'; + const monsterQuads: Quad[] = [ + { graph: metaGraph, subject, predicate: `${DKG_NS}publishedAt`, object: `"${fresh}"^^<${XSD_DT}>` }, + ]; + for (let index = 1; index <= 64_000; index += 1) { + monsterQuads.push({ + graph: metaGraph, subject, predicate: `${DKG_NS}note`, object: `"filler-${index}"`, + }); + } + await insertChunked(store, monsterQuads); + + const cap = registerTestSyncHandler(store, { sharedMemoryTtlMs: TTL_MS, syncPageSize: 500 }); + + await expect(cap.invoke({ + contextGraphId: cgId, + includeSharedMemory: true, + phase: 'meta', + offset: 0, + limit: 500, + syncSessionId: 'monster-session', + })).rejects.toThrow(/per-snapshot rows budget/); + await store.close(); + }, 120_000); + + it('legacy cutoff-less sessions keep the unfiltered store-paged compatibility fallback', async () => { + const cgId = 'meta-ceiling-legacy'; + const metaGraph = `did:dkg:context-graph:${cgId}/_shared_memory_meta`; + const store = new OxigraphStore(); + const iso = '2026-06-01T00:00:00.000Z'; + for (const opId of ['x', 'y', 'z']) { + await store.insert(workspaceOpQuads(cgId, opId, `urn:l:${opId}`, metaGraph, iso)); + } + + let legacyPagedQueries = 0; + const originalQuery = store.query.bind(store); + store.query = (async (sparql: string, options?: unknown) => { + const normalized = sparql.replace(/\s+/g, ' ').trim(); + if ( + normalized.includes('VALUES ?g') && + normalized.includes('_shared_memory_meta') && + normalized.includes('ORDER BY ?g ?s ?p ?o') && + /OFFSET \d+/.test(normalized) + ) { + // The legacy paged query must never carry the TTL join. + expect(normalized).not.toContain('publishedAt'); + expect(normalized).not.toContain('FILTER'); + legacyPagedQueries += 1; + } + return originalQuery(sparql, options as never); + }) as OxigraphStore['query']; + + // sharedMemoryTtlMs: 0 => cutoffIso == null (legacy lane), tiny budget + // forces the fallback. + const cap = registerTestSyncHandler(store, { + sharedMemoryTtlMs: 0, + syncPageSize: 4, + snapshotBudget: TINY_SNAPSHOT_BUDGET, + }); + const { lines } = await collectAllPages( + cap, + { contextGraphId: cgId, includeSharedMemory: true, phase: 'meta', limit: 4, syncSessionId: 'legacy' }, + 4, + ); + expect(lines.size).toBe(15); + expect(legacyPagedQueries).toBeGreaterThan(0); + await store.close(); + }); +}); + +describe('TTL meta session plans are charged to the responder snapshot budget (#1868 review)', () => { + const plan = (bytesEstimate: number) => ({ entries: [], totalRows: 0, bytesEstimate }); + + it('admits, LRU-evicts and rejects plans via the GLOBAL budget while exempting them from per-snapshot caps', async () => { + const budget = createSyncResponderSnapshotBudget({ + maxRows: 1_000, + maxBytesEstimate: 10_000, + // Deliberately tiny per-snapshot caps: plans are control-plane entries + // bounded by their own fixed construction caps, so per-snapshot limits + // must NOT reject them (shrinking those limits is how a session is + // forced into the plan-paged mode that NEEDS the plan). + maxSnapshotRows: 1, + maxSnapshotBytesEstimate: 1, + }); + const memo = createResponderFreshSwmMetaPlanMemo(60_000, 8, budget); + + await memo.get('k1', async () => plan(4_000)); + expect(budget.stats().snapshots).toBe(1); + expect(budget.stats().bytesEstimate).toBe(4_000); + + await memo.get('k2', async () => plan(4_000)); + expect(budget.stats().snapshots).toBe(2); + + // Global pressure: admitting k3 must evict the least-recently-used idle + // plan (k1) rather than growing past the global byte budget. + await memo.get('k3', async () => plan(4_000)); + expect(budget.stats().bytesEstimate).toBe(8_000); + expect(await memo.get('k1', async () => plan(1), { requireExisting: true })).toBeNull(); + + // A plan that cannot fit even after draining evictables is a typed + // global rejection (the requester's quiet retryable limit), never an + // uncharged retention. + await expect(memo.get('kX', async () => plan(50_000))) + .rejects.toThrow(/global estimated bytes budget/); + }); + + it('memo eviction and TTL expiry release the charged bytes', async () => { + const budget = createSyncResponderSnapshotBudget({ + maxRows: 1_000, + maxBytesEstimate: 100_000, + maxSnapshotRows: 1, + maxSnapshotBytesEstimate: 1, + }); + const memo = createResponderFreshSwmMetaPlanMemo(60_000, 2, budget); + await memo.get('a', async () => plan(1_000)); + await memo.get('b', async () => plan(1_000)); + expect(budget.stats().bytesEstimate).toBe(2_000); + // maxEntries=2: inserting c evicts the memo's oldest entry AND its charge. + await memo.get('c', async () => plan(1_000)); + expect(budget.stats().snapshots).toBe(2); + expect(budget.stats().bytesEstimate).toBe(2_000); + }); + + it('time-based TTL expiry prunes a plan AND releases its global charge, distinct from maxEntries eviction (#1868 review)', async () => { + const budget = createSyncResponderSnapshotBudget({ + maxRows: 1_000, + maxBytesEstimate: 100_000, + maxSnapshotRows: 1, + maxSnapshotBytesEstimate: 1, + }); + // maxEntries is deliberately roomy so ONLY the clock can remove entries: a + // regression in the time-based prune path ('expired') cannot hide behind + // the LRU/maxEntries eviction the previous test already proves. + const memo = createResponderFreshSwmMetaPlanMemo(60_000, 8, budget); + const nowSpy = vi.spyOn(Date, 'now'); + const epoch = 1_800_000_000_000; + try { + nowSpy.mockReturnValue(epoch); + await memo.get('a', async () => plan(1_000)); + expect(budget.stats().bytesEstimate).toBe(1_000); + + // One tick BEFORE the TTL boundary an unrelated get must NOT prune 'a'. + nowSpy.mockReturnValue(epoch + 59_999); + await memo.get('b', async () => plan(2_000)); + expect(budget.stats().snapshots).toBe(2); + expect(budget.stats().bytesEstimate).toBe(3_000); + + // AT the TTL boundary 'a' (still cached at epoch) must be pruned AND its + // global charge released; 'b' (age 1ms) must survive with its charge. + nowSpy.mockReturnValue(epoch + 60_000); + await memo.get('c', async () => plan(4_000)); + expect(budget.stats().snapshots).toBe(2); + expect(budget.stats().bytesEstimate).toBe(6_000); + expect(await memo.get('a', async () => plan(1), { requireExisting: true })).toBeNull(); + expect(await memo.get('b', async () => plan(1), { requireExisting: true })).not.toBeNull(); + } finally { + nowSpy.mockRestore(); + } + }); + + it('the sync handler wires the responder budget through to plan admission', async () => { + const cgId = 'meta-plan-budget-wire'; + const metaGraph = `did:dkg:context-graph:${cgId}/_shared_memory_meta`; + const fresh = freshIso(); + const store = new OxigraphStore(); + await store.insert(workspaceOpQuads(cgId, 'a', 'urn:w:a', metaGraph, fresh)); + + const cap = registerTestSyncHandler(store, { + sharedMemoryTtlMs: TTL_MS, + syncPageSize: 5, + snapshotBudget: { + maxRows: 1_000_000, + // Global byte budget below even one plan's scalar estimate: PLAN + // admission must fail typed through the handler (proving + // registerSyncHandler passes its budget into the meta plan memo, not + // an uncharged default). maxSnapshotRows=1 keeps the ROW snapshot on + // its memoized per-snapshot refusal so it never reaches the global + // budget itself — the plan memo is the only global-budget client here. + maxBytesEstimate: 100, + maxSnapshotRows: 1, + maxSnapshotBytesEstimate: Number.MAX_SAFE_INTEGER, + }, + }); + await expect(cap.invoke({ + contextGraphId: cgId, + includeSharedMemory: true, + phase: 'meta', + offset: 0, + limit: 5, + syncSessionId: 'plan-budget-wire', + })).rejects.toThrow(/global estimated bytes budget/); + await store.close(); + }); +}); + +describe('requester reassembly of the plan-paged SWM meta lane (#1847 x #1788)', () => { + function makeCtx(): OperationContext { + return { kind: 'system', id: 'meta-ceiling-requester', startedAt: Date.now() } as never; + } + const noop = () => {}; + + /** Minimal N-Quads line parser for the fixture vocabulary (IRIs + literals). */ + function parseNquads(text: string): Quad[] { + const quads: Quad[] = []; + for (const line of text.split('\n')) { + const trimmed = line.trim(); + if (!trimmed) continue; + const match = trimmed.match(/^<([^>]+)> <([^>]+)> (.+) <([^>]+)> \.$/); + if (!match) throw new Error(`unparseable line: ${trimmed}`); + quads.push({ subject: match[1], predicate: match[2], object: match[3], graph: match[4] }); + } + return quads; + } + + async function fetchAllMeta( + cap: CapturedSyncHandler, + cgId: string, + pageSize: number, + afterPage?: (pagesServed: number) => Promise, + ) { + let pagesServed = 0; + return fetchSyncPages({ + ctx: makeCtx(), + remotePeerId: '12D3KooWMetaCeilingRemote', + contextGraphId: cgId, + includeSharedMemory: true, + phase: 'meta', + graphUri: `did:dkg:context-graph:${cgId}/_shared_memory_meta`, + deadline: Date.now() + 60_000, + syncPageTimeoutMs: 10_000, + syncRouterAttempts: 1, + syncPageRetryAttempts: 1, + syncPageSize: pageSize, + syncDeniedResponse: 'sync-denied', + debugSyncProgress: false, + protocolSync: '/origintrail/dkg/sync/1.0.0', + checkpointStore: new MemorySyncCheckpointStore(), + buildSyncRequest: async (contextGraphId, offset, limit, includeSharedMemory, _peer, phase, _snap, _since, syncSessionId) => + new TextEncoder().encode(JSON.stringify({ + contextGraphId, offset, limit, includeSharedMemory, phase, syncSessionId, + })), + parseAndFilter: async (nquadsText) => { + const quads = parseNquads(nquadsText); + return { quads, totalQuads: quads.length }; + }, + send: async (_peerId, _protocolId, data) => { + const envelope = JSON.parse(new TextDecoder().decode(data)) as SyncRequestEnvelope; + const out = await cap.invoke(envelope); + pagesServed += 1; + await afterPage?.(pagesServed); + return new TextEncoder().encode(out); + }, + logWarn: noop, + logInfo: noop, + logDebug: noop, + }); + } + + /** Assert no subject group lost a field to a page boundary (#1788 class). */ + function assertNoStrippedFields(quads: readonly Quad[], expectedGroups: ReadonlyMap) { + const bySubject = new Map>(); + for (const quad of quads) { + const predicates = bySubject.get(quad.subject) ?? new Set(); + predicates.add(quad.predicate); + bySubject.set(quad.subject, predicates); + } + for (const [subject, expectedPredicates] of expectedGroups) { + const predicates = bySubject.get(subject); + expect(predicates, `subject ${subject} missing entirely`).toBeDefined(); + for (const predicate of expectedPredicates) { + expect( + predicates!.has(predicate), + `subject ${subject} lost <${predicate}> across a page boundary`, + ).toBe(true); + } + } + } + + const OP_PREDICATES = [ + RDF_TYPE, + `${DKG_NS}publishedAt`, + `${DKG_NS}rootEntity`, + `${DKG_NS}contextGraphId`, + `${DKG_NS}shareOperationId`, + ] as const; + const HEAD_PREDICATES = [ + `${DKG_NS}contentScopeVersion`, + `${DKG_NS}kaUal`, + `${DKG_NS}assertionVersion`, + `${DKG_NS}shareOperationId`, + `${DKG_NS}assertionGraph`, + ] as const; + + for (const [label, snapshotBudget] of [ + ['session snapshot lane (default budgets)', undefined], + ['plan-paged lane (oversized snapshot)', TINY_SNAPSHOT_BUDGET], + ] as const) { + it(`reassembles every seal/head row-group with no stripped fields via the ${label}`, async () => { + const cgId = `meta-reassembly-${snapshotBudget ? 'paged' : 'snap'}`; + const metaGraph = `did:dkg:context-graph:${cgId}/_shared_memory_meta`; + const fresh = freshIso(); + const store = new OxigraphStore(); + + const expectedGroups = new Map(); + const quads: Quad[] = []; + for (let index = 0; index < 30; index += 1) { + const opId = `op-${String(index).padStart(2, '0')}`; + quads.push(...workspaceOpQuads(cgId, opId, `urn:re:${opId}`, metaGraph, fresh)); + expectedGroups.set(`urn:dkg:share:${cgId}:${opId}`, OP_PREDICATES); + } + for (let index = 0; index < 4; index += 1) { + const ual = `did:dkg:hardhat:31337/0x00000000000000000000000000000000000000ab/${index + 1}`; + quads.push(...graphScopedHeadQuads(cgId, metaGraph, ual, `head-${index}`, fresh)); + expectedGroups.set(`${ual}#dkg-swm-head`, HEAD_PREDICATES); + } + await store.insert(quads); + + // Page size 4 vs 5- and 11-row groups: every group straddles a boundary. + const cap = registerTestSyncHandler(store, { + sharedMemoryTtlMs: TTL_MS, + syncPageSize: 4, + ...(snapshotBudget ? { snapshotBudget } : {}), + }); + const result = await fetchAllMeta(cap, cgId, 4); + + expect(result.completed).toBe(true); + expect(result.timedOut).toBe(false); + expect(result.quads.length).toBe(quads.length); + assertNoStrippedFields(result.quads, expectedGroups); + await store.close(); + }); + } + + it('never completes with a hybrid row-group when a split subject is replaced same-count mid-session (#1868 review repro)', async () => { + // lupuszr's reproduction shape: ONE five-row operation, page size 1, a + // same-count replacement between pages. A count-only guard accepted the + // reread and assembled a five-row hybrid of both versions (omitting + // publishedAt); the content binding must fail the session instead, and the + // requester must never report a completed phase carrying the hybrid. + const cgId = 'meta-samecount-requester'; + const metaGraph = `did:dkg:context-graph:${cgId}/_shared_memory_meta`; + const fresh = freshIso(); + const store = new OxigraphStore(); + await store.insert(workspaceOpQuads(cgId, 'solo', 'urn:sq:solo', metaGraph, fresh)); + const subject = `urn:dkg:share:${cgId}:solo`; + + const cap = registerTestSyncHandler(store, { + sharedMemoryTtlMs: TTL_MS, + syncPageSize: 1, + snapshotBudget: TINY_SNAPSHOT_BUDGET, // plan-paged mode: every page rereads the subject + }); + + const mutateAfterFirstPage = async (pagesServed: number) => { + if (pagesServed !== 1) return; + await store.delete([ + { graph: metaGraph, subject, predicate: `${DKG_NS}publishedAt`, object: `"${fresh}"^^` }, + ]); + await store.insert([ + { graph: metaGraph, subject, predicate: `${DKG_NS}note`, object: '"replacement"' }, + ]); + }; + + let threw = false; + let result: Awaited> | undefined; + try { + result = await fetchAllMeta(cap, cgId, 1, mutateAfterFirstPage); + } catch { + threw = true; + } + if (!threw) { + expect(result!.completed).toBe(false); + } + // Whatever partial rows the requester holds, they must not mix versions: + // the pre-mutation publishedAt row and the post-mutation replacement row + // can never coexist in one assembled row-group. + const objects = (result?.quads ?? []).map((quad) => quad.object).join('\n'); + expect(objects.includes('"replacement"') && objects.includes(fresh)).toBe(false); + await store.close(); + }); +}); diff --git a/packages/agent/vitest.unit.config.ts b/packages/agent/vitest.unit.config.ts index 3d6833a64d..2016fc64c3 100644 --- a/packages/agent/vitest.unit.config.ts +++ b/packages/agent/vitest.unit.config.ts @@ -55,6 +55,7 @@ export default defineConfig({ "test/sync-responder-snapshot-cache.test.ts", "test/sync-responder-cursor.test.ts", "test/sync-responder-oversized-fallback.test.ts", + "test/sync-responder-swm-meta-ceiling.test.ts", "test/sync-responder-large-graph-stack-overflow.test.ts", "test/sync-page-frame-budget.test.ts", "test/sync-byte-budget-pages.test.ts", @@ -111,6 +112,7 @@ export default defineConfig({ "test/workspace-crypto-delegatee-filter.test.ts", "test/swm-public-snapshot-materialization.test.ts", "test/swm-public-cg-plaintext.test.ts", + "test/swm-snapshot-materializer.test.ts", ], testTimeout: 60_000, maxWorkers: 1,