diff --git a/src/reference/ReferenceSourceReaderFile.test.ts b/src/reference/ReferenceSourceReaderFile.test.ts new file mode 100644 index 000000000..9f4a20dcd --- /dev/null +++ b/src/reference/ReferenceSourceReaderFile.test.ts @@ -0,0 +1,70 @@ +import { expect, it } from "vitest"; + +import { readFileContents } from "./ReferenceSourceReaderFile.js"; + +const reader = (bytes: Buffer, maximumRead = bytes.length) => { + let position = 0; + return { + read(buffer: Buffer, offset: number, length: number) { + const end = Math.min( + bytes.length, + position + length, + position + maximumRead, + ); + const bytesRead = bytes.copy(buffer, offset, position, end); + position = end; + return Promise.resolve({ bytesRead }); + }, + }; +}; + +it("retains all bytes across short reads through the declared size", async () => { + const bytes = Buffer.from("complete observation"); + const result = await readFileContents({ + handle: reader(bytes, 3), + path: "source.txt", + expectedSize: BigInt(bytes.length), + }); + if (result.status === "failed") throw result.entry; + expect(Buffer.concat(result.chunks, result.total)).toEqual(bytes); +}); + +it("reports growth instead of continuing to consume a changed file", async () => { + const result = await readFileContents({ + handle: reader(Buffer.from("larger than admitted")), + path: "source.txt", + expectedSize: 1n, + }); + expect(result).toEqual({ + status: "failed", + entry: { + status: "failed", + kind: "file", + path: "source.txt", + code: "changed", + message: "File grew while it was read", + size: 2, + }, + }); +}); + +it("preserves cancellation that arrives during a descriptor read", async () => { + const controller = new AbortController(); + const descriptor = reader(Buffer.from("observed")); + const result = await readFileContents({ + handle: { + async read(buffer, offset, length) { + const read = await descriptor.read(buffer, offset, length); + controller.abort(); + return read; + }, + }, + path: "source.txt", + expectedSize: 8n, + signal: controller.signal, + }); + expect(result).toMatchObject({ + status: "failed", + entry: { code: "cancelled", path: "source.txt" }, + }); +}); diff --git a/src/reference/ReferenceSourceReaderFile.ts b/src/reference/ReferenceSourceReaderFile.ts index b9c483e5d..11ee5f287 100644 --- a/src/reference/ReferenceSourceReaderFile.ts +++ b/src/reference/ReferenceSourceReaderFile.ts @@ -77,9 +77,18 @@ const prepareFileRead = async ( return { status: "ready", before }; }; -const readFileContents = async (request: { - readonly handle: FileHandle; +/** Read a stable-size file through its borrowed descriptor; growth is a change. */ +export const readFileContents = async (request: { + readonly handle: { + read( + buffer: Buffer, + offset: number, + length: number, + position: null, + ): Promise<{ readonly bytesRead: number }>; + }; readonly path: string; + readonly expectedSize: bigint; readonly signal?: AbortSignal; }): Promise => { const { handle, path, signal } = request; @@ -97,7 +106,13 @@ const readFileContents = async (request: { total, ), }; - const chunk = Buffer.allocUnsafe(READ_CHUNK_BYTES); + // One extra byte detects growth without allocating a full chunk at EOF. + const remaining = request.expectedSize - BigInt(total); + const chunkBytes = + remaining >= BigInt(READ_CHUNK_BYTES) + ? READ_CHUNK_BYTES + : Number(remaining) + 1; + const chunk = Buffer.allocUnsafe(chunkBytes); const read = await handle.read(chunk, 0, chunk.byteLength, null); if (isAborted(signal)) return { @@ -112,6 +127,17 @@ const readFileContents = async (request: { }; if (read.bytesRead === 0) break; total += read.bytesRead; + if (BigInt(total) > request.expectedSize) + return { + status: "failed", + entry: entryFailure( + path, + "file", + "changed", + "File grew while it was read", + total, + ), + }; chunks.push(chunk.subarray(0, read.bytesRead)); } return { status: "ok", chunks, total }; @@ -189,6 +215,7 @@ export const readStableFile = async ( const contents = await readFileContents({ handle, path, + expectedSize: prepared.before.size, ...(signal === undefined ? {} : { signal }), }); if (contents.status === "failed") return contents.entry;