From b0fea1ea953dbab20ca27b5b1506450d3a5496f8 Mon Sep 17 00:00:00 2001 From: David de Boer Date: Thu, 23 Jul 2026 14:31:52 +0200 Subject: [PATCH] feat(pipeline): add file-backed FileProvenanceStore - Persist processing records in a single JSON file keyed by dataset URI, for pipelines that run without a triplestore - Write atomically (temp file + rename) so a run killed mid-write cannot corrupt later skip decisions - Document the single-writer scope and durable-volume requirement in JSDoc and README --- packages/pipeline/README.md | 56 +++++--- .../src/provenance/fileProvenanceStore.ts | 70 ++++++++++ packages/pipeline/src/provenance/index.ts | 4 + .../provenance/fileProvenanceStore.test.ts | 129 ++++++++++++++++++ packages/pipeline/vite.config.ts | 8 +- 5 files changed, 241 insertions(+), 26 deletions(-) create mode 100644 packages/pipeline/src/provenance/fileProvenanceStore.ts create mode 100644 packages/pipeline/test/provenance/fileProvenanceStore.test.ts diff --git a/packages/pipeline/README.md b/packages/pipeline/README.md index 7e418612..c959cbcd 100644 --- a/packages/pipeline/README.md +++ b/packages/pipeline/README.md @@ -2,7 +2,7 @@ A framework for transforming large RDF datasets, primarily using [SPARQL](https://www.w3.org/TR/sparql11-query/) queries with TypeScript for the parts that are hard to express in SPARQL alone. -- **SPARQL-native.** Data transformations are plain SPARQL query files — portable, transparent, testable and version-controlled. +- **SPARQL-native.** Data transformations are plain SPARQL query files – portable, transparent, testable and version-controlled. - **Composable.** Executors are an interface: wrap a SPARQL reader with custom TypeScript to handle edge cases like date parsing or string normalisation (see [Reader](#reader)). - **Extensible.** A plugin system lets packages like [@lde/pipeline-void](../pipeline-void) (or your own plugins) hook into the pipeline lifecycle. @@ -44,9 +44,9 @@ const resolver = new ImportResolver(new SparqlDistributionResolver(), { }); ``` -- `'sparql'` (default) — use the dataset’s own SPARQL endpoint when one is available; import a data dump only when no endpoint responds. -- `'sparqlWithImportFallback'` — like `'sparql'`, but also fall back to the data dump when the endpoint passes probing yet a stage fails against it at runtime. The pipeline discards the endpoint-sourced partial output and re-runs all stages against the import. Use this when endpoints are present but unreliable for heavy aggregate queries. -- `'import'` — always import the data dump, even when a working endpoint is advertised. +- `'sparql'` (default) – use the dataset’s own SPARQL endpoint when one is available; import a data dump only when no endpoint responds. +- `'sparqlWithImportFallback'` – like `'sparql'`, but also fall back to the data dump when the endpoint passes probing yet a stage fails against it at runtime. The pipeline discards the endpoint-sourced partial output and re-runs all stages against the import. Use this when endpoints are present but unreliable for heavy aggregate queries. +- `'import'` – always import the data dump, even when a working endpoint is advertised. ### Stage @@ -90,7 +90,7 @@ new Stage({ `expectsOutput` (default: `false`) marks a stage whose query must yield at least one quad. A supported stage that produces none is then treated as a hard failure rather than a legitimately empty result. -Set it for scalar aggregates such as `SELECT (COUNT(*) AS ?n)`, which always return exactly one row — so zero output can only mean the endpoint truncated or aborted the response (e.g. a timeout surfaced as an empty `HTTP 200`). The failure flows through like any other hard stage failure, triggering the [reactive dump fallback](#distribution-resolver) when `strategy: 'sparqlWithImportFallback'` is configured. Leave it `false` for stages that may legitimately be empty, such as class or property partitions of a dataset that lacks that structure. +Set it for scalar aggregates such as `SELECT (COUNT(*) AS ?n)`, which always return exactly one row – so zero output can only mean the endpoint truncated or aborted the response (e.g. a timeout surfaced as an empty `HTTP 200`). The failure flows through like any other hard stage failure, triggering the [reactive dump fallback](#distribution-resolver) when `strategy: 'sparqlWithImportFallback'` is configured. Leave it `false` for stages that may legitimately be empty, such as class or property partitions of a dataset that lacks that structure. ### Item Selector @@ -115,7 +115,7 @@ new SparqlItemSelector({ #### Capping total results with `maxResults` -By default, `SparqlItemSelector` paginates through **all** matching rows: any `LIMIT` clause in the query is interpreted as the page size, then it walks pages with `OFFSET` until the source is exhausted. To cap the total bindings yielded across all pages — for sampling, testing, prototyping, or just safety — set `maxResults`: +By default, `SparqlItemSelector` paginates through **all** matching rows: any `LIMIT` clause in the query is interpreted as the page size, then it walks pages with `OFFSET` until the source is exhausted. To cap the total bindings yielded across all pages – for sampling, testing, prototyping, or just safety – set `maxResults`: ```typescript new SparqlItemSelector({ @@ -126,7 +126,7 @@ new SparqlItemSelector({ When `maxResults` is set: -- Pagination stops as soon as `maxResults` bindings have been yielded — no wasted page request after the cap is hit. +- Pagination stops as soon as `maxResults` bindings have been yielded – no wasted page request after the cap is hit. - The last (partial) page's `LIMIT` is shrunk to the remaining cap so the endpoint doesn't over-fetch on the remainder (e.g. with `maxResults: 85` and `pageSize: 10`, the 9th page request is `LIMIT 5`, not `LIMIT 10`). - The first page uses the configured page size as-is; `maxResults` and page size stay orthogonal. If `maxResults < pageSize`, the first page may return a few rows that aren't yielded. - `maxResults: 0` is a valid no-op; the selector yields nothing without issuing any SPARQL request. @@ -162,7 +162,7 @@ const reader = new SparqlConstructReader({ }); ``` -SPARQL CONSTRUCT queries can produce duplicate triples — for example, constant triples (like `?dataset a edm:ProvidedCHO`) are emitted for every solution row. Enable `deduplicate` to remove duplicates inline on the stream using a string-based identity set (inspired by [Comunica's `distinctConstruct`](https://comunica.dev/docs/query/advanced/context/#14--distinct-construct)): +SPARQL CONSTRUCT queries can produce duplicate triples – for example, constant triples (like `?dataset a edm:ProvidedCHO`) are emitted for every solution row. Enable `deduplicate` to remove duplicates inline on the stream using a string-based identity set (inspired by [Comunica's `distinctConstruct`](https://comunica.dev/docs/query/advanced/context/#14--distinct-construct)): ```typescript const reader = new SparqlConstructReader({ @@ -175,9 +175,9 @@ The dedup set is scoped to each `read()` call, so memory stays bounded to the nu ### Extending a stage with a quad transform -Some logic is hard to express in pure SPARQL — cleaning up messy date notations, converting locale-specific dates to ISO 8601, or sampling a reader’s output and firing follow-up queries. Rather than subclass `Reader`, attach a `QuadTransform` to it as data: a plain function `(quads, context) => quads` that post-processes one reader’s output before the stage merges it with its siblings. This is extension point 1 of [ADR 2](../../docs/decisions/0002-unify-pipeline-extension-on-quad-transforms.md). +Some logic is hard to express in pure SPARQL – cleaning up messy date notations, converting locale-specific dates to ISO 8601, or sampling a reader’s output and firing follow-up queries. Rather than subclass `Reader`, attach a `QuadTransform` to it as data: a plain function `(quads, context) => quads` that post-processes one reader’s output before the stage merges it with its siblings. This is extension point 1 of [ADR 2](../../docs/decisions/0002-unify-pipeline-extension-on-quad-transforms.md). -A transform receives an `ReaderContext` — the `dataset`, the `distribution` (so it can fire its own SPARQL queries), and the `stage` name. It runs once per reader call, so **write it to accept being called more than once**: a global stage calls it once over the reader’s complete output, but a per-class stage with batching enabled calls it once per batch (one class at `batchSize: 1`). Accumulate within an invocation, not across invocations — or keep the transform per-quad, where the number of calls makes no difference. +A transform receives an `ReaderContext` – the `dataset`, the `distribution` (so it can fire its own SPARQL queries), and the `stage` name. It runs once per reader call, so **write it to accept being called more than once**: a global stage calls it once over the reader’s complete output, but a per-class stage with batching enabled calls it once per batch (one class at `batchSize: 1`). Accumulate within an invocation, not across invocations – or keep the transform per-quad, where the number of calls makes no difference. ```typescript import { DataFactory } from 'n3'; @@ -218,7 +218,7 @@ new Stage({ #### Adaptive timeouts -By default, every SPARQL request uses the same 5-minute budget. When a pipeline runs against many third-party endpoints, that fixed budget can cost ~80 minutes on a single dataset whose endpoint times out repeatedly on heavy queries — light stages on the same endpoint then sit behind the heavy ones that will never succeed. +By default, every SPARQL request uses the same 5-minute budget. When a pipeline runs against many third-party endpoints, that fixed budget can cost ~80 minutes on a single dataset whose endpoint times out repeatedly on heavy queries – light stages on the same endpoint then sit behind the heavy ones that will never succeed. A `TimeoutPolicy` decides the budget for each SPARQL request and observes the outcome. Two are built in: @@ -252,11 +252,11 @@ Transitions are forwarded to the `ProgressReporter` via `timeoutTightened` / `ti Implement `TimeoutPolicy` directly for custom strategies (closing over shared state in the factory if you want it to span datasets). -Timeouts live at the pipeline level — neither `SparqlConstructReader` nor `SparqlItemSelector` accept their own `timeout` option. Per-endpoint state belongs in the adaptive policy, and per-stage budgets aren’t supported. Reusable stage facades (`@lde/pipeline-void`, `@lde/pipeline-shacl-sampler`) follow the same convention. +Timeouts live at the pipeline level – neither `SparqlConstructReader` nor `SparqlItemSelector` accept their own `timeout` option. Per-endpoint state belongs in the adaptive policy, and per-stage budgets aren’t supported. Reusable stage facades (`@lde/pipeline-void`, `@lde/pipeline-shacl-sampler`) follow the same convention. ### Validation -Stages can optionally validate their output quads against a `Validator`. Validation operates on the **combined output of all readers per batch**, not on individual quads or per-reader output. A batch produces a complete result set — a self-contained cluster of linked resources — that can be meaningfully matched against SHACL shapes. Even with a single reader, each batch is a complete unit; with multiple readers, shapes that reference triples from different readers are validated correctly. +Stages can optionally validate their output quads against a `Validator`. Validation operates on the **combined output of all readers per batch**, not on individual quads or per-reader output. A batch produces a complete result set – a self-contained cluster of linked resources – that can be meaningfully matched against SHACL shapes. Even with a single reader, each batch is a complete unit; with multiple readers, shapes that reference triples from different readers are validated correctly. Validating individual quads would be meaningless, since a single quad carries no structural context for shape matching. Validating the full pipeline output would also be problematic: because the pipeline streams results in batches, it doesn’t know where resource cluster boundaries fall. Batching the output could split a valid cluster across two batches, causing partial resources to fail validation even though the complete cluster is valid. @@ -291,14 +291,14 @@ new Stage({ #### Per-dataset reporting -After all stages for a dataset have run, the pipeline calls `validator.report(dataset)` once for each distinct validator attached to any stage and emits a `datasetValidated(dataset, report)` event on the reporter. The call happens **regardless of whether any stage actually invoked `validate()`** — for SHACL that means a dataset whose stages produced no input typically reports `quadsValidated: 0` and `conforms: true` (the SHACL vacuous-truth default). Consumers that want to distinguish ‘not tested’ from ‘tested and passed’ can read `quadsValidated`. +After all stages for a dataset have run, the pipeline calls `validator.report(dataset)` once for each distinct validator attached to any stage and emits a `datasetValidated(dataset, report)` event on the reporter. The call happens **regardless of whether any stage actually invoked `validate()`** – for SHACL that means a dataset whose stages produced no input typically reports `quadsValidated: 0` and `conforms: true` (the SHACL vacuous-truth default). Consumers that want to distinguish ‘not tested’ from ‘tested and passed’ can read `quadsValidated`. ### Writer Writes generated quads to a destination. A `Writer` is transactional: each pipeline run opens one run on it (`openRun(context)`), writes every dataset through the resulting `RunWriter`, and ends with exactly one `commit()` (on success) or `abort(error)` (on failure). The run lifecycle is the home of destination-level concerns such as atomic swaps, deletion sweeps and cross-pod locks; the pipeline drives `openRun → write* → commit/abort` uniformly and never branches on the writer’s update mode. After a dataset’s stages complete, its `flush(dataset, outcome)` says whether the dataset succeeded, so a writer can gate destructive finalization (e.g. an In-place stale-document sweep) on `'success'`. -- `SparqlUpdateWriter` — writes to a SPARQL endpoint via UPDATE queries; writes are visible as they land, so `commit`/`abort` are no-ops -- `FileWriter` — writes to local files; each file is streamed to a sibling temp file and atomically renamed on flush, `commit` finalizes any files still open, and `abort` discards temp output. In `n-quads` format it skolemises blank nodes to dataset-scoped IRIs, because such files are typically `cat`-indexed into one served graph where document-scoped blank-node labels from different datasets would otherwise fuse (see `blankNodes`/`failOnBlankNodes` and ldelements/lde#474); Turtle/N-Triples output is left unchanged +- `SparqlUpdateWriter` – writes to a SPARQL endpoint via UPDATE queries; writes are visible as they land, so `commit`/`abort` are no-ops +- `FileWriter` – writes to local files; each file is streamed to a sibling temp file and atomically renamed on flush, `commit` finalizes any files still open, and `abort` discards temp output. In `n-quads` format it skolemises blank nodes to dataset-scoped IRIs, because such files are typically `cat`-indexed into one served graph where document-scoped blank-node labels from different datasets would otherwise fuse (see `blankNodes`/`failOnBlankNodes` and ldelements/lde#474); Turtle/N-Triples output is left unchanged The `RunContext` handed to `openRun` carries the run’s identity (`runId`, `startedAt`), the full selection (`selectedSources()`, complete by commit time – including datasets skipped as unchanged) and the pipeline’s provenance store, when configured. @@ -320,7 +320,7 @@ async openRun(): Promise { A `ProgressReporter` observes the run, receiving lifecycle events such as `pipelineStart`, `stageComplete`, `datasetValidated` and `pipelineComplete`. Every method is optional, so a reporter implements only the events it cares about. -Pass a single reporter, or an array to have several observe the same run — for example a console reporter alongside one that collects validation verdicts: +Pass a single reporter, or an array to have several observe the same run – for example a console reporter alongside one that collects validation verdicts: ```typescript new Pipeline({ @@ -342,11 +342,11 @@ interface ProvenanceStore { } ``` -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. +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` -The reference implementation targets a triplestore that is served read-only and rebuilt by bulk-loading files (e.g. [QLever](https://github.com/ad-freiburg/qlever)). It reads through SPARQL queries against the live endpoint, and writes records as files for the next bulk-load — because the endpoint accepts no SPARQL UPDATE. +The reference implementation targets a triplestore that is served read-only and rebuilt by bulk-loading files (e.g. [QLever](https://github.com/ad-freiburg/qlever)). It reads through SPARQL queries against the live endpoint, and writes records as files for the next bulk-load – because the endpoint accepts no SPARQL UPDATE. ```typescript import { FileLoadedSparqlProvenanceStore } from '@lde/pipeline'; @@ -361,7 +361,19 @@ const store = new FileLoadedSparqlProvenanceStore({ - `get` runs a named-graph-scoped SPARQL `SELECT` against `queryEndpoint`, reading the records a previous run loaded. - `set` writes one flat [PROV-O](https://www.w3.org/TR/prov-o/) N-Quads file per dataset into `outputDir`, in the pipeline-scoped named graph, to be bulk-loaded after the run. -Each record is stored as flat PROV-O on the dataset entity — `prov:generatedAtTime` plus `sourceFingerprint`, `pipelineVersion` and `status` under the `https://w3id.org/lde/provenance#` namespace. Scoping every record by `pipelineIri` (used as the named graph) lets multiple pipelines share one triplestore without colliding. +Each record is stored as flat PROV-O on the dataset entity – `prov:generatedAtTime` plus `sourceFingerprint`, `pipelineVersion` and `status` under the `https://w3id.org/lde/provenance#` namespace. Scoping every record by `pipelineIri` (used as the named graph) lets multiple pipelines share one triplestore without colliding. + +#### `FileProvenanceStore` + +For pipelines that run without a triplestore, `FileProvenanceStore` persists all records to a single JSON file, keyed by dataset URI: + +```typescript +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. #### Enabling skipping @@ -382,11 +394,11 @@ skip iff recorded.sourceFingerprint === current.sourceFingerprint AND recorded.pipelineVersion === current.pipelineVersion ``` -Otherwise it imports (if needed), runs the stages, and writes an updated record. `pipelineVersion` is consumer-owned and opaque: rotate it only on releases that change output, and every dataset reprocesses on the next run. It is **required** when a `provenanceStore` is configured (a skip-enabled pipeline with no version would silently freeze); when no store is configured, every dataset is reprocessed — today’s behaviour. A dataset that failed but whose source is unchanged is recorded as `'failed'` and skipped on later runs until its source changes or the version rotates, so a deterministically failing import is not retried every run. +Otherwise it imports (if needed), runs the stages, and writes an updated record. `pipelineVersion` is consumer-owned and opaque: rotate it only on releases that change output, and every dataset reprocesses on the next run. It is **required** when a `provenanceStore` is configured (a skip-enabled pipeline with no version would silently freeze); when no store is configured, every dataset is reprocessed – today’s behaviour. A dataset that failed but whose source is unchanged is recorded as `'failed'` and skipped on later runs until its source changes or the version rotates, so a deterministically failing import is not retried every run. ### Source-change fingerprint -`sourceFingerprint(distribution, probeResult)` derives a cheap, opaque change signal for a distribution from metadata the probe already collected — no body download. For a data dump it combines the most recent of the register’s `dct:modified` and the artifact’s HTTP `Last-Modified` with the byte size (the probe’s `Content-Length`, falling back to the declared `dcat:byteSize`). It returns `null` for a live SPARQL endpoint, or when no date and no size can be established — a `null` fingerprint never compares equal, so such a distribution is always reprocessed. +`sourceFingerprint(distribution, probeResult)` derives a cheap, opaque change signal for a distribution from metadata the probe already collected – no body download. For a data dump it combines the most recent of the register’s `dct:modified` and the artifact’s HTTP `Last-Modified` with the byte size (the probe’s `Content-Length`, falling back to the declared `dcat:byteSize`). It returns `null` for a live SPARQL endpoint, or when no date and no size can be established – a `null` fingerprint never compares equal, so such a distribution is always reprocessed. ### Plugins diff --git a/packages/pipeline/src/provenance/fileProvenanceStore.ts b/packages/pipeline/src/provenance/fileProvenanceStore.ts new file mode 100644 index 00000000..cd8292ab --- /dev/null +++ b/packages/pipeline/src/provenance/fileProvenanceStore.ts @@ -0,0 +1,70 @@ +import { mkdir, readFile, rename, writeFile } from 'node:fs/promises'; +import { dirname } from 'node:path'; +import type { ProcessingRecord } from './record.js'; +import type { ProvenanceStore } from './store.js'; + +export interface FileProvenanceStoreOptions { + /** + * Path of the JSON file the records are persisted to. Parent directories are + * created on the first write. Must sit on a durable volume to survive across + * runs (a Kubernetes CronJob’s container filesystem is discarded). + */ + path: string; +} + +/** + * A {@link ProvenanceStore} that persists all records to a single JSON file, + * keyed by dataset URI. + * + * For pipelines that run without a triplestore – e.g. a single CronJob + * persisting to a mounted volume – a JSON file is far lighter than standing + * up a SPARQL server purely to remember processing records. + * + * 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 only: + * concurrent pipeline processes writing the same file lose each other’s + * updates – use {@link FileLoadedSparqlProvenanceStore} (or file locking) + * instead. Every access reads the whole file into memory; records are tiny, + * but at very large dataset counts prefer the SPARQL-backed store. + */ +export class FileProvenanceStore implements ProvenanceStore { + private readonly path: string; + + constructor(options: FileProvenanceStoreOptions) { + this.path = options.path; + } + + async get(datasetUri: URL): Promise { + const records = await this.readAll(); + return records[datasetUri.toString()] ?? null; + } + + async set(datasetUri: URL, record: ProcessingRecord): Promise { + const records = await this.readAll(); + records[datasetUri.toString()] = record; + await this.writeAll(records); + } + + private async readAll(): Promise> { + let json: string; + try { + json = await readFile(this.path, 'utf8'); + } catch (error) { + // A missing file is the empty store: nothing has been processed yet. + if ((error as NodeJS.ErrnoException).code === 'ENOENT') { + return {}; + } + throw error; + } + return JSON.parse(json) as Record; + } + + private async writeAll( + records: Record, + ): Promise { + await mkdir(dirname(this.path), { recursive: true }); + const temporaryPath = `${this.path}.${process.pid}.tmp`; + await writeFile(temporaryPath, `${JSON.stringify(records, null, 2)}\n`); + await rename(temporaryPath, this.path); // Atomic on POSIX filesystems. + } +} diff --git a/packages/pipeline/src/provenance/index.ts b/packages/pipeline/src/provenance/index.ts index d3d3781b..355274c8 100644 --- a/packages/pipeline/src/provenance/index.ts +++ b/packages/pipeline/src/provenance/index.ts @@ -6,3 +6,7 @@ export { FileLoadedSparqlProvenanceStore, type FileLoadedSparqlProvenanceStoreOptions, } from './fileLoadedSparqlProvenanceStore.js'; +export { + FileProvenanceStore, + type FileProvenanceStoreOptions, +} from './fileProvenanceStore.js'; diff --git a/packages/pipeline/test/provenance/fileProvenanceStore.test.ts b/packages/pipeline/test/provenance/fileProvenanceStore.test.ts new file mode 100644 index 00000000..a91907e5 --- /dev/null +++ b/packages/pipeline/test/provenance/fileProvenanceStore.test.ts @@ -0,0 +1,129 @@ +import { describe, it, expect, beforeEach, afterEach } from 'vitest'; +import { 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'; +import { FileProvenanceStore } from '../../src/index.js'; + +const DATASET_URI = new URL('http://example.org/dataset/1'); + +const RECORD: ProcessingRecord = { + sourceFingerprint: '2024-06-01T00:00:00.000Z|1000', + pipelineVersion: 'v1', + generatedAt: '2026-06-11T00:00:00.000Z', + status: 'success', +}; + +describe('FileProvenanceStore', () => { + let directory: string; + let path: string; + + beforeEach(async () => { + directory = await mkdtemp(join(tmpdir(), 'lde-file-provenance-')); + path = join(directory, 'provenance.json'); + }); + + afterEach(async () => { + await rm(directory, { recursive: true, force: true }); + }); + + it('returns null when the file does not exist yet', async () => { + const store = new FileProvenanceStore({ path }); + + expect(await store.get(DATASET_URI)).toBeNull(); + }); + + it('round-trips a record', async () => { + const store = new FileProvenanceStore({ path }); + + await store.set(DATASET_URI, RECORD); + + expect(await store.get(DATASET_URI)).toEqual(RECORD); + }); + + it('returns null for a dataset with no record', async () => { + const store = new FileProvenanceStore({ path }); + + await store.set(DATASET_URI, RECORD); + + expect(await store.get(new URL('http://example.org/absent'))).toBeNull(); + }); + + it('replaces the previous record for the same dataset', async () => { + const store = new FileProvenanceStore({ path }); + + await store.set(DATASET_URI, RECORD); + const updated: ProcessingRecord = { + ...RECORD, + pipelineVersion: 'v2', + status: 'failed', + }; + await store.set(DATASET_URI, updated); + + expect(await store.get(DATASET_URI)).toEqual(updated); + }); + + it('keeps records for other datasets when writing one', async () => { + const store = new FileProvenanceStore({ path }); + const otherUri = new URL('http://example.org/dataset/2'); + const otherRecord: ProcessingRecord = { + ...RECORD, + sourceFingerprint: null, + status: 'failed', + }; + + await store.set(DATASET_URI, RECORD); + await store.set(otherUri, otherRecord); + + expect(await store.get(DATASET_URI)).toEqual(RECORD); + expect(await store.get(otherUri)).toEqual(otherRecord); + }); + + it('persists across store instances', async () => { + await new FileProvenanceStore({ path }).set(DATASET_URI, RECORD); + + expect(await new FileProvenanceStore({ path }).get(DATASET_URI)).toEqual( + RECORD, + ); + }); + + it('creates missing parent directories on write', async () => { + const nestedPath = join(directory, 'state', 'nested', 'provenance.json'); + const store = new FileProvenanceStore({ path: nestedPath }); + + await store.set(DATASET_URI, RECORD); + + expect(await store.get(DATASET_URI)).toEqual(RECORD); + }); + + it('leaves no temp file behind after a write', async () => { + await new FileProvenanceStore({ path }).set(DATASET_URI, RECORD); + + expect(await readdir(directory)).toEqual(['provenance.json']); + }); + + it('writes human-readable JSON keyed by dataset URI', async () => { + await new FileProvenanceStore({ path }).set(DATASET_URI, RECORD); + + const contents = await readFile(path, 'utf8'); + expect(JSON.parse(contents)).toEqual({ [DATASET_URI.toString()]: RECORD }); + // Pretty-printed with a trailing newline, so the file diffs cleanly. + expect(contents).toContain('\n '); + expect(contents.endsWith('\n')).toBe(true); + }); + + it('propagates read errors other than a missing file', async () => { + // Reading a directory fails with EISDIR, not ENOENT, so it must throw. + const store = new FileProvenanceStore({ path: directory }); + + await expect(store.get(DATASET_URI)).rejects.toThrow(); + }); + + it('propagates corruption instead of silently starting over', async () => { + await writeFile(path, 'not json'); + const store = new FileProvenanceStore({ path }); + + await expect(store.get(DATASET_URI)).rejects.toThrow(); + await expect(store.set(DATASET_URI, RECORD)).rejects.toThrow(); + }); +}); diff --git a/packages/pipeline/vite.config.ts b/packages/pipeline/vite.config.ts index 60281770..c9518c83 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.38, - lines: 97.04, - branches: 91.73, - statements: 96.58, + functions: 97.43, + lines: 97.09, + branches: 91.79, + statements: 96.63, }, }, },