fix(task-runner-docker)!: refuse a second task while the named one runs

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.
 

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: <logs>`.
+
+## 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.

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<Container> {
   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<Container> {
   }
 
   async wait(task: Container): Promise<string> {
+    // 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<Container> {
   }
 
   async run(command: string): Promise<Container> {
+    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<Container> {
     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<Container> {
       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<string> {
+    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();
   }
 }

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);
+  });
 });