0a7e89b · d97a02a
@@ -1800,6 +1800,8 @@
async function handleHook(app: App, request: Request): Promise<Response> {
export interface Servers {
  readonly url: string;
  /** Requests the listener still holds: 0 once every one has settled. */
  readonly inFlight: number;
  readonly port: number;
  /**
   * Stop taking connections and let those in flight finish: what a process
@@ -2003,15 +2005,31 @@
export async function serve(app: App): Promise<Servers> {
    }
  };
  // Behind the floor, and a throw is answered here so it is counted too.
  // Behind the floor, and a throw is answered here so it is counted too. A
  // client that goes away settles the request as 499 (nginx's code for it)
  // instead of leaving a handler parked on a Response that never comes; a
  // body already on its way is cancelled. Only the client's own abort ends a
  // request: a git push is never cut by the forge.
  const answered = async (
    owner: App,
    request: Request,
    run: () => Promise<Response>,
  ): Promise<Response> => {
    const gone = Promise.withResolvers<"gone">();
    const onAbort = () => gone.resolve("gone");
    if (request.signal.aborted) onAbort();
    else request.signal.addEventListener("abort", onAbort, { once: true });
    const work = run().then(
      (response) => securityFloor(response),
      (error) => failure(owner, error),
    );
    try {
      return securityFloor(await run());
    } catch (error) {
      return failure(owner, error);
      const first = await Promise.race([work, gone.promise]);
      if (first !== "gone") return first;
      work.then((late) => late.body?.cancel().catch(() => {}));
      
        sentinelcrew
        29d ago
        
      
      No test covers this line: removing it leaves the suite green. Add a test where the handler resolves after the abort with a stream whose cancel sets a flag, and assert the flag is set. This is the issue's "streaming bodies are cancelled" bullet.
    
      
        sentinelcrew
        29d ago
        
      
      Minor, not blocking: after this 499 the handler is still running, and the caller goes straight on to settleBody(request). If a handler has called getReader() on a chunked body but hasn't read from it yet, bodyUsed is still false while the stream is locked, so drain's getReader() would throw out of the fetch handler. The client is already gone, so nothing visible breaks, but a guard (request.body.locked) in settleBody would make the edge case explicit.
    
      return securityFloor(new Response(null, { status: 499 }));
    } finally {
      request.signal.removeEventListener("abort", onAbort);
    }
  };
@@ -2061,7 +2079,7 @@
export async function serve(app: App): Promise<Servers> {
          }
          notePeer(request, server.requestIP(request)?.address, trust);
          armClose(request, () => server.timeout(request, 1));
          const response = await answered(child, () => http.handle(request));
          const response = await answered(child, request, () => http.handle(request));
          counted(child, request, response);
          logged(key, request, response, started);
          await settleBody(request);
@@ -2078,7 +2096,7 @@
export async function serve(app: App): Promise<Servers> {
      armClose(request, () => server.timeout(request, 1));
      // Everything the forge answers for itself leaves through this one
      // door, so a responder added next year cannot forget the floor.
      const response = await answered(app, () =>
      const response = await answered(app, request, () =>
        answer(app, router, apiRouter, limiter, request),
      );
      counted(app, request, response);
@@ -2135,6 +2153,9 @@
export async function serve(app: App): Promise<Servers> {
  return {
    url: `http://localhost:${port}`,
    port,
    get inFlight() {
      return web.pendingRequests;
    },
    async stopAccepting(budgetMs) {
      // Closing the listener leaves the port to the process beside this one;
      // the connections already here are answered to the end.
@@ -0,0 +1,106 @@
/** A client that goes away settles its request as 499 and frees the handler (#2250). */
import { afterAll, beforeAll, expect, spyOn, test } from "bun:test";
import { readdirSync, readFileSync } from "node:fs";
import { mkdtemp, rm } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { createApp, type App } from "../src/app.ts";
import { accessDir, flushAccess } from "../src/http/access_log.ts";
import * as health from "../src/http/health.ts";
import { serve } from "../src/http/server.ts";
import { readUsage } from "../src/core/usage_count.ts";
let dataDir: string;
let app: App;
beforeAll(async () => {
  dataDir = await mkdtemp(join(tmpdir(), "forge-abort-"));
  app = createApp({ dataDir, port: 0 });
});
afterAll(async () => {
  app.close();
  await rm(dataDir, { recursive: true, force: true });
});
test("an aborted slow request is logged and counted as 499, and nothing stays in flight", async () => {
  const servers = await serve(app);
  // A handler whose Response never comes: what an aborted request left behind.
  const stuck = spyOn(health, "readyResponse").mockImplementation(
    (() => new Promise(() => {})) as never,
  );
  try {
    const controller = new AbortController();
    const pending = fetch(`${servers.url}/readyz`, { signal: controller.signal }).catch(
      () => "aborted",
    );
    for (let i = 0; i < 100 && servers.inFlight === 0; i++) await Bun.sleep(10);
    expect(servers.inFlight).toBe(1);
    controller.abort();
    expect(await pending).toBe("aborted");
    for (let i = 0; i < 100 && servers.inFlight > 0; i++) await Bun.sleep(10);
      
        sentinelcrew
        29d ago
        outdated
      
      If an expectation above fails, the spy is never restored and the server never stops. Put stuck.mockRestore() and servers.stop() in a try/finally.
    
    expect(servers.inFlight).toBe(0);
    flushAccess();
    const kept = readdirSync(accessDir(dataDir))
      .flatMap((f) => readFileSync(join(accessDir(dataDir), f), "utf8").split("\n"))
      .filter((l) => l !== "")
      .map((l) => JSON.parse(l) as { path: string; status: number });
    expect(kept.find((l) => l.path === "/readyz")?.status).toBe(499);
  } finally {
    stuck.mockRestore();
    servers.stop();
  }
  // Stopping the server settles the usage counts.
  const rows = readUsage(app.db, "").filter((r) => r.pattern === "/readyz");
  expect(rows.map((r) => [r.status, r.n])).toEqual([["4xx", 1]]);
});
test("a body that arrives after the client left is cancelled", async () => {
  const servers = await serve(app);
      
        sentinelcrew
        29d ago
        outdated
      
      res's body is never read, so pendingRequests can still count this request here. Await res.text() first so the assertion does not depend on timing.
    
  let cancelled = false;
  let release!: (r: Response) => void;
  const stuck = spyOn(health, "readyResponse").mockImplementation(
    (() => new Promise<Response>((resolve) => (release = resolve))) as never,
  );
  try {
    const controller = new AbortController();
    const pending = fetch(`${servers.url}/readyz`, { signal: controller.signal }).catch(
      () => "aborted",
    );
    for (let i = 0; i < 100 && servers.inFlight === 0; i++) await Bun.sleep(10);
    expect(servers.inFlight).toBe(1);
    controller.abort();
    expect(await pending).toBe("aborted");
    for (let i = 0; i < 100 && servers.inFlight > 0; i++) await Bun.sleep(10);
    release(
      new Response(
        new ReadableStream({
          pull(c) {
            c.enqueue(new TextEncoder().encode("late"));
          },
          cancel() {
            cancelled = true;
          },
        }),
      ),
    );
    for (let i = 0; i < 100 && !cancelled; i++) await Bun.sleep(10);
    expect(cancelled).toBe(true);
    expect(servers.inFlight).toBe(0);
  } finally {
    stuck.mockRestore();
    servers.stop();
  }
});
test("a request that is not aborted is untouched, floor and all", async () => {
  const servers = await serve(app);
  const res = await fetch(`${servers.url}/readyz`);
  expect(res.status).toBe(200);
  expect(res.headers.get("x-content-type-options")).toBe("nosniff");
  await res.text();
  expect(servers.inFlight).toBe(0);
  servers.stop();
});
		@@ -1800,6 +1800,8 @@ async function handleHook(app: App, request: Request): Promise<Response> {
1800	1800
1801	1801	export interface Servers {
1802	1802	readonly url: string;
	1803	/** Requests the listener still holds: 0 once every one has settled. */
	1804	readonly inFlight: number;
1803	1805	readonly port: number;
1804	1806	/**
1805	1807	* Stop taking connections and let those in flight finish: what a process
		@@ -2003,15 +2005,31 @@ export async function serve(app: App): Promise<Servers> {
2003	2005	}
2004	2006	};
2005	2007
2006		// Behind the floor, and a throw is answered here so it is counted too.
	2008	// Behind the floor, and a throw is answered here so it is counted too. A
	2009	// client that goes away settles the request as 499 (nginx's code for it)
	2010	// instead of leaving a handler parked on a Response that never comes; a
	2011	// body already on its way is cancelled. Only the client's own abort ends a
	2012	// request: a git push is never cut by the forge.
2007	2013	const answered = async (
2008	2014	owner: App,
	2015	request: Request,
2009	2016	run: () => Promise<Response>,
2010	2017	): Promise<Response> => {
	2018	const gone = Promise.withResolvers<"gone">();
	2019	const onAbort = () => gone.resolve("gone");
	2020	if (request.signal.aborted) onAbort();
	2021	else request.signal.addEventListener("abort", onAbort, { once: true });
	2022	const work = run().then(
	2023	(response) => securityFloor(response),
	2024	(error) => failure(owner, error),
	2025	);
2011	2026	try {
2012		return securityFloor(await run());
2013		} catch (error) {
2014		return failure(owner, error);
	2027	const first = await Promise.race([work, gone.promise]);
	2028	if (first !== "gone") return first;
	2029	work.then((late) => late.body?.cancel().catch(() => {}));
		sentinelcrew 29d ago No test covers this line: removing it leaves the suite green. Add a test where the handler resolves after the abort with a stream whose `cancel` sets a flag, and assert the flag is set. This is the issue's "streaming bodies are cancelled" bullet. sentinelcrew 29d ago Minor, not blocking: after this 499 the handler is still running, and the caller goes straight on to `settleBody(request)`. If a handler has called `getReader()` on a chunked body but hasn't read from it yet, `bodyUsed` is still false while the stream is locked, so `drain`'s `getReader()` would throw out of the fetch handler. The client is already gone, so nothing visible breaks, but a guard (`request.body.locked`) in `settleBody` would make the edge case explicit.
	2030	return securityFloor(new Response(null, { status: 499 }));
	2031	} finally {
	2032	request.signal.removeEventListener("abort", onAbort);
2015	2033	}
2016	2034	};
2017	2035
		@@ -2061,7 +2079,7 @@ export async function serve(app: App): Promise<Servers> {
2061	2079	}
2062	2080	notePeer(request, server.requestIP(request)?.address, trust);
2063	2081	armClose(request, () => server.timeout(request, 1));
2064		const response = await answered(child, () => http.handle(request));
	2082	const response = await answered(child, request, () => http.handle(request));
2065	2083	counted(child, request, response);
2066	2084	logged(key, request, response, started);
2067	2085	await settleBody(request);
		@@ -2078,7 +2096,7 @@ export async function serve(app: App): Promise<Servers> {
2078	2096	armClose(request, () => server.timeout(request, 1));
2079	2097	// Everything the forge answers for itself leaves through this one
2080	2098	// door, so a responder added next year cannot forget the floor.
2081		const response = await answered(app, () =>
	2099	const response = await answered(app, request, () =>
2082	2100	answer(app, router, apiRouter, limiter, request),
2083	2101	);
2084	2102	counted(app, request, response);
		@@ -2135,6 +2153,9 @@ export async function serve(app: App): Promise<Servers> {
2135	2153	return {
2136	2154	url: `http://localhost:${port}`,
2137	2155	port,
	2156	get inFlight() {
	2157	return web.pendingRequests;
	2158	},
2138	2159	async stopAccepting(budgetMs) {
2139	2160	// Closing the listener leaves the port to the process beside this one;
2140	2161	// the connections already here are answered to the end.
		@@ -0,0 +1,106 @@
	1	/** A client that goes away settles its request as 499 and frees the handler (#2250). */
	2
	3	import { afterAll, beforeAll, expect, spyOn, test } from "bun:test";
	4	import { readdirSync, readFileSync } from "node:fs";
	5	import { mkdtemp, rm } from "node:fs/promises";
	6	import { tmpdir } from "node:os";
	7	import { join } from "node:path";
	8	import { createApp, type App } from "../src/app.ts";
	9	import { accessDir, flushAccess } from "../src/http/access_log.ts";
	10	import * as health from "../src/http/health.ts";
	11	import { serve } from "../src/http/server.ts";
	12	import { readUsage } from "../src/core/usage_count.ts";
	13
	14	let dataDir: string;
	15	let app: App;
	16
	17	beforeAll(async () => {
	18	dataDir = await mkdtemp(join(tmpdir(), "forge-abort-"));
	19	app = createApp({ dataDir, port: 0 });
	20	});
	21
	22	afterAll(async () => {
	23	app.close();
	24	await rm(dataDir, { recursive: true, force: true });
	25	});
	26
	27	test("an aborted slow request is logged and counted as 499, and nothing stays in flight", async () => {
	28	const servers = await serve(app);
	29	// A handler whose Response never comes: what an aborted request left behind.
	30	const stuck = spyOn(health, "readyResponse").mockImplementation(
	31	(() => new Promise(() => {})) as never,
	32	);
	33	try {
	34	const controller = new AbortController();
	35	const pending = fetch(`${servers.url}/readyz`, { signal: controller.signal }).catch(
	36	() => "aborted",
	37	);
	38	for (let i = 0; i < 100 && servers.inFlight === 0; i++) await Bun.sleep(10);
	39	expect(servers.inFlight).toBe(1);
	40	controller.abort();
	41	expect(await pending).toBe("aborted");
	42	for (let i = 0; i < 100 && servers.inFlight > 0; i++) await Bun.sleep(10);
		sentinelcrew 29d ago outdated If an expectation above fails, the spy is never restored and the server never stops. Put `stuck.mockRestore()` and `servers.stop()` in a `try/finally`.
	43
	44	expect(servers.inFlight).toBe(0);
	45	flushAccess();
	46	const kept = readdirSync(accessDir(dataDir))
	47	.flatMap((f) => readFileSync(join(accessDir(dataDir), f), "utf8").split("\n"))
	48	.filter((l) => l !== "")
	49	.map((l) => JSON.parse(l) as { path: string; status: number });
	50	expect(kept.find((l) => l.path === "/readyz")?.status).toBe(499);
	51	} finally {
	52	stuck.mockRestore();
	53	servers.stop();
	54	}
	55	// Stopping the server settles the usage counts.
	56	const rows = readUsage(app.db, "").filter((r) => r.pattern === "/readyz");
	57	expect(rows.map((r) => [r.status, r.n])).toEqual([["4xx", 1]]);
	58	});
	59
	60	test("a body that arrives after the client left is cancelled", async () => {
	61	const servers = await serve(app);
		sentinelcrew 29d ago outdated `res`'s body is never read, so `pendingRequests` can still count this request here. Await `res.text()` first so the assertion does not depend on timing.
	62	let cancelled = false;
	63	let release!: (r: Response) => void;
	64	const stuck = spyOn(health, "readyResponse").mockImplementation(
	65	(() => new Promise<Response>((resolve) => (release = resolve))) as never,
	66	);
	67	try {
	68	const controller = new AbortController();
	69	const pending = fetch(`${servers.url}/readyz`, { signal: controller.signal }).catch(
	70	() => "aborted",
	71	);
	72	for (let i = 0; i < 100 && servers.inFlight === 0; i++) await Bun.sleep(10);
	73	expect(servers.inFlight).toBe(1);
	74	controller.abort();
	75	expect(await pending).toBe("aborted");
	76	for (let i = 0; i < 100 && servers.inFlight > 0; i++) await Bun.sleep(10);
	77	release(
	78	new Response(
	79	new ReadableStream({
	80	pull(c) {
	81	c.enqueue(new TextEncoder().encode("late"));
	82	},
	83	cancel() {
	84	cancelled = true;
	85	},
	86	}),
	87	),
	88	);
	89	for (let i = 0; i < 100 && !cancelled; i++) await Bun.sleep(10);
	90	expect(cancelled).toBe(true);
	91	expect(servers.inFlight).toBe(0);
	92	} finally {
	93	stuck.mockRestore();
	94	servers.stop();
	95	}
	96	});
	97
	98	test("a request that is not aborted is untouched, floor and all", async () => {
	99	const servers = await serve(app);
	100	const res = await fetch(`${servers.url}/readyz`);
	101	expect(res.status).toBe(200);
	102	expect(res.headers.get("x-content-type-options")).toBe("nosniff");
	103	await res.text();
	104	expect(servers.inFlight).toBe(0);
	105	servers.stop();
	106	});