From 8ed8feed4b93eec42a211042cb0b3d43d089a20a Mon Sep 17 00:00:00 2001 From: Daniel Salazar Date: Thu, 13 Aug 2026 01:02:17 -0700 Subject: [PATCH] fix: don't let an aborted upload take down the process (#3557) --- src/backend/index.ts | 2 + src/backend/server.ts | 11 ++ src/backend/services/fs/FSService.test.ts | 37 +++++++ src/backend/services/fs/FSService.ts | 16 ++- src/backend/types.ts | 18 +++- src/backend/util/processGuards.test.ts | 122 ++++++++++++++++++++++ src/backend/util/processGuards.ts | 109 +++++++++++++++++++ 7 files changed, 308 insertions(+), 7 deletions(-) create mode 100644 src/backend/util/processGuards.test.ts create mode 100644 src/backend/util/processGuards.ts diff --git a/src/backend/index.ts b/src/backend/index.ts index 9d3ef3cee..f5bc1414f 100644 --- a/src/backend/index.ts +++ b/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. } diff --git a/src/backend/server.ts b/src/backend/server.ts index c0daea161..251f16eb1 100644 --- a/src/backend/server.ts +++ b/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; #server: ReturnType['listen']> | null = null; + #removeProcessGuards: (() => void) | null = null; #ready: Promise; @@ -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 diff --git a/src/backend/services/fs/FSService.test.ts b/src/backend/services/fs/FSService.test.ts index c4e8d27d6..eef0a94f1 100644 --- a/src/backend/services/fs/FSService.test.ts +++ b/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', () => { diff --git a/src/backend/services/fs/FSService.ts b/src/backend/services/fs/FSService.ts index 8d3619bc8..8ef917570 100644 --- a/src/backend/services/fs/FSService.ts +++ b/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, diff --git a/src/backend/types.ts b/src/backend/types.ts index 518f1209c..0cb84e359 100644 --- a/src/backend/types.ts +++ b/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[]> // eslint-disable-next-line @typescript-eslint/no-explicit-any | Record[]; diff --git a/src/backend/util/processGuards.test.ts b/src/backend/util/processGuards.test.ts new file mode 100644 index 000000000..51210b2e9 --- /dev/null +++ b/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 . + */ + +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, + ); + }); +}); diff --git a/src/backend/util/processGuards.ts b/src/backend/util/processGuards.ts new file mode 100644 index 000000000..0eb9a580c --- /dev/null +++ b/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 . + */ + +/** + * 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); + } + }; +};