diff --git a/packages/agent/src/dkg-agent-base.ts b/packages/agent/src/dkg-agent-base.ts index e764312b37..47072d2ad8 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, captureStructuredMutationEffects, createTripleStore, isExternalBackend, type TripleStore, type TripleStoreConfig, type Quad, type LargeLiteralStorageConfig, type QueryOptions } from '@origintrail-official/dkg-storage'; +import { GraphManager, PrivateContentStore, SystemRecordLaneForwarderV1, captureStructuredMutationSnapshot, 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, @@ -577,14 +577,17 @@ export function createListContextGraphsCacheInvalidatingStore( ) : undefined, structuredMutation: innerStore.structuredMutation - ? (mutation, options) => { - const effects = captureStructuredMutationEffects(mutation); - return invalidateAfterMutation( - () => innerStore.structuredMutation!(mutation, options), - () => effects !== undefined, - () => effects?.touchedGraphs.forEach( - (graph) => markProjectionDirty?.(undefined, graph), - ), + ? async (mutation, options) => { + const snapshot = captureStructuredMutationSnapshot(mutation); + await invalidateAfterMutation( + () => innerStore.structuredMutation!(snapshot.mutation, options), + () => snapshot.outcome !== 'noop', + () => { + if (snapshot.outcome === 'noop') return; + snapshot.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 03a2da2bad..6554a2c19f 100644 --- a/packages/agent/test/replace-subject-agent-wrapper.test.ts +++ b/packages/agent/test/replace-subject-agent-wrapper.test.ts @@ -22,6 +22,7 @@ import { createTripleStore, tryReplaceSubjectAtomically, type Quad, + type StructuredMutation, type TripleStore, } from '@origintrail-official/dkg-storage'; import { contextGraphCatalogUri, contextGraphMetaGraphUri } from '@origintrail-official/dkg-core'; @@ -156,13 +157,15 @@ describe('#1863 replaceSubject through the agent store wrapper', () => { const inFlight = new Promise((resolve) => { release = resolve; }); const options = { source: 'agent.test.structured-mutation-effects' }; let inner!: TripleStore; + let observedMutation: unknown; const structuredMutation = vi.fn(function ( this: TripleStore, - _mutation: unknown, + receivedMutation: unknown, receivedOptions: unknown, ) { expect(this).toBe(inner); expect(receivedOptions).toBe(options); + observedMutation = receivedMutation; return inFlight; }); inner = { structuredMutation } as unknown as TripleStore; @@ -193,6 +196,92 @@ describe('#1863 replaceSubject through the agent store wrapper', () => { expect(invalidate).toHaveBeenCalledOnce(); expect(markProjectionDirty).toHaveBeenCalledOnce(); expect(markProjectionDirty).toHaveBeenCalledWith(undefined, 'urn:test:target'); + expect(observedMutation).toEqual({ + kind: 'copy-subject-projection', + input: { + sourceGraphUris: ['urn:test:source'], + targetGraphUri: 'urn:test:target', + roots: ['urn:test:root'], + descendantSuffix: '/', + excludedPredicates: [], + }, + }); + expect(Object.isFrozen(observedMutation)).toBe(true); + }); + + it('forwards every structured mutation kind with exact scoped effects', async () => { + const structuredMutation = vi.fn(async () => undefined); + const inner = { structuredMutation } as unknown as TripleStore; + const invalidate = vi.fn(); + const markProjectionDirty = vi.fn(); + const store = createListContextGraphsCacheInvalidatingStore( + inner, + invalidate, + markProjectionDirty, + ); + const options = { source: 'agent.test.all-structured-mutations' }; + const graph = 'urn:test:agent:graph'; + const target = 'urn:test:agent:target'; + const predicate = 'urn:test:agent:predicate'; + const fixtures: Array<{ mutation: StructuredMutation; touched: string }> = [ + { mutation: { kind: 'delete-subjects', input: { + graphUri: graph, subjects: ['urn:test:agent:subject'], + } }, touched: graph }, + { mutation: { kind: 'prune-ranked-subjects', input: { + graphUri: graph, + subjectPrefix: 'urn:test:agent:ranked:', + eligibilityPredicate: predicate, + eligibleObjects: ['approved'], + primaryRankPredicate: 'urn:test:agent:rank:primary', + secondaryRankPredicate: 'urn:test:agent:rank:secondary', + retainNewest: 1, + maxDelete: 1, + } }, touched: graph }, + { mutation: { kind: 'prune-linked-record-closures', input: { + graphUri: graph, + matchObjectIris: ['urn:test:agent:member'], + linkPredicates: [predicate], + recordParentPredicate: 'urn:test:agent:parent', + descendantSeparator: '/', + } }, touched: graph }, + { mutation: { kind: 'replace-subject-predicates', input: { + graphUri: graph, + subject: 'urn:test:agent:subject', + predicates: [predicate], + replacementQuads: [{ + graph, + subject: 'urn:test:agent:subject', + predicate, + object: '"value"', + }], + } }, touched: graph }, + { mutation: { kind: 'replace-projection-from-graph', input: { + targetGraphUri: target, + stagingGraphUri: 'urn:test:agent:staging', + targetSubject: 'urn:test:agent:subject', + preservedTargetPredicates: [predicate], + targetSubjectPrefixes: [], + } }, touched: target }, + { mutation: { kind: 'copy-subject-projection', input: { + sourceGraphUris: [graph], + targetGraphUri: target, + roots: ['urn:test:agent:root'], + descendantSuffix: '/', + excludedPredicates: [], + } }, touched: target }, + ]; + + for (const [index, fixture] of fixtures.entries()) { + await store.structuredMutation!(fixture.mutation, options); + expect(structuredMutation.mock.calls[index][1]).toBe(options); + expect(Object.isFrozen(structuredMutation.mock.calls[index][0])).toBe(true); + expect(markProjectionDirty).toHaveBeenNthCalledWith( + index + 1, + undefined, + fixture.touched, + ); + } + expect(invalidate).toHaveBeenCalledTimes(fixtures.length); }); it('does not invalidate structured mutation failures or structural no-ops', async () => { @@ -211,6 +300,8 @@ describe('#1863 replaceSubject through the agent store wrapper', () => { kind: 'delete-subjects', input: { graphUri: 'urn:test:target', subjects: [] }, }); + expect(inner.structuredMutation).toHaveBeenCalledOnce(); + expect(Object.isFrozen(inner.structuredMutation.mock.calls[0][0])).toBe(true); expect(invalidate).not.toHaveBeenCalled(); expect(markProjectionDirty).not.toHaveBeenCalled(); @@ -222,4 +313,21 @@ describe('#1863 replaceSubject through the agent store wrapper', () => { expect(invalidate).not.toHaveBeenCalled(); expect(markProjectionDirty).not.toHaveBeenCalled(); }); + + it('converts synchronous snapshot validation failures into rejected Promises', async () => { + const inner = { + structuredMutation: vi.fn(async () => undefined), + } as unknown as TripleStore; + const store = createListContextGraphsCacheInvalidatingStore(inner, vi.fn(), vi.fn()); + let result!: Promise; + + expect(() => { + result = store.structuredMutation!({ + kind: 'delete-subjects', + input: { graphUri: 'relative', subjects: [] }, + }); + }).not.toThrow(); + await expect(result).rejects.toThrow(/absolute IRI/); + expect(inner.structuredMutation).not.toHaveBeenCalled(); + }); }); diff --git a/packages/storage/package.json b/packages/storage/package.json index 27901d10fd..fb3164f716 100644 --- a/packages/storage/package.json +++ b/packages/storage/package.json @@ -18,6 +18,7 @@ "./package.json": "./package.json" }, "scripts": { + "benchmark:structured-mutation": "node scripts/benchmark-structured-mutation.mjs", "build": "tsc", "test": "vitest run", "test:package-exports": "node scripts/verify-pack-exports.mjs", diff --git a/packages/storage/scripts/benchmark-structured-mutation.mjs b/packages/storage/scripts/benchmark-structured-mutation.mjs new file mode 100644 index 0000000000..3afb72d692 --- /dev/null +++ b/packages/storage/scripts/benchmark-structured-mutation.mjs @@ -0,0 +1,212 @@ +#!/usr/bin/env node + +import { spawnSync } from 'node:child_process'; +import { cpus } from 'node:os'; +import { createHash } from 'node:crypto'; +import { existsSync, mkdirSync, readFileSync, writeFileSync } from 'node:fs'; +import { dirname } from 'node:path'; +import { fileURLToPath } from 'node:url'; +import { execFileSync } from 'node:child_process'; + +const WARMUPS = 10; +const TRIALS = 30; +const REPETITIONS = 3; +const REGRESSION_LIMIT_PERCENT = 10; + +const fixture = Object.freeze({ + kind: 'delete-subjects', + input: Object.freeze({ + graphUri: 'urn:benchmark:structured-mutation', + subjects: Object.freeze(Array.from( + { length: 100_000 }, + (_, index) => `urn:benchmark:subject:${index.toString().padStart(6, '0')}`, + )), + }), +}); +const fixtureDigest = createHash('sha256') + .update(JSON.stringify(fixture)) + .digest('hex'); + +if (process.argv.includes('--child')) { + await runChild(); +} else { + runParent(); +} + +async function runChild() { + if (typeof globalThis.gc !== 'function') { + throw new Error('benchmark child requires --expose-gc'); + } + const bounded = await import('../dist/bounded-structured-mutation.js'); + const root = await import('../dist/index.js'); + let prepare; + if (typeof root.captureStructuredMutationSnapshot === 'function') { + const materialization = await import('../dist/structured-mutation-materialization-internal.js'); + prepare = () => { + const result = materialization.materializeStructuredMutation( + root.captureStructuredMutationSnapshot(fixture), + ); + // Match the pre-change path's returned lifetime: retain only the generated + // update, not the new API's diagnostic reference back to its input snapshot. + return result.outcome === 'execute' ? result.update : undefined; + }; + } else { + prepare = () => { + const normalized = bounded.normalizeStructuredMutation(fixture); + bounded.captureStructuredMutationEffects(normalized); + return bounded.buildStructuredMutationUpdate(normalized); + }; + } + + for (let index = 0; index < WARMUPS; index += 1) prepare(); + globalThis.gc(); + const heapBefore = process.memoryUsage().heapUsed; + const cpuBefore = process.cpuUsage(); + let witness = 0; + for (let index = 0; index < TRIALS; index += 1) { + const result = prepare(); + witness += typeof result === 'string' ? result.length : 0; + } + const cpu = process.cpuUsage(cpuBefore); + globalThis.gc(); + const heapAfter = process.memoryUsage().heapUsed; + const maxRss = process.resourceUsage().maxRSS; + process.stdout.write(`${JSON.stringify({ + cpuMicros: cpu.user + cpu.system, + retainedHeapBytes: Math.max(0, heapAfter - heapBefore), + maxRssBytes: process.platform === 'darwin' ? maxRss : maxRss * 1024, + witness, + })}\n`); +} + +function runParent() { + const options = parseArgs(process.argv.slice(2)); + assertNode22(); + if (existsSync(options.output)) { + throw new Error(`refusing to overwrite benchmark output ${options.output}`); + } + const repetitions = Array.from({ length: REPETITIONS }, () => runFreshChild()); + const metrics = Object.freeze({ + cpuMicros: median(repetitions.map(({ cpuMicros }) => cpuMicros)), + retainedHeapBytes: median(repetitions.map(({ retainedHeapBytes }) => retainedHeapBytes)), + maxRssBytes: median(repetitions.map(({ maxRssBytes }) => maxRssBytes)), + }); + const metadata = environmentMetadata(); + let comparison; + if (options.mode === 'compare') { + const baseline = readBenchmark(options.baseline); + assertComparable(baseline, metadata); + comparison = Object.freeze({ + baseline: options.baseline, + cpuPercent: percentage(metrics.cpuMicros, baseline.metrics.cpuMicros), + retainedHeapPercent: percentage( + metrics.retainedHeapBytes, + baseline.metrics.retainedHeapBytes, + ), + maxRssPercent: percentage(metrics.maxRssBytes, baseline.metrics.maxRssBytes), + }); + } + const result = Object.freeze({ + schemaVersion: 1, + mode: options.mode, + metadata, + fixture: Object.freeze({ + digest: fixtureDigest, + mutationKind: fixture.kind, + subjects: fixture.input.subjects.length, + warmups: WARMUPS, + trials: TRIALS, + repetitions: REPETITIONS, + }), + metrics, + repetitions, + ...(comparison === undefined ? {} : { comparison }), + }); + mkdirSync(dirname(options.output), { recursive: true }); + writeFileSync(options.output, `${JSON.stringify(result, null, 2)}\n`, { flag: 'wx' }); + process.stdout.write(`${JSON.stringify(result, null, 2)}\n`); + if (comparison !== undefined && Object.entries(comparison).some( + ([key, value]) => key.endsWith('Percent') && value > REGRESSION_LIMIT_PERCENT, + )) { + process.exitCode = 2; + } +} + +function parseArgs(args) { + let mode; + let output; + let baseline; + for (let index = 0; index < args.length; index += 1) { + const flag = args[index]; + const value = args[index + 1]; + if (flag === '--') continue; + if (flag === '--mode') mode = value; + else if (flag === '--output') output = value; + else if (flag === '--baseline') baseline = value; + else throw new Error(`unknown benchmark argument ${flag}`); + index += 1; + } + if (mode !== 'baseline' && mode !== 'compare') { + throw new Error('--mode must be baseline or compare'); + } + if (typeof output !== 'string' || output.length === 0) { + throw new Error('--output is required'); + } + if (mode === 'compare' && (typeof baseline !== 'string' || baseline.length === 0)) { + throw new Error('--baseline is required in compare mode'); + } + return Object.freeze({ mode, output, baseline }); +} + +function runFreshChild() { + const result = spawnSync( + process.execPath, + ['--expose-gc', fileURLToPath(import.meta.url), '--child'], + { encoding: 'utf8', maxBuffer: 16 * 1024 * 1024 }, + ); + if (result.status !== 0) { + throw new Error(`benchmark child failed: ${result.stderr || result.stdout}`); + } + return Object.freeze(JSON.parse(result.stdout)); +} + +function environmentMetadata() { + return Object.freeze({ + node: process.version, + architecture: process.arch, + platform: process.platform, + cpuModel: cpus()[0]?.model ?? 'unknown', + cpuCount: cpus().length, + gitSha: execFileSync('git', ['rev-parse', 'HEAD'], { encoding: 'utf8' }).trim(), + gitDirty: execFileSync('git', ['status', '--porcelain'], { encoding: 'utf8' }).length > 0, + }); +} + +function readBenchmark(path) { + return JSON.parse(readFileSync(path, 'utf8')); +} + +function assertComparable(baseline, metadata) { + if (baseline.mode !== 'baseline') throw new Error('comparison input is not a baseline result'); + if (baseline.metadata?.node !== metadata.node) throw new Error('benchmark Node version mismatch'); + if (baseline.metadata?.architecture !== metadata.architecture) { + throw new Error('benchmark architecture mismatch'); + } + if (baseline.fixture?.digest !== fixtureDigest) throw new Error('benchmark fixture mismatch'); +} + +function assertNode22() { + if (Number(process.versions.node.split('.')[0]) !== 22) { + throw new Error(`benchmark requires Node.js 22, received ${process.version}`); + } +} + +function median(values) { + const sorted = [...values].sort((left, right) => left - right); + return sorted[Math.floor(sorted.length / 2)]; +} + +function percentage(candidate, baseline) { + if (baseline === 0) return candidate === 0 ? 0 : Number.POSITIVE_INFINITY; + return ((candidate - baseline) / baseline) * 100; +} diff --git a/packages/storage/scripts/pack-gate/barrel-value-exports.json b/packages/storage/scripts/pack-gate/barrel-value-exports.json index 5dc2b5b3ed..f68558d3da 100644 --- a/packages/storage/scripts/pack-gate/barrel-value-exports.json +++ b/packages/storage/scripts/pack-gate/barrel-value-exports.json @@ -49,6 +49,7 @@ "buildAtomicSubjectReplaceUpdate", "canonicalSharedMemoryScopeWriteGraph", "captureStructuredMutationEffects", + "captureStructuredMutationSnapshot", "changelogSchemaQuad", "chunkCopySubjectProjectionInput", "createStoreControlBarrierKeyV1", diff --git a/packages/storage/scripts/pack-gate/type-fixture.ts b/packages/storage/scripts/pack-gate/type-fixture.ts index 2b29304d8d..96f5d27c0a 100644 --- a/packages/storage/scripts/pack-gate/type-fixture.ts +++ b/packages/storage/scripts/pack-gate/type-fixture.ts @@ -4,8 +4,14 @@ // error; a forbidden symbol returning to the barrel turns its suppression // into an unused directive (TS2578). The directive token must never begin a // wrapped comment line — tsc parses such a comment as a real directive. -import { ManagedOxigraphBackendUnownedError } from '@origintrail-official/dkg-storage'; +import { + ManagedOxigraphBackendUnownedError, + captureStructuredMutationSnapshot, + type StructuredMutationSnapshot, +} from '@origintrail-official/dkg-storage'; import type { ManagedOxigraphSupervisorHandoffV1 } from '@origintrail-official/dkg-storage/internal/managed-oxigraph-ownership-v1'; +// @ts-expect-error — final mutation materialization has no supported package subpath +import { materializeStructuredMutation } from '@origintrail-official/dkg-storage/structured-mutation-materialization-internal'; // @ts-expect-error — the ownership mint is not on the public barrel import { createManagedOxigraphOwnershipControllerV1 } from '@origintrail-official/dkg-storage'; // Every removed ownership type is pinned individually — runtime namespace @@ -25,4 +31,10 @@ const witness: [typeof ManagedOxigraphBackendUnownedError, ManagedOxigraphSuperv ManagedOxigraphBackendUnownedError, null, ]; +const snapshotWitness: StructuredMutationSnapshot = captureStructuredMutationSnapshot({ + kind: 'delete-subjects', + input: { graphUri: 'urn:test:pack-gate', subjects: [] }, +}); +void materializeStructuredMutation; +void snapshotWitness; export default witness; diff --git a/packages/storage/scripts/verify-pack-exports.mjs b/packages/storage/scripts/verify-pack-exports.mjs index 5ace453037..571fe06382 100644 --- a/packages/storage/scripts/verify-pack-exports.mjs +++ b/packages/storage/scripts/verify-pack-exports.mjs @@ -53,6 +53,8 @@ const GATE = Object.freeze({ ['@origintrail-official/dkg-storage/package.json', true], ['@origintrail-official/dkg-storage/dist/internal/managed-oxigraph-ownership-v1.js', false], ['@origintrail-official/dkg-storage/dist/store-priority-scheduler.js', false], + ['@origintrail-official/dkg-storage/structured-mutation-materialization-internal', false], + ['@origintrail-official/dkg-storage/dist/structured-mutation-materialization-internal.js', false], ['@origintrail-official/dkg-storage/dist/index.js', false], ], }); diff --git a/packages/storage/src/adapters/blazegraph.ts b/packages/storage/src/adapters/blazegraph.ts index f05cbeb330..7831326785 100644 --- a/packages/storage/src/adapters/blazegraph.ts +++ b/packages/storage/src/adapters/blazegraph.ts @@ -33,10 +33,9 @@ import { isAtomicGraphReplaceStagingGraph, } from '../atomic-graph-replace.js'; import { - buildStructuredMutationUpdate, - captureStructuredMutationEffects, - normalizeStructuredMutation, + captureStructuredMutationSnapshot, } from '../bounded-structured-mutation.js'; +import { materializeStructuredMutation } from '../structured-mutation-materialization-internal.js'; import { quadToNQuad } from '../bounded-rdf.js'; import { readResponseTextBounded } from '../http-response-limit.js'; @@ -416,18 +415,17 @@ export class BlazegraphStore implements TripleStore { mutation: StructuredMutation, options?: QueryOptions, ): Promise { - const normalized = normalizeStructuredMutation(mutation); - const effects = captureStructuredMutationEffects(normalized); - if (normalized.kind === 'replace-subject-predicates') { - assertQuadLiteralsMutf8Safe([...normalized.input.replacementQuads], { + const snapshot = captureStructuredMutationSnapshot(mutation); + if (snapshot.mutation.kind === 'replace-subject-predicates') { + assertQuadLiteralsMutf8Safe(snapshot.mutation.input.replacementQuads, { maxBytes: JAVA_WRITE_UTF_MAX_BYTES, label: 'BlazegraphStore.structuredMutation', }); } - const update = buildStructuredMutationUpdate(normalized); - if (!update || !effects) return; + const materialized = materializeStructuredMutation(snapshot); + if (materialized.outcome === 'noop') return; await this.sparqlUpdate( - update, + materialized.update, { ...options, source: options?.source ?? 'blazegraph.structuredMutation' }, 'structuredMutation', ); diff --git a/packages/storage/src/adapters/oxigraph-worker-impl.ts b/packages/storage/src/adapters/oxigraph-worker-impl.ts index 6bfce261ce..66d8c721da 100644 --- a/packages/storage/src/adapters/oxigraph-worker-impl.ts +++ b/packages/storage/src/adapters/oxigraph-worker-impl.ts @@ -1,5 +1,6 @@ import { parentPort, workerData } from 'node:worker_threads'; import { OxigraphStore } from './oxigraph.js'; +import { structuredMutationPreDispatchRefusalCode } from '../structured-mutation/refusal-internal.js'; const store = new OxigraphStore(workerData?.persistPath); @@ -7,12 +8,19 @@ parentPort!.on('message', async (msg: { id: number; method: string; args: unknow try { const fn = (store as any)[msg.method]; if (typeof fn !== 'function') { - parentPort!.postMessage({ id: msg.id, error: `Unknown method: ${msg.method}` }); + parentPort!.postMessage({ id: msg.id, error: { message: `Unknown method: ${msg.method}` } }); return; } const result = await fn.apply(store, msg.args); parentPort!.postMessage({ id: msg.id, result }); } catch (err) { - parentPort!.postMessage({ id: msg.id, error: err instanceof Error ? err.message : String(err) }); + const code = structuredMutationPreDispatchRefusalCode(err); + parentPort!.postMessage({ + id: msg.id, + error: { + message: err instanceof Error ? err.message : String(err), + ...(code === undefined ? {} : { code }), + }, + }); } }); diff --git a/packages/storage/src/adapters/oxigraph-worker.ts b/packages/storage/src/adapters/oxigraph-worker.ts index e612ec95e3..989805aae1 100644 --- a/packages/storage/src/adapters/oxigraph-worker.ts +++ b/packages/storage/src/adapters/oxigraph-worker.ts @@ -12,10 +12,13 @@ import type { } from '../triple-store.js'; import { registerTripleStoreAdapter } from '../triple-store.js'; import { GraphWriteGenTracker } from '../graph-write-gen.js'; +import { captureStructuredMutationSnapshot } from '../bounded-structured-mutation.js'; +import { assertQuadLiteralsMutf8Safe, JAVA_WRITE_UTF_MAX_BYTES } from '@origintrail-official/dkg-core'; +import { assertStructuredMutationSnapshotMaterializable } from '../structured-mutation-materialization-internal.js'; import { - captureStructuredMutationEffects, - normalizeStructuredMutation, -} from '../bounded-structured-mutation.js'; + STRUCTURED_MUTATION_PRE_DISPATCH_REFUSAL_CODE, + reconstructStructuredMutationPreDispatchRefusal, +} from '../structured-mutation/refusal-internal.js'; /** * Default per-operation timeout for the embedded worker store. The worker is @@ -120,6 +123,19 @@ export interface OxigraphWorkerTimeoutError extends Error { timeoutMs: number; } +interface WorkerErrorPayload { + readonly message: string; + readonly code?: string; +} + +function deserializeWorkerError(payload: string | WorkerErrorPayload): Error { + if (typeof payload === 'string') return new Error(payload); + if (payload.code === STRUCTURED_MUTATION_PRE_DISPATCH_REFUSAL_CODE) { + return reconstructStructuredMutationPreDispatchRefusal(payload.message); + } + return new Error(payload.message); +} + /** * Explicit worker POLICY state, replacing the old cluster of interdependent * booleans/promises that all had to agree. `respawnGaveUp` folds into the @@ -336,7 +352,11 @@ export class OxigraphWorkerStore implements TripleStore { // this: a spawn only happens in the constructor or within respawn(), which // bails the moment a close() is seen. this.markSpawnedLive(); - worker.on('message', (msg: { id: number; result?: unknown; error?: string }) => { + worker.on('message', (msg: { + id: number; + result?: unknown; + error?: string | WorkerErrorPayload; + }) => { if (this.worker !== worker) return; // Any successful reply proves this worker is healthy, which ends the // crash-loop accounting window (see MAX_CONSECUTIVE_RESPAWNS). @@ -344,7 +364,7 @@ export class OxigraphWorkerStore implements TripleStore { const p = this.pending.get(msg.id); if (!p) return; this.pending.delete(msg.id); - if (msg.error) p.reject(new Error(msg.error)); + if (msg.error) p.reject(deserializeWorkerError(msg.error)); else p.resolve(msg.result); }); worker.on('error', (err) => { @@ -540,6 +560,30 @@ export class OxigraphWorkerStore implements TripleStore { return this.postToWorker(timeoutMs, signal, method, args); } + /** Share call()'s lifecycle linearization without allocating a worker message. */ + private async preflightNoop(method: string): Promise { + while (this.respawnPromise) await this.respawnPromise; + if (this.lifecycle === 'live' && !this.workerExited) return; + throw this.workerUnavailableError(method); + } + + private workerUnavailableError(method: string): Error { + if (this.lifecycle === 'in_memory_lost') { + return new Error( + `oxigraph-worker: cannot run "${method}" — the IN-MEMORY store's worker crashed and its data was ` + + 'lost. An in-memory store cannot be recovered from a worker crash; every request now fails ' + + 'fast. Use a disk-persisted store (store.path) or restart the node to start from empty.', + ); + } + return new Error( + `oxigraph-worker: cannot run "${method}" — the store is closed.` + + (this.lifecycle === 'gave_up' + ? ' (The worker crashed repeatedly and automatic respawn gave up — restart the node and ' + + 'investigate the [oxigraph-worker] crash logs.)' + : ''), + ); + } + private postToWorker( timeoutMs: number, signal: AbortSignal | undefined, @@ -558,21 +602,7 @@ export class OxigraphWorkerStore implements TripleStore { // • gave_up — closed + the crash-loop guidance. // • otherwise — a plain operator close. if (this.workerExited) { - if (this.lifecycle === 'in_memory_lost') { - reject(new Error( - `oxigraph-worker: cannot run "${method}" — the IN-MEMORY store's worker crashed and its data was ` + - 'lost. An in-memory store cannot be recovered from a worker crash; every request now fails ' + - 'fast. Use a disk-persisted store (store.path) or restart the node to start from empty.', - )); - return; - } - reject(new Error( - `oxigraph-worker: cannot run "${method}" — the store is closed.` + - (this.lifecycle === 'gave_up' - ? ' (The worker crashed repeatedly and automatic respawn gave up — restart the node and ' + - 'investigate the [oxigraph-worker] crash logs.)' - : ''), - )); + reject(this.workerUnavailableError(method)); return; } if (signal?.aborted) { @@ -696,10 +726,20 @@ export class OxigraphWorkerStore implements TripleStore { // aborting the caller could report failure while the worker later commits. _options?: TripleStoreQueryOptions, ): Promise { - const normalized = normalizeStructuredMutation(mutation); - const effects = captureStructuredMutationEffects(normalized); - await this.call('structuredMutation', normalized); - if (effects) this.writeGen.recordGraphWrites(effects.touchedGraphs); + const snapshot = captureStructuredMutationSnapshot(mutation); + assertStructuredMutationSnapshotMaterializable(snapshot); + if (snapshot.outcome === 'noop') { + await this.preflightNoop('structuredMutation'); + return; + } + if (snapshot.mutation.kind === 'replace-subject-predicates') { + assertQuadLiteralsMutf8Safe(snapshot.mutation.input.replacementQuads, { + maxBytes: JAVA_WRITE_UTF_MAX_BYTES, + label: 'OxigraphWorkerStore.structuredMutation', + }); + } + await this.call('structuredMutation', snapshot.mutation); + this.writeGen.recordGraphWrites(snapshot.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 986a29c327..65aaf02c15 100644 --- a/packages/storage/src/adapters/oxigraph.ts +++ b/packages/storage/src/adapters/oxigraph.ts @@ -27,10 +27,9 @@ import { isAtomicGraphReplaceStagingGraph, } from '../atomic-graph-replace.js'; import { - buildStructuredMutationUpdate, - captureStructuredMutationEffects, - normalizeStructuredMutation, + captureStructuredMutationSnapshot, } from '../bounded-structured-mutation.js'; +import { materializeStructuredMutation } from '../structured-mutation-materialization-internal.js'; import { quadsToNQuads } from '../bounded-rdf.js'; import { assertQuadLiteralsMutf8Safe, JAVA_WRITE_UTF_MAX_BYTES } from '@origintrail-official/dkg-core'; @@ -419,19 +418,18 @@ export class OxigraphStore implements TripleStore { // dispatch; source/priority are advisory only for scheduled HTTP stores. _options?: TripleStoreQueryOptions, ): Promise { - const normalized = normalizeStructuredMutation(mutation); - const effects = captureStructuredMutationEffects(normalized); - if (normalized.kind === 'replace-subject-predicates') { - assertQuadLiteralsMutf8Safe([...normalized.input.replacementQuads], { + const snapshot = captureStructuredMutationSnapshot(mutation); + if (snapshot.mutation.kind === 'replace-subject-predicates') { + assertQuadLiteralsMutf8Safe(snapshot.mutation.input.replacementQuads, { maxBytes: JAVA_WRITE_UTF_MAX_BYTES, label: 'OxigraphStore.structuredMutation', }); } - const update = buildStructuredMutationUpdate(normalized); - if (!update || !effects) return; - this.store.update(update); + const materialized = materializeStructuredMutation(snapshot); + if (materialized.outcome === 'noop') return; + this.store.update(materialized.update); this.scheduleFlush(); - this.writeGen.recordGraphWrites(effects.touchedGraphs); + this.writeGen.recordGraphWrites(materialized.snapshot.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 7ac23f6981..13fc61e682 100644 --- a/packages/storage/src/adapters/sparql-http.ts +++ b/packages/storage/src/adapters/sparql-http.ts @@ -53,11 +53,9 @@ import { isAtomicGraphReplaceStagingGraph, } from '../atomic-graph-replace.js'; import { - buildStructuredMutationUpdate, - captureStructuredMutationEffects, - normalizeStructuredMutation, - structuredMutationGuardedGraphs, + captureStructuredMutationSnapshot, } from '../bounded-structured-mutation.js'; +import { materializeStructuredMutation } from '../structured-mutation-materialization-internal.js'; import { assertNotReservedInternalGraphV1, isInternalGraphUriV1, @@ -1288,20 +1286,20 @@ export class SparqlHttpStore implements TripleStore { mutation: StructuredMutation, options?: QueryOptions, ): Promise { - const normalized = normalizeStructuredMutation(mutation); - const effects = captureStructuredMutationEffects(normalized); - this.assertGenericMutationScope(structuredMutationGuardedGraphs(normalized), 'structuredMutation'); - if (normalized.kind === 'replace-subject-predicates') { - assertQuadLiteralsMutf8Safe([...normalized.input.replacementQuads], { + const snapshot = captureStructuredMutationSnapshot(mutation); + this.assertGenericMutationScope(snapshot.guardedGraphs, 'structuredMutation'); + if (snapshot.mutation.kind === 'replace-subject-predicates') { + assertQuadLiteralsMutf8Safe(snapshot.mutation.input.replacementQuads, { maxBytes: JAVA_WRITE_UTF_MAX_BYTES, label: 'SparqlHttpStore.structuredMutation', }); } - const update = buildStructuredMutationUpdate(normalized); - if (!update || !effects) return; + const materialized = materializeStructuredMutation(snapshot); + if (materialized.outcome === 'noop') return; + const effects = materialized.snapshot.effects; try { await this.postUpdate( - update, + materialized.update, { ...options, source: options?.source ?? 'sparql-http.structuredMutation' }, 'structuredMutation', effects.touchedGraphs, diff --git a/packages/storage/src/bounded-structured-mutation.ts b/packages/storage/src/bounded-structured-mutation.ts index bee5cbe9af..248384e9df 100644 --- a/packages/storage/src/bounded-structured-mutation.ts +++ b/packages/storage/src/bounded-structured-mutation.ts @@ -14,28 +14,41 @@ import { } from './structured-mutation/primitives.js'; import { buildDeleteSubjectsUpdate, + captureDeleteSubjectsInput, + deleteSubjectsSemantics, normalizeDeleteSubjectsInput, } from './structured-mutation/delete-subjects.js'; import { buildPruneLinkedRecordClosuresUpdate, buildPruneRankedSubjectsUpdate, + capturePruneLinkedRecordClosuresInput, + capturePruneRankedSubjectsInput, normalizePruneLinkedRecordClosuresInput, normalizePruneRankedSubjectsInput, + pruneLinkedRecordClosuresSemantics, + pruneRankedSubjectsSemantics, } from './structured-mutation/retention.js'; import { buildReplaceSubjectPredicatesUpdate, + captureReplaceSubjectPredicatesInput, normalizeReplaceSubjectPredicatesInput, normalizeReplaceSubjectPredicatesInputForObjectRewrite, + replaceSubjectPredicatesSemantics, } from './structured-mutation/replace-subject-predicates.js'; import { buildReplaceProjectionFromGraphUpdate, + captureReplaceProjectionFromGraphInput, normalizeReplaceProjectionFromGraphInput, + replaceProjectionFromGraphSemantics, } from './structured-mutation/replace-projection-from-graph.js'; import { buildCopySubjectProjectionUpdate, + captureCopySubjectProjectionInput, chunkCopySubjectProjectionInput, + copySubjectProjectionSemantics, normalizeCopySubjectProjectionInput, } from './structured-mutation/copy-subject-projection.js'; +import type { StructuredMutationSemantics } from './structured-mutation/capture-internal.js'; export { BOUNDED_MUTATION_MAX_IRIS, @@ -62,6 +75,186 @@ export { normalizeReplaceSubjectPredicatesInputForObjectRewrite, }; +export interface ReadonlyStructuredMutationQuad { + readonly subject: string; + readonly predicate: string; + readonly object: string; + readonly graph: string; +} + +export type ReadonlyStructuredMutation = + | Readonly<{ + kind: 'delete-subjects'; + input: Readonly<{ graphUri: string; subjects: readonly string[] }>; + }> + | Readonly<{ + kind: 'prune-ranked-subjects'; + input: Readonly<{ + graphUri: string; + subjectPrefix: string; + eligibilityPredicate: string; + eligibleObjects: readonly string[]; + primaryRankPredicate: string; + secondaryRankPredicate: string; + retainNewest: number; + maxDelete: number; + }>; + }> + | Readonly<{ + kind: 'prune-linked-record-closures'; + input: Readonly<{ + graphUri: string; + matchObjectIris: readonly string[]; + linkPredicates: readonly string[]; + recordParentPredicate: string; + protectedRecordIri?: string; + descendantSeparator: string; + }>; + }> + | Readonly<{ + kind: 'replace-subject-predicates'; + input: Readonly<{ + graphUri: string; + subject: string; + predicates: readonly string[]; + replacementQuads: readonly ReadonlyStructuredMutationQuad[]; + }>; + }> + | Readonly<{ + kind: 'replace-projection-from-graph'; + input: Readonly<{ + targetGraphUri: string; + stagingGraphUri: string; + targetSubject: string; + preservedTargetPredicates: readonly string[]; + targetSubjectPrefixes: readonly string[]; + }>; + }> + | Readonly<{ + kind: 'copy-subject-projection'; + input: Readonly<{ + sourceGraphUris: readonly string[]; + targetGraphUri: string; + roots: readonly string[]; + descendantSuffix: string; + excludedPredicates: readonly string[]; + }>; + }>; + +interface StructuredMutationSnapshotBase { + readonly mutation: ReadonlyStructuredMutation; + readonly guardedGraphs: readonly string[]; +} + +export type StructuredMutationSnapshot = + | Readonly + | Readonly; + +const STRUCTURED_MUTATION_SNAPSHOT_BRAND = Symbol('structured-mutation-snapshot'); +const STRUCTURED_MUTATION_SNAPSHOTS = new WeakMap(); + +/** Capture one immutable, caller-independent structured mutation observation. */ +export function captureStructuredMutationSnapshot( + mutation: StructuredMutation, +): StructuredMutationSnapshot { + if (typeof mutation === 'object' && mutation !== null) { + const trusted = STRUCTURED_MUTATION_SNAPSHOTS.get(mutation); + if (trusted !== undefined) return trusted; + } + const descriptor = mutation as unknown as Record; + const kind = descriptor?.kind; + const input = descriptor?.input; + let captured: ReadonlyStructuredMutation; + let semantics: StructuredMutationSemantics; + switch (kind) { + case 'delete-subjects': { + const capturedInput = captureDeleteSubjectsInput(input); + captured = brandMutation(kind, capturedInput); + semantics = deleteSubjectsSemantics(capturedInput); + break; + } + case 'prune-ranked-subjects': { + const capturedInput = capturePruneRankedSubjectsInput(input); + captured = brandMutation(kind, capturedInput); + semantics = pruneRankedSubjectsSemantics(capturedInput); + break; + } + case 'prune-linked-record-closures': { + const capturedInput = capturePruneLinkedRecordClosuresInput(input); + captured = brandMutation(kind, capturedInput); + semantics = pruneLinkedRecordClosuresSemantics(capturedInput); + break; + } + case 'replace-subject-predicates': { + const capturedInput = captureReplaceSubjectPredicatesInput(input); + captured = brandMutation(kind, capturedInput); + semantics = replaceSubjectPredicatesSemantics(capturedInput); + break; + } + case 'replace-projection-from-graph': { + const capturedInput = captureReplaceProjectionFromGraphInput(input); + captured = brandMutation(kind, capturedInput); + semantics = replaceProjectionFromGraphSemantics(capturedInput); + break; + } + case 'copy-subject-projection': { + const capturedInput = captureCopySubjectProjectionInput(input); + captured = brandMutation(kind, capturedInput); + semantics = copySubjectProjectionSemantics(capturedInput); + break; + } + default: + throw new Error(`Unsupported structured mutation kind ${String(kind)}`); + } + const guardedGraphs = Object.freeze([...semantics.guardedGraphs]); + const snapshot: StructuredMutationSnapshot = semantics.mightMutate + ? Object.freeze({ + mutation: captured, + guardedGraphs, + outcome: 'candidate' as const, + effects: Object.freeze({ + touchedGraphs: Object.freeze([...semantics.touchedGraphs]), + }), + }) + : Object.freeze({ + mutation: captured, + guardedGraphs, + outcome: 'noop' as const, + effects: undefined, + }); + STRUCTURED_MUTATION_SNAPSHOTS.set(captured, snapshot); + return snapshot; +} + +/** Storage-internal trust check used by final materialization. */ +export function assertTrustedStructuredMutationSnapshot( + snapshot: StructuredMutationSnapshot, +): void { + if (typeof snapshot !== 'object' + || snapshot === null + || typeof snapshot.mutation !== 'object' + || snapshot.mutation === null + || STRUCTURED_MUTATION_SNAPSHOTS.get(snapshot.mutation) !== snapshot) { + throw new Error('structured mutation snapshot is not trusted'); + } +} + +function brandMutation( + kind: K, + input: Extract['input'], +): Extract { + const mutation = { kind, input } as Extract; + Object.defineProperty(mutation, STRUCTURED_MUTATION_SNAPSHOT_BRAND, { value: true }); + Object.freeze(mutation); + return mutation; +} + function unsupportedMutation(mutation: never): never { throw new Error( `Unsupported structured mutation kind ${String((mutation as { kind?: unknown })?.kind)}`, @@ -125,26 +318,29 @@ export function buildStructuredMutationUpdate(mutation: StructuredMutation): str } export function structuredMutationGuardedGraphs(mutation: StructuredMutation): readonly string[] { - switch (mutation.kind) { - case 'replace-projection-from-graph': - return [mutation.input.targetGraphUri, mutation.input.stagingGraphUri]; - case 'copy-subject-projection': - return [...mutation.input.sourceGraphUris, mutation.input.targetGraphUri]; - default: - return [mutation.input.graphUri]; - } + return structuredMutationSemantics(mutation).guardedGraphs; } export function structuredMutationTouchedGraphs(mutation: StructuredMutation): readonly string[] { - switch (mutation.kind) { - case 'replace-projection-from-graph': return [mutation.input.targetGraphUri]; - case 'copy-subject-projection': return [mutation.input.targetGraphUri]; - default: return [mutation.input.graphUri]; - } + return structuredMutationSemantics(mutation).touchedGraphs; } export function structuredMutationMightMutate(mutation: StructuredMutation): boolean { - return mutation.kind !== 'delete-subjects' || mutation.input.subjects.length > 0; + return structuredMutationSemantics(mutation).mightMutate; +} + +function structuredMutationSemantics( + mutation: StructuredMutation, +): StructuredMutationSemantics { + switch (mutation.kind) { + case 'delete-subjects': return deleteSubjectsSemantics(mutation.input); + case 'prune-ranked-subjects': return pruneRankedSubjectsSemantics(mutation.input); + case 'prune-linked-record-closures': return pruneLinkedRecordClosuresSemantics(mutation.input); + case 'replace-subject-predicates': return replaceSubjectPredicatesSemantics(mutation.input); + case 'replace-projection-from-graph': return replaceProjectionFromGraphSemantics(mutation.input); + case 'copy-subject-projection': return copySubjectProjectionSemantics(mutation.input); + default: return unsupportedMutation(mutation); + } } /** Immutable graph-scoped effects captured before a structured mutation is dispatched. */ diff --git a/packages/storage/src/changelog-store.ts b/packages/storage/src/changelog-store.ts index 8c920ed982..7cbbd2948e 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, + captureStructuredMutationSnapshot, } from './bounded-structured-mutation.js'; +import { isStructuredMutationPreDispatchRefusal } from './structured-mutation/refusal-internal.js'; /** * ChangelogStore — an append-only per-node change log maintained on the write @@ -458,27 +457,29 @@ export class ChangelogStore implements TripleStore, ChangelogReader { } async structuredMutation(mutation: StructuredMutation, options?: QueryOptions): Promise { - const normalized = normalizeStructuredMutation(mutation); - const effects = captureStructuredMutationEffects(normalized); + const snapshot = captureStructuredMutationSnapshot(mutation); const operation = this.inner.structuredMutation; if (!operation) { throw new UnsupportedTripleStoreCapabilityError('structuredMutation', 'ChangelogStore'); } - if (!this.enabled) return operation.call(this.inner, normalized, options); - for (const graph of structuredMutationGuardedGraphs(normalized)) { + if (!this.enabled) return operation.call(this.inner, snapshot.mutation, options); + for (const graph of snapshot.guardedGraphs) { this.assertNotReserved(graph, 'structuredMutation'); } await this.runExclusive(async () => { try { - await operation.call(this.inner, normalized, options); + await operation.call(this.inner, snapshot.mutation, options); } catch (error) { - if (!isTripleStoreCapabilityRefusal(error, 'structuredMutation')) { + if ( + !isTripleStoreCapabilityRefusal(error, 'structuredMutation') && + !isStructuredMutationPreDispatchRefusal(error) + ) { this.flagReconcile('structuredMutation(indeterminate-failure)'); } throw error; } - if (effects) { - await this.markPostMutation(effects.touchedGraphs, options); + if (snapshot.outcome !== 'noop') { + await this.markPostMutation(snapshot.effects.touchedGraphs, options); } }); } diff --git a/packages/storage/src/graph-set-index-store.ts b/packages/storage/src/graph-set-index-store.ts index 136507da28..a700281b35 100644 --- a/packages/storage/src/graph-set-index-store.ts +++ b/packages/storage/src/graph-set-index-store.ts @@ -23,9 +23,9 @@ import { import { isAtomicGraphReplaceStagingGraph } from './atomic-graph-replace.js'; import { ManagedOxigraphBackendUnownedError } from './managed-oxigraph-backend-unowned-error.js'; import { - captureStructuredMutationEffects, - normalizeStructuredMutation, + captureStructuredMutationSnapshot, } from './bounded-structured-mutation.js'; +import { isStructuredMutationPreDispatchRefusal } from './structured-mutation/refusal-internal.js'; import { CACHED_READ_GATE_V1, asCachedReadGateV1, @@ -557,24 +557,26 @@ export class GraphSetIndexStore implements TripleStore { } async structuredMutation(mutation: StructuredMutation, options?: QueryOptions): Promise { - const normalized = normalizeStructuredMutation(mutation); - const effects = captureStructuredMutationEffects(normalized); + const snapshot = captureStructuredMutationSnapshot(mutation); const operation = this.inner.structuredMutation; if (!operation) { throw new UnsupportedTripleStoreCapabilityError('structuredMutation', 'GraphSetIndexStore'); } try { - await operation.call(this.inner, normalized, options); + await operation.call(this.inner, snapshot.mutation, options); } catch (error) { - if (!isTripleStoreCapabilityRefusal(error, 'structuredMutation')) { + if ( + !isTripleStoreCapabilityRefusal(error, 'structuredMutation') && + !isStructuredMutationPreDispatchRefusal(error) + ) { this.scheduleFullRefresh('structuredMutation'); } throw error; } - if (!this.enabled || !effects) return; + if (!this.enabled || snapshot.outcome === 'noop') return; this.bumpMutation(); await this.maintainTouchedGraphs( - [...effects.touchedGraphs], + [...snapshot.effects.touchedGraphs], 'structuredMutation', options, ); diff --git a/packages/storage/src/index.ts b/packages/storage/src/index.ts index 8bb5c41857..1c7f2b8b73 100644 --- a/packages/storage/src/index.ts +++ b/packages/storage/src/index.ts @@ -52,10 +52,14 @@ export { export { BOUNDED_MUTATION_MAX_PRUNE_DELETE, captureStructuredMutationEffects, + captureStructuredMutationSnapshot, chunkCopySubjectProjectionInput, structuredMutationMightMutate, structuredMutationTouchedGraphs, + type ReadonlyStructuredMutation, + type ReadonlyStructuredMutationQuad, type StructuredMutationEffects, + type StructuredMutationSnapshot, } 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/src/shared-memory-literal-blob-store.ts b/packages/storage/src/shared-memory-literal-blob-store.ts index 103cba57f7..0fdd9c13ff 100644 --- a/packages/storage/src/shared-memory-literal-blob-store.ts +++ b/packages/storage/src/shared-memory-literal-blob-store.ts @@ -18,7 +18,7 @@ import { UnsupportedTripleStoreCapabilityError, } from './unsupported-capability-error.js'; import { - rewriteStructuredMutationQuads, + captureStructuredMutationSnapshot, } from './bounded-structured-mutation.js'; export const EXTERNAL_LITERAL_REF_DATATYPE = 'http://dkg.io/ontology/externalLiteralRef'; @@ -195,11 +195,19 @@ export class SharedMemoryLiteralBlobStore implements TripleStore { 'SharedMemoryLiteralBlobStore', ); } - const rewritten = await rewriteStructuredMutationQuads( - mutation, - (quad) => this.externalizeInsertQuad(quad), + const snapshot = captureStructuredMutationSnapshot(mutation); + if (snapshot.outcome === 'noop' || snapshot.mutation.kind !== 'replace-subject-predicates') { + await this.inner.structuredMutation(snapshot.mutation, options); + return; + } + const replacementQuads = await Promise.all( + snapshot.mutation.input.replacementQuads.map((quad) => this.externalizeInsertQuad(quad)), ); - await this.inner.structuredMutation(rewritten, options); + const rewritten = captureStructuredMutationSnapshot({ + kind: snapshot.mutation.kind, + input: { ...snapshot.mutation.input, replacementQuads }, + }); + await this.inner.structuredMutation(rewritten.mutation, options); } async update(sparql: string, options?: UpdateOptions): Promise { diff --git a/packages/storage/src/structured-mutation-materialization-internal.ts b/packages/storage/src/structured-mutation-materialization-internal.ts new file mode 100644 index 0000000000..f8e29441ea --- /dev/null +++ b/packages/storage/src/structured-mutation-materialization-internal.ts @@ -0,0 +1,116 @@ +import { + assertTrustedStructuredMutationSnapshot, + type ReadonlyStructuredMutation, + type StructuredMutationSnapshot, +} from './bounded-structured-mutation.js'; +import { + assertCopySubjectProjectionInputMaterializable, + buildCopySubjectProjectionUpdateFromNormalized, +} from './structured-mutation/copy-subject-projection.js'; +import { + assertDeleteSubjectsInputMaterializable, + buildDeleteSubjectsUpdateFromNormalized, +} from './structured-mutation/delete-subjects.js'; +import { + assertReplaceProjectionFromGraphInputMaterializable, + buildReplaceProjectionFromGraphUpdateFromNormalized, +} from './structured-mutation/replace-projection-from-graph.js'; +import { + assertReplaceSubjectPredicatesInputMaterializable, + buildReplaceSubjectPredicatesUpdateFromNormalized, +} from './structured-mutation/replace-subject-predicates.js'; +import { + assertPruneLinkedRecordClosuresInputMaterializable, + assertPruneRankedSubjectsInputMaterializable, + buildPruneLinkedRecordClosuresUpdateFromNormalized, + buildPruneRankedSubjectsUpdateFromNormalized, +} from './structured-mutation/retention.js'; +import { markStructuredMutationPreDispatchRefusal } from './structured-mutation/refusal-internal.js'; + +export type MaterializedStructuredMutation = + | Readonly<{ + outcome: 'noop'; + snapshot: Extract; + }> + | Readonly<{ + outcome: 'execute'; + snapshot: Extract; + update: string; + }>; + +/** Validate deferred budgets and build one executable update from a trusted snapshot. */ +export function materializeStructuredMutation( + snapshot: StructuredMutationSnapshot, +): MaterializedStructuredMutation { + assertTrustedStructuredMutationSnapshot(snapshot); + try { + const mutation = snapshot.mutation; + assertSnapshotMaterializable(mutation); + const update = buildSnapshotUpdate(mutation); + if (update === undefined) { + if (snapshot.outcome !== 'noop') { + throw new Error('structured mutation candidate unexpectedly materialized as a no-op'); + } + return Object.freeze({ outcome: 'noop', snapshot }); + } + if (snapshot.outcome !== 'candidate') { + throw new Error('structured mutation no-op unexpectedly materialized an update'); + } + return Object.freeze({ outcome: 'execute', snapshot, update }); + } catch (error) { + markStructuredMutationPreDispatchRefusal(error); + throw error; + } +} + +/** Validate a worker-bound snapshot before structured clone without building backend text. */ +export function assertStructuredMutationSnapshotMaterializable( + snapshot: StructuredMutationSnapshot, +): void { + assertTrustedStructuredMutationSnapshot(snapshot); + try { + assertSnapshotMaterializable(snapshot.mutation); + } catch (error) { + markStructuredMutationPreDispatchRefusal(error); + throw error; + } +} + +function assertSnapshotMaterializable(mutation: ReadonlyStructuredMutation): void { + switch (mutation.kind) { + case 'delete-subjects': + assertDeleteSubjectsInputMaterializable(mutation.input); + return; + case 'prune-ranked-subjects': + assertPruneRankedSubjectsInputMaterializable(mutation.input); + return; + case 'prune-linked-record-closures': + assertPruneLinkedRecordClosuresInputMaterializable(mutation.input); + return; + case 'replace-subject-predicates': + assertReplaceSubjectPredicatesInputMaterializable(mutation.input); + return; + case 'replace-projection-from-graph': + assertReplaceProjectionFromGraphInputMaterializable(mutation.input); + return; + case 'copy-subject-projection': + assertCopySubjectProjectionInputMaterializable(mutation.input); + } +} + +function buildSnapshotUpdate(mutation: ReadonlyStructuredMutation): string | undefined { + switch (mutation.kind) { + case 'delete-subjects': + return buildDeleteSubjectsUpdateFromNormalized(mutation.input); + case 'prune-ranked-subjects': + return buildPruneRankedSubjectsUpdateFromNormalized(mutation.input); + case 'prune-linked-record-closures': + return buildPruneLinkedRecordClosuresUpdateFromNormalized(mutation.input); + case 'replace-subject-predicates': + return buildReplaceSubjectPredicatesUpdateFromNormalized(mutation.input); + case 'replace-projection-from-graph': + return buildReplaceProjectionFromGraphUpdateFromNormalized(mutation.input); + case 'copy-subject-projection': + return buildCopySubjectProjectionUpdateFromNormalized(mutation.input); + } +} diff --git a/packages/storage/src/structured-mutation/capture-internal.ts b/packages/storage/src/structured-mutation/capture-internal.ts new file mode 100644 index 0000000000..3a6bb0a208 --- /dev/null +++ b/packages/storage/src/structured-mutation/capture-internal.ts @@ -0,0 +1,66 @@ +import { + absoluteIri, + boundedString, +} from './primitives.js'; + +export interface StructuredMutationSemantics { + readonly guardedGraphs: readonly string[]; + readonly touchedGraphs: readonly string[]; + readonly mightMutate: boolean; +} + +export function captureInputRecord(value: unknown, label: string): Record { + if (typeof value !== 'object' || value === null || Array.isArray(value)) { + throw new Error(`${label} must be an object`); + } + return value as Record; +} + +export function captureUniqueIris( + value: unknown, + label: string, + max: number, + allowEmpty: boolean, +): readonly string[] { + const seen = new Set(); + return captureArray(value, label, allowEmpty ? 0 : 1, max, (candidate, index) => { + const iri = absoluteIri(candidate as string, `${label}[${index}]`); + if (seen.has(iri)) throw new Error(`${label} contains duplicate IRI ${iri}`); + seen.add(iri); + return iri; + }); +} + +export function captureUniqueStrings( + value: unknown, + label: string, + max: number, +): readonly string[] { + const seen = new Set(); + return captureArray(value, label, 1, max, (candidate, index) => { + const captured = boundedString(candidate as string, `${label}[${index}]`); + if (seen.has(captured)) throw new Error(`${label} contains duplicate value ${captured}`); + seen.add(captured); + return captured; + }); +} + +export function captureArray( + value: unknown, + label: string, + min: number, + max: number, + capture: (candidate: unknown, index: number) => T, +): readonly T[] { + if (!Array.isArray(value)) throw new Error(`${label} must be an array`); + const length = value.length; + if (length < min || length > max) { + throw new Error(`${label} must contain ${min}..${max} values`); + } + const result = new Array(length); + for (let index = 0; index < length; index += 1) { + if (!(index in value)) throw new Error(`${label} must be a dense array`); + result[index] = capture(value[index], index); + } + return Object.freeze(result); +} diff --git a/packages/storage/src/structured-mutation/copy-subject-projection.ts b/packages/storage/src/structured-mutation/copy-subject-projection.ts index dbc59512e6..3511d1feae 100644 --- a/packages/storage/src/structured-mutation/copy-subject-projection.ts +++ b/packages/storage/src/structured-mutation/copy-subject-projection.ts @@ -1,6 +1,7 @@ import { sparqlString } from '@origintrail-official/dkg-core'; import type { CopySubjectProjectionInput } from '../triple-store.js'; import { + BOUNDED_MUTATION_MAX_IRIS, BOUNDED_MUTATION_MAX_PREDICATES, BOUNDED_MUTATION_MAX_SOURCE_GRAPHS, BoundedMutationBudgetError, @@ -8,66 +9,107 @@ import { assertBoundedStructuredUpdate, assertOperandBudget, boundedString, - uniqueIris, } from './primitives.js'; +import { + captureInputRecord, + captureUniqueIris, + type StructuredMutationSemantics, +} from './capture-internal.js'; -export function normalizeCopySubjectProjectionInput( - input: CopySubjectProjectionInput, -): CopySubjectProjectionInput { - const sourceGraphUris = uniqueIris( - input.sourceGraphUris, +export function captureCopySubjectProjectionInput(input: unknown): CopySubjectProjectionInput { + const value = captureInputRecord(input, 'copySubjectProjection'); + const sourceGraphUris = captureUniqueIris( + value.sourceGraphUris, 'copySubjectProjection.sourceGraphUris', BOUNDED_MUTATION_MAX_SOURCE_GRAPHS, + false, + ); + const targetGraphUri = absoluteIri( + value.targetGraphUri as string, + 'copySubjectProjection.targetGraphUri', ); - const targetGraphUri = absoluteIri(input.targetGraphUri, 'copySubjectProjection.targetGraphUri'); if (sourceGraphUris.includes(targetGraphUri)) { throw new Error('copySubjectProjection target graph must not be a source graph'); } - const roots = uniqueIris(input.roots, 'copySubjectProjection.roots'); + const roots = captureUniqueIris( + value.roots, + 'copySubjectProjection.roots', + BOUNDED_MUTATION_MAX_IRIS, + false, + ); const descendantSuffix = boundedString( - input.descendantSuffix, + value.descendantSuffix as string, 'copySubjectProjection.descendantSuffix', 256, ); if (!descendantSuffix.startsWith('/')) { throw new Error('copySubjectProjection.descendantSuffix must start with /'); } - const excludedPredicates = uniqueIris( - input.excludedPredicates, + const excludedPredicates = captureUniqueIris( + value.excludedPredicates, 'copySubjectProjection.excludedPredicates', BOUNDED_MUTATION_MAX_PREDICATES, true, ); - assertOperandBudget('copySubjectProjection', [ - ...sourceGraphUris, - targetGraphUri, - ...roots, - descendantSuffix, - ...excludedPredicates, - ]); - return { + return Object.freeze({ sourceGraphUris, targetGraphUri, roots, descendantSuffix, excludedPredicates, + }); +} + +export function copySubjectProjectionSemantics( + input: CopySubjectProjectionInput, +): StructuredMutationSemantics { + return { + guardedGraphs: [...input.sourceGraphUris, input.targetGraphUri], + touchedGraphs: [input.targetGraphUri], + mightMutate: true, }; } +export function assertCopySubjectProjectionInputMaterializable( + input: CopySubjectProjectionInput, +): void { + assertOperandBudget('copySubjectProjection', [ + ...input.sourceGraphUris, + input.targetGraphUri, + ...input.roots, + input.descendantSuffix, + ...input.excludedPredicates, + ]); +} + +export function normalizeCopySubjectProjectionInput( + input: CopySubjectProjectionInput, +): CopySubjectProjectionInput { + const captured = captureCopySubjectProjectionInput(input); + assertCopySubjectProjectionInputMaterializable(captured); + return captured; +} + export function buildCopySubjectProjectionUpdate(input: CopySubjectProjectionInput): string { const normalized = normalizeCopySubjectProjectionInput(input); - const sources = normalized.sourceGraphUris.map((graph) => `<${graph}>`).join(' '); - const roots = normalized.roots.map((root) => `<${root}>`).join(' '); - const excluded = normalized.excludedPredicates.length > 0 - ? `FILTER(?predicate NOT IN (${normalized.excludedPredicates.map((iri) => `<${iri}>`).join(', ')}))` + return buildCopySubjectProjectionUpdateFromNormalized(normalized); +} + +export function buildCopySubjectProjectionUpdateFromNormalized( + input: CopySubjectProjectionInput, +): string { + const sources = input.sourceGraphUris.map((graph) => `<${graph}>`).join(' '); + const roots = input.roots.map((root) => `<${root}>`).join(' '); + const excluded = input.excludedPredicates.length > 0 + ? `FILTER(?predicate NOT IN (${input.excludedPredicates.map((iri) => `<${iri}>`).join(', ')}))` : ''; - return assertBoundedStructuredUpdate('copySubjectProjection', `INSERT { GRAPH <${normalized.targetGraphUri}> { ?subject ?predicate ?object } } + return assertBoundedStructuredUpdate('copySubjectProjection', `INSERT { GRAPH <${input.targetGraphUri}> { ?subject ?predicate ?object } } WHERE { VALUES ?sourceGraph { ${sources} } VALUES ?root { ${roots} } # sparql-scan-allow: R2 -- ?sourceGraph is VALUES-bound to at most 8 validated exact graph IRIs GRAPH ?sourceGraph { ?subject ?predicate ?object } - FILTER(?subject = ?root || STRSTARTS(STR(?subject), CONCAT(STR(?root), ${sparqlString(normalized.descendantSuffix)}))) + FILTER(?subject = ?root || STRSTARTS(STR(?subject), CONCAT(STR(?root), ${sparqlString(input.descendantSuffix)}))) ${excluded} }`); } @@ -76,7 +118,12 @@ WHERE { export function chunkCopySubjectProjectionInput( input: CopySubjectProjectionInput, ): CopySubjectProjectionInput[] { - const roots = uniqueIris(input.roots, 'copySubjectProjection.roots'); + const roots = captureUniqueIris( + input.roots, + 'copySubjectProjection.roots', + BOUNDED_MUTATION_MAX_IRIS, + false, + ); const normalized = normalizeCopySubjectProjectionInput({ ...input, roots: [roots[0]] }); const chunks: CopySubjectProjectionInput[] = []; diff --git a/packages/storage/src/structured-mutation/delete-subjects.ts b/packages/storage/src/structured-mutation/delete-subjects.ts index 561da9a341..4502dde085 100644 --- a/packages/storage/src/structured-mutation/delete-subjects.ts +++ b/packages/storage/src/structured-mutation/delete-subjects.ts @@ -1,24 +1,60 @@ import type { DeleteSubjectsInput } from '../triple-store.js'; import { + BOUNDED_MUTATION_MAX_IRIS, absoluteIri, assertBoundedStructuredUpdate, assertOperandBudget, - uniqueIris, } from './primitives.js'; +import { + captureInputRecord, + captureUniqueIris, + type StructuredMutationSemantics, +} from './capture-internal.js'; + +export function captureDeleteSubjectsInput(input: unknown): DeleteSubjectsInput { + const value = captureInputRecord(input, 'deleteSubjects'); + const graphUri = absoluteIri(value.graphUri as string, 'deleteSubjects.graphUri'); + const subjects = captureUniqueIris( + value.subjects, + 'deleteSubjects.subjects', + BOUNDED_MUTATION_MAX_IRIS, + true, + ); + return Object.freeze({ graphUri, subjects }); +} + +export function deleteSubjectsSemantics( + input: DeleteSubjectsInput, +): StructuredMutationSemantics { + return { + guardedGraphs: [input.graphUri], + touchedGraphs: [input.graphUri], + mightMutate: input.subjects.length > 0, + }; +} + +export function assertDeleteSubjectsInputMaterializable(input: DeleteSubjectsInput): void { + assertOperandBudget('deleteSubjects', [input.graphUri, ...input.subjects]); +} export function normalizeDeleteSubjectsInput(input: DeleteSubjectsInput): DeleteSubjectsInput { - const graphUri = absoluteIri(input.graphUri, 'deleteSubjects.graphUri'); - const subjects = uniqueIris(input.subjects, 'deleteSubjects.subjects', undefined, true); - assertOperandBudget('deleteSubjects', [graphUri, ...subjects]); - return { graphUri, subjects }; + const captured = captureDeleteSubjectsInput(input); + assertDeleteSubjectsInputMaterializable(captured); + return captured; } export function buildDeleteSubjectsUpdate(input: DeleteSubjectsInput): string | undefined { const normalized = normalizeDeleteSubjectsInput(input); - if (normalized.subjects.length === 0) return undefined; - const values = normalized.subjects.map((subject) => `<${subject}>`).join(' '); - return assertBoundedStructuredUpdate('deleteSubjects', `DELETE { GRAPH <${normalized.graphUri}> { ?subject ?predicate ?object } } -WHERE { GRAPH <${normalized.graphUri}> { + return buildDeleteSubjectsUpdateFromNormalized(normalized); +} + +export function buildDeleteSubjectsUpdateFromNormalized( + input: DeleteSubjectsInput, +): string | undefined { + if (input.subjects.length === 0) return undefined; + const values = input.subjects.map((subject) => `<${subject}>`).join(' '); + return assertBoundedStructuredUpdate('deleteSubjects', `DELETE { GRAPH <${input.graphUri}> { ?subject ?predicate ?object } } +WHERE { GRAPH <${input.graphUri}> { VALUES ?subject { ${values} } ?subject ?predicate ?object } }`); diff --git a/packages/storage/src/structured-mutation/primitives.ts b/packages/storage/src/structured-mutation/primitives.ts index 4ea6674e56..b195256ad5 100644 --- a/packages/storage/src/structured-mutation/primitives.ts +++ b/packages/storage/src/structured-mutation/primitives.ts @@ -13,6 +13,12 @@ const UTF8 = new TextEncoder(); export class BoundedMutationBudgetError extends Error {} +export function isBoundedMutationBudgetError( + error: unknown, +): error is BoundedMutationBudgetError { + return error instanceof BoundedMutationBudgetError; +} + export function boundedInteger(value: number, label: string, max: number): number { if (!Number.isSafeInteger(value) || value < 0 || value > max) { throw new Error(`${label} must be an integer in 0..${max}`); @@ -86,7 +92,7 @@ export function boundedUniqueStrings( return result; } -export function assertOperandBudget(label: string, values: readonly string[]): void { +export function assertOperandBudget(label: string, values: Iterable): void { let bytes = 0; for (const value of values) { bytes += UTF8.encode(value).byteLength; diff --git a/packages/storage/src/structured-mutation/refusal-internal.ts b/packages/storage/src/structured-mutation/refusal-internal.ts new file mode 100644 index 0000000000..3851404263 --- /dev/null +++ b/packages/storage/src/structured-mutation/refusal-internal.ts @@ -0,0 +1,40 @@ +import { isBoundedMutationBudgetError } from './primitives.js'; + +export const STRUCTURED_MUTATION_PRE_DISPATCH_REFUSAL_CODE = + 'STRUCTURED_MUTATION_PRE_DISPATCH_REFUSAL'; + +const TRUSTED_STRUCTURED_MUTATION_REFUSALS = new WeakSet(); + +export function isStructuredMutationPreDispatchRefusal( + error: unknown, +): boolean { + return typeof error === 'object' + && error !== null + && TRUSTED_STRUCTURED_MUTATION_REFUSALS.has(error); +} + +/** Mark a budget error only while still inside the explicit pre-dispatch boundary. */ +export function markStructuredMutationPreDispatchRefusal(error: unknown): void { + if (isBoundedMutationBudgetError(error)) { + TRUSTED_STRUCTURED_MUTATION_REFUSALS.add(error); + } +} + +/** Reconstruct the private refusal identity after a trusted worker reports it. */ +export function reconstructStructuredMutationPreDispatchRefusal(message: string): Error { + const error = new Error(message); + Object.defineProperty(error, 'code', { + value: STRUCTURED_MUTATION_PRE_DISPATCH_REFUSAL_CODE, + enumerable: true, + }); + TRUSTED_STRUCTURED_MUTATION_REFUSALS.add(error); + return error; +} + +export function structuredMutationPreDispatchRefusalCode( + error: unknown, +): typeof STRUCTURED_MUTATION_PRE_DISPATCH_REFUSAL_CODE | undefined { + return isStructuredMutationPreDispatchRefusal(error) + ? STRUCTURED_MUTATION_PRE_DISPATCH_REFUSAL_CODE + : undefined; +} diff --git a/packages/storage/src/structured-mutation/replace-projection-from-graph.ts b/packages/storage/src/structured-mutation/replace-projection-from-graph.ts index d74bdc68fe..1dd57db51c 100644 --- a/packages/storage/src/structured-mutation/replace-projection-from-graph.ts +++ b/packages/storage/src/structured-mutation/replace-projection-from-graph.ts @@ -6,90 +6,124 @@ import { absoluteIri, assertBoundedStructuredUpdate, assertOperandBudget, - uniqueIris, } from './primitives.js'; +import { + captureInputRecord, + captureUniqueIris, + type StructuredMutationSemantics, +} from './capture-internal.js'; -export function normalizeReplaceProjectionFromGraphInput( - input: ReplaceProjectionFromGraphInput, +export function captureReplaceProjectionFromGraphInput( + input: unknown, ): ReplaceProjectionFromGraphInput { + const value = captureInputRecord(input, 'replaceProjectionFromGraph'); const targetGraphUri = absoluteIri( - input.targetGraphUri, + value.targetGraphUri as string, 'replaceProjectionFromGraph.targetGraphUri', ); const stagingGraphUri = absoluteIri( - input.stagingGraphUri, + value.stagingGraphUri as string, 'replaceProjectionFromGraph.stagingGraphUri', ); if (targetGraphUri === stagingGraphUri) { throw new Error('replaceProjectionFromGraph requires distinct target and staging graphs'); } const targetSubject = absoluteIri( - input.targetSubject, + value.targetSubject as string, 'replaceProjectionFromGraph.targetSubject', ); - const preservedTargetPredicates = uniqueIris( - input.preservedTargetPredicates, + const preservedTargetPredicates = captureUniqueIris( + value.preservedTargetPredicates, 'replaceProjectionFromGraph.preservedTargetPredicates', BOUNDED_MUTATION_MAX_PREDICATES, true, ); - const targetSubjectPrefixes = uniqueIris( - input.targetSubjectPrefixes, + const targetSubjectPrefixes = captureUniqueIris( + value.targetSubjectPrefixes, 'replaceProjectionFromGraph.targetSubjectPrefixes', BOUNDED_MUTATION_MAX_PREFIXES, true, ); - assertOperandBudget('replaceProjectionFromGraph', [ - targetGraphUri, - stagingGraphUri, - targetSubject, - ...preservedTargetPredicates, - ...targetSubjectPrefixes, - ]); - return { + return Object.freeze({ targetGraphUri, stagingGraphUri, targetSubject, preservedTargetPredicates, targetSubjectPrefixes, + }); +} + +export function replaceProjectionFromGraphSemantics( + input: ReplaceProjectionFromGraphInput, +): StructuredMutationSemantics { + return { + guardedGraphs: [input.targetGraphUri, input.stagingGraphUri], + touchedGraphs: [input.targetGraphUri], + mightMutate: true, }; } +export function assertReplaceProjectionFromGraphInputMaterializable( + input: ReplaceProjectionFromGraphInput, +): void { + assertOperandBudget('replaceProjectionFromGraph', [ + input.targetGraphUri, + input.stagingGraphUri, + input.targetSubject, + ...input.preservedTargetPredicates, + ...input.targetSubjectPrefixes, + ]); +} + +export function normalizeReplaceProjectionFromGraphInput( + input: ReplaceProjectionFromGraphInput, +): ReplaceProjectionFromGraphInput { + const captured = captureReplaceProjectionFromGraphInput(input); + assertReplaceProjectionFromGraphInputMaterializable(captured); + return captured; +} + export function buildReplaceProjectionFromGraphUpdate( input: ReplaceProjectionFromGraphInput, ): string { const normalized = normalizeReplaceProjectionFromGraphInput(input); - const preserved = normalized.preservedTargetPredicates.length > 0 - ? ` && ?stalePredicate NOT IN (${normalized.preservedTargetPredicates.map((iri) => `<${iri}>`).join(', ')})` + return buildReplaceProjectionFromGraphUpdateFromNormalized(normalized); +} + +export function buildReplaceProjectionFromGraphUpdateFromNormalized( + input: ReplaceProjectionFromGraphInput, +): string { + const preserved = input.preservedTargetPredicates.length > 0 + ? ` && ?stalePredicate NOT IN (${input.preservedTargetPredicates.map((iri) => `<${iri}>`).join(', ')})` : ''; - const prefixes = normalized.targetSubjectPrefixes + const prefixes = input.targetSubjectPrefixes .map((prefix) => `STRSTARTS(STR(?staleSubject), ${sparqlString(prefix)})`); const staleScopes = [ - `(?staleSubject = <${normalized.targetSubject}>${preserved})`, + `(?staleSubject = <${input.targetSubject}>${preserved})`, ...prefixes, ].join(' || '); const freshScopes = [ - `?freshSubject = <${normalized.targetSubject}>`, - ...normalized.targetSubjectPrefixes.map( + `?freshSubject = <${input.targetSubject}>`, + ...input.targetSubjectPrefixes.map( (prefix) => `STRSTARTS(STR(?freshSubject), ${sparqlString(prefix)})`, ), ].join(' || '); return assertBoundedStructuredUpdate('replaceProjectionFromGraph', `DELETE { - GRAPH <${normalized.targetGraphUri}> { ?staleSubject ?stalePredicate ?staleObject } + GRAPH <${input.targetGraphUri}> { ?staleSubject ?stalePredicate ?staleObject } } INSERT { - GRAPH <${normalized.targetGraphUri}> { ?freshSubject ?freshPredicate ?freshObject } + GRAPH <${input.targetGraphUri}> { ?freshSubject ?freshPredicate ?freshObject } } WHERE { { - GRAPH <${normalized.targetGraphUri}> { + GRAPH <${input.targetGraphUri}> { ?staleSubject ?stalePredicate ?staleObject . FILTER(${staleScopes}) } } UNION { - GRAPH <${normalized.stagingGraphUri}> { + GRAPH <${input.stagingGraphUri}> { ?freshSubject ?freshPredicate ?freshObject . FILTER(${freshScopes}) } diff --git a/packages/storage/src/structured-mutation/replace-subject-predicates.ts b/packages/storage/src/structured-mutation/replace-subject-predicates.ts index 14c0408e89..79e8940046 100644 --- a/packages/storage/src/structured-mutation/replace-subject-predicates.ts +++ b/packages/storage/src/structured-mutation/replace-subject-predicates.ts @@ -6,100 +6,132 @@ import { assertBoundedStructuredUpdate, assertOperandBudget, normalizeStructuredRdfObject, - uniqueIris, } from './primitives.js'; +import { + captureArray, + captureInputRecord, + captureUniqueIris, + type StructuredMutationSemantics, +} from './capture-internal.js'; -function normalizeReplaceSubjectPredicatesInputInternal( - input: ReplaceSubjectPredicatesInput, - enforceOperandBudget: boolean, +export function captureReplaceSubjectPredicatesInput( + input: unknown, ): ReplaceSubjectPredicatesInput { - const graphUri = absoluteIri(input.graphUri, 'replaceSubjectPredicates.graphUri'); - const subject = absoluteIri(input.subject, 'replaceSubjectPredicates.subject'); - const predicates = uniqueIris( - input.predicates, + const value = captureInputRecord(input, 'replaceSubjectPredicates'); + const graphUri = absoluteIri(value.graphUri as string, 'replaceSubjectPredicates.graphUri'); + const subject = absoluteIri(value.subject as string, 'replaceSubjectPredicates.subject'); + const predicates = captureUniqueIris( + value.predicates, 'replaceSubjectPredicates.predicates', BOUNDED_MUTATION_MAX_PREDICATES, + false, ); - if (!Array.isArray(input.replacementQuads) - || input.replacementQuads.length > BOUNDED_MUTATION_MAX_IRIS) { - throw new Error( - `replaceSubjectPredicates.replacementQuads must contain 0..${BOUNDED_MUTATION_MAX_IRIS} quads`, - ); - } - for (let index = 0; index < input.replacementQuads.length; index++) { - if (!(index in input.replacementQuads)) { - throw new Error('replaceSubjectPredicates.replacementQuads must be a dense array'); - } - } const allowedPredicates = new Set(predicates); - const replacementQuads = input.replacementQuads.map((quad, index) => { - if (quad.graph !== graphUri || quad.subject !== subject) { - throw new Error( - `replaceSubjectPredicates quad ${index} must target subject ${subject} in graph ${graphUri}`, + const replacementQuads = captureArray( + value.replacementQuads, + 'replaceSubjectPredicates.replacementQuads', + 0, + BOUNDED_MUTATION_MAX_IRIS, + (candidate, index) => { + const quad = captureInputRecord( + candidate, + `replaceSubjectPredicates.replacementQuads[${index}]`, ); - } - const predicate = absoluteIri( - quad.predicate, - `replaceSubjectPredicates.replacementQuads[${index}].predicate`, - ); - if (!allowedPredicates.has(predicate)) { - throw new Error(`replaceSubjectPredicates quad ${index} targets undeclared predicate ${predicate}`); - } - const object = normalizeStructuredRdfObject( - quad.object, - `replaceSubjectPredicates.replacementQuads[${index}].object`, - ); - return { ...quad, predicate, object }; - }); - if (enforceOperandBudget) { - assertOperandBudget('replaceSubjectPredicates', [ - graphUri, - subject, - ...predicates, - ...replacementQuads.flatMap((quad) => [ - quad.subject, - quad.predicate, - quad.object, - quad.graph, - ]), - ]); + const quadSubject = quad.subject; + const quadPredicate = quad.predicate; + const quadObject = quad.object; + const quadGraph = quad.graph; + if (quadGraph !== graphUri || quadSubject !== subject) { + throw new Error( + `replaceSubjectPredicates quad ${index} must target subject ${subject} in graph ${graphUri}`, + ); + } + const predicate = absoluteIri( + quadPredicate as string, + `replaceSubjectPredicates.replacementQuads[${index}].predicate`, + ); + if (!allowedPredicates.has(predicate)) { + throw new Error( + `replaceSubjectPredicates quad ${index} targets undeclared predicate ${predicate}`, + ); + } + const object = normalizeStructuredRdfObject( + quadObject as string, + `replaceSubjectPredicates.replacementQuads[${index}].object`, + ); + return Object.freeze({ subject, predicate, object, graph: graphUri }); + }, + ); + return Object.freeze({ graphUri, subject, predicates, replacementQuads }); +} + +export function replaceSubjectPredicatesSemantics( + input: ReplaceSubjectPredicatesInput, +): StructuredMutationSemantics { + return { guardedGraphs: [input.graphUri], touchedGraphs: [input.graphUri], mightMutate: true }; +} + +export function assertReplaceSubjectPredicatesInputMaterializable( + input: ReplaceSubjectPredicatesInput, +): void { + assertOperandBudget('replaceSubjectPredicates', replaceSubjectPredicatesOperands(input)); +} + +function* replaceSubjectPredicatesOperands( + input: ReplaceSubjectPredicatesInput, +): Iterable { + yield input.graphUri; + yield input.subject; + yield* input.predicates; + for (const quad of input.replacementQuads) { + yield quad.subject; + yield quad.predicate; + yield quad.object; + yield quad.graph; } - return { graphUri, subject, predicates, replacementQuads }; } export function normalizeReplaceSubjectPredicatesInput( input: ReplaceSubjectPredicatesInput, ): ReplaceSubjectPredicatesInput { - return normalizeReplaceSubjectPredicatesInputInternal(input, true); + const captured = captureReplaceSubjectPredicatesInput(input); + assertReplaceSubjectPredicatesInputMaterializable(captured); + return captured; } /** Validate before a storage decorator rewrites RDF objects. */ export function normalizeReplaceSubjectPredicatesInputForObjectRewrite( input: ReplaceSubjectPredicatesInput, ): ReplaceSubjectPredicatesInput { - return normalizeReplaceSubjectPredicatesInputInternal(input, false); + return captureReplaceSubjectPredicatesInput(input); } export function buildReplaceSubjectPredicatesUpdate( input: ReplaceSubjectPredicatesInput, ): string { const normalized = normalizeReplaceSubjectPredicatesInput(input); - const predicateValues = normalized.predicates.map((predicate) => `<${predicate}>`).join(', '); - const insertion = normalized.replacementQuads.length > 0 + return buildReplaceSubjectPredicatesUpdateFromNormalized(normalized); +} + +export function buildReplaceSubjectPredicatesUpdateFromNormalized( + input: ReplaceSubjectPredicatesInput, +): string { + const predicateValues = input.predicates.map((predicate) => `<${predicate}>`).join(', '); + const insertion = input.replacementQuads.length > 0 ? `INSERT { - GRAPH <${normalized.graphUri}> { -${normalized.replacementQuads.map((quad) => ` <${quad.subject}> <${quad.predicate}> ${quad.object} .`).join('\n')} + GRAPH <${input.graphUri}> { +${input.replacementQuads.map((quad) => ` <${quad.subject}> <${quad.predicate}> ${quad.object} .`).join('\n')} } } ` : ''; return assertBoundedStructuredUpdate('replaceSubjectPredicates', `DELETE { - GRAPH <${normalized.graphUri}> { <${normalized.subject}> ?predicate ?oldObject } + GRAPH <${input.graphUri}> { <${input.subject}> ?predicate ?oldObject } } ${insertion}WHERE { OPTIONAL { - GRAPH <${normalized.graphUri}> { - <${normalized.subject}> ?predicate ?oldObject . + GRAPH <${input.graphUri}> { + <${input.subject}> ?predicate ?oldObject . FILTER(?predicate IN (${predicateValues})) } } diff --git a/packages/storage/src/structured-mutation/retention.ts b/packages/storage/src/structured-mutation/retention.ts index a46e488eb1..d040bc48c9 100644 --- a/packages/storage/src/structured-mutation/retention.ts +++ b/packages/storage/src/structured-mutation/retention.ts @@ -12,52 +12,50 @@ import { assertOperandBudget, boundedInteger, boundedString, - boundedUniqueStrings, - uniqueIris, } from './primitives.js'; +import { + captureInputRecord, + captureUniqueIris, + captureUniqueStrings, + type StructuredMutationSemantics, +} from './capture-internal.js'; -export function normalizePruneRankedSubjectsInput( - input: PruneRankedSubjectsInput, -): PruneRankedSubjectsInput { - const graphUri = absoluteIri(input.graphUri, 'pruneRankedSubjects.graphUri'); - const subjectPrefix = absoluteIri(input.subjectPrefix, 'pruneRankedSubjects.subjectPrefix'); +export function capturePruneRankedSubjectsInput(input: unknown): PruneRankedSubjectsInput { + const value = captureInputRecord(input, 'pruneRankedSubjects'); + const graphUri = absoluteIri(value.graphUri as string, 'pruneRankedSubjects.graphUri'); + const subjectPrefix = absoluteIri( + value.subjectPrefix as string, + 'pruneRankedSubjects.subjectPrefix', + ); const eligibilityPredicate = absoluteIri( - input.eligibilityPredicate, + value.eligibilityPredicate as string, 'pruneRankedSubjects.eligibilityPredicate', ); + const eligibleObjects = captureUniqueStrings( + value.eligibleObjects, + 'pruneRankedSubjects.eligibleObjects', + 16, + ); const primaryRankPredicate = absoluteIri( - input.primaryRankPredicate, + value.primaryRankPredicate as string, 'pruneRankedSubjects.primaryRankPredicate', ); const secondaryRankPredicate = absoluteIri( - input.secondaryRankPredicate, + value.secondaryRankPredicate as string, 'pruneRankedSubjects.secondaryRankPredicate', ); - const eligibleObjects = boundedUniqueStrings( - input.eligibleObjects, - 'pruneRankedSubjects.eligibleObjects', - 16, - ); const retainNewest = boundedInteger( - input.retainNewest, + value.retainNewest as number, 'pruneRankedSubjects.retainNewest', BOUNDED_MUTATION_MAX_IRIS, ); const maxDelete = boundedInteger( - input.maxDelete, + value.maxDelete as number, 'pruneRankedSubjects.maxDelete', BOUNDED_MUTATION_MAX_PRUNE_DELETE, ); if (maxDelete === 0) throw new Error('pruneRankedSubjects.maxDelete must be positive'); - assertOperandBudget('pruneRankedSubjects', [ - graphUri, - subjectPrefix, - eligibilityPredicate, - primaryRankPredicate, - secondaryRankPredicate, - ...eligibleObjects, - ]); - return { + return Object.freeze({ graphUri, subjectPrefix, eligibilityPredicate, @@ -66,31 +64,126 @@ export function normalizePruneRankedSubjectsInput( secondaryRankPredicate, retainNewest, maxDelete, - }; + }); +} + +export function pruneRankedSubjectsSemantics( + input: PruneRankedSubjectsInput, +): StructuredMutationSemantics { + return { guardedGraphs: [input.graphUri], touchedGraphs: [input.graphUri], mightMutate: true }; +} + +export function assertPruneRankedSubjectsInputMaterializable( + input: PruneRankedSubjectsInput, +): void { + assertOperandBudget('pruneRankedSubjects', [ + input.graphUri, + input.subjectPrefix, + input.eligibilityPredicate, + input.primaryRankPredicate, + input.secondaryRankPredicate, + ...input.eligibleObjects, + ]); +} + +export function capturePruneLinkedRecordClosuresInput( + input: unknown, +): PruneLinkedRecordClosuresInput { + const value = captureInputRecord(input, 'pruneLinkedRecordClosures'); + const graphUri = absoluteIri(value.graphUri as string, 'pruneLinkedRecordClosures.graphUri'); + const matchObjectIris = captureUniqueIris( + value.matchObjectIris, + 'pruneLinkedRecordClosures.matchObjectIris', + BOUNDED_MUTATION_MAX_IRIS, + false, + ); + const linkPredicates = captureUniqueIris( + value.linkPredicates, + 'pruneLinkedRecordClosures.linkPredicates', + BOUNDED_MUTATION_MAX_PREDICATES, + false, + ); + const recordParentPredicate = absoluteIri( + value.recordParentPredicate as string, + 'pruneLinkedRecordClosures.recordParentPredicate', + ); + const rawProtectedRecordIri = value.protectedRecordIri; + const protectedRecordIri = rawProtectedRecordIri === undefined + ? undefined + : absoluteIri( + rawProtectedRecordIri as string, + 'pruneLinkedRecordClosures.protectedRecordIri', + ); + const descendantSeparator = boundedString( + value.descendantSeparator as string, + 'pruneLinkedRecordClosures.descendantSeparator', + 64, + ); + return Object.freeze({ + graphUri, + matchObjectIris, + linkPredicates, + recordParentPredicate, + protectedRecordIri, + descendantSeparator, + }); +} + +export function pruneLinkedRecordClosuresSemantics( + input: PruneLinkedRecordClosuresInput, +): StructuredMutationSemantics { + return { guardedGraphs: [input.graphUri], touchedGraphs: [input.graphUri], mightMutate: true }; +} + +export function assertPruneLinkedRecordClosuresInputMaterializable( + input: PruneLinkedRecordClosuresInput, +): void { + assertOperandBudget('pruneLinkedRecordClosures', [ + input.graphUri, + ...input.matchObjectIris, + ...input.linkPredicates, + input.recordParentPredicate, + input.descendantSeparator, + ...(input.protectedRecordIri ? [input.protectedRecordIri] : []), + ]); +} + +export function normalizePruneRankedSubjectsInput( + input: PruneRankedSubjectsInput, +): PruneRankedSubjectsInput { + const captured = capturePruneRankedSubjectsInput(input); + assertPruneRankedSubjectsInputMaterializable(captured); + return captured; } export function buildPruneRankedSubjectsUpdate(input: PruneRankedSubjectsInput): string { const normalized = normalizePruneRankedSubjectsInput(input); - const eligibleObjects = normalized.eligibleObjects.map(sparqlString).join(' '); + return buildPruneRankedSubjectsUpdateFromNormalized(normalized); +} + +export function buildPruneRankedSubjectsUpdateFromNormalized( + input: PruneRankedSubjectsInput, +): string { + const eligibleObjects = input.eligibleObjects.map(sparqlString).join(' '); const eligibilityFilter = `VALUES ?eligibleObject { ${eligibleObjects} } FILTER NOT EXISTS { - ?subject <${normalized.eligibilityPredicate}> ?ineligibleObject . - FILTER(?ineligibleObject NOT IN (${normalized.eligibleObjects.map(sparqlString).join(', ')})) + ?subject <${input.eligibilityPredicate}> ?ineligibleObject . + FILTER(?ineligibleObject NOT IN (${input.eligibleObjects.map(sparqlString).join(', ')})) }`; return assertBoundedStructuredUpdate('pruneRankedSubjects', `PREFIX xsd: -DELETE { GRAPH <${normalized.graphUri}> { ?subject ?predicate ?object } } +DELETE { GRAPH <${input.graphUri}> { ?subject ?predicate ?object } } WHERE { { SELECT ?subject (MAX(?primaryRank) AS ?latestPrimaryRank) (MAX(?secondaryRank) AS ?latestSecondaryRank) WHERE { - GRAPH <${normalized.graphUri}> { - ?subject <${normalized.eligibilityPredicate}> ?eligibleObject . - OPTIONAL { ?subject <${normalized.primaryRankPredicate}> ?primaryRank } - OPTIONAL { ?subject <${normalized.secondaryRankPredicate}> ?secondaryRank } + GRAPH <${input.graphUri}> { + ?subject <${input.eligibilityPredicate}> ?eligibleObject . + OPTIONAL { ?subject <${input.primaryRankPredicate}> ?primaryRank } + OPTIONAL { ?subject <${input.secondaryRankPredicate}> ?secondaryRank } ${eligibilityFilter} - FILTER(STRSTARTS(STR(?subject), ${sparqlString(normalized.subjectPrefix)})) + FILTER(STRSTARTS(STR(?subject), ${sparqlString(input.subjectPrefix)})) } } # sparql-scan-allow: R3 -- one bounded retention prune; OFFSET <= 100000 and LIMIT <= 10000; never page-walked @@ -100,13 +193,13 @@ WHERE { xsd:integer(STR(?latestSecondaryRank)), 0 )) DESC(STR(?subject)) - OFFSET ${sparqlInt(normalized.retainNewest, { min: 0 })} - LIMIT ${sparqlInt(normalized.maxDelete, { min: 1 })} + OFFSET ${sparqlInt(input.retainNewest, { min: 0 })} + LIMIT ${sparqlInt(input.maxDelete, { min: 1 })} } - GRAPH <${normalized.graphUri}> { - ?subject <${normalized.eligibilityPredicate}> ?eligibleObject ; ?predicate ?object . + GRAPH <${input.graphUri}> { + ?subject <${input.eligibilityPredicate}> ?eligibleObject ; ?predicate ?object . ${eligibilityFilter} - FILTER(STRSTARTS(STR(?subject), ${sparqlString(normalized.subjectPrefix)})) + FILTER(STRSTARTS(STR(?subject), ${sparqlString(input.subjectPrefix)})) } }`); } @@ -114,64 +207,35 @@ WHERE { export function normalizePruneLinkedRecordClosuresInput( input: PruneLinkedRecordClosuresInput, ): PruneLinkedRecordClosuresInput { - const graphUri = absoluteIri(input.graphUri, 'pruneLinkedRecordClosures.graphUri'); - const matchObjectIris = uniqueIris( - input.matchObjectIris, - 'pruneLinkedRecordClosures.matchObjectIris', - ); - const linkPredicates = uniqueIris( - input.linkPredicates, - 'pruneLinkedRecordClosures.linkPredicates', - BOUNDED_MUTATION_MAX_PREDICATES, - ); - const recordParentPredicate = absoluteIri( - input.recordParentPredicate, - 'pruneLinkedRecordClosures.recordParentPredicate', - ); - const protectedRecordIri = input.protectedRecordIri === undefined - ? undefined - : absoluteIri(input.protectedRecordIri, 'pruneLinkedRecordClosures.protectedRecordIri'); - const descendantSeparator = boundedString( - input.descendantSeparator, - 'pruneLinkedRecordClosures.descendantSeparator', - 64, - ); - assertOperandBudget('pruneLinkedRecordClosures', [ - graphUri, - ...matchObjectIris, - ...linkPredicates, - recordParentPredicate, - descendantSeparator, - ...(protectedRecordIri ? [protectedRecordIri] : []), - ]); - return { - graphUri, - matchObjectIris, - linkPredicates, - recordParentPredicate, - protectedRecordIri, - descendantSeparator, - }; + const captured = capturePruneLinkedRecordClosuresInput(input); + assertPruneLinkedRecordClosuresInputMaterializable(captured); + return captured; } export function buildPruneLinkedRecordClosuresUpdate(input: PruneLinkedRecordClosuresInput): string { const normalized = normalizePruneLinkedRecordClosuresInput(input); - const matchObjects = normalized.matchObjectIris.map((root) => `<${root}>`).join(' '); - const linkPredicates = normalized.linkPredicates + return buildPruneLinkedRecordClosuresUpdateFromNormalized(normalized); +} + +export function buildPruneLinkedRecordClosuresUpdateFromNormalized( + input: PruneLinkedRecordClosuresInput, +): string { + const matchObjects = input.matchObjectIris.map((root) => `<${root}>`).join(' '); + const linkPredicates = input.linkPredicates .map((predicate) => `<${predicate}>`) .join(' '); - const keepFilter = normalized.protectedRecordIri - ? `FILTER(?record != <${normalized.protectedRecordIri}>)` + const keepFilter = input.protectedRecordIri + ? `FILTER(?record != <${input.protectedRecordIri}>)` : ''; - return assertBoundedStructuredUpdate('pruneLinkedRecordClosures', `DELETE { GRAPH <${normalized.graphUri}> { ?subject ?predicate ?object } } -WHERE { GRAPH <${normalized.graphUri}> { + return assertBoundedStructuredUpdate('pruneLinkedRecordClosures', `DELETE { GRAPH <${input.graphUri}> { ?subject ?predicate ?object } } +WHERE { GRAPH <${input.graphUri}> { VALUES ?matchObject { ${matchObjects} } VALUES ?linkPredicate { ${linkPredicates} } ?member ?linkPredicate ?matchObject . - OPTIONAL { ?member <${normalized.recordParentPredicate}> ?parent } + OPTIONAL { ?member <${input.recordParentPredicate}> ?parent } BIND(COALESCE(?parent, ?member) AS ?record) ${keepFilter} ?subject ?predicate ?object . - FILTER(?subject = ?record || STRSTARTS(STR(?subject), CONCAT(STR(?record), ${sparqlString(normalized.descendantSeparator)}))) + FILTER(?subject = ?record || STRSTARTS(STR(?subject), CONCAT(STR(?record), ${sparqlString(input.descendantSeparator)}))) } }`); } diff --git a/packages/storage/test/blazegraph.unit.test.ts b/packages/storage/test/blazegraph.unit.test.ts index 12946bb012..34c3796bd5 100644 --- a/packages/storage/test/blazegraph.unit.test.ts +++ b/packages/storage/test/blazegraph.unit.test.ts @@ -687,6 +687,17 @@ describe('BlazegraphStore (mocked HTTP)', () => { expect(body).toContain(' "approved"'); }); + it('structuredMutation sends no request for a valid empty delete', async () => { + const s = new BlazegraphStore(baseUrl); + + await s.structuredMutation({ + kind: 'delete-subjects', + input: { graphUri: 'http://ex.org/g', subjects: [] }, + }); + + expect(fetchCalls).toHaveLength(0); + }); + it('structuredMutation rejects oversized replacement literals before fetch', async () => { const s = new BlazegraphStore(baseUrl); await expect(s.structuredMutation({ kind: 'replace-subject-predicates', input: { diff --git a/packages/storage/test/bounded-structured-mutation.test.ts b/packages/storage/test/bounded-structured-mutation.test.ts index ed4ee8ff63..9ad3c2bcba 100644 --- a/packages/storage/test/bounded-structured-mutation.test.ts +++ b/packages/storage/test/bounded-structured-mutation.test.ts @@ -1,3 +1,5 @@ +import { createHash } from 'node:crypto'; + import { describe, expect, it, vi } from 'vitest'; import { @@ -20,9 +22,11 @@ import { } from '../src/index.js'; import { BOUNDED_MUTATION_MAX_IRIS, + BOUNDED_MUTATION_MAX_OPERAND_BYTES, BOUNDED_MUTATION_MAX_UPDATE_BYTES, buildCopySubjectProjectionUpdate, buildDeleteSubjectsUpdate, + buildStructuredMutationUpdate, chunkCopySubjectProjectionInput, normalizeCopySubjectProjectionInput, normalizeDeleteSubjectsInput, @@ -49,6 +53,80 @@ async function rows(store: TripleStore, graph: string): Promise { + it('preserves the canonical update bytes for every mutation kind and the no-op shape', () => { + const fixtures: ReadonlyArray = [ + ['delete-subjects', { + kind: 'delete-subjects', + input: { graphUri: GRAPH, subjects: ['urn:test:a', 'urn:test:b'] }, + }, 184, '1816e8bb2a177b1a5abc7b66b0d716b9ecfa39caccbc57280add47bc25b7f775'], + ['prune-ranked-subjects', { + kind: 'prune-ranked-subjects', + input: { + graphUri: GRAPH, + subjectPrefix: 'urn:test:req:', + eligibilityPredicate: STATUS, + eligibleObjects: ['approved', 'rejected'], + primaryRankPredicate: DECIDED_AT, + secondaryRankPredicate: REQUESTED_AT, + retainNewest: 5, + maxDelete: 10, + }, + }, 1_460, 'a4383f845f8eaa592d6e48932c9d1a37623d933e7c550970fea711d040a9f491'], + ['prune-linked-record-closures', { + kind: 'prune-linked-record-closures', + input: { + graphUri: GRAPH, + matchObjectIris: ['urn:test:agent'], + linkPredicates: [P], + recordParentPredicate: STATUS, + protectedRecordIri: 'urn:test:keep', + descendantSeparator: '/', + }, + }, 478, 'a5b262cba67f38f7467316f2e39f917c9c51a96593a10d55c8f1e589a1df9a1f'], + ['replace-subject-predicates', { + kind: 'replace-subject-predicates', + input: { + graphUri: GRAPH, + subject: 'urn:test:a', + predicates: [P], + replacementQuads: [quad('urn:test:a', P, '"value"')], + }, + }, 310, '63257772a038004b2043e8946de4f8da682e90e7bcf6156807520966e3f256b8'], + ['replace-projection-from-graph', { + kind: 'replace-projection-from-graph', + input: { + targetGraphUri: GRAPH, + stagingGraphUri: OTHER_GRAPH, + targetSubject: 'urn:test:a', + preservedTargetPredicates: [P], + targetSubjectPrefixes: ['urn:test:child:'], + }, + }, 618, '1f15cafd7c386ee0f32f742d9188bb53d04e8b026c0aa984302603a2b615fe21'], + ['copy-subject-projection', { + kind: 'copy-subject-projection', + input: { + sourceGraphUris: [GRAPH], + targetGraphUri: OTHER_GRAPH, + roots: ['urn:test:a'], + descendantSuffix: '/', + excludedPredicates: [P], + }, + }, 434, '16d90e11bc91e1aaa049a5f56b4f55ec03e27090205010fb3a76e64508685ec1'], + ['delete-subjects/noop', { + kind: 'delete-subjects', + input: { graphUri: GRAPH, subjects: [] }, + }, 0, 'e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855'], + ]; + + for (const [label, mutation, expectedBytes, expectedHash] of fixtures) { + const update = buildStructuredMutationUpdate(mutation); + const bytes = update === undefined ? Buffer.alloc(0) : Buffer.from(update, 'utf8'); + expect(bytes.byteLength, label).toBe(expectedBytes); + expect(createHash('sha256').update(bytes).digest('hex'), label).toBe(expectedHash); + expect(update === undefined, label).toBe(label.endsWith('/noop')); + } + }); + it('captures immutable graph effects before dispatch', () => { const mutation = { kind: 'copy-subject-projection' as const, @@ -363,6 +441,79 @@ describe('bounded structured mutation capabilities', () => { expect(update).toHaveBeenCalledTimes(1); }); + it('keeps valid empty deletes I/O-free and rejects oversized worker no-ops before posting', async () => { + const embedded = new OxigraphStore(); + const embeddedStore = (embedded as unknown as { + store: { update: (sparql: string) => void }; + }).store; + const update = vi.spyOn(embeddedStore, 'update'); + const embeddedGeneration = embedded.getWriteGen(GRAPH); + + await embedded.structuredMutation({ + kind: 'delete-subjects', + input: { graphUri: GRAPH, subjects: [] }, + }); + + expect(update).not.toHaveBeenCalled(); + expect(embedded.getWriteGen(GRAPH)).toBe(embeddedGeneration); + + const worker = new OxigraphWorkerStore(); + const workerInternals = worker as unknown as { nextId: number }; + try { + const nextId = workerInternals.nextId; + const workerGeneration = worker.getWriteGen(GRAPH); + + await worker.structuredMutation({ + kind: 'delete-subjects', + input: { graphUri: GRAPH, subjects: [] }, + }); + + expect(workerInternals.nextId).toBe(nextId); + expect(worker.getWriteGen(GRAPH)).toBe(workerGeneration); + + const oversizedGraph = `urn:test:${'x'.repeat(BOUNDED_MUTATION_MAX_OPERAND_BYTES + 1)}`; + await expect(worker.structuredMutation({ + kind: 'delete-subjects', + input: { graphUri: oversizedGraph, subjects: [] }, + })).rejects.toThrow(/operand bytes/); + expect(workerInternals.nextId).toBe(nextId); + expect(worker.getWriteGen(oversizedGraph)).toBe(0); + } finally { + await worker.close(); + } + }); + + it('rejects final worker-bound encoding and operand budgets before posting', async () => { + const worker = new OxigraphWorkerStore(); + const workerInternals = worker as unknown as { nextId: number }; + try { + const beforeMutf8 = workerInternals.nextId; + await expect(worker.structuredMutation({ + kind: 'replace-subject-predicates', + input: { + graphUri: GRAPH, + subject: 'urn:test:subject', + predicates: [P], + replacementQuads: [quad('urn:test:subject', P, `"${'x'.repeat(70_000)}"`)], + }, + })).rejects.toMatchObject({ code: 'OVERSIZED_RDF_LITERAL' }); + expect(workerInternals.nextId).toBe(beforeMutf8); + + const subjects = Array.from( + { length: 65_000 }, + (_, index) => `urn:test:operand:${index}:${'x'.repeat(55)}`, + ); + const beforeBudget = workerInternals.nextId; + await expect(worker.structuredMutation({ + kind: 'delete-subjects', + input: { graphUri: GRAPH, subjects }, + })).rejects.toThrow(/operand bytes/); + expect(workerInternals.nextId).toBe(beforeBudget); + } finally { + await worker.close(); + } + }); + it('rejects sparse, duplicate, oversized, and cross-scope descriptors before dispatch', () => { const sparse = Array(2) as string[]; sparse[1] = 'urn:test:a'; diff --git a/packages/storage/test/changelog-store.test.ts b/packages/storage/test/changelog-store.test.ts index 0186e5f778..55f2808b00 100644 --- a/packages/storage/test/changelog-store.test.ts +++ b/packages/storage/test/changelog-store.test.ts @@ -24,6 +24,8 @@ import { ChangelogStore, CHANGELOG_GRAPH, asChangelogReader, type ChangelogEraGu import { createTripleStore } from '../src/triple-store.js'; import type { Quad, QueryOptions, QueryResult, StructuredMutation, TripleStore, UpdateOptions } from '../src/triple-store.js'; import { UnsupportedTripleStoreCapabilityError } from '../src/unsupported-capability-error.js'; +import { captureStructuredMutationSnapshot } from '../src/bounded-structured-mutation.js'; +import { materializeStructuredMutation } from '../src/structured-mutation-materialization-internal.js'; const G1 = 'http://ex.org/g1'; const G2 = 'http://ex.org/g2'; @@ -286,6 +288,118 @@ describe('ChangelogStore — opaque update handling', () => { await base.close(); }); + it('flags reconcile when a no-op fails because the inner store lost data', async () => { + const base = new OxigraphStore(); + const failing = new Proxy(base, { + get(target, property, receiver) { + if (property === 'structuredMutation') { + return async () => { throw new Error('in-memory worker data was lost'); }; + } + const value = Reflect.get(target, property, receiver) as unknown; + return typeof value === 'function' ? value.bind(target) : value; + }, + }) as TripleStore; + const log = new ChangelogStore(failing); + + await expect(log.structuredMutation({ + kind: 'delete-subjects', + input: { graphUri: G1, subjects: [] }, + })).rejects.toThrow('in-memory worker data was lost'); + + expect(log.needsReconcile).toBe(true); + await base.close(); + }); + + it('does not flag reconcile after a deferred-budget refusal before backend I/O', async () => { + const base = new OxigraphStore(); + let backendUpdates = 0; + const validatingLeaf = new Proxy(base, { + get(target, property, receiver) { + if (property === 'structuredMutation') { + return async (mutation: StructuredMutation, options?: QueryOptions) => { + const materialized = materializeStructuredMutation( + captureStructuredMutationSnapshot(mutation), + ); + if (materialized.outcome === 'noop') return; + backendUpdates += 1; + await target.structuredMutation(mutation, options); + }; + } + const value = Reflect.get(target, property, receiver) as unknown; + return typeof value === 'function' ? value.bind(target) : value; + }, + }) as TripleStore; + const log = new ChangelogStore(validatingLeaf); + + await expect(log.structuredMutation(overBudgetDeleteMutation(G1))) + .rejects.toThrow(/operand bytes/); + + expect(backendUpdates).toBe(0); + expect(log.needsReconcile).toBe(false); + expect(await log.readChanges(0, 100)).toEqual([]); + await base.close(); + }); + + it('keeps an enabled no-op inside the write tail while close drains it', async () => { + const base = new OxigraphStore(); + await base.insert([q('http://ex.org/delete-me', G1)]); + let release!: () => void; + let started!: () => void; + const gate = new Promise((resolve) => { release = resolve; }); + const firstStarted = new Promise((resolve) => { started = resolve; }); + const calls: Array<{ mutation: StructuredMutation; options?: QueryOptions }> = []; + const gated = new Proxy(base, { + get(target, property, receiver) { + if (property === 'structuredMutation') { + return async (mutation: StructuredMutation, options?: QueryOptions) => { + calls.push({ mutation, options }); + if (calls.length === 1) { + started(); + await gate; + } + await target.structuredMutation(mutation, options); + }; + } + const value = Reflect.get(target, property, receiver) as unknown; + return typeof value === 'function' ? value.bind(target) : value; + }, + }) as TripleStore; + const appended: unknown[] = []; + const log = new ChangelogStore(gated, { onAppend: (record) => appended.push(record) }); + const options = { source: 'changelog.noop' }; + + const mutation = log.structuredMutation({ + kind: 'delete-subjects', + input: { graphUri: G1, subjects: ['http://ex.org/delete-me'] }, + }); + await firstStarted; + const noop = log.structuredMutation({ + kind: 'delete-subjects', + input: { graphUri: G1, subjects: [] }, + }, options); + const close = log.close(); + + await expect(Promise.race([ + noop.then(() => 'settled'), + new Promise((resolve) => setTimeout(() => resolve('pending'), 25)), + ])).resolves.toBe('pending'); + await expect(log.structuredMutation({ + kind: 'delete-subjects', + input: { graphUri: G1, subjects: [] }, + })).rejects.toThrow(/store is closing/); + + release(); + await mutation; + await noop; + await close; + + expect(calls).toHaveLength(2); + expect(calls[1].options).toBe(options); + expect(Object.isFrozen(calls[1].mutation)).toBe(true); + expect(appended).toHaveLength(1); + expect(log.needsReconcile).toBe(false); + }); + it('an update() with touchedGraphs emits markers; without, it flags reconcile', async () => { const base = new OxigraphStore(); const log = new ChangelogStore(base); @@ -329,6 +443,19 @@ describe('ChangelogStore — opaque update handling', () => { }); }); +function overBudgetDeleteMutation(graphUri: string): StructuredMutation { + return { + kind: 'delete-subjects', + input: { + graphUri, + subjects: Array.from( + { length: 65_000 }, + (_, index) => `urn:test:operand:${index}:${'x'.repeat(55)}`, + ), + }, + }; +} + describe('ChangelogStore over Blazegraph — single-request atomicity', () => { let server: Server; let url: string; diff --git a/packages/storage/test/external-literal-store.test.ts b/packages/storage/test/external-literal-store.test.ts index 78dfa42ca9..e38cc18419 100644 --- a/packages/storage/test/external-literal-store.test.ts +++ b/packages/storage/test/external-literal-store.test.ts @@ -335,6 +335,28 @@ describe('SharedMemoryLiteralBlobStore', () => { expect(hydrated.type === 'bindings' ? hydrated.bindings : []).toEqual([{ o: largeLiteral }]); }); + it('forwards a structural no-op without externalization and preserves options identity', async () => { + const blobDir = await tempBlobDir(); + const inner = new OxigraphStore(); + const mutationSpy = vi.spyOn(inner, 'structuredMutation'); + const store = new SharedMemoryLiteralBlobStore(inner, { blobDir, thresholdBytes: 20 }); + const externalize = vi.spyOn( + store as unknown as { externalizeInsertQuad: (quad: Quad) => Promise }, + 'externalizeInsertQuad', + ); + const options = { source: 'literal-blob.noop' }; + + await store.structuredMutation({ + kind: 'delete-subjects', + input: { graphUri: SWM_GRAPH, subjects: [] }, + }, options); + + expect(externalize).not.toHaveBeenCalled(); + expect(mutationSpy).toHaveBeenCalledOnce(); + expect(mutationSpy.mock.calls[0][1]).toBe(options); + expect(Object.isFrozen(mutationSpy.mock.calls[0][0])).toBe(true); + }); + it('externalizes an over-budget replacement literal before validating the rewritten mutation', async () => { const blobDir = await tempBlobDir(); const inner = new OxigraphStore(); diff --git a/packages/storage/test/graph-set-index-store.test.ts b/packages/storage/test/graph-set-index-store.test.ts index 66f86bf924..0ac205898d 100644 --- a/packages/storage/test/graph-set-index-store.test.ts +++ b/packages/storage/test/graph-set-index-store.test.ts @@ -15,7 +15,9 @@ import { type StructuredMutation, type QueryOptions, type StoreWorkPriority, + captureStructuredMutationSnapshot, } from '../src/index.js'; +import { materializeStructuredMutation } from '../src/structured-mutation-materialization-internal.js'; import { ControlledProbeStore, CountingStore, @@ -156,6 +158,78 @@ describe('GraphSetIndexStore', () => { await inner.close(); }); + it('invalidates a warm index when a no-op fails because the inner store lost data', async () => { + const graph = 'did:dkg:context-graph:noop'; + const inner = new OxigraphStore(); + await inner.insert([q(graph)]); + let observed: StructuredMutation | undefined; + let observedOptions: QueryOptions | undefined; + const failing = new (class extends CountingStore { + override async structuredMutation( + mutation: StructuredMutation, + options?: QueryOptions, + ): Promise { + observed = mutation; + observedOptions = options; + this.failListGraphs = true; + throw new Error('in-memory worker data was lost'); + } + })(inner); + const events: GraphSetMutationEvent[] = []; + const store = new GraphSetIndexStore(failing, { + onMutation: (event) => events.push(event), + }); + const options = { source: 'graph-set.noop' }; + + await expect(store.listGraphs()).resolves.toEqual([graph]); + events.length = 0; + await expect(store.structuredMutation({ + kind: 'delete-subjects', + input: { graphUri: graph, subjects: [] }, + }, options)).rejects.toThrow('in-memory worker data was lost'); + + expect(observedOptions).toBe(options); + expect(observed).toEqual({ + kind: 'delete-subjects', + input: { graphUri: graph, subjects: [] }, + }); + expect(Object.isFrozen(observed)).toBe(true); + expect(events).toEqual([]); + await expect(store.listGraphs()).rejects.toThrow('listGraphs failed'); + expect(failing.listGraphsCalls).toBe(2); + await inner.close(); + }); + + it('does not rebuild after a deterministic deferred-budget refusal before backend I/O', async () => { + const graph = 'did:dkg:context-graph:budget-refusal'; + const inner = new OxigraphStore(); + await inner.insert([q(graph)]); + let backendUpdates = 0; + const validatingLeaf = new (class extends CountingStore { + override async structuredMutation( + mutation: StructuredMutation, + options?: QueryOptions, + ): Promise { + const materialized = materializeStructuredMutation( + captureStructuredMutationSnapshot(mutation), + ); + if (materialized.outcome === 'noop') return; + backendUpdates += 1; + await super.structuredMutation(mutation, options); + } + })(inner); + const store = new GraphSetIndexStore(validatingLeaf); + + await expect(store.listGraphs()).resolves.toEqual([graph]); + await expect(store.structuredMutation(overBudgetDeleteMutation(graph))) + .rejects.toThrow(/operand bytes/); + + expect(backendUpdates).toBe(0); + await expect(store.listGraphs()).resolves.toEqual([graph]); + expect(validatingLeaf.listGraphsCalls).toBe(1); + await inner.close(); + }); + it('rebuilds a warm graph index after an indeterminate structured mutation failure', async () => { const source = 'did:dkg:context-graph:indeterminate-source'; const target = 'did:dkg:context-graph:indeterminate-target'; @@ -1360,3 +1434,16 @@ describe('GraphSetIndexStore', () => { await optedInStore.close(); }); }); + +function overBudgetDeleteMutation(graphUri: string): StructuredMutation { + return { + kind: 'delete-subjects', + input: { + graphUri, + subjects: Array.from( + { length: 65_000 }, + (_, index) => `urn:test:operand:${index}:${'x'.repeat(55)}`, + ), + }, + }; +} diff --git a/packages/storage/test/managed-backend-ownership-dispatch-v1.test.ts b/packages/storage/test/managed-backend-ownership-dispatch-v1.test.ts index 4f688e854e..a7138ce266 100644 --- a/packages/storage/test/managed-backend-ownership-dispatch-v1.test.ts +++ b/packages/storage/test/managed-backend-ownership-dispatch-v1.test.ts @@ -117,6 +117,19 @@ describe('managed backend ownership at mutation dispatch', () => { await store.close().catch(() => undefined); }); + it('completes a valid structured no-op before managed admission and I/O', async () => { + const store = await managedStore(); + ownership.invalidate('port-release-unproven'); + + await expect(store.structuredMutation!({ + kind: 'delete-subjects', + input: { graphUri: 'urn:dkg:test:g', subjects: [] }, + })).resolves.toBeUndefined(); + expect(requests).toEqual([]); + + await store.close().catch(() => undefined); + }); + it('refuses a mutation with ZERO I/O after a failed clean-generation start', async () => { // The reviewer's second named regression. Here the lease is NOT terminal — // the supervisor invalidated with `stop` and will revive — so a check keyed diff --git a/packages/storage/test/oxigraph-worker-respawn.test.ts b/packages/storage/test/oxigraph-worker-respawn.test.ts index 6d60206177..ecd642bff8 100644 --- a/packages/storage/test/oxigraph-worker-respawn.test.ts +++ b/packages/storage/test/oxigraph-worker-respawn.test.ts @@ -51,6 +51,7 @@ function internals(store: OxigraphWorkerStore): { closePromise: Promise | null; respawnPromise: Promise | null; consecutiveRespawnFailures: number; + nextId: number; } { return store as unknown as { worker: Worker; @@ -58,6 +59,7 @@ function internals(store: OxigraphWorkerStore): { closePromise: Promise | null; respawnPromise: Promise | null; consecutiveRespawnFailures: number; + nextId: number; }; } @@ -168,6 +170,32 @@ describe('OxigraphWorkerStore respawn after unexpected worker exit', () => { } }, 15_000); + it('parks a no-op through respawn without posting or advancing generations', async () => { + const store = makeStore(path); + try { + await killWorker(store); + await killWorker(store); + expect(internals(store).lifecycle).toBe('respawning'); + expect(internals(store).respawnPromise).not.toBeNull(); + + const nextId = internals(store).nextId; + const generation = store.getWriteGen('urn:test:g'); + const noop = store.structuredMutation({ + kind: 'delete-subjects', + input: { graphUri: 'urn:test:g', subjects: [] }, + }); + await new Promise((resolve) => setTimeout(resolve, 25)); + expect(internals(store).nextId).toBe(nextId); + + await expect(noop).resolves.toBeUndefined(); + expect(internals(store).lifecycle).toBe('live'); + expect(internals(store).nextId).toBe(nextId); + expect(store.getWriteGen('urn:test:g')).toBe(generation); + } finally { + await store.close().catch(() => {}); + } + }, 15_000); + it('a successful op resets the crash-loop counter', async () => { const store = makeStore(path); try { @@ -200,6 +228,27 @@ describe('OxigraphWorkerStore respawn after unexpected worker exit', () => { // Post-close ops fail exactly as before this fix — fast and permanent. await expect(store.countQuads('urn:test:g')).rejects.toThrow(/store is closed/); await expect(store.insert(quads(1))).rejects.toThrow(/store is closed/); + const nextId = internals(store).nextId; + await expect(store.structuredMutation({ + kind: 'delete-subjects', + input: { graphUri: 'urn:test:g', subjects: [] }, + })).rejects.toThrow(/store is closed/); + expect(internals(store).nextId).toBe(nextId); + }); + + it('rejects a no-op while close is draining without posting another message', async () => { + const store = makeStore(path); + const close = store.close(); + expect(internals(store).lifecycle).toBe('closing'); + const nextId = internals(store).nextId; + + await expect(store.structuredMutation({ + kind: 'delete-subjects', + input: { graphUri: 'urn:test:g', subjects: [] }, + })).rejects.toThrow(/store is closed/); + expect(internals(store).nextId).toBe(nextId); + + await close; }); it('gives up after MAX consecutive dead-on-arrival respawns and latches closed with guidance', async () => { @@ -224,6 +273,12 @@ describe('OxigraphWorkerStore respawn after unexpected worker exit', () => { // The give-up verdict is now the single, explicit terminal state — no // cluster of booleans that could disagree with the fail-fast error above. expect(internals(store).lifecycle).toBe('gave_up'); + const nextId = internals(store).nextId; + await expect(store.structuredMutation({ + kind: 'delete-subjects', + input: { graphUri: 'urn:test:g', subjects: [] }, + })).rejects.toThrow(/respawn gave up/); + expect(internals(store).nextId).toBe(nextId); } finally { // close() on a latched store must still resolve (idempotent teardown). await expect(store.close()).resolves.toBeUndefined(); @@ -245,6 +300,12 @@ describe('OxigraphWorkerStore respawn after unexpected worker exit', () => { // Wait out the backoff so a buggy respawn would have fired by now. await new Promise((r) => setTimeout(r, 1_500)); await expect(store.countQuads('urn:test:g')).rejects.toThrow(/store is closed/); + const nextId = internals(store).nextId; + await expect(store.structuredMutation({ + kind: 'delete-subjects', + input: { graphUri: 'urn:test:g', subjects: [] }, + })).rejects.toThrow(/store is closed/); + expect(internals(store).nextId).toBe(nextId); }, 15_000); }); @@ -289,6 +350,12 @@ describe('OxigraphWorkerStore in-memory fail-closed on unexpected worker exit', // Writes fail closed the same way — the store is unusable, not read-only. await expect(store.insert(quads(1))).rejects.toThrow(/IN-MEMORY store's worker crashed/); + const nextId = internals(store).nextId; + await expect(store.structuredMutation({ + kind: 'delete-subjects', + input: { graphUri: 'urn:test:g', subjects: [] }, + })).rejects.toThrow(/IN-MEMORY store's worker crashed/); + expect(internals(store).nextId).toBe(nextId); } finally { // close() on a data-lost store must still resolve (idempotent teardown) // and must not resurrect it. diff --git a/packages/storage/test/sparql-http.test.ts b/packages/storage/test/sparql-http.test.ts index b8e8fc4307..a4eea55334 100644 --- a/packages/storage/test/sparql-http.test.ts +++ b/packages/storage/test/sparql-http.test.ts @@ -875,6 +875,17 @@ describe('SparqlHttpStore (test server)', () => { expect(insertedQuads[0]).not.toContain(';'); }); + it('structuredMutation sends no request for a valid empty delete', async () => { + insertedQuads.length = 0; + + await store.structuredMutation!({ + kind: 'delete-subjects', + input: { graphUri: 'http://ex.org/g1', subjects: [] }, + }); + + expect(insertedQuads).toHaveLength(0); + }); + it('deleteByPattern sends DELETE WHERE to update endpoint', async () => { insertedQuads.length = 0; await store.deleteByPattern({ subject: 'http://ex.org/s', graph: 'http://ex.org/g' }); diff --git a/packages/storage/test/structured-mutation-composition.test.ts b/packages/storage/test/structured-mutation-composition.test.ts new file mode 100644 index 0000000000..b3e27da380 --- /dev/null +++ b/packages/storage/test/structured-mutation-composition.test.ts @@ -0,0 +1,372 @@ +import { createHash } from 'node:crypto'; +import { mkdtemp, rm } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; + +import { afterEach, describe, expect, it, vi } from 'vitest'; + +import { + CHANGELOG_GRAPH, + ChangelogStore, + EXTERNAL_LITERAL_REF_DATATYPE, + GraphSetIndexStore, + OxigraphStore, + OxigraphWorkerStore, + SharedMemoryLiteralBlobStore, + captureStructuredMutationSnapshot, + type GraphSetMutationEvent, + type QueryOptions, + type StructuredMutation, +} from '../src/index.js'; +import { + BOUNDED_MUTATION_MAX_OPERAND_BYTES, + buildStructuredMutationUpdate, +} from '../src/bounded-structured-mutation.js'; +import { BoundedMutationBudgetError } from '../src/structured-mutation/primitives.js'; + +const GRAPH = 'urn:test:composition'; +const TARGET = 'urn:test:composition:target'; +const PREDICATE = 'urn:test:composition:predicate'; +const SWM_GRAPH = 'did:dkg:context-graph:composition/_shared_memory'; + +function mutationFixtures(): StructuredMutation[] { + return [ + { kind: 'delete-subjects', input: { + graphUri: GRAPH, + subjects: ['urn:test:composition:delete'], + } }, + { kind: 'prune-ranked-subjects', input: { + graphUri: GRAPH, + subjectPrefix: 'urn:test:composition:ranked:', + eligibilityPredicate: PREDICATE, + eligibleObjects: ['approved'], + primaryRankPredicate: 'urn:test:composition:rank:primary', + secondaryRankPredicate: 'urn:test:composition:rank:secondary', + retainNewest: 1, + maxDelete: 1, + } }, + { kind: 'prune-linked-record-closures', input: { + graphUri: GRAPH, + matchObjectIris: ['urn:test:composition:agent'], + linkPredicates: [PREDICATE], + recordParentPredicate: 'urn:test:composition:parent', + descendantSeparator: '/', + } }, + { kind: 'replace-subject-predicates', input: { + graphUri: GRAPH, + subject: 'urn:test:composition:subject', + predicates: [PREDICATE], + replacementQuads: [{ + subject: 'urn:test:composition:subject', + predicate: PREDICATE, + object: '"value"', + graph: GRAPH, + }], + } }, + { kind: 'replace-projection-from-graph', input: { + targetGraphUri: TARGET, + stagingGraphUri: 'urn:test:composition:staging', + targetSubject: 'urn:test:composition:subject', + preservedTargetPredicates: [PREDICATE], + targetSubjectPrefixes: ['urn:test:composition:prefix:'], + } }, + { kind: 'copy-subject-projection', input: { + sourceGraphUris: [GRAPH], + targetGraphUri: TARGET, + roots: ['urn:test:composition:root'], + descendantSuffix: '/', + excludedPredicates: [PREDICATE], + } }, + ]; +} + +function redirectCallerMutation(mutation: StructuredMutation): void { + const input = mutation.input as unknown as Record; + if ('graphUri' in input) input.graphUri = 'urn:test:redirected'; + if ('targetGraphUri' in input) input.targetGraphUri = 'urn:test:redirected'; + for (const key of [ + 'subjects', + 'eligibleObjects', + 'matchObjectIris', + 'predicates', + 'preservedTargetPredicates', + 'roots', + ]) { + const values = input[key]; + if (Array.isArray(values) && values.length > 0) values[0] = 'urn:test:redirected'; + } + if (mutation.kind === 'replace-subject-predicates') { + (mutation.input.replacementQuads[0] as { object: string }).object = '"redirected"'; + } +} + +function sha256(value: string): string { + return createHash('sha256').update(value).digest('hex'); +} + +describe('structured mutation composition', () => { + const tempDirs: string[] = []; + + afterEach(async () => { + vi.restoreAllMocks(); + await Promise.all(tempDirs.splice(0).map((dir) => rm(dir, { recursive: true, force: true }))); + }); + + it('reuses one snapshot through unchanged wrappers for every mutation kind', async () => { + const blobDir = await mkdtemp(join(tmpdir(), 'structured-mutation-composition-')); + tempDirs.push(blobDir); + const leaf = new OxigraphStore(); + const literal = new SharedMemoryLiteralBlobStore(leaf, { + blobDir, + thresholdBytes: 1_024, + }); + const graphSet = new GraphSetIndexStore(literal); + const changelog = new ChangelogStore(graphSet); + const graphSetMutation = vi.spyOn(graphSet, 'structuredMutation'); + const literalMutation = vi.spyOn(literal, 'structuredMutation'); + const leafMutation = vi.spyOn(leaf, 'structuredMutation'); + const options: QueryOptions = { source: 'composition.identity' }; + + try { + for (const mutation of mutationFixtures()) { + const expected = JSON.parse(JSON.stringify(mutation)) as StructuredMutation; + const graphSetCalls = graphSetMutation.mock.calls.length; + const literalCalls = literalMutation.mock.calls.length; + const leafCalls = leafMutation.mock.calls.length; + + const pending = changelog.structuredMutation(mutation, options); + redirectCallerMutation(mutation); + await pending; + + const graphSetCall = graphSetMutation.mock.calls[graphSetCalls]; + const literalCall = literalMutation.mock.calls[literalCalls]; + const leafCall = leafMutation.mock.calls[leafCalls]; + expect(graphSetCall[0]).toEqual(expected); + expect(literalCall[0]).toEqual(expected); + expect(leafCall[0]).toEqual(expected); + expect(graphSetCall[0]).toBe(literalCall[0]); + expect(captureStructuredMutationSnapshot(graphSetCall[0])) + .toBe(captureStructuredMutationSnapshot(literalCall[0])); + if (expected.kind === 'replace-subject-predicates') { + expect(leafCall[0]).not.toBe(literalCall[0]); + expect(captureStructuredMutationSnapshot(leafCall[0])) + .not.toBe(captureStructuredMutationSnapshot(literalCall[0])); + } else { + expect(leafCall[0]).toBe(literalCall[0]); + expect(captureStructuredMutationSnapshot(leafCall[0])) + .toBe(captureStructuredMutationSnapshot(literalCall[0])); + } + expect(Object.isFrozen(graphSetCall[0])).toBe(true); + expect(Object.isFrozen(leafCall[0])).toBe(true); + expect(graphSetCall[1]).toBe(options); + expect(literalCall[1]).toBe(options); + expect(leafCall[1]).toBe(options); + } + } finally { + await changelog.close(); + } + }); + + it('runs no-op policy preflight but performs no I/O, maintenance, or generation advance', async () => { + const blobDir = await mkdtemp(join(tmpdir(), 'structured-mutation-noop-')); + tempDirs.push(blobDir); + const leaf = new OxigraphStore(); + const embedded = (leaf as unknown as { store: { update: (sparql: string) => void } }).store; + const update = vi.spyOn(embedded, 'update'); + const leafMutation = vi.spyOn(leaf, 'structuredMutation'); + const query = vi.spyOn(leaf, 'query'); + const hasGraph = vi.spyOn(leaf, 'hasGraph'); + const graphEvents: GraphSetMutationEvent[] = []; + const changelogEvents: unknown[] = []; + const literal = new SharedMemoryLiteralBlobStore(leaf, { blobDir, thresholdBytes: 20 }); + const graphSet = new GraphSetIndexStore(literal, { + onMutation: (event) => graphEvents.push(event), + }); + const changelog = new ChangelogStore(graphSet, { + onAppend: (event) => changelogEvents.push(event), + }); + const options = { source: 'composition.noop' }; + const generation = leaf.getWriteGen(GRAPH); + + try { + await changelog.structuredMutation({ + kind: 'delete-subjects', + input: { graphUri: GRAPH, subjects: [] }, + }, options); + + expect(leafMutation).toHaveBeenCalledOnce(); + expect(leafMutation.mock.calls[0][1]).toBe(options); + expect(update).not.toHaveBeenCalled(); + expect(query).not.toHaveBeenCalled(); + expect(hasGraph).not.toHaveBeenCalled(); + expect(graphEvents).toEqual([]); + expect(changelogEvents).toEqual([]); + expect(leaf.getWriteGen(GRAPH)).toBe(generation); + + leafMutation.mockClear(); + await expect(changelog.structuredMutation({ + kind: 'delete-subjects', + input: { graphUri: CHANGELOG_GRAPH, subjects: [] }, + })).rejects.toThrow(/reserved changelog plane/); + expect(leafMutation).not.toHaveBeenCalled(); + + const neutral = new GraphSetIndexStore(literal); + await expect(neutral.structuredMutation({ + kind: 'delete-subjects', + input: { graphUri: CHANGELOG_GRAPH, subjects: [] }, + })).resolves.toBeUndefined(); + expect(leafMutation).toHaveBeenCalledOnce(); + expect(update).not.toHaveBeenCalled(); + } finally { + await changelog.close(); + } + }); + + it('externalizes an over-budget literal before final validation through the full stack', async () => { + const blobDir = await mkdtemp(join(tmpdir(), 'structured-mutation-large-composition-')); + tempDirs.push(blobDir); + const leaf = new OxigraphStore(); + const leafMutation = vi.spyOn(leaf, 'structuredMutation'); + const literal = new SharedMemoryLiteralBlobStore(leaf, { blobDir, thresholdBytes: 20 }); + const graphSet = new GraphSetIndexStore(literal); + const changelog = new ChangelogStore(graphSet); + const subject = 'urn:test:composition:large-subject'; + const largeLiteral = `"${'x'.repeat(BOUNDED_MUTATION_MAX_OPERAND_BYTES + 1)}"`; + + try { + await changelog.structuredMutation({ + kind: 'replace-subject-predicates', + input: { + graphUri: SWM_GRAPH, + subject, + predicates: [PREDICATE], + replacementQuads: [{ + subject, + predicate: PREDICATE, + object: largeLiteral, + graph: SWM_GRAPH, + }], + }, + }); + + expect(leafMutation).toHaveBeenCalledOnce(); + const forwarded = leafMutation.mock.calls[0][0]; + expect(forwarded.kind).toBe('replace-subject-predicates'); + if (forwarded.kind !== 'replace-subject-predicates') return; + expect(forwarded.input.replacementQuads[0].object).toBe( + `"sha256:${sha256(largeLiteral)}"^^<${EXTERNAL_LITERAL_REF_DATATYPE}>`, + ); + expect(await leaf.hasGraph(SWM_GRAPH)).toBe(true); + expect(changelog.needsReconcile).toBe(false); + } finally { + await changelog.close(); + } + }); + + it('does not trust forged or bare low-level refusals after the inner store commits', async () => { + const failures: ReadonlyArray Error]> = [ + ['forged-code', () => Object.assign( + new Error('committed then forged a clean refusal'), + { code: 'STRUCTURED_MUTATION_PRE_DISPATCH_REFUSAL' }, + )], + ['bare-budget', () => new BoundedMutationBudgetError( + 'committed then threw a bare budget error', + )], + ]; + + for (const [suffix, createFailure] of failures) { + const source = `urn:test:composition:${suffix}:source`; + const target = `urn:test:composition:${suffix}:target`; + const root = `urn:test:composition:${suffix}:root`; + const failure = createFailure(); + const leaf = new (class extends OxigraphStore { + override async structuredMutation( + mutation: StructuredMutation, + options?: QueryOptions, + ): Promise { + await super.structuredMutation(mutation, options); + throw failure; + } + })(); + const listGraphs = vi.spyOn(leaf, 'listGraphs'); + const graphSet = new GraphSetIndexStore(leaf); + const changelog = new ChangelogStore(graphSet); + + try { + await leaf.insert([{ + subject: root, + predicate: PREDICATE, + object: '"source"', + graph: source, + }]); + await expect(changelog.listGraphs()).resolves.toEqual([source]); + expect(listGraphs).toHaveBeenCalledOnce(); + + const received = await changelog.structuredMutation({ + kind: 'copy-subject-projection', + input: { + sourceGraphUris: [source], + targetGraphUri: target, + roots: [root], + descendantSuffix: '/', + excludedPredicates: [], + }, + }).then(() => undefined, (error: unknown) => error); + expect(received).toBe(failure); + + expect(changelog.needsReconcile).toBe(true); + expect(await leaf.hasGraph(target)).toBe(true); + await expect(changelog.listGraphs()) + .resolves.toEqual(expect.arrayContaining([source, target])); + expect(listGraphs).toHaveBeenCalledTimes(2); + } finally { + await changelog.close(); + } + } + }); + + it('preserves worker-side serialized-budget refusals as mutation-free', async () => { + const leaf = new OxigraphWorkerStore(); + const graphSet = new GraphSetIndexStore(leaf); + const changelog = new ChangelogStore(graphSet); + const listGraphs = vi.spyOn(leaf, 'listGraphs'); + const mutation: StructuredMutation = { + kind: 'delete-subjects', + input: { + graphUri: GRAPH, + subjects: Array.from( + { length: 75_000 }, + (_, index) => `urn:test:${index.toString().padStart(5, '0')}:${'x'.repeat(40)}`, + ), + }, + }; + + try { + await leaf.insert([{ + subject: 'urn:test:composition:retained', + predicate: PREDICATE, + object: '"retained"', + graph: GRAPH, + }]); + await expect(changelog.listGraphs()).resolves.toEqual([GRAPH]); + expect(listGraphs).toHaveBeenCalledOnce(); + const generation = leaf.getWriteGen(GRAPH); + expect(() => buildStructuredMutationUpdate(mutation)) + .toThrow(/serialized update exceeds/); + + await expect(changelog.structuredMutation(mutation)) + .rejects.toMatchObject({ + code: 'STRUCTURED_MUTATION_PRE_DISPATCH_REFUSAL', + message: expect.stringMatching(/serialized update exceeds/), + }); + + expect(changelog.needsReconcile).toBe(false); + expect(leaf.getWriteGen(GRAPH)).toBe(generation); + expect(await leaf.countQuads(GRAPH)).toBe(1); + await expect(changelog.listGraphs()).resolves.toEqual([GRAPH]); + expect(listGraphs).toHaveBeenCalledOnce(); + } finally { + await changelog.close(); + } + }); +}); diff --git a/packages/storage/test/structured-mutation-preparation.test.ts b/packages/storage/test/structured-mutation-preparation.test.ts new file mode 100644 index 0000000000..c59d734f84 --- /dev/null +++ b/packages/storage/test/structured-mutation-preparation.test.ts @@ -0,0 +1,377 @@ +import { describe, expect, it, vi } from 'vitest'; + +import { + BOUNDED_MUTATION_MAX_SOURCE_GRAPHS, + buildStructuredMutationUpdate, + captureStructuredMutationEffects, + captureStructuredMutationSnapshot, + type StructuredMutationSnapshot, +} from '../src/bounded-structured-mutation.js'; +import { GraphSetIndexStore } from '../src/graph-set-index-store.js'; +import { CHANGELOG_GRAPH, ChangelogStore } from '../src/changelog-store.js'; +import { materializeStructuredMutation } from '../src/structured-mutation-materialization-internal.js'; +import type { StructuredMutation, TripleStore } from '../src/triple-store.js'; + +const GRAPH = 'urn:test:preparation'; +const TARGET = 'urn:test:preparation:target'; +const PREDICATE = 'urn:test:preparation:predicate'; + +function deleteMutation(subjects: readonly string[] = ['urn:test:subject']): StructuredMutation { + return { kind: 'delete-subjects', input: { graphUri: GRAPH, subjects } }; +} + +describe('structured mutation preparation', () => { + it('captures one deeply frozen caller-independent snapshot and reuses its identity', () => { + const subjects = ['urn:test:subject']; + const mutation = deleteMutation(subjects); + const snapshot = captureStructuredMutationSnapshot(mutation); + + subjects[0] = 'urn:test:redirected'; + expect(snapshot.mutation).toEqual({ + kind: 'delete-subjects', + input: { graphUri: GRAPH, subjects: ['urn:test:subject'] }, + }); + expect(Object.isFrozen(snapshot)).toBe(true); + expect(Object.isFrozen(snapshot.mutation)).toBe(true); + expect(Object.isFrozen(snapshot.mutation.input)).toBe(true); + expect(snapshot.mutation.kind === 'delete-subjects' + && Object.isFrozen(snapshot.mutation.input.subjects)).toBe(true); + expect(captureStructuredMutationSnapshot(snapshot.mutation)).toBe(snapshot); + expect(captureStructuredMutationSnapshot({ ...snapshot.mutation })).not.toBe(snapshot); + expect(() => { + (snapshot.mutation.input as { graphUri: string }).graphUri = TARGET; + }).toThrow(TypeError); + }); + + it('keeps final backend materialization off the package-root surface', async () => { + const storage = await import('../src/index.js'); + expect('materializeStructuredMutation' in storage).toBe(false); + }); + + it('reads accessor-backed descriptors and proxy-array entries exactly once', () => { + const reads = new Map(); + const read = (key: string, value: T): T => { + reads.set(key, (reads.get(key) ?? 0) + 1); + return value; + }; + const subjects = new Proxy(['urn:test:subject'], { + get(target, property, receiver) { + if (property === 'length' || property === '0') read(`subjects.${String(property)}`, null); + return Reflect.get(target, property, receiver); + }, + }); + const input = Object.defineProperties({}, { + graphUri: { enumerable: true, get: () => read('graphUri', GRAPH) }, + subjects: { enumerable: true, get: () => read('subjects', subjects) }, + }); + const mutation = Object.defineProperties({}, { + kind: { enumerable: true, get: () => read('kind', 'delete-subjects') }, + input: { enumerable: true, get: () => read('input', input) }, + }) as StructuredMutation; + + const snapshot = captureStructuredMutationSnapshot(mutation); + + expect(snapshot.mutation).toEqual(deleteMutation()); + expect(Object.fromEntries(reads)).toEqual({ + kind: 1, + input: 1, + graphUri: 1, + subjects: 1, + 'subjects.length': 1, + 'subjects.0': 1, + }); + }); + + it('copies exact quad fields and cannot be redirected after capture', () => { + const quad = { + subject: 'urn:test:subject', + predicate: PREDICATE, + object: '"before"', + graph: GRAPH, + ignored: 'caller-only', + }; + const snapshot = captureStructuredMutationSnapshot({ + kind: 'replace-subject-predicates', + input: { + graphUri: GRAPH, + subject: quad.subject, + predicates: [PREDICATE], + replacementQuads: [quad], + }, + }); + quad.object = '"after"'; + quad.graph = TARGET; + + expect(snapshot.mutation.kind).toBe('replace-subject-predicates'); + if (snapshot.mutation.kind !== 'replace-subject-predicates') return; + expect(snapshot.mutation.input.replacementQuads).toEqual([{ + subject: 'urn:test:subject', + predicate: PREDICATE, + object: '"before"', + graph: GRAPH, + }]); + expect(Object.isFrozen(snapshot.mutation.input.replacementQuads[0])).toBe(true); + expect('ignored' in snapshot.mutation.input.replacementQuads[0]).toBe(false); + }); + + it('materializes every trusted snapshot without replacing its operands', () => { + const mutations: StructuredMutation[] = [ + deleteMutation(), + { kind: 'prune-ranked-subjects', input: { + graphUri: GRAPH, + subjectPrefix: 'urn:test:ranked:', + eligibilityPredicate: PREDICATE, + eligibleObjects: ['approved'], + primaryRankPredicate: 'urn:test:rank:primary', + secondaryRankPredicate: 'urn:test:rank:secondary', + retainNewest: 1, + maxDelete: 1, + } }, + { kind: 'prune-linked-record-closures', input: { + graphUri: GRAPH, + matchObjectIris: ['urn:test:agent'], + linkPredicates: [PREDICATE], + recordParentPredicate: 'urn:test:parent', + descendantSeparator: '/', + } }, + { kind: 'replace-subject-predicates', input: { + graphUri: GRAPH, + subject: 'urn:test:subject', + predicates: [PREDICATE], + replacementQuads: [{ + subject: 'urn:test:subject', predicate: PREDICATE, object: '"value"', graph: GRAPH, + }], + } }, + { kind: 'replace-projection-from-graph', input: { + targetGraphUri: TARGET, + stagingGraphUri: 'urn:test:preparation:staging', + targetSubject: 'urn:test:subject', + preservedTargetPredicates: [PREDICATE], + targetSubjectPrefixes: [], + } }, + { kind: 'copy-subject-projection', input: { + sourceGraphUris: [GRAPH], + targetGraphUri: TARGET, + roots: ['urn:test:subject'], + descendantSuffix: '/', + excludedPredicates: [], + } }, + ]; + + for (const mutation of mutations) { + const snapshot = captureStructuredMutationSnapshot(mutation); + const capturedInput = snapshot.mutation.input; + const materialized = materializeStructuredMutation(snapshot); + expect(materialized.outcome).toBe('execute'); + if (materialized.outcome !== 'execute') continue; + expect(materialized.snapshot).toBe(snapshot); + expect(materialized.snapshot.mutation.input).toBe(capturedInput); + expect(materialized.update).toBe(buildStructuredMutationUpdate(mutation)); + } + }); + + it('classifies no-op only after trusted deferred validation', () => { + const snapshot = captureStructuredMutationSnapshot(deleteMutation([])); + expect(snapshot.outcome).toBe('noop'); + expect(snapshot.effects).toBeUndefined(); + expect(materializeStructuredMutation(snapshot)).toEqual({ outcome: 'noop', snapshot }); + }); + + it('rejects copied or forged snapshots at the internal materialization boundary', () => { + const snapshot = captureStructuredMutationSnapshot(deleteMutation()); + const copied = { ...snapshot } as StructuredMutationSnapshot; + const forged = { + ...snapshot, + mutation: { ...snapshot.mutation }, + } as StructuredMutationSnapshot; + + expect(() => materializeStructuredMutation(copied)).toThrow(/not trusted/); + expect(() => materializeStructuredMutation(forged)).toThrow(/not trusted/); + }); + + it('defers the aggregate replacement budget until after snapshot capture', () => { + const object = JSON.stringify('x'.repeat(4 * 1024 * 1024)); + const snapshot = captureStructuredMutationSnapshot({ + kind: 'replace-subject-predicates', + input: { + graphUri: GRAPH, + subject: 'urn:test:subject', + predicates: [PREDICATE], + replacementQuads: [{ + subject: 'urn:test:subject', predicate: PREDICATE, object, graph: GRAPH, + }], + }, + }); + + expect(snapshot.outcome).toBe('candidate'); + expect(() => materializeStructuredMutation(snapshot)).toThrow(/operand bytes/); + }); + + it('preserves the non-validating synchronous effects compatibility helper', () => { + const malformed = { + kind: 'delete-subjects', + input: { graphUri: 'relative', subjects: [] }, + } as StructuredMutation; + expect(captureStructuredMutationEffects(malformed)).toBeUndefined(); + expect(() => captureStructuredMutationSnapshot(malformed)).toThrow(/absolute IRI/); + }); + + it.each(invalidSnapshotDescriptors())( + 'rejects the $name descriptor through snapshot capture', + ({ mutation, expected }) => { + expect(() => captureStructuredMutationSnapshot(mutation)).toThrow(expected); + }, + ); + + it('rejects an invalid descriptor in a decorator before inner I/O', async () => { + const operation = vi.fn(async () => {}); + const store = new GraphSetIndexStore({ + structuredMutation: operation, + } as unknown as TripleStore); + + await expect(store.structuredMutation({ + kind: 'copy-subject-projection', + input: { + sourceGraphUris: [GRAPH], + targetGraphUri: GRAPH, + roots: ['urn:test:subject'], + descendantSuffix: '/', + excludedPredicates: [], + }, + })).rejects.toThrow(/must not be a source/); + expect(operation).not.toHaveBeenCalled(); + }); + + it('separates multi-graph guarded scopes from target-only effects', () => { + const copy = captureStructuredMutationSnapshot({ + kind: 'copy-subject-projection', + input: { + sourceGraphUris: [GRAPH, 'urn:test:preparation:source-2'], + targetGraphUri: TARGET, + roots: ['urn:test:subject'], + descendantSuffix: '/', + excludedPredicates: [], + }, + }); + expect(copy.guardedGraphs).toEqual([GRAPH, 'urn:test:preparation:source-2', TARGET]); + expect(copy.outcome).toBe('candidate'); + if (copy.outcome === 'candidate') { + expect(copy.effects.touchedGraphs).toEqual([TARGET]); + } + + const replacement = captureStructuredMutationSnapshot({ + kind: 'replace-projection-from-graph', + input: { + targetGraphUri: TARGET, + stagingGraphUri: GRAPH, + targetSubject: 'urn:test:subject', + preservedTargetPredicates: [], + targetSubjectPrefixes: [], + }, + }); + expect(replacement.guardedGraphs).toEqual([TARGET, GRAPH]); + expect(replacement.outcome).toBe('candidate'); + if (replacement.outcome === 'candidate') { + expect(replacement.effects.touchedGraphs).toEqual([TARGET]); + } + }); + + it('rejects reserved multi-graph sources and staging graphs before inner I/O', async () => { + const operation = vi.fn(async () => {}); + const store = new ChangelogStore({ + structuredMutation: operation, + } as unknown as TripleStore); + + await expect(store.structuredMutation({ + kind: 'copy-subject-projection', + input: { + sourceGraphUris: [CHANGELOG_GRAPH], + targetGraphUri: TARGET, + roots: ['urn:test:subject'], + descendantSuffix: '/', + excludedPredicates: [], + }, + })).rejects.toThrow(/reserved changelog plane/); + await expect(store.structuredMutation({ + kind: 'replace-projection-from-graph', + input: { + targetGraphUri: TARGET, + stagingGraphUri: CHANGELOG_GRAPH, + targetSubject: 'urn:test:subject', + preservedTargetPredicates: [], + targetSubjectPrefixes: [], + }, + })).rejects.toThrow(/reserved changelog plane/); + expect(operation).not.toHaveBeenCalled(); + }); +}); + +function invalidSnapshotDescriptors(): Array<{ + name: string; + mutation: StructuredMutation; + expected: RegExp; +}> { + const sparseSubjects = Array(2) as string[]; + sparseSubjects[1] = 'urn:test:subject'; + return [ + { + name: 'sparse', + mutation: deleteMutation(sparseSubjects), + expected: /dense array/, + }, + { + name: 'duplicate', + mutation: deleteMutation(['urn:test:subject', 'urn:test:subject']), + expected: /duplicate/, + }, + { + name: 'oversized', + mutation: { + kind: 'copy-subject-projection', + input: { + sourceGraphUris: Array.from( + { length: BOUNDED_MUTATION_MAX_SOURCE_GRAPHS + 1 }, + (_, index) => `urn:test:source:${index}`, + ), + targetGraphUri: TARGET, + roots: ['urn:test:subject'], + descendantSuffix: '/', + excludedPredicates: [], + }, + }, + expected: /must contain/, + }, + { + name: 'cross-scope', + mutation: { + kind: 'replace-subject-predicates', + input: { + graphUri: GRAPH, + subject: 'urn:test:subject', + predicates: [PREDICATE], + replacementQuads: [{ + subject: 'urn:test:subject', + predicate: PREDICATE, + object: '"value"', + graph: TARGET, + }], + }, + }, + expected: /must target subject/, + }, + { + name: 'same source and target', + mutation: { + kind: 'copy-subject-projection', + input: { + sourceGraphUris: [GRAPH], + targetGraphUri: GRAPH, + roots: ['urn:test:subject'], + descendantSuffix: '/', + excludedPredicates: [], + }, + }, + expected: /must not be a source/, + }, + ]; +} diff --git a/packages/storage/test/structured-mutation-snapshot.typetest.ts b/packages/storage/test/structured-mutation-snapshot.typetest.ts new file mode 100644 index 0000000000..cccf0199a2 --- /dev/null +++ b/packages/storage/test/structured-mutation-snapshot.typetest.ts @@ -0,0 +1,43 @@ +import { captureStructuredMutationSnapshot } from '../src/index.js'; + +const snapshot = captureStructuredMutationSnapshot({ + kind: 'replace-subject-predicates', + input: { + graphUri: 'urn:test:graph', + subject: 'urn:test:subject', + predicates: ['urn:test:predicate'], + replacementQuads: [{ + subject: 'urn:test:subject', + predicate: 'urn:test:predicate', + object: '"value"', + graph: 'urn:test:graph', + }], + }, +}); + +if (snapshot.outcome === 'noop') { + const effects: undefined = snapshot.effects; + void effects; +} else { + const touchedGraphs: readonly string[] = snapshot.effects.touchedGraphs; + void touchedGraphs; +} + +if (snapshot.mutation.kind === 'replace-subject-predicates') { + // @ts-expect-error snapshot input fields are immutable + snapshot.mutation.input.graphUri = 'urn:test:redirected'; + // @ts-expect-error snapshot arrays are immutable + snapshot.mutation.input.predicates[0] = 'urn:test:redirected'; + // @ts-expect-error snapshot quad fields are immutable + snapshot.mutation.input.replacementQuads[0].object = '"redirected"'; + // @ts-expect-error snapshot arrays cannot be extended + snapshot.mutation.input.replacementQuads.push({ + subject: 'urn:test:subject', + predicate: 'urn:test:predicate', + object: '"extra"', + graph: 'urn:test:graph', + }); +} + +// @ts-expect-error snapshot mutation tags are immutable +snapshot.mutation.kind = 'delete-subjects'; diff --git a/packages/storage/tsconfig.typetests.json b/packages/storage/tsconfig.typetests.json index 5f212a75cb..81bb027a30 100644 --- a/packages/storage/tsconfig.typetests.json +++ b/packages/storage/tsconfig.typetests.json @@ -7,7 +7,8 @@ "include": [ "src", "test/store-control-barrier-contract-v1.test.ts", - "test/store-control-barrier-contract-v1.typetest.ts" + "test/store-control-barrier-contract-v1.typetest.ts", + "test/structured-mutation-snapshot.typetest.ts" ], "references": [{ "path": "../core" }, { "path": "../rdf-utils" }] }