From 4f96e7dfbf36bf6f3ce425f0b779ededd72998a6 Mon Sep 17 00:00:00 2001 From: ARUNAVO RAY Date: Thu, 15 Oct 2026 18:46:56 +0530 Subject: [PATCH] fix: survive SSE client disconnects on Bun 1.4 (#465) (#466) On Bun 1.4 the node adapter's reader cancel closes the stream before the request abort fires, so the abort listener's controller.close() threw ERR_INVALID_STATE as an uncaught exception and the process shut itself down whenever a dashboard tab disconnected. Both SSE routes now guard the close and clear their timers on cancel too. A shutdown forced by an uncaught exception or unhandled rejection now exits with status 1 instead of 0, so a unit with Restart=on-failure (the Proxmox community script) starts the service again. --- src/lib/shutdown-manager.test.ts | 15 ++++++ src/lib/shutdown-manager.ts | 11 +++- src/pages/api/events/index.test.ts | 85 ++++++++++++++++++++++++++++++ src/pages/api/events/index.ts | 25 +++++++-- src/pages/api/sse/index.ts | 19 +++++-- 5 files changed, 146 insertions(+), 9 deletions(-) create mode 100644 src/lib/shutdown-manager.test.ts create mode 100644 src/pages/api/events/index.test.ts diff --git a/src/lib/shutdown-manager.test.ts b/src/lib/shutdown-manager.test.ts new file mode 100644 index 0000000..b108d12 --- /dev/null +++ b/src/lib/shutdown-manager.test.ts @@ -0,0 +1,15 @@ +import { describe, expect, test } from "bun:test"; +import { shutdownExitCode } from "./shutdown-manager"; + +describe("shutdownExitCode (#465)", () => { + test("a crash-driven shutdown exits non-zero so Restart=on-failure restarts the service", () => { + expect(shutdownExitCode("UNCAUGHT_EXCEPTION")).toBe(1); + expect(shutdownExitCode("UNHANDLED_REJECTION")).toBe(1); + }); + + test("a signal-driven stop is a clean exit", () => { + expect(shutdownExitCode("SIGTERM")).toBe(0); + expect(shutdownExitCode("SIGINT")).toBe(0); + expect(shutdownExitCode("SIGHUP")).toBe(0); + }); +}); diff --git a/src/lib/shutdown-manager.ts b/src/lib/shutdown-manager.ts index f96d196..46313e8 100644 --- a/src/lib/shutdown-manager.ts +++ b/src/lib/shutdown-manager.ts @@ -147,6 +147,11 @@ async function executeShutdownCallbacks(): Promise { /** * Perform graceful shutdown of the application */ +/** Exit status for a completed shutdown: 1 after a crash, 0 after a signal. */ +export function shutdownExitCode(signal: string): number { + return signal === 'UNCAUGHT_EXCEPTION' || signal === 'UNHANDLED_REJECTION' ? 1 : 0; +} + export async function gracefulShutdown(signal: string = 'UNKNOWN'): Promise { if (shutdownInProgress) { console.log('⚠️ Shutdown already in progress, ignoring additional signal'); @@ -187,8 +192,10 @@ export async function gracefulShutdown(signal: string = 'UNKNOWN'): Promise { + const res = Bun.spawnSync({ + cmd: [process.execPath, "test", import.meta.path], + env: { ...process.env, [CHILD_FLAG]: "1" }, + stdout: "pipe", + stderr: "pipe", + }); + if (res.exitCode !== 0) { + console.error(res.stdout.toString()); + console.error(res.stderr.toString()); + } + expect(res.exitCode).toBe(0); + }, 60_000); +} + +if (isChild) { + mock.module("@/lib/auth-guards", () => ({ + requireAuthenticatedUserId: async () => ({ userId: "user-1" }), + })); + mock.module("@/lib/events", () => ({ + getNewEvents: async () => [], + })); +} + +const routes = isChild + ? { + events: (await import("./index")).GET, + sse: (await import("../sse/index")).GET, + } + : ({} as Record); + +/** + * Opens the stream, then closes it the way the node adapter does on a + * disconnect: the reader is cancelled first, the abort signal fires second. + */ +async function disconnect(route: any) { + const abort = new AbortController(); + const request = new Request("http://localhost/api/x", { signal: abort.signal }); + const response = await route({ request, locals: {} }); + const reader = response.body.getReader(); + const first = await reader.read(); + expect(new TextDecoder().decode(first.value)).toContain(": connected"); + + const uncaught: unknown[] = []; + const record = (error: unknown) => uncaught.push(error); + process.on("uncaughtException", record); + try { + await reader.cancel(); + abort.abort(); + await new Promise((resolve) => setTimeout(resolve, 20)); + } finally { + process.off("uncaughtException", record); + } + return uncaught; +} + +describe.skipIf(!isChild)("SSE routes survive a client disconnect (#465)", () => { + test("/api/events", async () => { + expect(await disconnect(routes.events)).toEqual([]); + }); + + test("/api/sse", async () => { + expect(await disconnect(routes.sse)).toEqual([]); + }); +}); diff --git a/src/pages/api/events/index.ts b/src/pages/api/events/index.ts index 781cd5a..c08fc0c 100644 --- a/src/pages/api/events/index.ts +++ b/src/pages/api/events/index.ts @@ -8,6 +8,7 @@ export const GET: APIRoute = async ({ request, locals }) => { const userId = authResult.userId; // Create a new ReadableStream for SSE + let cancelStream: (() => void) | undefined; const stream = new ReadableStream({ start(controller) { const encoder = new TextEncoder(); @@ -47,12 +48,30 @@ export const GET: APIRoute = async ({ request, locals }) => { } }, 30000); - // Cleanup on close - request.signal.addEventListener("abort", () => { + // Cleanup on close. When the client disconnects, the node adapter + // cancels the stream reader and then fires this abort; on Bun 1.4 the + // cancel has already closed the stream, so close() throws + // ERR_INVALID_STATE inside an event listener, an uncaught exception + // that took the whole process down (#465). + let closed = false; + const stop = () => { + if (closed) return; + closed = true; clearInterval(pollInterval); clearInterval(heartbeatInterval); - controller.close(); + }; + request.signal.addEventListener("abort", () => { + stop(); + try { + controller.close(); + } catch { + // Already closed by the reader cancel. + } }); + cancelStream = stop; + }, + cancel() { + cancelStream?.(); }, }); diff --git a/src/pages/api/sse/index.ts b/src/pages/api/sse/index.ts index 11ccbed..4bcef94 100644 --- a/src/pages/api/sse/index.ts +++ b/src/pages/api/sse/index.ts @@ -11,6 +11,7 @@ export const GET: APIRoute = async ({ request, locals }) => { let isClosed = false; const POLL_INTERVAL = 5000; // Poll every 5 seconds (reduced from 2 seconds for low-traffic usage) + let cancelStream: (() => void) | undefined; const stream = new ReadableStream({ start(controller) { const encoder = new TextEncoder(); @@ -73,20 +74,30 @@ export const GET: APIRoute = async ({ request, locals }) => { } }, 30000); - // Handle client disconnection + // Handle client disconnection. The reader cancel runs first on Bun 1.4 + // and closes the stream, so close() can throw ERR_INVALID_STATE here; + // inside an event listener that is an uncaught exception (#465). request.signal?.addEventListener("abort", () => { if (!isClosed) { isClosed = true; if (pollIntervalId) { clearInterval(pollIntervalId); } - controller.close(); + try { + controller.close(); + } catch { + // Already closed by the reader cancel. + } } }); + cancelStream = () => { + isClosed = true; + if (pollIntervalId) clearInterval(pollIntervalId); + clearInterval(heartbeatInterval); + }; }, cancel() { - // Extra safety in case cancel is triggered - isClosed = true; + cancelStream?.(); }, });