fix(sparql-anything): concatenate outputs without blank lines or listener leaks

docs/reference/sparql-anything.md

@@ -159,4 +159,4 @@ For each chunk – or once, for a job that has none – the converter:
 
 Converting an empty list of jobs is an error rather than an empty output: a step that produced none has already failed.
 
-Finally, the `.nt` files are concatenated, in the order the jobs and their chunks were given, into the output path. The concatenation streams, so multi-gigabyte outputs do not have to fit in memory. N-Triples has no prefixes or document structure, so concatenating per-chunk files always yields a single valid document.
+Finally, the `.nt` files are concatenated, in the order the jobs and their chunks were given, into the output path. The concatenation streams, so multi-gigabyte outputs do not have to fit in memory, and it is byte for byte what `cat` would give: a newline is inserted only after a file that does not end in one, so two triples never share a line and no blank line is added between files. N-Triples has no prefixes or document structure, so concatenating per-chunk files always yields a single valid document.

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

@@ -1,7 +1,7 @@
 import { shellQuote, TaskRunner } from '@lde/task-runner';
 import { createReadStream, createWriteStream } from 'node:fs';
 import { mkdtemp, readFile, rm, stat, writeFile } from 'node:fs/promises';
-import { finished, pipeline } from 'node:stream/promises';
+import { pipeline } from 'node:stream/promises';
 import { basename, join } from 'node:path';
 
 /** Placeholder in the query file that is replaced with each chunk's path. */
@@ -445,27 +445,37 @@ async function assertNonEmpty(
   }
 }
 
+/** The byte that ends an N-Triples line. */
+const NEWLINE = '\n'.charCodeAt(0);
+
 /**
  * Concatenates `inputPaths` into `outputPath`, streaming so multi-GB outputs do
  * not have to fit in memory. N-Triples has no prefixes or document structure, so
  * concatenating per-chunk files yields a single valid document.
+ *
+ * One pipeline over every input rather than one per input into a shared
+ * destination: each pipeline() leaves its listeners on the destination, and
+ * past ten of them Node warns of a leak.
  */
 async function concatenate(
   inputPaths: string[],
   outputPath: string,
 ): Promise<void> {
-  const output = createWriteStream(outputPath);
-  for (const [index, inputPath] of inputPaths.entries()) {
-    // A newline between files, in case one does not end in one: N-Triples
-    // tolerates the blank line, but not two triples sharing a line.
-    if (index > 0) {
-      output.write('\n');
+  await pipeline(async function* () {
+    let endsInNewline = true;
+    for (const inputPath of inputPaths) {
+      // A newline between files only when one does not end in one, so the
+      // result is byte for byte what `cat` gives: N-Triples tolerates a blank
+      // line, but not two triples sharing a line.
+      if (!endsInNewline) {
+        yield '\n';
+      }
+      for await (const chunk of createReadStream(
+        inputPath,
+      ) as AsyncIterable<Buffer>) {
+        yield chunk;
+        endsInNewline = chunk.at(-1) === NEWLINE;
+      }
     }
-    await pipeline(createReadStream(inputPath), output, { end: false });
-  }
-  output.end();
-  // pipeline() with `end: false` resolves once the source ends, not once the
-  // destination is flushed and closed, so await that before reporting success.
-  // On an earlier rejection pipeline() has already destroyed the stream.
-  await finished(output);
+  }, createWriteStream(outputPath));
 }

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

@@ -30,6 +30,11 @@ interface FakeTaskRunnerOptions {
    * all, as SPARQL Anything leaves behind when it crashes before writing.
    */
   missingOutputContaining?: string;
+  /**
+   * When set, chunks whose output path contains this get an `--output` that
+   * does not end in a newline, as a writer other than Jena’s might leave.
+   */
+  unterminatedOutputContaining?: string;
   /**
    * When set, chunks whose output path contains this get an `--output` that
    * cannot be inspected – a symlink to itself, which fails `stat` with ELOOP.
@@ -93,6 +98,7 @@ class FakeTaskRunner implements TaskRunner<{ command: string }> {
     const {
       emptyOutputContaining,
       missingOutputContaining,
+      unterminatedOutputContaining,
       unreadableOutputContaining,
     } = this.options;
     if (matches(outputFile, missingOutputContaining)) {
@@ -105,7 +111,11 @@ class FakeTaskRunner implements TaskRunner<{ command: string }> {
     }
     await writeFile(
       path,
-      matches(outputFile, emptyOutputContaining) ? '' : `${outputFile}\n`,
+      matches(outputFile, emptyOutputContaining)
+        ? ''
+        : matches(outputFile, unterminatedOutputContaining)
+          ? outputFile
+          : `${outputFile}\n`,
     );
   }
 
@@ -386,10 +396,58 @@ describe('SparqlAnythingConverter', () => {
     // The FakeTaskRunner writes each chunk's output path as that file's
     // content, so the result reflects the order the chunks were processed.
     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$/,
+      /^sparql-anything-\S+\/output-0\.nt\nsparql-anything-\S+\/output-1\.nt\nsparql-anything-\S+\/output-2\.nt\n$/,
     );
   });
 
+  it('concatenates outputs byte for byte, as cat would', async () => {
+    const taskRunner = new FakeTaskRunner(workDir);
+    const chunks = await writeChunks(3);
+    const outputPath = join(workDir, 'output.nt');
+
+    await converterFor(taskRunner).convert(jobsFor(chunks), outputPath);
+
+    const outputs = taskRunner.commands.map(
+      (command) => `${tokenAfter(command, '--output')}\n`,
+    );
+    expect(await readFile(outputPath, 'utf-8')).toBe(outputs.join(''));
+  });
+
+  it('separates an output that does not end in a newline from the next', async () => {
+    const taskRunner = new FakeTaskRunner(workDir, {
+      unterminatedOutputContaining: 'output-0.nt',
+    });
+    const chunks = await writeChunks(2);
+    const outputPath = join(workDir, 'output.nt');
+
+    await converterFor(taskRunner).convert(jobsFor(chunks), outputPath);
+
+    expect(await readFile(outputPath, 'utf-8')).toMatch(
+      /^sparql-anything-\S+\/output-0\.nt\nsparql-anything-\S+\/output-1\.nt\n$/,
+    );
+  });
+
+  it('concatenates more than ten outputs without a listener leak warning', async () => {
+    const taskRunner = new FakeTaskRunner(workDir);
+    const chunks = await writeChunks(12);
+    const warnings: Error[] = [];
+    const onWarning = (warning: Error) => warnings.push(warning);
+    process.on('warning', onWarning);
+
+    try {
+      await converterFor(taskRunner).convert(
+        jobsFor(chunks),
+        join(workDir, 'output.nt'),
+      );
+      // Node emits process warnings on a later tick, so give them one.
+      await new Promise((resolve) => setImmediate(resolve));
+    } finally {
+      process.off('warning', onWarning);
+    }
+
+    expect(warnings).toEqual([]);
+  });
+
   it('leaves nothing behind in the working directory', async () => {
     const taskRunner = new FakeTaskRunner(workDir);
     const chunks = await writeChunks(2);
@@ -540,7 +598,7 @@ describe('SparqlAnythingConverter', () => {
     }).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$/,
+      /^sparql-anything-\S+\/output-0\.nt\nsparql-anything-\S+\/output-1\.nt\nsparql-anything-\S+\/output-2\.nt\n$/,
     );
   });