From 70463fee117c91b26da3b5000bfc8142224f4480 Mon Sep 17 00:00:00 2001 From: David de Boer Date: Sat, 25 Jul 2026 20:35:35 +0200 Subject: [PATCH] fix(pipeline): fail fast when the provenance store cannot persist records - Add an optional check() to ProvenanceStore; the pipeline calls it once at the start of a run, so a store that would fail every write fails the run immediately instead of silently disabling skip-unchanged - FileProvenanceStore.check() probes the file's directory with a real write, catching a mount the runtime user cannot write to - Report a record write that fails mid-run through the reporter (stageFailed, stage 'provenance') instead of swallowing it - search-indexer image: pre-create /provenance owned by the runtime user, so a fresh named volume mounted there is writable by uid 1000 --- packages/pipeline/README.md | 5 +- packages/pipeline/src/pipeline.ts | 18 ++++- .../src/provenance/fileProvenanceStore.ts | 22 ++++- packages/pipeline/src/provenance/store.ts | 8 ++ packages/pipeline/test/pipeline.test.ts | 81 ++++++++++++++++++- .../provenance/fileProvenanceStore.test.ts | 53 +++++++++++- packages/pipeline/vite.config.ts | 8 +- packages/search-indexer/Dockerfile | 5 +- packages/search-indexer/README.md | 6 +- 9 files changed, 193 insertions(+), 13 deletions(-) diff --git a/packages/pipeline/README.md b/packages/pipeline/README.md index 4b7845ae..6ae74c09 100644 --- a/packages/pipeline/README.md +++ b/packages/pipeline/README.md @@ -339,11 +339,14 @@ A `ProvenanceStore` gives the pipeline a small per-dataset memory, so a future r ```typescript interface ProvenanceStore { + check?(): Promise; get(datasetUri: URL): Promise; set(datasetUri: URL, record: ProcessingRecord): Promise; } ``` +The optional `check` verifies the store can persist records; the pipeline calls it once at the start of a run, so a store that would fail every write fails the run immediately instead of silently disabling skip-unchanged. A record write that fails mid-run is reported through the reporter (`stageFailed` with stage `'provenance'`) and does not abort the run. + A `ProcessingRecord` holds the two opaque change fields – `sourceFingerprint` (derived automatically from source metadata) and `pipelineVersion` (consumer-declared) – plus `generatedAt` and a `status` of `'success'` or `'failed'`. The two change fields are compared only for equality, never parsed or ordered. #### `FileLoadedSparqlProvenanceStore` @@ -375,7 +378,7 @@ import { FileProvenanceStore } from '@lde/pipeline'; const store = new FileProvenanceStore({ path: './state/provenance.json' }); ``` -Writes are atomic (temp file + rename), so a run killed mid-write cannot corrupt the next run’s skip decisions. Safe for a single writer: concurrent pipeline processes writing the same file lose each other’s updates – use the SPARQL-backed store (or file locking) instead. The file must sit on a durable volume to survive across runs, the same requirement the SPARQL-backed store already has. +Writes are atomic (temp file + rename), so a run killed mid-write cannot corrupt the next run’s skip decisions. Safe for a single writer: concurrent pipeline processes writing the same file lose each other’s updates – use the SPARQL-backed store (or file locking) instead. The file must sit on a durable volume to survive across runs, the same requirement the SPARQL-backed store already has, and its directory must be writable by the user the pipeline runs as – `check` probes this with a real write at the start of the run, so a read-only or root-owned mount fails loudly instead of silently disabling skipping. #### Enabling skipping diff --git a/packages/pipeline/src/pipeline.ts b/packages/pipeline/src/pipeline.ts index b1556e8b..28cc3b20 100644 --- a/packages/pipeline/src/pipeline.ts +++ b/packages/pipeline/src/pipeline.ts @@ -466,6 +466,12 @@ export class Pipeline { this.reporter?.pipelineStart?.(this.name); + // Fail fast on a store that cannot persist records (e.g. a provenance + // file on a mount the runtime user cannot write to): every dataset would + // process fine, but the records would be lost and skip-unchanged would + // silently never engage on any later run. + await this.provenanceStore?.check?.(); + const selectStart = Date.now(); const datasets = await this.datasetSelector.select(); this.reporter?.datasetsSelected?.(datasets.total, Date.now() - selectStart); @@ -751,9 +757,15 @@ export class Pipeline { generatedAt: new Date().toISOString(), status, }); - } catch { - // A failed write must not abort the run; the dataset simply reprocesses - // next run, its record not yet updated. + } catch (error) { + // A failed write must not abort the run – the dataset simply + // reprocesses next run, its record not yet updated – but it must be + // surfaced: a store that fails every write silently disables + // skip-unchanged. + this.reporter?.stageFailed?.( + 'provenance', + error instanceof Error ? error : new Error(String(error)), + ); } } diff --git a/packages/pipeline/src/provenance/fileProvenanceStore.ts b/packages/pipeline/src/provenance/fileProvenanceStore.ts index cd8292ab..2995a9bd 100644 --- a/packages/pipeline/src/provenance/fileProvenanceStore.ts +++ b/packages/pipeline/src/provenance/fileProvenanceStore.ts @@ -1,4 +1,4 @@ -import { mkdir, readFile, rename, writeFile } from 'node:fs/promises'; +import { mkdir, readFile, rename, unlink, writeFile } from 'node:fs/promises'; import { dirname } from 'node:path'; import type { ProcessingRecord } from './record.js'; import type { ProvenanceStore } from './store.js'; @@ -34,6 +34,26 @@ export class FileProvenanceStore implements ProvenanceStore { this.path = options.path; } + /** + * Verify the file can be written by actually creating (and removing) a + * probe file next to it. Catches the mount that a non-root runtime user + * cannot write to (e.g. a root-owned Docker volume, see issue #661) at the + * start of the run, before writes would silently fail per dataset. + */ + async check(): Promise { + try { + await mkdir(dirname(this.path), { recursive: true }); + const probePath = `${this.path}.${process.pid}.check`; + await writeFile(probePath, ''); + await unlink(probePath); + } catch (error) { + throw new Error( + `Provenance file ${this.path} is not writable; make its directory writable by the user the pipeline runs as: ${String(error)}`, + { cause: error }, + ); + } + } + async get(datasetUri: URL): Promise { const records = await this.readAll(); return records[datasetUri.toString()] ?? null; diff --git a/packages/pipeline/src/provenance/store.ts b/packages/pipeline/src/provenance/store.ts index 9afe1594..7ff85544 100644 --- a/packages/pipeline/src/provenance/store.ts +++ b/packages/pipeline/src/provenance/store.ts @@ -9,6 +9,14 @@ import type { ProcessingRecord } from './record.js'; * triplestore, files, or anything else. */ export interface ProvenanceStore { + /** + * Verify the store can persist records, throwing when it cannot. The + * pipeline calls this once at the start of a run, so a store that would + * fail every write – e.g. a provenance file on a mount the runtime user + * cannot write to – fails the run immediately instead of silently + * disabling skip-unchanged. + */ + check?(): Promise; /** * The record from the dataset’s last processing, or `null` if it has never * been processed (or the store was wiped). A `null` result drives a diff --git a/packages/pipeline/test/pipeline.test.ts b/packages/pipeline/test/pipeline.test.ts index 4b876191..76680d6b 100644 --- a/packages/pipeline/test/pipeline.test.ts +++ b/packages/pipeline/test/pipeline.test.ts @@ -2768,7 +2768,7 @@ describe('Pipeline', () => { expect(stage.run).toHaveBeenCalledTimes(1); }); - it('continues the run when the store write fails', async () => { + it('continues the run when the store write fails, reporting each failure', async () => { const dataset1 = makeDataset('http://example.org/dataset/1'); const dataset2 = makeDataset('http://example.org/dataset/2'); const resolver = makeDumpResolver(); @@ -2777,6 +2777,7 @@ describe('Pipeline', () => { set: vi.fn().mockRejectedValue(new Error('disk full')), }; const stage = makeStage('stage1'); + const reporter = makeReporter(); const pipeline = new Pipeline({ datasetSelector: makeDatasetSelector(dataset1, dataset2), @@ -2785,12 +2786,90 @@ describe('Pipeline', () => { distributionResolver: resolver, provenanceStore: store, pipelineVersion: 'v1', + reporter, }); await expect(pipeline.run()).resolves.toBeUndefined(); // Both datasets are processed despite each write throwing. expect(stage.run).toHaveBeenCalledTimes(2); expect(store.set).toHaveBeenCalledTimes(2); + // But the failures are surfaced, not swallowed: a store that fails + // every write silently disables skip-unchanged. + expect(reporter.stageFailed).toHaveBeenCalledTimes(2); + expect(reporter.stageFailed).toHaveBeenCalledWith( + 'provenance', + expect.objectContaining({ message: 'disk full' }), + ); + }); + + it('wraps a non-Error write failure for the reporter', async () => { + const resolver = makeDumpResolver(); + const store: ProvenanceStore & { set: ReturnType } = { + get: vi.fn().mockResolvedValue(null), + set: vi.fn().mockRejectedValue('disk full'), + }; + const reporter = makeReporter(); + + const pipeline = new Pipeline({ + datasetSelector: makeDatasetSelector(dataset), + stages: [makeStage('stage1')], + writers: writer, + distributionResolver: resolver, + provenanceStore: store, + pipelineVersion: 'v1', + reporter, + }); + + await expect(pipeline.run()).resolves.toBeUndefined(); + expect(reporter.stageFailed).toHaveBeenCalledWith( + 'provenance', + expect.objectContaining({ message: 'disk full' }), + ); + }); + + it('fails the run when the store check fails, before selecting datasets', async () => { + const resolver = makeDumpResolver(); + const store: ProvenanceStore & { check: ReturnType } = { + check: vi.fn().mockRejectedValue(new Error('not writable')), + get: vi.fn().mockResolvedValue(null), + set: vi.fn().mockResolvedValue(undefined), + }; + + const pipeline = new Pipeline({ + datasetSelector: makeDatasetSelector(dataset), + stages: [makeStage('stage1')], + writers: writer, + distributionResolver: resolver, + provenanceStore: store, + pipelineVersion: 'v1', + }); + + // A store that cannot persist records would process every dataset but + // lose the outcome, so skip-unchanged would silently never engage. + await expect(pipeline.run()).rejects.toThrow('not writable'); + expect(resolver.probe).not.toHaveBeenCalled(); + }); + + it('checks the store once at the start of the run', async () => { + const resolver = makeDumpResolver(); + const store: ProvenanceStore & { check: ReturnType } = { + check: vi.fn().mockResolvedValue(undefined), + get: vi.fn().mockResolvedValue(null), + set: vi.fn().mockResolvedValue(undefined), + }; + + const pipeline = new Pipeline({ + datasetSelector: makeDatasetSelector(dataset), + stages: [makeStage('stage1')], + writers: writer, + distributionResolver: resolver, + provenanceStore: store, + pipelineVersion: 'v1', + }); + + await pipeline.run(); + + expect(store.check).toHaveBeenCalledTimes(1); }); it('does not gate or record when no store is configured', async () => { diff --git a/packages/pipeline/test/provenance/fileProvenanceStore.test.ts b/packages/pipeline/test/provenance/fileProvenanceStore.test.ts index a91907e5..2dcc1b5d 100644 --- a/packages/pipeline/test/provenance/fileProvenanceStore.test.ts +++ b/packages/pipeline/test/provenance/fileProvenanceStore.test.ts @@ -1,5 +1,12 @@ import { describe, it, expect, beforeEach, afterEach } from 'vitest'; -import { mkdtemp, readdir, readFile, rm, writeFile } from 'node:fs/promises'; +import { + chmod, + mkdtemp, + readdir, + readFile, + rm, + writeFile, +} from 'node:fs/promises'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; import type { ProcessingRecord } from '../../src/index.js'; @@ -126,4 +133,48 @@ describe('FileProvenanceStore', () => { await expect(store.get(DATASET_URI)).rejects.toThrow(); await expect(store.set(DATASET_URI, RECORD)).rejects.toThrow(); }); + + describe('check', () => { + it('passes on a writable directory, leaving nothing behind', async () => { + const store = new FileProvenanceStore({ path }); + + await store.check(); + + expect(await readdir(directory)).toEqual([]); + }); + + it('creates missing parent directories', async () => { + const nestedPath = join(directory, 'state', 'nested', 'provenance.json'); + const store = new FileProvenanceStore({ path: nestedPath }); + + await store.check(); + + expect(await readdir(join(directory, 'state', 'nested'))).toEqual([]); + }); + + it('leaves an existing provenance file untouched', async () => { + const store = new FileProvenanceStore({ path }); + await store.set(DATASET_URI, RECORD); + + await store.check(); + + expect(await store.get(DATASET_URI)).toEqual(RECORD); + expect(await readdir(directory)).toEqual(['provenance.json']); + }); + + it('rejects when the directory is not writable', async () => { + // The root-owned volume case from issue #661: the directory exists but + // the runtime user cannot create files in it. + const store = new FileProvenanceStore({ path }); + await chmod(directory, 0o555); + + try { + await expect(store.check()).rejects.toThrow( + `Provenance file ${path} is not writable`, + ); + } finally { + await chmod(directory, 0o755); + } + }); + }); }); diff --git a/packages/pipeline/vite.config.ts b/packages/pipeline/vite.config.ts index 3a830371..b43f8775 100644 --- a/packages/pipeline/vite.config.ts +++ b/packages/pipeline/vite.config.ts @@ -11,10 +11,10 @@ export default mergeConfig( coverage: { thresholds: { autoUpdate: true, - functions: 97.44, - lines: 97.1, - branches: 91.87, - statements: 96.65, + functions: 97.45, + lines: 97.12, + branches: 91.9, + statements: 96.67, }, }, }, diff --git a/packages/search-indexer/Dockerfile b/packages/search-indexer/Dockerfile index f342bf8d..585c27f9 100644 --- a/packages/search-indexer/Dockerfile +++ b/packages/search-indexer/Dockerfile @@ -23,7 +23,10 @@ COPY dist-docker/ ./ # Default DATA_DIR for the QLever import path; a real deployment mounts a # volume here (and the same host path into the sibling QLever container). -RUN mkdir -p /data && chown node:node /data +# /provenance is the conventional home of PROVENANCE_FILE. Both are owned by +# the runtime user so a fresh named volume mounted there – which inherits the +# image directory’s ownership – is writable by uid 1000 (issue #661). +RUN mkdir -p /data /provenance && chown node:node /data /provenance USER node CMD ["node", "dist/cli.js"] diff --git a/packages/search-indexer/README.md b/packages/search-indexer/README.md index b3f39a8f..d31ca1fa 100644 --- a/packages/search-indexer/README.md +++ b/packages/search-indexer/README.md @@ -79,7 +79,11 @@ images. A misconfigured boot reports **all** problems in one error, not one per crash loop. `PROVENANCE_FILE` must sit on a durable volume, and cannot be combined with `REBUILD_MODE=blue-green`: a skipped dataset would be missing from the -fresh collection the swap makes live. +fresh collection the swap makes live. Its directory must be writable by the +image’s runtime user (uid 1000) – the image pre-creates `/provenance` with +that ownership, so a named volume mounted there just works, while a bind +mount must be `chown`ed on the host. A run fails at start when the file is +not writable, instead of silently never skipping. ## The QLever import path