From b842ba860cf05b4708803741634ae25516544f18 Mon Sep 17 00:00:00 2001 From: Branimir Rakic Date: Thu, 9 Jul 2026 02:13:51 +0200 Subject: [PATCH] refactor(storage): retire oxigraph-worker backend --- CHANGELOG.md | 4 + bench/store-read-latency.bench.ts | 188 +---------- docs/UPGRADE_TO_RC17.md | 4 + docs/use-dkg/storage-sparql-http.md | 4 +- packages/agent/src/dkg-agent-helpers.ts | 1 - packages/agent/src/dkg-agent-types.ts | 2 +- packages/agent/src/dkg-agent.ts | 4 +- .../agent/src/sync/responder/sync-handler.ts | 2 +- packages/cli/src/commands/hermes.ts | 2 +- packages/cli/src/commands/init.ts | 9 +- packages/cli/src/commands/lifecycle.ts | 2 +- packages/cli/src/commands/mcp.ts | 2 +- packages/cli/src/commands/openclaw.ts | 2 +- packages/cli/src/config.ts | 14 +- packages/cli/src/daemon/chain-reset-wipe.ts | 6 +- packages/cli/src/daemon/lifecycle.ts | 58 ++-- packages/cli/src/daemon/routes/status.ts | 2 +- packages/cli/src/daemon/store-runtime.ts | 93 ++++++ packages/cli/src/publisher-runner.ts | 3 +- packages/cli/src/store-wizard.ts | 275 +++++++++++----- packages/cli/test/chain-reset-wipe.test.ts | 30 +- .../test/daemon-http-behavior-extra.test.ts | 2 +- .../test/daemon-startup-validation.test.ts | 85 ++++- .../test/daemon/plugin-routes-api.e2e.test.ts | 2 +- packages/cli/test/helpers/live-daemon.ts | 2 +- packages/cli/test/oxigraph-managed.test.ts | 4 +- packages/cli/test/publisher-cli-smoke.test.ts | 2 +- packages/cli/test/store-health-check.test.ts | 2 +- packages/cli/test/store-identity-tag.test.ts | 2 +- packages/cli/test/store-wizard.test.ts | 77 +++-- .../cli/test/validate-store-config.test.ts | 24 +- .../core/test/ensure-dkg-node-config.test.ts | 2 +- .../test/kafka-plugin-api.e2e.test.ts | 2 +- packages/storage/README.md | 34 -- .../src/adapters/oxigraph-worker-impl.ts | 18 - .../storage/src/adapters/oxigraph-worker.ts | 311 ------------------ packages/storage/src/adapters/oxigraph.ts | 4 +- packages/storage/src/index.ts | 2 - packages/storage/src/triple-store.ts | 13 +- .../storage/test/is-external-backend.test.ts | 1 - .../test/oxigraph-worker-resilience.test.ts | 261 --------------- packages/storage/test/storage.test.ts | 50 +-- scripts/devnet.sh | 8 +- scripts/publisher-smoke-test.sh | 2 +- 44 files changed, 558 insertions(+), 1059 deletions(-) create mode 100644 packages/cli/src/daemon/store-runtime.ts delete mode 100644 packages/storage/src/adapters/oxigraph-worker-impl.ts delete mode 100644 packages/storage/src/adapters/oxigraph-worker.ts delete mode 100644 packages/storage/test/oxigraph-worker-resilience.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 9b534fcc15..e60751dacd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,10 @@ All notable changes to the DKG V9 node are documented here. The format is based ## [Unreleased] +### Removed — `oxigraph-worker` backend support + +- **The embedded `oxigraph-worker` backend is retired.** The storage package no longer exports or registers the worker-thread adapter, `createTripleStore({ backend: "oxigraph-worker" })` fails with an actionable migration error, and CLI config validation refuses explicit `store.backend: "oxigraph-worker"` before daemon boot. Block-less configs now resolve to the daemon-managed `oxigraph-server` default; if a legacy `store.nq` file exists, daemon boot requires `DKG_ACCEPT_STORE_RESET=1` so operators acknowledge the fresh-store cutover. + ### Fixed — `dkg init` on monorepo checkouts + oxigraph-server default for adapter/MCP setups (#960) - **V10 publish/update allowance recovery now translates raw ethers custom-error data before `TooLowAllowance` classification** (`packages/chain/src/evm-adapter.ts`): the shared populate-and-sign recovery gate decodes raw provider revert data first, so a fresh node can force one re-approval and retry when RPC allowance reads lag behind a mined TRAC approval. diff --git a/bench/store-read-latency.bench.ts b/bench/store-read-latency.bench.ts index 28d16edd52..9f2ec700ce 100644 --- a/bench/store-read-latency.bench.ts +++ b/bench/store-read-latency.bench.ts @@ -1,5 +1,3 @@ -import { existsSync } from 'node:fs'; -import { fileURLToPath } from 'node:url'; import { defineSuite } from 'esbench'; // Import the store classes + types from their SPECIFIC source modules rather // than the storage barrel: the barrel also re-exports GraphManager / @@ -8,7 +6,6 @@ import { defineSuite } from 'esbench'; // clean checkout. These adapter modules depend only on `oxigraph` + the local // triple-store types, so the bench needs nothing beyond the storage build. import { OxigraphStore } from '../packages/storage/src/adapters/oxigraph.ts'; -import { OxigraphWorkerStore } from '../packages/storage/src/adapters/oxigraph-worker.ts'; import type { Quad, QueryResult } from '../packages/storage/src/triple-store.ts'; import { GET_TOTAL_TRIPLES_SPARQL, parseRdfInt } from '../packages/cli/src/daemon/metrics-queries.ts'; import { benchAsyncWithHooks } from './support/esbench-case-hooks.ts'; @@ -24,33 +21,11 @@ import { benchAsyncWithHooks } from './support/esbench-case-hooks.ts'; * - the production `getTotalTriples` `COUNT(*)` aggregate the 30s metrics * collector runs (`packages/cli/src/daemon/lifecycle.ts`). * - * Backends (env `DKG_BENCH_STORE_BACKENDS`, comma-separated): - * - `inprocess` — `OxigraphStore`, no build step required. - * - `worker` — `OxigraphWorkerStore`, the PRODUCTION backend; requires - * `pnpm --filter @origintrail-official/dkg-storage build` first (it spawns a - * compiled worker artefact). - * Default: `inprocess`, plus `worker` automatically when its compiled artefact - * is present (so a built tree / CI measures both; an unbuilt tree degrades to - * `inprocess`-only instead of erroring). - * - * CONTENTION — the `(under write load)` cases — is measured ONLY on the `worker` - * backend, by design. The production read-starvation is a property of the - * single-writer oxigraph WORKER, whose message queue serialises a read behind - * queued writes — exactly what an out-of-process MVCC server (#938) relieves. - * The in-process `OxigraphStore` runs its insert/query work SYNCHRONOUSLY on one - * thread, so a read and a write can never truly overlap; a same-thread "writer" - * would only measure event-loop interleaving (not store contention) and could - * report misleadingly idle reads. So `inprocess` runs the idle read-latency - * baselines only. - * * Store sizes via env `DKG_BENCH_STORE_SIZES` (default `1k,50k`). */ -type Backend = 'inprocess' | 'worker'; - interface ReadStore { insert(quads: Quad[]): Promise; - delete(quads: Quad[]): Promise; query(sparql: string): Promise; close(): Promise; } @@ -63,13 +38,6 @@ const READ_LIMIT1 = `SELECT ?s WHERE { GRAPH <${GRAPH}> { ?s ?p ?o } } LIMIT 1`; // synthetic data lives in a named graph, so the `GRAPH ?g` branch carries the scan. const READ_TOTAL_TRIPLES = GET_TOTAL_TRIPLES_SPARQL; -// Bounded write churn: the background writer repeatedly inserts then deletes a -// fixed batch in a region disjoint from the pre-populated base, so it generates -// sustained write work WITHOUT drifting the store size (which would otherwise -// confound the getTotalTriples-under-load measurement). -const CHURN_BATCH = 50; -const CHURN_OFFSET = 1_000_000_000; - const STORE_SIZES: Record = { '1k': 1_000, '10k': 10_000, '50k': 50_000, '200k': 200_000 }; const INSERT_CHUNK = 1_000; @@ -111,46 +79,6 @@ function makeQuads(count: number, offset: number): Quad[] { return quads; } -// The compiled worker artefact sits next to the storage dist build; its presence -// means the `worker` backend can be constructed without throwing. -function workerArtifactAvailable(): boolean { - try { - return existsSync(fileURLToPath(new URL('../packages/storage/dist/adapters/oxigraph-worker-impl.js', import.meta.url))); - } catch { - return false; - } -} - -function resolveBackends(): Backend[] { - const raw = process.env.DKG_BENCH_STORE_BACKENDS?.trim(); - if (!raw) { - if (workerArtifactAvailable()) return ['inprocess', 'worker']; - // Loud, because the worker backend is the one this regression is about — a - // silent inprocess-only run would look like it measured the production path. - console.warn( - '[store-read-latency] worker backend SKIPPED (compiled adapter missing) — ' + - 'measuring `inprocess` idle read latency ONLY, NOT the production write-contention path. ' + - 'Run `pnpm --filter @origintrail-official/dkg-storage build` (or `pnpm bench:store-read`, ' + - 'which builds it) to include the worker backend.', - ); - return ['inprocess']; - } - const known = new Set(['inprocess', 'worker']); - const requested = raw.split(',').map((p) => p.trim().toLowerCase()).filter(Boolean); - for (const b of requested) { - if (!known.has(b as Backend)) { - throw new Error(`Unknown DKG_BENCH_STORE_BACKENDS entry "${b}". Expected: inprocess, worker`); - } - } - if (requested.includes('worker') && !workerArtifactAvailable()) { - throw new Error( - 'DKG_BENCH_STORE_BACKENDS requested "worker", but the compiled adapter is missing. ' + - 'Run `pnpm --filter @origintrail-official/dkg-storage build` first.', - ); - } - return requested.length > 0 ? (requested as Backend[]) : ['inprocess']; -} - function resolveStoreSizeLabels(): string[] { const raw = process.env.DKG_BENCH_STORE_SIZES?.trim(); const labels = raw ? raw.split(',').map((p) => p.trim().toLowerCase()).filter(Boolean) : ['1k', '50k']; @@ -162,15 +90,8 @@ function resolveStoreSizeLabels(): string[] { return labels; } -function createStore(backend: Backend): ReadStore { - // `worker` availability is gated in resolveBackends(), so by the time we get - // here the compiled adapter is present. - return backend === 'worker' ? new OxigraphWorkerStore() : new OxigraphStore(); -} - export default defineSuite({ params: { - backend: resolveBackends(), storeSize: resolveStoreSizeLabels(), }, baseline: { @@ -185,63 +106,20 @@ export default defineSuite({ warmup: 1, }, async setup(scene) { - const backend = scene.params.backend as Backend; const sizeLabel = scene.params.storeSize as string; const quadCount = STORE_SIZES[sizeLabel]; - const store = createStore(backend); + const store: ReadStore = new OxigraphStore(); - const churn = makeQuads(CHURN_BATCH, CHURN_OFFSET); - let writerActive = false; - let writerDone: Promise | undefined; - let writerError: unknown; - - const stopWriter = async (): Promise => { - if (!writerDone) return; - writerActive = false; - const done = writerDone; - writerDone = undefined; - // The loop captures its own failures into `writerError`, so this never rejects. - await done; - // Remove the churn batch so a half-applied cycle (an insert without its - // matching delete) can't leak into the next iteration's getTotalTriples count. - try { - await store.delete(churn); - } catch { - /* store may be mid-teardown */ - } - // Fail fast if the writer died: a stopped writer must never let an - // `(under write load)` case record a successful (effectively idle) sample. - if (writerError !== undefined) { - const err = writerError; - writerError = undefined; - throw new Error(`background writer died during the under-load iteration: ${errorText(err)}`); - } - }; - - // ONE ordered teardown for the scene: stop the writer and await its loop - // BEFORE closing the store, so an in-flight insert/delete can never race a - // closed store/worker. esbench may run scene teardown hooks concurrently, so - // all ordering lives inside this single callback. Registered up-front so the - // store is still closed (and any worker thread terminated) even if the - // population below throws. + // Registered up-front so the store is still closed even if population or a + // benchmark case throws. scene.teardown(async () => { - let stopErr: unknown; - try { - await stopWriter(); - } catch (err) { - stopErr = err; // capture; still close the store below - } - // `close()` is the only place the worker thread is joined, so a failure - // here is a real teardown bug — surface it (log + reject) rather than - // swallow it and let a broken worker benchmark look green. try { await store.close(); } catch (closeErr) { - console.error(`[store-read-latency] store.close() failed (worker thread may not have joined): ${errorText(closeErr)}`); + console.error(`[store-read-latency] store.close() failed: ${errorText(closeErr)}`); throw closeErr; } - if (stopErr !== undefined) throw stopErr; }); // Pre-populate the base graph the reads scan. @@ -256,63 +134,5 @@ export default defineSuite({ benchAsyncWithHooks(scene, 'read getTotalTriples (idle)', async () => { assertCountAtLeast(await store.query(READ_TOTAL_TRIPLES), quadCount, 'read getTotalTriples'); }, {}); - - // Contention is meaningful only on the worker backend (see file docstring): - // the in-process store is single-threaded + synchronous, so reads and writes - // cannot truly overlap. Skip the `(under write load)` cases for it rather - // than report a misleading same-thread number. - if (backend !== 'worker') return; - - // Background writer for the worker-backend `(under write load)` cases, - // scoped to each loaded iteration (not to case-execution order): - // - `beforeIteration` (startWriter) AWAITS one full insert/delete cycle so - // writes are provably queued in the worker before the measured read. - // - a writer that dies is recorded in `writerError` and surfaced as a - // FAILED case — startWriter throws if it died before the first cycle, - // the workload throws if it died mid-measurement, and stopWriter throws - // if it died just after — so a broken writer can never silently degrade - // into an idle read. - // - `afterIteration` (stopWriter) stops it and clears the churn batch. - const startWriter = async (): Promise => { - if (writerActive) return; - writerActive = true; - writerError = undefined; - let signalFirstCycle!: () => void; - const firstCycle = new Promise((resolve) => { signalFirstCycle = resolve; }); - writerDone = (async () => { - try { - let signalled = false; - while (writerActive) { - await store.insert(churn); - await store.delete(churn); - if (!signalled) { signalled = true; signalFirstCycle(); } - } - } catch (err) { - writerError = err; - } finally { - // Unblock the barrier even if the first cycle threw, so a writer - // failure surfaces (below) instead of hanging the benchmark. - signalFirstCycle(); - } - })(); - await firstCycle; - if (writerError !== undefined) { - throw new Error(`background writer failed before its first write cycle: ${errorText(writerError)}`); - } - }; - - benchAsyncWithHooks(scene, 'read LIMIT 1 (under write load)', async () => { - assertNonEmptySelect(await store.query(READ_LIMIT1), 'read LIMIT 1'); - if (writerError !== undefined) { - throw new Error(`background writer died during the measurement: ${errorText(writerError)}`); - } - }, { beforeIteration: startWriter, afterIteration: stopWriter }); - - benchAsyncWithHooks(scene, 'read getTotalTriples (under write load)', async () => { - assertCountAtLeast(await store.query(READ_TOTAL_TRIPLES), quadCount, 'read getTotalTriples'); - if (writerError !== undefined) { - throw new Error(`background writer died during the measurement: ${errorText(writerError)}`); - } - }, { beforeIteration: startWriter, afterIteration: stopWriter }); }, }); diff --git a/docs/UPGRADE_TO_RC17.md b/docs/UPGRADE_TO_RC17.md index a54fb3d6d5..bdeb1c6e01 100644 --- a/docs/UPGRADE_TO_RC17.md +++ b/docs/UPGRADE_TO_RC17.md @@ -1,5 +1,9 @@ # Upgrading to `v10.0.0-rc.17` — land it clean +> **Historical note:** this guide describes the rc.17 storage behavior. Current +> v10 builds have retired `oxigraph-worker`; use `oxigraph-server` or an +> external SPARQL backend for daemon storage. + **Audience:** DKG **node operators** (edge + core) on V10 Testnet (Base Sepolia) upgrading from any pre-rc.17 build (rc.12 → rc.16), plus anyone spinning up a node for the first time on rc.17. diff --git a/docs/use-dkg/storage-sparql-http.md b/docs/use-dkg/storage-sparql-http.md index 0094f229c4..3a3cd811cb 100644 --- a/docs/use-dkg/storage-sparql-http.md +++ b/docs/use-dkg/storage-sparql-http.md @@ -7,7 +7,7 @@ doc_type: how-to # Using an external SPARQL store (Oxigraph server, etc.) -The DKG node can use any **SPARQL 1.1 Protocol**–compliant store you run yourself, instead of its default daemon-managed local Oxigraph server (or the embedded `oxigraph-worker` fallback). That gives you: +The DKG node can use any **SPARQL 1.1 Protocol**–compliant store you run yourself, instead of its default daemon-managed local Oxigraph server. That gives you: - **Real on-disk persistence** (e.g. Oxigraph server with RocksDB) - **Larger graphs** without holding everything in the Node process @@ -91,4 +91,4 @@ await agent.start(); New installs default to a **daemon-managed local Oxigraph server** (`store.backend: "oxigraph-server"`): `dkg init`, `dkg openclaw/hermes/mcp setup`, or accepting the wizard default writes this block. The daemon fetches the pinned `oxigraph` binary on first boot and runs it on loopback, giving MVCC concurrent reads and incremental RocksDB persistence. -If a config has **no** `store` block at all, the runtime falls back to the embedded in-process **`oxigraph-worker`** (a single-writer store that rewrites its on-disk N-Quads dump under `dataDir` on every flush) — fine for development and small nodes. For very large graphs or existing infrastructure, use `sparql-http` with an external store. +If a config has **no** `store` block at all, the runtime now uses the same daemon-managed **`oxigraph-server`** default. The old embedded **`oxigraph-worker`** backend has been retired; configs that still name it fail fast with a migration message. For very large graphs or existing infrastructure, use `sparql-http` with an external store. diff --git a/packages/agent/src/dkg-agent-helpers.ts b/packages/agent/src/dkg-agent-helpers.ts index 3d53b4157a..5e7a39b814 100644 --- a/packages/agent/src/dkg-agent-helpers.ts +++ b/packages/agent/src/dkg-agent-helpers.ts @@ -403,7 +403,6 @@ export function applyDefaultLargeLiteralStorage( export function isLocalOxigraphConfig(storeConfig: TripleStoreConfig): boolean { return storeConfig.backend === 'oxigraph' - || storeConfig.backend === 'oxigraph-worker' || storeConfig.backend === 'oxigraph-persistent'; } diff --git a/packages/agent/src/dkg-agent-types.ts b/packages/agent/src/dkg-agent-types.ts index a1378d9285..64a099a579 100644 --- a/packages/agent/src/dkg-agent-types.ts +++ b/packages/agent/src/dkg-agent-types.ts @@ -870,7 +870,7 @@ export interface DKGAgentConfig { }>; dataDir?: string; store?: TripleStore; - /** Triple store backend configuration (e.g. oxigraph-worker, blazegraph). If omitted, defaults to oxigraph-worker when dataDir is set. */ + /** Triple store backend configuration (e.g. oxigraph-server runtime view, oxigraph-persistent, blazegraph). If omitted, dataDir agents use oxigraph-persistent. */ storeConfig?: TripleStoreConfig; /** Out-of-line storage for large public SWM RDF literal object terms. Defaults on for local Oxigraph-backed dataDir stores. */ largeLiteralStorage?: LargeLiteralStorageConfig; diff --git a/packages/agent/src/dkg-agent.ts b/packages/agent/src/dkg-agent.ts index 72947a212a..1c1d3f2a70 100644 --- a/packages/agent/src/dkg-agent.ts +++ b/packages/agent/src/dkg-agent.ts @@ -503,11 +503,11 @@ export class DKGAgent extends DKGAgentBase { const { join } = await import('node:path'); const persistPath = join(config.dataDir, 'store.nq'); store = await createTripleStore({ - backend: 'oxigraph-worker', + backend: 'oxigraph-persistent', options: { path: persistPath }, largeLiteralStorage: defaultLargeLiteralStorage(config.dataDir, config.largeLiteralStorage), }); - log.info(ctx, `Persistent triple store (worker thread): ${persistPath}`); + log.info(ctx, `Persistent triple store: ${persistPath}`); } else { store = await createTripleStore({ backend: 'oxigraph' }); log.warn(ctx, `No dataDir — triple store is in-memory (data will be lost on restart)`); diff --git a/packages/agent/src/sync/responder/sync-handler.ts b/packages/agent/src/sync/responder/sync-handler.ts index 5c00015fa2..6f48e42444 100644 --- a/packages/agent/src/sync/responder/sync-handler.ts +++ b/packages/agent/src/sync/responder/sync-handler.ts @@ -351,7 +351,7 @@ export function registerSyncHandler(params: RegisterSyncHandlerParams): void { warnedPreDispatchCancellation = true; logWarn( createOperationContext('sync'), - 'Sync responder is using a store backend whose query AbortSignal is pre-dispatch only; in-flight sync queries cannot release responder capacity until the synchronous store call returns. Use oxigraph-worker or an HTTP SPARQL backend for interruptible long-query cancellation.', + 'Sync responder is using a store backend whose query AbortSignal is pre-dispatch only; in-flight sync queries cannot release responder capacity until the synchronous store call returns. Use an HTTP SPARQL backend for interruptible long-query cancellation.', ); } if (isWorkspace) { diff --git a/packages/cli/src/commands/hermes.ts b/packages/cli/src/commands/hermes.ts index fdf840e212..56e6778074 100644 --- a/packages/cli/src/commands/hermes.ts +++ b/packages/cli/src/commands/hermes.ts @@ -185,7 +185,7 @@ hermesCmd ) .option( '--store ', - 'Triple-store backend (oxigraph | blazegraph | sparql-http). Validates the URL and persists the store block after setup.', + 'Triple-store backend (oxigraph-server | oxigraph | blazegraph | sparql-http). Validates the URL and persists the store block after setup.', ) .option( '--store-url ', diff --git a/packages/cli/src/commands/init.ts b/packages/cli/src/commands/init.ts index d7c869f45c..2121a81592 100644 --- a/packages/cli/src/commands/init.ts +++ b/packages/cli/src/commands/init.ts @@ -176,7 +176,7 @@ program .option('--role ', "Node role: 'edge' (default; personal laptop / behind NAT) or 'core' (24/7 relay / SLA)") .option( '--store ', - 'Pre-fill the triple-store backend prompt (oxigraph | blazegraph | sparql-http).', + 'Pre-fill the triple-store backend prompt (oxigraph-server | oxigraph | blazegraph | sparql-http).', ) .option( '--store-url ', @@ -436,9 +436,8 @@ program chain: chainSection ?? existing.chain, auth: { enabled: enableAuth, tokens: existing.auth?.tokens }, // Persist the chosen backend. `storeBlock === null` from the - // wizard means "use the local default" — we explicitly clear any - // existing block so re-running `dkg init` to switch from - // blazegraph back to oxigraph actually applies. + // wizard means "leave the store block omitted"; daemon boot treats + // that as the managed `oxigraph-server` default. store: storeBlock ?? undefined, }; await saveConfig(config); @@ -475,7 +474,7 @@ program const endpoint = o?.url ?? o?.queryEndpoint; return `${storeBlock.backend}${endpoint ? ` (${endpoint})` : ''}`; })() - : 'oxigraph (local default)' + : 'oxigraph-server (default)' }`, ); { diff --git a/packages/cli/src/commands/lifecycle.ts b/packages/cli/src/commands/lifecycle.ts index f1d28f69fd..98e05039ca 100644 --- a/packages/cli/src/commands/lifecycle.ts +++ b/packages/cli/src/commands/lifecycle.ts @@ -290,7 +290,7 @@ program // remote store. Quad count = null is rare in practice (cached // every 30 s on the daemon side) so when it shows up the // operator should treat it as an alert, not a no-op. - const backend = s.storeBackend ?? 'oxigraph-worker'; + const backend = s.storeBackend ?? 'oxigraph-server'; if (s.storeUrl) { const quads = s.storeQuads == null ? 'UNREACHABLE' : `${s.storeQuads.toLocaleString()} quads`; console.log(` Store: ${backend} (${s.storeUrl}) — ${quads}`); diff --git a/packages/cli/src/commands/mcp.ts b/packages/cli/src/commands/mcp.ts index 765bebce5c..dd21b19f91 100644 --- a/packages/cli/src/commands/mcp.ts +++ b/packages/cli/src/commands/mcp.ts @@ -156,7 +156,7 @@ mcpCmd .option('--yes', 'Auto-confirm per-client registrations (default false: prompt interactively in TTY mode; non-TTY auto-confirms — pass `--yes` in scripts for the safer scripted-environment posture)') .option( '--store ', - 'Triple-store backend (oxigraph | blazegraph | sparql-http). Validates the URL and persists the store block after setup.', + 'Triple-store backend (oxigraph-server | oxigraph | blazegraph | sparql-http). Validates the URL and persists the store block after setup.', ) .option( '--store-url ', diff --git a/packages/cli/src/commands/openclaw.ts b/packages/cli/src/commands/openclaw.ts index 9fd4aaf67e..d1c001e02d 100644 --- a/packages/cli/src/commands/openclaw.ts +++ b/packages/cli/src/commands/openclaw.ts @@ -123,7 +123,7 @@ openclawCmd .option('--fund', 'Fund wallets via testnet faucet (default)') .option( '--store ', - 'Triple-store backend (oxigraph | blazegraph | sparql-http). Validates the URL via an ASK probe and persists the store block after setup completes.', + 'Triple-store backend (oxigraph-server | oxigraph | blazegraph | sparql-http). Validates the URL via an ASK probe and persists the store block after setup completes.', ) .option( '--store-url ', diff --git a/packages/cli/src/config.ts b/packages/cli/src/config.ts index 8211bd9a7f..71f53160dd 100644 --- a/packages/cli/src/config.ts +++ b/packages/cli/src/config.ts @@ -533,7 +533,7 @@ export interface DkgConfig { drag?: DragConfig; /** Block explorer URL for TX links (default: derived from chainId). */ blockExplorerUrl?: string; - /** Triple store backend override (default: oxigraph-worker with file persistence). */ + /** Triple store backend override (default: daemon-managed oxigraph-server). */ store?: { backend: string; options?: Record; graphSetIndex?: boolean | GraphSetIndexConfig }; /** * Intentional cap on how many persisted context-graph subscriptions a node @@ -1505,6 +1505,16 @@ export interface StoreConfigValidationError { export function validateStoreConfig(config: DkgConfig): StoreConfigValidationError[] { const errors: StoreConfigValidationError[] = []; const backend = config.store?.backend; + if (backend === 'oxigraph-worker') { + return [{ + field: 'store.backend', + message: + `${EXTERNAL_VALIDATION_PREFIX} "oxigraph-worker" is no longer supported. ` + + `Set store.backend to "oxigraph-server" for the daemon-managed local store, ` + + `"sparql-http" or "blazegraph" for an external store, or "oxigraph" ` + + `for an embedded development store.`, + }]; + } // Mirror of `isExternalBackend` from @origintrail-official/dkg-storage. // Duplicated here to keep config.ts free of upward dependencies on the // storage package (config.ts is leaf-imported by many other modules). @@ -1521,7 +1531,7 @@ export function validateStoreConfig(config: DkgConfig): StoreConfigValidationErr `${EXTERNAL_VALIDATION_PREFIX} is "blazegraph" but ` + `store.options.url is missing. Set it to the SPARQL endpoint URL ` + `(e.g. http://127.0.0.1:9999/bigdata/namespace/mynode/sparql) or ` + - `switch backend to oxigraph-worker.`, + `switch backend to oxigraph-server.`, }); } } else if (backend === 'sparql-http') { diff --git a/packages/cli/src/daemon/chain-reset-wipe.ts b/packages/cli/src/daemon/chain-reset-wipe.ts index f990ed9966..7e5066aa79 100644 --- a/packages/cli/src/daemon/chain-reset-wipe.ts +++ b/packages/cli/src/daemon/chain-reset-wipe.ts @@ -452,7 +452,7 @@ export interface BackendSwitchDetectOptions { /** * Backend name from the current config. Pass the effective value * including the default — e.g. when `config.store?.backend` is - * undefined, callers should pass `'oxigraph-worker'` so the check + * undefined, callers should pass `'oxigraph-server'` so the check * is symmetric across "no store block" ↔ "explicit store block". */ currentBackend: string; @@ -491,8 +491,8 @@ export function detectBackendSwitch( : null; // First boot or legacy state file: silently record and move on. We - // explicitly do NOT treat null-previous as a "switch from - // oxigraph-worker"; that would re-warn every operator who upgrades + // explicitly do NOT treat null-previous as a "switch from the old + // implicit backend"; that would re-warn every operator who upgrades // into this release without ever having touched their store // configuration. Only operator-visible config changes between two // recorded backends count as a switch. diff --git a/packages/cli/src/daemon/lifecycle.ts b/packages/cli/src/daemon/lifecycle.ts index 822a06c26b..47e7a644fc 100644 --- a/packages/cli/src/daemon/lifecycle.ts +++ b/packages/cli/src/daemon/lifecycle.ts @@ -39,7 +39,7 @@ import { import { execSync, exec, execFile } from "node:child_process"; import { promisify } from "node:util"; import { join, dirname, resolve } from 'node:path'; -import { existsSync, readdirSync, readFileSync, openSync, closeSync, writeFileSync as fsWriteFileSync, unlinkSync } from 'node:fs'; +import { readdirSync, readFileSync, openSync, closeSync, writeFileSync as fsWriteFileSync, unlinkSync } from 'node:fs'; // Namespace import: our Phase-8 install-context builder (~line 290) calls // `osModule.homedir()`, and the later agent-identity probe (~line 6851) // uses `osModule.hostname()` + `osModule.userInfo()`. v10-rc's new @@ -290,6 +290,10 @@ import { } from './store-health-check.js'; import { startManagedOxigraph } from './oxigraph-managed.js'; import type { OxigraphServerHandle } from './oxigraph-server.js'; +import { + resolveDaemonStoreBootPlan, + resolveDaemonStoreRuntime, +} from './store-runtime.js'; import { resetNatStatus, startNatStatusWatcher } from './nat-status.js'; import { OPENCLAW_UI_CONNECT_TIMEOUT_MS, @@ -916,6 +920,23 @@ export async function runDaemonInner( ...resolveNetworkDefaultContextGraphs(network), ]), ]; + const acceptStoreReset = process.env.DKG_ACCEPT_STORE_RESET === '1'; + const storeBoot = resolveDaemonStoreBootPlan({ + config, + dataDir: dkgDir(), + acceptStoreReset, + }); + + if (storeBoot.validateOperatorConfigFirst) { + exitOnStoreConfigErrors(storeBoot.operatorConfig, log); + } + if (storeBoot.abortMessage) { + log(storeBoot.abortMessage); + process.exit(1); + } + if (storeBoot.notice) { + log(storeBoot.notice); + } // Auto-wipe per-node chain-state derived files (oxigraph store, publish // journal, random-sampling WAL) when the maintainer bumps @@ -932,8 +953,8 @@ export async function runDaemonInner( // silently would look like data loss to the operator. const backendSwitch = detectBackendSwitch({ dataDir: dkgDir(), - currentBackend: config.store?.backend ?? 'oxigraph-worker', - acceptStoreReset: process.env.DKG_ACCEPT_STORE_RESET === '1', + currentBackend: storeBoot.effectiveStore.backend, + acceptStoreReset, log, }); if (backendSwitch.aborted) { @@ -952,7 +973,7 @@ export async function runDaemonInner( let managedOxigraph: OxigraphServerHandle | null = null; let managed: Awaited> = null; try { - managed = await startManagedOxigraph({ config, dataDir: dkgDir(), log }); + managed = await startManagedOxigraph({ config: storeBoot.effectiveConfig, dataDir: dkgDir(), log }); if (managed) { managedOxigraph = managed.handle; // Every remaining fatal boot path (config validation, store health @@ -965,7 +986,7 @@ export async function runDaemonInner( } catch (err) { log( `[STORE] failed to start managed Oxigraph server: ${(err as Error).message}\n` + - `Fix the cause, or switch \`store.backend\` to oxigraph-worker (embedded) or ` + + `Fix the cause, or switch \`store.backend\` to ` + `sparql-http (operator-managed endpoint) in ~/.dkg/config.json.`, ); process.exit(1); @@ -983,22 +1004,13 @@ export async function runDaemonInner( // For the directory-backed blob/snapshot stores we use the managed // defaults (the rewritten sparql-http backend has no `options.path` to // infer a directory from, unlike the local Oxigraph backend). - const runtimeStore = managed?.storeConfig ?? config.store; - const runtimeLargeLiteralStorage = - managed?.largeLiteralStorage ?? config.largeLiteralStorage; - const runtimeSnapshotStorage = - managed?.sharedMemoryPublicSnapshotStorage ?? config.sharedMemoryPublicSnapshotStorage; - // Config view used only for the boot-time store validation/health steps - // below: same as `config` but with the runtime store/blob/snapshot values - // swapped in, so a managed config validates against what actually runs. - const runtimeStoreConfig: DkgConfig = managed - ? { - ...config, - store: runtimeStore, - largeLiteralStorage: runtimeLargeLiteralStorage, - sharedMemoryPublicSnapshotStorage: runtimeSnapshotStorage, - } - : config; + const storeRuntime = resolveDaemonStoreRuntime(storeBoot, managed); + const { + runtimeStore, + runtimeConfig: runtimeStoreConfig, + runtimeLargeLiteralStorage, + runtimeSnapshotStorage, + } = storeRuntime; // Refuse to start on invalid external-backend config (missing URL, // missing blob/snapshot directory). This fires before the health @@ -1681,7 +1693,7 @@ export async function runDaemonInner( try { const runtime = await startPublisherRuntimeIfEnabled({ dataDir: dkgDir(), - config, + config: storeBoot.effectiveConfig, store: agent.store, keypair: agent.wallet.keypair, chainBase: publisherChainBase, @@ -2977,7 +2989,7 @@ export async function runDaemonInner( agent, publisherControl, publisherRuntime, - config, + storeBoot.effectiveConfig, startedAt, dashDb, opWallets, diff --git a/packages/cli/src/daemon/routes/status.ts b/packages/cli/src/daemon/routes/status.ts index 9535ea21d3..48a2128bfa 100644 --- a/packages/cli/src/daemon/routes/status.ts +++ b/packages/cli/src/daemon/routes/status.ts @@ -664,7 +664,7 @@ export async function handleStatusRoutes(ctx: RequestContext): Promise { networkConfig: resolveNetworkConfigName(config), networkId, networkName: network?.networkName ?? null, - storeBackend: config.store?.backend ?? "oxigraph-worker", + storeBackend: config.store?.backend ?? "oxigraph-server", // External backend visibility (RFC 120 / plan PR 1 item 3). For // local backends both fields stay null so the response shape is // stable across deployments. diff --git a/packages/cli/src/daemon/store-runtime.ts b/packages/cli/src/daemon/store-runtime.ts new file mode 100644 index 0000000000..606b401326 --- /dev/null +++ b/packages/cli/src/daemon/store-runtime.ts @@ -0,0 +1,93 @@ +import { existsSync } from 'node:fs'; +import { join } from 'node:path'; +import type { DkgConfig } from '../config.js'; +import type { ManagedOxigraphResult } from './oxigraph-managed.js'; + +type StoreConfig = NonNullable; + +export interface DaemonStoreBootPlan { + /** Operator-facing config exactly as loaded from disk / CLI. */ + operatorConfig: DkgConfig; + /** Config with the implicit daemon default materialized for boot steps. */ + effectiveConfig: DkgConfig; + /** Store backend used for backend-switch detection and managed startup. */ + effectiveStore: StoreConfig; + /** True when explicit validation should run before migration gates. */ + validateOperatorConfigFirst: boolean; + /** Fatal startup message; caller logs and exits before touching runtime store. */ + abortMessage?: string; + /** Non-fatal startup notice, e.g. acknowledged legacy default cutover. */ + notice?: string; +} + +export interface DaemonStoreRuntimePlan extends DaemonStoreBootPlan { + /** Store config consumed by validation, health probes, wipe, and the agent. */ + runtimeStore: StoreConfig; + /** Config view with runtime store/blob/snapshot values swapped in. */ + runtimeConfig: DkgConfig; + runtimeLargeLiteralStorage: DkgConfig['largeLiteralStorage']; + runtimeSnapshotStorage: DkgConfig['sharedMemoryPublicSnapshotStorage']; +} + +function defaultDaemonStore(): StoreConfig { + return { backend: 'oxigraph-server', options: {} }; +} + +export function resolveDaemonStoreBootPlan(opts: { + config: DkgConfig; + dataDir: string; + acceptStoreReset: boolean; +}): DaemonStoreBootPlan { + const { config, dataDir, acceptStoreReset } = opts; + const effectiveStore = config.store ?? defaultDaemonStore(); + const effectiveConfig = config.store ? config : { ...config, store: effectiveStore }; + const legacyStorePath = join(dataDir, 'store.nq'); + + const plan: DaemonStoreBootPlan = { + operatorConfig: config, + effectiveConfig, + effectiveStore, + validateOperatorConfigFirst: config.store?.backend === 'oxigraph-worker', + }; + + if (!config.store && existsSync(legacyStorePath) && !acceptStoreReset) { + plan.abortMessage = + `[STORE] oxigraph-worker support has been removed, but this node has a legacy ` + + `store.nq from the old implicit worker default.\n` + + `Set store.backend to "oxigraph-server" (or an external SPARQL backend) and ` + + `restart with DKG_ACCEPT_STORE_RESET=1 to acknowledge the fresh-store cutover. ` + + `The legacy store.nq file is left untouched for manual backup or migration.`; + } else if (!config.store && acceptStoreReset) { + plan.notice = + '[STORE] no store block found; using oxigraph-server. Legacy store.nq, if present, is left untouched.'; + } + + return plan; +} + +export function resolveDaemonStoreRuntime( + bootPlan: DaemonStoreBootPlan, + managed: ManagedOxigraphResult | null, +): DaemonStoreRuntimePlan { + const runtimeStore = managed?.storeConfig ?? bootPlan.effectiveStore; + const runtimeLargeLiteralStorage = + managed?.largeLiteralStorage ?? bootPlan.operatorConfig.largeLiteralStorage; + const runtimeSnapshotStorage = + managed?.sharedMemoryPublicSnapshotStorage ?? bootPlan.operatorConfig.sharedMemoryPublicSnapshotStorage; + const runtimeConfig: DkgConfig = managed + ? { + ...bootPlan.effectiveConfig, + store: runtimeStore, + largeLiteralStorage: runtimeLargeLiteralStorage, + sharedMemoryPublicSnapshotStorage: runtimeSnapshotStorage, + } + : bootPlan.effectiveConfig; + + return { + ...bootPlan, + runtimeStore, + runtimeConfig, + runtimeLargeLiteralStorage, + runtimeSnapshotStorage, + }; +} diff --git a/packages/cli/src/publisher-runner.ts b/packages/cli/src/publisher-runner.ts index f68b70834d..395a109c66 100644 --- a/packages/cli/src/publisher-runner.ts +++ b/packages/cli/src/publisher-runner.ts @@ -536,7 +536,7 @@ async function createPublisherStore(dataDir: string, config: DkgConfig): Promise } return await createTripleStore({ - backend: 'oxigraph-worker', + backend: 'oxigraph-persistent', options: { path: join(dataDir, 'store.nq') }, largeLiteralStorage: defaultLargeLiteralStorage(dataDir, config), }); @@ -563,7 +563,6 @@ export function createPublicSnapshotStore( function isLocalOxigraphStoreConfig(storeConfig: { backend?: unknown }): boolean { return storeConfig.backend === 'oxigraph' - || storeConfig.backend === 'oxigraph-worker' || storeConfig.backend === 'oxigraph-persistent'; } diff --git a/packages/cli/src/store-wizard.ts b/packages/cli/src/store-wizard.ts index 4eac50b144..b209989041 100644 --- a/packages/cli/src/store-wizard.ts +++ b/packages/cli/src/store-wizard.ts @@ -67,9 +67,9 @@ export interface PromptStoreBackendOptions { export interface PromptStoreBackendResult { /** - * Persisted store block. `null` means "use the local default" (caller - * should omit the field or set it to undefined; saveConfig drops - * undefined keys). + * Persisted store block. `null` means "leave the store block omitted"; + * daemon boot treats an omitted store block as the managed `oxigraph-server` + * default. * * `managedByDkg: true` is set by the Docker provisioner branch only; * manual URLs always get `managedByDkg: false`. The chain-reset-wipe @@ -80,15 +80,8 @@ export interface PromptStoreBackendResult { storeBlock: ExternalStoreBlock | LocalStoreBlock | null; } -// An explicit embedded/local store block carried through verbatim on an -// Enter-through re-init. `null` still means "no block — runtime default", but -// when a node already pinned a local backend with custom `options` (e.g. the -// `options.path` the oxigraph-worker adapter reads for its persistence file), -// returning that block instead of `null` keeps re-init idempotent: cli init -// writes `store: storeBlock ?? undefined`, so a `null` would clear the block -// and relocate the store on the next boot. export type LocalStoreBlock = { - backend: 'oxigraph' | 'oxigraph-worker' | 'oxigraph-persistent'; + backend: 'oxigraph' | 'oxigraph-persistent'; options?: Record; }; @@ -136,6 +129,108 @@ function externalStoreBlock( }; } +const STORE_BACKENDS = { + 'oxigraph-server': { + kind: 'managed-local', + retired: false, + menu: true, + label: 'oxigraph-server (managed local server — recommended)', + }, + oxigraph: { + kind: 'local', + retired: false, + menu: true, + label: 'oxigraph (embedded in-memory store — development only)', + requiresExistingPath: false, + }, + 'oxigraph-persistent': { + kind: 'local', + retired: false, + menu: false, + requiresExistingPath: true, + }, + blazegraph: { + kind: 'external', + retired: false, + menu: true, + label: 'blazegraph (external SPARQL endpoint)', + }, + 'sparql-http': { + kind: 'external', + retired: false, + menu: false, + }, + 'oxigraph-worker': { + kind: 'retired', + retired: true, + menu: false, + }, +} as const; + +type StoreBackend = keyof typeof STORE_BACKENDS; +type SupportedStoreBackend = Exclude; +type MenuStoreBackend = 'oxigraph-server' | 'oxigraph' | 'blazegraph'; +type ExternalStoreBackend = 'blazegraph' | 'sparql-http'; + +const DEFAULT_STORE_BACKEND: SupportedStoreBackend = 'oxigraph-server'; +const UNSUPPORTED_WORKER_BACKEND: StoreBackend = 'oxigraph-worker'; + +function storeBackendNames(): StoreBackend[] { + return Object.keys(STORE_BACKENDS) as StoreBackend[]; +} + +function supportedBackendNames(): SupportedStoreBackend[] { + return storeBackendNames().filter((backend): backend is SupportedStoreBackend => ( + !STORE_BACKENDS[backend].retired + )); +} + +function supportedBackendList(): string { + return supportedBackendNames().join(', '); +} + +function menuBackendChoices(): MenuStoreBackend[] { + return storeBackendNames().filter((backend): backend is MenuStoreBackend => ( + !STORE_BACKENDS[backend].retired && STORE_BACKENDS[backend].menu + )); +} + +function isKnownStoreBackend(backend: string | undefined): backend is StoreBackend { + return backend !== undefined && Object.prototype.hasOwnProperty.call(STORE_BACKENDS, backend); +} + +function isSupportedExistingBackend(backend: string | undefined): backend is SupportedStoreBackend { + return isKnownStoreBackend(backend) && !STORE_BACKENDS[backend].retired; +} + +function unsupportedWorkerBackendError(source: string): Error { + return new Error( + `${source} "oxigraph-worker" is no longer supported. ` + + `Use one of: ${supportedBackendList()}.`, + ); +} + +function unknownBackendError(source: 'prompt' | '--store', backend: string): Error { + if (source === '--store') { + return new Error(`--store must be one of: ${supportedBackendList()} (got "${backend}")`); + } + return new Error(`Unknown store backend "${backend}". Expected one of: ${supportedBackendList()}.`); +} + +function parseBackendAnswer( + input: string, + defaultBackend: string, + choices: readonly MenuStoreBackend[], +): string { + return /^\d+$/.test(input) + ? (choices[parseInt(input, 10) - 1] ?? defaultBackend) + : input.toLowerCase(); +} + +function hasStorePath(store: PromptStoreBackendOptions['existingStore'] | undefined): boolean { + return typeof store?.options?.path === 'string' && store.options.path.trim().length > 0; +} + export async function promptStoreBackend( opts: PromptStoreBackendOptions, ): Promise { @@ -156,35 +251,17 @@ export async function promptStoreBackend( : undefined; // `oxigraph-server` (daemon-managed local RocksDB server) is the default - // local backend: it gives MVCC concurrent reads + incremental persistence, - // whereas `oxigraph` (the embedded in-process worker) rewrites the whole - // N-Quads dump on every flush. The in-process worker stays available as a - // minimal-footprint / single-reader option. NOTE: this only changes what a - // *fresh / block-less* `dkg init` writes — the runtime fallback for configs - // with no `store` block stays `oxigraph-worker`, so the existing fleet keeps - // booting unchanged on auto-update (only an explicit re-init flips a node, - // and the daemon's STORE-SWITCH guard makes that an opt-in, not silent). - // Keep ANY explicit existing backend as the default answer — including the - // embedded `oxigraph` / `oxigraph-worker` / `oxigraph-persistent` variants. - // Keeping the *exact* variant (rather than normalising the worker variants - // onto the listed `oxigraph` choice) matters for distinguishing intent: an - // Enter-through resolves the default back to that exact backend (so the - // preserve branch below keeps the block + custom options), whereas explicitly - // picking option `2` ("oxigraph") resolves to a *different* answer and is - // treated as a real switch to the plain embedded worker. Only a truly absent - // store config (fresh install / block-less node) falls through to the new - // `oxigraph-server` default. + // local backend: it gives MVCC concurrent reads + incremental persistence. + // The old `oxigraph-worker` fallback is retired; configs that still name it + // are not preserved on Enter-through. + if (existingBackend === UNSUPPORTED_WORKER_BACKEND) { + log(' Existing store.backend "oxigraph-worker" is retired; defaulting to oxigraph-server.'); + } const defaultBackend = opts.flagBackend - ?? (existingBackend === 'blazegraph' || existingBackend === 'sparql-http' || existingBackend === 'oxigraph-server' - || existingBackend === 'oxigraph' || existingBackend === 'oxigraph-worker' || existingBackend === 'oxigraph-persistent' + ?? (isSupportedExistingBackend(existingBackend) ? existingBackend - : 'oxigraph-server'); - const backendChoices = ['oxigraph-server', 'oxigraph', 'blazegraph'] as const; - const backendLabels: Record = { - 'oxigraph-server': 'oxigraph-server (managed local server — recommended)', - 'oxigraph': 'oxigraph (embedded in-process worker)', - 'blazegraph': 'blazegraph (external SPARQL endpoint)', - }; + : DEFAULT_STORE_BACKEND); + const backendChoices = menuBackendChoices(); // `sparql-http` is intentionally not listed (advanced bring-your-own-server // option) but is still accepted when typed or inherited from an existing // config / `--store` flag. Resolve the default *answer* by name for unlisted @@ -198,7 +275,7 @@ export async function promptStoreBackend( log(' Triple store backend:'); for (let i = 0; i < backendChoices.length; i++) { const choice = backendChoices[i]; - log(` ${i + 1}) ${backendLabels[choice] ?? choice}`); + log(` ${i + 1}) ${STORE_BACKENDS[choice].label}`); } // When the inherited/flagged backend isn't one of the numbered choices // (e.g. `sparql-http`), spell out that pressing Enter keeps it and that a @@ -216,15 +293,21 @@ export async function promptStoreBackend( // out-of-range number (typo like "4") falls back to `defaultBackend` — // i.e. the recommended option shown to the operator — rather than a // hard-coded `oxigraph`, so a fat-fingered digit on a fresh install no - // longer silently downgrades the node to the embedded worker. - const backendAnswer = /^\d+$/.test(backendInput) - ? (backendChoices[parseInt(backendInput, 10) - 1] ?? defaultBackend) - : backendInput.toLowerCase(); + // longer silently downgrades the node to an embedded backend. + const backendAnswer = parseBackendAnswer(backendInput, defaultBackend, backendChoices); + + if (backendAnswer === UNSUPPORTED_WORKER_BACKEND) { + throw unsupportedWorkerBackendError('store backend'); + } + if (!isSupportedExistingBackend(backendAnswer)) { + throw unknownBackendError('prompt', backendAnswer); + } + const backendPolicy = STORE_BACKENDS[backendAnswer]; // `oxigraph-server` (daemon-managed local server) is the default numbered // choice and is also accepted by name. No URL prompt or probe: the endpoint // doesn't exist until the daemon spawns it at boot. - if (backendAnswer === 'oxigraph-server') { + if (backendPolicy.kind === 'managed-local') { log(' Using a daemon-managed local Oxigraph server (started on first daemon boot).'); // Preserve existing managed-server overrides (port/location/cacheDir) on an // Enter-through: `dkg init` persists this block, so returning empty options @@ -237,28 +320,25 @@ export async function promptStoreBackend( return { storeBlock: { backend: 'oxigraph-server', options: prevOptions } }; } - if (backendAnswer !== 'blazegraph' && backendAnswer !== 'sparql-http') { - // Embedded in-process worker. Preserve an existing explicit local store - // block verbatim ONLY when the operator kept that same backend — i.e. the - // resolved answer equals the existing backend (an Enter-through, which - // resolves the default back to the exact existing variant). That keeps a - // re-init idempotent and never drops custom `options` (e.g. the worker's - // `options.path`) or relocates the store. An *explicit* switch — picking - // option `2`/"oxigraph" on a node currently using `oxigraph-worker` / - // `oxigraph-persistent`, a switch from an external/server backend, or a - // fresh / block-less init — resolves to a different answer and falls - // through to `null`, so `dkg init` clears the old block as intended. - if ( - (backendAnswer === 'oxigraph' || - backendAnswer === 'oxigraph-worker' || - backendAnswer === 'oxigraph-persistent') && - backendAnswer === existingBackend - ) { - return { storeBlock: { backend: existingBackend, options: opts.existingStore?.options } }; + if (backendPolicy.kind === 'local') { + const localBackend = backendAnswer as LocalStoreBlock['backend']; + if (STORE_BACKENDS[localBackend].requiresExistingPath && !hasStorePath(opts.existingStore)) { + throw new Error( + 'store backend "oxigraph-persistent" requires store.options.path; ' + + 'set it manually in config.json or use "oxigraph-server".', + ); } - return { storeBlock: null }; + const prevOptions = + localBackend === existingBackend && opts.existingStore?.options + ? opts.existingStore.options + : {}; + return { storeBlock: { backend: localBackend, options: prevOptions } }; } - const backend = backendAnswer as 'blazegraph' | 'sparql-http'; + + if (backendPolicy.kind !== 'external') { + throw unknownBackendError('prompt', backendAnswer); + } + const backend = backendAnswer as ExternalStoreBackend; // URL prompt loop: validate each attempt, surface the operator-facing // failure message, allow retry or abort. @@ -317,7 +397,7 @@ export async function promptStoreBackend( } const retry = (await opts.ask('Retry with a URL? (y/n)', 'y')).toLowerCase(); if (retry === 'n') { - log(' Aborting store setup; defaulting to local Oxigraph.'); + log(' Aborting store setup; using the oxigraph-server default.'); return { storeBlock: null }; } continue; @@ -352,7 +432,7 @@ export async function promptStoreBackend( 'y', )).toLowerCase(); if (retry === 'n') { - log(' Aborting store setup; defaulting to local Oxigraph.'); + log(' Aborting store setup; using the oxigraph-server default.'); return { storeBlock: null }; } } @@ -391,33 +471,49 @@ export async function applyStoreFlagsToConfig( opts: ApplyStoreFlagsOptions, ): Promise { const log = opts.log ?? console.log; - const backend = opts.storeFlag; + const backend = opts.storeFlag?.trim().toLowerCase(); if (!backend) return; const load = opts.loadConfig ?? loadConfig; const save = opts.saveConfig ?? saveConfig; - // Operators who pass `--store oxigraph` may be trying to FORCE local - // even though their existing config has a `store` block — honour - // that by clearing the block. - if ( - backend === 'oxigraph' || - backend === 'oxigraph-worker' || - backend === 'oxigraph-persistent' - ) { + if (backend === UNSUPPORTED_WORKER_BACKEND) { + throw unsupportedWorkerBackendError('--store'); + } + if (!isSupportedExistingBackend(backend)) { + throw unknownBackendError('--store', backend); + } + const backendPolicy = STORE_BACKENDS[backend]; + + // Operators who pass `--store oxigraph` are explicitly opting into the + // embedded development store. Persist it because an omitted store block now + // means the managed `oxigraph-server` default. + if (backendPolicy.kind === 'local') { + const localBackend = backend as LocalStoreBlock['backend']; const existing = await load(); - if (existing.store) { - log(` Removing existing store block (--store ${backend} → local default).`); - const next = { ...existing }; - delete next.store; - await save(next); + if (STORE_BACKENDS[localBackend].requiresExistingPath) { + if (existing.store?.backend === localBackend && hasStorePath(existing.store)) { + await save({ ...existing, store: { backend: localBackend, options: existing.store.options } }); + log(` Store configured: ${localBackend}.`); + return; + } + throw new Error( + `--store ${localBackend} requires an existing store.options.path; ` + + 'set it manually in config.json or use --store oxigraph-server.', + ); + } + await save({ ...existing, store: { backend: localBackend, options: {} } }); + if (localBackend === 'oxigraph') { + log(' Store configured: oxigraph (embedded development store).'); + } else { + log(` Store configured: ${localBackend}.`); } return; } // Daemon-managed local Oxigraph server: no URL to validate (the daemon // brings it up at boot). Write the block and return. - if (backend === 'oxigraph-server') { + if (backendPolicy.kind === 'managed-local') { const existing = await load(); // Preserve any existing managed-server overrides (port/location/cacheDir) // that planManagedOxigraph reads at boot — re-running setup with @@ -431,21 +527,20 @@ export async function applyStoreFlagsToConfig( return; } - if (backend !== 'blazegraph' && backend !== 'sparql-http') { - throw new Error( - `--store must be one of: oxigraph, blazegraph, sparql-http, oxigraph-server (got "${backend}")`, - ); + if (backendPolicy.kind !== 'external') { + throw unknownBackendError('--store', backend); } + const externalBackend = backend as ExternalStoreBackend; const url = opts.storeUrlFlag?.trim(); if (!url) { - throw new Error(`--store ${backend} requires --store-url `); + throw new Error(`--store ${externalBackend} requires --store-url `); } const optionsForProbe = - backend === 'blazegraph' ? { url } : { queryEndpoint: url }; + externalBackend === 'blazegraph' ? { url } : { queryEndpoint: url }; const health = await checkExternalStoreReachable({ - storeConfig: { backend, options: optionsForProbe }, + storeConfig: { backend: externalBackend, options: optionsForProbe }, fetch: opts.fetch, }); if (!health.ok) { @@ -455,8 +550,8 @@ export async function applyStoreFlagsToConfig( const existing = await load(); const next: DkgConfig = { ...existing, - store: externalStoreBlock(backend, url, false), + store: externalStoreBlock(externalBackend, url, false), }; await save(next); - log(` Store configured: ${backend} (${url}) — verified reachable.`); + log(` Store configured: ${externalBackend} (${url}) — verified reachable.`); } diff --git a/packages/cli/test/chain-reset-wipe.test.ts b/packages/cli/test/chain-reset-wipe.test.ts index 657d2ad79f..c9e1564043 100644 --- a/packages/cli/test/chain-reset-wipe.test.ts +++ b/packages/cli/test/chain-reset-wipe.test.ts @@ -521,7 +521,7 @@ describe('chainResetWipe — external SPARQL wipe', () => { expect(result.failedFiles.some((f) => f.error.includes('ECONNREFUSED'))).toBe(true); }); - it('skips external wipe entirely for local backends (storeConfig present, backend oxigraph-worker)', async () => { + it('skips external wipe entirely for local backends (storeConfig present, backend oxigraph-persistent)', async () => { writeFileSync(join(dataDir, 'store.nq'), '

.'); const { fn, calls } = mockFetch(() => new Response(null, { status: 200 })); @@ -529,7 +529,7 @@ describe('chainResetWipe — external SPARQL wipe', () => { dataDir, currentMarker: NEW_MARKER, storeConfig: { - backend: 'oxigraph-worker', + backend: 'oxigraph-persistent', // Options that LOOK like an external URL: these must be ignored // because the backend itself is local. Otherwise an operator who // hand-tuned options would see surprise SPARQL requests. @@ -584,7 +584,7 @@ describe('detectBackendSwitch', () => { const logs: string[] = []; const result = detectBackendSwitch({ dataDir, - currentBackend: 'oxigraph-worker', + currentBackend: 'oxigraph-persistent', acceptStoreReset: false, log: (m) => logs.push(m), }); @@ -594,7 +594,7 @@ describe('detectBackendSwitch', () => { expect(result.aborted).toBe(false); const persisted = JSON.parse(readFileSync(join(dataDir, STATE_FILE), 'utf8')); - expect(persisted.lastBackend).toBe('oxigraph-worker'); + expect(persisted.lastBackend).toBe('oxigraph-persistent'); // First-boot path is silent — no STORE-SWITCH warning header. expect(logs.find((l) => l.includes('STORE-SWITCH'))).toBeUndefined(); }); @@ -628,19 +628,19 @@ describe('detectBackendSwitch', () => { it('is a no-op when backend matches previous boot', () => { writeFileSync( join(dataDir, STATE_FILE), - JSON.stringify({ chainResetMarker: null, lastBackend: 'oxigraph-worker', savedAt: Date.now() }), + JSON.stringify({ chainResetMarker: null, lastBackend: 'oxigraph-persistent', savedAt: Date.now() }), ); const logs: string[] = []; const result = detectBackendSwitch({ dataDir, - currentBackend: 'oxigraph-worker', + currentBackend: 'oxigraph-persistent', acceptStoreReset: false, log: (m) => logs.push(m), }); expect(result.changed).toBe(false); - expect(result.previous).toBe('oxigraph-worker'); + expect(result.previous).toBe('oxigraph-persistent'); expect(result.aborted).toBe(false); expect(logs).toEqual([]); }); @@ -648,7 +648,7 @@ describe('detectBackendSwitch', () => { it('aborts boot on mismatch without acceptStoreReset, surfaces multi-line warning', () => { writeFileSync( join(dataDir, STATE_FILE), - JSON.stringify({ chainResetMarker: null, lastBackend: 'oxigraph-worker', savedAt: Date.now() }), + JSON.stringify({ chainResetMarker: null, lastBackend: 'oxigraph-persistent', savedAt: Date.now() }), ); const logs: string[] = []; @@ -660,25 +660,25 @@ describe('detectBackendSwitch', () => { }); expect(result.changed).toBe(true); - expect(result.previous).toBe('oxigraph-worker'); + expect(result.previous).toBe('oxigraph-persistent'); expect(result.aborted).toBe(true); const joined = logs.join('\n'); expect(joined).toMatch(/STORE-SWITCH/); - expect(joined).toMatch(/previous: oxigraph-worker/); + expect(joined).toMatch(/previous: oxigraph-persistent/); expect(joined).toMatch(/current:\s+blazegraph/); expect(joined).toMatch(/DKG_ACCEPT_STORE_RESET=1/); // State file MUST still record the old backend so a corrected // config (operator reverts the edit) sees a match on next boot. const persisted = JSON.parse(readFileSync(join(dataDir, STATE_FILE), 'utf8')); - expect(persisted.lastBackend).toBe('oxigraph-worker'); + expect(persisted.lastBackend).toBe('oxigraph-persistent'); }); it('proceeds and updates state when acceptStoreReset=true', () => { writeFileSync( join(dataDir, STATE_FILE), - JSON.stringify({ chainResetMarker: null, lastBackend: 'oxigraph-worker', savedAt: Date.now() }), + JSON.stringify({ chainResetMarker: null, lastBackend: 'oxigraph-persistent', savedAt: Date.now() }), ); const logs: string[] = []; @@ -707,12 +707,12 @@ describe('detectBackendSwitch', () => { // Establish a baseline by running detectBackendSwitch first. detectBackendSwitch({ dataDir, - currentBackend: 'oxigraph-worker', + currentBackend: 'oxigraph-persistent', acceptStoreReset: false, }); expect( JSON.parse(readFileSync(join(dataDir, STATE_FILE), 'utf8')).lastBackend, - ).toBe('oxigraph-worker'); + ).toBe('oxigraph-persistent'); // Now run a marker change — the wipe path persists chainResetMarker // and MUST preserve lastBackend. @@ -724,6 +724,6 @@ describe('detectBackendSwitch', () => { const persisted = JSON.parse(readFileSync(join(dataDir, STATE_FILE), 'utf8')); expect(persisted.chainResetMarker).toBe(NEW_MARKER); - expect(persisted.lastBackend).toBe('oxigraph-worker'); + expect(persisted.lastBackend).toBe('oxigraph-persistent'); }); }); diff --git a/packages/cli/test/daemon-http-behavior-extra.test.ts b/packages/cli/test/daemon-http-behavior-extra.test.ts index 76217485dd..75ba9e4469 100644 --- a/packages/cli/test/daemon-http-behavior-extra.test.ts +++ b/packages/cli/test/daemon-http-behavior-extra.test.ts @@ -101,7 +101,7 @@ async function writeDaemonConfig( relay: 'none', auth: { enabled: authEnabled }, store: { - backend: 'oxigraph-worker', + backend: 'oxigraph-persistent', options: { path: join(home, 'store.nq') }, }, // Real EVM adapter against the shared Hardhat node (port 9548 per diff --git a/packages/cli/test/daemon-startup-validation.test.ts b/packages/cli/test/daemon-startup-validation.test.ts index d7da0d4564..b18f639bb1 100644 --- a/packages/cli/test/daemon-startup-validation.test.ts +++ b/packages/cli/test/daemon-startup-validation.test.ts @@ -1,5 +1,5 @@ -import { afterEach, describe, expect, it, vi } from 'vitest'; -import { mkdtemp, rm } from 'node:fs/promises'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { mkdtemp, rm, writeFile } from 'node:fs/promises'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { computeNetworkId } from '../../core/src/genesis.js'; @@ -8,6 +8,7 @@ const mocks = vi.hoisted(() => ({ agentCreate: vi.fn(), loadOpWallets: vi.fn(), loadNetworkConfig: vi.fn(), + startManagedOxigraph: vi.fn(), })); vi.mock('@origintrail-official/dkg-agent', async importOriginal => { @@ -28,6 +29,14 @@ vi.mock('../src/config.js', async importOriginal => { }; }); +vi.mock('../src/daemon/oxigraph-managed.js', async importOriginal => { + const actual = await importOriginal(); + return { + ...actual, + startManagedOxigraph: mocks.startManagedOxigraph, + }; +}); + const { runDaemonInner } = await import('../src/daemon/lifecycle.js'); describe('daemon startup network validation', () => { @@ -37,6 +46,13 @@ describe('daemon startup network validation', () => { let stderrWrite: typeof process.stderr.write = process.stderr.write; let uncaughtExceptionListeners: NodeJS.UncaughtExceptionListener[] = []; let unhandledRejectionListeners: NodeJS.UnhandledRejectionListener[] = []; + const originalAcceptStoreReset = process.env.DKG_ACCEPT_STORE_RESET; + + beforeEach(() => { + mocks.loadNetworkConfig.mockResolvedValue(undefined); + mocks.loadOpWallets.mockResolvedValue({ adminWallet: undefined, wallets: [] }); + mocks.startManagedOxigraph.mockResolvedValue(null); + }); afterEach(async () => { vi.restoreAllMocks(); @@ -56,10 +72,75 @@ describe('daemon startup network validation', () => { } else { process.env.DKG_HOME = originalDkgHome; } + if (originalAcceptStoreReset === undefined) { + delete process.env.DKG_ACCEPT_STORE_RESET; + } else { + process.env.DKG_ACCEPT_STORE_RESET = originalAcceptStoreReset; + } if (tempHome) await rm(tempHome, { recursive: true, force: true }); tempHome = undefined; }); + async function useTempHome(prefix: string) { + tempHome = await mkdtemp(join(tmpdir(), prefix)); + originalDkgHome = process.env.DKG_HOME; + process.env.DKG_HOME = tempHome; + stdoutWrite = process.stdout.write; + stderrWrite = process.stderr.write; + uncaughtExceptionListeners = process.listeners('uncaughtException') as NodeJS.UncaughtExceptionListener[]; + unhandledRejectionListeners = process.listeners('unhandledRejection') as NodeJS.UnhandledRejectionListener[]; + return vi.spyOn(process.stdout, 'write').mockImplementation(() => true); + } + + it('exits before managed store startup when a blockless config has legacy store.nq and no reset acknowledgement', async () => { + const stdoutSpy = await useTempHome('dkg-legacy-store-gate-'); + await writeFile(join(tempHome!, 'store.nq'), '

.'); + vi + .spyOn(process, 'exit') + .mockImplementation(((code?: string | number | null) => { + throw new Error(`process.exit:${code}`); + }) as never); + + await expect(runDaemonInner(true, { + name: 'legacy-store-gate-test', + listenPort: 0, + nodeRole: 'edge', + } as any, Date.now())).rejects.toThrow('process.exit:1'); + + const output = stdoutSpy.mock.calls.map(call => String(call[0])).join(''); + expect(output).toContain('legacy store.nq from the old implicit worker default'); + expect(output).toContain('DKG_ACCEPT_STORE_RESET=1'); + expect(mocks.startManagedOxigraph).not.toHaveBeenCalled(); + expect(mocks.agentCreate).not.toHaveBeenCalled(); + }); + + it('continues with the effective oxigraph-server store after legacy store.nq is acknowledged', async () => { + const stdoutSpy = await useTempHome('dkg-legacy-store-ack-'); + process.env.DKG_ACCEPT_STORE_RESET = '1'; + await writeFile(join(tempHome!, 'store.nq'), '

.'); + mocks.agentCreate.mockRejectedValue(new Error('after-agent-create')); + + await expect(runDaemonInner(true, { + name: 'legacy-store-ack-test', + listenPort: 0, + nodeRole: 'edge', + } as any, Date.now())).rejects.toThrow('after-agent-create'); + + const output = stdoutSpy.mock.calls.map(call => String(call[0])).join(''); + expect(output).toContain('using oxigraph-server'); + expect(mocks.startManagedOxigraph).toHaveBeenCalledTimes(1); + expect(mocks.startManagedOxigraph.mock.calls[0]?.[0]).toMatchObject({ + dataDir: tempHome, + config: { + store: { backend: 'oxigraph-server', options: {} }, + }, + }); + expect(mocks.agentCreate).toHaveBeenCalledTimes(1); + expect(mocks.agentCreate.mock.calls[0]?.[0]).toMatchObject({ + storeConfig: { backend: 'oxigraph-server', options: {} }, + }); + }); + it('exits before agent creation when the selected network is pre-deployment', async () => { tempHome = await mkdtemp(join(tmpdir(), 'dkg-predeployment-startup-')); originalDkgHome = process.env.DKG_HOME; diff --git a/packages/cli/test/daemon/plugin-routes-api.e2e.test.ts b/packages/cli/test/daemon/plugin-routes-api.e2e.test.ts index c1617aa888..3204585822 100644 --- a/packages/cli/test/daemon/plugin-routes-api.e2e.test.ts +++ b/packages/cli/test/daemon/plugin-routes-api.e2e.test.ts @@ -62,7 +62,7 @@ async function writeDaemonConfig( relay: 'none', auth: { enabled: true }, store: { - backend: 'oxigraph-worker', + backend: 'oxigraph-persistent', options: { path: join(home, 'store.nq') }, }, chain: { diff --git a/packages/cli/test/helpers/live-daemon.ts b/packages/cli/test/helpers/live-daemon.ts index b2ddf0846e..94362f5913 100644 --- a/packages/cli/test/helpers/live-daemon.ts +++ b/packages/cli/test/helpers/live-daemon.ts @@ -87,7 +87,7 @@ export async function startLiveDaemon(opts: StartDaemonOpts = {}): Promise { it('returns null for non-oxigraph-server backends', () => { - expect(planManagedOxigraph({ store: { backend: 'oxigraph-worker' } }, '/data')).toBeNull(); + expect(planManagedOxigraph({ store: { backend: 'oxigraph' } }, '/data')).toBeNull(); expect(planManagedOxigraph({ store: { backend: 'sparql-http' } }, '/data')).toBeNull(); expect(planManagedOxigraph({}, '/data')).toBeNull(); }); @@ -182,7 +182,7 @@ describe('startManagedOxigraph (real download + real server)', () => { // Contract: a non-managed backend is a no-op — null result, and nothing // observable happens (no cache dir is created, nothing binds a port). const result = await startManagedOxigraph({ - config: { store: { backend: 'oxigraph-worker' } }, + config: { store: { backend: 'oxigraph' } }, dataDir: '/data', }); expect(result).toBeNull(); diff --git a/packages/cli/test/publisher-cli-smoke.test.ts b/packages/cli/test/publisher-cli-smoke.test.ts index f537a5ecd6..fe35a526ad 100644 --- a/packages/cli/test/publisher-cli-smoke.test.ts +++ b/packages/cli/test/publisher-cli-smoke.test.ts @@ -46,7 +46,7 @@ describe.sequential.skip('publisher CLI smoke', () => { contextGraphs: [], auth: { enabled: false }, store: { - backend: 'oxigraph-worker', + backend: 'oxigraph-persistent', options: { path: join(dkgHome, 'store.nq') }, }, }), diff --git a/packages/cli/test/store-health-check.test.ts b/packages/cli/test/store-health-check.test.ts index b01d80ba5a..49797a2ea8 100644 --- a/packages/cli/test/store-health-check.test.ts +++ b/packages/cli/test/store-health-check.test.ts @@ -37,7 +37,7 @@ describe('checkExternalStoreReachable', () => { it('passes through with no I/O for local backends', async () => { const { fn, calls } = mockFetch(() => new Response(null, { status: 200 })); const result = await checkExternalStoreReachable({ - storeConfig: { backend: 'oxigraph-worker' }, + storeConfig: { backend: 'oxigraph' }, fetch: fn, }); expect(result.ok).toBe(true); diff --git a/packages/cli/test/store-identity-tag.test.ts b/packages/cli/test/store-identity-tag.test.ts index fd8b19d354..f438be62d2 100644 --- a/packages/cli/test/store-identity-tag.test.ts +++ b/packages/cli/test/store-identity-tag.test.ts @@ -43,7 +43,7 @@ describe('checkOrSetStoreIdentity', () => { it('skips for local oxigraph backend', async () => { const { fn, calls } = mockFetch(() => new Response(null, { status: 200 })); const result = await checkOrSetStoreIdentity({ - storeConfig: { backend: 'oxigraph-worker' }, + storeConfig: { backend: 'oxigraph' }, nodeName: 'mynode', fetch: fn, }); diff --git a/packages/cli/test/store-wizard.test.ts b/packages/cli/test/store-wizard.test.ts index 02a2ac794a..6a3b4c1956 100644 --- a/packages/cli/test/store-wizard.test.ts +++ b/packages/cli/test/store-wizard.test.ts @@ -10,7 +10,7 @@ * - Blazegraph + valid URL: store block persisted with * `managedByDkg: false`. * - Blazegraph + unreachable URL: surfaces formatted failure, allows - * retry, abort returns to local default. + * retry, abort leaves the managed local default in place. * - Blazegraph + 404 URL: namespace-missing branch fires; message * mentions namespace, not network. * - Blank URL prompt: PR 2's "no Docker yet" message + retry. @@ -65,31 +65,26 @@ describe('promptStoreBackend', () => { expect(calls).toHaveLength(0); // no URL probe issued for a local backend }); - it('returns no store block (embedded worker) when operator picks "oxigraph" by name', async () => { + it('persists an explicit embedded development store when operator picks "oxigraph" by name', async () => { const result = await promptStoreBackend({ ask: mockAsk(['oxigraph']), log: () => {}, }); - expect(result.storeBlock).toBeNull(); + expect(result.storeBlock).toEqual({ backend: 'oxigraph', options: {} }); }); - it('returns no store block (embedded worker) when operator picks the worker by number', async () => { + it('persists an explicit embedded development store when operator picks oxigraph by number', async () => { // Menu is now `1) oxigraph-server 2) oxigraph 3) blazegraph` — picking - // option 2 must opt down to the embedded in-process worker (no block). + // option 2 opts into the embedded in-memory store explicitly. const result = await promptStoreBackend({ ask: mockAsk(['2']), log: () => {}, }); - expect(result.storeBlock).toBeNull(); + expect(result.storeBlock).toEqual({ backend: 'oxigraph', options: {} }); }); - it('preserves an explicit embedded backend verbatim on Enter-through (no flip, no option loss)', async () => { - // Codex #946 — only a *block-less* config should fall through to the new - // oxigraph-server default. A node that explicitly chose a local worker - // variant must keep it on a re-init Enter-through, AND keep its custom - // `options` (e.g. the worker's `options.path`): returning `null` would let - // `dkg init` write `store: undefined` and relocate the store on next boot. - for (const backend of ['oxigraph', 'oxigraph-worker', 'oxigraph-persistent'] as const) { + it('preserves a supported explicit embedded backend verbatim on Enter-through (no flip, no option loss)', async () => { + for (const backend of ['oxigraph', 'oxigraph-persistent'] as const) { const existingStore = { backend, options: { path: '/custom/store' } }; const result = await promptStoreBackend({ ask: mockAsk(['']), // Enter @@ -100,11 +95,7 @@ describe('promptStoreBackend', () => { } }); - it('switches to the default embedded worker when an oxigraph-persistent node EXPLICITLY picks oxigraph', async () => { - // Codex #946 — preservation must be gated on a true keep. An operator who - // explicitly selects option `2` / "oxigraph" to move a worker/persistent - // node back to the plain embedded default must NOT have the old backend + - // options silently retained. Both the numeric and named selection switch. + it('does not preserve oxigraph-persistent options when the operator explicitly picks oxigraph', async () => { const existingStore = { backend: 'oxigraph-persistent', options: { path: '/custom/store' } }; for (const answer of ['2', 'oxigraph']) { const result = await promptStoreBackend({ @@ -112,13 +103,33 @@ describe('promptStoreBackend', () => { existingStore, log: () => {}, }); - expect(result.storeBlock).toBeNull(); + expect(result.storeBlock).toEqual({ backend: 'oxigraph', options: {} }); } }); + it('does not preserve retired oxigraph-worker configs on Enter-through', async () => { + const logs: string[] = []; + const result = await promptStoreBackend({ + ask: mockAsk(['']), + existingStore: { backend: 'oxigraph-worker', options: { path: '/custom/store' } }, + log: (m) => logs.push(m), + }); + expect(result.storeBlock).toEqual({ backend: 'oxigraph-server', options: {} }); + expect(logs.join('\n')).toMatch(/retired/); + }); + + it('rejects oxigraph-worker when typed explicitly', async () => { + await expect( + promptStoreBackend({ + ask: mockAsk(['oxigraph-worker']), + log: () => {}, + }), + ).rejects.toThrow(/no longer supported/); + }); + it('falls back to the recommended default (oxigraph-server) on an out-of-range number', async () => { // Codex #946 — a typo'd digit ("9") must not silently downgrade a fresh - // install to the embedded worker; it resolves to defaultBackend (option 1). + // install to an embedded backend; it resolves to defaultBackend (option 1). const result = await promptStoreBackend({ ask: mockAsk(['9']), log: () => {}, @@ -172,7 +183,7 @@ describe('promptStoreBackend', () => { expect(logs.some((l) => l.includes('STORE-HEALTH'))).toBe(true); }); - it('aborts to local default when operator declines retry on unreachable URL', async () => { + it('aborts to the managed local default when operator declines retry on unreachable URL', async () => { const { fn } = mockFetch(() => new Response('boom', { status: 500 })); const logs: string[] = []; const result = await promptStoreBackend({ @@ -283,7 +294,7 @@ describe('promptStoreBackend', () => { 'blazegraph', '', // blank URL 'n', // decline Docker - 'n', // decline retry-with-URL → abort to local default + 'n', // decline retry-with-URL → abort to managed local default ]), isDockerAvailable: async () => true, provisionBlazegraphDocker: async () => { @@ -616,7 +627,7 @@ describe('applyStoreFlagsToConfig', () => { storeFlag: 'neptune', log: () => {}, }), - ).rejects.toThrow(/oxigraph, blazegraph, sparql-http/); + ).rejects.toThrow(/oxigraph-server, oxigraph, oxigraph-persistent, blazegraph, sparql-http/); }); it('persists a daemon-managed oxigraph-server block (no URL required)', async () => { @@ -654,7 +665,7 @@ describe('applyStoreFlagsToConfig', () => { expect(store.saved[0].store).toEqual({ backend: 'oxigraph-server', options: {} }); }); - it('clears existing store block when --store oxigraph is passed', async () => { + it('persists an explicit oxigraph block when --store oxigraph is passed', async () => { const store = newMockConfig({ ...baseConfig, store: { @@ -669,10 +680,10 @@ describe('applyStoreFlagsToConfig', () => { log: () => {}, }); expect(store.saved).toHaveLength(1); - expect(store.saved[0].store).toBeUndefined(); + expect(store.saved[0].store).toEqual({ backend: 'oxigraph', options: {} }); }); - it('is a no-op when --store oxigraph is passed and no existing store block', async () => { + it('persists oxigraph when --store oxigraph is passed and no existing store block', async () => { const store = newMockConfig(baseConfig); const io = mockConfigIO(store); await applyStoreFlagsToConfig({ @@ -680,6 +691,20 @@ describe('applyStoreFlagsToConfig', () => { storeFlag: 'oxigraph', log: () => {}, }); + expect(store.saved).toHaveLength(1); + expect(store.saved[0].store).toEqual({ backend: 'oxigraph', options: {} }); + }); + + it('rejects --store oxigraph-worker', async () => { + const store = newMockConfig(baseConfig); + const io = mockConfigIO(store); + await expect( + applyStoreFlagsToConfig({ + ...io, + storeFlag: 'oxigraph-worker', + log: () => {}, + }), + ).rejects.toThrow(/no longer supported/); expect(store.saved).toEqual([]); }); diff --git a/packages/cli/test/validate-store-config.test.ts b/packages/cli/test/validate-store-config.test.ts index 39ff79e24c..fefd180a61 100644 --- a/packages/cli/test/validate-store-config.test.ts +++ b/packages/cli/test/validate-store-config.test.ts @@ -8,8 +8,8 @@ * when paired with an external backend (no local store path to * infer from). * - * Local backends (default Oxigraph) are unaffected — the function is a - * no-op for them. + * Supported local backends are unaffected. The retired `oxigraph-worker` + * backend is rejected before boot. * * Plan: `.cursor/plans/blazegraph_v10_support_178da670.plan.md` §PR 1 item 6. */ @@ -33,9 +33,12 @@ describe('validateStoreConfig', () => { }); describe('local backends', () => { - it('no-op for oxigraph-worker', () => { + it('no-op for supported local backends', () => { expect( - validateStoreConfig(mk({ store: { backend: 'oxigraph-worker' } })), + validateStoreConfig(mk({ store: { backend: 'oxigraph' } })), + ).toEqual([]); + expect( + validateStoreConfig(mk({ store: { backend: 'oxigraph-persistent', options: { path: '/tmp/store.nq' } } })), ).toEqual([]); }); @@ -44,10 +47,19 @@ describe('validateStoreConfig', () => { // if the backend is local; the wipe + health check honour // isExternalBackend the same way. const errors = validateStoreConfig( - mk({ store: { backend: 'oxigraph-worker', options: { url: 'irrelevant' } } }), + mk({ store: { backend: 'oxigraph', options: { url: 'irrelevant' } } }), ); expect(errors).toEqual([]); }); + + it('rejects the retired oxigraph-worker backend', () => { + const errors = validateStoreConfig( + mk({ store: { backend: 'oxigraph-worker' } }), + ); + expect(errors).toHaveLength(1); + expect(errors[0].field).toBe('store.backend'); + expect(errors[0].message).toMatch(/no longer supported/); + }); }); describe('blazegraph', () => { @@ -136,7 +148,7 @@ describe('validateStoreConfig', () => { it('does not enforce the directory requirement for local backends', () => { const errors = validateStoreConfig( mk({ - store: { backend: 'oxigraph-worker' }, + store: { backend: 'oxigraph-persistent', options: { path: '/tmp/store.nq' } }, largeLiteralStorage: { enabled: true }, }), ); diff --git a/packages/core/test/ensure-dkg-node-config.test.ts b/packages/core/test/ensure-dkg-node-config.test.ts index 25bb061e55..0799d9a622 100644 --- a/packages/core/test/ensure-dkg-node-config.test.ts +++ b/packages/core/test/ensure-dkg-node-config.test.ts @@ -35,7 +35,7 @@ describe('ensureDkgNodeConfig — store-backend default (issue #960)', () => { it('does NOT flip an existing (block-less) node onto a new backend', () => { // Simulate an existing node: a config.json is already on disk (it had been - // running on the oxigraph-worker runtime fallback). Re-running setup must + // running without an explicit store block). Re-running setup must // not silently switch its backend (which would force a store reset). writeFileSync(join(tempHome, 'config.json'), JSON.stringify({ name: 'node-a', nodeRole: 'edge' }) + '\n'); ensureDkgNodeConfig({ diff --git a/packages/kafka-plugin/test/kafka-plugin-api.e2e.test.ts b/packages/kafka-plugin/test/kafka-plugin-api.e2e.test.ts index dd08785081..3abebdcd3f 100644 --- a/packages/kafka-plugin/test/kafka-plugin-api.e2e.test.ts +++ b/packages/kafka-plugin/test/kafka-plugin-api.e2e.test.ts @@ -124,7 +124,7 @@ async function writeDaemonConfig( relay: 'none', auth: { enabled: true }, store: { - backend: 'oxigraph-worker', + backend: 'oxigraph-persistent', options: { path: join(home, 'store.nq') }, }, chain: { diff --git a/packages/storage/README.md b/packages/storage/README.md index 698e3c2a05..08c2263bf4 100644 --- a/packages/storage/README.md +++ b/packages/storage/README.md @@ -6,7 +6,6 @@ Triple store abstraction layer for DKG V10. Provides a unified API over multiple - **Backend adapters** — pluggable triple store implementations: - `OxigraphStore` — embedded WASM/native store, no external dependencies - - `OxigraphWorkerStore` — worker-thread variant; keeps the daemon event loop free, with a per-read-operation timeout (see below) - `BlazegraphStore` — connects to a running Blazegraph SPARQL endpoint - `SparqlHttpStore` — generic adapter for any SPARQL 1.1 compliant endpoint - **Graph manager** — named graph lifecycle (create, drop, list) with contextGraph-scoped data and metadata graphs @@ -33,39 +32,6 @@ await store.insert(quads); const result = await store.query('SELECT * WHERE { ?s ?p ?o } LIMIT 10'); ``` -## Embedded worker store (`oxigraph-worker`) tuning - -The embedded worker runs **all** store operations on a single worker thread, so -a long-running or stuck op (a huge import, an expensive query) blocks every -other store-backed request behind it. Under real load this surfaces as the -daemon's `/api/status` staying green while `/api/query`, -`/api/context-graph/list`, and `/api/assertion/create` hang. A `store.options` -knob bounds that blast radius: - -| Option | Default | Purpose | -|---|---|---| -| `operationTimeoutMs` | `120000` | Reject a **read-only** op (`query`, `hasGraph`, `listGraphs`, `countQuads`) that exceeds this instead of hanging forever — that's where the user-visible hang shows up. `0` disables (restores unbounded behaviour). `close` is exempt — its final flush always runs to completion so shutdown can't drop pending writes. | - -Mutations (`insert`, `delete`, …) are intentionally **not** bounded by this -timeout. The bound only drops the *caller's* promise — the single worker thread -keeps running the op — so a "timed-out" write could still commit afterwards, -and the rest of the codebase treats a rejected `insert`/`delete` as a clean -failure. Bounding only reads surfaces a wedged worker on the paths that hang -without inventing an indeterminate write outcome. `insert()` therefore stays -strictly atomic (all quads commit or the call fails), which callers rely on. - -```jsonc -// ~/.dkg/config.json -"store": { - "backend": "oxigraph-worker", - "options": { "operationTimeoutMs": 120000 } -} -``` - -For heavy / production workloads, prefer an out-of-process SPARQL server -(`sparql-http` or `blazegraph`), which handles reads and writes concurrently -and keeps the daemon responsive under load. - ## Internal Dependencies - `@origintrail-official/dkg-core` — configuration types, logging, constants diff --git a/packages/storage/src/adapters/oxigraph-worker-impl.ts b/packages/storage/src/adapters/oxigraph-worker-impl.ts deleted file mode 100644 index 6bfce261ce..0000000000 --- a/packages/storage/src/adapters/oxigraph-worker-impl.ts +++ /dev/null @@ -1,18 +0,0 @@ -import { parentPort, workerData } from 'node:worker_threads'; -import { OxigraphStore } from './oxigraph.js'; - -const store = new OxigraphStore(workerData?.persistPath); - -parentPort!.on('message', async (msg: { id: number; method: string; args: unknown[] }) => { - try { - const fn = (store as any)[msg.method]; - if (typeof fn !== 'function') { - parentPort!.postMessage({ id: msg.id, error: `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) }); - } -}); diff --git a/packages/storage/src/adapters/oxigraph-worker.ts b/packages/storage/src/adapters/oxigraph-worker.ts deleted file mode 100644 index e2e7a6ca0f..0000000000 --- a/packages/storage/src/adapters/oxigraph-worker.ts +++ /dev/null @@ -1,311 +0,0 @@ -import { Worker } from 'node:worker_threads'; -import { existsSync } from 'node:fs'; -import { sep } from 'node:path'; -import { fileURLToPath } from 'node:url'; -import type { TripleStore, Quad, TripleStoreQueryOptions, QueryResult } from '../triple-store.js'; -import { registerTripleStoreAdapter } from '../triple-store.js'; - -/** - * Default per-operation timeout for the embedded worker store. The worker is - * a SINGLE thread that processes store ops FIFO, so one slow / wedged op (a - * huge import, an expensive query, or a genuinely hung worker) blocks every - * other store-backed request queued behind it. Without a bound, the caller — - * an API route, the publisher, gossip ingest — waits FOREVER. That is the - * exact signature behind issues #997 / #999 / #1002 / #1005 / #1008: - * `/api/status` (no store) stays green while `/api/query`, - * `/api/context-graph/list`, `/api/assertion/create` never return. - * - * A bounded wait turns an indefinite hang into a surfaced error the operator - * can act on (the message points at the real fix: use an external SPARQL - * server for heavy workloads). 120s is generous enough not to trip normal - * operations yet finite. Set `operationTimeoutMs: 0` to restore the old - * unbounded behaviour. - * - * IMPORTANT — the timeout is applied to READ-ONLY ops only (see - * READ_ONLY_METHODS). A read is side-effect-free, so bounding the caller's wait - * and rejecting is always a clean, determinate failure. A mutation is NOT - * bounded: the timeout only drops the caller's promise while the single worker - * thread keeps running the op, so a "timed-out" insert/delete could STILL - * commit afterwards — and the rest of the codebase treats a rejected - * insert/delete as a clean failure, which would leave partial state visible and - * retries ambiguous. The user-visible hang in the issues above is on the read - * paths (`/api/query`, `/api/context-graph/list`); bounding reads surfaces a - * wedged worker there without inventing an indeterminate mutation outcome. - */ -const DEFAULT_OPERATION_TIMEOUT_MS = 120_000; - -export interface OxigraphWorkerStoreOptions { - /** - * Per-operation timeout in milliseconds for READ-ONLY ops. Default 120_000. - * 0 disables it. Mutations are intentionally never bounded (see - * DEFAULT_OPERATION_TIMEOUT_MS) so a timed-out write can't be reported as a - * clean failure while it is still in flight. - */ - operationTimeoutMs?: number; -} - -/** - * Accept only a finite, non-negative override; otherwise fall back. The result - * is floored to an INTEGER — the timeout is a millisecond count, so a fractional - * value is meaningless noise. - */ -function normalizeNonNegativeInt(value: number | undefined, fallback: number): number { - return typeof value === 'number' && Number.isFinite(value) && value >= 0 - ? Math.floor(value) - : fallback; -} - -function asAbortError(reason: unknown): Error { - return reason instanceof Error ? reason : new Error(String(reason ?? 'aborted')); -} - -/** - * Side-effect-free read methods. ONLY these are bounded by the per-op timeout: - * rejecting a read after the bound is a clean, determinate failure (nothing was - * written), so it safely surfaces a wedged worker on the exact paths that hang - * in production. Every other method mutates persisted state and is left - * unbounded — see DEFAULT_OPERATION_TIMEOUT_MS for why timing out a mutation - * would be unsafe with the current call sites. - */ -const READ_ONLY_METHODS = new Set([ - 'query', 'hasGraph', 'listGraphs', 'countQuads', -]); - -/** Rejection raised when a read op exceeds its per-op timeout. */ -export interface OxigraphWorkerTimeoutError extends Error { - code: 'OXIGRAPH_WORKER_OP_TIMEOUT'; - /** Which store method timed out. */ - method: string; - /** The bound that was exceeded. */ - timeoutMs: number; -} - -export class OxigraphWorkerStore implements TripleStore { - readonly queryCancellation = 'interruptible' as const; - - private worker: Worker; - private nextId = 0; - private pending = new Map void; reject: (e: Error) => void }>(); - private readonly operationTimeoutMs: number; - /** Set once the worker thread has exited (graceful close, crash, or kill). */ - private workerExited = false; - /** Memoized close so repeat/concurrent close() calls share one teardown. */ - private closePromise: Promise | null = null; - - constructor(persistPath?: string, opts?: OxigraphWorkerStoreOptions) { - this.operationTimeoutMs = normalizeNonNegativeInt(opts?.operationTimeoutMs, DEFAULT_OPERATION_TIMEOUT_MS); - - // Resolve the worker impl with a small search path so this keeps - // working in all three deployment shapes we actually run in: - // - // 1. Production / npm install / built monorepo — this module is - // loaded from `dist/adapters/oxigraph-worker.js`, so the - // sibling `./oxigraph-worker-impl.js` resolves correctly. - // 2. vitest against raw source — this module is loaded from - // `src/adapters/oxigraph-worker.ts`, so the sibling - // `./oxigraph-worker-impl.js` does NOT exist, but its compiled - // twin in `dist/adapters/` does as long as the caller ran - // `pnpm --filter ...dkg-storage build` first. Redirect to - // that path so the adapter is runnable in dev loops. - // 3. Neither file exists — genuinely unbuilt tree. Throw a loud, - // actionable error explaining the fix (`pnpm build`), matching - // the expectation in `test/storage.test.ts`. - const siblingJsUrl = new URL('./oxigraph-worker-impl.js', import.meta.url); - const siblingJsPath = fileURLToPath(siblingJsUrl); - let workerPath: string | null = existsSync(siblingJsPath) ? siblingJsPath : null; - if (!workerPath) { - const srcAdapters = `${sep}src${sep}adapters${sep}`; - const distAdapters = `${sep}dist${sep}adapters${sep}`; - if (siblingJsPath.includes(srcAdapters)) { - const candidate = siblingJsPath.replace(srcAdapters, distAdapters); - if (existsSync(candidate)) workerPath = candidate; - } - } - if (!workerPath) { - throw new Error( - `oxigraph-worker adapter: compiled worker artefact ` + - `\`oxigraph-worker-impl.js\` was not found next to ` + - `${siblingJsPath} or in the sibling \`dist/adapters/\` ` + - `directory. Run \`pnpm --filter @origintrail-official/dkg-storage build\` ` + - `before using this adapter.`, - ); - } - this.worker = new Worker(workerPath, { - workerData: { persistPath }, - }); - this.worker.on('message', (msg: { id: number; result?: unknown; error?: string }) => { - const p = this.pending.get(msg.id); - if (!p) return; - this.pending.delete(msg.id); - if (msg.error) p.reject(new Error(msg.error)); - else p.resolve(msg.result); - }); - this.worker.on('error', (err) => { - for (const [, p] of this.pending) p.reject(err); - this.pending.clear(); - }); - // Once the worker exits, no pending op can ever get a reply. Settle them all - // (reject) instead of leaving callers hung forever — this is what makes the - // unbounded `close()` safe: if a second close (or any op) is still queued - // when terminate() kills the thread, it rejects here rather than hanging. - this.worker.on('exit', () => { - this.workerExited = true; - if (this.pending.size > 0) { - const err = new Error('oxigraph-worker: worker exited before the operation completed (store closed)'); - for (const [, p] of this.pending) p.reject(err); - this.pending.clear(); - } - }); - } - - private call(method: string, ...args: unknown[]): Promise { - // Only read-only ops are bounded; mutations run unbounded (timeoutMs 0) so a - // timed-out write is never reported as a clean failure while still in flight. - const timeoutMs = READ_ONLY_METHODS.has(method) ? this.operationTimeoutMs : 0; - return this.callWithTimeout(timeoutMs, undefined, method, ...args); - } - - /** - * Post one op to the worker and await its reply, bounding the caller's wait by - * `timeoutMs` (0 = wait indefinitely). The bound is per-CALLER: on timeout or - * caller abort we reject and drop the pending entry, but the single-threaded worker is STILL - * running the op — the late reply is then ignored (the message handler no-ops - * on a missing id) rather than double-settling this promise. Only read-only - * ops are ever given a non-zero timeout (see `call`), so a fired timeout is - * always a determinate, side-effect-free failure. - */ - private callWithTimeout( - timeoutMs: number, - signal: AbortSignal | undefined, - method: string, - ...args: unknown[] - ): Promise { - const id = this.nextId++; - return new Promise((resolve, reject) => { - // The worker is gone — a posted message would never be answered, so fail - // fast instead of registering a pending entry that can only ever hang. - if (this.workerExited) { - reject(new Error(`oxigraph-worker: cannot run "${method}" — the store is closed.`)); - return; - } - if (signal?.aborted) { - reject(asAbortError(signal.reason)); - return; - } - let timer: ReturnType | undefined; - let onAbort: (() => void) | undefined; - const cleanup = () => { - if (timer) clearTimeout(timer); - if (signal && onAbort) signal.removeEventListener('abort', onAbort); - }; - if (timeoutMs > 0) { - timer = setTimeout(() => { - if (this.pending.delete(id)) { - cleanup(); - const err = new Error( - `oxigraph-worker: "${method}" timed out after ${timeoutMs}ms. ` + - `The embedded store runs on a single worker thread, so a long-running or ` + - `stuck operation blocks all reads queued behind it. For heavy workloads ` + - `point the node at an external SPARQL server (store.backend "sparql-http" ` + - `/ "blazegraph"), or raise / disable store.options.operationTimeoutMs.`, - ) as OxigraphWorkerTimeoutError; - err.code = 'OXIGRAPH_WORKER_OP_TIMEOUT'; - err.method = method; - err.timeoutMs = timeoutMs; - reject(err); - } - }, timeoutMs); - // A pending-op timer must not keep the process alive on its own. - if (typeof timer.unref === 'function') timer.unref(); - } - this.pending.set(id, { - resolve: (v) => { cleanup(); resolve(v as T); }, - reject: (e) => { cleanup(); reject(e); }, - }); - if (signal) { - onAbort = () => { - if (this.pending.delete(id)) { - cleanup(); - reject(asAbortError(signal.reason)); - } - }; - signal.addEventListener('abort', onAbort, { once: true }); - if (signal.aborted) { - onAbort(); - return; - } - } - try { - this.worker.postMessage({ id, method, args }); - } catch (err) { - if (this.pending.delete(id)) { - cleanup(); - reject(err instanceof Error ? err : new Error(String(err))); - } - } - }); - } - - // A single atomic worker message: all quads commit together or the call - // fails. The contract is "all-or-nothing" and every caller can rely on it - // (e.g. FinalizationHandler packs both canonical copies into one insert so one - // can't land without the other). Large idempotent bulk imports that want - // head-of-line fairness should run on an external SPARQL server, not by - // silently fragmenting this insert into non-atomic chunks. - async insert(quads: Quad[]): Promise { return this.call('insert', quads); } - async delete(quads: Quad[]): Promise { return this.call('delete', quads); } - async deleteByPattern(pattern: Partial): Promise { return this.call('deleteByPattern', pattern); } - // Server-side SPARQL UPDATE forwarded to the worker's OxigraphStore (which - // implements `update`); same atomic single-message contract as `insert`. - async update(sparql: string): Promise { return this.call('update', sparql); } - async query(sparql: string, options?: TripleStoreQueryOptions): Promise { - return this.callWithTimeout(this.operationTimeoutMs, options?.signal, 'query', sparql); - } - async hasGraph(graphUri: string): Promise { return this.call('hasGraph', graphUri); } - async createGraph(graphUri: string): Promise { return this.call('createGraph', graphUri); } - async dropGraph(graphUri: string): Promise { return this.call('dropGraph', graphUri); } - async listGraphs(options?: TripleStoreQueryOptions): Promise { - return this.callWithTimeout(this.operationTimeoutMs, options?.signal, 'listGraphs'); - } - async deleteBySubjectPrefix(graphUri: string, prefix: string): Promise { return this.call('deleteBySubjectPrefix', graphUri, prefix); } - async countQuads(graphUri?: string): Promise { return this.call('countQuads', graphUri); } - async flush(): Promise { return this.call('flush'); } - - async close(): Promise { - // Memoized + serialized: every close() call shares ONE teardown promise, so - // we issue exactly one close RPC. (A second unbounded close RPC would be - // orphaned when terminate() kills the worker after the first resolves, and - // — without the exit handler — would hang forever.) - if (!this.closePromise) this.closePromise = this.doClose(); - return this.closePromise; - } - - private async doClose(): Promise { - // Worker already gone (crash/kill) — there's nothing to flush; just make - // sure the thread is reaped. Keeps close() idempotent and non-throwing. - if (this.workerExited) { - try { await this.worker.terminate(); } catch { /* already terminated */ } - return; - } - // `close` runs the worker's FINAL synchronous flush (insert() only schedules - // a 50ms debounced flush, so close is what guarantees durability). It is - // therefore EXEMPT from the per-op timeout (timeoutMs 0): bounding it could - // fire the timeout while the worker is mid-flush, and the `finally` would - // then terminate() the thread before pending writes hit disk — losing data. - // terminate() always runs in `finally`; the worker's 'exit' handler then - // rejects anything still pending, so nothing leaks or hangs. - try { - await this.callWithTimeout(0, undefined, 'close'); - } finally { - await this.worker.terminate(); - } - } -} - -registerTripleStoreAdapter('oxigraph-worker', async (opts) => { - const filePath = opts?.path as string | undefined; - return new OxigraphWorkerStore(filePath, { - operationTimeoutMs: - typeof opts?.operationTimeoutMs === 'number' ? (opts.operationTimeoutMs as number) : undefined, - }); -}); diff --git a/packages/storage/src/adapters/oxigraph.ts b/packages/storage/src/adapters/oxigraph.ts index ce0effa3b1..3f858e28fa 100644 --- a/packages/storage/src/adapters/oxigraph.ts +++ b/packages/storage/src/adapters/oxigraph.ts @@ -218,8 +218,8 @@ export class OxigraphStore implements TripleStore { async query(sparql: string, options?: TripleStoreQueryOptions): Promise { throwIfAborted(options?.signal); // The embedded Oxigraph binding executes synchronously, so a caller abort - // cannot interrupt this native call mid-flight. Use oxigraph-worker or an - // HTTP backend when long sync queries need prompt cancellation. + // cannot interrupt this native call mid-flight. Use an HTTP backend when + // long sync queries need prompt cancellation. const result = this.store.query(sparql); throwIfAborted(options?.signal); diff --git a/packages/storage/src/index.ts b/packages/storage/src/index.ts index ef453ae86e..44c7e8a1bd 100644 --- a/packages/storage/src/index.ts +++ b/packages/storage/src/index.ts @@ -32,7 +32,6 @@ export { } from './graph-set-index-store.js'; export { OxigraphStore } from './adapters/oxigraph.js'; -export { OxigraphWorkerStore } from './adapters/oxigraph-worker.js'; export { BlazegraphStore } from './adapters/blazegraph.js'; export { SparqlHttpStore, @@ -45,6 +44,5 @@ export { PrivateContentStore } from './private-store.js'; // Side-effect: register built-in adapters import './adapters/oxigraph.js'; -import './adapters/oxigraph-worker.js'; import './adapters/blazegraph.js'; import './adapters/sparql-http.js'; diff --git a/packages/storage/src/triple-store.ts b/packages/storage/src/triple-store.ts index ab8476334b..222e7ac4f9 100644 --- a/packages/storage/src/triple-store.ts +++ b/packages/storage/src/triple-store.ts @@ -103,7 +103,7 @@ export interface TripleStore { close(): Promise; } -export type TripleStoreBackend = 'oxigraph' | 'oxigraph-persistent' | 'oxigraph-worker' | 'blazegraph' | 'sparql-http' | string; +export type TripleStoreBackend = 'oxigraph' | 'oxigraph-persistent' | 'blazegraph' | 'sparql-http' | string; // Backends that talk to a remote SPARQL endpoint over HTTP rather than // owning local files. The local/external split governs three pieces of @@ -201,6 +201,14 @@ export function registerTripleStoreAdapter( export async function createTripleStore( config: TripleStoreConfig, ): Promise { + if (config.backend === 'oxigraph-worker') { + throw new Error( + 'TripleStore backend "oxigraph-worker" is no longer supported. ' + + 'Use the daemon-managed "oxigraph-server" config, an external ' + + '"sparql-http"/"blazegraph" store, or "oxigraph-persistent" for ' + + 'local embedded persistence.', + ); + } const factory = adapterRegistry.get(config.backend); if (!factory) { throw new Error( @@ -254,8 +262,7 @@ function shouldEnableGraphSetIndex(config: TripleStoreConfig): boolean { function isDefaultLocalGraphSetIndexBackend(backend: TripleStoreBackend): boolean { return backend === 'oxigraph' - || backend === 'oxigraph-persistent' - || backend === 'oxigraph-worker'; + || backend === 'oxigraph-persistent'; } function resolveLargeLiteralStorageOptions( diff --git a/packages/storage/test/is-external-backend.test.ts b/packages/storage/test/is-external-backend.test.ts index 759e08ec35..daae5378d4 100644 --- a/packages/storage/test/is-external-backend.test.ts +++ b/packages/storage/test/is-external-backend.test.ts @@ -18,7 +18,6 @@ describe('isExternalBackend', () => { it('returns false for the oxigraph family', () => { expect(isExternalBackend('oxigraph')).toBe(false); - expect(isExternalBackend('oxigraph-worker')).toBe(false); expect(isExternalBackend('oxigraph-persistent')).toBe(false); }); diff --git a/packages/storage/test/oxigraph-worker-resilience.test.ts b/packages/storage/test/oxigraph-worker-resilience.test.ts deleted file mode 100644 index 89c7c6cb98..0000000000 --- a/packages/storage/test/oxigraph-worker-resilience.test.ts +++ /dev/null @@ -1,261 +0,0 @@ -import { describe, it, expect } from 'vitest'; -import { mkdtempSync, rmSync } from 'node:fs'; -import { tmpdir } from 'node:os'; -import { join } from 'node:path'; -import { OxigraphWorkerStore, createTripleStore, type Quad } from '../src/index.js'; - -// These exercise the embedded worker adapter's resilience guards added to stop -// a single slow/wedged store op from hanging every other store-backed request -// behind it (issues #997 / #999 / #1002 / #1005 / #1008). They need the -// compiled worker artifact (`dist/adapters/oxigraph-worker-impl.js`); if it's -// missing we fail loudly with the remediation hint rather than silently skip, -// matching `storage.test.ts`'s convention. -// -// Design note: the per-op timeout is applied to READ-ONLY ops only. A read is -// side-effect-free, so rejecting it after the bound is a clean, determinate -// failure that surfaces a wedged worker on the exact paths that hang in prod. -// Mutations are left unbounded — timing one out would only drop the caller's -// promise while the write is still in flight, which the rest of the codebase -// would mis-read as a clean failure. So the timeout tests provoke a timeout by -// queuing a READ behind a busy worker, not by bounding a write. -function makeStore(opts?: { operationTimeoutMs?: number }, persistPath?: string): OxigraphWorkerStore { - try { - return new OxigraphWorkerStore(persistPath, opts); - } catch (err) { - const msg = err instanceof Error ? err.message : String(err); - if (/oxigraph-worker-impl/.test(msg)) { - throw new Error( - `oxigraph-worker adapter is not runnable — run ` + - `\`pnpm --filter @origintrail-official/dkg-storage build\` first. Underlying: ${msg}`, - ); - } - throw err; - } -} - -// For the timeout tests the worker is left mid-operation; close() forcibly -// terminates the thread, but the graceful close reply may itself time out, so -// swallow any error during teardown. -async function closeQuietly(store: OxigraphWorkerStore): Promise { - await store.close().catch(() => {}); -} - -function quads(n: number): Quad[] { - const out: Quad[] = new Array(n); - for (let i = 0; i < n; i += 1) { - out[i] = { - subject: `urn:test:s:${i}`, - predicate: 'http://schema.org/name', - object: `"v${i}"`, - graph: 'urn:test:g', - }; - } - return out; -} - -function abortDuringListenerRegistration(message: string): AbortSignal { - let aborted = false; - let reason: Error | undefined; - return { - get aborted() { - return aborted; - }, - get reason() { - return reason; - }, - addEventListener(type: string, listener: EventListenerOrEventListenerObject) { - if (type !== 'abort') return; - aborted = true; - reason = new Error(message); - if (typeof listener === 'function') listener(new Event('abort')); - else listener.handleEvent(new Event('abort')); - }, - removeEventListener: () => undefined, - dispatchEvent: () => true, - onabort: null, - } as unknown as AbortSignal; -} - -// Occupy the single worker thread with a large UNBOUNDED insert (mutations are -// not bounded by the timeout), so a read posted right after it queues behind a -// busy worker — the production "wedged worker" signature in miniature. -const BUSY_QUERY = 'ASK { GRAPH { ?s ?p ?o } }'; - -describe('OxigraphWorkerStore resilience', () => { - it('rejects a READ queued behind a busy worker after operationTimeoutMs', async () => { - // 50k inserts take hundreds of ms; a 5ms bound on the queued read must - // reject well before the worker frees up. - const store = makeStore({ operationTimeoutMs: 5 }); - try { - const busy = store.insert(quads(50_000)); // occupies the worker (unbounded) - await expect(store.query(BUSY_QUERY)).rejects.toThrow(/timed out after 5ms/); - await busy.catch(() => {}); - } finally { - await closeQuietly(store); - } - }); - - it('surfaces the timeout as a typed OXIGRAPH_WORKER_OP_TIMEOUT error', async () => { - const store = makeStore({ operationTimeoutMs: 5 }); - try { - const busy = store.insert(quads(50_000)); - const err: any = await store.query(BUSY_QUERY).then(() => null, (e) => e); - expect(err).toBeTruthy(); - expect(err.code).toBe('OXIGRAPH_WORKER_OP_TIMEOUT'); - expect(err.method).toBe('query'); - expect(err.timeoutMs).toBe(5); - await busy.catch(() => {}); - } finally { - await closeQuietly(store); - } - }); - - it('rejects a queued READ promptly when the caller aborts', async () => { - const store = makeStore({ operationTimeoutMs: 60_000 }); - try { - const busy = store.insert(quads(50_000)); - const controller = new AbortController(); - const read = store.query(BUSY_QUERY, { signal: controller.signal }); - controller.abort(new Error('caller aborted queued read')); - await expect(read).rejects.toThrow(/caller aborted queued read/); - await busy.catch(() => {}); - } finally { - await closeQuietly(store); - } - }); - - it('rejects when the caller aborts while registering the abort listener', async () => { - const store = makeStore({ operationTimeoutMs: 60_000 }); - try { - await expect( - store.query(BUSY_QUERY, { signal: abortDuringListenerRegistration('listener registration aborted') }), - ).rejects.toThrow(/listener registration aborted/); - } finally { - await closeQuietly(store); - } - }); - - it('does NOT bound mutations — a large insert under a tiny timeout still completes', async () => { - // The whole point of read-only scoping: a write is never reported as a clean - // failure while it's still in flight. With a 5ms bound a 50k insert (>>5ms) - // would reject if it were bounded; it must resolve instead. - const store = makeStore({ operationTimeoutMs: 5 }); - try { - await expect(store.insert(quads(50_000))).resolves.toBeUndefined(); - } finally { - await closeQuietly(store); - } - }); - - it('completes normally within a generous timeout', async () => { - const store = makeStore({ operationTimeoutMs: 60_000 }); - try { - await store.insert(quads(10)); - expect(await store.countQuads('urn:test:g')).toBe(10); - // The data lives in a NAMED graph, so the ASK must scope to it (a bare - // `ASK { ?s ?p ?o }` only matches the default graph). - const r = await store.query(BUSY_QUERY); - expect(r.type).toBe('boolean'); - if (r.type === 'boolean') expect(r.value).toBe(true); - } finally { - await closeQuietly(store); - } - }); - - it('operationTimeoutMs: 0 disables the timeout (a queued read still completes)', async () => { - const store = makeStore({ operationTimeoutMs: 0 }); - try { - const busy = store.insert(quads(50_000)); - // With the bound disabled, the read WAITS for the worker instead of - // rejecting, then returns true once the import lands. - const r = await store.query(BUSY_QUERY); - expect(r.type).toBe('boolean'); - if (r.type === 'boolean') expect(r.value).toBe(true); - await busy; - expect(await store.countQuads('urn:test:g')).toBe(50_000); - } finally { - await closeQuietly(store); - } - }); - - it('a fractional operationTimeoutMs is floored to an integer', async () => { - // normalizeNonNegativeInt floors the ms bound: 5.9 behaves like 5, so the - // surfaced error reports "after 5ms" (no fractional noise). - const store = makeStore({ operationTimeoutMs: 5.9 }); - try { - const busy = store.insert(quads(50_000)); - await expect(store.query(BUSY_QUERY)).rejects.toThrow(/timed out after 5ms/); - await busy.catch(() => {}); - } finally { - await closeQuietly(store); - } - }); - - it('close() is exempt from the per-op timeout so the final flush is never cut short', async () => { - // Codex review: close runs the worker's final flush; bounding it by the - // per-op timeout could terminate() the thread mid-flush and lose writes. - // With a tiny operationTimeoutMs and the worker still busy on a large - // insert, close() must WAIT for the worker to drain rather than reject — - // AND the in-flight write must actually land, not get killed mid-flush. - // We prove durability end-to-end on a persistent path: the regression this - // guards (close-after-debounced-flush race) was a SILENT data loss, so the - // test must assert the quads survive a reopen, not just that close resolves. - const dir = mkdtempSync(join(tmpdir(), 'oxigraph-worker-close-')); - const path = join(dir, 'store.nq'); - try { - const store = makeStore({ operationTimeoutMs: 10 }, path); - const busy = store.insert(quads(50_000)); // occupies the worker - // A read queued behind it times out at 10ms — proves the worker is busy. - await expect(store.query(BUSY_QUERY)).rejects.toThrow(/timed out/); - // close() must resolve cleanly (not reject with a 10ms timeout): it waits - // for the in-flight op to drain, then flushes + terminates. - await expect(store.close()).resolves.toBeUndefined(); - // The in-flight insert must have COMPLETED (drained), not been terminated - // mid-flight — otherwise close() silently cut it short. - await expect(busy).resolves.toBeUndefined(); - // And the write must be durable: a fresh worker on the same path hydrates - // all 50k quads, proving close() flushed before terminating. - const reopened = makeStore({ operationTimeoutMs: 60_000 }, path); - try { - expect(await reopened.countQuads('urn:test:g')).toBe(50_000); - } finally { - await closeQuietly(reopened); - } - } finally { - try { rmSync(dir, { recursive: true, force: true }); } catch { /* */ } - } - }); - - it('concurrent and repeated close() calls all resolve without hanging', async () => { - // Codex review: close() goes through an UNBOUNDED worker RPC, so a second - // close racing the first was orphaned when terminate() killed the worker and - // never settled. close() is now memoized + the worker 'exit' handler rejects - // anything still pending, so concurrent/repeat closes all settle. - const store = makeStore({ operationTimeoutMs: 60_000 }); - await store.insert(quads(5)); - const results = await Promise.all([store.close(), store.close(), store.close()]); - expect(results).toEqual([undefined, undefined, undefined]); - // Ops issued after close fail fast (store closed) instead of hanging. - await new Promise((r) => setImmediate(r)); - await expect(store.insert(quads(1))).rejects.toThrow(/closed/i); - }); - - it('store.options reach the adapter through createTripleStore (factory path)', async () => { - // Codex review: the user-facing path is createTripleStore({ backend, options }), - // not the constructor — assert the option forwarding in the adapter factory - // actually takes effect so a typo there can't silently drop the knob. - // operationTimeoutMs forwarded: a 5ms bound rejects a read queued behind a - // busy worker. - const store = await createTripleStore({ - backend: 'oxigraph-worker', - options: { operationTimeoutMs: 5 }, - }); - try { - const busy = store.insert(quads(50_000)); - await expect(store.query(BUSY_QUERY)).rejects.toThrow(/timed out after 5ms/); - await busy.catch(() => {}); - } finally { - await store.close().catch(() => {}); - } - }); -}); diff --git a/packages/storage/test/storage.test.ts b/packages/storage/test/storage.test.ts index 6374df6f5d..0af0660c1b 100644 --- a/packages/storage/test/storage.test.ts +++ b/packages/storage/test/storage.test.ts @@ -245,9 +245,8 @@ describe('createTripleStore factory', () => { it('all built-in backends are registered (factory throws something other than "Unknown TripleStore backend")', async () => { // The *registry* contract being tested here is: every built-in // backend name is recognized. The construction itself may require - // options (blazegraph needs `url`, sparql-http needs `queryEndpoint`) - // or worker artifacts (oxigraph-worker needs the compiled worker - // impl). So a backend passes this test iff calling `createTripleStore` + // options (blazegraph needs `url`, sparql-http needs `queryEndpoint`). + // So a backend passes this test iff calling `createTripleStore` // either succeeds OR throws a *non*-"Unknown TripleStore backend" // error. // @@ -256,7 +255,7 @@ describe('createTripleStore factory', () => { // effectively assert "a promise settled" — noise. This version // asserts the positive contract explicitly and points at the // specific failing backend if the registry regresses. - const backends = ['oxigraph', 'oxigraph-worker', 'blazegraph', 'sparql-http']; + const backends = ['oxigraph', 'blazegraph', 'sparql-http']; for (const backend of backends) { let outcome: 'constructed' | Error; try { @@ -302,45 +301,10 @@ describe('createTripleStore factory', () => { ).rejects.toThrow('queryEndpoint'); }); - it('oxigraph-worker adapter is registered and round-trips an insert', async () => { - // The worker adapter resolves `./oxigraph-worker-impl.js` relative to - // the module loaded at runtime. When vitest runs against raw source - // without a prior `pnpm build`, that URL lands in `src/adapters/` where - // only the .ts files live, so the Worker constructor throws - // "Cannot find module … oxigraph-worker-impl.js". - // - // This used to be caught and converted to `ctx.skip()`, which meant a - // green CI run even when the worker artifact was missing — i.e. a - // broken build never triggered a test failure. We now FAIL LOUDLY in - // that case with a remediation hint, so: - // • locally, the developer sees "run pnpm build first" instead of a - // silent skip; - // • in CI, if `pnpm build` was not wired into the lane (or the build - // regresses), this test surfaces it as a red failure. - let store: Awaited>; - try { - store = await createTripleStore({ backend: 'oxigraph-worker' }); - } catch (err: unknown) { - const msg = err instanceof Error ? err.message : String(err); - if (msg.includes('Cannot find module') && msg.includes('oxigraph-worker-impl')) { - throw new Error( - `oxigraph-worker adapter is not runnable — the compiled ` + - `oxigraph-worker-impl.js artifact is missing from ` + - `packages/storage/dist/adapters/. Run ` + - `\`pnpm --filter @origintrail-official/dkg-storage build\` ` + - `before running this test. Underlying error: ${msg}`, - ); - } - throw err; - } - await store.insert([{ - subject: 'http://ex.org/s', - predicate: 'http://ex.org/p', - object: '"hi"', - graph: 'http://ex.org/g', - }]); - expect(await store.countQuads()).toBe(1); - await store.close(); + it('rejects the retired oxigraph-worker backend with an actionable message', async () => { + await expect(createTripleStore({ backend: 'oxigraph-worker' })).rejects.toThrow( + /no longer supported/, + ); }); it('throws on unknown backend', async () => { diff --git a/scripts/devnet.sh b/scripts/devnet.sh index fc3e3a063b..6dd5d952f9 100755 --- a/scripts/devnet.sh +++ b/scripts/devnet.sh @@ -438,7 +438,7 @@ create_node_config() { # and spawns it on its own port — NO Docker. Node 1 is the node the # UI/e2e suite drives, so the suite now exercises the REAL default # backend and deterministically reproduces SPARQL-over-HTTP-only - # bugs such as #996 — which the old `oxigraph-worker` default hid.) + # bugs such as #996.) # Node 3-4: blazegraph (if Docker) else oxigraph (in-process baseline) # Node 5-6: sparql-http → external Dockerized Oxigraph (EXTRA coverage of the # generic external-endpoint path; Docker-only, optional) @@ -460,6 +460,8 @@ create_node_config() { local ox_port_var="OXIGRAPH_SERVER_PORT_${node_num}" local ox_port="${!ox_port_var}" store_block="\"store\": { \"backend\": \"sparql-http\", \"options\": { \"queryEndpoint\": \"http://127.0.0.1:${ox_port}/query\", \"updateEndpoint\": \"http://127.0.0.1:${ox_port}/update\" } }," + else + store_block="\"store\": { \"backend\": \"oxigraph\" }," fi fi @@ -1343,12 +1345,12 @@ cmd_start() { local api_port=$((API_PORT_BASE + i - 1)) local role="edge" [ "$i" -le "$NUM_CORE_NODES" ] && role="core" - local store_label="oxigraph-worker" + local store_label="oxigraph-server" if [ "$i" -ge 3 ] && [ "$i" -le 4 ]; then [ "$BLAZEGRAPH_AVAILABLE" = true ] && store_label="blazegraph" || store_label="oxigraph" fi if [ "$i" -ge 5 ]; then - [ "$OXIGRAPH_SERVER_AVAILABLE" = true ] && store_label="oxigraph-server" || store_label="oxigraph-worker" + [ "$OXIGRAPH_SERVER_AVAILABLE" = true ] && store_label="sparql-http" || store_label="oxigraph" fi log "Node $i ($role, $store_label): http://127.0.0.1:$api_port/ui" done diff --git a/scripts/publisher-smoke-test.sh b/scripts/publisher-smoke-test.sh index 1086fe0cc5..28472b5660 100755 --- a/scripts/publisher-smoke-test.sh +++ b/scripts/publisher-smoke-test.sh @@ -76,7 +76,7 @@ import { DKGPublisher, TripleStoreAsyncLiftPublisher } from '@origintrail-offici const dkgHome = process.env.DKG_HOME; const privateKey = process.env.SMOKE_PRIVATE_KEY; const store = await createTripleStore({ - backend: 'oxigraph-worker', + backend: 'oxigraph-persistent', options: { path: join(dkgHome, 'store.nq') }, });