From 894611a8891b434a52fc33737925473b060fabbb Mon Sep 17 00:00:00 2001 From: David de Boer Date: Mon, 31 Aug 2026 20:04:06 +0200 Subject: [PATCH 1/2] fix(task-runner-docker)!: refuse a second task while the named one runs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit run() removes any container with the configured containerName before starting a task, which is what makes starting one again idempotent: a container left by an earlier run cannot block the next. It did that to a container this runner had started and was still running too, so a second task took the name out from under the first, and the first's wait() failed with nothing to explain why. Nobody hit it while tasks ran one at a time. @lde/sparql-anything now converts chunks in parallel, and Docker is the deployment its concurrency is for, so each chunk would have removed the container of the chunk before it. A name cannot be shared: on a network it is how other containers address the one that has it. So the runner keeps the name for one task at a time and says so, rather than picking a unique name per task, which would leave the address pointing at nothing. Leaving containerName unset runs tasks alongside each other, with Docker naming each container itself. Breaking: a second run() under a name whose task is still running now throws where it used to replace it. Nothing in this repo does that – sparql-qlever shares one named runner between its importer and server, and the pipeline stops the server in a finally before the next dataset – but a consumer relying on replacement will see the error rather than a container disappearing. --- docs/reference/sparql-anything.md | 4 +- docs/reference/task-runner-docker.md | 10 ++- packages/task-runner-docker/src/index.ts | 73 +++++++++++++------ .../task-runner-docker/test/index.test.ts | 48 ++++++++++++ 4 files changed, 111 insertions(+), 24 deletions(-) diff --git a/docs/reference/sparql-anything.md b/docs/reference/sparql-anything.md index cf439fec..326c4198 100644 --- a/docs/reference/sparql-anything.md +++ b/docs/reference/sparql-anything.md @@ -96,8 +96,8 @@ Chunks of every job are converted through one pool, in the order the jobs and th 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. +> [!NOTE] +> A `DockerTaskRunner` configured with a `containerName` runs one task at a time – the name is how other containers address it – so it rejects a second chunk rather than taking the name from the first. 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/docs/reference/task-runner-docker.md b/docs/reference/task-runner-docker.md index b8b90a4e..6944faf6 100644 --- a/docs/reference/task-runner-docker.md +++ b/docs/reference/task-runner-docker.md @@ -36,7 +36,7 @@ await runner.stop(container); | Option | Type | Required | Description | | --------------- | -------- | -------- | ---------------------------------------------------------------------------------------------------------------------------------- | | `image` | `string` | Yes | Docker image to use | -| `containerName` | `string` | No | Name for the container (auto-removed on restart) | +| `containerName` | `string` | No | Name for the container (auto-removed on restart). One task at a time; see [Naming containers](#naming-containers) | | `mountDir` | `string` | No | Host directory to mount at `/mount` in the container | | `port` | `number` | No | Port to expose from the container | | `network` | `string` | No | Docker network to attach the container to (`HostConfig.NetworkMode`); on a user-defined network the container is reachable by name | @@ -55,3 +55,11 @@ await runner.stop(container); ## Output and errors `wait()` and `stop()` do not stream logs while the container runs. Instead, each fetches the container’s stdout and stderr in one go (`follow: false`) and returns them as a string. When the container exits with a non-zero status code, `wait()` throws an error of the form `Task failed with status code N: `. + +## Naming containers + +`containerName` names the container, and on a `network` that name is how other containers reach it. It therefore belongs to one container at a time, and the runner enforces that: starting a task while a named task is still running is an error rather than a silent replacement. + +That matters because the runner removes a container of that name before starting a task, which is what makes restarting idempotent – a container left behind by an earlier run cannot block the next one. Without the check, a second task would remove the container of the first _while it was still running_, and the first task's `wait()` would fail with no explanation of why. + +Leave `containerName` unset to run tasks alongside each other. Docker then names each container itself, and nothing is shared for them to take from one another. diff --git a/packages/task-runner-docker/src/index.ts b/packages/task-runner-docker/src/index.ts index 3e7aebfe..a9e73972 100644 --- a/packages/task-runner-docker/src/index.ts +++ b/packages/task-runner-docker/src/index.ts @@ -4,6 +4,13 @@ import Docker, { Container, ContainerCreateOptions } from 'dockerode'; export interface DockerTaskRunnerOptions { image: string; + /** + * Name for the container. Other containers on a `network` address it by this + * name, so it belongs to one container at a time: a runner that has it runs + * one task at a time, and rejects a second while the first is still going. + * Leave it unset to run tasks alongside each other, and Docker names each + * container itself. + */ containerName?: string; /** Publish this container port to the same port on the host. */ port?: number; @@ -19,6 +26,12 @@ export interface DockerTaskRunnerOptions { export class DockerTaskRunner implements TaskRunner { private readonly options; + /** + * Containers started under {@link DockerTaskRunnerOptions.containerName} + * that have not been awaited or stopped, so a second task cannot take the + * name out from under one that is still running. + */ + private readonly running = new Set(); constructor(options: DockerTaskRunnerOptions) { this.options = { @@ -28,27 +41,38 @@ export class DockerTaskRunner implements TaskRunner { } async wait(task: Container): Promise { - const result = await task.wait(); - const logs = ( - await task.logs({ - stdout: true, - stderr: true, - follow: false, - }) - ).toString(); + try { + const result = await task.wait(); + const logs = ( + await task.logs({ + stdout: true, + stderr: true, + follow: false, + }) + ).toString(); + + if (result.StatusCode !== 0) { + throw new Error( + `Task failed with status code ${result.StatusCode}: ${logs})`, + ); + } - if (result.StatusCode !== 0) { - throw new Error( - `Task failed with status code ${result.StatusCode}: ${logs})`, - ); + return logs; + } finally { + this.running.delete(task); } - - return logs; } async run(command: string): Promise { if (this.options.containerName) { + if (this.running.size > 0) { + throw new Error( + `A task is already running as ‘${this.options.containerName}’. A runner with a containerName runs one task at a time, because that name is how other containers address it; leave containerName unset to run tasks alongside each other.`, + ); + } try { + // A container of this name left behind by an earlier run: removing it + // is what makes starting a task again idempotent. await this.options.docker .getContainer(this.options.containerName) .remove({ force: true }); @@ -107,17 +131,24 @@ export class DockerTaskRunner implements TaskRunner { await this.options.docker.createContainer(containerOptions); await container.start(); + if (this.options.containerName) { + this.running.add(container); + } return container; } async stop(task: Container): Promise { - const logs = await task.logs({ - stdout: true, - stderr: true, - follow: false, - }); - await task.stop(); - return logs.toString(); + try { + const logs = await task.logs({ + stdout: true, + stderr: true, + follow: false, + }); + await task.stop(); + return logs.toString(); + } finally { + this.running.delete(task); + } } } diff --git a/packages/task-runner-docker/test/index.test.ts b/packages/task-runner-docker/test/index.test.ts index 9992c12d..aa99172d 100644 --- a/packages/task-runner-docker/test/index.test.ts +++ b/packages/task-runner-docker/test/index.test.ts @@ -27,6 +27,9 @@ function createFakeDocker(): Docker & { created: ContainerCreateOptions[] } { fake.created.push(options); return { start: () => Promise.resolve(), + wait: () => Promise.resolve({ StatusCode: 0 }), + logs: () => Promise.resolve(Buffer.from('')), + stop: () => Promise.resolve(), }; }, }; @@ -64,4 +67,49 @@ describe('DockerTaskRunner', () => { expect(docker.created[0].HostConfig?.NetworkMode).toBe('app_default'); expect(docker.created[0].HostConfig?.PortBindings).toBeUndefined(); }); + + it('runs tasks alongside each other when they are not named', async () => { + const docker = createFakeDocker(); + const runner = new DockerTaskRunner({ image: 'example/image', docker }); + + await runner.run('first'); + await runner.run('second'); + + // Docker names each container itself, so neither can displace the other. + expect(docker.created).toHaveLength(2); + expect(docker.created[0].name).toBeUndefined(); + }); + + it('refuses a second task while the named one is still running', async () => { + const docker = createFakeDocker(); + const runner = new DockerTaskRunner({ + image: 'example/image', + containerName: 'example', + docker, + }); + await runner.run('first'); + + // Starting it would have force-removed the container of the task that is + // still running under that name. + await expect(runner.run('second')).rejects.toThrow( + 'A task is already running as ‘example’', + ); + expect(docker.created).toHaveLength(1); + }); + + it('reuses the name once the task it belonged to has finished', async () => { + const docker = createFakeDocker(); + const runner = new DockerTaskRunner({ + image: 'example/image', + containerName: 'example', + docker, + }); + + await runner.wait(await runner.run('first')); + await runner.stop(await runner.run('second')); + await runner.run('third'); + + // Starting a task again stays idempotent: each removes what the last left. + expect(docker.created).toHaveLength(3); + }); }); From b3f2a984752d54fc3c4ec4fc00b271fb3fb3e76e Mon Sep 17 00:00:00 2001 From: David de Boer Date: Mon, 31 Aug 2026 20:11:57 +0200 Subject: [PATCH 2/2] fix(task-runner-docker): claim the container name before the first await MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The guard added in this branch checked that the name was free and then took it four awaits later, after the pull, the create and the start. Two run() calls that overlap both found it free, and the second went on to remove the first's container – the behaviour the guard exists to prevent, in the only case that produces it. Sequential callers were never the problem. @lde/sparql-anything's pool is exactly such a caller: its workers call run() alongside each other, so with a containerName it would still have killed the first chunk mid-run. The name is now claimed synchronously, before anything is awaited, and given back if the container fails to start. Freeing it is tighter too, in both directions: - wait() frees it once the container has exited, not in a finally. A wait() that fails for a reason of its own – a dropped connection – leaves the container running, and freeing the name would let the next task remove it. - stop() frees it once the container has actually stopped, treating Docker's 304 as the state it asked for. It used to read logs first and free the name in a finally, so a failure to read logs freed the name of a container that was never stopped. --- packages/task-runner-docker/src/index.ts | 99 +++++++++++++------ .../task-runner-docker/test/index.test.ts | 79 ++++++++++++++- 2 files changed, 143 insertions(+), 35 deletions(-) diff --git a/packages/task-runner-docker/src/index.ts b/packages/task-runner-docker/src/index.ts index a9e73972..2fa6c4a1 100644 --- a/packages/task-runner-docker/src/index.ts +++ b/packages/task-runner-docker/src/index.ts @@ -27,11 +27,14 @@ export interface DockerTaskRunnerOptions { export class DockerTaskRunner implements TaskRunner { private readonly options; /** - * Containers started under {@link DockerTaskRunnerOptions.containerName} - * that have not been awaited or stopped, so a second task cannot take the - * name out from under one that is still running. + * What holds {@link DockerTaskRunnerOptions.containerName}: the container + * running under it, or `'starting'` while one is being created, so a second + * task cannot take the name out from under one that is still running. + * + * Claimed before the first await in {@link run}, because two calls that + * overlap would otherwise both find it free. */ - private readonly running = new Set(); + private nameHolder?: Container | 'starting'; constructor(options: DockerTaskRunnerOptions) { this.options = { @@ -41,35 +44,53 @@ export class DockerTaskRunner implements TaskRunner { } async wait(task: Container): Promise { - try { - const result = await task.wait(); - const logs = ( - await task.logs({ - stdout: true, - stderr: true, - follow: false, - }) - ).toString(); - - if (result.StatusCode !== 0) { - throw new Error( - `Task failed with status code ${result.StatusCode}: ${logs})`, - ); - } + // Only once the container has exited: a wait() that fails for a reason of + // its own – a dropped connection, say – leaves it running, and freeing the + // name would let the next task remove a container still doing its work. + const result = await task.wait(); + this.releaseName(task); + + const logs = ( + await task.logs({ + stdout: true, + stderr: true, + follow: false, + }) + ).toString(); - return logs; - } finally { - this.running.delete(task); + if (result.StatusCode !== 0) { + throw new Error( + `Task failed with status code ${result.StatusCode}: ${logs})`, + ); } + + return logs; } async run(command: string): Promise { if (this.options.containerName) { - if (this.running.size > 0) { + if (this.nameHolder !== undefined) { throw new Error( `A task is already running as ‘${this.options.containerName}’. A runner with a containerName runs one task at a time, because that name is how other containers address it; leave containerName unset to run tasks alongside each other.`, ); } + // Before anything is awaited, so two overlapping calls cannot both take + // the name for themselves. + this.nameHolder = 'starting'; + } + try { + return await this.start(command); + } catch (error) { + if (this.nameHolder === 'starting') { + this.nameHolder = undefined; + } + throw error; + } + } + + /** Creates and starts the container for `command`. */ + private async start(command: string): Promise { + if (this.options.containerName) { try { // A container of this name left behind by an earlier run: removing it // is what makes starting a task again idempotent. @@ -132,23 +153,37 @@ export class DockerTaskRunner implements TaskRunner { await container.start(); if (this.options.containerName) { - this.running.add(container); + this.nameHolder = container; } return container; } + /** Frees the container name, once the task holding it is no longer running. */ + private releaseName(task: Container): void { + if (this.nameHolder === task) { + this.nameHolder = undefined; + } + } + async stop(task: Container): Promise { try { - const logs = await task.logs({ - stdout: true, - stderr: true, - follow: false, - }); await task.stop(); - return logs.toString(); - } finally { - this.running.delete(task); + } catch (error) { + // 304 says it had already stopped, which is the state being asked for. + if ((error as { statusCode?: number }).statusCode !== 304) { + throw error; + } } + // Only now: a stop that did not happen leaves the container running, and + // its name is still its own. + this.releaseName(task); + + const logs = await task.logs({ + stdout: true, + stderr: true, + follow: false, + }); + return logs.toString(); } } diff --git a/packages/task-runner-docker/test/index.test.ts b/packages/task-runner-docker/test/index.test.ts index aa99172d..7da16bcd 100644 --- a/packages/task-runner-docker/test/index.test.ts +++ b/packages/task-runner-docker/test/index.test.ts @@ -4,9 +4,15 @@ import type { ContainerCreateOptions } from 'dockerode'; import { DockerTaskRunner } from '../src/index.js'; /** A fake Docker daemon that records the container options passed to it. */ -function createFakeDocker(): Docker & { created: ContainerCreateOptions[] } { +function createFakeDocker(): Docker & { + created: ContainerCreateOptions[]; + failNextCreate: boolean; + failNextWait: boolean; +} { const fake = { created: [] as ContainerCreateOptions[], + failNextCreate: false, + failNextWait: false, async pull() { return {}; }, @@ -24,16 +30,27 @@ function createFakeDocker(): Docker & { created: ContainerCreateOptions[] } { }; }, async createContainer(options: ContainerCreateOptions) { + if (fake.failNextCreate) { + fake.failNextCreate = false; + throw new Error('no space left on device'); + } fake.created.push(options); return { start: () => Promise.resolve(), - wait: () => Promise.resolve({ StatusCode: 0 }), + wait: () => + fake.failNextWait + ? Promise.reject(new Error('connection reset')) + : Promise.resolve({ StatusCode: 0 }), logs: () => Promise.resolve(Buffer.from('')), stop: () => Promise.resolve(), }; }, }; - return fake as unknown as Docker & { created: ContainerCreateOptions[] }; + return fake as unknown as Docker & { + created: ContainerCreateOptions[]; + failNextCreate: boolean; + failNextWait: boolean; + }; } describe('DockerTaskRunner', () => { @@ -80,6 +97,62 @@ describe('DockerTaskRunner', () => { expect(docker.created[0].name).toBeUndefined(); }); + it('refuses a second task started alongside the named one', async () => { + const docker = createFakeDocker(); + const runner = new DockerTaskRunner({ + image: 'example/image', + containerName: 'example', + docker, + }); + + // Overlapping calls, which is what a pool makes: both would otherwise find + // the name free, and the second would remove the first's container. + const results = await Promise.allSettled([ + runner.run('first'), + runner.run('second'), + ]); + + expect(results.map((result) => result.status)).toEqual([ + 'fulfilled', + 'rejected', + ]); + expect(docker.created).toHaveLength(1); + }); + + it('frees the name when the task fails to start', async () => { + const docker = createFakeDocker(); + docker.failNextCreate = true; + const runner = new DockerTaskRunner({ + image: 'example/image', + containerName: 'example', + docker, + }); + + await expect(runner.run('first')).rejects.toThrow('no space left'); + + // The name was claimed before the container existed; a failure to create + // one must give it back. + await expect(runner.run('second')).resolves.toBeDefined(); + }); + + it('keeps the name when waiting fails without the container exiting', async () => { + const docker = createFakeDocker(); + docker.failNextWait = true; + const runner = new DockerTaskRunner({ + image: 'example/image', + containerName: 'example', + docker, + }); + const task = await runner.run('first'); + + await expect(runner.wait(task)).rejects.toThrow('connection reset'); + + // The container is still running, so its name is still its own. + await expect(runner.run('second')).rejects.toThrow( + 'A task is already running', + ); + }); + it('refuses a second task while the named one is still running', async () => { const docker = createFakeDocker(); const runner = new DockerTaskRunner({