diff --git a/packages/agent/src/dkg-agent-base.ts b/packages/agent/src/dkg-agent-base.ts index 072387a896..e764312b37 100644 --- a/packages/agent/src/dkg-agent-base.ts +++ b/packages/agent/src/dkg-agent-base.ts @@ -110,7 +110,7 @@ import { pickNetworkTunables, isSparqlUpdateOperation, } from '@origintrail-official/dkg-core'; -import { GraphManager, PrivateContentStore, SystemRecordLaneForwarderV1, createTripleStore, isExternalBackend, structuredMutationMightMutate, structuredMutationTouchedGraphs, type TripleStore, type TripleStoreConfig, type Quad, type LargeLiteralStorageConfig, type QueryOptions } from '@origintrail-official/dkg-storage'; +import { GraphManager, PrivateContentStore, SystemRecordLaneForwarderV1, captureStructuredMutationEffects, createTripleStore, isExternalBackend, type TripleStore, type TripleStoreConfig, type Quad, type LargeLiteralStorageConfig, type QueryOptions } from '@origintrail-official/dkg-storage'; import { emptyRpcUsageWindow, EVMChainAdapter, NoChainAdapter, enrichEvmError, buildKnowledgeAssetUal, type EVMAdapterConfig, type ChainAdapter, type CreateContextGraphParams, type CreateOnChainContextGraphParams, type CreateOnChainContextGraphResult, type TxResult, type V10PublishingConvictionAccountInfo, type RpcUsageWindow } from '@origintrail-official/dkg-chain'; import { DKGPublisher, PublishHandler, SharedMemoryHandler, UpdateHandler, ChainEventPoller, AccessHandler, AccessClient, @@ -578,14 +578,13 @@ export function createListContextGraphsCacheInvalidatingStore( : undefined, structuredMutation: innerStore.structuredMutation ? (mutation, options) => { - // Capture scope before the first await so caller-side mutation cannot - // redirect cache invalidation after the backend has committed. - const targetGraphs = [...structuredMutationTouchedGraphs(mutation)]; - const mightMutate = structuredMutationMightMutate(mutation); + const effects = captureStructuredMutationEffects(mutation); return invalidateAfterMutation( () => innerStore.structuredMutation!(mutation, options), - () => mightMutate, - () => targetGraphs.forEach((graph) => markProjectionDirty?.(undefined, graph)), + () => effects !== undefined, + () => effects?.touchedGraphs.forEach( + (graph) => markProjectionDirty?.(undefined, graph), + ), ); } : undefined, diff --git a/packages/agent/test/replace-subject-agent-wrapper.test.ts b/packages/agent/test/replace-subject-agent-wrapper.test.ts index 9d5d56824a..03a2da2bad 100644 --- a/packages/agent/test/replace-subject-agent-wrapper.test.ts +++ b/packages/agent/test/replace-subject-agent-wrapper.test.ts @@ -15,7 +15,7 @@ import { mkdtemp, rm } from 'node:fs/promises'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; -import { afterEach, describe, expect, it } from 'vitest'; +import { afterEach, describe, expect, it, vi } from 'vitest'; import { CHANGELOG_GRAPH, OxigraphStore, @@ -150,4 +150,76 @@ describe('#1863 replaceSubject through the agent store wrapper', () => { proj.markDirtyForGraph('urn:dkg:publisher:control-plane'); expect(entries.has('urn:dkg:publisher:control-plane')).toBe(false); }); + + it('invalidates structured mutation effects after success without decoding them in Agent', async () => { + let release!: () => void; + const inFlight = new Promise((resolve) => { release = resolve; }); + const options = { source: 'agent.test.structured-mutation-effects' }; + let inner!: TripleStore; + const structuredMutation = vi.fn(function ( + this: TripleStore, + _mutation: unknown, + receivedOptions: unknown, + ) { + expect(this).toBe(inner); + expect(receivedOptions).toBe(options); + return inFlight; + }); + inner = { structuredMutation } as unknown as TripleStore; + const invalidate = vi.fn(); + const markProjectionDirty = vi.fn(); + const store = createListContextGraphsCacheInvalidatingStore( + inner, + invalidate, + markProjectionDirty, + ); + const mutation = { + kind: 'copy-subject-projection' as const, + input: { + sourceGraphUris: ['urn:test:source'], + targetGraphUri: 'urn:test:target', + roots: ['urn:test:root'], + descendantSuffix: '/', + excludedPredicates: [], + }, + }; + + const pending = store.structuredMutation!(mutation, options); + mutation.input.targetGraphUri = 'urn:test:redirected'; + expect(invalidate).not.toHaveBeenCalled(); + release(); + await pending; + + expect(invalidate).toHaveBeenCalledOnce(); + expect(markProjectionDirty).toHaveBeenCalledOnce(); + expect(markProjectionDirty).toHaveBeenCalledWith(undefined, 'urn:test:target'); + }); + + it('does not invalidate structured mutation failures or structural no-ops', async () => { + const invalidate = vi.fn(); + const markProjectionDirty = vi.fn(); + const inner = { + structuredMutation: vi.fn(async () => undefined), + } as unknown as TripleStore; + const store = createListContextGraphsCacheInvalidatingStore( + inner, + invalidate, + markProjectionDirty, + ); + + await store.structuredMutation!({ + kind: 'delete-subjects', + input: { graphUri: 'urn:test:target', subjects: [] }, + }); + expect(invalidate).not.toHaveBeenCalled(); + expect(markProjectionDirty).not.toHaveBeenCalled(); + + inner.structuredMutation = vi.fn(async () => { throw new Error('commit failed'); }); + await expect(store.structuredMutation!({ + kind: 'delete-subjects', + input: { graphUri: 'urn:test:target', subjects: ['urn:test:subject'] }, + })).rejects.toThrow('commit failed'); + expect(invalidate).not.toHaveBeenCalled(); + expect(markProjectionDirty).not.toHaveBeenCalled(); + }); }); diff --git a/packages/storage/src/adapters/blazegraph.ts b/packages/storage/src/adapters/blazegraph.ts index 04a046f9a4..f05cbeb330 100644 --- a/packages/storage/src/adapters/blazegraph.ts +++ b/packages/storage/src/adapters/blazegraph.ts @@ -34,8 +34,8 @@ import { } from '../atomic-graph-replace.js'; import { buildStructuredMutationUpdate, + captureStructuredMutationEffects, normalizeStructuredMutation, - structuredMutationMightMutate, } from '../bounded-structured-mutation.js'; import { quadToNQuad } from '../bounded-rdf.js'; import { readResponseTextBounded } from '../http-response-limit.js'; @@ -417,6 +417,7 @@ export class BlazegraphStore implements TripleStore { options?: QueryOptions, ): Promise { const normalized = normalizeStructuredMutation(mutation); + const effects = captureStructuredMutationEffects(normalized); if (normalized.kind === 'replace-subject-predicates') { assertQuadLiteralsMutf8Safe([...normalized.input.replacementQuads], { maxBytes: JAVA_WRITE_UTF_MAX_BYTES, @@ -424,7 +425,7 @@ export class BlazegraphStore implements TripleStore { }); } const update = buildStructuredMutationUpdate(normalized); - if (!update || !structuredMutationMightMutate(normalized)) return; + if (!update || !effects) return; await this.sparqlUpdate( update, { ...options, source: options?.source ?? 'blazegraph.structuredMutation' }, diff --git a/packages/storage/src/adapters/oxigraph-worker.ts b/packages/storage/src/adapters/oxigraph-worker.ts index 23009634ed..e612ec95e3 100644 --- a/packages/storage/src/adapters/oxigraph-worker.ts +++ b/packages/storage/src/adapters/oxigraph-worker.ts @@ -13,9 +13,8 @@ import type { import { registerTripleStoreAdapter } from '../triple-store.js'; import { GraphWriteGenTracker } from '../graph-write-gen.js'; import { + captureStructuredMutationEffects, normalizeStructuredMutation, - structuredMutationMightMutate, - structuredMutationTouchedGraphs, } from '../bounded-structured-mutation.js'; /** @@ -698,10 +697,9 @@ export class OxigraphWorkerStore implements TripleStore { _options?: TripleStoreQueryOptions, ): Promise { const normalized = normalizeStructuredMutation(mutation); + const effects = captureStructuredMutationEffects(normalized); await this.call('structuredMutation', normalized); - if (structuredMutationMightMutate(normalized)) { - this.writeGen.recordGraphWrites(structuredMutationTouchedGraphs(normalized)); - } + if (effects) this.writeGen.recordGraphWrites(effects.touchedGraphs); } async query(sparql: string, options?: TripleStoreQueryOptions): Promise { return this.callWithTimeout(this.operationTimeoutMs, options?.signal, 'query', sparql); diff --git a/packages/storage/src/adapters/oxigraph.ts b/packages/storage/src/adapters/oxigraph.ts index 67729bb55c..986a29c327 100644 --- a/packages/storage/src/adapters/oxigraph.ts +++ b/packages/storage/src/adapters/oxigraph.ts @@ -28,9 +28,8 @@ import { } from '../atomic-graph-replace.js'; import { buildStructuredMutationUpdate, + captureStructuredMutationEffects, normalizeStructuredMutation, - structuredMutationMightMutate, - structuredMutationTouchedGraphs, } from '../bounded-structured-mutation.js'; import { quadsToNQuads } from '../bounded-rdf.js'; import { assertQuadLiteralsMutf8Safe, JAVA_WRITE_UTF_MAX_BYTES } from '@origintrail-official/dkg-core'; @@ -421,6 +420,7 @@ export class OxigraphStore implements TripleStore { _options?: TripleStoreQueryOptions, ): Promise { const normalized = normalizeStructuredMutation(mutation); + const effects = captureStructuredMutationEffects(normalized); if (normalized.kind === 'replace-subject-predicates') { assertQuadLiteralsMutf8Safe([...normalized.input.replacementQuads], { maxBytes: JAVA_WRITE_UTF_MAX_BYTES, @@ -428,10 +428,10 @@ export class OxigraphStore implements TripleStore { }); } const update = buildStructuredMutationUpdate(normalized); - if (!update || !structuredMutationMightMutate(normalized)) return; + if (!update || !effects) return; this.store.update(update); this.scheduleFlush(); - this.writeGen.recordGraphWrites(structuredMutationTouchedGraphs(normalized)); + this.writeGen.recordGraphWrites(effects.touchedGraphs); } async listGraphs(options?: TripleStoreQueryOptions): Promise { diff --git a/packages/storage/src/adapters/sparql-http.ts b/packages/storage/src/adapters/sparql-http.ts index 0d551595b9..cf5ed9f424 100644 --- a/packages/storage/src/adapters/sparql-http.ts +++ b/packages/storage/src/adapters/sparql-http.ts @@ -54,10 +54,9 @@ import { } from '../atomic-graph-replace.js'; import { buildStructuredMutationUpdate, + captureStructuredMutationEffects, normalizeStructuredMutation, structuredMutationGuardedGraphs, - structuredMutationMightMutate, - structuredMutationTouchedGraphs, } from '../bounded-structured-mutation.js'; import { assertNotReservedInternalGraphV1, @@ -1299,7 +1298,7 @@ export class SparqlHttpStore implements TripleStore { options?: QueryOptions, ): Promise { const normalized = normalizeStructuredMutation(mutation); - const touchedGraphs = structuredMutationTouchedGraphs(normalized); + const effects = captureStructuredMutationEffects(normalized); this.assertGenericMutationScope(structuredMutationGuardedGraphs(normalized), 'structuredMutation'); if (normalized.kind === 'replace-subject-predicates') { assertQuadLiteralsMutf8Safe([...normalized.input.replacementQuads], { @@ -1308,24 +1307,24 @@ export class SparqlHttpStore implements TripleStore { }); } const update = buildStructuredMutationUpdate(normalized); - if (!update || !structuredMutationMightMutate(normalized)) return; + if (!update || !effects) return; try { await this.postUpdate( update, { ...options, source: options?.source ?? 'sparql-http.structuredMutation' }, 'structuredMutation', - touchedGraphs, + effects.touchedGraphs, ); } catch (error) { // A remote endpoint may commit before its response is lost. Fail open for // cache coherence: invalidate graph enumeration and advance each affected // write generation even though the caller still receives the failure. this.invalidateListGraphsCache(); - this.writeGen.recordGraphWrites(touchedGraphs); + this.writeGen.recordGraphWrites(effects.touchedGraphs); throw error; } this.invalidateListGraphsCache(); - this.writeGen.recordGraphWrites(touchedGraphs); + this.writeGen.recordGraphWrites(effects.touchedGraphs); } async query(sparql: string, options?: SparqlHttpQueryOptions): Promise { diff --git a/packages/storage/src/bounded-structured-mutation.ts b/packages/storage/src/bounded-structured-mutation.ts index 93923c80aa..2dfea21334 100644 --- a/packages/storage/src/bounded-structured-mutation.ts +++ b/packages/storage/src/bounded-structured-mutation.ts @@ -680,3 +680,17 @@ export function structuredMutationTouchedGraphs(mutation: StructuredMutation): r export function structuredMutationMightMutate(mutation: StructuredMutation): boolean { return mutation.kind !== 'delete-subjects' || mutation.input.subjects.length > 0; } + +/** Immutable graph-scoped effects captured before a structured mutation is dispatched. */ +export interface StructuredMutationEffects { + readonly touchedGraphs: readonly string[]; +} + +/** Capture canonical effects without executing or probing a store capability. */ +export function captureStructuredMutationEffects( + mutation: StructuredMutation, +): StructuredMutationEffects | undefined { + if (!structuredMutationMightMutate(mutation)) return undefined; + const touchedGraphs = Object.freeze([...structuredMutationTouchedGraphs(mutation)]); + return Object.freeze({ touchedGraphs }); +} diff --git a/packages/storage/src/changelog-store.ts b/packages/storage/src/changelog-store.ts index cfcb26a5cf..8c920ed982 100644 --- a/packages/storage/src/changelog-store.ts +++ b/packages/storage/src/changelog-store.ts @@ -24,10 +24,9 @@ import { } from './store-chain-capability.js'; import type { SystemRecordLaneControllerV1 } from './system-record-materializer-v1.js'; import { + captureStructuredMutationEffects, normalizeStructuredMutation, structuredMutationGuardedGraphs, - structuredMutationMightMutate, - structuredMutationTouchedGraphs, } from './bounded-structured-mutation.js'; /** @@ -460,6 +459,7 @@ export class ChangelogStore implements TripleStore, ChangelogReader { async structuredMutation(mutation: StructuredMutation, options?: QueryOptions): Promise { const normalized = normalizeStructuredMutation(mutation); + const effects = captureStructuredMutationEffects(normalized); const operation = this.inner.structuredMutation; if (!operation) { throw new UnsupportedTripleStoreCapabilityError('structuredMutation', 'ChangelogStore'); @@ -477,8 +477,8 @@ export class ChangelogStore implements TripleStore, ChangelogReader { } throw error; } - if (structuredMutationMightMutate(normalized)) { - await this.markPostMutation(structuredMutationTouchedGraphs(normalized), options); + if (effects) { + await this.markPostMutation(effects.touchedGraphs, options); } }); } diff --git a/packages/storage/src/graph-set-index-store.ts b/packages/storage/src/graph-set-index-store.ts index c522f97a7a..2c5b249dd2 100644 --- a/packages/storage/src/graph-set-index-store.ts +++ b/packages/storage/src/graph-set-index-store.ts @@ -23,9 +23,8 @@ import { import { isAtomicGraphReplaceStagingGraph } from './atomic-graph-replace.js'; import { ManagedOxigraphBackendUnownedError } from './managed-oxigraph-ownership-v1-internal.js'; import { + captureStructuredMutationEffects, normalizeStructuredMutation, - structuredMutationMightMutate, - structuredMutationTouchedGraphs, } from './bounded-structured-mutation.js'; import { CACHED_READ_GATE_V1, @@ -559,6 +558,7 @@ export class GraphSetIndexStore implements TripleStore { async structuredMutation(mutation: StructuredMutation, options?: QueryOptions): Promise { const normalized = normalizeStructuredMutation(mutation); + const effects = captureStructuredMutationEffects(normalized); const operation = this.inner.structuredMutation; if (!operation) { throw new UnsupportedTripleStoreCapabilityError('structuredMutation', 'GraphSetIndexStore'); @@ -571,10 +571,10 @@ export class GraphSetIndexStore implements TripleStore { } throw error; } - if (!this.enabled || !structuredMutationMightMutate(normalized)) return; + if (!this.enabled || !effects) return; this.bumpMutation(); await this.maintainTouchedGraphs( - [...structuredMutationTouchedGraphs(normalized)], + [...effects.touchedGraphs], 'structuredMutation', options, ); diff --git a/packages/storage/src/index.ts b/packages/storage/src/index.ts index 46a41ce01d..b7b69312ce 100644 --- a/packages/storage/src/index.ts +++ b/packages/storage/src/index.ts @@ -51,9 +51,11 @@ export { } from './atomic-graph-replace.js'; export { BOUNDED_MUTATION_MAX_PRUNE_DELETE, + captureStructuredMutationEffects, chunkCopySubjectProjectionInput, structuredMutationMightMutate, structuredMutationTouchedGraphs, + type StructuredMutationEffects, } from './bounded-structured-mutation.js'; // System-record V1 (#2052 Stack B2). Default-unused: these modules perform no // I/O, scheduling, timer, or per-store lane work until the daemon supervisor diff --git a/packages/storage/test/bounded-structured-mutation.test.ts b/packages/storage/test/bounded-structured-mutation.test.ts index a28d8ee116..ed4ee8ff63 100644 --- a/packages/storage/test/bounded-structured-mutation.test.ts +++ b/packages/storage/test/bounded-structured-mutation.test.ts @@ -8,12 +8,14 @@ import { OxigraphWorkerStore, TRIPLE_STORE_CAPABILITY_SUPPORT, UnsupportedTripleStoreCapabilityError, + captureStructuredMutationEffects, supportsReplaceSubjectPredicatesAtomically, supportsTripleStoreCapability, tryCopySubjectProjection, tryDeleteSubjects, tryReplaceProjectionFromGraphAtomically, type Quad, + type StructuredMutation, type TripleStore, } from '../src/index.js'; import { @@ -47,6 +49,71 @@ async function rows(store: TripleStore, graph: string): Promise { + it('captures immutable graph effects before dispatch', () => { + const mutation = { + kind: 'copy-subject-projection' as const, + input: { + sourceGraphUris: [GRAPH], + targetGraphUri: OTHER_GRAPH, + roots: ['urn:test:a'], + descendantSuffix: '/', + excludedPredicates: [], + }, + }; + + const effects = captureStructuredMutationEffects(mutation); + mutation.input.targetGraphUri = 'urn:test:redirected'; + expect(effects).toEqual({ touchedGraphs: [OTHER_GRAPH] }); + expect(Object.isFrozen(effects)).toBe(true); + expect(Object.isFrozen(effects.touchedGraphs)).toBe(true); + }); + + it('classifies structural no-ops without store orchestration', () => { + expect(captureStructuredMutationEffects({ + kind: 'delete-subjects', + input: { graphUri: GRAPH, subjects: [] }, + })).toBeUndefined(); + expect(captureStructuredMutationEffects({ + kind: 'delete-subjects', + input: { graphUri: GRAPH, subjects: ['urn:test:a'] }, + })).toEqual({ touchedGraphs: [GRAPH] }); + }); + + it('reports the canonical target graph for every structured mutation kind', async () => { + const mutations: Array = [ + [{ kind: 'delete-subjects', input: { + graphUri: GRAPH, subjects: ['urn:test:a'], + } }, GRAPH], + [{ kind: 'prune-ranked-subjects', input: { + graphUri: GRAPH, subjectPrefix: 'urn:test:req:', eligibilityPredicate: STATUS, + eligibleObjects: ['approved'], primaryRankPredicate: DECIDED_AT, + secondaryRankPredicate: REQUESTED_AT, retainNewest: 1, maxDelete: 1, + } }, GRAPH], + [{ kind: 'prune-linked-record-closures', input: { + graphUri: GRAPH, matchObjectIris: ['urn:test:agent'], linkPredicates: [P], + recordParentPredicate: STATUS, descendantSeparator: '/', + } }, GRAPH], + [{ kind: 'replace-subject-predicates', input: { + graphUri: GRAPH, subject: 'urn:test:a', predicates: [P], + replacementQuads: [quad('urn:test:a', P, '"replacement"')], + } }, GRAPH], + [{ kind: 'replace-projection-from-graph', input: { + targetGraphUri: GRAPH, stagingGraphUri: 'urn:test:staging', + targetSubject: 'urn:test:a', preservedTargetPredicates: [], + targetSubjectPrefixes: [], + } }, GRAPH], + [{ kind: 'copy-subject-projection', input: { + sourceGraphUris: [GRAPH], targetGraphUri: OTHER_GRAPH, + roots: ['urn:test:a'], descendantSuffix: '/', excludedPredicates: [], + } }, OTHER_GRAPH], + ]; + for (const [mutation, expectedGraph] of mutations) { + expect(captureStructuredMutationEffects(mutation)).toEqual({ + touchedGraphs: [expectedGraph], + }); + } + }); + it('deletes one explicit subject set and preserves co-located rows', async () => { const store = new OxigraphStore(); await store.insert([