feat(sparql-anything): report each chunk as it is converted

docs/reference/sparql-anything.md

@@ -41,14 +41,15 @@ await converter.convert(
 
 ### Options
 
-| Option        | Type               | Description                                                                                                                   |
-| ------------- | ------------------ | ----------------------------------------------------------------------------------------------------------------------------- |
-| `jarPath`     | `string`           | Path to the SPARQL Anything CLI jar, as the task runner sees it                                                               |
-| `workDir`     | `string`           | The task runner's working directory; see [Where files are written](#where-files-are-written)                                  |
-| `heap`        | `string`           | Maximum JVM heap per chunk process, as `-Xmx` takes it (default `'2g'`); see [Memory](#memory)                                |
-| `cliArgs`     | `string[]`         | Further arguments for the SPARQL Anything CLI; see [Memory](#memory)                                                          |
-| `concurrency` | `number`           | How many chunks to convert at once (default `1`); see [Converting several chunks at once](#converting-several-chunks-at-once) |
-| `taskRunner`  | `TaskRunner<Task>` | Runs the SPARQL Anything process for each chunk                                                                               |
+| Option             | Type                 | Description                                                                                                                   |
+| ------------------ | -------------------- | ----------------------------------------------------------------------------------------------------------------------------- |
+| `jarPath`          | `string`             | Path to the SPARQL Anything CLI jar, as the task runner sees it                                                               |
+| `workDir`          | `string`             | The task runner's working directory; see [Where files are written](#where-files-are-written)                                  |
+| `heap`             | `string`             | Maximum JVM heap per chunk process, as `-Xmx` takes it (default `'2g'`); see [Memory](#memory)                                |
+| `cliArgs`          | `string[]`           | Further arguments for the SPARQL Anything CLI; see [Memory](#memory)                                                          |
+| `concurrency`      | `number`             | How many chunks to convert at once (default `1`); see [Converting several chunks at once](#converting-several-chunks-at-once) |
+| `onChunkConverted` | `(progress) => void` | Called as each chunk finishes; see [Following a conversion](#following-a-conversion)                                          |
+| `taskRunner`       | `TaskRunner<Task>`   | Runs the SPARQL Anything process for each chunk                                                                               |
 
 ### Jobs
 
@@ -101,6 +102,19 @@ The first failure aborts the run: no further chunk is started, and the processes
 
 `cliArgs` is the escape hatch for CLI flags the converter does not model itself, appended to the arguments it sets. It cannot repeat those: `-q`, `-f`, `-o` and `-l` are rejected, in their long and `--flag=value` forms too, because the converter reads back the `--output` it named, in the `--format` it asked for – overriding either leaves it reporting an empty conversion, or concatenating fragments that are not N-Triples. SPARQL Anything documents repetition only for `-v` and `-c`, so a repeated flag has no defined winner to rely on.
 
+### Following a conversion
+
+A conversion says nothing for as long as it takes – a quarter of an hour, over eighteen chunks, for the GeoNames dumps. `onChunkConverted` is called as each one finishes:
+
+```typescript
+onChunkConverted: ({ index, total, chunk }) =>
+  console.log(`Converted ${index}/${total}${chunk === undefined ? '' : `: ${chunk}`}`),
+```
+
+It is called once per chunk, in the order they finish rather than the order they were given, and not at all for a chunk that failed – a failure arrives as the rejection instead. A callback that throws aborts the run, like any other failure.
+
+It is wiring rather than configuration: a runtime that already has somewhere to report progress passes a function that forwards to it, which is why this is a plain callback rather than a reporter interface of its own.
+
 ## Chunking
 
 `convert()` takes chunks it does not create, because a caller often has work to do first – filtering rows out, say. `chunk()` produces them:

packages/sparql-anything/src/index.ts

@@ -1,6 +1,7 @@
 export { chunk, type ChunkOptions } from './chunk.js';
 export {
   SparqlAnythingConverter,
+  type ChunkProgress,
   type ConversionJob,
   type SparqlAnythingConverterOptions,
 } from './sparql-anything-converter.js';

packages/sparql-anything/src/sparql-anything-converter.ts

@@ -58,6 +58,18 @@ export interface ConversionJob {
   load?: string;
 }
 
+/** What a chunk's conversion reports when it is done. */
+export interface ChunkProgress {
+  /** Position of this process in the run, counting from one. */
+  index: number;
+  /** How many processes the run holds in all. */
+  total: number;
+  /** The chunk converted, for a job that has chunks. */
+  chunk?: string;
+  /** The query the job ran, which is what tells two jobs apart. */
+  queryFile: string;
+}
+
 /** Configuration for a {@link SparqlAnythingConverter}. */
 export interface SparqlAnythingConverterOptions<Task> {
   /** Path to the SPARQL Anything CLI jar, as the task runner sees it. */
@@ -94,6 +106,14 @@ export interface SparqlAnythingConverterOptions<Task> {
   concurrency?: number;
   /** Runs the SPARQL Anything process for each chunk. */
   taskRunner: TaskRunner<Task>;
+  /**
+   * Called as each chunk finishes, for a conversion that would otherwise say
+   * nothing for as long as it takes – the GeoNames run is a quarter of an hour
+   * over eighteen chunks. Called once per chunk, in the order they finish
+   * rather than the order they were given, and not at all for a chunk that
+   * failed. A callback that throws aborts the run, like any other failure.
+   */
+  onChunkConverted?: (progress: ChunkProgress) => void;
 }
 
 /**
@@ -108,6 +128,7 @@ export class SparqlAnythingConverter<Task> {
   private readonly cliArgs: string[];
   private readonly concurrency: number;
   private readonly taskRunner: TaskRunner<Task>;
+  private readonly onChunkConverted?: (progress: ChunkProgress) => void;
 
   constructor(options: SparqlAnythingConverterOptions<Task>) {
     this.jarPath = options.jarPath;
@@ -137,6 +158,7 @@ export class SparqlAnythingConverter<Task> {
     }
     this.concurrency = concurrency;
     this.taskRunner = options.taskRunner;
+    this.onChunkConverted = options.onChunkConverted;
   }
 
   /**
@@ -189,7 +211,10 @@ export class SparqlAnythingConverter<Task> {
     runDirName: string,
   ): Promise<number> {
     const pending = processesOf(planned);
-    const state: RunState<Task> = { inFlight: new Set() };
+    const state: RunState<Task> = {
+      total: countOf(planned),
+      inFlight: new Set(),
+    };
 
     // Pulled one at a time rather than with `for...of`: leaving a for-of early
     // closes the iterator, so the first worker to give up would end the queue
@@ -216,7 +241,7 @@ export class SparqlAnythingConverter<Task> {
     if (state.failure !== undefined) {
       throw state.failure;
     }
-    return countOf(planned);
+    return state.total;
   }
 
   /** Converts one chunk, writing `output-<index>.nt` in the run directory. */
@@ -253,6 +278,12 @@ export class SparqlAnythingConverter<Task> {
       state.inFlight.delete(task);
     }
     await assertNonEmpty(join(this.workDir, output), job, chunk);
+    this.onChunkConverted?.({
+      index: index + 1,
+      total: state.total,
+      chunk,
+      queryFile: job.queryFile,
+    });
   }
 
   /** Stops every process still running, so none outlives the run. */
@@ -329,6 +360,8 @@ function countOf(planned: PlannedJob[]): number {
 
 /** What the workers of one run share. */
 interface RunState<Task> {
+  /** How many processes the run holds, for what reports progress. */
+  total: number;
   /** Tasks that have been started and not yet finished. */
   inFlight: Set<Task>;
   /** The first failure, which aborts the run. */

packages/sparql-anything/test/sparql-anything-converter.test.ts

@@ -1,4 +1,8 @@
-import { ConversionJob, SparqlAnythingConverter } from '../src/index.js';
+import {
+  ChunkProgress,
+  ConversionJob,
+  SparqlAnythingConverter,
+} from '../src/index.js';
 import { TaskRunner } from '@lde/task-runner';
 import { describe, it, expect, beforeEach, afterEach } from 'vitest';
 import {
@@ -620,6 +624,55 @@ describe('SparqlAnythingConverter', () => {
     ).toThrow('is not a number of chunks to convert at once');
   });
 
+  it('reports each chunk as it finishes', async () => {
+    const taskRunner = new FakeTaskRunner(workDir);
+    const chunks = await writeChunks(2);
+    const ontologyQuery = join(workDir, 'ontology.rq');
+    await writeFile(ontologyQuery, 'CONSTRUCT { ?s ?p ?o } WHERE { ?s ?p ?o }');
+    const reported: ChunkProgress[] = [];
+
+    await new SparqlAnythingConverter({
+      jarPath: '/bin/sparql-anything.jar',
+      workDir,
+      taskRunner,
+      onChunkConverted: (progress) => reported.push(progress),
+    }).convert(
+      [
+        { queryFile, chunks },
+        { queryFile: ontologyQuery, load: '/data/ontology.rdf' },
+      ],
+      join(workDir, 'output.nt'),
+    );
+
+    expect(reported).toEqual([
+      { index: 1, total: 3, chunk: chunks[0], queryFile },
+      { index: 2, total: 3, chunk: chunks[1], queryFile },
+      // A job without chunks reports too, with none to name.
+      { index: 3, total: 3, chunk: undefined, queryFile: ontologyQuery },
+    ]);
+  });
+
+  it('does not report a chunk that failed', async () => {
+    const chunks = await writeChunks(2);
+    const taskRunner = new FakeTaskRunner(workDir, {
+      failOutputContaining: 'output-1.nt',
+    });
+    const reported: ChunkProgress[] = [];
+
+    await expect(
+      new SparqlAnythingConverter({
+        jarPath: '/bin/sparql-anything.jar',
+        workDir,
+        taskRunner,
+        onChunkConverted: (progress) => reported.push(progress),
+      }).convert([{ queryFile, chunks }], join(workDir, 'output.nt')),
+    ).rejects.toThrow('Process failed');
+
+    expect(reported).toEqual([
+      { index: 1, total: 2, chunk: chunks[0], queryFile },
+    ]);
+  });
+
   it('refuses an empty job list rather than writing an empty output', async () => {
     const taskRunner = new FakeTaskRunner(workDir);
     const outputPath = join(workDir, 'output.nt');