From 4e4948ee2273e167c4ba2b5e4007307cf74985ce Mon Sep 17 00:00:00 2001 From: Daniel Salazar Date: Tue, 13 Oct 2026 19:55:26 -0400 Subject: [PATCH] fix(fs): keep S3 read stream errors from crashing the process getObjectStream now attaches an 'error' listener to every body it returns. A body that fails before its consumer attaches (socket timeout, reset, or a checksum mismatch on a truncated read) used to emit 'error' with no listener and exit the process, failing every in-flight request on the task. It now logs a warning; consumers that attach later still receive the error. --- .../controllers/fs/FSController.http.test.ts | 112 +++++++++++++++++- .../core/http/middleware/puterSite.test.ts | 63 +++++++++- src/backend/drivers/util/fileInput.test.ts | 39 ++++++ src/backend/stores/fs/S3ObjectStore.test.ts | 97 +++++++++++++++ src/backend/stores/fs/S3ObjectStore.ts | 9 ++ 5 files changed, 318 insertions(+), 2 deletions(-) diff --git a/src/backend/controllers/fs/FSController.http.test.ts b/src/backend/controllers/fs/FSController.http.test.ts index cec71cfad..1b2255bd7 100644 --- a/src/backend/controllers/fs/FSController.http.test.ts +++ b/src/backend/controllers/fs/FSController.http.test.ts @@ -17,7 +17,9 @@ * along with this program. If not, see . */ -import { afterAll, beforeAll, describe, expect, it } from 'vitest'; +import { GetObjectCommand } from '@aws-sdk/client-s3'; +import { Readable } from 'node:stream'; +import { afterAll, beforeAll, describe, expect, it, vi } from 'vitest'; import { setupPuterTestEnv, type PuterTestEnv } from '../../testUtil.js'; /** @@ -549,3 +551,111 @@ describe('POST /fs/startWrite storage quota over HTTP', () => { expect(responseBody.code).toBe('bad_request'); }); }); + +describe('file reads over HTTP when storage fails mid-download', () => { + let env: PuterTestEnv; + + beforeAll(async () => { + env = await setupPuterTestEnv(); + }, 120_000); + + afterAll(async () => { + await env?.shutdown(); + }); + + const writeFile = async (name: string, content: string) => { + const { username } = env.users.user; + const user = await env.server.stores.user.getByUsername(username); + const body = Buffer.from(content); + const { fsEntry } = await env.server.services.fs.write(user!.id, { + fileMetadata: { + path: `/${username}/Documents/${name}`, + size: body.byteLength, + contentType: 'text/plain', + }, + fileContent: body, + }); + return fsEntry; + }; + + // Storage hands over one chunk and then fails, the way a dropped + // connection or a checksum mismatch surfaces on a read in flight. + const breakReadsOf = (objectKey: string) => { + const s3 = env.server.stores.s3Object; + const client = env.server.clients.s3.get(s3.resolveRegion(null)); + const send = client.send.bind(client); + return vi.spyOn(client, 'send').mockImplementation((( + command: unknown, + ) => { + if ( + !(command instanceof GetObjectCommand) || + command.input.Key !== objectKey + ) { + return send(command as never); + } + let reads = 0; + const body = new Readable({ + read() { + reads += 1; + if (reads === 1) this.push('partial'); + // Late enough for the headers and first chunk to go out. + else if (reads === 2) + setTimeout( + () => this.destroy(new Error('Checksum mismatch')), + 25, + ); + }, + }); + return Promise.resolve({ + Body: body, + ContentLength: 64, + ContentType: 'text/plain', + }); + }) as never); + }; + + it.each([ + ['/fs/read', 'path'], + ['/read', 'file'], + ])( + 'GET %s fails that response and keeps serving others', + async (route, param) => { + const { username, token } = env.users.user; + const urlFor = (name: string) => { + const url = new URL(route, env.apiOrigin); + url.searchParams.set(param, `/${username}/Documents/${name}`); + return url; + }; + const suffix = route.replace(/\W/g, ''); + const broken = await writeFile( + `broken-${suffix}.txt`, + 'x'.repeat(64), + ); + await writeFile(`healthy-${suffix}.txt`, 'still fine'); + + const send = breakReadsOf(broken.uuid); + let status = 0; + let completed = false; + try { + const response = await fetch(urlFor(`broken-${suffix}.txt`), { + headers: { Authorization: `Bearer ${token}` }, + }); + status = response.status; + await response.arrayBuffer(); + completed = true; + } catch { + // Expected: the transfer is cut short. + } finally { + send.mockRestore(); + } + expect(status).toBe(200); + expect(completed).toBe(false); + + const healthy = await fetch(urlFor(`healthy-${suffix}.txt`), { + headers: { Authorization: `Bearer ${token}` }, + }); + expect(healthy.status).toBe(200); + expect(await healthy.text()).toBe('still fine'); + }, + ); +}); diff --git a/src/backend/core/http/middleware/puterSite.test.ts b/src/backend/core/http/middleware/puterSite.test.ts index da20f6418..311e8a9c9 100644 --- a/src/backend/core/http/middleware/puterSite.test.ts +++ b/src/backend/core/http/middleware/puterSite.test.ts @@ -17,7 +17,8 @@ * along with this program. If not, see . */ -import { Writable } from 'node:stream'; +import { GetObjectCommand } from '@aws-sdk/client-s3'; +import { Readable, Writable } from 'node:stream'; import type { Request, Response } from 'express'; import { afterAll, beforeAll, describe, expect, it, vi } from 'vitest'; import { v4 as uuidv4 } from 'uuid'; @@ -654,6 +655,66 @@ describe('createPuterSiteMiddleware — file serving', () => { expect(out.headers['Accept-Ranges']).toBe('bytes'); }); + it('cuts the response off when storage fails mid-download', async () => { + const owner = await makeUserWithHome(); + const homePath = `/${owner.username}`; + const homeEntry = await server.stores.fsEntry.getEntryByPath(homePath); + const sub = `cut-${Math.random().toString(36).slice(2, 8)}`; + await server.stores.subdomain.create({ + userId: owner.id, + subdomain: sub, + rootDirId: homeEntry!.id, + }); + const entry = await writeFile( + owner.id, + `${homePath}/movie.bin`, + Buffer.alloc(64, 0x42), + ); + + // One chunk, then the storage connection drops. + let reads = 0; + const body = new Readable({ + read() { + reads += 1; + if (reads === 1) this.push('partial'); + else this.destroy(new Error('aborted')); + }, + }); + const client = server.clients.s3.get( + server.stores.s3Object.resolveRegion(entry!.bucketRegion), + ); + const send = client.send.bind(client); + const spy = vi + .spyOn(client, 'send') + .mockImplementation(((command: unknown) => + command instanceof GetObjectCommand + ? Promise.resolve({ Body: body, ContentLength: 64 }) + : send(command as never)) as never); + + const mw = buildMiddleware(); + const { res } = makeRes(); + // A real response is torn down through its socket and never emits + // 'error' itself; the stand-in Writable does, so absorb it. + res.on('error', () => undefined); + try { + await mw( + makeReq({ + hostname: `${sub}.site.puter.localhost`, + path: '/movie.bin', + }), + res, + vi.fn(), + ); + await new Promise((resolve) => setImmediate(resolve)); + } finally { + spy.mockRestore(); + } + + expect(res.destroyed).toBe(true); + expect(res.writableFinished).toBe(false); + expect(body.destroyed).toBe(true); + }); + it("rewrites a trailing slash request to /index.html under the site's root", async () => { const owner = await makeUserWithHome(); const homePath = `/${owner.username}`; diff --git a/src/backend/drivers/util/fileInput.test.ts b/src/backend/drivers/util/fileInput.test.ts index 968244ad0..7f5801050 100644 --- a/src/backend/drivers/util/fileInput.test.ts +++ b/src/backend/drivers/util/fileInput.test.ts @@ -17,6 +17,8 @@ * along with this program. If not, see . */ +import { GetObjectCommand } from '@aws-sdk/client-s3'; +import { Readable } from 'node:stream'; import { afterAll, beforeAll, describe, expect, it, vi } from 'vitest'; import { v4 as uuidv4 } from 'uuid'; import type { Actor } from '../../core/actor.js'; @@ -420,6 +422,43 @@ describe('loadFileInput FS path', () => { legacyCode: 'input_too_large', }); }); + + it('rejects the load when storage fails partway through the read', async () => { + const { actor, userId } = await makeUser(); + const path = `/${actor.user!.username!}/Documents/cut-off.bin`; + const entry = await withActor(actor, () => + writeFile(userId, path, Buffer.alloc(64, 0x42)), + ); + // The storage client is the boundary: one chunk, then the + // connection drops. + const client = server.clients.s3.get( + server.stores.s3Object.resolveRegion(entry.bucketRegion), + ); + const send = client.send.bind(client); + const spy = vi.spyOn(client, 'send').mockImplementation((( + command: unknown, + ) => { + if (!(command instanceof GetObjectCommand)) { + return send(command as never); + } + let reads = 0; + const body = new Readable({ + read() { + reads += 1; + if (reads === 1) this.push(Buffer.alloc(16, 0x42)); + else this.destroy(new Error('aborted')); + }, + }); + return Promise.resolve({ Body: body, ContentLength: 64 }); + }) as never); + try { + await expect(callLoadFileInput(actor, path)).rejects.toThrow( + 'aborted', + ); + } finally { + spy.mockRestore(); + } + }); }); // ── resolveFileInput ─────────────────────────────────────────────── diff --git a/src/backend/stores/fs/S3ObjectStore.test.ts b/src/backend/stores/fs/S3ObjectStore.test.ts index 38112b7d5..860ce3461 100644 --- a/src/backend/stores/fs/S3ObjectStore.test.ts +++ b/src/backend/stores/fs/S3ObjectStore.test.ts @@ -1277,6 +1277,103 @@ describe('S3ObjectStore failure handling', () => { expect(result.lastModified).toBeNull(); }); + // An 'error' event with no listener is rethrown as an uncaught + // exception, so these watch the process as well as the reader. + const collectUncaught = async (run: () => Promise) => { + const uncaught: unknown[] = []; + const record = (error: unknown) => uncaught.push(error); + process.on('uncaughtException', record); + try { + await run(); + await new Promise((resolve) => setImmediate(resolve)); + } finally { + process.off('uncaughtException', record); + } + return uncaught; + }; + + it('keeps a body that fails before anyone reads it from crashing the process', async () => { + const body = new Readable({ read() {} }); + const { store: fake } = makeFakeStore({ + respond: () => ({ Body: body, ContentLength: 10 }), + }); + const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}); + + let result: Awaited>; + const uncaught = await collectUncaught(async () => { + result = await fake.getObjectStream( + { bucket, objectKey: 'k' }, + region, + ); + // The connection drops while the caller is busy elsewhere. + body.destroy(new Error('aborted')); + }); + + expect(uncaught).toEqual([]); + expect(warn).toHaveBeenCalledWith( + '[s3] object stream failed with no reader', + expect.objectContaining({ message: 'aborted' }), + ); + // A reader that turns up afterwards still gets the failure. + await expect(readAll(result!.body)).rejects.toThrow('aborted'); + warn.mockRestore(); + }); + + it('fails the reader, and only the reader, when the body breaks mid-read', async () => { + let reads = 0; + const body = new Readable({ + read() { + reads += 1; + if (reads === 1) this.push('partial'); + else this.destroy(new Error('Checksum mismatch')); + }, + }); + const { store: fake } = makeFakeStore({ + respond: () => ({ Body: body, ContentLength: 10 }), + }); + const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}); + + const uncaught = await collectUncaught(async () => { + const result = await fake.getObjectStream( + { bucket, objectKey: 'k' }, + region, + ); + await expect(readAll(result.body)).rejects.toThrow( + 'Checksum mismatch', + ); + }); + + expect(uncaught).toEqual([]); + // Someone was reading, so the failure is theirs to report. + expect(warn).not.toHaveBeenCalled(); + warn.mockRestore(); + }); + + it('fails the reader when a web stream body errors mid-read', async () => { + const { store: fake } = makeFakeStore({ + respond: () => ({ + Body: new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode('part')); + controller.error(new Error('connection reset')); + }, + }), + }), + }); + + const uncaught = await collectUncaught(async () => { + const result = await fake.getObjectStream( + { bucket, objectKey: 'k' }, + region, + ); + await expect(readAll(result.body)).rejects.toThrow( + 'connection reset', + ); + }); + + expect(uncaught).toEqual([]); + }); + it('reports a null size when S3 omits ContentLength on HEAD', async () => { const { store: fake } = makeFakeStore({ respond: () => ({}) }); await expect( diff --git a/src/backend/stores/fs/S3ObjectStore.ts b/src/backend/stores/fs/S3ObjectStore.ts index ded327de6..adf844c21 100644 --- a/src/backend/stores/fs/S3ObjectStore.ts +++ b/src/backend/stores/fs/S3ObjectStore.ts @@ -649,6 +649,15 @@ export class S3ObjectStore extends PuterStore { : Readable.fromWeb( body as unknown as import('node:stream/web').ReadableStream, ); + // The body can fail before its consumer attaches (the connection drops + // while the caller is still awaiting something else), and an 'error' + // with no listener exits the process. Consumers still get the error + // through their own listener, pipeline or iterator. + stream.on('error', (err) => { + if (stream.listenerCount('error') === 1) { + console.warn('[s3] object stream failed with no reader', err); + } + }); return { body: stream,