From 5086c30af294bba46d0a6808b520e6a0f4f78bfd Mon Sep 17 00:00:00 2001 From: David de Boer Date: Mon, 31 Aug 2026 16:05:42 +0200 Subject: [PATCH 1/2] feat(sparql-anything): convert several chunks at once MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit map.sh runs an xargs -P pool over its chunks; the converter ran them one at a time, which on allCountries multiplies a job that already takes about a day. concurrency says how many to convert at once, over the chunks of every job in one pool, so a long places chunk and a short alternate-names one pack together rather than draining in phases. The output is concatenated by position, not by completion, so it stays in the order the jobs and their chunks were given. The first failure aborts the run: no further chunk is started, and the processes still going are stopped rather than left writing into a directory convert() is about to delete. The default is 1, and deliberately not derived from the CPU count or a memory limit. Each chunk is a JVM capped by `heap`, so a pool needs concurrency × heap on the machine the TASK RUNNER uses – which is not this process's machine once the runner is Docker or remote. The converter cannot see that machine, so the number belongs to the caller, who can do map.sh's arithmetic for the deployment they actually have. Workers pull from the queue one at a time rather than sharing a for...of over it: leaving a for-of early closes the iterator, so the first worker to give up would silently end the queue for the others. --- docs/reference/sparql-anything.md | 25 ++- .../src/sparql-anything-converter.ts | 158 +++++++++++++++--- .../test/sparql-anything-converter.test.ts | 110 +++++++++++- 3 files changed, 254 insertions(+), 39 deletions(-) diff --git a/docs/reference/sparql-anything.md b/docs/reference/sparql-anything.md index 6887d23f..b95bd8a4 100644 --- a/docs/reference/sparql-anything.md +++ b/docs/reference/sparql-anything.md @@ -41,13 +41,14 @@ 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) | -| `taskRunner` | `TaskRunner` | 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) | +| `taskRunner` | `TaskRunner` | Runs the SPARQL Anything process for each chunk | ### Jobs @@ -85,6 +86,16 @@ heap: '4g', // -Xmx4g Size it with the chunk size. A chunk that outgrows the heap fails loudly – the JVM's `OutOfMemoryError` arrives in the output of a non-zero exit, which aborts the conversion – where an uncapped JVM instead grows until the OOM killer takes the whole container. +### Converting several chunks at once + +`concurrency` is how many chunks are converted at the same time. Each one is a JVM of its own, so it multiplies against `heap`: a run needs `concurrency × heap`, on the machine the **task runner** uses – which is not this process's machine when the runner is Docker or remote. + +That is why the default is `1` rather than something derived from the CPU count or a memory limit: the converter cannot see the machine its processes run on, so the number is the caller's to choose. `map.sh` sizes its pool from `nproc` capped by the cgroup limit, budgeting ~3 GB a worker; a caller who knows their deployment can do the same arithmetic and pass the result. + +Chunks of every job are converted through one pool, in the order the jobs and their chunks were given – a long job and a short one pack together rather than draining in phases. The output is concatenated in that same order, however the processes happened to finish. + +The first failure aborts the run: no further chunk is started, and the processes still going are stopped rather than left writing into a directory the converter is about to delete. + `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. ## How a conversion runs diff --git a/packages/sparql-anything/src/sparql-anything-converter.ts b/packages/sparql-anything/src/sparql-anything-converter.ts index ba2e3b45..4ef2a511 100644 --- a/packages/sparql-anything/src/sparql-anything-converter.ts +++ b/packages/sparql-anything/src/sparql-anything-converter.ts @@ -84,6 +84,14 @@ export interface SparqlAnythingConverterOptions { * are the converter's own, and are rejected here. */ cliArgs?: string[]; + /** + * How many chunks to convert at once. Each one is a JVM of its own, so this + * multiplies against {@link heap}: the memory a run needs is `concurrency × + * heap`, and the machine that has to hold it is the task runner's, not this + * process's. Left at one, chunks are converted one after another. + * @default 1 + */ + concurrency?: number; /** Runs the SPARQL Anything process for each chunk. */ taskRunner: TaskRunner; } @@ -98,6 +106,7 @@ export class SparqlAnythingConverter { private readonly workDir: string; private readonly heap: string; private readonly cliArgs: string[]; + private readonly concurrency: number; private readonly taskRunner: TaskRunner; constructor(options: SparqlAnythingConverterOptions) { @@ -120,6 +129,13 @@ export class SparqlAnythingConverter { ); } this.cliArgs = options.cliArgs ?? []; + const concurrency = options.concurrency ?? 1; + if (!Number.isInteger(concurrency) || concurrency < 1) { + throw new Error( + `‘${concurrency}’ is not a number of chunks to convert at once; give a whole number of one or more`, + ); + } + this.concurrency = concurrency; this.taskRunner = options.taskRunner; } @@ -146,38 +162,94 @@ export class SparqlAnythingConverter { const runDir = await mkdtemp(join(this.workDir, 'sparql-anything-')); const runDirName = basename(runDir); try { - const outputs: string[] = []; - let index = 0; - for (const { job, query } of planned) { - // Interpolated per chunk, at the point of use: holding a copy of the - // query for every chunk up front would scale with the input. - for (const chunk of job.chunks ?? [undefined]) { - const queryPath = join(runDirName, `query-${index}.rq`); - await writeFile( - join(this.workDir, queryPath), - chunk === undefined - ? query - : // A replacer function, so `$&` and friends in a chunk path are - // the characters they look like rather than replacement patterns. - query.replaceAll(SOURCE_PLACEHOLDER, () => chunk), - ); - const processOutput = join(runDirName, `output-${index}.nt`); - const task = await this.taskRunner.run( - this.command(queryPath, processOutput, job), + const count = await this.runAll(planned, runDirName); + // By index, not by completion: the order the jobs and their chunks were + // given is the order of the triples, however the processes finished. + await concatenate( + Array.from({ length: count }, (_, index) => + join(this.workDir, runDirName, `output-${index}.nt`), + ), + outputPath, + ); + } finally { + await rm(runDir, { recursive: true, force: true }); + } + } + + /** + * Runs every chunk, at most {@link concurrency} at a time, and returns how + * many processes that was. + * + * The first failure aborts the run: no further chunk is started, and the + * processes still going are stopped rather than left writing into a + * directory this is about to delete. + */ + private async runAll( + planned: PlannedJob[], + runDirName: string, + ): Promise { + const pending = processesOf(planned); + const inFlight = new Set(); + let failure: unknown; + + // 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 + // for the others, whatever the reason it stopped. + const convertChunks = async (): Promise => { + while (failure === undefined) { + const next = pending.next(); + if (next.done === true) { + return; + } + try { + await this.convertChunk(next.value, runDirName, inFlight); + } catch (error) { + failure ??= error; + await Promise.all( + [...inFlight].map((task) => this.taskRunner.stop(task)), ); - // wait() rejects on a non-zero exit, aborting convert() before the - // crashed process's missing output can be silently concatenated. - await this.taskRunner.wait(task); - const processOutputPath = join(this.workDir, processOutput); - await assertNonEmpty(processOutputPath, job, chunk); - outputs.push(processOutputPath); - index++; + return; } } - await concatenate(outputs, outputPath); + }; + + await Promise.all( + Array.from({ length: this.concurrency }, () => convertChunks()), + ); + if (failure !== undefined) { + throw failure; + } + return countOf(planned); + } + + /** Converts one chunk, writing `output-.nt` in the run directory. */ + private async convertChunk( + { index, job, chunk, query }: PlannedProcess, + runDirName: string, + inFlight: Set, + ): Promise { + const queryPath = join(runDirName, `query-${index}.rq`); + await writeFile( + join(this.workDir, queryPath), + chunk === undefined + ? query + : // A replacer function, so `$&` and friends in a chunk path are the + // characters they look like rather than replacement patterns. + query.replaceAll(SOURCE_PLACEHOLDER, () => chunk), + ); + const output = join(runDirName, `output-${index}.nt`); + const task = await this.taskRunner.run( + this.command(queryPath, output, job), + ); + inFlight.add(task); + try { + // wait() rejects on a non-zero exit, aborting convert() before the + // crashed chunk's missing output can be silently concatenated. + await this.taskRunner.wait(task); } finally { - await rm(runDir, { recursive: true, force: true }); + inFlight.delete(task); } + await assertNonEmpty(join(this.workDir, output), job, chunk); } /** The SPARQL Anything invocation for one job. */ @@ -203,6 +275,38 @@ export class SparqlAnythingConverter { } } +/** One SPARQL Anything process: a job's query, to run over one of its chunks. */ +interface PlannedProcess { + /** Position in the run, which orders the outputs and names their files. */ + index: number; + job: ConversionJob; + chunk?: string; + /** The job's query as written, with `{SOURCE}` still in it. */ + query: string; +} + +/** + * The processes the planned jobs call for, in order: one per chunk, and one + * for a job that has none. Lazy, so the workers pulling from it hold one + * process each rather than the whole run. + */ +function* processesOf(planned: PlannedJob[]): Generator { + let index = 0; + for (const { job, query } of planned) { + for (const chunk of job.chunks ?? [undefined]) { + yield { index: index++, job, chunk, query }; + } + } +} + +/** How many processes the planned jobs call for. */ +function countOf(planned: PlannedJob[]): number { + return planned.reduce( + (total, { job }) => total + (job.chunks?.length ?? 1), + 0, + ); +} + /** A job whose query has been read, and checked against its chunks. */ interface PlannedJob { job: ConversionJob; diff --git a/packages/sparql-anything/test/sparql-anything-converter.test.ts b/packages/sparql-anything/test/sparql-anything-converter.test.ts index bda1cb11..0f3e202f 100644 --- a/packages/sparql-anything/test/sparql-anything-converter.test.ts +++ b/packages/sparql-anything/test/sparql-anything-converter.test.ts @@ -31,6 +31,8 @@ interface FakeTaskRunnerOptions { * cannot be inspected – a symlink to itself, which fails `stat` with ELOOP. */ unreadableOutputContaining?: string; + /** Milliseconds `wait()` takes, per output path, to order completions. */ + waitFor?: Record; } /** @@ -45,6 +47,11 @@ interface FakeTaskRunnerOptions { class FakeTaskRunner implements TaskRunner<{ command: string }> { readonly commands: string[] = []; readonly queries: string[] = []; + /** Commands that were stopped before they finished. */ + readonly stopped: string[] = []; + /** The most processes that were ever in flight at the same time. */ + peakInFlight = 0; + private inFlight = 0; constructor( private readonly workDir: string, @@ -91,14 +98,23 @@ class FakeTaskRunner implements TaskRunner<{ command: string }> { } async wait(task: { command: string }): Promise { - const { failOutputContaining } = this.options; - if (matches(task.command, failOutputContaining)) { - throw new Error('Process failed with code 1'); + this.inFlight++; + this.peakInFlight = Math.max(this.peakInFlight, this.inFlight); + try { + const output = tokenAfter(task.command, '--output') ?? ''; + const delay = this.options.waitFor?.[output] ?? 0; + await new Promise((resolve) => setTimeout(resolve, delay)); + if (matches(task.command, this.options.failOutputContaining)) { + throw new Error('Process failed with code 1'); + } + return ''; + } finally { + this.inFlight--; } - return ''; } - async stop(): Promise { + async stop(task: { command: string }): Promise { + this.stopped.push(task.command); return null; } } @@ -446,6 +462,90 @@ describe('SparqlAnythingConverter', () => { ).rejects.toThrow('would go unread'); }); + it('converts several chunks at once, up to the configured concurrency', async () => { + const taskRunner = new FakeTaskRunner(workDir); + const chunks = await writeChunks(6); + + await new SparqlAnythingConverter({ + jarPath: '/bin/sparql-anything.jar', + workDir, + concurrency: 3, + taskRunner, + }).convert([{ queryFile, chunks }], join(workDir, 'output.nt')); + + expect(taskRunner.commands).toHaveLength(6); + expect(taskRunner.peakInFlight).toBe(3); + }); + + it('converts one chunk at a time by default', async () => { + const taskRunner = new FakeTaskRunner(workDir); + const chunks = await writeChunks(3); + + await converterFor(taskRunner).convert( + [{ queryFile, chunks }], + join(workDir, 'output.nt'), + ); + + expect(taskRunner.peakInFlight).toBe(1); + }); + + it('concatenates in the order given, not the order they finished', async () => { + const chunks = await writeChunks(3); + // The first chunk finishes last. + const taskRunner = new FakeTaskRunner(workDir, { + waitFor: { 'output-0.nt': 30, 'output-1.nt': 10 }, + }); + const outputPath = join(workDir, 'output.nt'); + + await new SparqlAnythingConverter({ + jarPath: '/bin/sparql-anything.jar', + workDir, + concurrency: 3, + taskRunner, + }).convert([{ queryFile, chunks }], outputPath); + + expect(await readFile(outputPath, 'utf-8')).toMatch( + /^sparql-anything-\S+\/output-0\.nt\n\nsparql-anything-\S+\/output-1\.nt\n\nsparql-anything-\S+\/output-2\.nt\n$/, + ); + }); + + it('stops the chunks still running when one fails', async () => { + const chunks = await writeChunks(4); + const taskRunner = new FakeTaskRunner(workDir, { + failOutputContaining: 'output-0.nt', + // The others outlive the failure, so they have to be stopped rather than + // left writing into the directory convert() is about to delete. + waitFor: { 'output-1.nt': 200, 'output-2.nt': 200 }, + }); + + await expect( + new SparqlAnythingConverter({ + jarPath: '/bin/sparql-anything.jar', + workDir, + concurrency: 3, + taskRunner, + }).convert([{ queryFile, chunks }], join(workDir, 'output.nt')), + ).rejects.toThrow('Process failed'); + + expect(taskRunner.stopped).toHaveLength(2); + // The fourth chunk was never started. + expect(taskRunner.commands).toHaveLength(3); + }); + + it('rejects a concurrency that is not a whole number of processes', () => { + const taskRunner = new FakeTaskRunner(workDir); + + expect( + () => + new SparqlAnythingConverter({ + jarPath: '/bin/sparql-anything.jar', + workDir, + concurrency: 0, + taskRunner, + }), + ).toThrow('is not a number of chunks to convert at once'); + }); + it('refuses an empty job list rather than writing an empty output', async () => { const taskRunner = new FakeTaskRunner(workDir); const outputPath = join(workDir, 'output.nt'); From f312c657413adf3544e748f2de566f524772ac6f Mon Sep 17 00:00:00 2001 From: David de Boer Date: Mon, 31 Aug 2026 19:54:05 +0200 Subject: [PATCH 2/2] fix(sparql-anything): keep the abort path from outliving or hiding the failure MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three ways the abort could go wrong, all of them found by review: - Stopping was unguarded. A task runner cannot stop what has already exited – DockerTaskRunner's stop() reads logs and stops the container, and both reject for one that is gone – so a stop failure replaced the conversion failure and rejected the join, leaving the other workers unawaited while the run directory was deleted underneath them. Stopping is best effort now. - A chunk could be started after the run had failed. The guard sits before two awaits, so a worker could pass it, then start a process while another chunk was failing; that process was not in the set the failure stopped, and the run waited out its whole conversion. Each worker now checks once its process exists, and stops it. - A DockerTaskRunner with a containerName force-removes any container of that name before starting a task, so chunks in parallel would destroy each other's containers. Documented as a warning; the runner needs a per-task name before that combination can work. The fake task runner's delays never applied: they were keyed by file name while the command names the output relative to the run directory, so every lookup missed and every delay was zero. Two tests that read as though they staged an ordering were passing on whatever order the event loop produced. --- docs/reference/sparql-anything.md | 5 +- .../src/sparql-anything-converter.ts | 54 ++++++++++--- .../test/sparql-anything-converter.test.ts | 80 ++++++++++++++++++- 3 files changed, 122 insertions(+), 17 deletions(-) diff --git a/docs/reference/sparql-anything.md b/docs/reference/sparql-anything.md index b95bd8a4..cf439fec 100644 --- a/docs/reference/sparql-anything.md +++ b/docs/reference/sparql-anything.md @@ -94,7 +94,10 @@ That is why the default is `1` rather than something derived from the CPU count Chunks of every job are converted through one pool, in the order the jobs and their chunks were given – a long job and a short one pack together rather than draining in phases. The output is concatenated in that same order, however the processes happened to finish. -The first failure aborts the run: no further chunk is started, and the processes still going are stopped rather than left writing into a directory the converter is about to delete. +The first failure aborts the run: no further chunk is started, and the processes still going are stopped rather than left writing into a directory the converter is about to delete. A process that cannot be stopped – one that has just exited, say – does not change what is reported: the conversion failure is the one worth reading. + +> [!WARNING] +> A `DockerTaskRunner` configured with a `containerName` cannot be used with `concurrency` above one. It force-removes any container of that name before starting a task, so each chunk would destroy the container of the chunk before it. Leave `containerName` unset for a converter that runs chunks in parallel. `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. diff --git a/packages/sparql-anything/src/sparql-anything-converter.ts b/packages/sparql-anything/src/sparql-anything-converter.ts index 4ef2a511..5c1c8a29 100644 --- a/packages/sparql-anything/src/sparql-anything-converter.ts +++ b/packages/sparql-anything/src/sparql-anything-converter.ts @@ -189,25 +189,22 @@ export class SparqlAnythingConverter { runDirName: string, ): Promise { const pending = processesOf(planned); - const inFlight = new Set(); - let failure: unknown; + const state: RunState = { 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 // for the others, whatever the reason it stopped. const convertChunks = async (): Promise => { - while (failure === undefined) { + while (state.failure === undefined) { const next = pending.next(); if (next.done === true) { return; } try { - await this.convertChunk(next.value, runDirName, inFlight); + await this.convertChunk(next.value, runDirName, state); } catch (error) { - failure ??= error; - await Promise.all( - [...inFlight].map((task) => this.taskRunner.stop(task)), - ); + state.failure ??= error; + await this.stopInFlight(state); return; } } @@ -216,8 +213,8 @@ export class SparqlAnythingConverter { await Promise.all( Array.from({ length: this.concurrency }, () => convertChunks()), ); - if (failure !== undefined) { - throw failure; + if (state.failure !== undefined) { + throw state.failure; } return countOf(planned); } @@ -226,7 +223,7 @@ export class SparqlAnythingConverter { private async convertChunk( { index, job, chunk, query }: PlannedProcess, runDirName: string, - inFlight: Set, + state: RunState, ): Promise { const queryPath = join(runDirName, `query-${index}.rq`); await writeFile( @@ -241,17 +238,40 @@ export class SparqlAnythingConverter { const task = await this.taskRunner.run( this.command(queryPath, output, job), ); - inFlight.add(task); + state.inFlight.add(task); try { + if (state.failure !== undefined) { + // Started in the window between another chunk failing and this worker + // seeing it, so it is not in the set that failure stopped. + await this.stopQuietly(task); + return; + } // wait() rejects on a non-zero exit, aborting convert() before the // crashed chunk's missing output can be silently concatenated. await this.taskRunner.wait(task); } finally { - inFlight.delete(task); + state.inFlight.delete(task); } await assertNonEmpty(join(this.workDir, output), job, chunk); } + /** Stops every process still running, so none outlives the run. */ + private async stopInFlight(state: RunState): Promise { + await Promise.all( + [...state.inFlight].map((task) => this.stopQuietly(task)), + ); + } + + /** + * Stops a task, ignoring a failure to stop it. Stopping is best effort by + * nature – a process that has just exited cannot be stopped, and reporting + * that would replace the failure that is actually worth reporting, and leave + * the other workers unawaited while the run directory is deleted. + */ + private async stopQuietly(task: Task): Promise { + await this.taskRunner.stop(task).catch(() => undefined); + } + /** The SPARQL Anything invocation for one job. */ private command( queryPath: string, @@ -307,6 +327,14 @@ function countOf(planned: PlannedJob[]): number { ); } +/** What the workers of one run share. */ +interface RunState { + /** Tasks that have been started and not yet finished. */ + inFlight: Set; + /** The first failure, which aborts the run. */ + failure?: unknown; +} + /** A job whose query has been read, and checked against its chunks. */ interface PlannedJob { job: ConversionJob; diff --git a/packages/sparql-anything/test/sparql-anything-converter.test.ts b/packages/sparql-anything/test/sparql-anything-converter.test.ts index 0f3e202f..6eb2feb8 100644 --- a/packages/sparql-anything/test/sparql-anything-converter.test.ts +++ b/packages/sparql-anything/test/sparql-anything-converter.test.ts @@ -33,6 +33,12 @@ interface FakeTaskRunnerOptions { unreadableOutputContaining?: string; /** Milliseconds `wait()` takes, per output path, to order completions. */ waitFor?: Record; + /** Milliseconds `wait()` takes for any output not named in `waitFor`. */ + waitForAll?: number; + /** Milliseconds `run()` takes, per output path, before the task exists. */ + runFor?: Record; + /** When set, `stop()` rejects, as a runner does for a container already gone. */ + stopFails?: boolean; } /** @@ -67,6 +73,8 @@ class FakeTaskRunner implements TaskRunner<{ command: string }> { const outputFile = tokenAfter(command, '--output'); if (outputFile) { await this.writeOutput(outputFile); + const delay = delayFor(this.options.runFor, outputFile) ?? 0; + await new Promise((resolve) => setTimeout(resolve, delay)); } return { command }; } @@ -101,8 +109,10 @@ class FakeTaskRunner implements TaskRunner<{ command: string }> { this.inFlight++; this.peakInFlight = Math.max(this.peakInFlight, this.inFlight); try { - const output = tokenAfter(task.command, '--output') ?? ''; - const delay = this.options.waitFor?.[output] ?? 0; + const delay = + delayFor(this.options.waitFor, tokenAfter(task.command, '--output')) ?? + this.options.waitForAll ?? + 0; await new Promise((resolve) => setTimeout(resolve, delay)); if (matches(task.command, this.options.failOutputContaining)) { throw new Error('Process failed with code 1'); @@ -115,10 +125,30 @@ class FakeTaskRunner implements TaskRunner<{ command: string }> { async stop(task: { command: string }): Promise { this.stopped.push(task.command); + if (this.options.stopFails === true) { + throw new Error('No such container'); + } return null; } } +/** + * The delay configured for an output path. Keyed by file name, while the + * command names it relative to the run directory. + */ +function delayFor( + delays: Record | undefined, + outputPath?: string, +): number | undefined { + if (delays === undefined || outputPath === undefined) { + return undefined; + } + const named = Object.entries(delays).find(([name]) => + outputPath.endsWith(name), + ); + return named?.[1]; +} + /** Whether `needle` is configured and `haystack` contains it. */ function matches(haystack: string, needle?: string): boolean { return needle !== undefined && haystack.includes(needle); @@ -463,7 +493,8 @@ describe('SparqlAnythingConverter', () => { }); it('converts several chunks at once, up to the configured concurrency', async () => { - const taskRunner = new FakeTaskRunner(workDir); + // A delay every chunk shares, so the overlap does not depend on timing. + const taskRunner = new FakeTaskRunner(workDir, { waitForAll: 20 }); const chunks = await writeChunks(6); await new SparqlAnythingConverter({ @@ -532,6 +563,49 @@ describe('SparqlAnythingConverter', () => { expect(taskRunner.commands).toHaveLength(3); }); + it('reports the conversion failure even when stopping the others fails', async () => { + const chunks = await writeChunks(3); + // A task runner cannot stop a container that has already gone; saying so + // must not replace the failure that is worth reporting. + const taskRunner = new FakeTaskRunner(workDir, { + failOutputContaining: 'output-0.nt', + waitFor: { 'output-1.nt': 100, 'output-2.nt': 100 }, + stopFails: true, + }); + + await expect( + new SparqlAnythingConverter({ + jarPath: '/bin/sparql-anything.jar', + workDir, + concurrency: 3, + taskRunner, + }).convert([{ queryFile, chunks }], join(workDir, 'output.nt')), + ).rejects.toThrow('Process failed'); + }); + + it('stops a chunk that started while the run was failing', async () => { + const chunks = await writeChunks(2); + // The second chunk's process appears only after the first has failed, so + // it is not among the ones that failure stopped. + const taskRunner = new FakeTaskRunner(workDir, { + failOutputContaining: 'output-0.nt', + runFor: { 'output-1.nt': 50 }, + }); + + await expect( + new SparqlAnythingConverter({ + jarPath: '/bin/sparql-anything.jar', + workDir, + concurrency: 2, + taskRunner, + }).convert([{ queryFile, chunks }], join(workDir, 'output.nt')), + ).rejects.toThrow('Process failed'); + + expect(taskRunner.stopped).toEqual([ + expect.stringContaining('output-1.nt'), + ]); + }); + it('rejects a concurrency that is not a whole number of processes', () => { const taskRunner = new FakeTaskRunner(workDir);