bfebc4c · 9c0ccd2 · be14185
@@ -6,6 +6,13 @@
it (`none` when no request did) — the same reading as the `event loop blocked`
log lines (`src/perf/loop.ts`), so a fix's before and after reads from the
forge itself, not the pod log.
A request still running when the wake comes (#11478) is charged too, when it
began at or before the stall did: its row is `<route> (in flight)`, apart from
the same route's finished requests, in every place below (MCP `system`
`stalls` as `{ cause, ms, count }` with `cause` that name). The open set is
`src/perf/in_flight.ts`: one map entry per request being answered, added and
removed in O(1), read only at a stall.
Where to look (a forge admin who is a person; anyone else gets 404, or no
palette entry):
@@ -47,5 +54,5 @@
figure for the last hour; for the same 6-minute window,
in Prometheus, compared with 69,600.
Caveats: a stall over two requests counts against both, so routes sum past
the wall time; a request still running when the wake comes is not yet named;
a restart starts the figures at zero.
the wall time; a request held open across many stalls (a slow upload, say)
is charged with each; a restart starts the figures at zero.
@@ -10,7 +10,7 @@
import { good, bad, type Tool } from "../shared.ts";
export const tool: Tool = {
  name: "system",
  doc: "the machine the forge runs on, now: CPU, load, memory, disk, network, processes, `stalls`: the last hour's main-thread stalls by route or `job:<name>` as {cause, ms, count}, `readings`: the last 20 minutes' worst loop lag as {at, ms, what} beside `barMs` (100) and `over`, and `readers`: who polls the telemetry and how fast; a forge admin, or an agent granted `telemetry:read`. `{ kind: \"csp\" }`: the last 24 h of Content-Security-Policy violations browsers reported, `{ directive, n, newest }` per directive (`newest` the blocked host last counted), most first, counted per UTC hour from `since`, and `scriptErrors` `{ page, where, message, n, last }`, as the security page reads them; a forge admin only",
  doc: "the machine the forge runs on, now: CPU, load, memory, disk, network, processes, `stalls`: the last hour's main-thread stalls by route (`<route> (in flight)` when still running at the wake), or `job:<name>`, as {cause, ms, count}, `readings`: the last 20 minutes' worst loop lag as {at, ms, what} beside `barMs` (100) and `over`, and `readers`: who polls the telemetry and how fast; a forge admin, or an agent granted `telemetry:read`. `{ kind: \"csp\" }`: the last 24 h of Content-Security-Policy violations browsers reported, `{ directive, n, newest }` per directive (`newest` the blocked host last counted), most first, counted per UTC hour from `since`, and `scriptErrors` `{ page, where, message, n, last }`, as the security page reads them; a forge admin only",
  description:
    "The forge machine now: CPU, memory, disk, network. Admin or telemetry:read; `kind: csp`: 24 h of CSP violations (admin).",
  inputSchema: { type: "object", properties: { kind: { type: "string", enum: ["csp"] } } },
@@ -519,7 +519,7 @@
import {
  takeDbTime,
  type RouteTiming,
} from "../core/route_timing.ts";
import { captureStore, loopWatch, perfRecord, readLedger, stallBlame } from "../perf/live.ts";
import { captureStore, inFlight, loopWatch, perfRecord, readLedger, stallBlame } from "../perf/live.ts";
import { during } from "../perf/jobs.ts";
import { classOf, countryOf, repoOf, signatureOf } from "../perf/traffic.ts";
import { createTrafficWriter } from "../perf/traffic_store.ts";
@@ -2759,7 +2759,8 @@
export async function serve(app: App): Promise<Servers> {
    // The only sockets the forge holds are an app's, carried through the
    // proxy; its own live surfaces are SSE (`sse.ts`) and stay so.
    websocket: withShells(appWebSocketWith(app.config.wsMaxMessage)),
    fetch: handle,
    // 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)),
    // A last resort: `handle` answers a throw itself, so it is counted.
    error: (error) => failure(app, error),
  });
@@ -43,6 +43,11 @@
export function noteUsage(request: Request, tag: UsageTag): void {
  noted.set(request, tag);
}
/** The pattern noted for this request so far, left in place; `(unmatched)` before the router names it. */
export function patternOf(request: Request): string {
  return noted.get(request)?.pattern ?? UNMATCHED;
}
/** What was noted for this request, or the unmatched, anonymous default. */
export function takeUsage(request: Request): UsageTag {
  const tag = noted.get(request);
@@ -30,7 +30,7 @@
export function stallFamilies(tally: StallTally): Family[] {
    {
      type: "counter",
      name: "forge_loop_stall_milliseconds_total",
      help: "Milliseconds the main thread stalled past 200 ms, by the route running across the stall (route=\"none\" when none), since the process started.",
      help: "Milliseconds the main thread stalled past 200 ms, by the route running across the stall (\"<route> (in flight)\" when still running at the wake, route=\"none\" when none), since the process started.",
      samples: tally.total.map((r) => ({ labels: { route: r.route }, value: Math.round(r.ms) })),
    },
    {
@@ -0,0 +1,51 @@
/**
 * The requests the forge is answering right now (#11478): each one's start,
 * so a loop stall can name the request still running when the wake comes,
 * not only the ones that finished across it (`loop.ts`).
 *
 * One `Map` entry per open request, keyed by the request itself: `begin` and
 * `end` are a `set` and a `delete`, O(1), and the map holds only what is
 * open — it never grows with the requests already answered. The pattern is
 * read only at a wake (`nameOf`), since the router names a request after it
 * starts; a stall is rare, so the walk over the open set costs nothing on
 * the common request.
 */
export interface InFlight<K> {
  /** `key` started at `at` (`performance.now`). */
  begin(key: K, at: number): void;
  /** `key` is answered; a key never begun is ignored. */
  end(key: K): void;
  /** Run `work` with `key` open from `at` until it settles, however it settles. */
  during<T>(key: K, at: number, work: () => Promise<T>): Promise<T>;
  /** The patterns of the requests begun at or before `from` and still open, oldest first, each once. */
  since(from: number): string[];
  /** How many are open now. */
  size(): number;
}
export function createInFlight<K>(nameOf: (key: K) => string): InFlight<K> {
  const open = new Map<K, number>();
  return {
    begin(key, at) {
      open.set(key, at);
    },
    end(key) {
      open.delete(key);
    },
    async during(key, at, work) {
      open.set(key, at);
      try {
        return await work();
      } finally {
        open.delete(key);
      }
    },
    since(from) {
      const old: [number, string][] = [];
      for (const [key, at] of open) if (at <= from) old.push([at, nameOf(key)]);
      return [...new Set(old.sort((a, b) => a[0] - b[0]).map(([, name]) => name))];
    },
    size: () => open.size,
  };
}
@@ -4,6 +4,8 @@
import { createBootStall } from "./boot_stall.ts";
import { createGitRecord } from "./git.ts";
import { jobWatch } from "./jobs.ts";
import { createLoopWatch } from "./loop.ts";
import { createInFlight } from "./in_flight.ts";
import { patternOf } from "../http/usage_tag.ts";
import { createCaptureStore } from "./capture.ts";
import { createReadLedger } from "./reads.ts";
import { createStallBlame } from "./stall_blame.ts";
@@ -11,8 +13,13 @@
import { createPerfRecord } from "./record.ts";
import { createLagMinutes } from "./lag_minutes.ts";
export const perfRecord = createPerfRecord();
/** A stall no request ran across names the background jobs that did (#10834). */
export const loopWatch = createLoopWatch(undefined, jobWatch.across);
/** The requests the server is answering now (#11478): begun and ended around each one in `http/server.ts`. */
export const inFlight = createInFlight<Request>(patternOf);
/**
 * A stall names the requests that ran across it, finished or still in flight
 * (#11478); when none did, the background jobs that did (#10834).
 */
export const loopWatch = createLoopWatch(undefined, jobWatch.across, inFlight.since);
export const gitRecord = createGitRecord();
/** The worst stall in this process's first five minutes, read by the release keeper (#8848). */
export const bootStall = createBootStall();
@@ -14,6 +14,10 @@
 * A stall no request ran across is counted against the background jobs that
 * did (#10834, `jobs.ts`), each as `job:<name>`; `NO_REQUEST` is left for a
 * stall nothing named ran across.
 *
 * A request still running when the wake comes (#11478), begun at or before
 * the stall began, is charged with it too, as `<pattern> (in flight)`: a
 * stall that never ends in a finished request is not blamed on nothing.
 */
/** A wake later than this is a stall, ms (#11069: 200, so a stall the probes would feel is named). */
@@ -29,6 +33,12 @@
export const NO_REQUEST = "none";
export const JOB_PREFIX = "job:";
/** The jobs that ran across `[from, to]` (`JobWatch.across`). */
export type JobsAcross = (from: number, to: number) => readonly string[];
/** The patterns of the requests begun at or before `from` and still open (`InFlight.since`). */
export type OpenSince = (from: number) => readonly string[];
/** What marks a route charged while still running, apart from one that finished. */
export const IN_FLIGHT = " (in flight)";
/** The name an open request's pattern is charged under. */
export const inFlightName = (pattern: string): string => pattern + IN_FLIGHT;
export interface RouteStall {
  readonly route: string;
@@ -47,6 +57,8 @@
export interface Stall {
  readonly at: number;
  readonly lateMs: number;
  readonly routes: readonly string[];
  /** The patterns of the requests begun before it and still open at the wake, oldest first (#11478). */
  readonly inFlight?: readonly string[];
  /** The background jobs running across it; read only when no request was. */
  readonly jobs: readonly string[];
}
@@ -68,9 +80,12 @@
export interface LoopWatch {
  clear(): void;
}
/** What a stall is counted against: its routes, else its jobs, else `NO_REQUEST`. */
export const blamed = (s: Stall): readonly string[] =>
  s.routes.length > 0 ? s.routes : s.jobs.length > 0 ? s.jobs.map((j) => JOB_PREFIX + j) : [NO_REQUEST];
/** What a stall is counted against: its routes and its requests in flight, else its jobs, else `NO_REQUEST`. */
export function blamed(s: Stall): readonly string[] {
  const open = s.inFlight ?? [];
  if (open.length > 0) return [...s.routes, ...open.map(inFlightName)];
  return s.routes.length > 0 ? s.routes : s.jobs.length > 0 ? s.jobs.map((j) => JOB_PREFIX + j) : [NO_REQUEST];
}
function byRoute(stalls: Iterable<Stall>): RouteStall[] {
  const sums = new Map<string, { ms: number; count: number }>();
@@ -84,7 +99,7 @@
function byRoute(stalls: Iterable<Stall>): RouteStall[] {
  return [...sums].map(([route, t]) => ({ route, ms: t.ms, count: t.count }));
}
export function createLoopWatch(limitMs = LIMIT_MS, jobsAcross: JobsAcross = () => []): LoopWatch {
export function createLoopWatch(limitMs = LIMIT_MS, jobsAcross: JobsAcross = () => [], openSince: OpenSince = () => []): LoopWatch {
  const recent: { pattern: string; start: number; end: number }[] = [];
  const totals = new Map<string, { ms: number; count: number }>();
  const hour: Stall[] = [];
@@ -104,8 +119,9 @@
export function createLoopWatch(limitMs = LIMIT_MS, jobsAcross: JobsAcross = ()
            .map((r) => r.pattern),
        ),
      ];
      const jobs = routes.length === 0 ? jobsAcross(from, at) : [];
      const stall = { at, lateMs, routes, jobs };
      const inFlight = openSince(from);
      const jobs = routes.length === 0 && inFlight.length === 0 ? jobsAcross(from, at) : [];
      const stall: Stall = inFlight.length > 0 ? { at, lateMs, routes, inFlight, jobs } : { at, lateMs, routes, jobs };
      for (const r of byRoute([stall])) {
        const t = totals.get(r.route) ?? { ms: 0, count: 0 };
        t.ms += r.ms;
@@ -143,5 +159,9 @@
export function createLoopWatch(limitMs = LIMIT_MS, jobsAcross: JobsAcross = ()
/** The stall as a finding line. */
export const stallLine = (s: Stall): string =>
  `event loop blocked ${Math.round(s.lateMs)} ms; running then: ${
    s.routes.length > 0 ? s.routes.join(", ") : s.jobs.length > 0 ? `no request; jobs ${s.jobs.join(", ")}` : "no request"
    s.routes.length > 0 || (s.inFlight ?? []).length > 0
      ? [...s.routes, ...(s.inFlight ?? []).map(inFlightName)].join(", ")
      : s.jobs.length > 0
        ? `no request; jobs ${s.jobs.join(", ")}`
        : "no request"
  }`;
@@ -4,10 +4,11 @@
 * can say what held the loop behind its `loopLagMs`.
 *
 * Who ran is the loop watch's finding (`loop.ts`): the route patterns whose
 * request ran across the blocked window, else the background jobs marked
 * request ran across the blocked window, then those still running at the
 * wake as `<pattern> (in flight)` (#11478), else the background jobs marked
 * with `jobWatch.during` (`jobs.ts`), else nothing. `what` is the first of
 * those — the longest-running route, or `job:<name>` — and `unknown` when
 * nothing marked ran across the stall.
 * those — the longest-running finished route, the oldest open one, or
 * `job:<name>` — and `unknown` when nothing marked ran across the stall.
 *
 * The ring holds `KEPT` stalls and never grows: a note is three array
 * writes and an index step. At one stall per 200 ms an hour would be 18,000,
@@ -0,0 +1,181 @@
/** A loop stall names the request still running when the loop woke, not only the ones that finished (#11478). */
import { afterAll, beforeAll, expect, test } from "bun:test";
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 { createToken } from "../src/core/tokens.ts";
import { createUser } from "../src/core/users.ts";
import { serve, type Servers } from "../src/http/server.ts";
import { stallFamilies } from "../src/metrics/stalls.ts";
import { createInFlight } from "../src/perf/in_flight.ts";
import { inFlight, loopWatch } from "../src/perf/live.ts";
import { blamed, createLoopWatch, LIMIT_MS, NO_REQUEST, stallLine } from "../src/perf/loop.ts";
import { whatOf } from "../src/perf/stall_blame.ts";
import { quietLogs } from "./helpers/globals.ts";
import { freshStallState } from "./helpers/stall-state.ts";
quietLogs();
freshStallState();
/** A watch over a hand-held open set, on a clock the test names. */
function watched() {
  const open = createInFlight<string>((key) => key.split("#")[0]!);
  return { open, watch: createLoopWatch(250, () => ["sampler"], open.since) };
}
test("a request held open across a stall is charged with it, labelled in flight", () => {
  const { open, watch } = watched();
  open.begin("/explore#1", 1_000);
  // The wake at 2_000 is 400 ms late: the loop was held from 1_600.
  const stall = watch.stalled(400, 2_000)!;
  expect(stall.inFlight).toEqual(["/explore"]);
  expect(blamed(stall)).toEqual(["/explore (in flight)"]);
  expect(whatOf(stall)).toBe("/explore (in flight)");
  expect(stallLine(stall)).toBe("event loop blocked 400 ms; running then: /explore (in flight)");
  // Charged in place of the job: a request ran, so the job is not read.
  expect(stall.jobs).toEqual([]);
  expect(watch.tally(2_000).hour).toEqual([{ route: "/explore (in flight)", ms: 400, count: 1 }]);
});
test("a request that started after the stall began is not charged", () => {
  const { open, watch } = watched();
  open.begin("/late#1", 1_700);
  const stall = watch.stalled(400, 2_000)!;
  expect(stall.inFlight).toBeUndefined();
  expect(blamed(stall)).toEqual(["job:sampler"]);
});
test("one started exactly when the stall began is older than it, and charged", () => {
  const { open, watch } = watched();
  open.begin("/edge#1", 1_600);
  expect(blamed(watch.stalled(400, 2_000)!)).toEqual(["/edge (in flight)"]);
});
test("an ended request is no longer in flight; finished and open both carry the stall", () => {
  const { open, watch } = watched();
  open.begin("/gone#1", 1_000);
  open.end("/gone#1");
  expect(blamed(watch.stalled(400, 2_000)!)).toEqual(["job:sampler"]);
  open.begin("/held#1", 2_100);
  watch.finished("/done", 2_500, 2_900);
  expect(blamed(watch.stalled(400, 3_000)!)).toEqual(["/done", "/held (in flight)"]);
});
test("open requests are named oldest first, each pattern once", () => {
  const open = createInFlight<string>((key) => key.split("#")[0]!);
  open.begin("/b#1", 30);
  open.begin("/a#1", 10);
  open.begin("/b#2", 20);
  open.begin("/c#1", 90);
  expect(open.since(50)).toEqual(["/a", "/b"]);
});
test("the open set holds only what is open: a million requests begun and ended leave it empty", () => {
  const open = createInFlight<object>(() => "/x");
  for (let i = 0; i < 1_000_000; i++) {
    const key = {};
    open.begin(key, i);
    open.end(key);
  }
  expect(open.size()).toBe(0);
  // Ending what was never begun is harmless.
  open.end({});
  expect(open.size()).toBe(0);
});
test("`during` holds a request open while it runs and lets it go however it settles", async () => {
  const open = createInFlight<object>(() => "/x");
  const key = {};
  const seen = await open.during(key, 5, async () => open.since(5));
  expect(seen).toEqual(["/x"]);
  expect(open.size()).toBe(0);
  await expect(open.during({}, 5, async () => { throw new Error("boom"); })).rejects.toThrow("boom");
  expect(open.size()).toBe(0);
});
test("a stall nothing ran across is still `none`", () => {
  const watch = createLoopWatch(250);
  expect(blamed(watch.stalled(400, 2_000)!)).toEqual([NO_REQUEST]);
});
test("the in-flight row is a metrics series like any route", () => {
  const { open, watch } = watched();
  open.begin("/explore#1", 1_000);
  watch.stalled(400, 2_000);
  const total = stallFamilies(watch.tally(2_000)).find((f) => f.name === "forge_loop_stall_milliseconds_total")!;
  expect(total.samples).toContainEqual({ labels: { route: "/explore (in flight)" }, value: 400 });
});
describe_server();
function describe_server() {
  let dataDir = "";
  let app: App;
  let servers: Servers;
  let token = "";
  beforeAll(async () => {
    dataDir = await mkdtemp(join(tmpdir(), "forge-stall-inflight-"));
    app = createApp({ dataDir, port: 0 });
    servers = await serve(app);
    const u = await createUser(app.db, { origin: "seed", trusted: true, username: "alice", email: "[email protected]", password: "correct horse battery 9", isAdmin: true });
    if (!u.ok) throw new Error("alice");
    token = createToken(app.db, u.value.id, "t", "admin").token;
  }, 30_000);
  afterAll(async () => {
    servers.stop();
    app.close();
    await rm(dataDir, { recursive: true, force: true });
  });
  const system = async (): Promise<any> => {
    const res = await fetch(`${servers.url}/mcp`, {
      method: "POST",
      headers: { "content-type": "application/json", accept: "application/json, text/event-stream", authorization: `Bearer ${token}` },
      body: JSON.stringify({ jsonrpc: "2.0", id: 1, method: "tools/call", params: { name: "system", arguments: {} } }),
    });
    const body = (await res.json()) as any;
    return JSON.parse(body.result.content[0].text);
  };
  test("a real request held open across a stall is in MCP `system` `stalls`; one begun after is not", async () => {
    loopWatch.clear();
    // A request whose body has not finished arriving: the server is answering it, and it is open.
    let push: ((chunk: Uint8Array) => void) | undefined;
    let close: (() => void) | undefined;
    const body = new ReadableStream<Uint8Array>({
      start(c) {
        push = (chunk) => c.enqueue(chunk);
        close = () => c.close();
      },
    });
    const held = fetch(`${servers.url}/mcp`, {
      method: "POST",
      headers: { "content-type": "application/json", accept: "application/json, text/event-stream", authorization: `Bearer ${token}` },
      body,
      duplex: "half",
    } as RequestInit);
    push!(new TextEncoder().encode('{"jsonrpc":"2.0",'));
    const deadline = Date.now() + 5_000;
    while (inFlight.size() === 0 && Date.now() < deadline) await Bun.sleep(5);
    expect(inFlight.size()).toBe(1);
    await Bun.sleep(LIMIT_MS + 60);
    // The wake: the loop held for longer than the stall limit, beginning after the request did.
    const now = performance.now();
    const stall = loopWatch.stalled(LIMIT_MS + 10, now)!;
    const named = stall.inFlight ?? [];
    expect(named).toEqual(["/mcp"]);
    // A stall shorter than the request's age is charged; one reaching back before it began is not.
    expect(loopWatch.stalled(60_000, now)!.inFlight).toBeUndefined();
    push!(new TextEncoder().encode('"id":9,"method":"tools/list"}'));
    close!();
    await (await held).text();
    expect(inFlight.size()).toBe(0);
    const out = await system();
    expect(out.stalls).toContainEqual({ cause: `${named[0]} (in flight)`, ms: LIMIT_MS + 10, count: 1 });
  });
}
		@@ -6,6 +6,13 @@ it (`none` when no request did) — the same reading as the `event loop blocked`
6	6	log lines (`src/perf/loop.ts`), so a fix's before and after reads from the
7	7	forge itself, not the pod log.
8	8
	9	A request still running when the wake comes (#11478) is charged too, when it
	10	began at or before the stall did: its row is `<route> (in flight)`, apart from
	11	the same route's finished requests, in every place below (MCP `system`
	12	`stalls` as `{ cause, ms, count }` with `cause` that name). The open set is
	13	`src/perf/in_flight.ts`: one map entry per request being answered, added and
	14	removed in O(1), read only at a stall.
	15
9	16	Where to look (a forge admin who is a person; anyone else gets 404, or no
10	17	palette entry):
11	18
		@@ -47,5 +54,5 @@ figure for the last hour; for the same 6-minute window,
47	54	in Prometheus, compared with 69,600.
48	55
49	56	Caveats: a stall over two requests counts against both, so routes sum past
50		the wall time; a request still running when the wake comes is not yet named;
51		a restart starts the figures at zero.
	57	the wall time; a request held open across many stalls (a slow upload, say)
	58	is charged with each; a restart starts the figures at zero.
		@@ -10,7 +10,7 @@ import { good, bad, type Tool } from "../shared.ts";
10	10
11	11	export const tool: Tool = {
12	12	name: "system",
13		doc: "the machine the forge runs on, now: CPU, load, memory, disk, network, processes, `stalls`: the last hour's main-thread stalls by route or `job:<name>` as {cause, ms, count}, `readings`: the last 20 minutes' worst loop lag as {at, ms, what} beside `barMs` (100) and `over`, and `readers`: who polls the telemetry and how fast; a forge admin, or an agent granted `telemetry:read`. `{ kind: \"csp\" }`: the last 24 h of Content-Security-Policy violations browsers reported, `{ directive, n, newest }` per directive (`newest` the blocked host last counted), most first, counted per UTC hour from `since`, and `scriptErrors` `{ page, where, message, n, last }`, as the security page reads them; a forge admin only",
	13	doc: "the machine the forge runs on, now: CPU, load, memory, disk, network, processes, `stalls`: the last hour's main-thread stalls by route (`<route> (in flight)` when still running at the wake), or `job:<name>`, as {cause, ms, count}, `readings`: the last 20 minutes' worst loop lag as {at, ms, what} beside `barMs` (100) and `over`, and `readers`: who polls the telemetry and how fast; a forge admin, or an agent granted `telemetry:read`. `{ kind: \"csp\" }`: the last 24 h of Content-Security-Policy violations browsers reported, `{ directive, n, newest }` per directive (`newest` the blocked host last counted), most first, counted per UTC hour from `since`, and `scriptErrors` `{ page, where, message, n, last }`, as the security page reads them; a forge admin only",
14	14	description:
15	15	"The forge machine now: CPU, memory, disk, network. Admin or telemetry:read; `kind: csp`: 24 h of CSP violations (admin).",
16	16	inputSchema: { type: "object", properties: { kind: { type: "string", enum: ["csp"] } } },
		@@ -519,7 +519,7 @@ import {
519	519	takeDbTime,
520	520	type RouteTiming,
521	521	} from "../core/route_timing.ts";
522		import { captureStore, loopWatch, perfRecord, readLedger, stallBlame } from "../perf/live.ts";
	522	import { captureStore, inFlight, loopWatch, perfRecord, readLedger, stallBlame } from "../perf/live.ts";
523	523	import { during } from "../perf/jobs.ts";
524	524	import { classOf, countryOf, repoOf, signatureOf } from "../perf/traffic.ts";
525	525	import { createTrafficWriter } from "../perf/traffic_store.ts";
		@@ -2759,7 +2759,8 @@ export async function serve(app: App): Promise<Servers> {
2759	2759	// The only sockets the forge holds are an app's, carried through the
2760	2760	// proxy; its own live surfaces are SSE (`sse.ts`) and stay so.
2761	2761	websocket: withShells(appWebSocketWith(app.config.wsMaxMessage)),
2762		fetch: handle,
	2762	// Open until it is answered, so a stall while it runs names it (#11478).
	2763	fetch: (request, server) => inFlight.during(request, performance.now(), () => handle(request, server)),
2763	2764	// A last resort: `handle` answers a throw itself, so it is counted.
2764	2765	error: (error) => failure(app, error),
2765	2766	});
		@@ -43,6 +43,11 @@ export function noteUsage(request: Request, tag: UsageTag): void {
43	43	noted.set(request, tag);
44	44	}
45	45
	46	/** The pattern noted for this request so far, left in place; `(unmatched)` before the router names it. */
	47	export function patternOf(request: Request): string {
	48	return noted.get(request)?.pattern ?? UNMATCHED;
	49	}
	50
46	51	/** What was noted for this request, or the unmatched, anonymous default. */
47	52	export function takeUsage(request: Request): UsageTag {
48	53	const tag = noted.get(request);
		@@ -30,7 +30,7 @@ export function stallFamilies(tally: StallTally): Family[] {
30	30	{
31	31	type: "counter",
32	32	name: "forge_loop_stall_milliseconds_total",
33		help: "Milliseconds the main thread stalled past 200 ms, by the route running across the stall (route=\"none\" when none), since the process started.",
	33	help: "Milliseconds the main thread stalled past 200 ms, by the route running across the stall (\"<route> (in flight)\" when still running at the wake, route=\"none\" when none), since the process started.",
34	34	samples: tally.total.map((r) => ({ labels: { route: r.route }, value: Math.round(r.ms) })),
35	35	},
36	36	{
		@@ -0,0 +1,51 @@
	1	/**
	2	* The requests the forge is answering right now (#11478): each one's start,
	3	* so a loop stall can name the request still running when the wake comes,
	4	* not only the ones that finished across it (`loop.ts`).
	5	*
	6	* One `Map` entry per open request, keyed by the request itself: `begin` and
	7	* `end` are a `set` and a `delete`, O(1), and the map holds only what is
	8	* open — it never grows with the requests already answered. The pattern is
	9	* read only at a wake (`nameOf`), since the router names a request after it
	10	* starts; a stall is rare, so the walk over the open set costs nothing on
	11	* the common request.
	12	*/
	13
	14	export interface InFlight<K> {
	15	/** `key` started at `at` (`performance.now`). */
	16	begin(key: K, at: number): void;
	17	/** `key` is answered; a key never begun is ignored. */
	18	end(key: K): void;
	19	/** Run `work` with `key` open from `at` until it settles, however it settles. */
	20	during<T>(key: K, at: number, work: () => Promise<T>): Promise<T>;
	21	/** The patterns of the requests begun at or before `from` and still open, oldest first, each once. */
	22	since(from: number): string[];
	23	/** How many are open now. */
	24	size(): number;
	25	}
	26
	27	export function createInFlight<K>(nameOf: (key: K) => string): InFlight<K> {
	28	const open = new Map<K, number>();
	29	return {
	30	begin(key, at) {
	31	open.set(key, at);
	32	},
	33	end(key) {
	34	open.delete(key);
	35	},
	36	async during(key, at, work) {
	37	open.set(key, at);
	38	try {
	39	return await work();
	40	} finally {
	41	open.delete(key);
	42	}
	43	},
	44	since(from) {
	45	const old: [number, string][] = [];
	46	for (const [key, at] of open) if (at <= from) old.push([at, nameOf(key)]);
	47	return [...new Set(old.sort((a, b) => a[0] - b[0]).map(([, name]) => name))];
	48	},
	49	size: () => open.size,
	50	};
	51	}
		@@ -4,6 +4,8 @@ import { createBootStall } from "./boot_stall.ts";
4	4	import { createGitRecord } from "./git.ts";
5	5	import { jobWatch } from "./jobs.ts";
6	6	import { createLoopWatch } from "./loop.ts";
	7	import { createInFlight } from "./in_flight.ts";
	8	import { patternOf } from "../http/usage_tag.ts";
7	9	import { createCaptureStore } from "./capture.ts";
8	10	import { createReadLedger } from "./reads.ts";
9	11	import { createStallBlame } from "./stall_blame.ts";
		@@ -11,8 +13,13 @@ import { createPerfRecord } from "./record.ts";
11	13	import { createLagMinutes } from "./lag_minutes.ts";
12	14
13	15	export const perfRecord = createPerfRecord();
14		/** A stall no request ran across names the background jobs that did (#10834). */
15		export const loopWatch = createLoopWatch(undefined, jobWatch.across);
	16	/** The requests the server is answering now (#11478): begun and ended around each one in `http/server.ts`. */
	17	export const inFlight = createInFlight<Request>(patternOf);
	18	/**
	19	* A stall names the requests that ran across it, finished or still in flight
	20	* (#11478); when none did, the background jobs that did (#10834).
	21	*/
	22	export const loopWatch = createLoopWatch(undefined, jobWatch.across, inFlight.since);
16	23	export const gitRecord = createGitRecord();
17	24	/** The worst stall in this process's first five minutes, read by the release keeper (#8848). */
18	25	export const bootStall = createBootStall();
		@@ -14,6 +14,10 @@
14	14	* A stall no request ran across is counted against the background jobs that
15	15	* did (#10834, `jobs.ts`), each as `job:<name>`; `NO_REQUEST` is left for a
16	16	* stall nothing named ran across.
	17	*
	18	* A request still running when the wake comes (#11478), begun at or before
	19	* the stall began, is charged with it too, as `<pattern> (in flight)`: a
	20	* stall that never ends in a finished request is not blamed on nothing.
17	21	*/
18	22
19	23	/** A wake later than this is a stall, ms (#11069: 200, so a stall the probes would feel is named). */
		@@ -29,6 +33,12 @@ export const NO_REQUEST = "none";
29	33	export const JOB_PREFIX = "job:";
30	34	/** The jobs that ran across `[from, to]` (`JobWatch.across`). */
31	35	export type JobsAcross = (from: number, to: number) => readonly string[];
	36	/** The patterns of the requests begun at or before `from` and still open (`InFlight.since`). */
	37	export type OpenSince = (from: number) => readonly string[];
	38	/** What marks a route charged while still running, apart from one that finished. */
	39	export const IN_FLIGHT = " (in flight)";
	40	/** The name an open request's pattern is charged under. */
	41	export const inFlightName = (pattern: string): string => pattern + IN_FLIGHT;
32	42
33	43	export interface RouteStall {
34	44	readonly route: string;
		@@ -47,6 +57,8 @@ export interface Stall {
47	57	readonly at: number;
48	58	readonly lateMs: number;
49	59	readonly routes: readonly string[];
	60	/** The patterns of the requests begun before it and still open at the wake, oldest first (#11478). */
	61	readonly inFlight?: readonly string[];
50	62	/** The background jobs running across it; read only when no request was. */
51	63	readonly jobs: readonly string[];
52	64	}
		@@ -68,9 +80,12 @@ export interface LoopWatch {
68	80	clear(): void;
69	81	}
70	82
71		/** What a stall is counted against: its routes, else its jobs, else `NO_REQUEST`. */
72		export const blamed = (s: Stall): readonly string[] =>
73		s.routes.length > 0 ? s.routes : s.jobs.length > 0 ? s.jobs.map((j) => JOB_PREFIX + j) : [NO_REQUEST];
	83	/** What a stall is counted against: its routes and its requests in flight, else its jobs, else `NO_REQUEST`. */
	84	export function blamed(s: Stall): readonly string[] {
	85	const open = s.inFlight ?? [];
	86	if (open.length > 0) return [...s.routes, ...open.map(inFlightName)];
	87	return s.routes.length > 0 ? s.routes : s.jobs.length > 0 ? s.jobs.map((j) => JOB_PREFIX + j) : [NO_REQUEST];
	88	}
74	89
75	90	function byRoute(stalls: Iterable<Stall>): RouteStall[] {
76	91	const sums = new Map<string, { ms: number; count: number }>();
		@@ -84,7 +99,7 @@ function byRoute(stalls: Iterable<Stall>): RouteStall[] {
84	99	return [...sums].map(([route, t]) => ({ route, ms: t.ms, count: t.count }));
85	100	}
86	101
87		export function createLoopWatch(limitMs = LIMIT_MS, jobsAcross: JobsAcross = () => []): LoopWatch {
	102	export function createLoopWatch(limitMs = LIMIT_MS, jobsAcross: JobsAcross = () => [], openSince: OpenSince = () => []): LoopWatch {
88	103	const recent: { pattern: string; start: number; end: number }[] = [];
89	104	const totals = new Map<string, { ms: number; count: number }>();
90	105	const hour: Stall[] = [];
		@@ -104,8 +119,9 @@ export function createLoopWatch(limitMs = LIMIT_MS, jobsAcross: JobsAcross = ()
104	119	.map((r) => r.pattern),
105	120	),
106	121	];
107		const jobs = routes.length === 0 ? jobsAcross(from, at) : [];
108		const stall = { at, lateMs, routes, jobs };
	122	const inFlight = openSince(from);
	123	const jobs = routes.length === 0 && inFlight.length === 0 ? jobsAcross(from, at) : [];
	124	const stall: Stall = inFlight.length > 0 ? { at, lateMs, routes, inFlight, jobs } : { at, lateMs, routes, jobs };
109	125	for (const r of byRoute([stall])) {
110	126	const t = totals.get(r.route) ?? { ms: 0, count: 0 };
111	127	t.ms += r.ms;
		@@ -143,5 +159,9 @@ export function createLoopWatch(limitMs = LIMIT_MS, jobsAcross: JobsAcross = ()
143	159	/** The stall as a finding line. */
144	160	export const stallLine = (s: Stall): string =>
145	161	`event loop blocked ${Math.round(s.lateMs)} ms; running then: ${
146		s.routes.length > 0 ? s.routes.join(", ") : s.jobs.length > 0 ? `no request; jobs ${s.jobs.join(", ")}` : "no request"
	162	s.routes.length > 0 \|\| (s.inFlight ?? []).length > 0
	163	? [...s.routes, ...(s.inFlight ?? []).map(inFlightName)].join(", ")
	164	: s.jobs.length > 0
	165	? `no request; jobs ${s.jobs.join(", ")}`
	166	: "no request"
147	167	}`;
		@@ -4,10 +4,11 @@
4	4	* can say what held the loop behind its `loopLagMs`.
5	5	*
6	6	* Who ran is the loop watch's finding (`loop.ts`): the route patterns whose
7		* request ran across the blocked window, else the background jobs marked
	7	* request ran across the blocked window, then those still running at the
	8	* wake as `<pattern> (in flight)` (#11478), else the background jobs marked
8	9	* with `jobWatch.during` (`jobs.ts`), else nothing. `what` is the first of
9		* those — the longest-running route, or `job:<name>` — and `unknown` when
10		* nothing marked ran across the stall.
	10	* those — the longest-running finished route, the oldest open one, or
	11	* `job:<name>` — and `unknown` when nothing marked ran across the stall.
11	12	*
12	13	* The ring holds `KEPT` stalls and never grows: a note is three array
13	14	* writes and an index step. At one stall per 200 ms an hour would be 18,000,
		@@ -0,0 +1,181 @@
	1	/** A loop stall names the request still running when the loop woke, not only the ones that finished (#11478). */
	2
	3	import { afterAll, beforeAll, expect, test } from "bun:test";
	4	import { mkdtemp, rm } from "node:fs/promises";
	5	import { tmpdir } from "node:os";
	6	import { join } from "node:path";
	7	import { createApp, type App } from "../src/app.ts";
	8	import { createToken } from "../src/core/tokens.ts";
	9	import { createUser } from "../src/core/users.ts";
	10	import { serve, type Servers } from "../src/http/server.ts";
	11	import { stallFamilies } from "../src/metrics/stalls.ts";
	12	import { createInFlight } from "../src/perf/in_flight.ts";
	13	import { inFlight, loopWatch } from "../src/perf/live.ts";
	14	import { blamed, createLoopWatch, LIMIT_MS, NO_REQUEST, stallLine } from "../src/perf/loop.ts";
	15	import { whatOf } from "../src/perf/stall_blame.ts";
	16	import { quietLogs } from "./helpers/globals.ts";
	17	import { freshStallState } from "./helpers/stall-state.ts";
	18
	19	quietLogs();
	20	freshStallState();
	21
	22	/** A watch over a hand-held open set, on a clock the test names. */
	23	function watched() {
	24	const open = createInFlight<string>((key) => key.split("#")[0]!);
	25	return { open, watch: createLoopWatch(250, () => ["sampler"], open.since) };
	26	}
	27
	28	test("a request held open across a stall is charged with it, labelled in flight", () => {
	29	const { open, watch } = watched();
	30	open.begin("/explore#1", 1_000);
	31	// The wake at 2_000 is 400 ms late: the loop was held from 1_600.
	32	const stall = watch.stalled(400, 2_000)!;
	33	expect(stall.inFlight).toEqual(["/explore"]);
	34	expect(blamed(stall)).toEqual(["/explore (in flight)"]);
	35	expect(whatOf(stall)).toBe("/explore (in flight)");
	36	expect(stallLine(stall)).toBe("event loop blocked 400 ms; running then: /explore (in flight)");
	37	// Charged in place of the job: a request ran, so the job is not read.
	38	expect(stall.jobs).toEqual([]);
	39	expect(watch.tally(2_000).hour).toEqual([{ route: "/explore (in flight)", ms: 400, count: 1 }]);
	40	});
	41
	42	test("a request that started after the stall began is not charged", () => {
	43	const { open, watch } = watched();
	44	open.begin("/late#1", 1_700);
	45	const stall = watch.stalled(400, 2_000)!;
	46	expect(stall.inFlight).toBeUndefined();
	47	expect(blamed(stall)).toEqual(["job:sampler"]);
	48	});
	49
	50	test("one started exactly when the stall began is older than it, and charged", () => {
	51	const { open, watch } = watched();
	52	open.begin("/edge#1", 1_600);
	53	expect(blamed(watch.stalled(400, 2_000)!)).toEqual(["/edge (in flight)"]);
	54	});
	55
	56	test("an ended request is no longer in flight; finished and open both carry the stall", () => {
	57	const { open, watch } = watched();
	58	open.begin("/gone#1", 1_000);
	59	open.end("/gone#1");
	60	expect(blamed(watch.stalled(400, 2_000)!)).toEqual(["job:sampler"]);
	61	open.begin("/held#1", 2_100);
	62	watch.finished("/done", 2_500, 2_900);
	63	expect(blamed(watch.stalled(400, 3_000)!)).toEqual(["/done", "/held (in flight)"]);
	64	});
	65
	66	test("open requests are named oldest first, each pattern once", () => {
	67	const open = createInFlight<string>((key) => key.split("#")[0]!);
	68	open.begin("/b#1", 30);
	69	open.begin("/a#1", 10);
	70	open.begin("/b#2", 20);
	71	open.begin("/c#1", 90);
	72	expect(open.since(50)).toEqual(["/a", "/b"]);
	73	});
	74
	75	test("the open set holds only what is open: a million requests begun and ended leave it empty", () => {
	76	const open = createInFlight<object>(() => "/x");
	77	for (let i = 0; i < 1_000_000; i++) {
	78	const key = {};
	79	open.begin(key, i);
	80	open.end(key);
	81	}
	82	expect(open.size()).toBe(0);
	83	// Ending what was never begun is harmless.
	84	open.end({});
	85	expect(open.size()).toBe(0);
	86	});
	87
	88	test("`during` holds a request open while it runs and lets it go however it settles", async () => {
	89	const open = createInFlight<object>(() => "/x");
	90	const key = {};
	91	const seen = await open.during(key, 5, async () => open.since(5));
	92	expect(seen).toEqual(["/x"]);
	93	expect(open.size()).toBe(0);
	94	await expect(open.during({}, 5, async () => { throw new Error("boom"); })).rejects.toThrow("boom");
	95	expect(open.size()).toBe(0);
	96	});
	97
	98	test("a stall nothing ran across is still `none`", () => {
	99	const watch = createLoopWatch(250);
	100	expect(blamed(watch.stalled(400, 2_000)!)).toEqual([NO_REQUEST]);
	101	});
	102
	103	test("the in-flight row is a metrics series like any route", () => {
	104	const { open, watch } = watched();
	105	open.begin("/explore#1", 1_000);
	106	watch.stalled(400, 2_000);
	107	const total = stallFamilies(watch.tally(2_000)).find((f) => f.name === "forge_loop_stall_milliseconds_total")!;
	108	expect(total.samples).toContainEqual({ labels: { route: "/explore (in flight)" }, value: 400 });
	109	});
	110
	111	describe_server();
	112
	113	function describe_server() {
	114	let dataDir = "";
	115	let app: App;
	116	let servers: Servers;
	117	let token = "";
	118
	119	beforeAll(async () => {
	120	dataDir = await mkdtemp(join(tmpdir(), "forge-stall-inflight-"));
	121	app = createApp({ dataDir, port: 0 });
	122	servers = await serve(app);
	123	const u = await createUser(app.db, { origin: "seed", trusted: true, username: "alice", email: "[email protected]", password: "correct horse battery 9", isAdmin: true });
	124	if (!u.ok) throw new Error("alice");
	125	token = createToken(app.db, u.value.id, "t", "admin").token;
	126	}, 30_000);
	127
	128	afterAll(async () => {
	129	servers.stop();
	130	app.close();
	131	await rm(dataDir, { recursive: true, force: true });
	132	});
	133
	134	const system = async (): Promise<any> => {
	135	const res = await fetch(`${servers.url}/mcp`, {
	136	method: "POST",
	137	headers: { "content-type": "application/json", accept: "application/json, text/event-stream", authorization: `Bearer ${token}` },
	138	body: JSON.stringify({ jsonrpc: "2.0", id: 1, method: "tools/call", params: { name: "system", arguments: {} } }),
	139	});
	140	const body = (await res.json()) as any;
	141	return JSON.parse(body.result.content[0].text);
	142	};
	143
	144	test("a real request held open across a stall is in MCP `system` `stalls`; one begun after is not", async () => {
	145	loopWatch.clear();
	146	// A request whose body has not finished arriving: the server is answering it, and it is open.
	147	let push: ((chunk: Uint8Array) => void) \| undefined;
	148	let close: (() => void) \| undefined;
	149	const body = new ReadableStream<Uint8Array>({
	150	start(c) {
	151	push = (chunk) => c.enqueue(chunk);
	152	close = () => c.close();
	153	},
	154	});
	155	const held = fetch(`${servers.url}/mcp`, {
	156	method: "POST",
	157	headers: { "content-type": "application/json", accept: "application/json, text/event-stream", authorization: `Bearer ${token}` },
	158	body,
	159	duplex: "half",
	160	} as RequestInit);
	161	push!(new TextEncoder().encode('{"jsonrpc":"2.0",'));
	162	const deadline = Date.now() + 5_000;
	163	while (inFlight.size() === 0 && Date.now() < deadline) await Bun.sleep(5);
	164	expect(inFlight.size()).toBe(1);
	165	await Bun.sleep(LIMIT_MS + 60);
	166	// The wake: the loop held for longer than the stall limit, beginning after the request did.
	167	const now = performance.now();
	168	const stall = loopWatch.stalled(LIMIT_MS + 10, now)!;
	169	const named = stall.inFlight ?? [];
	170	expect(named).toEqual(["/mcp"]);
	171	// A stall shorter than the request's age is charged; one reaching back before it began is not.
	172	expect(loopWatch.stalled(60_000, now)!.inFlight).toBeUndefined();
	173	push!(new TextEncoder().encode('"id":9,"method":"tools/list"}'));
	174	close!();
	175	await (await held).text();
	176	expect(inFlight.size()).toBe(0);
	177
	178	const out = await system();
	179	expect(out.stalls).toContainEqual({ cause: `${named[0]} (in flight)`, ms: LIMIT_MS + 10, count: 1 });
	180	});
	181	}