Ship a file-backed ProvenanceStore in @lde/pipeline

[ddeboer]

Summary

Ship a file-backed ProvenanceStore in @lde/pipeline, alongside the interface and the existing FileLoadedSparqlProvenanceStore.

Motivation

The skip-unchanged change-gate is a core pipeline feature, but provenanceStore currently has only one shipped implementation – FileLoadedSparqlProvenanceStore (SPARQL-backed). A consumer that runs the pipeline without a triplestore therefore has to either stand one up purely to remember processing records, or hand-roll a store.

The record it holds is tiny and domain-neutral: a datasetURI → ProcessingRecord map. For the common case – a single CronJob persisting to a mounted volume – a JSON file is far lighter than a SPARQL server, and it is exactly the generic-toolkit layer @lde/pipeline's provenance module already owns. (Same reasoning as @lde/pipeline-console-reporter existing next to the ProgressReporter interface.)

Proposal

A FileProvenanceStore implementing ProvenanceStore (get/set a ProcessingRecord by dataset URI), persisting a single JSON object to a configurable path, with atomic writes (temp file + rename) so a run killed mid-write cannot corrupt the next run's skip decision.

Caveats to document (not blockers)

• Concurrency: safe for a single writer. Concurrent pipeline pods writing the same file need the SPARQL-backed store (or file locking) – worth a note in the JSDoc.
• Persistence: must sit on a durable volume to survive across runs (a Kubernetes CronJob's container filesystem is discarded) – the same requirement the SPARQL-backed store already has, so no regression.

Reference implementation

Linked Open Limburg already runs exactly this (it would drop its copy and import the LDE one). The whole thing:

import { mkdir, readFile, rename, writeFile } from 'node:fs/promises';
import { dirname } from 'node:path';
import type { ProcessingRecord, ProvenanceStore } from '@lde/pipeline';

type Records = Record<string, ProcessingRecord>;

export class FileProvenanceStore implements ProvenanceStore {
  constructor(private readonly filePath: string) {}

  async get(datasetUri: URL): Promise<ProcessingRecord | null> {
    const records = await this.readAll();
    return records[datasetUri.href] ?? null;
  }

  async set(datasetUri: URL, record: ProcessingRecord): Promise<void> {
    const records = await this.readAll();
    records[datasetUri.href] = record;
    await this.writeAll(records);
  }

  private async readAll(): Promise<Records> {
    try {
      return JSON.parse(await readFile(this.filePath, 'utf8')) as Records;
    } catch (error) {
      if ((error as NodeJS.ErrnoException).code === 'ENOENT') return {};
      throw error;
    }
  }

  private async writeAll(records: Records): Promise<void> {
    await mkdir(dirname(this.filePath), { recursive: true });
    const temp = `${this.filePath}.${process.pid}.tmp`;
    await writeFile(temp, `${JSON.stringify(records, null, 2)}\n`);
    await rename(temp, this.filePath); // atomic
  }
}

Found while wiring the pipeline in a downstream project (Linked Open Limburg); happy to open the PR if useful.