diff --git a/packages/agent/src/dkg-agent-lifecycle.ts b/packages/agent/src/dkg-agent-lifecycle.ts index 373dc3e0b8..d46b9342da 100644 --- a/packages/agent/src/dkg-agent-lifecycle.ts +++ b/packages/agent/src/dkg-agent-lifecycle.ts @@ -235,6 +235,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, @@ -4762,6 +4763,25 @@ export class LifecycleSyncMethods extends DKGAgentBase { const graphManager = new GraphManager(this.store); await graphManager.ensureContextGraph(contextGraphId); }, + // Everything needed to materialize verified public SWM snapshots, + // as ONE dependency (a loose optional trio allowed a silent + // half-configured mode). Graph-scoped (contentScopeVersion 2) KAs + // carry no dkg:rootEntity, so the aggregate data phase returns 0 + // 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. + // 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 f5350100d1..9e0edf6d0d 100644 --- a/packages/agent/src/sync/requester/shared-memory-sync.ts +++ b/packages/agent/src/sync/requester/shared-memory-sync.ts @@ -6,6 +6,12 @@ import type { SyncPhase } from '../auth/request-build.js'; import { didSyncPeerRespond, isSyncBackoffWorthyError, isSyncPermanentRejection, isSyncTransportFailure } from '../error-tags.js'; import { isSharedMemoryBucketDescendantDataGraph } from '../shared-memory-graphs.js'; import type { SyncPageResult } from './page-fetch.js'; +import { + materializeGraphScopedSwmRecoveryAsset, + parseGraphScopedSwmRecoveryDescriptors, + type GraphScopedSwmRecoveryDescriptor, +} from '../graph-scoped-swm-recovery.js'; +import type { SharedMemorySnapshotMaterializer } from './swm-snapshot-materializer.js'; const DKG = 'http://dkg.io/ontology/'; @@ -62,6 +68,19 @@ interface SharedMemorySyncContext { }>; ensureContextGraph: (contextGraphId: string) => Promise; storeInsert: (quads: Quad[]) => Promise; + /** + * Everything needed to MATERIALIZE verified public SWM snapshots into the + * 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?: SharedMemorySnapshotMaterializer; publicSnapshotStore?: WorkspacePublicSnapshotStore; getRegisteredSubGraphNames?: (contextGraphId: string) => Promise; getExcludedSubGraphNames?: (contextGraphId: string) => Promise; @@ -74,6 +93,21 @@ interface SharedMemorySyncContext { logDebug: (ctx: OperationContext, message: string) => void; } + +/** + * True when the locally stored head version outranks the descriptor we are + * about to materialize. BigInt-compared when both parse; anything unparseable + * is treated as OUTRANKING — failing safe means never destroying local state + * whose ordering we cannot establish. + */ +function storedVersionOutranksDescriptor(stored: string, descriptorVersion: string): boolean { + try { + return BigInt(stored) > BigInt(descriptorVersion); + } catch { + return true; + } +} + export async function runSharedMemorySync(context: SharedMemorySyncContext): Promise { const { ctx, @@ -84,6 +118,7 @@ export async function runSharedMemorySync(context: SharedMemorySyncContext): Pro processSharedMemoryBatch, ensureContextGraph, storeInsert, + snapshotMaterializer, publicSnapshotStore, getRegisteredSubGraphNames, getExcludedSubGraphNames, @@ -223,6 +258,136 @@ export async function runSharedMemorySync(context: SharedMemorySyncContext): Pro summary.droppedDataTriples += dropped; } + // MATERIALIZE verified snapshots into the store, mirroring the private + // recovery lane (`swm-recovery.ts` materializeReadySnapshot). + // + // Descriptors are parsed ONLY from verified meta, and only when the meta + // phase completed: parseGraphScopedSwmRecoveryDescriptors throws on + // incomplete metadata, and this lane pages meta, so a timed-out page would + // otherwise abort the whole CG fanout. A parse failure here must degrade to + // "no materialization this round" — never take down the sync. + const snapshotDescriptorsByRef = new Map(); + if (snapshotMaterializer && publicSnapshotStore && wsMetaResult.completed) { + try { + for (const descriptor of parseGraphScopedSwmRecoveryDescriptors({ + contextGraphId: pid, + metaQuads: processed.verifiedMeta, + // Without the subgraph admission context every KA under a + // REGISTERED subgraph is judged to live in an unregistered metadata + // graph. The parser then throws, the catch clears ALL descriptors, + // and materialization is silently disabled for the whole context + // graph — not just for the subgraph KA that triggered it. + ...(registeredSubGraphNames ? { registeredSubGraphNames } : {}), + ...(excludedSubGraphNames ? { excludedSubGraphNames } : {}), + })) { + const ref = descriptor.publicSnapshotRef; + if (!ref) continue; // no immutable snapshot for this KA + const list = snapshotDescriptorsByRef.get(ref) ?? []; + list.push(descriptor); + snapshotDescriptorsByRef.set(ref, list); + } + } catch (err) { + logWarn(ctx, `SWM sync could not parse graph-scoped snapshot descriptors for "${pid}": ` + + `${err instanceof Error ? err.message : String(err)}`); + snapshotDescriptorsByRef.clear(); + } + } + let materializedGraphs = 0; + let materializationFailures = 0; + let materializedQuads = 0; + const materializedKeys = new Set(); + const materializeReadySnapshot = async (snapshotRef: string): Promise => { + const descriptors = snapshotDescriptorsByRef.get(snapshotRef); + if (!descriptors?.length || !snapshotMaterializer || !publicSnapshotStore) return; + for (const descriptor of descriptors) { + const graphKey = `${descriptor.metaGraph}\u0000${descriptor.assertionGraph}`; + if (materializedKeys.has(graphKey)) continue; + try { + await snapshotMaterializer.withKaWriteLock( + pid, + descriptor.subGraphName, + descriptor.kaUal, + async () => { + // ALL decisions live INSIDE the lock. Between our pre-lock view + // of the world and acquisition, live gossip may have committed + // this KA — the lock stops the interleaving, and the two + // re-checks below stop the other failure the lock alone cannot: + // replacing newer content with an older verified snapshot. + // + // (a) Version ordering. A stored head newer than the descriptor + // means gossip advanced this KA past our snapshot; replacing + // 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. 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 ${storedHead.version} (descriptor ${descriptor.assertionVersion}); skipping`); + return; + } + // (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; + } + const asset = await materializeGraphScopedSwmRecoveryAsset({ + descriptor, + fetchedDataQuads: [], + publicSnapshotStore, + }); + 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; + logInfo(ctx, `SWM sync for "${pid}": materialized snapshot ${snapshotRef} ` + + `as ${asset.assertionGraph} (${asset.quads.length} triples)`); + }, + ); + } catch (err) { + // A failed replace must never be able to look materialized later. + // Suppressing it here while the surrounding sync still inserts the + // graph-scoped head marker makes the loss PERMANENT: the next pass + // sees that marker, isGraphAssetMaterialized returns true, and the + // missing assertion graph is skipped forever. Record the failure so + // the caller keeps the phase incomplete and withholds the metadata + // that would otherwise certify a graph that was never written. + materializationFailures += 1; + logWarn(ctx, `SWM sync failed to materialize snapshot ${snapshotRef} for "${pid}": ` + + `${err instanceof Error ? err.message : String(err)}`); + } + } + }; + const snapshotStartedAt = Date.now(); const snapshotSync = await syncPublicSnapshotsForMeta({ ctx, @@ -234,14 +399,39 @@ export async function runSharedMemorySync(context: SharedMemorySyncContext): Pro fetchSyncPages, deleteCheckpoint, setCheckpoint, + // Fires for BOTH 'cache' and 'network' sources, so a node whose earlier + // runs already cached the blobs materializes them on the next pass + // without refetching a byte. + ...(snapshotDescriptorsByRef.size > 0 + ? { onSnapshotReady: (snapshot: PublicSnapshotMetadata) => materializeReadySnapshot(snapshot.ref) } + : {}), }); + if (materializedGraphs > 0) { + summary.insertedTriples += materializedQuads; + // Also data progress: lifecycle readiness classifies a round with zero + // insertedDataTriples as metadata-only, which would mis-report a + // successful graph-scoped materialization as "no data". + summary.insertedDataTriples += materializedQuads; + logInfo(ctx, `SWM sync for "${pid}": materialized ${materializedGraphs} graph-scoped ` + + `KA snapshot(s) totalling ${materializedQuads} triples`); + } summary.bytesReceived += snapshotSync.bytesReceived; summary.resumedPhases += snapshotSync.resumedPhases; summary.timedOutPhases += snapshotSync.timedOutPhases; summary.completedPhases += snapshotSync.completedPhases; summary.checkpointAdvances += snapshotSync.checkpointAdvances; const snapshotDurationMs = Date.now() - snapshotStartedAt; - if (!snapshotSync.completed) { + // A snapshot that verified but could not be written must be treated + // exactly like a snapshot phase that did not complete. Otherwise the meta + // insert below stamps a graph-scoped head marker for an assertion graph + // that was never materialized, and every later pass skips it as already + // present — turning a transient store error into permanent, silent loss. + const snapshotPhaseUsable = snapshotSync.completed && materializationFailures === 0; + if (materializationFailures > 0) { + logWarn(ctx, `SWM sync for "${pid}": ${materializationFailures} snapshot(s) verified but ` + + `not materialized — holding the phase incomplete so metadata cannot certify them`); + } + if (!snapshotPhaseUsable) { // The responder was reachable, but the snapshot phase did not produce // a complete, verified snapshot. Preserve any verified data prefix // below, while keeping the overall sync result non-successful so the 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/test/swm-public-snapshot-materialization.test.ts b/packages/agent/test/swm-public-snapshot-materialization.test.ts new file mode 100644 index 0000000000..276e279440 --- /dev/null +++ b/packages/agent/test/swm-public-snapshot-materialization.test.ts @@ -0,0 +1,372 @@ +/** + * Public SWM catch-up snapshot MATERIALIZATION — the behavior that turns a + * 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) + * and asserts the DECISIONS around the destructive `replaceGraph`: + * + * 1. a held-out node materializes a cached snapshot into the store + * 2. the race with live gossip is closed: catch-up blocks on the SAME + * per-KA write lock (real `withKeyedLocks` + `swmKaWriteLockKey`, shared + * map), and the in-lock version re-check skips when gossip advanced the + * KA while catch-up waited — replace is never called + * 3. the pre-fix broken state (head marker present, graph never written) + * is HEALED, because the materialized-check is content-based + * 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 { + GRAPH_KA_CONTENT_SCOPE_VERSION, + MemoryLayer, + createGraphKnowledgeAssetScope, + contextGraphWorkspaceGraphUri, + contextGraphWorkspaceMetaGraphUri, + knowledgeAssetLayerGraphUri, + type OperationContext, +} from '@origintrail-official/dkg-core'; +import { + generateKnowledgeAssetShareMetadata, + workspacePublicQuadsDigest, + withKeyedLocks, + swmKaWriteLockKey, + type WorkspacePublicSnapshotStore, +} from '@origintrail-official/dkg-publisher'; +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); +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/0xaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa/7'; +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; + } +} + +function page(quads: Quad[], completed = true): SyncPageResult { + 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(subGraphName?: string) { + const scope = createGraphKnowledgeAssetScope(UAL, 1); + 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`; + const payload: Quad[] = [ + { subject: 'urn:snap:a', predicate: 'http://schema.org/status', object: '"held-out"', graph: '' }, + { subject: 'urn:snap:b', predicate: 'http://schema.org/status', object: '"held-out"', graph: '' }, + ]; + const digest = workspacePublicQuadsDigest(payload); + const meta: Quad[] = [ + ...generateKnowledgeAssetShareMetadata({ + shareOperationId: operationId, + contextGraphId: CG, + kaUal: UAL, + assertionVersion: 1, + publicTripleCount: payload.length, + privateTripleCount: 0, + publisherPeerId: 'peer-source', + timestamp: new Date(0), + ...(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, metaGraph, assertionGraph }; +} + +interface HarnessOverrides { + 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(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 (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, _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, + totalFetchedDataQuads: wsDataQuads.length, + totalFetchedMetaQuads: wsMetaQuads.length, + droppedDataTriples: 0, + emptyResponses: 0, + entityCreators: [], + }), + ...(overrides.subGraphName + ? { + getRegisteredSubGraphNames: async () => [overrides.subGraphName!], + getExcludedSubGraphNames: async () => [], + } + : {}), + ensureContextGraph: async () => {}, + storeInsert: async (quads) => { + events.push('meta-inserted'); + inserted.push(quads); + }, + snapshotMaterializer: { + withKaWriteLock: (contextGraphId, subGraphName, kaUal, fn) => { + events.push('lock-requested'); + overrides.onLockRequested?.(); + return withKeyedLocks(lockMap, [swmKaWriteLockKey(contextGraphId, subGraphName, kaUal)], fn); + }, + isGraphAssetMaterialized: async () => { + events.push('content-checked'); + return overrides.contentPresent?.() ?? false; + }, + readStoredHead: async () => { + events.push('version-read'); + 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: () => {}, + setCheckpoint: () => {}, + ensureOwnedMap: () => new Map(), + logInfo: () => {}, + logWarn: () => {}, + logDebug: () => {}, + }); + }; + return { fx, run, events, replaced, headSwaps, inserted, snapshotFetches, lockMap }; +} + +describe('public SWM snapshot materialization', () => { + it('materializes a cached snapshot into the assertion graph for a held-out node', async () => { + const h = harness(); + const summary = await h.run(); + expect(h.replaced).toHaveLength(1); + expect(h.replaced[0]!.graphUri).toBe(h.fx.assertionGraph); + expect(h.replaced[0]!.quads).toHaveLength(h.fx.payload.length); + expect(h.replaced[0]!.quads.every((q) => q.graph === h.fx.assertionGraph)).toBe(true); + // Counted as DATA progress, not metadata-only. + expect(summary.insertedDataTriples).toBeGreaterThanOrEqual(h.fx.payload.length); + expect(summary.failedPhases).toBe(0); + // Meta made it to the store: the phase was usable. + 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. + const lockMap = new Map>(); + let releaseGossip!: () => void; + const gossipDone = new Promise((r) => { releaseGossip = r; }); + let storedVersion: string | null = null; + let sawLockRequest!: () => void; + const lockRequested = new Promise((r) => { sawLockRequest = r; }); + + const h = harness({ + lockMap, + storedHead: () => ({ version: storedVersion, needsRepair: false }), + onLockRequested: () => sawLockRequest(), + }); + + // Gossip enters first and holds the per-KA critical section. + const gossipHold = withKeyedLocks( + lockMap, + // Checksummed-case UAL on purpose: the shared key helper lowercases, so + // a case difference between gossip's UAL and the descriptor's must still + // serialize on ONE key. An under-merged key would let this test pass + // vacuously with no contention at all. + [swmKaWriteLockKey(CG, undefined, UAL.toUpperCase().replace('DID:DKG:HARDHAT', 'did:dkg:hardhat'))], + async () => { await gossipDone; }, + ); + + const syncPromise = h.run(); + await lockRequested; // catch-up has asked for the lock… + storedVersion = '2'; // …gossip commits version 2 meanwhile… + h.events.push('gossip-committed'); + releaseGossip(); // …and leaves the critical section. + const summary = await syncPromise; + await gossipHold; + + // Catch-up proceeded only after gossip, saw the newer stored version, and + // 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 () => { + // 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({ 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({ 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); + }); + + it('withholds the meta insert when a replace fails, and marks the phase failed', async () => { + const h = harness({ replaceImpl: async () => { throw new Error('store unavailable'); } }); + const summary = await h.run(); + expect(summary.failedPhases).toBe(1); + // No meta batch reached the store: a head marker must never certify an + // assertion graph that was not written, or the next pass would classify + // 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/vitest.unit.config.ts b/packages/agent/vitest.unit.config.ts index f89c5a1511..0cbf40ed9c 100644 --- a/packages/agent/vitest.unit.config.ts +++ b/packages/agent/vitest.unit.config.ts @@ -103,6 +103,8 @@ export default defineConfig({ "test/cg-resolve-refresh.test.ts", "test/private-cg-membership-bootstrap.test.ts", "test/workspace-crypto-delegatee-filter.test.ts", + "test/swm-public-snapshot-materialization.test.ts", + "test/swm-snapshot-materializer.test.ts", ], testTimeout: 60_000, maxWorkers: 1, diff --git a/packages/publisher/src/index.ts b/packages/publisher/src/index.ts index caef25d7e2..1377467433 100644 --- a/packages/publisher/src/index.ts +++ b/packages/publisher/src/index.ts @@ -325,3 +325,4 @@ export { ChainEventPoller, type ChainEventPollerConfig, type CursorPersistence, export { AccessHandler, type AccessPolicy } from './access-handler.js'; export { AccessClient, type AccessResult } from './access-client.js'; export * from './share-batching.js'; +export { withKeyedLocks, swmKaWriteLockKey } from './keyed-lock.js'; diff --git a/packages/publisher/src/keyed-lock.ts b/packages/publisher/src/keyed-lock.ts index f9d7defb5b..4899e994f6 100644 --- a/packages/publisher/src/keyed-lock.ts +++ b/packages/publisher/src/keyed-lock.ts @@ -37,3 +37,36 @@ export async function withKeyedLocks( } } } + +/** + * The per-KA SWM write-lock key. + * + * EVERY path that writes a shared-working-memory per-KA layer graph must + * serialize on this exact key, against the SAME lock map (the agent owns it and + * injects it into SharedMemoryHandler). Live gossip already does; the public + * catch-up materializer must too, or this interleaving destroys data: + * + * 1. catch-up observes the KA's assertion graph is absent + * 2. gossip acquires its lock and commits a newer, richer graph + * 3. catch-up replaces that graph with its older verified snapshot + * + * Deriving the key in ONE exported function is what makes drift impossible — + * two hand-rolled copies of this string format would fail silently, as an + * unequal key does not error, it just stops serializing. + * + * The UAL segment is lowercased: UAL address segments appear in both + * checksummed and lowercase forms depending on the source (chain read vs head + * subject), and a case mismatch would under-merge the lock. Over-merging two + * distinct KAs into one key would merely coarsen serialization; under-merging + * recreates the race. Lowercase is therefore the safe direction. + */ +export function swmKaWriteLockKey( + contextGraphId: string, + subGraphName: string | undefined, + kaUal: string, +): string { + const lockNamespace = subGraphName + ? `${contextGraphId}\0${subGraphName}` + : contextGraphId; + return `${lockNamespace}\0ka\0${kaUal.toLowerCase()}`; +} diff --git a/packages/publisher/src/workspace-handler.ts b/packages/publisher/src/workspace-handler.ts index 9f26f23f05..7731cf0a1f 100644 --- a/packages/publisher/src/workspace-handler.ts +++ b/packages/publisher/src/workspace-handler.ts @@ -26,7 +26,7 @@ import { import type { EncryptedWorkspacePayloadMsg, GossipEnvelopeMsg, OperationContext, SwmSenderKeyMessageMsg, WorkspaceCASConditionMsg, WorkspacePublishRequestMsg, WorkspaceRecipientEncryptionKey } from '@origintrail-official/dkg-core'; import { ethers } from 'ethers'; import { validateKnowledgeAssetPublishRequest } from './validation.js'; -import { withKeyedLocks } from './keyed-lock.js'; +import { withKeyedLocks, swmKaWriteLockKey } from './keyed-lock.js'; import { generateSubGraphRegistration } from './metadata.js'; import { parseSimpleNQuads } from './publish-handler.js'; import { @@ -1304,13 +1304,12 @@ export class SharedMemoryHandler { contentScope, subGraphName, ); - const lockNamespace = subGraphName - ? `${contextGraphId}\0${subGraphName}` - : contextGraphId; // All assertion versions currently replace the same exact per-KA layer // graph. Lock by UAL, not subject and not version, so concurrent version - // deliveries cannot interleave a DROP/INSERT pair. - const lockKeys = [`${lockNamespace}\0ka\0${contentScope.ual}`]; + // deliveries cannot interleave a DROP/INSERT pair. The key is derived by + // the SHARED helper so the public catch-up materializer serializes on the + // identical string — see swmKaWriteLockKey for why drift here is silent. + const lockKeys = [swmKaWriteLockKey(contextGraphId, subGraphName, contentScope.ual)]; onPhase?.('store', 'start'); const applied = await this.withWriteLocks(lockKeys, async (): Promise => {