fix: drain server requests and update deployment smoke

apps/server/src/index.test.ts

@@ -1,6 +1,9 @@
+import { once } from "node:events";
 import { afterEach, describe, expect, it, vi } from "vitest";
+import { serve } from "@hono/node-server";
 
 const events = vi.hoisted(() => [] as string[]);
+const appFetch = vi.hoisted(() => vi.fn());
 vi.mock("./startup/checks", () => ({
 	runStartupChecks: async () => {
 		await Promise.resolve();
@@ -12,7 +15,7 @@ vi.mock("./http/app", () => {
 	return {
 		createApp: () => {
 			events.push("app created");
-			return { fetch: vi.fn() };
+			return { fetch: appFetch };
 		},
 	};
 });
@@ -23,19 +26,63 @@ vi.mock("@reactive-resume/auth/config", () => ({
 	},
 }));
 vi.mock("@hono/node-server", () => ({
-	serve: () => {
+	serve: vi.fn(() => {
 		events.push("server listening");
-	},
+	}),
 }));
-vi.mock("@reactive-resume/env/server", () => ({ env: { SERVER_PORT: 3001 } }));
+vi.mock("@reactive-resume/env/server", () => ({ env: { SERVER_PORT: 0 } }));
 afterEach(() => vi.restoreAllMocks());
 
 describe("server startup", () => {
 	it("finishes migrations before importing auth and seeding OAuth resources", async () => {
 		vi.spyOn(process, "on").mockReturnValue(process);
+		vi.spyOn(process, "once").mockReturnValue(process);
 		const entry = await import("./index");
 		expect(events).toEqual([]);
 		await entry.main();
 		expect(events).toEqual(["migrations complete", "auth imported", "auth ready", "app created", "server listening"]);
 	});
+
+	it.each(["SIGTERM", "SIGINT"])("drains active requests before exiting on %s", async (signal) => {
+		vi.spyOn(process, "on").mockReturnValue(process);
+		const signals = vi.spyOn(process, "once").mockReturnValue(process);
+		const exit = vi.spyOn(process, "exit").mockImplementation(() => undefined as never);
+		const started = Promise.withResolvers<void>();
+		const finished = Promise.withResolvers<Response>();
+		appFetch.mockImplementation(() => {
+			started.resolve();
+			return finished.promise;
+		});
+		const { serve: realServe } = await vi.importActual<typeof import("@hono/node-server")>("@hono/node-server");
+		let server: ReturnType<typeof serve> | undefined;
+		vi.mocked(serve).mockImplementationOnce((options, callback) => {
+			server = realServe(options, callback);
+			return server;
+		});
+		await (await import("./index")).main();
+		if (!server) throw new Error("Server did not start");
+		const runningServer = server;
+		try {
+			if (!server.listening) await once(server, "listening");
+			const address = server.address();
+			if (!address || typeof address === "string") throw new Error("Missing HTTP address");
+			const url = `http://127.0.0.1:${address.port}`;
+			const response = fetch(url);
+			await started.promise;
+			const shutdown = signals.mock.calls.find(([name]) => name === signal)?.[1];
+			expect(shutdown).toBeTypeOf("function");
+			const closed = once(server, "close");
+			shutdown?.();
+			shutdown?.();
+			expect(exit).not.toHaveBeenCalled();
+			await expect(fetch(url)).rejects.toThrow();
+			finished.resolve(new Response("drained"));
+			expect(await (await response).text()).toBe("drained");
+			await closed;
+			expect(exit).toHaveBeenCalledExactlyOnceWith(0);
+		} finally {
+			finished.resolve(new Response("drained"));
+			await new Promise<void>((resolve) => runningServer.close(() => resolve()));
+		}
+	});
 });

apps/server/src/index.ts

@@ -24,7 +24,7 @@ export async function main() {
 
 	const app = createApp();
 
-	serve(
+	const server = serve(
 		{
 			fetch: app.fetch,
 			port,
@@ -33,6 +33,22 @@ export async function main() {
 			console.info(`🚀 Up and running on http://localhost:${info.port}`);
 		},
 	);
+
+	let shuttingDown = false;
+	const shutdown = () => {
+		if (shuttingDown) return;
+		shuttingDown = true;
+		// Stop accepting connections, then wait for active requests before exiting.
+		server.close((error) => {
+			if (error) {
+				console.error("Failed to drain HTTP requests", error);
+				process.exit(1);
+			}
+			process.exit(0);
+		});
+	};
+	process.once("SIGTERM", shutdown);
+	process.once("SIGINT", shutdown);
 }
 
 if (process.argv[1] && import.meta.url === pathToFileURL(process.argv[1]).href) {

tooling/deployment/smoke.mjs

@@ -114,7 +114,7 @@ try {
 			apiKey: process.env.SMOKE_AI_API_KEY,
 		});
 		await rpc("aiProviders/test", { id: provider.id });
-		const thread = await rpc("agent/threads/create", { aiProviderId: provider.id });
+		const thread = await rpc("agent/threads/start", { resumeId: id, aiProviderId: provider.id });
 		const attachment = await rpc(
 			"agent/attachments/create",
 			{