fix(sparql-anything): stop in-flight processes and clean up when the run is interrupted

docs/reference/sparql-anything.md

@@ -97,6 +97,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.
 
+An interrupted run ends the same way. For as long as a conversion runs, the converter listens for `SIGINT` and `SIGTERM` – a Ctrl-C, or a CI job being cancelled – and on either it stops the processes still going, removes its run directory, and then lets the signal end the process as it would have anyway. Without that, the JVMs would outlive the process that started them: the task runner spawns each one in a process group of its own, which is what lets it stop them as a whole, but nothing would tell it to.
+
 > [!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.
 

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

@@ -22,6 +22,13 @@ const HEAP_SIZE = /^(?!0+[kmg]?$)\d+[kmg]?$/i;
  */
 const DEFAULT_HEAP = '2g';
 
+/**
+ * The signals that end this process on a Ctrl-C or a cancelled job. The
+ * processes a run started would survive them: a task runner spawns each in a
+ * process group of its own, which is what lets it stop them as a whole.
+ */
+const STOP_SIGNALS: NodeJS.Signals[] = ['SIGINT', 'SIGTERM'];
+
 /**
  * Arguments the converter sets itself, with their aliases. Passing one again
  * through `cliArgs` would break what the converter does around the process:
@@ -183,8 +190,13 @@ export class SparqlAnythingConverter<Task> {
     // otherwise satisfy the non-empty check below with stale triples.
     const runDir = await mkdtemp(join(this.workDir, 'sparql-anything-'));
     const runDirName = basename(runDir);
+    const state: RunState<Task> = {
+      total: countOf(planned),
+      inFlight: new Set(),
+    };
+    const stopListeningForSignals = this.stopOnSignal(state, runDir);
     try {
-      const count = await this.runAll(planned, runDirName);
+      const count = await this.runAll(planned, runDirName, state);
       // 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(
@@ -194,8 +206,47 @@ export class SparqlAnythingConverter<Task> {
         outputPath,
       );
     } finally {
+      stopListeningForSignals();
+      await rm(runDir, { recursive: true, force: true });
+    }
+  }
+
+  /**
+   * Stops the run when this process is told to stop, so that a Ctrl-C or a
+   * cancelled job does not leave the processes running and the run directory
+   * behind, which `convert()`’s cleanup never gets to. The signal is then
+   * raised again with the listeners gone, so the process ends as it would have
+   * without them, with the same exit status.
+   *
+   * Returns what removes the listeners again. Each run listens for itself, so
+   * one that ends does not stop listening for another still going.
+   */
+  private stopOnSignal(state: RunState<Task>, runDir: string): () => void {
+    let stopping = false;
+    const onSignal = async (signal: NodeJS.Signals): Promise<void> => {
+      // A second signal while stopping changes nothing: the processes have
+      // been told to stop, and the run is about to end.
+      if (stopping) {
+        return;
+      }
+      stopping = true;
+      // The run’s failure now, whatever else went wrong: no further chunk is
+      // started while the processes are being stopped.
+      state.failure = new Error(`Interrupted by ${signal}`);
+      await this.stopInFlight(state);
       await rm(runDir, { recursive: true, force: true });
+      stopListening();
+      process.kill(process.pid, signal);
+    };
+    const stopListening = (): void => {
+      for (const signal of STOP_SIGNALS) {
+        process.off(signal, onSignal);
+      }
+    };
+    for (const signal of STOP_SIGNALS) {
+      process.on(signal, onSignal);
     }
+    return stopListening;
   }
 
   /**
@@ -209,12 +260,9 @@ export class SparqlAnythingConverter<Task> {
   private async runAll(
     planned: PlannedJob[],
     runDirName: string,
+    state: RunState<Task>,
   ): Promise<number> {
     const pending = processesOf(planned);
-    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

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

@@ -4,7 +4,7 @@ import {
   SparqlAnythingConverter,
 } from '../src/index.js';
 import { TaskRunner } from '@lde/task-runner';
-import { describe, it, expect, beforeEach, afterEach } from 'vitest';
+import { describe, it, expect, beforeEach, afterEach, vi } from 'vitest';
 import {
   mkdtemp,
   readdir,
@@ -13,6 +13,7 @@ import {
   symlink,
   writeFile,
 } from 'node:fs/promises';
+import { readdirSync } from 'node:fs';
 import { isAbsolute, join } from 'node:path';
 import { tmpdir } from 'node:os';
 
@@ -179,6 +180,7 @@ describe('SparqlAnythingConverter', () => {
   });
 
   afterEach(async () => {
+    vi.restoreAllMocks();
     await rm(workDir, { recursive: true, force: true });
   });
 
@@ -610,6 +612,126 @@ describe('SparqlAnythingConverter', () => {
     ]);
   });
 
+  /**
+   * Stands in for `process.kill`, which would end the test process when the
+   * converter raises the signal again. Records what the converter had done by
+   * then: the process exits right after, so nothing later counts. Restored
+   * after each test, which also forgets its calls, so assert before that.
+   */
+  function stubProcessKill() {
+    const atKill: { files?: string[]; listeners?: number }[] = [];
+    const kill = vi.spyOn(process, 'kill').mockImplementation(() => {
+      atKill.push({
+        files: readdirSync(workDir),
+        listeners: process.listenerCount('SIGINT'),
+      });
+      return true;
+    });
+    return { kill, atKill };
+  }
+
+  /** Delivers `signal` to this process as Node does, without the OS. */
+  function interrupt(signal: NodeJS.Signals): void {
+    process.emit(signal, signal);
+  }
+
+  it('stops the running chunks and cleans up when the process is interrupted', async () => {
+    const chunks = await writeChunks(3);
+    const taskRunner = new FakeTaskRunner(workDir, { waitForAll: 200 });
+    const listenersBefore = process.listenerCount('SIGINT');
+    const { kill, atKill } = stubProcessKill();
+    const run = new SparqlAnythingConverter({
+      jarPath: '/bin/sparql-anything.jar',
+      workDir,
+      concurrency: 2,
+      taskRunner,
+    }).convert([{ queryFile, chunks }], join(workDir, 'output.nt'));
+    await new Promise((resolve) => setTimeout(resolve, 50));
+    interrupt('SIGINT');
+
+    await expect(run).rejects.toThrow('Interrupted by SIGINT');
+
+    expect(taskRunner.stopped).toHaveLength(2);
+    // The third chunk was never started.
+    expect(taskRunner.commands).toHaveLength(2);
+    // Raised again, with the run directory and the listener gone, so the
+    // process ends as it would have without the converter listening.
+    expect(kill).toHaveBeenCalledExactlyOnceWith(process.pid, 'SIGINT');
+    expect(atKill).toEqual([
+      {
+        files: ['chunk-0.csv', 'chunk-1.csv', 'chunk-2.csv', 'places.rq'],
+        listeners: listenersBefore,
+      },
+    ]);
+  });
+
+  it('ignores a second signal while it is already stopping', async () => {
+    const chunks = await writeChunks(2);
+    const taskRunner = new FakeTaskRunner(workDir, { waitForAll: 200 });
+    const { kill } = stubProcessKill();
+    const run = new SparqlAnythingConverter({
+      jarPath: '/bin/sparql-anything.jar',
+      workDir,
+      concurrency: 2,
+      taskRunner,
+    }).convert([{ queryFile, chunks }], join(workDir, 'output.nt'));
+    await new Promise((resolve) => setTimeout(resolve, 50));
+    interrupt('SIGINT');
+    interrupt('SIGINT');
+
+    await expect(run).rejects.toThrow('Interrupted by SIGINT');
+
+    expect(taskRunner.stopped).toHaveLength(2);
+    expect(kill).toHaveBeenCalledOnce();
+  });
+
+  it('keeps listening for a run still going when another has finished', async () => {
+    const chunks = await writeChunks(3);
+    const finishing = new FakeTaskRunner(workDir);
+    const running = new FakeTaskRunner(workDir, { waitForAll: 200 });
+    const { kill } = stubProcessKill();
+    const run = new SparqlAnythingConverter({
+      jarPath: '/bin/sparql-anything.jar',
+      workDir,
+      concurrency: 2,
+      taskRunner: running,
+    }).convert(
+      [{ queryFile, chunks: chunks.slice(1) }],
+      join(workDir, 'output.nt'),
+    );
+    // Finishing removes only its own listener, not the other run’s.
+    await converterFor(finishing).convert(
+      [{ queryFile, chunks: chunks.slice(0, 1) }],
+      join(workDir, 'finished.nt'),
+    );
+    interrupt('SIGTERM');
+
+    await expect(run).rejects.toThrow('Interrupted by SIGTERM');
+
+    expect(finishing.stopped).toHaveLength(0);
+    expect(running.stopped).toHaveLength(2);
+    expect(kill).toHaveBeenCalledExactlyOnceWith(process.pid, 'SIGTERM');
+  });
+
+  it('stops listening for signals once the run is done', async () => {
+    const taskRunner = new FakeTaskRunner(workDir);
+    const chunks = await writeChunks(1);
+    const listenersBefore = {
+      SIGINT: process.listenerCount('SIGINT'),
+      SIGTERM: process.listenerCount('SIGTERM'),
+    };
+
+    await converterFor(taskRunner).convert(
+      jobsFor(chunks),
+      join(workDir, 'output.nt'),
+    );
+
+    expect({
+      SIGINT: process.listenerCount('SIGINT'),
+      SIGTERM: process.listenerCount('SIGTERM'),
+    }).toEqual(listenersBefore);
+  });
+
   it('rejects a concurrency that is not a whole number of processes', () => {
     const taskRunner = new FakeTaskRunner(workDir);