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.

src/backend/controllers/fs/FSController.http.test.ts

@@ -17,7 +17,9 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
-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');
+        },
+    );
+});

src/backend/core/http/middleware/puterSite.test.ts

@@ -17,7 +17,8 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
-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<void>((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}`;

src/backend/drivers/util/fileInput.test.ts

@@ -17,6 +17,8 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
+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 ───────────────────────────────────────────────

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<void>) => {
+        const uncaught: unknown[] = [];
+        const record = (error: unknown) => uncaught.push(error);
+        process.on('uncaughtException', record);
+        try {
+            await run();
+            await new Promise<void>((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<ReturnType<S3ObjectStore['getObjectStream']>>;
+        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(

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,