ce298d8 · 7c7efff
@@ -2246,7 +2246,9 @@
export interface Servers {
   * Stop taking connections and let those in flight finish: what a process
   * handing its port to a release beside it does at drain start (#2040).
   * Resolves once no request is in flight, or `budgetMs` has passed; `stop`
   * then cuts whatever is left.
   * then cuts whatever is left. From then on every answer still given on a
   * kept-alive connection says `connection: close`, so a router holding one
   * redials and reaches the release instead of this process (#11682).
   */
  stopAccepting(budgetMs: number): Promise<void>;
  stop(): void;
@@ -2768,6 +2770,8 @@
export async function serve(app: App): Promise<Servers> {
      return withHeader(response, "connection", "close");
  };
  // Set by `stopAccepting`: the port is the release's now.
  let handedOver = false;
  const web = Bun.serve({
    port: app.config.port ?? 3000,
    // A release swaps in beside this process on the same port (#1663).
@@ -2781,7 +2785,12 @@
export async function serve(app: App): Promise<Servers> {
    // proxy; its own live surfaces are SSE (`sse.ts`) and stay so.
    websocket: withShells(appWebSocketWith(app.config.wsMaxMessage)),
    // Open until it is answered, so a stall while it runs names it (#11478).
    fetch: (request, server) => inFlight.during(request, performance.now(), () => handle(request, server)),
    // Once the port is handed over, an answer on a connection opened before ends it (#11682).
    fetch: (request, server) =>
      inFlight.during(request, performance.now(), async () => {
        const response = await handle(request, server);
        return handedOver && response !== undefined ? withHeader(response, "connection", "close") : response;
      }),
    // A last resort: `handle` answers a throw itself, so it is counted.
    error: (error) => failure(app, error),
  });
@@ -2796,7 +2805,10 @@
export async function serve(app: App): Promise<Servers> {
    },
    async stopAccepting(budgetMs) {
      // Closing the listener leaves the port to the process beside this one;
      // the connections already here are answered to the end.
      // the connections already here are answered to the end, each closed
      // after its next answer: a router's pooled connection would otherwise
      // keep hearing this version for the whole drain (#11682).
      handedOver = true;
      web.stop(false).catch(() => {});
      const deadline = Date.now() + budgetMs;
      while (web.pendingRequests > 0 && Date.now() < deadline)
@@ -0,0 +1,66 @@
/**
 * A handed-over port takes its kept-alive connections with it (#11682).
 * `stopAccepting` closes the old forge's listener, but a connection opened
 * before (a router's pool) was still answered by the old process for the
 * whole drain, so the public URL kept answering as the old version and the
 * release proof gave up. Now the old forge's next answer on such a
 * connection says `connection: close`, and the client's redial reaches the
 * release.
 */
import { afterAll, expect, test } from "bun:test";
import { mkdtemp, rm } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { createApp } from "../src/app.ts";
import { createUser } from "../src/core/users.ts";
import { serve } from "../src/http/server.ts";
const dirs: string[] = [];
afterAll(async () => {
  for (const d of dirs) await rm(d, { recursive: true, force: true });
});
test("after stopAccepting, a kept-alive connection is closed after one answer and the next request reaches the release", async () => {
  const oldDir = await mkdtemp(join(tmpdir(), "forge-keepalive-old-"));
  const newDir = await mkdtemp(join(tmpdir(), "forge-keepalive-new-"));
  dirs.push(oldDir, newDir);
  // Only the old forge knows alice, so an answer says which forge gave it.
  const oldApp = createApp({ dataDir: oldDir, port: 0, selfSwap: true });
  const oldServers = await serve(oldApp);
  const made = await createUser(oldApp.db, {
    origin: "seed",
    trusted: true,
    username: "alice",
    email: "[email protected]",
    password: "a-good-passphrase",
    adminIfFirst: true,
  });
  expect(made.ok).toBe(true);
  const url = `http://127.0.0.1:${oldServers.port}/api/v1/users/alice`;
  const ask = async () => {
    const res = await fetch(url, { signal: AbortSignal.timeout(3_000) });
    await res.arrayBuffer();
    return { status: res.status, connection: res.headers.get("connection") };
  };
  // Warm a kept-alive connection while the old forge is the only listener.
  for (let i = 0; i < 3; i++) expect((await ask()).status).toBe(200);
  const newApp = createApp({ dataDir: newDir, port: oldServers.port, selfSwap: true });
  const newServers = await serve(newApp);
  const drained = oldServers.stopAccepting(5_000);
  // The old forge answers what was already its connection once more, and ends it.
  const last = await ask();
  expect(last.status).toBe(200);
  expect(last.connection).toBe("close");
  // Every request after that is the release's.
  for (let i = 0; i < 5; i++) expect((await ask()).status).toBe(404);
  await drained;
  oldServers.stop();
  newServers.stop();
  oldApp.close();
  newApp.close();
}, 20_000);
		@@ -2246,7 +2246,9 @@ export interface Servers {
2246	2246	* Stop taking connections and let those in flight finish: what a process
2247	2247	* handing its port to a release beside it does at drain start (#2040).
2248	2248	* Resolves once no request is in flight, or `budgetMs` has passed; `stop`
2249		* then cuts whatever is left.
	2249	* then cuts whatever is left. From then on every answer still given on a
	2250	* kept-alive connection says `connection: close`, so a router holding one
	2251	* redials and reaches the release instead of this process (#11682).
2250	2252	*/
2251	2253	stopAccepting(budgetMs: number): Promise<void>;
2252	2254	stop(): void;
		@@ -2768,6 +2770,8 @@ export async function serve(app: App): Promise<Servers> {
2768	2770	return withHeader(response, "connection", "close");
2769	2771	};
2770	2772
	2773	// Set by `stopAccepting`: the port is the release's now.
	2774	let handedOver = false;
2771	2775	const web = Bun.serve({
2772	2776	port: app.config.port ?? 3000,
2773	2777	// A release swaps in beside this process on the same port (#1663).
		@@ -2781,7 +2785,12 @@ export async function serve(app: App): Promise<Servers> {
2781	2785	// proxy; its own live surfaces are SSE (`sse.ts`) and stay so.
2782	2786	websocket: withShells(appWebSocketWith(app.config.wsMaxMessage)),
2783	2787	// Open until it is answered, so a stall while it runs names it (#11478).
2784		fetch: (request, server) => inFlight.during(request, performance.now(), () => handle(request, server)),
	2788	// Once the port is handed over, an answer on a connection opened before ends it (#11682).
	2789	fetch: (request, server) =>
	2790	inFlight.during(request, performance.now(), async () => {
	2791	const response = await handle(request, server);
	2792	return handedOver && response !== undefined ? withHeader(response, "connection", "close") : response;
	2793	}),
2785	2794	// A last resort: `handle` answers a throw itself, so it is counted.
2786	2795	error: (error) => failure(app, error),
2787	2796	});
		@@ -2796,7 +2805,10 @@ export async function serve(app: App): Promise<Servers> {
2796	2805	},
2797	2806	async stopAccepting(budgetMs) {
2798	2807	// Closing the listener leaves the port to the process beside this one;
2799		// the connections already here are answered to the end.
	2808	// the connections already here are answered to the end, each closed
	2809	// after its next answer: a router's pooled connection would otherwise
	2810	// keep hearing this version for the whole drain (#11682).
	2811	handedOver = true;
2800	2812	web.stop(false).catch(() => {});
2801	2813	const deadline = Date.now() + budgetMs;
2802	2814	while (web.pendingRequests > 0 && Date.now() < deadline)
		@@ -0,0 +1,66 @@
	1	/**
	2	* A handed-over port takes its kept-alive connections with it (#11682).
	3	* `stopAccepting` closes the old forge's listener, but a connection opened
	4	* before (a router's pool) was still answered by the old process for the
	5	* whole drain, so the public URL kept answering as the old version and the
	6	* release proof gave up. Now the old forge's next answer on such a
	7	* connection says `connection: close`, and the client's redial reaches the
	8	* release.
	9	*/
	10
	11	import { afterAll, expect, test } from "bun:test";
	12	import { mkdtemp, rm } from "node:fs/promises";
	13	import { tmpdir } from "node:os";
	14	import { join } from "node:path";
	15	import { createApp } from "../src/app.ts";
	16	import { createUser } from "../src/core/users.ts";
	17	import { serve } from "../src/http/server.ts";
	18
	19	const dirs: string[] = [];
	20	afterAll(async () => {
	21	for (const d of dirs) await rm(d, { recursive: true, force: true });
	22	});
	23
	24	test("after stopAccepting, a kept-alive connection is closed after one answer and the next request reaches the release", async () => {
	25	const oldDir = await mkdtemp(join(tmpdir(), "forge-keepalive-old-"));
	26	const newDir = await mkdtemp(join(tmpdir(), "forge-keepalive-new-"));
	27	dirs.push(oldDir, newDir);
	28	// Only the old forge knows alice, so an answer says which forge gave it.
	29	const oldApp = createApp({ dataDir: oldDir, port: 0, selfSwap: true });
	30	const oldServers = await serve(oldApp);
	31	const made = await createUser(oldApp.db, {
	32	origin: "seed",
	33	trusted: true,
	34	username: "alice",
	35	email: "[email protected]",
	36	password: "a-good-passphrase",
	37	adminIfFirst: true,
	38	});
	39	expect(made.ok).toBe(true);
	40	const url = `http://127.0.0.1:${oldServers.port}/api/v1/users/alice`;
	41	const ask = async () => {
	42	const res = await fetch(url, { signal: AbortSignal.timeout(3_000) });
	43	await res.arrayBuffer();
	44	return { status: res.status, connection: res.headers.get("connection") };
	45	};
	46
	47	// Warm a kept-alive connection while the old forge is the only listener.
	48	for (let i = 0; i < 3; i++) expect((await ask()).status).toBe(200);
	49
	50	const newApp = createApp({ dataDir: newDir, port: oldServers.port, selfSwap: true });
	51	const newServers = await serve(newApp);
	52	const drained = oldServers.stopAccepting(5_000);
	53
	54	// The old forge answers what was already its connection once more, and ends it.
	55	const last = await ask();
	56	expect(last.status).toBe(200);
	57	expect(last.connection).toBe("close");
	58	// Every request after that is the release's.
	59	for (let i = 0; i < 5; i++) expect((await ask()).status).toBe(404);
	60
	61	await drained;
	62	oldServers.stop();
	63	newServers.stop();
	64	oldApp.close();
	65	newApp.close();
	66	}, 20_000);