fix(pipeline): fail fast when the provenance store cannot persist records

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<void>;
   get(datasetUri: URL): Promise<ProcessingRecord | null>;
   set(datasetUri: URL, record: ProcessingRecord): Promise<void>;
 }
 ```
 
+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
 

packages/pipeline/src/pipeline.ts

@@ -466,6 +466,12 @@ export class Pipeline<Out = Quad> {
 
     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<Out = Quad> {
         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)),
+      );
     }
   }
 

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<void> {
+    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<ProcessingRecord | null> {
     const records = await this.readAll();
     return records[datasetUri.toString()] ?? null;

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<void>;
   /**
    * 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

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<typeof vi.fn> } = {
+        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<typeof vi.fn> } = {
+        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<typeof vi.fn> } = {
+        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 () => {

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);
+      }
+    });
+  });
 });

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,
         },
       },
     },

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"]

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