feat(pipeline)!: transaction-aware Writer and the Executor→Reader rename

README.md

@@ -39,7 +39,7 @@ Data transformations are expressed as plain SPARQL queries: portable, transparen
 import {
   Pipeline,
   Stage,
-  SparqlConstructExecutor,
+  SparqlConstructReader,
   SparqlItemSelector,
   SparqlUpdateWriter,
   ManualDatasetSelection,
@@ -53,7 +53,7 @@ const pipeline = new Pipeline({
       itemSelector: new SparqlItemSelector({
         query: 'SELECT DISTINCT ?class WHERE { ?s a ?class }',
       }),
-      executors: new SparqlConstructExecutor({
+      readers: new SparqlConstructReader({
         query:
           'CONSTRUCT { ?class a <http://example.org/Class> } WHERE { ?s a ?class }',
       }),

docs/decisions/0006-make-the-writer-transaction-aware.md

@@ -0,0 +1,102 @@
+# 6. Make the Writer transaction-aware
+
+Date: 2026-07-06
+
+## Status
+
+Proposed
+
+Amends the pipeline model of
+[ADR 1 (Merge pipeline approaches)](./0001-merge-pipeline-approaches.md) and
+[ADR 2 (Unify pipeline extension on quad transforms)](./0002-unify-pipeline-extension-on-quad-transforms.md);
+groundwork for search as a Configurable Pipeline instance
+([#534](https://github.com/ldelements/lde/issues/534)).
+
+## Context
+
+The pipeline’s `Writer` was a per-dataset sink (`write`/`flush`/`reset`)
+with no notion of a run. Run-level lifecycle had no home: a search index
+rebuild needs an atomic alias swap after all datasets are written
+(Blue/green Rebuild), an in-place object index needs a deletion sweep over
+everything the run did not touch plus a registry-membership sweep
+(In-place Rebuild), and both need a single-flight cross-pod lock. The
+existing `@lde/search-typesense` `rebuild()` bundles load + swap + lock in
+a one-shot function outside the pipeline; QLever bulk-load + dir-swap has
+the same shape. Without a run boundary on the writer, each consumer
+rebuilds that lifecycle bespoke, or the pipeline would have to branch on
+the destination’s update mode.
+
+The input unit was named `Executor`, which says nothing about its role.
+
+## Decision
+
+### Reader / Writer as the pipeline’s I/O pair
+
+Rename `Executor` → `Reader` (and `execute()` → `read()`): a `Reader`
+produces quads for a dataset from outside the pipeline – a SPARQL endpoint,
+a dump, or a non-RDF facade. `Reader`/`Writer` map to the NDE Stack’s
+conceptual `source`/`sink`; we use the more familiar I/O terms. Only the
+input renames – `Writer`, `FileWriter` and `SparqlUpdateWriter` keep their
+names.
+
+### One run, one transaction
+
+`Writer` becomes a factory of per-run transactions, generic over its
+payload (`Writer<Item = Quad>` – KG writers stream quads unchanged; search
+writers will consume pre-framed documents):
+
+```typescript
+interface Writer<Item = Quad> {
+  openRun(context: RunContext): Promise<RunWriter<Item>>;
+}
+interface RunWriter<Item = Quad> extends DatasetWriter<Item> {
+  flush?(dataset: Dataset): Promise<void>; // per-dataset finalize
+  reset?(dataset: Dataset): Promise<void>; // discard a dataset’s pass
+  commit(): Promise<void>; // swap / sweep / release lock
+  abort(error: unknown): Promise<void>; // leave the live destination as-was
+}
+```
+
+`Pipeline.run` drives `openRun → write* → commit/abort` uniformly and
+never branches on the writer’s update mode: an alias swap is a private
+step of a Blue/green writer’s `commit()`, a sweep of an In-place writer’s.
+Stages write through the narrower `DatasetWriter` (just `write`), so a
+stage can never commit or abort the run.
+
+### RunContext
+
+`openRun` receives the run’s identity and selection scope: `runId` (stamps
+`last_seen` for In-place, names the blue collection for Blue/green),
+`startedAt` (injected – writers need no clock), `selectedSources()` (every
+selected dataset IRI **including ones skipped as unchanged**, complete by
+commit time – the input to a registry-membership sweep), and the
+pipeline’s `provenance` store when skip-unchanged is enabled.
+
+### One interface, no second path
+
+A destination without run-level state implements the same contract with
+no-op `commit`/`abort` (a five-line `openRun`; `SparqlUpdateWriter` shows
+the pattern). There is deliberately **no** second `TransactionalWriter`
+interface – it would reintroduce the dual code path the single contract
+exists to remove – and no wrapper helper either: every in-repo destination
+turned out to want real lifecycle behaviour, so a helper had no consumers
+and was dropped for a minimal API surface.
+
+## Consequences
+
+- Breaking for `@lde/pipeline` consumers: `Executor` call sites rename, and
+  custom writers implement `openRun` (with no-op `commit`/`abort` when they
+  have no run-level state). The Dataset Knowledge Graph pipeline adapts on
+  its next upgrade.
+- `FileWriter` and `SparqlUpdateWriter` are transactional: per-run state
+  (open files, cleared graphs) lives in the run, so re-running a pipeline
+  on the same writer instance replaces output instead of appending –
+  previously a latent bug. `FileWriter.commit` finalizes files still open;
+  `abort` discards temp output, never leaving a truncated final file.
+- Chained (sub-stage) scratch `FileWriter`s are now flushed before their
+  output is resolved – previously the chain read a final path that only
+  materialized on a flush nobody issued.
+- The search writers (`BlueGreenRebuild`, `InPlaceRebuild` in
+  `@lde/search-<engine>`) get a home for load + finalize as one atomic
+  unit, replacing the one-shot `rebuild()`; the two update modes become
+  `Writer` implementations, not pipeline branches.

packages/pipeline-shacl-sampler/src/sampleStages.ts

@@ -1,6 +1,6 @@
 import {
   Stage,
-  SparqlConstructExecutor,
+  SparqlConstructReader,
   SparqlItemSelector,
   type ItemSelector,
   type StageOptions,
@@ -51,13 +51,13 @@ export interface ShaclSampleStagesOptions {
    */
   samplesPerClass?: number;
   /**
-   * Maximum number of sampled subjects per executor call. Defaults to
+   * Maximum number of sampled subjects per reader call. Defaults to
    * {@link samplesPerClass} so the whole sample fits in one CONSTRUCT
    * round-trip; lower to spread work across multiple parallel queries.
    */
   batchSize?: number;
   /**
-   * Maximum concurrent in-flight executor batches per stage. @default 10
+   * Maximum concurrent in-flight reader batches per stage. @default 10
    */
   maxConcurrency?: number;
   /**
@@ -105,7 +105,7 @@ export interface ShaclSampleStagesOptions {
 /**
  * Build one sampling {@link Stage} per `sh:targetClass` declared in the SHACL
  * shapes file. Each stage pairs a SELECT-based {@link ItemSelector} that picks
- * N instances of its target class with a CONSTRUCT executor that, for every
+ * N instances of its target class with a CONSTRUCT reader that, for every
  * path chain the SHACL declares (recursively, stopping at leaf constraints
  * or cycles), pulls in the triples reachable along that chain’s terminal
  * node.
@@ -145,7 +145,7 @@ export async function shaclSampleStages(
           namespaceAliases,
           options.excludeResources?.(shape.targetClass),
         ),
-        executors: new SparqlConstructExecutor({
+        readers: new SparqlConstructReader({
           query: buildSampleQuery(shape),
         }),
         batchSize,

packages/pipeline-shacl-validator/README.md

@@ -13,7 +13,7 @@ RDF serialization (Turtle, JSON-LD, N-Triples etc.).
 import {
   Pipeline,
   Stage,
-  SparqlConstructExecutor,
+  SparqlConstructReader,
   FileWriter,
   SparqlUpdateWriter,
 } from '@lde/pipeline';
@@ -32,7 +32,7 @@ const pipeline = new Pipeline({
   stages: [
     new Stage({
       name: 'transform',
-      executors: new SparqlConstructExecutor({ query: '...' }),
+      readers: new SparqlConstructReader({ query: '...' }),
       validation: {
         validator,
         onInvalid: 'write', // 'write' | 'skip' | 'halt'

packages/pipeline-shacl-validator/src/shacl-validator.ts

@@ -1,10 +1,12 @@
+import { randomUUID } from 'node:crypto';
 import type { Quad } from '@rdfjs/types';
 import type { Dataset } from '@lde/dataset';
 
 import type {
   Validator,
   ValidationResult,
   ValidationReport,
+  RunWriter,
   Writer,
 } from '@lde/pipeline';
 // @ts-expect-error -- shacl-engine has no type declarations.
@@ -19,9 +21,10 @@ export interface ShaclValidatorOptions {
   /** Path to an RDF file containing SHACL shapes (any format supported by rdf-dereference). */
   shapesFile: string;
   /**
-   * Writers that receive the per-dataset SHACL validation report quads. Each
-   * batch with violations is streamed to every writer via {@link Writer.write};
-   * each writer's {@link Writer.flush} is called from {@link ShaclValidator.report}.
+   * Writers that receive the per-dataset SHACL validation report quads. The
+   * validator opens one run per writer (lazily, on the first violation) and
+   * streams each batch with violations to it; each run is flushed per dataset
+   * from {@link ShaclValidator.report}.
    *
    * Pass a {@link FileWriter} to mirror the previous on-disk behaviour, a
    * {@link SparqlUpdateWriter} to land reports in a named graph, or any custom
@@ -46,6 +49,7 @@ interface DatasetAccumulator {
 export class ShaclValidator implements Validator {
   private readonly shapesFile: string;
   private readonly reportWriters: Writer[];
+  private reportRuns: RunWriter[] | undefined;
   // eslint-disable-next-line @typescript-eslint/no-explicit-any
   private shapesDataset: any | undefined;
   private readonly accumulators = new Map<string, DatasetAccumulator>();
@@ -91,8 +95,8 @@ export class ShaclValidator implements Validator {
         report.dataset,
         dataset.iri.toString(),
       );
-      for (const writer of this.reportWriters) {
-        await writer.write(dataset, asyncIterableOf(reportQuads));
+      for (const run of await this.runs()) {
+        await run.write(dataset, asyncIterableOf(reportQuads));
       }
     }
 
@@ -107,8 +111,10 @@ export class ShaclValidator implements Validator {
   }
 
   async report(dataset: Dataset): Promise<ValidationReport> {
-    for (const writer of this.reportWriters) {
-      await writer.flush?.(dataset);
+    // Flush is the per-dataset completion hook regardless of whether the
+    // dataset produced violations, so open the runs here if no write did.
+    for (const run of await this.runs()) {
+      await run.flush?.(dataset);
     }
 
     const key = dataset.iri.toString();
@@ -123,6 +129,26 @@ export class ShaclValidator implements Validator {
     };
   }
 
+  /**
+   * The report writers' runs, opened lazily on first use. The report stream is
+   * the validator's own long-lived run: each dataset's report is finalized by
+   * the per-dataset flush in {@link report}, so the run needs no commit, and
+   * the fabricated context's selection stays empty (report writers do not
+   * sweep by membership).
+   */
+  private async runs(): Promise<RunWriter[]> {
+    this.reportRuns ??= await Promise.all(
+      this.reportWriters.map((writer) =>
+        writer.openRun({
+          runId: randomUUID(),
+          startedAt: new Date().toISOString(),
+          selectedSources: () => [],
+        }),
+      ),
+    );
+    return this.reportRuns;
+  }
+
   // eslint-disable-next-line @typescript-eslint/no-explicit-any
   private async getShapes(): Promise<any> {
     if (!this.shapesDataset) {