From 5078b8a87475a36d61af29564feb303a5f36b39d Mon Sep 17 00:00:00 2001 From: Sabyasachi Date: Fri, 14 Aug 2026 12:09:49 +0000 Subject: [PATCH] fix(file-store): make read failures reject instead of surfacing late MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit GCS getAsStream returned createReadStream() without any I/O, so a missing object resolved successfully and failed later as a stream 'error'. A caller doing reply.send(await getAsStream(k)) sent a 200 with a broken body, or crashed the process on an unhandled error. Every other provider rejects. GCS is the production backend. It now checks existence first. Plugin dispatch read Configure[opts.type] straight off an object literal, so type:"toString" resolved Object.prototype.toString and registered successfully with no FileStore decorated — a typo became a silent no-op boot rather than a failure. streamToBuffer only handled data/error/end. A stream destroyed without an error emits close alone, leaving the promise pending forever and hanging the request that awaited getAsBuffer. It now rejects, and a settled guard keeps late events from settling twice. Local getAsBuffer/getAsStream wrapped reads in if (await stat(p)), which can never be falsy since stat throws — so the File not found branch was unreachable and callers got raw ENOENT. Two specs reached it only by mocking stat to resolve undefined, a state the real filesystem never produces, and asserted an error the code cannot throw. getAsBuffer now reads directly; getAsStream keeps the stat so a missing file rejects rather than deferring to a stream error, matching the GCS fix. Co-Authored-By: Claude Opus 5 (1M context) --- src/file-store.gcs.spec.ts | 12 ++++++++ src/file-store.local.spec.ts | 32 +++++++++++++------- src/file-store.ts | 31 ++++++++++++------- src/utils.spec.ts | 58 +++++++++++++++++++++++++++++++----- src/utils.ts | 24 ++++++++++++--- 5 files changed, 125 insertions(+), 32 deletions(-) diff --git a/src/file-store.gcs.spec.ts b/src/file-store.gcs.spec.ts index e423b18..7a48193 100644 --- a/src/file-store.gcs.spec.ts +++ b/src/file-store.gcs.spec.ts @@ -118,10 +118,22 @@ describe("GCPFileStore", () => { describe("getAsStream", () => { it("should return the read stream", async () => { const rs = Readable.from(["data"]); + mockFile.exists.mockResolvedValueOnce([true]); mockFile.createReadStream.mockReturnValueOnce(rs); expect(await store.getAsStream("test.txt")).toBe(rs); }); + + // createReadStream does no I/O, so without an existence check a missing object + // resolves and fails later as a stream 'error'. Every other provider rejects. + it("rejects for a missing object instead of returning a doomed stream", async () => { + mockFile.exists.mockResolvedValueOnce([false]); + + await expect(store.getAsStream("missing.txt")).rejects.toThrow( + "File not found: missing.txt", + ); + expect(mockFile.createReadStream).not.toHaveBeenCalled(); + }); }); describe("copyFromStream", () => { diff --git a/src/file-store.local.spec.ts b/src/file-store.local.spec.ts index cf03d90..afec7ac 100644 --- a/src/file-store.local.spec.ts +++ b/src/file-store.local.spec.ts @@ -75,6 +75,19 @@ describe("LocalFileStore", () => { fastify.register(FileStorePlugin, { type: "nope" as any }), ).rejects.toThrow("Unknown storage type: nope"); }); + + // A plain lookup finds Object.prototype.toString and registers with no FileStore + // decorated — a typo becoming a silent no-op boot instead of a failure. + it.each(["toString", "constructor", "valueOf", "hasOwnProperty"])( + "rejects the inherited Object.prototype name %s", + async (type) => { + const f = Fastify({ logger: false }); + await expect( + f.register(FileStorePlugin, { type: type as never }), + ).rejects.toThrow(`Unknown storage type: ${type}`); + await f.close(); + }, + ); }); describe("exists", () => { @@ -165,12 +178,11 @@ describe("LocalFileStore", () => { expect((await store.getAsBuffer("read.txt")).toString()).toBe("buffered"); }); - it("throws File not found when stat yields nothing", async () => { + it("rejects with ENOENT for a missing file", async () => { const store = await register(); - jest.spyOn(fs.promises, "stat").mockResolvedValue(undefined); - await expect(store.getAsBuffer("ghost.txt")).rejects.toThrow( - `File not found: ${path.join(tempDir, "ghost.txt")}`, - ); + await expect(store.getAsBuffer("ghost.txt")).rejects.toMatchObject({ + code: "ENOENT", + }); }); }); @@ -182,12 +194,12 @@ describe("LocalFileStore", () => { expect((await streamToBuffer(rs)).toString()).toBe("streamed"); }); - it("throws File not found when stat yields nothing", async () => { + // Rejects rather than deferring to a stream 'error': createReadStream is lazy. + it("rejects with ENOENT for a missing file", async () => { const store = await register(); - jest.spyOn(fs.promises, "stat").mockResolvedValue(undefined); - await expect(store.getAsStream("ghost.txt")).rejects.toThrow( - `File not found: ${path.join(tempDir, "ghost.txt")}`, - ); + await expect(store.getAsStream("ghost.txt")).rejects.toMatchObject({ + code: "ENOENT", + }); }); }); diff --git a/src/file-store.ts b/src/file-store.ts index f08bd80..9f48012 100644 --- a/src/file-store.ts +++ b/src/file-store.ts @@ -79,7 +79,11 @@ const Configure = { const plugin: FastifyPluginAsync<{ type: keyof typeof Configure; }> = async function (f, opts): Promise { - const configure = Configure[opts.type]; + // hasOwn, not a plain lookup: `Configure["toString"]` finds Object.prototype.toString + // and registers with no FileStore decorated, turning a typo into a silent no-op boot. + const configure = Object.prototype.hasOwnProperty.call(Configure, opts.type) + ? Configure[opts.type] + : undefined; if (!configure) { throw new Error(`Unknown storage type: ${opts.type}`); } @@ -127,11 +131,10 @@ class LocalFileStore implements FileStore { await fs.promises.writeFile(p, data); } async getAsBuffer(filepath: string): Promise { - const p = path.join(this.dir, filepath); - if (await fs.promises.stat(p)) { - return fs.promises.readFile(p); - } - throw new Error(`File not found: ${p}`); + // No stat guard: a fulfilled stat is never falsy, so the old "File not found" + // branch was unreachable and readFile already rejects with ENOENT. Dropping it + // also removes a redundant stat and the TOCTOU window between the two calls. + return fs.promises.readFile(path.join(this.dir, filepath)); } async copyFromLocalFile( filepath: string, @@ -145,10 +148,10 @@ class LocalFileStore implements FileStore { } async getAsStream(filepath: string): Promise { const p = path.join(this.dir, filepath); - if (await fs.promises.stat(p)) { - return fs.createReadStream(p); - } - throw new Error(`File not found: ${p}`); + // stat first so a missing file rejects the promise rather than surfacing later as + // a stream 'error' — createReadStream is lazy, like the GCS reader. + await fs.promises.stat(p); + return fs.createReadStream(p); } async copyFromStream( filepath: string, @@ -302,6 +305,14 @@ class GCPFileStore implements FileStore { async getAsStream(filepath: string): Promise { const gcsfile = this.storage.bucket(this.bucket).file(filepath); + // createReadStream does no I/O before returning, so without this check a missing + // object resolves and only fails later as a stream 'error' — which a caller doing + // `reply.send(await getAsStream(k))` turns into a 200 with a broken body, or an + // unhandled 'error' that takes the process down. Every other provider rejects. + const [exists] = await gcsfile.exists(); + if (!exists) { + throw new Error(`File not found: ${filepath}`); + } return gcsfile.createReadStream(); } diff --git a/src/utils.spec.ts b/src/utils.spec.ts index 9e40900..5462bf5 100644 --- a/src/utils.spec.ts +++ b/src/utils.spec.ts @@ -1,3 +1,4 @@ +import { EventEmitter } from "node:events"; import { Readable } from "node:stream"; import { DataStream, streamToBuffer } from "./utils"; @@ -163,15 +164,56 @@ describe("Utils", () => { }); }); - describe("DataStream interface", () => { - it("should define the correct interface structure", () => { - // This is a compile-time test to ensure the interface is properly defined - const mockStream: DataStream = { - on: jest.fn().mockReturnThis(), - }; + describe("premature close", () => { + // A stream destroyed without an error emits only 'close'. Before this was handled + // the promise stayed pending forever and hung the request awaiting getAsBuffer. + it("rejects when the stream is destroyed without an error", async () => { + const stream = new Readable({ read() {} }); + stream.push("partial"); + setImmediate(() => stream.destroy()); + + await expect(streamToBuffer(stream)).rejects.toThrow( + "stream closed before end", + ); + }); + + it("rejects with the original error when destroyed with one", async () => { + const stream = new Readable({ read() {} }); + setImmediate(() => stream.destroy(new Error("boom"))); + + await expect(streamToBuffer(stream)).rejects.toThrow("boom"); + }); + + it("still resolves normally when close follows end", async () => { + const stream: DataStream = Readable.from(["a", "b"]); + + expect((await streamToBuffer(stream)).toString()).toBe("ab"); + }); + + // Late events must not settle the promise a second time. EventEmitter is used + // directly so the exact sequence can be driven; a real Readable will not emit + // 'error' after 'end'. + it("ignores an error arriving after end", async () => { + const em = new EventEmitter() as unknown as DataStream; + const p = streamToBuffer(em); + const e = em as unknown as EventEmitter; + e.emit("data", Buffer.from("ok")); + e.emit("end"); + e.emit("error", new Error("too late")); + e.emit("close"); + + expect((await p).toString()).toBe("ok"); + }); + + it("ignores end and close arriving after an error", async () => { + const em = new EventEmitter() as unknown as DataStream; + const p = streamToBuffer(em); + const e = em as unknown as EventEmitter; + e.emit("error", new Error("first")); + e.emit("end"); + e.emit("close"); - expect(typeof mockStream.on).toBe("function"); - expect(mockStream.on("test", () => {})).toBe(mockStream); + await expect(p).rejects.toThrow("first"); }); }); }); diff --git a/src/utils.ts b/src/utils.ts index 6468f0d..346b7a7 100644 --- a/src/utils.ts +++ b/src/utils.ts @@ -8,13 +8,29 @@ export interface DataStream { export function streamToBuffer(stream: DataStream): Promise { return new Promise((resolve, reject) => { const chunks: Buffer[] = []; + let settled = false; + stream.on("data", (chunk: Buffer | string) => chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)), ); // Typed as Error so the rejection carries a stack; node streams always emit one. - stream.on("error", (err: Error) => reject(err)); - stream.on("end", () => - resolve(chunks.length === 1 ? chunks[0] : Buffer.concat(chunks)), - ); + stream.on("error", (err: Error) => { + if (settled) return; + settled = true; + reject(err); + }); + stream.on("end", () => { + if (settled) return; + settled = true; + resolve(chunks.length === 1 ? chunks[0] : Buffer.concat(chunks)); + }); + // A stream destroyed without an error argument emits only 'close', which would + // otherwise leave this promise pending forever and hang the awaiting request. + // 'close' follows 'end' in the normal path, where the settled flag ignores it. + stream.on("close", () => { + if (settled) return; + settled = true; + reject(new Error("stream closed before end")); + }); }); }