fix: don't let an aborted upload take down the process (#3557)

src/backend/index.ts

@@ -208,4 +208,6 @@ if (require.main === module) {
     };
     process.on('SIGINT', shutDownProcess);
     process.on('SIGTERM', shutDownProcess);
+    // Uncaught exceptions and unhandled rejections are reported by the guards
+    // `PuterServer.start()` installs — see util/processGuards.ts.
 }

src/backend/server.ts

@@ -51,6 +51,7 @@ import { guiOriginGate } from './core/http/middleware/originGate';
 import { requireCreditsGate } from './core/http/middleware/credits';
 import { createStepUpGate } from './core/http/middleware/stepUpSession';
 import { createNotFoundHandler } from './core/http/middleware/notFoundHandler';
+import { installProcessGuards } from './util/processGuards';
 import {
     requireAntiCsrf,
     setAntiCsrfRedis,
@@ -102,6 +103,7 @@ export class PuterServer {
     #config: IConfig;
     #app!: ReturnType<typeof express>;
     #server: ReturnType<ReturnType<typeof express>['listen']> | null = null;
+    #removeProcessGuards: (() => void) | null = null;
 
     #ready: Promise<boolean>;
 
@@ -1230,6 +1232,13 @@ export class PuterServer {
     async start(noHttpServer = false) {
         await this.#ready;
 
+        // Installed before anything starts serving, so a fault during boot is
+        // reported too. Logging is unconditional; whether an uncaught exception
+        // ends the process is a deployment decision, hence the config gate.
+        this.#removeProcessGuards = installProcessGuards({
+            keepAliveOnUncaught: this.#config.keep_alive_on_uncaught ?? false,
+        });
+
         // Create the http server explicitly (instead of `app.listen()`) so we
         // have the server reference BEFORE listen starts — anything that needs
         // to hook into the raw server (socket.io upgrades, WebSockets, …) runs
@@ -1410,6 +1419,8 @@ export class PuterServer {
     }
 
     async shutdown() {
+        this.#removeProcessGuards?.();
+        this.#removeProcessGuards = null;
         if (this.#server) {
             console.log('PuterServer is shutting down');
             // Prepare hooks come first: SocketService's hook closes

src/backend/services/fs/FSService.test.ts

@@ -422,6 +422,43 @@ describe('FSService write payload handling', () => {
         expect(error.statusCode).toBe(400);
         expect(error.message).toBe('Unsupported file content payload');
     });
+
+    it('fails only the request when the source aborts mid-stream', async () => {
+        // A client disconnecting mid-upload destroys the source. The byte
+        // counter sitting between it and the upload must not turn that into an
+        // unhandled 'error' event, which would end the whole process.
+        const uncaught: unknown[] = [];
+        const onUncaught = (error: unknown) => uncaught.push(error);
+        process.on('uncaughtException', onUncaught);
+
+        const source = new Readable({
+            read() {
+                this.push('partial');
+                this.destroy(
+                    Object.assign(new Error('aborted'), {
+                        code: 'ECONNRESET',
+                    }),
+                );
+            },
+        });
+
+        try {
+            const outcome = await write('aborted.bin', source).then(
+                () => null,
+                (error: unknown) => error,
+            );
+            expect(outcome).toBeInstanceOf(Error);
+            // Drain the microtask and nextTick queues so a stray 'error'
+            // event has somewhere to land before the assertion below.
+            await new Promise((resolve) => setImmediate(resolve));
+        } finally {
+            process.off('uncaughtException', onUncaught);
+        }
+
+        expect(uncaught).toEqual([]);
+        // Generous timeout: the object-store client retries the torn-off body
+        // before giving up, which puts the rejection just past the default.
+    }, 20_000);
 });
 
 describe('FSService overwrite and dedupe resolution', () => {

src/backend/services/fs/FSService.ts

@@ -20,7 +20,7 @@
 import { createHash } from 'node:crypto';
 import { posix as pathPosix } from 'node:path';
 import type { TransformCallback } from 'node:stream';
-import { Readable, Transform } from 'node:stream';
+import { pipeline, Readable, Transform } from 'node:stream';
 import { v4 as uuidv4 } from 'uuid';
 import {
     BinaryPayload,
@@ -886,10 +886,18 @@ export class FSService extends PuterService {
             },
         });
 
-        source.on('error', (error) => {
-            countingStream.destroy(error);
+        // `pipeline` rather than `pipe` plus a hand-rolled 'error' forward: it
+        // keeps an 'error' listener attached to `countingStream` for the whole
+        // lifetime of the stream, and tears down both ends whichever one fails.
+        // The consumer is an object-store upload that may not have subscribed
+        // yet when a client disconnects mid-request, and destroying a Transform
+        // that nobody is listening to raises an unhandled 'error' event — which
+        // ends the process rather than just the request.
+        pipeline(source, countingStream, (error) => {
+            if (error) {
+                console.warn('upload stream ended early:', error.message);
+            }
         });
-        source.pipe(countingStream);
 
         return {
             stream: countingStream,

src/backend/types.ts

@@ -239,7 +239,12 @@ export interface IPreludeConfig {
      * an RCS agent provisioned in the Prelude account to actually use RCS.
      */
     preferredChannel?:
-        'sms' | 'rcs' | 'whatsapp' | 'viber' | 'zalo' | 'telegram';
+        | 'sms'
+        | 'rcs'
+        | 'whatsapp'
+        | 'viber'
+        | 'zalo'
+        | 'telegram';
 }
 
 /**
@@ -631,6 +636,14 @@ interface IConfigOptional {
      * (the default) leaves console output human-readable for local/dev.
      */
     log_format: 'json' | 'text';
+    /**
+     * Keep serving after an uncaught exception instead of exiting. Uncaught
+     * exceptions are always logged either way; this only decides whether one
+     * ends the process. Default: false, matching Node's own behavior. Set it
+     * where losing the node costs more than running a possibly-degraded one — a
+     * small pool behind a health check that replaces bad nodes anyway.
+     */
+    keep_alive_on_uncaught: boolean;
     /** Server version. Falls back to `npm_package_version`. */
     version: string;
     /**
@@ -1026,8 +1039,7 @@ export interface WithLifecycle extends Object {
 }
 
 export interface WithCostsReporting extends WithLifecycle {
-    getReportedCosts?: () =>
-        // eslint-disable-next-line @typescript-eslint/no-explicit-any
+    getReportedCosts?: () => // eslint-disable-next-line @typescript-eslint/no-explicit-any
         | Promise<Record<string, any>[]>
         // eslint-disable-next-line @typescript-eslint/no-explicit-any
         | Record<string, any>[];

src/backend/util/processGuards.test.ts

@@ -0,0 +1,122 @@
+/*
+ * Copyright (C) 2024-present Puter Technologies Inc.
+ *
+ * This file is part of Puter.
+ *
+ * Puter is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published
+ * by the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+import { afterEach, describe, expect, it, vi } from 'vitest';
+import { installProcessGuards } from './processGuards.js';
+
+// The guards mutate global process state, so every test tears its own down and
+// asserts against listener counts taken before installing.
+const counts = () => ({
+    monitor: process.listenerCount('uncaughtExceptionMonitor'),
+    uncaught: process.listenerCount('uncaughtException'),
+    rejection: process.listenerCount('unhandledRejection'),
+});
+
+let uninstall: (() => void) | null = null;
+
+afterEach(() => {
+    uninstall?.();
+    uninstall = null;
+    vi.restoreAllMocks();
+});
+
+describe('installProcessGuards', () => {
+    it('watches exceptions without suppressing them by default', () => {
+        const before = counts();
+        uninstall = installProcessGuards();
+        const after = counts();
+
+        // The monitor observes; it does not stop Node from exiting.
+        expect(after.monitor).toBe(before.monitor + 1);
+        expect(after.uncaught).toBe(before.uncaught);
+        expect(after.rejection).toBe(before.rejection);
+    });
+
+    it('suppresses the exit only when asked to keep serving', () => {
+        const before = counts();
+        uninstall = installProcessGuards({ keepAliveOnUncaught: true });
+        const after = counts();
+
+        expect(after.monitor).toBe(before.monitor + 1);
+        expect(after.uncaught).toBe(before.uncaught + 1);
+        expect(after.rejection).toBe(before.rejection + 1);
+    });
+
+    it('removes every listener it added', () => {
+        const before = counts();
+        installProcessGuards({ keepAliveOnUncaught: true })();
+        expect(counts()).toEqual(before);
+    });
+
+    it('logs the fault and reports it to onFault', () => {
+        const consoleError = vi
+            .spyOn(console, 'error')
+            .mockImplementation(() => {});
+        const onFault = vi.fn();
+        uninstall = installProcessGuards({ onFault });
+
+        const error = new Error('boom');
+        process.emit('uncaughtExceptionMonitor', error, 'uncaughtException');
+
+        expect(onFault).toHaveBeenCalledWith(
+            'uncaughtException',
+            error,
+            'uncaughtException',
+        );
+        expect(consoleError).toHaveBeenCalledWith('uncaughtException', error);
+    });
+
+    it('survives an onFault that throws', () => {
+        const consoleError = vi
+            .spyOn(console, 'error')
+            .mockImplementation(() => {});
+        uninstall = installProcessGuards({
+            onFault: () => {
+                throw new Error('alarm unavailable');
+            },
+        });
+
+        expect(() =>
+            process.emit(
+                'uncaughtExceptionMonitor',
+                new Error('boom'),
+                'uncaughtException',
+            ),
+        ).not.toThrow();
+        expect(consoleError).toHaveBeenCalledWith(
+            'process fault handler threw',
+            expect.any(Error),
+        );
+    });
+
+    it('labels a rejection surfaced through the monitor', () => {
+        const consoleError = vi
+            .spyOn(console, 'error')
+            .mockImplementation(() => {});
+        uninstall = installProcessGuards();
+
+        const error = new Error('rejected');
+        process.emit('uncaughtExceptionMonitor', error, 'unhandledRejection');
+
+        expect(consoleError).toHaveBeenCalledWith(
+            'uncaughtException (unhandledRejection)',
+            error,
+        );
+    });
+});

src/backend/util/processGuards.ts

@@ -0,0 +1,109 @@
+/*
+ * Copyright (C) 2024-present Puter Technologies Inc.
+ *
+ * This file is part of Puter.
+ *
+ * Puter is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published
+ * by the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+/**
+ * Process-level error visibility.
+ *
+ * A request handler can kill the whole process without any of its own code
+ * throwing: an `'error'` event on a stream nobody is listening to becomes an
+ * uncaught exception, and the default response is to exit. Every other
+ * in-flight request dies with it.
+ *
+ * The guards here make that visible, and optionally survivable.
+ */
+
+export type ProcessFaultKind = 'uncaughtException' | 'unhandledRejection';
+
+export interface ProcessGuardOptions {
+    /**
+     * Stay alive after an uncaught exception instead of exiting.
+     *
+     * Off by default, because a process that resumes after an uncaught
+     * exception may be holding half-applied state, and Node treats resuming as
+     * undefined behavior. Turn it on where an exit is the more expensive
+     * failure — a small pool of nodes behind a health check, where one bad
+     * request would otherwise take out the whole pool's worth of live
+     * requests.
+     */
+    keepAliveOnUncaught?: boolean;
+    /**
+     * Called for every fault, before the exit decision. Use it to raise an
+     * alarm; it must not throw.
+     */
+    onFault?: (kind: ProcessFaultKind, error: unknown, origin?: string) => void;
+}
+
+/**
+ * Install process-level fault logging. Returns a function that removes every
+ * listener it added, so a server can install per boot without leaking listeners
+ * across restarts (or across test files).
+ */
+export const installProcessGuards = (
+    options: ProcessGuardOptions = {},
+): (() => void) => {
+    const { keepAliveOnUncaught = false, onFault } = options;
+
+    const report = (
+        kind: ProcessFaultKind,
+        error: unknown,
+        origin?: string,
+    ) => {
+        // One console call per fault: with structured logging installed, that
+        // keeps the whole stack in a single log event.
+        console.error(
+            `${kind}${origin && origin !== kind ? ` (${origin})` : ''}`,
+            error,
+        );
+        if (!onFault) return;
+        try {
+            onFault(kind, error, origin);
+        } catch (faultHandlerError) {
+            console.error('process fault handler threw', faultHandlerError);
+        }
+    };
+
+    // `uncaughtExceptionMonitor` sees every uncaught exception — including the
+    // ones that are about to be fatal — without suppressing Node's default
+    // handling. It is the only way to log a crash and still crash.
+    const onMonitor = (error: Error, origin: string) =>
+        report('uncaughtException', error, origin);
+    process.on('uncaughtExceptionMonitor', onMonitor);
+
+    // Registering an 'uncaughtException' listener is what actually stops the
+    // exit; the monitor above has already logged, so this body stays empty.
+    const onUncaught = () => {};
+    // Node's default for an unhandled rejection is to raise it as an uncaught
+    // exception, which the monitor already reports. Only take it over when the
+    // process is meant to survive.
+    const onRejection = (reason: unknown) =>
+        report('unhandledRejection', reason);
+
+    if (keepAliveOnUncaught) {
+        process.on('uncaughtException', onUncaught);
+        process.on('unhandledRejection', onRejection);
+    }
+
+    return () => {
+        process.off('uncaughtExceptionMonitor', onMonitor);
+        if (keepAliveOnUncaught) {
+            process.off('uncaughtException', onUncaught);
+            process.off('unhandledRejection', onRejection);
+        }
+    };
+};