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..2fa6c4a1 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,15 @@ export interface DockerTaskRunnerOptions { export class DockerTaskRunner implements TaskRunner { private readonly options; + /** + * 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 nameHolder?: Container | 'starting'; constructor(options: DockerTaskRunnerOptions) { this.options = { @@ -28,7 +44,12 @@ export class DockerTaskRunner implements TaskRunner { } async wait(task: Container): Promise { + // 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, @@ -47,8 +68,32 @@ export class DockerTaskRunner implements TaskRunner { } async run(command: string): Promise { + if (this.options.containerName) { + 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. await this.options.docker .getContainer(this.options.containerName) .remove({ force: true }); @@ -107,17 +152,38 @@ export class DockerTaskRunner implements TaskRunner { await this.options.docker.createContainer(containerOptions); await container.start(); + if (this.options.containerName) { + 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 { + await task.stop(); + } 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, }); - await task.stop(); return logs.toString(); } } diff --git a/packages/task-runner-docker/test/index.test.ts b/packages/task-runner-docker/test/index.test.ts index 9992c12d..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,13 +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: () => + 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', () => { @@ -64,4 +84,105 @@ 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 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({ + 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); + }); });