From 0762e2496ce28409f4ad42c57185119d43f14c0a Mon Sep 17 00:00:00 2001
From: David de Boer <david@ddeboer.nl>
Date: Wed, 2 Sep 2026 10:38:09 +0200
Subject: [PATCH 1/2] feat(sparql-anything): split a file into the chunks a
 conversion can hold
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

convert() takes chunks it does not create, so every consumer was left to
produce them – the shell script this package replaces does it with split(1)
and a cat loop per chunk.

chunk() streams a line-oriented file into fixed-row pieces and returns their
paths in order, which is what a job's `chunks` takes. It belongs here rather
than in a package of its own: LDE's pipeline streams and has no use for a file
in pieces. Chunking exists because SPARQL Anything materialises a chunk's whole
result graph before writing it, so it is part of working around this tool, and
this is the package that does that.

Splitting is by line, which is what the format in hand allows: a tab-separated
export has no quoting mechanism, so a record cannot span lines. A delimited
format that wraps a field in quotes to carry a newline would be cut in two, and
the contract says so rather than parsing every field to guard against an input
this cannot receive. Measured, line splitting also runs a single pass at about
6x a parser's throughput.

`extension` exists for tools that read the format from the file name: SPARQL
Anything does, so a .txt export of a CSV has to be chunked as .csv to be read
as one. It defaults to the input's own extension, so chunking N-Triples for
something that expects N-Triples needs nothing.
---
 docs/reference/sparql-anything.md           |  31 +++++
 packages/sparql-anything/src/chunk.ts       | 109 ++++++++++++++++
 packages/sparql-anything/src/index.ts       |   1 +
 packages/sparql-anything/test/chunk.test.ts | 136 ++++++++++++++++++++
 4 files changed, 277 insertions(+)
 create mode 100644 packages/sparql-anything/src/chunk.ts
 create mode 100644 packages/sparql-anything/test/chunk.test.ts

diff --git a/docs/reference/sparql-anything.md b/docs/reference/sparql-anything.md
index 326c4198..5415aa4b 100644
--- a/docs/reference/sparql-anything.md
+++ b/docs/reference/sparql-anything.md
@@ -101,6 +101,37 @@ The first failure aborts the run: no further chunk is started, and the processes
 
 `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.
 
+## Chunking
+
+`convert()` takes chunks it does not create, because a caller often has work to do first – filtering rows out, say. `chunk()` produces them:
+
+```typescript
+import { chunk } from '@lde/sparql-anything';
+
+const chunks = await chunk('data/allCountries.txt', {
+  rows: 1_000_000,
+  into: 'data/chunks',
+  header: 'geonameid\tname\tlatitude\tlongitude',
+  extension: '.csv',
+});
+// → ['data/chunks/allCountries-0000.csv', 'data/chunks/allCountries-0001.csv', …]
+```
+
+It streams, so the file never has to fit in memory, and returns the chunk paths in order – which is what a job's `chunks` takes.
+
+| Option      | Type     | Description                                                                               |
+| ----------- | -------- | ----------------------------------------------------------------------------------------- |
+| `rows`      | `number` | Data rows per chunk, chosen together with `heap`: a chunk is what one process has to hold |
+| `into`      | `string` | Directory the chunks are written to, created if it does not exist                         |
+| `header`    | `string` | Line repeated at the top of every chunk; leave out for a format without a header          |
+| `extension` | `string` | Extension for the chunk files (default: the input's own); see below                       |
+
+Set `extension` for a tool that reads the format from the file name – SPARQL Anything does, so a `.txt` export of a CSV has to be chunked as `.csv` to be read as one.
+
+**Splitting is by line**, so every record must be one line. A delimited format that wraps a field in quotes to carry a newline inside it would be cut in two; tab-separated exports, N-Triples and NDJSON are one record per line by definition. Line endings are normalised to `\n`.
+
+An input with no rows is an error rather than an empty set of chunks: a step that produced an empty file has already failed.
+
 ## How a conversion runs
 
 For each chunk – or once, for a job that has none – the converter:
diff --git a/packages/sparql-anything/src/chunk.ts b/packages/sparql-anything/src/chunk.ts
new file mode 100644
index 00000000..07298852
--- /dev/null
+++ b/packages/sparql-anything/src/chunk.ts
@@ -0,0 +1,109 @@
+import { createReadStream, createWriteStream, WriteStream } from 'node:fs';
+import { mkdir } from 'node:fs/promises';
+import { once } from 'node:events';
+import { createInterface } from 'node:readline';
+import { basename, extname, join } from 'node:path';
+
+/** Configuration for {@link chunk}. */
+export interface ChunkOptions {
+  /**
+   * Data rows per chunk. Chosen together with the converter's `heap`: a chunk
+   * is what one process has to hold at once.
+   */
+  rows: number;
+  /** Directory the chunks are written to. Created if it does not exist. */
+  into: string;
+  /**
+   * Line repeated at the top of every chunk, for a format whose columns are
+   * named. Leave it out for a format without a header, such as N-Triples.
+   */
+  header?: string;
+  /**
+   * Extension for the chunk files, `'.csv'` and so on. Defaults to the input's
+   * own, and is worth setting for a tool that reads the format from the name –
+   * SPARQL Anything does, so a `.txt` export of a CSV has to be chunked as
+   * `.csv` to be read as one.
+   */
+  extension?: string;
+}
+
+/**
+ * Splits a line-oriented file into chunks of `rows` rows each, and returns
+ * their paths in order.
+ *
+ * SPARQL Anything materialises a chunk's whole result graph before writing it,
+ * so what a conversion can hold is a chunk rather than a file; this is how a
+ * file that does not fit becomes chunks that do.
+ *
+ * Splitting is by line, so every record must be one line: a delimited format
+ * that wraps a field in quotes to carry a newline inside it would be cut in
+ * two. Tab-separated exports, N-Triples and NDJSON are all one record per line
+ * by definition. Line endings are normalised to `\n`.
+ */
+export async function chunk(
+  inputPath: string,
+  options: ChunkOptions,
+): Promise<string[]> {
+  const { rows, into, header } = options;
+  if (!Number.isInteger(rows) || rows < 1) {
+    throw new Error(
+      `‘${rows}’ is not a number of rows to a chunk; give a whole number of one or more`,
+    );
+  }
+  await mkdir(into, { recursive: true });
+
+  const extension = options.extension ?? extname(inputPath);
+  const name = basename(inputPath, extname(inputPath));
+  const lines = createInterface({
+    input: createReadStream(inputPath),
+    crlfDelay: Infinity,
+  });
+
+  const paths: string[] = [];
+  let chunkFile: WriteStream | undefined;
+  let rowsWritten = 0;
+  for await (const line of lines) {
+    if (chunkFile === undefined) {
+      const path = join(
+        into,
+        `${name}-${String(paths.length).padStart(4, '0')}${extension}`,
+      );
+      paths.push(path);
+      chunkFile = createWriteStream(path);
+      if (header !== undefined) {
+        await write(chunkFile, `${header}\n`);
+      }
+    }
+    await write(chunkFile, `${line}\n`);
+    rowsWritten++;
+    if (rowsWritten === rows) {
+      await close(chunkFile);
+      chunkFile = undefined;
+      rowsWritten = 0;
+    }
+  }
+  if (chunkFile !== undefined) {
+    await close(chunkFile);
+  }
+
+  if (paths.length === 0) {
+    throw new Error(
+      `‘${inputPath}’ holds no rows to chunk; a step that produced an empty file has failed upstream, and converting nothing would hide that`,
+    );
+  }
+
+  return paths;
+}
+
+/** Writes `text`, waiting for the stream to drain when it asks to. */
+async function write(chunkFile: WriteStream, text: string): Promise<void> {
+  if (!chunkFile.write(text)) {
+    await once(chunkFile, 'drain');
+  }
+}
+
+/** Closes a chunk, so it is complete before its path is handed on. */
+async function close(chunkFile: WriteStream): Promise<void> {
+  chunkFile.end();
+  await once(chunkFile, 'close');
+}
diff --git a/packages/sparql-anything/src/index.ts b/packages/sparql-anything/src/index.ts
index 28950649..f7e27c7f 100644
--- a/packages/sparql-anything/src/index.ts
+++ b/packages/sparql-anything/src/index.ts
@@ -1,3 +1,4 @@
+export { chunk, type ChunkOptions } from './chunk.js';
 export {
   SparqlAnythingConverter,
   type ConversionJob,
diff --git a/packages/sparql-anything/test/chunk.test.ts b/packages/sparql-anything/test/chunk.test.ts
new file mode 100644
index 00000000..df78d85f
--- /dev/null
+++ b/packages/sparql-anything/test/chunk.test.ts
@@ -0,0 +1,136 @@
+import { chunk } from '../src/index.js';
+import { describe, it, expect, beforeEach, afterEach } from 'vitest';
+import { mkdtemp, readdir, readFile, rm, writeFile } from 'node:fs/promises';
+import { join } from 'node:path';
+import { tmpdir } from 'node:os';
+
+describe('chunk', () => {
+  let workDir: string;
+
+  beforeEach(async () => {
+    workDir = await mkdtemp(join(tmpdir(), 'chunk-test-'));
+  });
+
+  afterEach(async () => {
+    await rm(workDir, { recursive: true, force: true });
+  });
+
+  /** Writes an input file of `rows` numbered rows. */
+  async function writeInput(
+    rows: number,
+    name = 'places.txt',
+  ): Promise<string> {
+    const path = join(workDir, name);
+    await writeFile(
+      path,
+      `${Array.from({ length: rows }, (_, index) => `row-${index}`).join('\n')}\n`,
+    );
+    return path;
+  }
+
+  it('splits into chunks of the given number of rows', async () => {
+    const input = await writeInput(5);
+
+    const paths = await chunk(input, {
+      rows: 2,
+      into: join(workDir, 'chunks'),
+    });
+
+    expect(paths).toHaveLength(3);
+    expect(await readFile(paths[0], 'utf-8')).toBe('row-0\nrow-1\n');
+    // The last chunk holds what is left rather than being padded.
+    expect(await readFile(paths[2], 'utf-8')).toBe('row-4\n');
+  });
+
+  it('repeats the header at the top of every chunk', async () => {
+    const input = await writeInput(3);
+
+    const paths = await chunk(input, {
+      rows: 2,
+      into: join(workDir, 'chunks'),
+      header: 'id\tname',
+    });
+
+    expect(await readFile(paths[0], 'utf-8')).toBe('id\tname\nrow-0\nrow-1\n');
+    expect(await readFile(paths[1], 'utf-8')).toBe('id\tname\nrow-2\n');
+  });
+
+  it('writes no header for a format that has none', async () => {
+    const input = await writeInput(2, 'graph.nt');
+
+    const paths = await chunk(input, {
+      rows: 1,
+      into: join(workDir, 'chunks'),
+    });
+
+    expect(await readFile(paths[0], 'utf-8')).toBe('row-0\n');
+    // N-Triples keeps its extension, so what reads the chunk still knows it.
+    expect(paths[0].endsWith('.nt')).toBe(true);
+  });
+
+  it('names chunks after the input, in order, and returns their paths', async () => {
+    const input = await writeInput(4);
+
+    const paths = await chunk(input, {
+      rows: 2,
+      into: join(workDir, 'chunks'),
+    });
+
+    expect(paths.map((path) => path.replace(`${workDir}/`, ''))).toEqual([
+      'chunks/places-0000.txt',
+      'chunks/places-0001.txt',
+    ]);
+    expect(await readdir(join(workDir, 'chunks'))).toHaveLength(2);
+  });
+
+  it('gives the chunks an extension of their own when asked', async () => {
+    const input = await writeInput(2);
+
+    const paths = await chunk(input, {
+      rows: 2,
+      into: join(workDir, 'chunks'),
+      extension: '.csv',
+    });
+
+    // SPARQL Anything reads the format from the name, so a .txt export of a
+    // CSV has to arrive as .csv.
+    expect(paths[0].endsWith('places-0000.csv')).toBe(true);
+  });
+
+  it('reads a chunk back complete, however large the writes were', async () => {
+    const path = join(workDir, 'wide.txt');
+    const row = 'x'.repeat(100_000);
+    await writeFile(path, `${row}\n${row}\n${row}\n`);
+
+    const paths = await chunk(path, { rows: 2, into: join(workDir, 'chunks') });
+
+    // Backpressure: the chunk is closed before its path is handed back.
+    expect(await readFile(paths[0], 'utf-8')).toBe(`${row}\n${row}\n`);
+  });
+
+  it('normalises CRLF line endings', async () => {
+    const path = join(workDir, 'windows.txt');
+    await writeFile(path, 'row-0\r\nrow-1\r\n');
+
+    const paths = await chunk(path, { rows: 2, into: join(workDir, 'chunks') });
+
+    expect(await readFile(paths[0], 'utf-8')).toBe('row-0\nrow-1\n');
+  });
+
+  it('refuses an input with no rows rather than producing no chunks', async () => {
+    const path = join(workDir, 'empty.txt');
+    await writeFile(path, '');
+
+    await expect(
+      chunk(path, { rows: 2, into: join(workDir, 'chunks') }),
+    ).rejects.toThrow('holds no rows to chunk');
+  });
+
+  it('refuses a chunk size that is not a whole number of rows', async () => {
+    const input = await writeInput(2);
+
+    await expect(
+      chunk(input, { rows: 0, into: join(workDir, 'chunks') }),
+    ).rejects.toThrow('is not a number of rows to a chunk');
+  });
+});

From 339d54d6a5b6e916432964fa9fa36b362e493168 Mon Sep 17 00:00:00 2001
From: David de Boer <david@ddeboer.nl>
Date: Wed, 2 Sep 2026 11:16:33 +0200
Subject: [PATCH 2/2] fix(sparql-anything): report a failed chunk write instead
 of crashing on it
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

A review of chunk() found five things, four of them about what happens when
writing goes wrong:

- No 'error' listener was ever attached to the chunk file. once() covers only
  the instants an await is pending; a write that fails while the loop waits on
  the next line – ENOSPC part way through a multi-GB export – was an unhandled
  'error', which ends the process rather than this call.
- The open chunk was left open on any failure, keeping its handle and half a
  row, so a caller that caught and retried accumulated both.
- `extension` was concatenated unchecked, so 'csv' without its dot produced
  places-0000csv, which is exactly the naming the option exists to get right.
- Chunks of the same input from an earlier call were left in place, so a
  shorter re-run left a longer one's tail for a caller globbing the directory
  to pick up. Removed first, files only, and only those matching this input's
  own chunk names.

Closing a chunk now waits with finished() rather than once('close'): it
reports a stream that has already failed, and returns for one that has already
closed, where waiting for the event waits for one that will not come again.
That single reporting path replaced a stored error and a throw, which no test
could reach because the failure always arrived through an await first.

The fifth is documented rather than fixed: the input must hold data only, since
every line becomes a row. A file carrying its own header would repeat it inside
the first chunk, and no caller needs the alternative yet.
---
 docs/reference/sparql-anything.md           |   2 +
 packages/sparql-anything/src/chunk.ts       | 128 ++++++++++++++------
 packages/sparql-anything/test/chunk.test.ts |  71 ++++++++++-
 3 files changed, 164 insertions(+), 37 deletions(-)

diff --git a/docs/reference/sparql-anything.md b/docs/reference/sparql-anything.md
index 5415aa4b..262e8be4 100644
--- a/docs/reference/sparql-anything.md
+++ b/docs/reference/sparql-anything.md
@@ -130,6 +130,8 @@ Set `extension` for a tool that reads the format from the file name – SPARQL A
 
 **Splitting is by line**, so every record must be one line. A delimited format that wraps a field in quotes to carry a newline inside it would be cut in two; tab-separated exports, N-Triples and NDJSON are one record per line by definition. Line endings are normalised to `\n`.
 
+Chunks of the same input left by an earlier call are removed first, so a re-run cannot leave a longer run's tail behind for something to pick up. Only those: everything else in the directory is the caller's.
+
 An input with no rows is an error rather than an empty set of chunks: a step that produced an empty file has already failed.
 
 ## How a conversion runs
diff --git a/packages/sparql-anything/src/chunk.ts b/packages/sparql-anything/src/chunk.ts
index 07298852..cc10435f 100644
--- a/packages/sparql-anything/src/chunk.ts
+++ b/packages/sparql-anything/src/chunk.ts
@@ -1,6 +1,7 @@
 import { createReadStream, createWriteStream, WriteStream } from 'node:fs';
-import { mkdir } from 'node:fs/promises';
+import { mkdir, readdir, rm } from 'node:fs/promises';
 import { once } from 'node:events';
+import { finished } from 'node:stream/promises';
 import { createInterface } from 'node:readline';
 import { basename, extname, join } from 'node:path';
 
@@ -11,18 +12,25 @@ export interface ChunkOptions {
    * is what one process has to hold at once.
    */
   rows: number;
-  /** Directory the chunks are written to. Created if it does not exist. */
+  /**
+   * Directory the chunks are written to, created if it does not exist. Chunks
+   * of this input left by an earlier call are removed first, so a re-run
+   * cannot leave a longer run's tail behind for something to pick up.
+   */
   into: string;
   /**
    * Line repeated at the top of every chunk, for a format whose columns are
    * named. Leave it out for a format without a header, such as N-Triples.
+   *
+   * The input itself must hold data only: every line it has becomes a row, so
+   * a file that carries its own header would repeat it inside the first chunk.
    */
   header?: string;
   /**
-   * Extension for the chunk files, `'.csv'` and so on. Defaults to the input's
-   * own, and is worth setting for a tool that reads the format from the name –
-   * SPARQL Anything does, so a `.txt` export of a CSV has to be chunked as
-   * `.csv` to be read as one.
+   * Extension for the chunk files, leading dot included: `'.csv'`. Defaults to
+   * the input's own, and is worth setting for a tool that reads the format
+   * from the name – SPARQL Anything does, so a `.txt` export of a CSV has to
+   * be chunked as `.csv` to be read as one.
    */
   extension?: string;
 }
@@ -50,10 +58,17 @@ export async function chunk(
       `‘${rows}’ is not a number of rows to a chunk; give a whole number of one or more`,
     );
   }
-  await mkdir(into, { recursive: true });
-
   const extension = options.extension ?? extname(inputPath);
+  if (extension !== '' && !extension.startsWith('.')) {
+    throw new Error(
+      `‘${extension}’ is not an extension; give one with its leading dot, such as ‘.csv’`,
+    );
+  }
   const name = basename(inputPath, extname(inputPath));
+
+  await mkdir(into, { recursive: true });
+  await removeChunksOf(name, extension, into);
+
   const lines = createInterface({
     input: createReadStream(inputPath),
     crlfDelay: Infinity,
@@ -62,28 +77,55 @@ export async function chunk(
   const paths: string[] = [];
   let chunkFile: WriteStream | undefined;
   let rowsWritten = 0;
-  for await (const line of lines) {
-    if (chunkFile === undefined) {
-      const path = join(
-        into,
-        `${name}-${String(paths.length).padStart(4, '0')}${extension}`,
-      );
-      paths.push(path);
-      chunkFile = createWriteStream(path);
-      if (header !== undefined) {
-        await write(chunkFile, `${header}\n`);
+
+  const write = async (text: string): Promise<void> => {
+    if (!chunkFile!.write(text)) {
+      await once(chunkFile!, 'drain');
+    }
+  };
+
+  const closeChunk = async (): Promise<void> => {
+    const closing = chunkFile!;
+    chunkFile = undefined;
+    closing.end();
+    // finished(), not once('close'): it reports a stream that has already
+    // failed, and returns for one that has already closed, where waiting for
+    // the event would wait for one that will not come again.
+    await finished(closing);
+  };
+
+  try {
+    for await (const line of lines) {
+      if (chunkFile === undefined) {
+        const path = join(
+          into,
+          `${name}-${String(paths.length).padStart(4, '0')}${extension}`,
+        );
+        paths.push(path);
+        chunkFile = createWriteStream(path);
+        // A write can fail while this is waiting on the next line rather than
+        // on the stream, and an 'error' nobody listens for ends the process
+        // instead of this call. Stop reading; closing the chunk reports it.
+        chunkFile.on('error', () => lines.close());
+        if (header !== undefined) {
+          await write(`${header}\n`);
+        }
+      }
+      await write(`${line}\n`);
+      rowsWritten++;
+      if (rowsWritten === rows) {
+        await closeChunk();
+        rowsWritten = 0;
       }
     }
-    await write(chunkFile, `${line}\n`);
-    rowsWritten++;
-    if (rowsWritten === rows) {
-      await close(chunkFile);
-      chunkFile = undefined;
-      rowsWritten = 0;
+    if (chunkFile !== undefined) {
+      await closeChunk();
     }
-  }
-  if (chunkFile !== undefined) {
-    await close(chunkFile);
+  } finally {
+    // Whatever went wrong – a write, or the read that feeds it – the chunk
+    // still open would otherwise keep its handle and its half of a row.
+    chunkFile?.destroy();
+    lines.close();
   }
 
   if (paths.length === 0) {
@@ -95,15 +137,29 @@ export async function chunk(
   return paths;
 }
 
-/** Writes `text`, waiting for the stream to drain when it asks to. */
-async function write(chunkFile: WriteStream, text: string): Promise<void> {
-  if (!chunkFile.write(text)) {
-    await once(chunkFile, 'drain');
-  }
+/**
+ * Removes the chunks an earlier call made of this input. Only those: the
+ * directory is the caller's, and everything else in it is theirs.
+ */
+async function removeChunksOf(
+  name: string,
+  extension: string,
+  into: string,
+): Promise<void> {
+  const chunkFile = new RegExp(
+    `^${escapeForRegExp(name)}-\\d{4}${escapeForRegExp(extension)}$`,
+  );
+  const entries = await readdir(into, { withFileTypes: true });
+  await Promise.all(
+    entries
+      // Files only: something else of that name is not a chunk this made, and
+      // removing it is not this function's business.
+      .filter((entry) => entry.isFile() && chunkFile.test(entry.name))
+      .map((entry) => rm(join(into, entry.name), { force: true })),
+  );
 }
 
-/** Closes a chunk, so it is complete before its path is handed on. */
-async function close(chunkFile: WriteStream): Promise<void> {
-  chunkFile.end();
-  await once(chunkFile, 'close');
+/** Quotes the characters a file name may hold that a pattern would read. */
+function escapeForRegExp(value: string): string {
+  return value.replace(/[.*+?^${}()|[\]\\]/g, '\\$&');
 }
diff --git a/packages/sparql-anything/test/chunk.test.ts b/packages/sparql-anything/test/chunk.test.ts
index df78d85f..ddce13c9 100644
--- a/packages/sparql-anything/test/chunk.test.ts
+++ b/packages/sparql-anything/test/chunk.test.ts
@@ -1,6 +1,13 @@
 import { chunk } from '../src/index.js';
 import { describe, it, expect, beforeEach, afterEach } from 'vitest';
-import { mkdtemp, readdir, readFile, rm, writeFile } from 'node:fs/promises';
+import {
+  mkdir,
+  mkdtemp,
+  readdir,
+  readFile,
+  rm,
+  writeFile,
+} from 'node:fs/promises';
 import { join } from 'node:path';
 import { tmpdir } from 'node:os';
 
@@ -126,6 +133,68 @@ describe('chunk', () => {
     ).rejects.toThrow('holds no rows to chunk');
   });
 
+  it('surfaces a failure to write as a rejection, not a crash', async () => {
+    const input = await writeInput(4);
+    const into = join(workDir, 'chunks');
+    // A directory where the first chunk's file belongs: writing to it fails
+    // with EISDIR, and the failure arrives while the loop waits on a line.
+    await mkdir(join(into, 'places-0000.txt'), { recursive: true });
+
+    await expect(chunk(input, { rows: 2, into })).rejects.toThrow('EISDIR');
+  });
+
+  it('surfaces a write that fails while it is waiting on the next line', async () => {
+    const input = await writeInput(6);
+    const into = join(workDir, 'chunks');
+    // The third chunk cannot be opened, so the failure arrives once the first
+    // two have been written and the loop is reading again.
+    await mkdir(join(into, 'places-0002.txt'), { recursive: true });
+
+    await expect(chunk(input, { rows: 2, into })).rejects.toThrow('EISDIR');
+
+    // The chunks written before it are complete, not truncated.
+    expect(await readFile(join(into, 'places-0001.txt'), 'utf-8')).toBe(
+      'row-2\nrow-3\n',
+    );
+  });
+
+  it('leaves no chunk open when the input cannot be read', async () => {
+    await expect(
+      chunk(join(workDir, 'absent.txt'), {
+        rows: 2,
+        into: join(workDir, 'chunks'),
+      }),
+    ).rejects.toThrow('ENOENT');
+  });
+
+  it('removes the chunks an earlier call made of the same input', async () => {
+    const into = join(workDir, 'chunks');
+    await mkdir(into, { recursive: true });
+    // A longer run's tail, and a file of the caller's that is not ours.
+    await writeFile(join(into, 'places-0007.txt'), 'stale\n');
+    await writeFile(join(into, 'notes.txt'), 'keep me\n');
+    const input = await writeInput(2);
+
+    await chunk(input, { rows: 2, into });
+
+    expect((await readdir(into)).sort()).toEqual([
+      'notes.txt',
+      'places-0000.txt',
+    ]);
+  });
+
+  it('refuses an extension without its leading dot', async () => {
+    const input = await writeInput(2);
+
+    await expect(
+      chunk(input, {
+        rows: 2,
+        into: join(workDir, 'chunks'),
+        extension: 'csv',
+      }),
+    ).rejects.toThrow('is not an extension');
+  });
+
   it('refuses a chunk size that is not a whole number of rows', async () => {
     const input = await writeInput(2);