From ccb0eaaa5a30295054c23bbb7d893ae6b81a100f Mon Sep 17 00:00:00 2001 From: Bojan Date: Tue, 4 Aug 2026 10:54:55 +0200 Subject: [PATCH] fix(agent): prevent RS-heal store starvation --- packages/agent/src/dkg-agent-base.ts | 18 + packages/agent/src/dkg-agent-cg-resolve.ts | 73 ++- packages/agent/src/dkg-agent-swm-host.ts | 211 ++++++-- packages/agent/test/core-fills-gap.test.ts | 8 +- .../rs-heal-stranded-kc-decorated.test.ts | 54 +- .../agent/test/rs-heal-stranded-kc.test.ts | 270 +++++++++- packages/cli/src/daemon/http-utils.ts | 106 +++- .../test/write-preflight-resilience.test.ts | 496 ++++++++++++++++++ packages/publisher/src/metadata.ts | 14 +- 9 files changed, 1156 insertions(+), 94 deletions(-) diff --git a/packages/agent/src/dkg-agent-base.ts b/packages/agent/src/dkg-agent-base.ts index 55bd1850a7..1ed8f980f4 100644 --- a/packages/agent/src/dkg-agent-base.ts +++ b/packages/agent/src/dkg-agent-base.ts @@ -934,6 +934,22 @@ export class DKGAgentBase { */ static readonly VM_RECONCILE_BATCH_SIZE = Math.max(1, Number(process.env['DKG_VM_RECONCILE_BATCH_SIZE']) || 10); + /** Hard ceiling: RS heal is best-effort maintenance and must stay bounded. */ + static readonly RS_HEAL_BATCH_MAX = 64; + /** + * Maximum stranded KCs one RS-heal pass may inspect before yielding. RS heal + * is periodic repair work, so bounding each pass keeps foreground publish, + * SWM and gossip operations from competing with an entire historical backlog. + */ + static readonly RS_HEAL_BATCH_SIZE = Math.min( + DKGAgentBase.RS_HEAL_BATCH_MAX, + readPositiveSafeIntegerEnv('DKG_RS_HEAL_BATCH_SIZE', 8), + ); + /** Bounded per-CG keyset cursor retention for the independent RS-heal pager. */ + static readonly RS_HEAL_CG_STATE_MAX_ENTRIES = Math.min( + 10_000, + readPositiveSafeIntegerEnv('DKG_RS_HEAL_CG_STATE_MAX_ENTRIES', 1_000), + ); /** * Parallel ordinal work per CG. Combined with the default two-CG dispatcher * concurrency this caps chain/store pressure at ten in-flight ordinals. @@ -1043,6 +1059,8 @@ export class DKGAgentBase { protected vmReconcileLifecycleController = new AbortController(); /** Phase D/A4 — per-CG active-fetch cooldown so one sweep cannot fan out repeated fetches. */ protected readonly vmReconcileFetchCooldownAt = new Map(); + /** Last stranded UAL visited by the bounded RS-heal sweep for each CG. */ + protected readonly rsHealCursorByCg = new Map(); /** Phase D/A4 — round-robin cursor over the already ordered catch-up peer list. */ protected readonly vmReconcileCatchupPeerCursor = new Map(); protected readonly vmReconcileCatchupPeerOrder = new Map <${contextGraphUri}>) + (<${agentsGraph}> <${contextGraphUri}>) + (<${cgMetaGraph}> <${metaSubjectUri}>) + } { - GRAPH <${ontologyGraph}> { - <${contextGraphUri}> <${DKG_ONTOLOGY.RDF_TYPE}> <${DKG_ONTOLOGY.DKG_CONTEXT_GRAPH}> . - OPTIONAL { <${contextGraphUri}> <${DKG_ONTOLOGY.DKG_ACCESS_POLICY}> ?access } - OPTIONAL { <${contextGraphUri}> <${DKG_ONTOLOGY.DKG_CURATOR}> ?curator } + GRAPH ?sourceGraph { + ?sourceSubject <${DKG_ONTOLOGY.RDF_TYPE}> <${DKG_ONTOLOGY.DKG_CONTEXT_GRAPH}> . } + BIND("declared" AS ?kind) } UNION { - GRAPH <${agentsGraph}> { - <${contextGraphUri}> <${DKG_ONTOLOGY.RDF_TYPE}> <${DKG_ONTOLOGY.DKG_CONTEXT_GRAPH}> . - OPTIONAL { <${contextGraphUri}> <${DKG_ONTOLOGY.DKG_ACCESS_POLICY}> ?access } - OPTIONAL { <${contextGraphUri}> <${DKG_ONTOLOGY.DKG_CURATOR}> ?curator } + GRAPH ?sourceGraph { + ?sourceSubject <${DKG_ONTOLOGY.DKG_ACCESS_POLICY}> ?value . } + BIND("access" AS ?kind) } UNION { - GRAPH <${cgMetaGraph}> { - <${metaSubjectUri}> <${DKG_ONTOLOGY.RDF_TYPE}> <${DKG_ONTOLOGY.DKG_CONTEXT_GRAPH}> . - OPTIONAL { <${metaSubjectUri}> <${DKG_ONTOLOGY.DKG_ACCESS_POLICY}> ?access } - OPTIONAL { <${metaSubjectUri}> <${DKG_ONTOLOGY.DKG_CURATOR}> ?curator } + GRAPH ?sourceGraph { + ?sourceSubject <${DKG_ONTOLOGY.DKG_CURATOR}> ?value . } + BIND("curator" AS ?kind) + } UNION { + FILTER EXISTS { + VALUES ?gatePredicate { + <${DKG_ONTOLOGY.DKG_ALLOWED_AGENT}> + <${DKG_ONTOLOGY.DKG_ALLOWED_PEER}> + <${DKG_ONTOLOGY.DKG_PARTICIPANT_AGENT}> + <${DKG_ONTOLOGY.DKG_PARTICIPANT_IDENTITY_ID}> + } + GRAPH ?sourceGraph { + ?sourceSubject ?gatePredicate ?gateValue . + } + } + BIND("gate" AS ?kind) } } `); @@ -955,26 +978,36 @@ export class ContextGraphResolveMethods extends DKGAgentBase { : noteStoreFailure(declarationRead.reason); let accessPolicy: 'public' | 'private' | undefined; - // Tri-state: stays `undefined` (unknown) when the declaration read + // Tri-state: stays `undefined` (unknown) when the metadata projection read // failed, so a store outage can never masquerade as "no declaration". let declarationFound: boolean | undefined = declarationResult === undefined ? undefined : false; const curators: string[] = []; if (declarationResult && declarationResult.type === 'bindings') { - declarationFound = declarationResult.bindings.length > 0; let sawPublic = false; let sawPrivate = false; + let sawGate = false; for (const row of declarationResult.bindings as Record[]) { - const access = row['access']; - if (typeof access === 'string') { - const normalized = stripLiteral(access).trim().toLowerCase(); + const kind = typeof row['kind'] === 'string' + ? stripLiteral(row['kind']).trim().toLowerCase() + : ''; + const value = row['value']; + if (kind === 'declared') declarationFound = true; + if (kind === 'access' && typeof value === 'string') { + const normalized = stripLiteral(value).trim().toLowerCase(); if (normalized === 'private') sawPrivate = true; if (normalized === 'public') sawPublic = true; } - const curator = row['curator']; - if (typeof curator === 'string' && curator.trim()) curators.push(curator); + if (kind === 'curator' && typeof value === 'string' && value.trim()) { + curators.push(value); + } + if (kind === 'gate') sawGate = true; } if (sawPrivate) accessPolicy = 'private'; else if (sawPublic) accessPolicy = 'public'; + // Match list/projection semantics: an explicit public policy wins, but + // legacy/implicit allowlist gates make a graph private when no policy + // literal exists. This is a bounded point query for one canonical id. + else if (sawGate) accessPolicy = 'private'; } let checksum: string | null = null; diff --git a/packages/agent/src/dkg-agent-swm-host.ts b/packages/agent/src/dkg-agent-swm-host.ts index f6a3f3040f..09d5c66214 100644 --- a/packages/agent/src/dkg-agent-swm-host.ts +++ b/packages/agent/src/dkg-agent-swm-host.ts @@ -95,7 +95,7 @@ import { SUBSCRIPTION_SOURCES, pickNetworkTunables, } from '@origintrail-official/dkg-core'; -import { GraphManager, PrivateContentStore, asChangelogReader, asGraphWriteGenSource, createTripleStore, tryUpdateWithTouchedGraphs, type TripleStore, type TripleStoreConfig, type Quad, type LargeLiteralStorageConfig } from '@origintrail-official/dkg-storage'; +import { GraphManager, PrivateContentStore, StoreSchedulerBusyError, asChangelogReader, asGraphWriteGenSource, createTripleStore, tryUpdateWithTouchedGraphs, type TripleStore, type TripleStoreConfig, type QueryOptions, type Quad, type LargeLiteralStorageConfig, type SelectResult } from '@origintrail-official/dkg-storage'; import { EVMChainAdapter, NoChainAdapter, enrichEvmError, type EVMAdapterConfig, type ChainAdapter, type CreateContextGraphParams, type CreateOnChainContextGraphParams, type CreateOnChainContextGraphResult, type TxResult, type V10PublishingConvictionAccountInfo } from '@origintrail-official/dkg-chain'; import { DKGPublisher, PublishHandler, SharedMemoryHandler, UpdateHandler, ChainEventPoller, AccessHandler, AccessClient, @@ -250,6 +250,81 @@ import { type VmReconcileSource, } from './vm-reconcile-service.js'; import { createCursorState, type CursorState } from './reconcile-cursor.js'; + +function rsHealStoreOptions(operation: string, signal?: AbortSignal): QueryOptions { + return { + priority: 'background', + source: `agent.swm.rsHeal.${operation}`, + ...(signal ? { signal } : {}), + }; +} + +function isStoreSchedulerBusyError(err: unknown): boolean { + return err instanceof StoreSchedulerBusyError || ( + typeof err === 'object' && err !== null && + (err as { code?: unknown }).code === 'STORE_SCHEDULER_BUSY' + ); +} + +export type RsHealPassResult = + | { status: 'completed'; inspected: number } + | { status: 'skipped'; reason: 'not-current' | 'unsupported-store' | 'no-work' | 'invalid-result' | 'failed' } + | { status: 'deferred'; reason: 'store-busy' }; + +async function readRsHealStrandedPage( + store: TripleStore, + legacyMeta: string, + scopedMeta: string, + dkgNamespace: string, + cursor: string | undefined, + batchSize: number, + signal?: AbortSignal, +): Promise { + const result = await store.query( + `SELECT ?ual ?b WHERE { + GRAPH <${legacyMeta}> { ?ual <${dkgNamespace}batchId> ?b } + FILTER(isIRI(?ual)) + FILTER NOT EXISTS { + GRAPH <${scopedMeta}> { + ?ual <${dkgNamespace}batchId> ?b ; <${dkgNamespace}materializedVersion> ?version + } + } + ${cursor ? `FILTER(STR(?ual) > ${sparqlString(cursor)})` : ''} + } + ORDER BY STR(?ual) + LIMIT ${batchSize}`, + rsHealStoreOptions('enumerate', signal), + ); + return result.type === 'bindings' ? result : null; +} + +function advanceRsHealCursor( + cursorMap: Map, + cursorKey: string, + bindings: SelectResult['bindings'], + batchSize: number, + maxEntries: number, +): void { + if (bindings.length === 0 || bindings.length < batchSize) { + cursorMap.delete(cursorKey); + return; + } + const lastUal = stripBindingQuotes(bindings[bindings.length - 1]?.['ual'] ?? ''); + if (!lastUal || !isSafeIri(lastUal)) { + // The query constrains ?ual to an IRI, but fail open to a fresh scan rather + // than pinning a corrupt cursor if an adapter returns malformed bindings. + cursorMap.delete(cursorKey); + return; + } + cursorMap.delete(cursorKey); + cursorMap.set(cursorKey, lastUal); + while (cursorMap.size > maxEntries) { + const oldest = cursorMap.keys().next().value; + if (oldest === undefined) break; + cursorMap.delete(oldest); + } +} + // rc.9 PR-10: JoinApprovalRetryQueue removed — substrate outbox // (durable, SQLite-backed) replaces it. We keep a minimal local // type alias so listPendingJoinApprovalRetries() retains its old @@ -2894,16 +2969,6 @@ export class SwmHostModeMethods extends DKGAgentBase { && this.reconcileCursors.get(localCgId) === target.cursor; if (!isTargetCurrent()) throw new VmReconcileQueueClosedError(); - // Keep the legacy-label -> scoped-VM migration in the admitted lane and - // before the evidence gate. A current watermark can still need this repair. - await this.healStrandedScopedKCs( - localCgId, - target.sub, - isTargetCurrent, - lifecycleSignal, - ); - if (!isTargetCurrent()) throw new VmReconcileQueueClosedError(); - // Reconcile on a private cursor snapshot. The caller-facing abort race may // finish before an adapter physically settles; a stale continuation must // never mutate the live cursor or persist a watermark into a new binding. @@ -2954,8 +3019,31 @@ export class SwmHostModeMethods extends DKGAgentBase { // Queue one trailing slice while this key is still active. The dispatcher // places it behind already-waiting live CGs, so a large graph makes steady // progress without monopolising the only VM worker. - if (isLifecycleCurrent() && (result.hasMore || result.staleTarget)) { + const hasImmediateTrailingWork = result.hasMore || result.staleTarget; + if (isLifecycleCurrent() && hasImmediateTrailingWork) { this.vmReconcileDispatcher?.triggerLive(localCgId); + } else if (isTargetCurrent()) { + // RS heal is bounded, best-effort maintenance. Run it only after the + // useful VM slice completed and only when that slice has no urgent + // continuation. Store pressure must defer maintenance, never erase the + // main reconcile result or prevent foreground ordinal progress. + try { + await this.healStrandedScopedKCs( + localCgId, + target.sub, + isTargetCurrent, + lifecycleSignal, + ); + } catch (err) { + // Defensive isolation at the dispatcher boundary: the heal method + // reduces known pressure to a deferred result, but a future repair + // regression must still never erase an already-computed VM result. + this.log.warn( + createOperationContext('system'), + `RS heal after VM reconcile for "${localCgId}" was skipped: ${err instanceof Error ? err.message : String(err)}`, + ); + } + if (!isTargetCurrent()) throw new VmReconcileQueueClosedError(); } return response; })(); @@ -3225,17 +3313,21 @@ export class SwmHostModeMethods extends DKGAgentBase { sub: ContextGraphSub, isCurrent: () => boolean = () => true, signal?: AbortSignal, - ): Promise { + ): Promise { try { const capturedOnChainId = sub.onChainId; const canApply = () => isCurrent() && (!(this.subscribedContextGraphs instanceof Map) || this.subscribedContextGraphs.get(localCgId) === sub) && sub.onChainId === capturedOnChainId; - if (!canApply() || !capturedOnChainId) return; + if (!canApply() || !capturedOnChainId) { + return { status: 'skipped', reason: 'not-current' }; + } // Server-side byte-safe copy is the ONLY safe relocation mechanism; if the // backend can't do SPARQL UPDATE we bail rather than risk a lossy JS round-trip. - if (typeof this.store.update !== 'function') return; + if (typeof this.store.update !== 'function') { + return { status: 'skipped', reason: 'unsupported-store' }; + } // #1549: every server-side INSERT in this RS-heal path has a statically-known // target graph, so `touchedGraphs` is REQUIRED — the index then maintains // itself incrementally (a bounded `hasGraph`) instead of marking the whole @@ -3247,7 +3339,7 @@ export class SwmHostModeMethods extends DKGAgentBase { this.store, sparql, touchedGraphs, - { signal, source: 'agent.swm.rsHeal.materialize' }, + rsHealStoreOptions('materialize', signal), ); if (!updated) throw new Error('RS heal requires server-side update() support'); }; @@ -3277,26 +3369,42 @@ export class SwmHostModeMethods extends DKGAgentBase { } } }`, - { signal, source: 'agent.swm.rsHeal.findLegacyOnly' }, + rsHealStoreOptions('guard', signal), ); - if (!canApply() || askGuard.type !== 'boolean' || !askGuard.value) return; - - // 2b: enumerate the stranded UALs. - const stranded = await this.store.query( - `SELECT ?ual ?b WHERE { - GRAPH <${legacyMeta}> { ?ual <${DKG}batchId> ?b } - FILTER NOT EXISTS { - GRAPH <${scopedMeta}> { - ?ual <${DKG}batchId> ?b ; <${DKG}materializedVersion> ?version - } - } - }`, - { signal, source: 'agent.swm.rsHeal.listLegacyOnly' }, + if (!canApply()) return { status: 'skipped', reason: 'not-current' }; + if (askGuard.type !== 'boolean') return { status: 'skipped', reason: 'invalid-result' }; + if (!askGuard.value) return { status: 'skipped', reason: 'no-work' }; + + // 2b: enumerate one bounded page. A per-CG lexical cursor means a + // permanently incomplete KC cannot pin the first page forever; after the + // final page the cursor wraps and the next sweep retries earlier gaps. + const cursorKey = `${localCgId}\u0000${sub.onChainId}`; + const cursorMap = this.rsHealCursorByCg ?? new Map(); + const cursor = cursorMap.get(cursorKey); + const stranded = await readRsHealStrandedPage( + this.store, + legacyMeta, + scopedMeta, + DKG, + cursor, + DKGAgentBase.RS_HEAL_BATCH_SIZE, + signal, ); - if (!canApply() || stranded.type !== 'bindings') return; + if (!canApply()) return { status: 'skipped', reason: 'not-current' }; + if (!stranded) return { status: 'skipped', reason: 'invalid-result' }; + if (stranded.bindings.length === 0) { + advanceRsHealCursor( + cursorMap, + cursorKey, + stranded.bindings, + DKGAgentBase.RS_HEAL_BATCH_SIZE, + DKGAgentBase.RS_HEAL_CG_STATE_MAX_ENTRIES, + ); + return { status: 'skipped', reason: 'no-work' }; + } for (const row of stranded.bindings) { - if (!canApply()) return; + if (!canApply()) return { status: 'skipped', reason: 'not-current' }; // Bindings come back stripped to bare values by the store adapters // (oxigraph/sparql-http both emit IRIs unwrapped); strip + validate // exactly as the extractor does for its `ual`. @@ -3331,10 +3439,22 @@ export class SwmHostModeMethods extends DKGAgentBase { // stamping the LOWEST version {0,0}: the GH#842 ordering guard then // lets any real update (block>0) win over this floor and never the // reverse, so it can never clobber a genuine update. - const version = (await readMaterializedVersion(this.store, legacyMeta, ual)) + const version = (await readMaterializedVersion( + this.store, + legacyMeta, + ual, + rsHealStoreOptions('version.readLegacy', signal), + )) ?? { blockNumber: 0, txIndex: 0 }; if (!canApply()) return; - if (!(await shouldApplyMaterialization(this.store, scopedMeta, ual, version))) return; // idempotent + if (!(await shouldApplyMaterialization( + this.store, + scopedMeta, + ual, + version, + undefined, + rsHealStoreOptions('version.checkScoped', signal), + ))) return; // idempotent if (!canApply()) return; assertSafeIri(ual); @@ -3348,7 +3468,7 @@ export class SwmHostModeMethods extends DKGAgentBase { { <${ual}> <${DKG}rootEntity> ?root . } } }`, - { signal, source: 'agent.swm.rsHeal.readRoots' }, + rsHealStoreOptions('roots', signal), ); if (!canApply() || rootsRes.type !== 'bindings') return; const roots: string[] = []; @@ -3382,7 +3502,7 @@ export class SwmHostModeMethods extends DKGAgentBase { } } }`, - { signal, source: 'agent.swm.rsHeal.checkRootData' }, + rsHealStoreOptions('rootPresent', signal), ); if (!canApply() || present.type !== 'boolean' || !present.value) return; } @@ -3459,17 +3579,36 @@ export class SwmHostModeMethods extends DKGAgentBase { } }, { signal }); } catch (err) { + if (isStoreSchedulerBusyError(err)) throw err; + if (signal?.aborted || !canApply()) { + return { status: 'skipped', reason: 'not-current' }; + } this.log.warn( createOperationContext('system'), `RS heal: relocate failed for ${ual} (cg=${capturedOnChainId}): ${err instanceof Error ? err.message : String(err)}`, ); } } + advanceRsHealCursor( + cursorMap, + cursorKey, + stranded.bindings, + DKGAgentBase.RS_HEAL_BATCH_SIZE, + DKGAgentBase.RS_HEAL_CG_STATE_MAX_ENTRIES, + ); + return { status: 'completed', inspected: stranded.bindings.length }; } catch (err) { this.log.warn( createOperationContext('system'), `RS heal sweep for "${localCgId}" failed: ${err instanceof Error ? err.message : String(err)}`, ); + // A scheduler rejection means this maintenance sweep has lost admission. + // Stop immediately; the periodic reconciler will retry on its next tick + // instead of flooding the remaining backlog into the queue. + if (isStoreSchedulerBusyError(err)) { + return { status: 'deferred', reason: 'store-busy' }; + } + return { status: 'skipped', reason: 'failed' }; } } diff --git a/packages/agent/test/core-fills-gap.test.ts b/packages/agent/test/core-fills-gap.test.ts index 56133766fe..eb815271a9 100644 --- a/packages/agent/test/core-fills-gap.test.ts +++ b/packages/agent/test/core-fills-gap.test.ts @@ -4715,7 +4715,7 @@ describe('Phase D — reconcile gate + core-fill telemetry', () => { expect(receivedSignal).toBe(controller.signal); }); - it('abandons a same-object context-graph rebind that lands during stranded-KC repair', async () => { + it('keeps the main VM slice ahead of repair and abandons a rebind during repair', async () => { const chain = new MockChainAdapter(); agent = await DKGAgent.create({ name: 'CoreFillHealBindingFence', chainAdapter: chain }); stubNode(agent); @@ -4744,7 +4744,11 @@ describe('Phase D — reconcile gate + core-fill telemetry', () => { releaseHeal(); await expect(reconcile).rejects.toMatchObject({ name: 'VmReconcileQueueClosedError' }); - expect(getCount).not.toHaveBeenCalled(); + // The primary reconcile must complete its useful bounded slice before the + // best-effort repair begins; the post-repair binding fence still rejects a + // stale continuation after the same subscription object is rebound. + expect(getCount).toHaveBeenCalledOnce(); + expect(getCount).toHaveBeenCalledWith(321n); expect(sub.lastReconciledOrdinal).toBe(0); }); diff --git a/packages/agent/test/rs-heal-stranded-kc-decorated.test.ts b/packages/agent/test/rs-heal-stranded-kc-decorated.test.ts index 4838dd39bc..f913e46fe5 100644 --- a/packages/agent/test/rs-heal-stranded-kc-decorated.test.ts +++ b/packages/agent/test/rs-heal-stranded-kc-decorated.test.ts @@ -302,12 +302,35 @@ describe('healStrandedScopedKCs — through the production store decorator stack // heal sends through the top of the production stack and assert each INSERT // declares its scoped target graph + source tag. const updateCalls: Array<{ sparql: string; options?: { source?: string; touchedGraphs?: readonly string[] } }> = []; + const operationOptions: Array<{ method: string; options?: QueryOptions }> = []; const capturing = new Proxy(store, { get(target, prop, receiver) { + if (prop === 'query') { + const orig = Reflect.get(target, prop, receiver) as TripleStore['query']; + return (sparql: string, options?: QueryOptions) => { + operationOptions.push({ method: 'query', options }); + return orig.call(target, sparql, options); + }; + } + if (prop === 'insert') { + const orig = Reflect.get(target, prop, receiver) as TripleStore['insert']; + return (quads: Quad[], options?: QueryOptions) => { + operationOptions.push({ method: 'insert', options }); + return orig.call(target, quads, options); + }; + } + if (prop === 'deleteByPattern') { + const orig = Reflect.get(target, prop, receiver) as TripleStore['deleteByPattern']; + return (pattern: Partial, options?: QueryOptions) => { + operationOptions.push({ method: 'deleteByPattern', options }); + return orig.call(target, pattern, options); + }; + } if (prop === 'update') { const orig = Reflect.get(target, prop, receiver) as NonNullable; return (sparql: string, options?: { source?: string; touchedGraphs?: readonly string[] }) => { updateCalls.push({ sparql, options }); + operationOptions.push({ method: 'update', options }); return orig.call(target, sparql, options); }; } @@ -316,7 +339,11 @@ describe('healStrandedScopedKCs — through the production store decorator stack }) as TripleStore; await SwmHostModeMethods.prototype.healStrandedScopedKCs.call( - { store: capturing, log: { info: () => undefined, warn: () => undefined, error: () => undefined } } as never, + { + store: capturing, + rsHealCursorByCg: new Map(), + log: { info: () => undefined, warn: () => undefined, error: () => undefined }, + } as never, TEST_CG, { subscribed: true, synced: true, onChainId: TEST_ONCHAIN } as never, ); @@ -325,8 +352,19 @@ describe('healStrandedScopedKCs — through the production store decorator stack const scopedMeta = contextGraphMetaUri(TEST_CG, TEST_ONCHAIN); const dataInsert = updateCalls.find((c) => /INSERT/i.test(c.sparql) && c.sparql.includes(scopedData)); const metaInsert = updateCalls.find((c) => /INSERT/i.test(c.sparql) && c.sparql.includes(scopedMeta)); - expect(dataInsert?.options).toMatchObject({ source: 'agent.swm.rsHeal.materialize', touchedGraphs: [scopedData] }); - expect(metaInsert?.options).toMatchObject({ source: 'agent.swm.rsHeal.materialize', touchedGraphs: [scopedMeta] }); + expect(dataInsert?.options).toMatchObject({ + priority: 'background', + source: 'agent.swm.rsHeal.materialize', + touchedGraphs: [scopedData], + }); + expect(metaInsert?.options).toMatchObject({ + priority: 'background', + source: 'agent.swm.rsHeal.materialize', + touchedGraphs: [scopedMeta], + }); + expect(operationOptions.length).toBeGreaterThan(0); + expect(operationOptions.every(({ options }) => options?.priority === 'background')).toBe(true); + expect(operationOptions.every(({ options }) => options?.source?.startsWith('agent.swm.rsHeal.'))).toBe(true); }); it('labels RS-heal reads by caller operation through the decorator stack', async () => { @@ -358,10 +396,12 @@ describe('healStrandedScopedKCs — through the production store decorator stack expect(new Set(querySources.filter((source) => source?.startsWith('agent.swm.rsHeal.')))) .toEqual(new Set([ - 'agent.swm.rsHeal.findLegacyOnly', - 'agent.swm.rsHeal.listLegacyOnly', - 'agent.swm.rsHeal.readRoots', - 'agent.swm.rsHeal.checkRootData', + 'agent.swm.rsHeal.guard', + 'agent.swm.rsHeal.enumerate', + 'agent.swm.rsHeal.version.readLegacy', + 'agent.swm.rsHeal.version.checkScoped', + 'agent.swm.rsHeal.roots', + 'agent.swm.rsHeal.rootPresent', ])); }); diff --git a/packages/agent/test/rs-heal-stranded-kc.test.ts b/packages/agent/test/rs-heal-stranded-kc.test.ts index 1e3873bf84..baa2275964 100644 --- a/packages/agent/test/rs-heal-stranded-kc.test.ts +++ b/packages/agent/test/rs-heal-stranded-kc.test.ts @@ -37,7 +37,13 @@ * control below is the honest substitute that proves byte-sensitivity. */ import { describe, it, expect, beforeEach } from 'vitest'; -import { OxigraphStore, type Quad } from '@origintrail-official/dkg-storage'; +import { + OxigraphStore, + StorePriorityScheduler, + StoreSchedulerBusyError, + type Quad, + type QueryOptions, +} from '@origintrail-official/dkg-storage'; import { V10MerkleTree, contextGraphDataUri, @@ -48,6 +54,7 @@ import { import { extractV10KCFromStore } from '@origintrail-official/dkg-random-sampling'; import { writeMaterializedVersion, readMaterializedVersion } from '@origintrail-official/dkg-publisher'; import { SwmHostModeMethods } from '../src/dkg-agent-swm-host.js'; +import { DKGAgentBase } from '../src/dkg-agent-base.js'; const DKG = 'http://dkg.io/ontology/'; const XSD = 'http://www.w3.org/2001/XMLSchema#'; @@ -100,6 +107,7 @@ async function seedOntology(store: OxigraphStore, localCgId: string, onChainId: function makeAgentLike(store: OxigraphStore): unknown { return { store, + rsHealCursorByCg: new Map(), log: { info: () => undefined, warn: () => undefined, error: () => undefined }, }; } @@ -218,6 +226,266 @@ describe('healStrandedScopedKCs — content-binding gate', () => { expect(legacyStill.type === 'boolean' && legacyStill.value).toBe(true); }); + it('bounds each sweep and resumes the remaining stranded KCs on the next pass', async () => { + const batchSize = DKGAgentBase.RS_HEAL_BATCH_SIZE; + const total = batchSize + 1; + const cg = 'bounded-heal-cg'; + const onChainId = '23'; + const legacyMeta = contextGraphMetaUri(cg); + const legacyData = contextGraphDataUri(cg); + const scopedMeta = contextGraphMetaUri(cg, onChainId); + const quads: Quad[] = []; + await seedOntology(store, cg, onChainId); + for (let i = 1; i <= total; i += 1) { + const suffix = String(i).padStart(4, '0'); + const ual = `did:dkg:hardhat:31337/0xbatch/${suffix}`; + const root = `urn:entity:bounded-${suffix}`; + quads.push( + { subject: ual, predicate: `${DKG}batchId`, object: `"${i}"^^<${XSD}integer>`, graph: legacyMeta }, + { subject: `${ual}/1`, predicate: `${DKG}partOf`, object: ual, graph: legacyMeta }, + { subject: `${ual}/1`, predicate: `${DKG}rootEntity`, object: root, graph: legacyMeta }, + { subject: root, predicate: 'urn:p:value', object: `"${i}"`, graph: legacyData }, + ); + } + await store.insert(quads); + const agentLike = makeAgentLike(store); + const run = () => SwmHostModeMethods.prototype.healStrandedScopedKCs.call( + agentLike as never, + cg, + { subscribed: true, synced: true, onChainId } as never, + ); + const healedCount = async (): Promise => { + const result = await store.query( + `SELECT (COUNT(DISTINCT ?ual) AS ?c) WHERE { GRAPH <${scopedMeta}> { ?ual <${DKG}batchId> ?b } }`, + ); + const raw = result.type === 'bindings' ? result.bindings[0]?.['c'] ?? '0' : '0'; + return Number(/^"?(\d+)/.exec(raw)?.[1] ?? 0); + }; + + await run(); + expect(await healedCount()).toBe(batchSize); + + await run(); + expect(await healedCount()).toBe(total); + + await expect(run()).resolves.toMatchObject({ status: 'skipped', reason: 'no-work' }); + }); + + it('advances past an unhealable full page, repairs a later KC, then wraps to retry gaps', async () => { + const batchSize = DKGAgentBase.RS_HEAL_BATCH_SIZE; + const cg = 'skipped-page-heal-cg'; + const onChainId = '24'; + const legacyMeta = contextGraphMetaUri(cg); + const legacyData = contextGraphDataUri(cg); + const scopedMeta = contextGraphMetaUri(cg, onChainId); + const quads: Quad[] = []; + await seedOntology(store, cg, onChainId); + + for (let i = 1; i <= batchSize; i += 1) { + const suffix = String(i).padStart(4, '0'); + const ual = `did:dkg:hardhat:31337/0xskip/${suffix}`; + const missingRoot = `urn:entity:missing-${suffix}`; + quads.push( + { subject: ual, predicate: `${DKG}batchId`, object: `"${i}"^^<${XSD}integer>`, graph: legacyMeta }, + { subject: `${ual}/1`, predicate: `${DKG}partOf`, object: ual, graph: legacyMeta }, + { subject: `${ual}/1`, predicate: `${DKG}rootEntity`, object: missingRoot, graph: legacyMeta }, + ); + } + + const laterUal = 'did:dkg:hardhat:31337/0xskip/zzzz'; + const laterRoot = 'urn:entity:later-healable'; + quads.push( + { subject: laterUal, predicate: `${DKG}batchId`, object: `"999"^^<${XSD}integer>`, graph: legacyMeta }, + { subject: `${laterUal}/1`, predicate: `${DKG}partOf`, object: laterUal, graph: legacyMeta }, + { subject: `${laterUal}/1`, predicate: `${DKG}rootEntity`, object: laterRoot, graph: legacyMeta }, + { subject: laterRoot, predicate: 'urn:p:value', object: '"healable"', graph: legacyData }, + ); + await store.insert(quads); + + const agentLike = makeAgentLike(store) as any; + const cursorKey = `${cg}\u0000${onChainId}`; + const run = () => SwmHostModeMethods.prototype.healStrandedScopedKCs.call( + agentLike, + cg, + { subscribed: true, synced: true, onChainId } as never, + ); + const laterMaterialized = async (): Promise => { + const result = await store.query( + `ASK { GRAPH <${scopedMeta}> { <${laterUal}> <${DKG}materializedVersion> ?version } }`, + ); + return result.type === 'boolean' && result.value; + }; + + await expect(run()).resolves.toMatchObject({ status: 'completed', inspected: batchSize }); + expect(await laterMaterialized()).toBe(false); + expect(agentLike.rsHealCursorByCg.get(cursorKey)).toBe( + `did:dkg:hardhat:31337/0xskip/${String(batchSize).padStart(4, '0')}`, + ); + + await expect(run()).resolves.toMatchObject({ status: 'completed', inspected: 1 }); + expect(await laterMaterialized()).toBe(true); + expect(agentLike.rsHealCursorByCg.has(cursorKey)).toBe(false); + + // The short later page wrapped the cursor. The next pass revisits the + // incomplete first page instead of forgetting it permanently. + await expect(run()).resolves.toMatchObject({ status: 'completed', inspected: batchSize }); + expect(agentLike.rsHealCursorByCg.get(cursorKey)).toBe( + `did:dkg:hardhat:31337/0xskip/${String(batchSize).padStart(4, '0')}`, + ); + }); + + it('stops the sweep immediately when the background scheduler rejects admission', async () => { + const options: QueryOptions[] = []; + let queryCalls = 0; + const busy = new StoreSchedulerBusyError( + 'queue_full', + 'background', + 'agent.swm.rsHeal.version.readLegacy', + ); + const fakeStore = { + update: async () => undefined, + query: async (_sparql: string, queryOptions?: QueryOptions) => { + queryCalls += 1; + options.push(queryOptions ?? {}); + if (queryCalls === 1) return { type: 'boolean', value: true } as const; + if (queryCalls === 2) { + return { + type: 'bindings', + bindings: [{ + ual: 'did:dkg:hardhat:31337/0xbusy/1', + b: `"1"^^<${XSD}integer>`, + }], + } as const; + } + throw busy; + }, + }; + const cursorKey = 'busy-cg\u000029'; + const cursorMap = new Map([[cursorKey, 'did:dkg:hardhat:31337/0xbefore/1']]); + const agentLike = { + store: fakeStore, + rsHealCursorByCg: cursorMap, + log: { info: () => undefined, warn: () => undefined, error: () => undefined }, + }; + + await expect(SwmHostModeMethods.prototype.healStrandedScopedKCs.call( + agentLike as never, + 'busy-cg', + { subscribed: true, synced: true, onChainId: '29' } as never, + )).resolves.toEqual({ status: 'deferred', reason: 'store-busy' }); + + expect(queryCalls).toBe(3); + expect(cursorMap.get(cursorKey)).toBe('did:dkg:hardhat:31337/0xbefore/1'); + expect(options).toHaveLength(3); + expect(options.every((entry) => entry.priority === 'background')).toBe(true); + expect(options.every((entry) => entry.source?.startsWith('agent.swm.rsHeal.'))).toBe(true); + }); + + it('cancels a queued legacy-version read through the scheduler before teardown completes', async () => { + const scheduler = new StorePriorityScheduler({ + maxConcurrent: 1, + ackReservedSlots: 0, + healthReservedSlots: 0, + backgroundReservedSlots: 0, + queueLimits: 2, + queueWaitTimeoutMs: 10_000, + }); + let releaseBlocker!: () => void; + const blocker = scheduler.run('background', 'test.blocker', async () => { + await new Promise((resolve) => { releaseBlocker = resolve; }); + }); + let legacyReadStarted = false; + const fakeStore = { + update: async () => undefined, + query: async (_sparql: string, options?: QueryOptions) => { + if (options?.source === 'agent.swm.rsHeal.guard') { + return { type: 'boolean', value: true } as const; + } + if (options?.source === 'agent.swm.rsHeal.enumerate') { + return { + type: 'bindings', + bindings: [{ + ual: 'did:dkg:hardhat:31337/0xabort/1', + b: `"1"^^<${XSD}integer>`, + }], + } as const; + } + if (options?.source === 'agent.swm.rsHeal.version.readLegacy') { + return scheduler.run('background', options.source, async () => { + legacyReadStarted = true; + return { type: 'bindings', bindings: [] } as const; + }, options.signal); + } + throw new Error(`unexpected store operation ${String(options?.source)}`); + }, + }; + const controller = new AbortController(); + const cursorMap = new Map(); + const heal = SwmHostModeMethods.prototype.healStrandedScopedKCs.call( + { + store: fakeStore, + rsHealCursorByCg: cursorMap, + log: { info: () => undefined, warn: () => undefined, error: () => undefined }, + } as never, + 'abort-cg', + { subscribed: true, synced: true, onChainId: '30' } as never, + () => !controller.signal.aborted, + controller.signal, + ); + + while (scheduler.snapshot.backgroundQueued === 0) { + await new Promise((resolve) => setTimeout(resolve, 0)); + } + controller.abort(new Error('lifecycle retired')); + await expect(heal).resolves.toEqual({ status: 'skipped', reason: 'not-current' }); + expect(scheduler.snapshot.backgroundQueued).toBe(0); + expect(legacyReadStarted).toBe(false); + expect(cursorMap.size).toBe(0); + + releaseBlocker(); + await blocker; + }); + + it('returns the primary VM result when the later RS-heal pass is deferred', async () => { + const order: string[] = []; + const agentLike = makeAgentLike(store) as any; + agentLike.subscribedContextGraphs = new Map([[ + TEST_CG, + { subscribed: true, synced: true, onChainId: TEST_ONCHAIN, lastReconciledOrdinal: 0 }, + ]]); + agentLike.contextGraphBindingGenerations = new Map(); + agentLike.reconcileCursors = new Map(); + agentLike.vmReconcilePhysicalRuns = new Set(); + agentLike.vmReconcileEnabled = () => true; + agentLike.chain = { + getContextGraphKCCount: async () => { + order.push('main-reconcile'); + return 0n; + }, + }; + agentLike.resolveVmReconcileTarget = SwmHostModeMethods.prototype.resolveVmReconcileTarget; + agentLike.createVmReconcileDeps = SwmHostModeMethods.prototype.createVmReconcileDeps; + agentLike.toContextGraphReconcileResult = SwmHostModeMethods.prototype.toContextGraphReconcileResult; + agentLike.emitVmReconcileTelemetry = () => undefined; + agentLike.healStrandedScopedKCs = async () => { + order.push('rs-heal-deferred'); + throw new StoreSchedulerBusyError( + 'queue_full', + 'background', + 'agent.swm.rsHeal.guard', + ); + }; + + const result = await SwmHostModeMethods.prototype.executeVmReconcileForCg.call( + agentLike, + TEST_CG, + 'manual', + ); + + expect(result).toMatchObject({ status: 'current', attempted: false }); + expect(order).toEqual(['main-reconcile', 'rs-heal-deferred']); + }); + it('repairs a stranded KC inside the admitted current-watermark reconcile path', async () => { await expect(extractV10KCFromStore(store, BigInt(TEST_ONCHAIN), KA_ID)).rejects.toBeTruthy(); diff --git a/packages/cli/src/daemon/http-utils.ts b/packages/cli/src/daemon/http-utils.ts index 1504db1309..7878b14c15 100644 --- a/packages/cli/src/daemon/http-utils.ts +++ b/packages/cli/src/daemon/http-utils.ts @@ -888,20 +888,22 @@ async function rescueWriteTargetWithoutStore( * - `accept` — fast-accept; the caller returns the candidate id. * - `rejectUnknown` — a definitive, store-backed deny of a non-bare id; the * caller returns 404. - * - `deferReject` — a definitive deny of a BARE id; continue to the list leg - * and reject only if that leg ALSO misses (a name may - * resolve differently there). - * - `unavailable` — the probe threw or degraded (`storeUnavailable`); this - * is the ONLY verdict that makes the both-legs-failed - * store-free rescue eligible. `errorMessage` feeds the 503. - * - `continueToList`— no probe, or a definitive miss with nothing to carry. + * - canonical outcomes are fully reduced to accept/reject/unavailable here; + * the HTTP resolver never re-interprets raw probe fields. + * - the three `list*` variants are mutually exclusive: ordinary list/name + * resolution, an authoritative exact deny deferred for bare-name suffix + * resolution, or an unavailable exact probe whose list failure may use + * the store-free rescue. */ type ExactPreflightDecision = | { kind: "accept" } | { kind: "rejectUnknown" } - | { kind: "deferReject" } - | { kind: "unavailable"; errorMessage: string } - | { kind: "continueToList" }; + | { kind: "rejectNonWritable" } + | { kind: "validationUnavailable"; errorMessage: string } + | { kind: "unavailableWithRescue"; errorMessage: string } + | { kind: "listFallback" } + | { kind: "listDeferredReject" } + | { kind: "listUnavailable"; errorMessage: string }; /** * Run the exact write-preflight probe and reduce it to one {@link @@ -925,13 +927,18 @@ async function evaluateExactWritePreflight( allowLocalExactFallback: boolean; }, ): Promise { - if (!agent.probeContextGraphWritePreflight) return { kind: "continueToList" }; const { callerAgentAddress, requireLocalWritable, isBareCandidateId, allowLocalExactFallback, } = opts; + if (!agent.probeContextGraphWritePreflight) { + // Compatibility for narrow providers that do not expose the exact point + // capability. Production agents do expose it; their canonical path below + // never enumerates the catalog. + return { kind: "listFallback" }; + } try { const probe = await agent.probeContextGraphWritePreflight(candidateId, { callerAgentAddress, @@ -951,10 +958,10 @@ async function evaluateExactWritePreflight( // fields to UNKNOWN. Deny-ish verdicts are NOT trustworthy from unknowns // (that would turn a store outage into a 400), so carry the store error // for the both-legs-failed 503 and leave the verdict to the list/rescue. - return { - kind: "unavailable", - errorMessage: probe.storeErrorMessage ?? "local store unavailable", - }; + const errorMessage = probe.storeErrorMessage ?? "local store unavailable"; + return isBareCandidateId + ? { kind: "listUnavailable", errorMessage } + : { kind: "unavailableWithRescue", errorMessage }; } // Store answered definitively — a deny-ish verdict is authoritative. Bare // ids defer (the list leg may resolve the name); qualified ids reject now. @@ -962,17 +969,35 @@ async function evaluateExactWritePreflight( exactProbeIsStaleSubscription(probe) || exactProbeIsAuthoritativeBearerDeny(probe, callerAgentAddress) ) { - return isBareCandidateId ? { kind: "deferReject" } : { kind: "rejectUnknown" }; + return isBareCandidateId + ? { kind: "listDeferredReject" } + : { kind: "rejectUnknown" }; + } + if (!isBareCandidateId) { + if (probe.exists === false) return { kind: "rejectUnknown" }; + if (probe.exists !== true) { + return { + kind: "validationUnavailable", + errorMessage: "exact context graph existence was incomplete", + }; + } + if (!exactProbeIsLocallyWritable(probe, requireLocalWritable)) { + return { kind: "rejectNonWritable" }; + } + return { + kind: "validationUnavailable", + errorMessage: "exact context graph metadata was incomplete", + }; } - return { kind: "continueToList" }; + return { kind: "listFallback" }; } catch (err) { // The exact probe could not answer local existence at all (store down / read // broke). The ONLY degraded case that makes the both-legs-failed rescue // eligible: there is no authoritative local-miss verdict to override. - return { - kind: "unavailable", - errorMessage: err instanceof Error ? err.message : String(err), - }; + const errorMessage = err instanceof Error ? err.message : String(err); + return isBareCandidateId + ? { kind: "listUnavailable", errorMessage } + : { kind: "unavailableWithRescue", errorMessage }; } } @@ -1052,15 +1077,46 @@ export async function resolveRequiredWriteContextGraphId( if (exactDecision.kind === "rejectUnknown") { return rejectUnknownContextGraph(res, raw); } + if (exactDecision.kind === "rejectNonWritable") { + return rejectKnownNonWritableContextGraph(res, raw); + } + if (exactDecision.kind === "validationUnavailable") { + return contextGraphValidationUnavailable(res, exactDecision.errorMessage); + } + // Qualified ids are already canonical and unambiguous. Never turn their + // point-validation into a full ONTOLOGY/AGENTS catalog scan: under store + // pressure that O(catalog) fallback was the operation that timed out and + // blocked an otherwise unrelated publish (#2066). Bare names still need the + // list leg below for suffix resolution and ambiguity reporting. + if (exactDecision.kind === "unavailableWithRescue") { + const message = `exact preflight failed: ${exactDecision.errorMessage}`; + if (await rescueWriteTargetWithoutStore( + agent, + candidateId, + message, + opts.chainRescueTimeoutMs, + )) { + return candidateId; + } + return contextGraphValidationUnavailable(res, message); + } + if ( + exactDecision.kind !== "listFallback" && + exactDecision.kind !== "listDeferredReject" && + exactDecision.kind !== "listUnavailable" + ) { + return contextGraphValidationUnavailable(res, 'exact context graph preflight was inconclusive'); + } // Immutable carry-forward for the list leg. Track B (rescue gating): the // store-free on-chain rescue may run ONLY on an `unavailable` verdict — the // exact probe THREW or degraded (`storeUnavailable`) and so has no // authoritative local-miss to override. A definitive local miss (any other // kind) keeps the fail-closed 503 if the list leg then fails. - const deferredExactProbeReject = exactDecision.kind === "deferReject"; - const exactProbeUnavailable = exactDecision.kind === "unavailable"; - const exactProbeErrorMessage = - exactDecision.kind === "unavailable" ? exactDecision.errorMessage : null; + const deferredExactProbeReject = exactDecision.kind === "listDeferredReject"; + const exactProbeUnavailable = exactDecision.kind === "listUnavailable"; + const exactProbeErrorMessage = exactDecision.kind === "listUnavailable" + ? exactDecision.errorMessage + : null; let contextGraphs: ExistingContextGraphRow[]; try { diff --git a/packages/cli/test/write-preflight-resilience.test.ts b/packages/cli/test/write-preflight-resilience.test.ts index 89e271248b..91b1686f7f 100644 --- a/packages/cli/test/write-preflight-resilience.test.ts +++ b/packages/cli/test/write-preflight-resilience.test.ts @@ -38,11 +38,13 @@ import { WRITE_PREFLIGHT_CHAIN_RESCUE_TIMEOUT_MS, } from '../src/daemon/http-utils.js'; import { ContextGraphResolveMethods } from '../../agent/src/dkg-agent-cg-resolve.js'; +import type { ContextGraphWritePreflightProbe } from '@origintrail-official/dkg-agent'; import { OxigraphWorkerStore, createTripleStore, type TripleStore } from '@origintrail-official/dkg-storage'; import { DKG_ONTOLOGY, SYSTEM_CONTEXT_GRAPHS, contextGraphDataGraphUri, + contextGraphMetaGraphUri, } from '@origintrail-official/dkg-core'; const CG = 'resilience-cg'; @@ -556,6 +558,145 @@ describe('resolveRequiredWriteContextGraphId — healthy-store paths unchanged', expect(probe.exists).toBe(true); }); + it('accepts an authenticated caller of an explicitly public canonical graph without listing', async () => { + const seededCg = '0x1111111111111111111111111111111111111111/public-scoped-allow'; + const ontologyGraph = contextGraphDataGraphUri(SYSTEM_CONTEXT_GRAPHS.ONTOLOGY); + const cgUri = `did:dkg:context-graph:${seededCg}`; + await healthyStore.insert([ + { + subject: cgUri, + predicate: DKG_ONTOLOGY.RDF_TYPE, + object: DKG_ONTOLOGY.DKG_CONTEXT_GRAPH, + graph: ontologyGraph, + }, + { + subject: cgUri, + predicate: DKG_ONTOLOGY.DKG_ACCESS_POLICY, + object: '"public"', + graph: ontologyGraph, + }, + ]); + const harness = agentHarness( + healthyStore, + new Map([[seededCg, { subscribed: true, synced: true }]]), + ); + const { provider, listCalls } = providerFor(harness); + const { res, out } = captureRes(); + + const resolved = await resolveRequiredWriteContextGraphId(provider, seededCg, res, SCOPED); + + expect(resolved).toBe(seededCg); + expect(out.status).toBeUndefined(); + expect(listCalls).toEqual([]); + await expect(harness.probeContextGraphWritePreflight( + seededCg, + { callerAgentAddress: CALLER }, + )).resolves.toMatchObject({ + accessPolicy: 'public', + callerAuthorized: true, + }); + }); + + it('never falls back to the full catalog when a canonical exact probe is unavailable', async () => { + const qualified = '0x1111111111111111111111111111111111111111/qualified-cg'; + let listCalls = 0; + const provider = { + async listContextGraphs() { + listCalls += 1; + return []; + }, + async probeContextGraphWritePreflight() { + return { + storeAvailable: false, + storeUnavailable: true, + storeErrorMessage: 'Store scheduler queue wait timeout', + } as ContextGraphWritePreflightProbe; + }, + async validateWriteTargetDuringStoreOutage() { + return false; + }, + }; + const { res, out } = captureRes(); + + const resolved = await resolveRequiredWriteContextGraphId( + provider, + qualified, + res, + UNSCOPED, + ); + + expect(resolved).toBeNull(); + expect(listCalls).toBe(0); + expect(out.status).toBe(503); + expect(out.body).toMatchObject({ code: 'CONTEXT_GRAPH_VALIDATION_UNAVAILABLE' }); + expect(out.body.error).toContain('exact preflight failed: Store scheduler queue wait timeout'); + }); + + it('rescues a canonical id from positive on-chain proof without scanning the catalog', async () => { + const qualified = '0x1111111111111111111111111111111111111111/rescued-cg'; + let listCalls = 0; + const provider = { + async listContextGraphs() { + listCalls += 1; + return []; + }, + async probeContextGraphWritePreflight() { + return { + storeAvailable: false, + storeUnavailable: true, + storeErrorMessage: 'Store scheduler queue wait timeout', + } as ContextGraphWritePreflightProbe; + }, + async validateWriteTargetDuringStoreOutage() { + return true; + }, + }; + const { res, out } = captureRes(); + + const resolved = await resolveRequiredWriteContextGraphId( + provider, + qualified, + res, + UNSCOPED, + ); + + expect(resolved).toBe(qualified); + expect(listCalls).toBe(0); + expect(out.status).toBeUndefined(); + }); + + it('rejects an unknown canonical id from exact evidence without scanning the catalog', async () => { + const qualified = '0x1111111111111111111111111111111111111111/missing-cg'; + let listCalls = 0; + const provider = { + async listContextGraphs() { + listCalls += 1; + return []; + }, + async probeContextGraphWritePreflight() { + return { + storeAvailable: true, + exists: false, + hasLocalContent: false, + declarationFound: false, + } as ContextGraphWritePreflightProbe; + }, + }; + const { res, out } = captureRes(); + + const resolved = await resolveRequiredWriteContextGraphId( + provider, + qualified, + res, + UNSCOPED, + ); + + expect(resolved).toBeNull(); + expect(listCalls).toBe(0); + expect(out.status).toBe(400); + expect(out.body).toMatchObject({ code: 'CONTEXT_GRAPH_NOT_FOUND' }); + }); + it('fast-accepts an exact private synced CG for a trusted local caller without listing every CG', async () => { const seededCg = '0x1111111111111111111111111111111111111111/private-fast-path'; const ontologyGraph = contextGraphDataGraphUri(SYSTEM_CONTEXT_GRAPHS.ONTOLOGY); @@ -636,6 +777,361 @@ describe('resolveRequiredWriteContextGraphId — healthy-store paths unchanged', expect(listCalls).toEqual([]); }); + it('accepts an explicitly private canonical graph for an authorized scoped caller without listing', async () => { + const seededCg = '0x1111111111111111111111111111111111111111/private-scoped-allow'; + const ontologyGraph = contextGraphDataGraphUri(SYSTEM_CONTEXT_GRAPHS.ONTOLOGY); + const cgUri = `did:dkg:context-graph:${seededCg}`; + await healthyStore.insert([ + { + subject: cgUri, + predicate: DKG_ONTOLOGY.RDF_TYPE, + object: DKG_ONTOLOGY.DKG_CONTEXT_GRAPH, + graph: ontologyGraph, + }, + { + subject: cgUri, + predicate: DKG_ONTOLOGY.DKG_ACCESS_POLICY, + object: '"private"', + graph: ontologyGraph, + }, + ]); + const harness = agentHarness( + healthyStore, + new Map([[seededCg, { subscribed: true, synced: true }]]), + ); + harness.callerIsAllowlistedAgentParticipant = async () => true; + const { provider, listCalls } = providerFor(harness); + const { res, out } = captureRes(); + + const resolved = await resolveRequiredWriteContextGraphId(provider, seededCg, res, SCOPED); + + expect(resolved).toBe(seededCg); + expect(out.status).toBeUndefined(); + expect(listCalls).toEqual([]); + }); + + it('authorizes a canonical gate-only private graph through the bounded exact probe', async () => { + const seededCg = '0x1111111111111111111111111111111111111111/gate-only-authorized'; + const ontologyGraph = contextGraphDataGraphUri(SYSTEM_CONTEXT_GRAPHS.ONTOLOGY); + const metaGraph = contextGraphMetaGraphUri(seededCg); + const cgUri = `did:dkg:context-graph:${seededCg}`; + await healthyStore.insert([ + { + subject: cgUri, + predicate: DKG_ONTOLOGY.RDF_TYPE, + object: DKG_ONTOLOGY.DKG_CONTEXT_GRAPH, + graph: ontologyGraph, + }, + { + subject: cgUri, + predicate: DKG_ONTOLOGY.DKG_ALLOWED_AGENT, + object: `"${CALLER}"`, + graph: metaGraph, + }, + ]); + const harness = agentHarness( + healthyStore, + new Map([[seededCg, { subscribed: true, synced: true }]]), + ); + harness.callerIsAllowlistedAgentParticipant = async () => true; + const { provider, listCalls } = providerFor(harness); + const { res, out } = captureRes(); + + const resolved = await resolveRequiredWriteContextGraphId( + provider, + seededCg, + res, + SCOPED, + ); + + expect(resolved).toBe(seededCg); + expect(out.status).toBeUndefined(); + expect(listCalls).toEqual([]); + await expect(harness.probeContextGraphWritePreflight( + seededCg, + { callerAgentAddress: CALLER }, + )).resolves.toMatchObject({ + storeAvailable: true, + exists: true, + accessPolicy: 'private', + callerAuthorized: true, + }); + }); + + it('denies an unauthorized caller of a canonical gate-only private graph without a catalog scan', async () => { + const seededCg = '0x1111111111111111111111111111111111111111/gate-only-denied'; + const ontologyGraph = contextGraphDataGraphUri(SYSTEM_CONTEXT_GRAPHS.ONTOLOGY); + const metaGraph = contextGraphMetaGraphUri(seededCg); + const cgUri = `did:dkg:context-graph:${seededCg}`; + await healthyStore.insert([ + { + subject: cgUri, + predicate: DKG_ONTOLOGY.RDF_TYPE, + object: DKG_ONTOLOGY.DKG_CONTEXT_GRAPH, + graph: ontologyGraph, + }, + { + subject: cgUri, + predicate: DKG_ONTOLOGY.DKG_ALLOWED_AGENT, + object: '"0x3333333333333333333333333333333333333333"', + graph: metaGraph, + }, + ]); + const harness = agentHarness( + healthyStore, + new Map([[seededCg, { subscribed: true, synced: true }]]), + ); + const { provider, listCalls } = providerFor(harness); + const { res, out } = captureRes(); + + const resolved = await resolveRequiredWriteContextGraphId( + provider, + seededCg, + res, + SCOPED, + ); + + expect(resolved).toBeNull(); + expect(listCalls).toEqual([]); + expect(out.status).toBe(400); + expect(out.body).toMatchObject({ code: 'CONTEXT_GRAPH_NOT_FOUND' }); + }); + + it('keeps an explicit public policy public when a legacy gate lives in another metadata source', async () => { + const seededCg = '0x1111111111111111111111111111111111111111/public-cross-source-gate'; + const ontologyGraph = contextGraphDataGraphUri(SYSTEM_CONTEXT_GRAPHS.ONTOLOGY); + const metaGraph = contextGraphMetaGraphUri(seededCg); + const cgUri = `did:dkg:context-graph:${seededCg}`; + await healthyStore.insert([ + { + subject: cgUri, + predicate: DKG_ONTOLOGY.RDF_TYPE, + object: DKG_ONTOLOGY.DKG_CONTEXT_GRAPH, + graph: ontologyGraph, + }, + { + subject: cgUri, + predicate: DKG_ONTOLOGY.DKG_ACCESS_POLICY, + object: '"public"', + graph: ontologyGraph, + }, + { + subject: cgUri, + predicate: DKG_ONTOLOGY.DKG_ALLOWED_AGENT, + object: '"0x3333333333333333333333333333333333333333"', + graph: metaGraph, + }, + ]); + const harness = agentHarness( + healthyStore, + new Map([[seededCg, { subscribed: true, synced: true }]]), + ); + harness.callerIsAllowlistedAgentParticipant = async () => { + throw new Error('public policy must not consult the allowlist'); + }; + const { provider, listCalls } = providerFor(harness); + const { res, out } = captureRes(); + + const resolved = await resolveRequiredWriteContextGraphId(provider, seededCg, res, SCOPED); + + expect(resolved).toBe(seededCg); + expect(out.status).toBeUndefined(); + expect(listCalls).toEqual([]); + await expect(harness.probeContextGraphWritePreflight( + seededCg, + { callerAgentAddress: CALLER }, + )).resolves.toMatchObject({ + declarationFound: true, + accessPolicy: 'public', + callerAuthorized: true, + }); + }); + + it('applies the private-policy ratchet across sources and denies an unauthorized caller', async () => { + const seededCg = '0x1111111111111111111111111111111111111111/private-ratchet-cross-source'; + const ontologyGraph = contextGraphDataGraphUri(SYSTEM_CONTEXT_GRAPHS.ONTOLOGY); + const metaGraph = contextGraphMetaGraphUri(seededCg); + const cgUri = `did:dkg:context-graph:${seededCg}`; + await healthyStore.insert([ + { + subject: cgUri, + predicate: DKG_ONTOLOGY.RDF_TYPE, + object: DKG_ONTOLOGY.DKG_CONTEXT_GRAPH, + graph: ontologyGraph, + }, + { + subject: cgUri, + predicate: DKG_ONTOLOGY.DKG_ACCESS_POLICY, + object: '"public"', + graph: ontologyGraph, + }, + { + subject: cgUri, + predicate: DKG_ONTOLOGY.DKG_ACCESS_POLICY, + object: '"private"', + graph: metaGraph, + }, + ]); + const harness = agentHarness( + healthyStore, + new Map([[seededCg, { subscribed: true, synced: true }]]), + ); + const { provider, listCalls } = providerFor(harness); + const { res, out } = captureRes(); + + const resolved = await resolveRequiredWriteContextGraphId(provider, seededCg, res, SCOPED); + + expect(resolved).toBeNull(); + expect(listCalls).toEqual([]); + expect(out.status).toBe(400); + expect(out.body).toMatchObject({ code: 'CONTEXT_GRAPH_NOT_FOUND' }); + await expect(harness.probeContextGraphWritePreflight( + seededCg, + { callerAgentAddress: CALLER }, + )).resolves.toMatchObject({ + declarationFound: true, + accessPolicy: 'private', + callerAuthorized: false, + }); + }); + + it('authorizes a cross-source curator when only the declaration source repeats rdf:type', async () => { + const seededCg = '0x1111111111111111111111111111111111111111/curator-cross-source'; + const ontologyGraph = contextGraphDataGraphUri(SYSTEM_CONTEXT_GRAPHS.ONTOLOGY); + const metaGraph = contextGraphMetaGraphUri(seededCg); + const cgUri = `did:dkg:context-graph:${seededCg}`; + await healthyStore.insert([ + { + subject: cgUri, + predicate: DKG_ONTOLOGY.RDF_TYPE, + object: DKG_ONTOLOGY.DKG_CONTEXT_GRAPH, + graph: ontologyGraph, + }, + { + subject: cgUri, + predicate: DKG_ONTOLOGY.DKG_ACCESS_POLICY, + object: '"private"', + graph: metaGraph, + }, + { + subject: cgUri, + predicate: DKG_ONTOLOGY.DKG_CURATOR, + object: `did:dkg:agent:${CALLER}`, + graph: metaGraph, + }, + ]); + const harness = agentHarness( + healthyStore, + new Map([[seededCg, { subscribed: true, synced: true }]]), + ); + harness.curatorDidMatchesChecksumAgent = (curator: string, checksum: string) => + curator.includes(CALLER) && checksum === CALLER; + harness.callerIsAllowlistedAgentParticipant = async () => { + throw new Error('curator match must short-circuit the allowlist'); + }; + const { provider, listCalls } = providerFor(harness); + const { res, out } = captureRes(); + + const resolved = await resolveRequiredWriteContextGraphId(provider, seededCg, res, SCOPED); + + expect(resolved).toBe(seededCg); + expect(out.status).toBeUndefined(); + expect(listCalls).toEqual([]); + await expect(harness.probeContextGraphWritePreflight( + seededCg, + { callerAgentAddress: CALLER }, + )).resolves.toMatchObject({ + declarationFound: true, + accessPolicy: 'private', + callerAuthorized: true, + }); + }); + + it('fails closed for healthy but incomplete canonical metadata without a catalog scan', async () => { + const qualified = '0x1111111111111111111111111111111111111111/incomplete-metadata'; + let listCalls = 0; + const provider = { + async listContextGraphs() { + listCalls += 1; + return []; + }, + async probeContextGraphWritePreflight() { + return { + storeAvailable: true, + exists: true, + hasLocalContent: true, + declarationFound: true, + } as ContextGraphWritePreflightProbe; + }, + }; + const { res, out } = captureRes(); + + const resolved = await resolveRequiredWriteContextGraphId(provider, qualified, res, SCOPED); + + expect(resolved).toBeNull(); + expect(listCalls).toBe(0); + expect(out.status).toBe(503); + expect(out.body).toMatchObject({ code: 'CONTEXT_GRAPH_VALIDATION_UNAVAILABLE' }); + expect(out.body.error).toContain('exact context graph metadata was incomplete'); + }); + + it('keeps a throwing canonical exact probe on the bounded rescue path without a catalog scan', async () => { + const qualified = '0x1111111111111111111111111111111111111111/throwing-probe'; + let listCalls = 0; + let rescueCalls = 0; + const provider = { + async listContextGraphs() { + listCalls += 1; + return []; + }, + async probeContextGraphWritePreflight() { + throw new Error('Store scheduler queue wait timeout'); + }, + async validateWriteTargetDuringStoreOutage() { + rescueCalls += 1; + return false; + }, + }; + const { res, out } = captureRes(); + + const resolved = await resolveRequiredWriteContextGraphId(provider, qualified, res, UNSCOPED); + + expect(resolved).toBeNull(); + expect(listCalls).toBe(0); + expect(rescueCalls).toBe(1); + expect(out.status).toBe(503); + expect(out.body).toMatchObject({ code: 'CONTEXT_GRAPH_VALIDATION_UNAVAILABLE' }); + expect(out.body.error).toContain('exact preflight failed: Store scheduler queue wait timeout'); + }); + + it('rejects a known but locally non-writable canonical graph without a catalog scan', async () => { + const qualified = '0x1111111111111111111111111111111111111111/non-writable'; + let listCalls = 0; + const provider = { + async listContextGraphs() { + listCalls += 1; + return []; + }, + async probeContextGraphWritePreflight() { + return { + storeAvailable: true, + exists: true, + hasLocalContent: false, + declarationFound: true, + accessPolicy: 'public', + } as ContextGraphWritePreflightProbe; + }, + }; + const { res, out } = captureRes(); + + const resolved = await resolveRequiredWriteContextGraphId(provider, qualified, res, UNSCOPED); + + expect(resolved).toBeNull(); + expect(listCalls).toBe(0); + expect(out.status).toBe(400); + expect(out.body).toMatchObject({ code: 'CONTEXT_GRAPH_NOT_WRITABLE' }); + }); + it('(e) a storeUnavailable exact probe does not poison a healthy list leg (list evidence still accepts)', async () => { // Partial outage: the probe's store reads fail, but the composite list // succeeds (e.g. store recovered between the legs). Pre-fix the probe diff --git a/packages/publisher/src/metadata.ts b/packages/publisher/src/metadata.ts index bc3d225072..857462c49f 100644 --- a/packages/publisher/src/metadata.ts +++ b/packages/publisher/src/metadata.ts @@ -1230,11 +1230,13 @@ export async function readMaterializedVersion( store: TripleStore, metaGraph: string, ual: string, + options: QueryOptions = {}, ): Promise { assertSafeGraphIriForSparql(metaGraph); assertSafeGraphIriForSparql(ual); const res = await store.query( `SELECT ?v WHERE { GRAPH <${metaGraph}> { <${ual}> <${MATERIALIZED_VERSION_PRED}> ?v } } LIMIT 1`, + options, ); if (res.type !== 'bindings' || res.bindings.length === 0) return null; return parseMaterializedVersion(res.bindings[0]['v']); @@ -1251,12 +1253,14 @@ export async function shouldApplyMaterialization( ual: string, incoming: MaterializedVersion, incomingAssertionVersion?: bigint, + options: QueryOptions = {}, ): Promise { if (incomingAssertionVersion !== undefined) { assertSafeGraphIriForSparql(metaGraph); assertSafeGraphIriForSparql(ual); const assertionVersions = await store.query( `SELECT ?v WHERE { GRAPH <${metaGraph}> { <${ual}> <${ASSERTION_VERSION_PRED}> ?v } }`, + options, ); if (assertionVersions.type === 'bindings') { for (const row of assertionVersions.bindings) { @@ -1269,7 +1273,7 @@ export async function shouldApplyMaterialization( } } } - const current = await readMaterializedVersion(store, metaGraph, ual); + const current = await readMaterializedVersion(store, metaGraph, ual, options); if (!current) return true; return compareMaterializedVersion(incoming, current) >= 0; } @@ -1279,11 +1283,15 @@ export async function writeMaterializedVersion( metaGraph: string, ual: string, version: MaterializedVersion, + options: QueryOptions = {}, ): Promise { assertSafeGraphIriForSparql(metaGraph); assertSafeGraphIriForSparql(ual); - await store.deleteByPattern({ graph: metaGraph, subject: ual, predicate: MATERIALIZED_VERSION_PRED }); - await store.insert([materializedVersionQuad(metaGraph, ual, version)]); + await store.deleteByPattern( + { graph: metaGraph, subject: ual, predicate: MATERIALIZED_VERSION_PRED }, + options, + ); + await store.insert([materializedVersionQuad(metaGraph, ual, version)], options); } export function materializedVersionQuad(