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?.(); }, });