7a8ba68 · 1eab46b · a6aa056 · d10d991 · c7532d4
@@ -846,6 +846,15 @@
file), and if you named an agent, the sentence became issue #1, assigned.
  names it exactly and a site admin has granted it: on the
  app's settings page, or in the terminal (About `I`, then `e`, `g`
  to grant, `x` to revoke). See `docs/APPS.md`.
- **Egress connections are bounded.** The proxy closes the upstream socket
  the moment the cell side closes, errors or stops, and the other way round.
  An app holds at most 8 open connections to one destination (the ninth
  gets a `429`, logged), and a connection that moves no bytes, either way, for 5 min is
  cut, so a tunnel quiet for longer than that (a websocket or SSE stream with
  very long gaps) needs a keepalive. Open connections per app and
  destination, with the bytes moved, are on the Capacity page
  (`/settings/capacity`, admins) and in the terminal: About `I`, then `c`
  (site admins).
- **Production is the default branch.** The only way code reaches it is that
  branch moving, and an agent cannot move it: the crew proposes through
  pulls, a person merges.
@@ -1914,3 +1923,8 @@
same rule. A guard installed by hand without the prefix answers `refused`
until it is re-created with `forge: ` in front of its message.
`reach` is a boards source for an admin only (anyone else gets 403, an agent too): `visitors` per day, `lcp_p75` and `inp_p75` per route per day, `server_p50` and `server_p95` per route per day (the beacon's time to first byte, the only per-route server timing the forge keeps), and `accounts_active_1d`, `accounts_active_7d` and `accounts_total` as of now. Filter by `metric`, group by `route` or a time bucket. Percentiles are nearest-rank within one route and day, so a `max` or `avg` over several days is a summary of daily percentiles, not a percentile of the period. Account activity comes from a session's last-seen time, so it is a count as of now, not a history. No dependency was added.
`GET /metrics` (site admin only; repository names are in the labels) serves
`forge_egress_connections`, `forge_egress_bytes_in` and `forge_egress_bytes_out`
per `repo` and `destination` as Prometheus text, so a leaked upstream socket
shows as a gauge that never falls (#3719).
@@ -362,6 +362,7 @@
export function openCell(
      {
        hosts: spec.egress,
        granted: spec.granted,
        label: spec.label,
        origins,
        report:
          (spec.egressMode ?? egressMode()) === "report"
@@ -38,6 +38,14 @@
export interface EgressPolicy {
   * refused. Private addresses are refused either way.
   */
  readonly report?: ((host: string, port: number) => void) | undefined;
  /** Whose proxy this is (`app:<slug>/<env>`), for the connection gauges. */
  readonly label?: string | undefined;
  /** Most connections open to one destination at once; default 8. */
  readonly maxPerDestination?: number | undefined;
  /** Close a connection that moved no bytes for this long; default 5 min. */
  readonly idleMs?: number | undefined;
  /** Bytes held for a client that does not read before it is cut; default 4 MiB. */
  readonly queueMax?: number | undefined;
}
export interface Egress {
@@ -45,6 +53,59 @@
export interface Egress {
}
const HEAD_MAX = 16 * 1024;
const MAX_PER_DESTINATION = 8;
const IDLE_MS = 5 * 60_000;
/** Bytes held for a client that does not read before its upstream is cut. */
const QUEUE_MAX = 4 * 1024 * 1024;
interface Tally {
  open: number;
  bytesIn: number;
  bytesOut: number;
}
/**
 * label -> `host:port` -> counts, for the connections open right now. A
 * destination's entry goes when its last connection closes (and the label
 * with its last destination), so it only ever holds what is live.
 */
const tallies = new Map<string, Map<string, Tally>>();
function tally(label: string, destination: string): Tally {
  let byDestination = tallies.get(label);
  if (byDestination === undefined) tallies.set(label, (byDestination = new Map()));
  let t = byDestination.get(destination);
  if (t === undefined)
    byDestination.set(destination, (t = { open: 0, bytesIn: 0, bytesOut: 0 }));
  return t;
}
/** One connection closed: drop its destination, then its label, when empty. */
function untally(label: string, destination: string): void {
  const byDestination = tallies.get(label);
  const t = byDestination?.get(destination);
  if (byDestination === undefined || t === undefined) return;
  if (--t.open > 0) return;
  byDestination.delete(destination);
  if (byDestination.size === 0) tallies.delete(label);
}
export interface EgressRow {
  readonly label: string;
  readonly destination: string;
  readonly open: number;
  readonly bytesIn: number;
  readonly bytesOut: number;
}
/** Every destination an app holds connections to now, busiest first. */
export function egressSnapshot(): EgressRow[] {
  const rows: EgressRow[] = [];
  for (const [label, byDestination] of tallies)
    for (const [destination, t] of byDestination)
      rows.push({ label, destination, open: t.open, bytesIn: t.bytesIn, bytesOut: t.bytesOut });
  return rows.sort((a, b) => b.open - a.open || a.label.localeCompare(b.label));
}
/** The ports a declared host may be reached on when its entry names none. */
export const WEB_PORTS: readonly number[] = [80, 443];
@@ -165,8 +226,26 @@
interface Conn {
  toPeer: Uint8Array[];
  toClient: Uint8Array[];
  done: boolean;
  /** The destination's counts once dialled; the data path adds to it directly. */
  tally?: Tally;
  /** When bytes last moved either way (ms). */
  last: number;
  /** Frees the destination's slot and counts it closed; set once dialled. */
  release: (() => void) | undefined;
}
/** Cut both sides now: the upstream must never outlive the client. */
function drop(client: Socket<Conn>): void {
  const conn = client.data;
  conn.done = true;
  conn.release?.();
  conn.release = undefined;
  conn.peer?.terminate();
}
const queued = (queue: Uint8Array[]): number =>
  queue.reduce((n, c) => n + c.length, 0);
/** Write what the socket takes now; keep the rest for its `drain`. */
function send(
  socket: { write(data: Uint8Array): number },
@@ -295,31 +374,59 @@
async function open(
    log(`egress reported: ${host}:${port} is not in forge.json "egress" (let through while egress is only reported)`);
    policy.report?.(host.toLowerCase(), port);
  }
  const destination = `${host.toLowerCase()}:${port}`;
  const count = tally(policy.label ?? "", destination);
  const cap = policy.maxPerDestination ?? MAX_PER_DESTINATION;
  if (count.open >= cap) {
    log(`egress capped: ${policy.label ?? "app"} already has ${count.open} connections to ${destination}`);
    refuse(
      client,
      "429 Too Many Requests",
      `${destination} already has ${cap} open connections from this app`,
    );
    return;
  }
  // The slot is taken before the dial so a burst cannot all pass the check.
  count.open++;
  client.data.tally = count;
  const label = policy.label ?? "";
  client.data.release = () => untally(label, destination);
  const ended = () => {
    if (!client.data.done) {
      drop(client);
      client.end();
    }
  };
  try {
    const peer = await Bun.connect<undefined>({
      hostname: verdict.address,
      port,
      socket: {
        data(_peer, chunk) {
          if (!client.data.done) send(client, client.data.toClient, chunk);
          if (client.data.done) return;
          client.data.last = Date.now();
          count.bytesIn += chunk.length;
          send(client, client.data.toClient, chunk);
          if (queued(client.data.toClient) > (policy.queueMax ?? QUEUE_MAX)) {
            drop(client);
            client.terminate();
          }
        },
        drain(peer) {
          flush(peer, client.data.toPeer);
        },
        close() {
          if (!client.data.done) {
            client.data.done = true;
            client.end();
          }
        },
        error() {
          if (!client.data.done) {
            client.data.done = true;
            client.end();
          }
        },
        close: ended,
        error: ended,
      },
    });
    // The client went away while the dial was in flight: nobody is left to
    // take this connection, so it must not stay open.
    if (client.data.done) {
      
        sentinelcrew
        24d ago
        
      
      This branch is the leak the issue names: the client leaves while the dial is in flight. No test reaches it, because every leak test waits for the 200 first. Add one that drops the client before the upstream accepts and asserts the upstream closes and the tally is empty.
    
      peer.terminate();
      client.data.release?.();
      client.data.release = undefined;
      return;
    }
    client.data.peer = peer;
    if (tunnel)
      send(
@@ -330,67 +437,111 @@
async function open(
    // Everything after the head — including what arrived during the dial,
    // kept with it — goes out now; from here `data` forwards directly.
    const first = concat(prefix, client.data.head.subarray(end));
    if (first.length > 0) send(peer, client.data.toPeer, first);
    if (first.length > 0) {
      count.bytesOut += first.length;
      send(peer, client.data.toPeer, first);
    }
  } catch {
    client.data.release?.();
    client.data.release = undefined;
    refuse(client, "502 Bad Gateway", `${host}:${port} did not answer`);
  }
}
/** Listen at `path` and forward only what `policy` allows. */
/**
 * Listen at `path` and forward only what `policy` allows. `{ tcp }` (a
 * loopback port) exists for the tests, where a unix socket cannot be bound.
 */
export function startEgress(
  path: string,
  path: string | { tcp: number },
  policy: EgressPolicy,
  log: (line: string) => void = () => {},
): Egress {
  const server = Bun.listen<Conn>({
    unix: path,
    socket: {
      open(client) {
        client.data = {
          head: new Uint8Array(0),
          peer: undefined,
          toPeer: [],
          toClient: [],
          done: false,
        };
      },
      data(client, chunk) {
        const conn = client.data;
        if (conn.peer !== undefined) {
          send(conn.peer, conn.toPeer, chunk);
          return;
        }
        if (conn.done) return;
        const before = conn.head.length;
        conn.head = concat(conn.head, chunk);
        // Waiting on the upstream dial: whatever else arrives is kept with
        // the head, and goes out after it.
        if (before > 0 && headEnd(conn.head.subarray(0, before)) !== -1) return;
        if (headEnd(conn.head) === -1) {
          if (conn.head.length > HEAD_MAX)
            refuse(
              client,
              "431 Request Header Fields Too Large",
              "head too large",
            );
          return;
        }
        void open(client, policy, log);
      },
      drain(client) {
        flush(client, client.data.toClient);
      },
      close(client) {
        client.data.done = true;
        client.data.peer?.end();
      },
      error(client) {
        client.data.done = true;
        client.data.peer?.end();
      },
  const live = new Set<Socket<Conn>>();
  const handlers = {
    open(client: Socket<Conn>) {
      live.add(client);
      client.data = {
        head: new Uint8Array(0),
        peer: undefined,
        toPeer: [],
        toClient: [],
        done: false,
        last: Date.now(),
        release: undefined,
      };
    },
  });
  return { close: () => server.stop(true) };
    data(client: Socket<Conn>, chunk: Uint8Array) {
      const conn = client.data;
      conn.last = Date.now();
      if (conn.peer !== undefined) {
        if (conn.done) return;
        if (conn.tally !== undefined) conn.tally.bytesOut += chunk.length;
        send(conn.peer, conn.toPeer, chunk);
        return;
      }
      if (conn.done) return;
      const before = conn.head.length;
      conn.head = concat(conn.head, chunk);
      // Waiting on the upstream dial: whatever else arrives is kept with
      // the head, and goes out after it.
      if (before > 0 && headEnd(conn.head.subarray(0, before)) !== -1) return;
      if (headEnd(conn.head) === -1) {
        if (conn.head.length > HEAD_MAX)
          refuse(
            client,
            "431 Request Header Fields Too Large",
            "head too large",
          );
        return;
      }
      void open(client, policy, log);
    },
    drain(client: Socket<Conn>) {
      client.data.last = Date.now();
      flush(client, client.data.toClient);
    },
    close(client: Socket<Conn>) {
      live.delete(client);
      drop(client);
    },
    error(client: Socket<Conn>) {
      live.delete(client);
      drop(client);
    },
  };
  const server =
    typeof path === "string"
      ? Bun.listen<Conn>({ unix: path, socket: handlers })
      : Bun.listen<Conn>({ hostname: "127.0.0.1", port: path.tcp, socket: handlers });
  // A tunnel that moves nothing (a client that stopped reading, an upstream
  // that stopped sending) is cut, so it cannot hold an upstream open.
  const idleMs = policy.idleMs ?? IDLE_MS;
  const sweep = setInterval(
    () => {
      const now = Date.now();
      for (const client of live)
        if (now - client.data.last > idleMs) {
          log(`egress idle: closed a connection quiet for ${idleMs} ms`);
          drop(client);
          client.terminate();
          live.delete(client);
        }
    },
    Math.max(50, Math.min(5_000, idleMs / 2)),
  );
  sweep.unref();
  return {
    close() {
      clearInterval(sweep);
      for (const client of live) {
        drop(client);
        client.terminate();
      }
      live.clear();
      server.stop(true);
    },
  };
}
/**
@@ -472,11 +623,11 @@
export function relay(
    },
    close(client: Socket<Pair>) {
      client.data.done = true;
      client.data.peer?.end();
      client.data.peer?.terminate();
    },
    error(client: Socket<Pair>) {
      client.data.done = true;
      client.data.peer?.end();
      client.data.peer?.terminate();
    },
  };
  const server =
@@ -0,0 +1,26 @@
/**
 * The egress tally as Prometheus text (#3719): one line per app and
 * destination for open connections and the bytes they have moved, so a leaked
 * upstream socket shows as a gauge that never falls.
 */
import type { EgressRow } from "./egress.ts";
/** A label value, escaped as the exposition format asks. */
function label(value: string): string {
  return value.replace(/\\/g, "\\\\").replace(/"/g, '\\"').replace(/\n/g, "\\n");
}
export function egressMetrics(rows: readonly EgressRow[]): string {
  const series: [string, string, (r: EgressRow) => number][] = [
    ["forge_egress_connections", "Open egress connections.", (r) => r.open],
    ["forge_egress_bytes_in", "Bytes received from the destination over open connections.", (r) => r.bytesIn],
    ["forge_egress_bytes_out", "Bytes sent to the destination over open connections.", (r) => r.bytesOut],
  ];
  const lines: string[] = [];
  for (const [name, help, value] of series) {
    lines.push(`# HELP ${name} ${help}`, `# TYPE ${name} gauge`);
    for (const r of rows)
      lines.push(`${name}{repo="${label(r.label)}",destination="${label(r.destination)}"} ${value(r)}`);
  }
  return `${lines.join("\n")}\n`;
}
@@ -427,6 +427,8 @@
import { passportRoutes } from "./passport_edge.ts";
import type { Router } from "./router.ts";
import { clientAddress, clientBucket } from "./peer.ts";
import { manifestFor } from "../actions/here.ts";
import { egressSnapshot } from "../apps/egress.ts";
import { egressMetrics } from "../apps/egress_metrics.ts";
import {
  acceptReport,
  dismissReport,
@@ -4821,6 +4823,20 @@
export function apiRoutes(router: Router<Handler>): void {
    ),
  );
  // Open egress connections per app and destination, as Prometheus text
  // (#3719). Repository names are in the labels, so a site admin only.
  router.add("GET", "/metrics", (ctx) => {
      
        sentinelcrew
        24d ago
        
      
      High: this admin-gated route has no test. Removing accountAdmin would leak every app's repository name and destinations, and no test would fail. Test it for anonymous, non-admin, agent-without-scope and admin.
    
    const denied = accountAdmin(ctx, "ro");
    if (denied) return denied;
    return new Response(egressMetrics(egressSnapshot()), {
      headers: {
        "content-type": "text/plain; version=0.0.4; charset=utf-8",
        "cache-control": "no-store",
        vary: "authorization, cookie",
      },
    });
  });
  // Explore (#607): the rows /explore shows, for the same one line.
  router.add("GET", "/api/v1/explore", async (ctx) => {
    const query = parseExploreQuery(ctx.url.searchParams.get("q") ?? "");
@@ -2425,7 +2425,8 @@
async function answerGet(
  // generic as `/:owner/:name` can never claim it — a route added there
  // next year still can't bring the HTML 404 back for an API caller.
  const isApiPath = url.pathname === "/api" || url.pathname.startsWith("/api/");
  const matched = isApiPath
  // `/metrics` is the API router's too: a scraper, not a page.
  const matched = isApiPath || url.pathname === "/metrics"
    ? apiRouter.match(routeMethod, url.pathname)
    : router.match(routeMethod, url.pathname);
  if (!matched) {
@@ -32,7 +32,7 @@
import { truncate, wrap } from "../term/width.ts";
import { badge, card } from "../ui/widgets.ts";
import type { Session } from "../session.ts";
import { PIN_MARK, ago, drawFacts, plural } from "./common.ts";
import { egressScreen } from "./egress.ts";
import { connectionsScreen, egressScreen } from "./egress.ts";
import { forkOf, hostOf, liveOrigin, localRepo, sourceName } from "./where.ts";
/**
@@ -98,7 +98,7 @@
export function aboutScreen(
          : []),
        // A site admin also reaches the repository's private egress grants.
        ...(f.user?.isAdmin === true && repo !== undefined
          ? [binding(["e"], "egress")]
          ? [binding(["e"], "egress"), binding(["c"], "connections")]
          : []),
        binding(["esc"], "close"),
      ];
@@ -209,6 +209,10 @@
export function aboutScreen(
        ctx.push(egressScreen(f, ctx, repo));
        return true;
      }
      if (e.key === "c" && f.user?.isAdmin === true) {
        ctx.push(connectionsScreen(f, ctx));
        return true;
      }
      if (e.key === "esc" || e.key === "q" || e.key === "I") {
        ctx.pop();
        return true;
@@ -10,6 +10,8 @@
import {
  revokeAppEgress,
  type EgressRow,
} from "../../actions/egress_grants.ts";
import { egressSnapshot } from "../../apps/egress.ts";
import { bytes } from "../../system/format.ts";
import type { Repo } from "../../core/repos.ts";
import type { Ctx, Screen } from "../app/app.ts";
import { binding } from "../app/keys.ts";
@@ -74,3 +76,38 @@
export function egressScreen(f: Session, ctx: Ctx, repo: Repo): Screen {
  };
  return screen;
}
/**
 * The egress connections open right now, per app and destination: the
 * Capacity page's card, for a site admin. Read again on every draw, so it
 * moves as connections do; anyone else is offered nothing.
 */
export function connectionsScreen(f: Session, ctx: Ctx): Screen {
  const state = new ListState();
  const lines = (): ListRow[] =>
    f.user?.isAdmin !== true
      ? []
      : egressSnapshot().map((r) => ({
          title: r.label === "" ? "(unlabelled)" : r.label,
          meta: `${r.destination} · ${r.open} open · ${bytes(r.bytesIn)} in · ${bytes(r.bytesOut)} out`,
        }));
  return {
    title: "Egress connections",
    keys: [binding(["j", "k"], "move cursor")],
    render(area, s) {
      const c = ctx.theme.c;
      const rows = lines();
      state.clamp(rows.length);
      const [head, list] = split(area, "v", [len(1), fill(1)]);
      const note =
        rows.length === 0
          ? "No app holds an egress connection open."
          : `${rows.length} destination${rows.length === 1 ? "" : "s"} in use`;
      s.text(head!.x, head!.y, truncate(note, head!.w), style(c.mutedFg, c.bg), head!.x + head!.w);
      drawList(s, list!, c, state, rows);
    },
    key(e) {
      return state.key(e, lines().length);
    },
  };
}
@@ -8,6 +8,8 @@
import { changeCapacity, listCapacity, type CapacityRow } from "../../actions/capacity.ts";
import { CAPACITY, type CapacityKey } from "../../core/capacity.ts";
import { bytes } from "../../system/format.ts";
import { egressSnapshot } from "../../apps/egress.ts";
import type { Ctx } from "../../http/context.ts";
import { notePhase, takeSessionTime, timePhase } from "../../core/route_timing.ts";
import { notFoundPage, signInFirst } from "../chrome.ts";
@@ -155,7 +157,32 @@
export function capacityPage(ctx: Ctx): Response {
      pad: false,
    },
  );
  const body = html`${happened}${when(
  const open = egressSnapshot();
  const connections = card(
    open.length === 0
      ? html`<div class="card-body">
          <p class="muted small" id="egress-open">
            No app holds an egress connection open.
          </p>
        </div>`
      : html`<ul class="list" id="egress-open">
          ${open.map((r) =>
            row({
              lead: icon("activity"),
              title: r.label === "" ? "(unlabelled)" : r.label,
              meta: html`<span class="mono">${r.destination}</span> ·
                ${bytes(r.bytesIn)} in · ${bytes(r.bytesOut)} out`,
              trail: html`<span class="mono">${r.open} open</span>`,
            }),
          )}
        </ul>`,
    {
      head: html`${icon("activity")}
        <h2>Egress connections</h2>`,
      pad: false,
    },
  );
  const body = html`${happened}${connections}${when(
      reported,
      () => html`<p class="muted small mb" id="egress-reported">
        ${icon("activity")} Egress is reported, not enforced: every new app,
@@ -18,6 +18,7 @@
import { addToCrew, createAgent } from "../src/actions/crew.ts";
import { assignIssue, openIssue } from "../src/actions/issues.ts";
import { commitToBranch, createRepo } from "../src/actions/repos.ts";
import { enableApp } from "../src/actions/apps.ts";
import { startEgress } from "../src/apps/egress.ts";
import { AppsRuntime } from "../src/apps/runtime.ts";
import { startChecks, type ChecksHandle } from "../src/checks/service.ts";
import {
@@ -384,6 +385,39 @@
describe("report mode is loud to admins, and to nobody else", () => {
    expect(shown.replace(/\s+/g, " ")).toContain("may reach any public host on ports 80 and 443");
  });
  test("the page shows the egress connections an app holds open, and none once they close", async () => {
    const camera = Bun.listen<undefined>({ hostname: "127.0.0.1", port: 0, socket: { data() {} } });
    const front = Bun.listen({ hostname: "127.0.0.1", port: 0, socket: { data() {} } });
    const listen = front.port;
    front.stop(true);
    const egress = startEgress(
      { tcp: listen },
      { hosts: [], origins: [{ host: "127.0.0.1", port: camera.port }], label: "app:cam/main" },
    );
    try {
      let reply = "";
      const client = await Bun.connect<undefined>({
        hostname: "127.0.0.1",
        port: listen,
        socket: { data: (_s, c) => void (reply += new TextDecoder().decode(c)) },
      });
      client.write(`CONNECT 127.0.0.1:${camera.port} HTTP/1.1\r\n\r\n`);
      while (!reply.includes("200")) await Bun.sleep(10);
      const shown = (await page()).replace(/\s+/g, " ");
      expect(shown).toContain("app:cam/main");
      expect(shown).toContain(`127.0.0.1:${camera.port}`);
      expect(shown).toContain("1 open");
      expect(shown).toMatch(/\d+ B in · \d+ B out/);
      expect(shown).not.toContain("→ 127.0.0.1");
      egress.close();
      expect(await page()).toContain("No app holds an egress connection open.");
      client.terminate();
    } finally {
      egress.close();
      camera.stop(true);
    }
  });
  test("reported: the public health check, a member and a stranger hear nothing of it", async () => {
    const health = await fetch(`${servers.url}/healthz`);
    expect(health.status).toBe(200);
@@ -13,6 +13,7 @@
import { commitToBranch, createRepo } from "../src/actions/repos.ts";
import { listEgressGrants } from "../src/core/egress_grants.ts";
import type { Repo } from "../src/core/repos.ts";
import { createUser, type User } from "../src/core/users.ts";
import { startEgress } from "../src/apps/egress.ts";
import { createForgeApp } from "../src/tui/forge.ts";
import { headlessTerminal } from "../src/tui/term/terminal.ts";
import { quietLogs } from "./helpers/globals.ts";
@@ -107,4 +108,41 @@
describe("egress grants in the terminal", () => {
      h.tui.stop();
    }
  });
  test("an admin sees the open egress connections from About; a non-admin does not", async () => {
    const camera = Bun.listen<undefined>({ hostname: "127.0.0.1", port: 0, socket: { data() {} } });
    const probe = Bun.listen({ hostname: "127.0.0.1", port: 0, socket: { data() {} } });
    const listen = probe.port;
    probe.stop(true);
    const egress = startEgress(
      { tcp: listen },
      { hosts: [], origins: [{ host: "127.0.0.1", port: camera.port }], label: "app:cam/main" },
    );
    let reply = "";
    const client = await Bun.connect<undefined>({
      hostname: "127.0.0.1",
      port: listen,
      socket: { data: (_s, c) => void (reply += new TextDecoder().decode(c)) },
    });
    client.write(`CONNECT 127.0.0.1:${camera.port} HTTP/1.1\r\n\r\n`);
    while (!reply.includes("200")) await Bun.sleep(10);
    const admin = boot(ada);
    const other = boot(bob);
    try {
      await admin.openAbout();
      await admin.press("c");
      await settleUntil(admin.tui, "app:cam/main");
      expect(admin.text()).toContain(`127.0.0.1:${camera.port} · 1 open`);
      await other.openAbout();
      await other.press("c");
      expect(other.text()).not.toContain("Egress connections");
      expect(other.text()).not.toContain("app:cam/main");
    } finally {
      admin.tui.stop();
      other.tui.stop();
      client.terminate();
      egress.close();
      camera.stop(true);
    }
  });
});
@@ -0,0 +1,231 @@
/**
 * The egress proxy never leaves an upstream socket open after the cell side
 * is gone (#3719): a client that drops, a proxy that closes, a client that
 * stops reading, and a cap on connections to one destination.
 */
import { afterEach, describe, expect, test } from "bun:test";
import type { Socket } from "bun";
import { egressSnapshot, relay, startEgress } from "../src/apps/egress.ts";
const openNow = (label: string, destination?: string): number =>
  egressSnapshot()
    .filter((r) => r.label === label && (destination === undefined || r.destination === destination))
    .reduce((n, r) => n + r.open, 0);
const pending: (() => void)[] = [];
afterEach(() => {
  while (pending.length > 0) pending.pop()?.();
});
/** A fake MJPEG camera: streams forever, counts live connections. */
function camera(frame = 1024) {
  const open = new Set<Socket<undefined>>();
  // Half-open allowed: like a real camera, it keeps streaming after the
  // other side has said it is done writing, until the socket is destroyed.
  const server = Bun.listen<undefined>({
    hostname: "127.0.0.1",
    port: 0,
    allowHalfOpen: true,
    socket: {
      open(s) {
        open.add(s);
        s.data = undefined;
      },
      data(s) {
        const timer = setInterval(() => {
          if (!open.has(s)) return clearInterval(timer);
          s.write("--frame\r\n" + "x".repeat(frame) + "\r\n");
        }, 5);
      },
      close(s) {
        open.delete(s);
      },
      error(s) {
        open.delete(s);
      },
    },
  });
  pending.push(() => server.stop(true));
  return { port: server.port, count: () => open.size };
}
async function until(check: () => boolean, ms = 2_000): Promise<boolean> {
  const end = Date.now() + ms;
  while (Date.now() < end) {
    if (check()) return true;
    await Bun.sleep(20);
  }
  return check();
}
describe("egress leaves nothing open", () => {
  // Bun.listen with port 0 gives no handle back through `startEgress`, so
  // these tests listen on a fixed free port found first.
  async function freePort(): Promise<number> {
    const s = Bun.listen({ hostname: "127.0.0.1", port: 0, socket: { data() {} } });
    const p = s.port;
    s.stop(true);
    return p;
  }
  async function setup(extra: Record<string, unknown> = {}, frame = 1024, log?: (line: string) => void) {
    const cam = camera(frame);
    const listen = await freePort();
    const egress = startEgress(
      { tcp: listen },
      {
        hosts: [],
        origins: [{ host: "127.0.0.1", port: cam.port }],
        label: `t${listen}`,
        ...extra,
      },
      log,
    );
    pending.push(() => egress.close());
    const stream = async (request = true, hold = false) => {
      const got = Promise.withResolvers<string>();
      let text = "";
      const sock = await Bun.connect<undefined>({
        hostname: "127.0.0.1",
        port: listen,
        socket: {
          data(_s, chunk) {
            text += new TextDecoder().decode(chunk);
            if (text.includes("200 Connection Established")) got.resolve(text);
          },
          close() {},
          error() {},
        },
      });
      sock.write(`CONNECT 127.0.0.1:${cam.port} HTTP/1.1\r\n\r\n${request ? "GET /\r\n\r\n" : ""}`);
      await got.promise;
      if (hold) sock.pause();
      return sock;
    };
    return { cam, egress, stream, listen, label: `t${listen}`, dest: `127.0.0.1:${cam.port}` };
  }
  test("a client that disconnects mid-stream closes its upstream", async () => {
    const { cam, stream } = await setup();
    const sock = await stream();
    expect(await until(() => cam.count() === 1)).toBe(true);
    sock.terminate();
    expect(await until(() => cam.count() === 0)).toBe(true);
  });
  test("closing the proxy (the cell stops) leaves 0 of 20 streams open", async () => {
    const { cam, egress, stream, label } = await setup({ maxPerDestination: 20 });
    const socks = await Promise.all(Array.from({ length: 20 }, () => stream()));
    expect(await until(() => cam.count() === 20)).toBe(true);
    expect(openNow(label)).toBe(20);
    egress.close();
    expect(await until(() => cam.count() === 0)).toBe(true);
    expect(openNow(label)).toBe(0);
    for (const s of socks) s.terminate();
  });
  test("clients that drop right after connecting leave no upstream behind", async () => {
    const { cam, stream } = await setup({ maxPerDestination: 10 });
    const socks = await Promise.all(Array.from({ length: 10 }, () => stream()));
    for (const s of socks) s.terminate();
    expect(await until(() => cam.count() === 0, 3_000)).toBe(true);
  });
  test("connections past the per-destination cap are refused, and counted", async () => {
    const lines: string[] = [];
    const { cam, stream, listen, label, dest } = await setup({ maxPerDestination: 2 }, 1024, (l) => lines.push(l));
    const a = await stream();
    const b = await stream();
    expect(await until(() => cam.count() === 2)).toBe(true);
    let reply = "";
    const third = await Bun.connect<undefined>({
      hostname: "127.0.0.1",
      port: listen,
      socket: {
        data(_s, chunk) {
          reply += new TextDecoder().decode(chunk);
        },
      },
    });
    third.write(`CONNECT 127.0.0.1:${cam.port} HTTP/1.1\r\n\r\n`);
    expect(await until(() => reply.startsWith("HTTP/1.1 429"))).toBe(true);
    expect(lines.some((l) => l.startsWith("egress capped:") && l.includes(dest))).toBe(true);
    expect(cam.count()).toBe(2);
    expect(openNow(label, dest)).toBe(2);
    const row = egressSnapshot().find((r) => r.label === label && r.destination === dest);
    expect(row?.open).toBe(2);
    expect(row?.bytesIn).toBeGreaterThan(0);
    a.terminate();
    b.terminate();
    expect(await until(() => openNow(label, dest) === 0)).toBe(true);
    // Nothing of a closed app stays in the tally.
    expect(egressSnapshot().some((r) => r.label === label)).toBe(false);
  });
  test("a connection that moves no bytes is cut after the idle timeout", async () => {
    const { cam, stream } = await setup({ idleMs: 150 });
    const sock = await stream(false);
    expect(await until(() => cam.count() === 1)).toBe(true);
    expect(await until(() => cam.count() === 0)).toBe(true);
    sock.terminate();
  });
  test("a client that stops reading is cut once its backlog passes the limit", async () => {
    const { cam, stream } = await setup({ queueMax: 64 * 1024 }, 32 * 1024);
    const sock = await stream(true, true);
    expect(await until(() => cam.count() === 0, 5_000)).toBe(true);
    sock.terminate();
  });
});
describe("the inbound and loopback relay closes its target with the client", () => {
  test("a client that leaves mid-stream closes the connection it was relayed to", async () => {
    const cam = camera();
    const find = Bun.listen({ hostname: "127.0.0.1", port: 0, socket: { data() {} } });
    const port = find.port;
    find.stop(true);
    const r = relay({ tcp: port }, { tcp: cam.port });
    pending.push(() => r.close());
    let seen = 0;
    const client = await Bun.connect<undefined>({
      hostname: "127.0.0.1",
      port,
      socket: { data: (_s, c) => void (seen += c.length) },
    });
    client.write("go");
    expect(await until(() => cam.count() === 1 && seen > 0)).toBe(true);
    client.terminate();
    expect(await until(() => cam.count() === 0)).toBe(true);
  });
});
describe("a client that leaves while the dial is in flight", () => {
  test("strands no upstream and leaves nothing in the tally", async () => {
    const cam = camera();
    const probe = Bun.listen({ hostname: "127.0.0.1", port: 0, socket: { data() {} } });
    const listen = probe.port;
    probe.stop(true);
    const label = `dial${listen}`;
    const egress = startEgress(
      { tcp: listen },
      { hosts: [], origins: [{ host: "127.0.0.1", port: cam.port }], label, maxPerDestination: 1000 },
    );
    pending.push(() => egress.close());
    // The dial is a few ticks long: leave at each of several moments inside it.
    const socks: Socket<undefined>[] = [];
    for (let i = 0; i < 60; i++) {
      const sock = await Bun.connect<undefined>({
        hostname: "127.0.0.1",
        port: listen,
        socket: { data() {}, close() {}, error() {} },
      });
      sock.write(`CONNECT 127.0.0.1:${cam.port} HTTP/1.1\r\n\r\n`);
      socks.push(sock);
      if (i % 3 > 0) await Bun.sleep(i % 3 - 1);
      sock.terminate();
    }
    expect(await until(() => cam.count() === 0, 3_000)).toBe(true);
    expect(await until(() => !egressSnapshot().some((r) => r.label === label), 3_000)).toBe(true);
  });
});
@@ -0,0 +1,107 @@
/**
 * `GET /metrics` serves the open egress connections with repository names in
 * its labels, so it is a site admin's alone (#3719).
 */
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 { createAgent } from "../src/actions/crew.ts";
import { startEgress, type Egress } from "../src/apps/egress.ts";
import { createToken } from "../src/core/tokens.ts";
import { createUser, type User } from "../src/core/users.ts";
import { serve, type Servers } from "../src/http/server.ts";
let dir: string;
let app: App;
let servers: Servers;
let admin: string;
let adminRo: string;
let member: string;
let agent: string;
let memberCookie: string;
let egress: Egress;
let camera: { port: number; stop(close?: boolean): void };
let client: Awaited<ReturnType<typeof Bun.connect>>;
beforeAll(async () => {
  dir = await mkdtemp(join(tmpdir(), "egress-metrics-"));
  app = createApp({ dataDir: dir, port: 0 });
  servers = await serve(app);
  const mk = async (username: string, adminIfFirst = false): Promise<User> => {
    const made = await createUser(app.db, {
      origin: "seed",
      trusted: true,
      username,
      email: `${username}@forge.local`,
      password: "a-good-passphrase",
      adminIfFirst,
    });
    if (!made.ok) throw new Error(made.error);
    return made.value;
  };
  const alice = await mk("alice", true);
  const bob = await mk("bob");
  const made = await createAgent(app, alice, { username: "scout" });
  if (!made.ok) throw new Error(String(made.error));
  admin = createToken(app.db, alice.id, "t", "admin").token;
  adminRo = createToken(app.db, alice.id, "t", "ro").token;
  member = createToken(app.db, bob.id, "t", "ro").token;
  agent = createToken(app.db, made.value.user.id, "t", "ro").token;
  const res = await fetch(`${servers.url}/signin`, {
    method: "POST",
    redirect: "manual",
    headers: { "content-type": "application/x-www-form-urlencoded", origin: servers.url },
    body: new URLSearchParams({ username: "bob", password: "a-good-passphrase" }),
  });
  memberCookie = (res.headers.get("set-cookie") ?? "").split(";")[0] ?? "";
  camera = Bun.listen<undefined>({ hostname: "127.0.0.1", port: 0, socket: { data() {} } });
  const probe = Bun.listen({ hostname: "127.0.0.1", port: 0, socket: { data() {} } });
  const listen = probe.port;
  probe.stop(true);
  egress = startEgress(
    { tcp: listen },
    { hosts: [], origins: [{ host: "127.0.0.1", port: camera.port }], label: "app:cam/main" },
  );
  let reply = "";
  client = await Bun.connect<undefined>({
    hostname: "127.0.0.1",
    port: listen,
    socket: { data: (_s, c) => void (reply += new TextDecoder().decode(c)) },
  });
  client.write(`CONNECT 127.0.0.1:${camera.port} HTTP/1.1\r\n\r\n`);
  while (!reply.includes("200")) await Bun.sleep(10);
}, 60_000);
afterAll(async () => {
  client.terminate();
  egress.close();
  camera.stop(true);
  servers.stop();
  app.close();
  await rm(dir, { recursive: true, force: true });
});
const get = (headers: Record<string, string> = {}) =>
  fetch(`${servers.url}/metrics`, { headers });
test("anonymous, a member and an agent are told it is not there", async () => {
  expect((await get()).status).toBe(404);
  expect((await get({ cookie: memberCookie })).status).toBe(404);
  expect((await get({ authorization: `Bearer ${member}` })).status).toBe(404);
  expect((await get({ authorization: `Bearer ${agent}` })).status).toBe(404);
});
test("a site admin reads the open connections as Prometheus text", async () => {
  const res = await get({ authorization: `Bearer ${adminRo}` });
  expect(res.status).toBe(200);
  expect(res.headers.get("content-type")).toStartWith("text/plain; version=0.0.4");
  expect(res.headers.get("cache-control")).toBe("no-store");
  expect(await res.text()).toContain(
    `forge_egress_connections{repo="app:cam/main",destination="127.0.0.1:${camera.port}"} 1`,
  );
  expect((await get({ authorization: `Bearer ${admin}` })).status).toBe(200);
});
@@ -0,0 +1,15 @@
import { expect, test } from "bun:test";
import { egressMetrics } from "../src/apps/egress_metrics.ts";
test("every destination gets its open count and bytes, labels escaped", () => {
  const text = egressMetrics([
    { label: 'a/"b"', destination: "pi:8080", open: 3, bytesIn: 10, bytesOut: 2 },
  ]);
  expect(text).toContain('forge_egress_connections{repo="a/\\"b\\"",destination="pi:8080"} 3');
  expect(text).toContain('forge_egress_bytes_in{repo="a/\\"b\\"",destination="pi:8080"} 10');
  expect(text).toContain('forge_egress_bytes_out{repo="a/\\"b\\"",destination="pi:8080"} 2');
});
test("no connections leaves only the headers", () => {
  expect(egressMetrics([])).not.toContain("{");
});
		@@ -846,6 +846,15 @@ file), and if you named an agent, the sentence became issue #1, assigned.
846	846	names it exactly and a site admin has granted it: on the
847	847	app's settings page, or in the terminal (About `I`, then `e`, `g`
848	848	to grant, `x` to revoke). See `docs/APPS.md`.
	849	- Egress connections are bounded. The proxy closes the upstream socket
	850	the moment the cell side closes, errors or stops, and the other way round.
	851	An app holds at most 8 open connections to one destination (the ninth
	852	gets a `429`, logged), and a connection that moves no bytes, either way, for 5 min is
	853	cut, so a tunnel quiet for longer than that (a websocket or SSE stream with
	854	very long gaps) needs a keepalive. Open connections per app and
	855	destination, with the bytes moved, are on the Capacity page
	856	(`/settings/capacity`, admins) and in the terminal: About `I`, then `c`
	857	(site admins).
849	858	- Production is the default branch. The only way code reaches it is that
850	859	branch moving, and an agent cannot move it: the crew proposes through
851	860	pulls, a person merges.
		@@ -1914,3 +1923,8 @@ same rule. A guard installed by hand without the prefix answers `refused`
1914	1923	until it is re-created with `forge: ` in front of its message.
1915	1924
1916	1925	`reach` is a boards source for an admin only (anyone else gets 403, an agent too): `visitors` per day, `lcp_p75` and `inp_p75` per route per day, `server_p50` and `server_p95` per route per day (the beacon's time to first byte, the only per-route server timing the forge keeps), and `accounts_active_1d`, `accounts_active_7d` and `accounts_total` as of now. Filter by `metric`, group by `route` or a time bucket. Percentiles are nearest-rank within one route and day, so a `max` or `avg` over several days is a summary of daily percentiles, not a percentile of the period. Account activity comes from a session's last-seen time, so it is a count as of now, not a history. No dependency was added.
	1926
	1927	`GET /metrics` (site admin only; repository names are in the labels) serves
	1928	`forge_egress_connections`, `forge_egress_bytes_in` and `forge_egress_bytes_out`
	1929	per `repo` and `destination` as Prometheus text, so a leaked upstream socket
	1930	shows as a gauge that never falls (#3719).
		@@ -362,6 +362,7 @@ export function openCell(
362	362	{
363	363	hosts: spec.egress,
364	364	granted: spec.granted,
	365	label: spec.label,
365	366	origins,
366	367	report:
367	368	(spec.egressMode ?? egressMode()) === "report"
		@@ -38,6 +38,14 @@ export interface EgressPolicy {
38	38	* refused. Private addresses are refused either way.
39	39	*/
40	40	readonly report?: ((host: string, port: number) => void) \| undefined;
	41	/** Whose proxy this is (`app:<slug>/<env>`), for the connection gauges. */
	42	readonly label?: string \| undefined;
	43	/** Most connections open to one destination at once; default 8. */
	44	readonly maxPerDestination?: number \| undefined;
	45	/** Close a connection that moved no bytes for this long; default 5 min. */
	46	readonly idleMs?: number \| undefined;
	47	/** Bytes held for a client that does not read before it is cut; default 4 MiB. */
	48	readonly queueMax?: number \| undefined;
41	49	}
42	50
43	51	export interface Egress {
		@@ -45,6 +53,59 @@ export interface Egress {
45	53	}
46	54
47	55	const HEAD_MAX = 16 * 1024;
	56	const MAX_PER_DESTINATION = 8;
	57	const IDLE_MS = 5 * 60_000;
	58	/** Bytes held for a client that does not read before its upstream is cut. */
	59	const QUEUE_MAX = 4 * 1024 * 1024;
	60
	61	interface Tally {
	62	open: number;
	63	bytesIn: number;
	64	bytesOut: number;
	65	}
	66
	67	/**
	68	* label -> `host:port` -> counts, for the connections open right now. A
	69	* destination's entry goes when its last connection closes (and the label
	70	* with its last destination), so it only ever holds what is live.
	71	*/
	72	const tallies = new Map<string, Map<string, Tally>>();
	73
	74	function tally(label: string, destination: string): Tally {
	75	let byDestination = tallies.get(label);
	76	if (byDestination === undefined) tallies.set(label, (byDestination = new Map()));
	77	let t = byDestination.get(destination);
	78	if (t === undefined)
	79	byDestination.set(destination, (t = { open: 0, bytesIn: 0, bytesOut: 0 }));
	80	return t;
	81	}
	82
	83	/** One connection closed: drop its destination, then its label, when empty. */
	84	function untally(label: string, destination: string): void {
	85	const byDestination = tallies.get(label);
	86	const t = byDestination?.get(destination);
	87	if (byDestination === undefined \|\| t === undefined) return;
	88	if (--t.open > 0) return;
	89	byDestination.delete(destination);
	90	if (byDestination.size === 0) tallies.delete(label);
	91	}
	92
	93	export interface EgressRow {
	94	readonly label: string;
	95	readonly destination: string;
	96	readonly open: number;
	97	readonly bytesIn: number;
	98	readonly bytesOut: number;
	99	}
	100
	101	/** Every destination an app holds connections to now, busiest first. */
	102	export function egressSnapshot(): EgressRow[] {
	103	const rows: EgressRow[] = [];
	104	for (const [label, byDestination] of tallies)
	105	for (const [destination, t] of byDestination)
	106	rows.push({ label, destination, open: t.open, bytesIn: t.bytesIn, bytesOut: t.bytesOut });
	107	return rows.sort((a, b) => b.open - a.open \|\| a.label.localeCompare(b.label));
	108	}
48	109
49	110	/** The ports a declared host may be reached on when its entry names none. */
50	111	export const WEB_PORTS: readonly number[] = [80, 443];
		@@ -165,8 +226,26 @@ interface Conn {
165	226	toPeer: Uint8Array[];
166	227	toClient: Uint8Array[];
167	228	done: boolean;
	229	/** The destination's counts once dialled; the data path adds to it directly. */
	230	tally?: Tally;
	231	/** When bytes last moved either way (ms). */
	232	last: number;
	233	/** Frees the destination's slot and counts it closed; set once dialled. */
	234	release: (() => void) \| undefined;
168	235	}
169	236
	237	/** Cut both sides now: the upstream must never outlive the client. */
	238	function drop(client: Socket<Conn>): void {
	239	const conn = client.data;
	240	conn.done = true;
	241	conn.release?.();
	242	conn.release = undefined;
	243	conn.peer?.terminate();
	244	}
	245
	246	const queued = (queue: Uint8Array[]): number =>
	247	queue.reduce((n, c) => n + c.length, 0);
	248
170	249	/** Write what the socket takes now; keep the rest for its `drain`. */
171	250	function send(
172	251	socket: { write(data: Uint8Array): number },
		@@ -295,31 +374,59 @@ async function open(
295	374	log(`egress reported: ${host}:${port} is not in forge.json "egress" (let through while egress is only reported)`);
296	375	policy.report?.(host.toLowerCase(), port);
297	376	}
	377	const destination = `${host.toLowerCase()}:${port}`;
	378	const count = tally(policy.label ?? "", destination);
	379	const cap = policy.maxPerDestination ?? MAX_PER_DESTINATION;
	380	if (count.open >= cap) {
	381	log(`egress capped: ${policy.label ?? "app"} already has ${count.open} connections to ${destination}`);
	382	refuse(
	383	client,
	384	"429 Too Many Requests",
	385	`${destination} already has ${cap} open connections from this app`,
	386	);
	387	return;
	388	}
	389	// The slot is taken before the dial so a burst cannot all pass the check.
	390	count.open++;
	391	client.data.tally = count;
	392	const label = policy.label ?? "";
	393	client.data.release = () => untally(label, destination);
	394	const ended = () => {
	395	if (!client.data.done) {
	396	drop(client);
	397	client.end();
	398	}
	399	};
298	400	try {
299	401	const peer = await Bun.connect<undefined>({
300	402	hostname: verdict.address,
301	403	port,
302	404	socket: {
303	405	data(_peer, chunk) {
304		if (!client.data.done) send(client, client.data.toClient, chunk);
	406	if (client.data.done) return;
	407	client.data.last = Date.now();
	408	count.bytesIn += chunk.length;
	409	send(client, client.data.toClient, chunk);
	410	if (queued(client.data.toClient) > (policy.queueMax ?? QUEUE_MAX)) {
	411	drop(client);
	412	client.terminate();
	413	}
305	414	},
306	415	drain(peer) {
307	416	flush(peer, client.data.toPeer);
308	417	},
309		close() {
310		if (!client.data.done) {
311		client.data.done = true;
312		client.end();
313		}
314		},
315		error() {
316		if (!client.data.done) {
317		client.data.done = true;
318		client.end();
319		}
320		},
	418	close: ended,
	419	error: ended,
321	420	},
322	421	});
	422	// The client went away while the dial was in flight: nobody is left to
	423	// take this connection, so it must not stay open.
	424	if (client.data.done) {
		sentinelcrew 24d ago This branch is the leak the issue names: the client leaves while the dial is in flight. No test reaches it, because every leak test waits for the 200 first. Add one that drops the client before the upstream accepts and asserts the upstream closes and the tally is empty.
	425	peer.terminate();
	426	client.data.release?.();
	427	client.data.release = undefined;
	428	return;
	429	}
323	430	client.data.peer = peer;
324	431	if (tunnel)
325	432	send(
		@@ -330,67 +437,111 @@ async function open(
330	437	// Everything after the head — including what arrived during the dial,
331	438	// kept with it — goes out now; from here `data` forwards directly.
332	439	const first = concat(prefix, client.data.head.subarray(end));
333		if (first.length > 0) send(peer, client.data.toPeer, first);
	440	if (first.length > 0) {
	441	count.bytesOut += first.length;
	442	send(peer, client.data.toPeer, first);
	443	}
334	444	} catch {
	445	client.data.release?.();
	446	client.data.release = undefined;
335	447	refuse(client, "502 Bad Gateway", `${host}:${port} did not answer`);
336	448	}
337	449	}
338	450
339		/** Listen at `path` and forward only what `policy` allows. */
	451	/**
	452	* Listen at `path` and forward only what `policy` allows. `{ tcp }` (a
	453	* loopback port) exists for the tests, where a unix socket cannot be bound.
	454	*/
340	455	export function startEgress(
341		path: string,
	456	path: string \| { tcp: number },
342	457	policy: EgressPolicy,
343	458	log: (line: string) => void = () => {},
344	459	): Egress {
345		const server = Bun.listen<Conn>({
346		unix: path,
347		socket: {
348		open(client) {
349		client.data = {
350		head: new Uint8Array(0),
351		peer: undefined,
352		toPeer: [],
353		toClient: [],
354		done: false,
355		};
356		},
357		data(client, chunk) {
358		const conn = client.data;
359		if (conn.peer !== undefined) {
360		send(conn.peer, conn.toPeer, chunk);
361		return;
362		}
363		if (conn.done) return;
364		const before = conn.head.length;
365		conn.head = concat(conn.head, chunk);
366		// Waiting on the upstream dial: whatever else arrives is kept with
367		// the head, and goes out after it.
368		if (before > 0 && headEnd(conn.head.subarray(0, before)) !== -1) return;
369		if (headEnd(conn.head) === -1) {
370		if (conn.head.length > HEAD_MAX)
371		refuse(
372		client,
373		"431 Request Header Fields Too Large",
374		"head too large",
375		);
376		return;
377		}
378		void open(client, policy, log);
379		},
380		drain(client) {
381		flush(client, client.data.toClient);
382		},
383		close(client) {
384		client.data.done = true;
385		client.data.peer?.end();
386		},
387		error(client) {
388		client.data.done = true;
389		client.data.peer?.end();
390		},
	460	const live = new Set<Socket<Conn>>();
	461	const handlers = {
	462	open(client: Socket<Conn>) {
	463	live.add(client);
	464	client.data = {
	465	head: new Uint8Array(0),
	466	peer: undefined,
	467	toPeer: [],
	468	toClient: [],
	469	done: false,
	470	last: Date.now(),
	471	release: undefined,
	472	};
391	473	},
392		});
393		return { close: () => server.stop(true) };
	474	data(client: Socket<Conn>, chunk: Uint8Array) {
	475	const conn = client.data;
	476	conn.last = Date.now();
	477	if (conn.peer !== undefined) {
	478	if (conn.done) return;
	479	if (conn.tally !== undefined) conn.tally.bytesOut += chunk.length;
	480	send(conn.peer, conn.toPeer, chunk);
	481	return;
	482	}
	483	if (conn.done) return;
	484	const before = conn.head.length;
	485	conn.head = concat(conn.head, chunk);
	486	// Waiting on the upstream dial: whatever else arrives is kept with
	487	// the head, and goes out after it.
	488	if (before > 0 && headEnd(conn.head.subarray(0, before)) !== -1) return;
	489	if (headEnd(conn.head) === -1) {
	490	if (conn.head.length > HEAD_MAX)
	491	refuse(
	492	client,
	493	"431 Request Header Fields Too Large",
	494	"head too large",
	495	);
	496	return;
	497	}
	498	void open(client, policy, log);
	499	},
	500	drain(client: Socket<Conn>) {
	501	client.data.last = Date.now();
	502	flush(client, client.data.toClient);
	503	},
	504	close(client: Socket<Conn>) {
	505	live.delete(client);
	506	drop(client);
	507	},
	508	error(client: Socket<Conn>) {
	509	live.delete(client);
	510	drop(client);
	511	},
	512	};
	513	const server =
	514	typeof path === "string"
	515	? Bun.listen<Conn>({ unix: path, socket: handlers })
	516	: Bun.listen<Conn>({ hostname: "127.0.0.1", port: path.tcp, socket: handlers });
	517	// A tunnel that moves nothing (a client that stopped reading, an upstream
	518	// that stopped sending) is cut, so it cannot hold an upstream open.
	519	const idleMs = policy.idleMs ?? IDLE_MS;
	520	const sweep = setInterval(
	521	() => {
	522	const now = Date.now();
	523	for (const client of live)
	524	if (now - client.data.last > idleMs) {
	525	log(`egress idle: closed a connection quiet for ${idleMs} ms`);
	526	drop(client);
	527	client.terminate();
	528	live.delete(client);
	529	}
	530	},
	531	Math.max(50, Math.min(5_000, idleMs / 2)),
	532	);
	533	sweep.unref();
	534	return {
	535	close() {
	536	clearInterval(sweep);
	537	for (const client of live) {
	538	drop(client);
	539	client.terminate();
	540	}
	541	live.clear();
	542	server.stop(true);
	543	},
	544	};
394	545	}
395	546
396	547	/**
		@@ -472,11 +623,11 @@ export function relay(
472	623	},
473	624	close(client: Socket<Pair>) {
474	625	client.data.done = true;
475		client.data.peer?.end();
	626	client.data.peer?.terminate();
476	627	},
477	628	error(client: Socket<Pair>) {
478	629	client.data.done = true;
479		client.data.peer?.end();
	630	client.data.peer?.terminate();
480	631	},
481	632	};
482	633	const server =
		@@ -0,0 +1,26 @@
	1	/**
	2	* The egress tally as Prometheus text (#3719): one line per app and
	3	* destination for open connections and the bytes they have moved, so a leaked
	4	* upstream socket shows as a gauge that never falls.
	5	*/
	6	import type { EgressRow } from "./egress.ts";
	7
	8	/** A label value, escaped as the exposition format asks. */
	9	function label(value: string): string {
	10	return value.replace(/\\/g, "\\\\").replace(/"/g, '\\"').replace(/\n/g, "\\n");
	11	}
	12
	13	export function egressMetrics(rows: readonly EgressRow[]): string {
	14	const series: [string, string, (r: EgressRow) => number][] = [
	15	["forge_egress_connections", "Open egress connections.", (r) => r.open],
	16	["forge_egress_bytes_in", "Bytes received from the destination over open connections.", (r) => r.bytesIn],
	17	["forge_egress_bytes_out", "Bytes sent to the destination over open connections.", (r) => r.bytesOut],
	18	];
	19	const lines: string[] = [];
	20	for (const [name, help, value] of series) {
	21	lines.push(`# HELP ${name} ${help}`, `# TYPE ${name} gauge`);
	22	for (const r of rows)
	23	lines.push(`${name}{repo="${label(r.label)}",destination="${label(r.destination)}"} ${value(r)}`);
	24	}
	25	return `${lines.join("\n")}\n`;
	26	}
		@@ -427,6 +427,8 @@ import { passportRoutes } from "./passport_edge.ts";
427	427	import type { Router } from "./router.ts";
428	428	import { clientAddress, clientBucket } from "./peer.ts";
429	429	import { manifestFor } from "../actions/here.ts";
	430	import { egressSnapshot } from "../apps/egress.ts";
	431	import { egressMetrics } from "../apps/egress_metrics.ts";
430	432	import {
431	433	acceptReport,
432	434	dismissReport,
		@@ -4821,6 +4823,20 @@ export function apiRoutes(router: Router<Handler>): void {
4821	4823	),
4822	4824	);
4823	4825
	4826	// Open egress connections per app and destination, as Prometheus text
	4827	// (#3719). Repository names are in the labels, so a site admin only.
	4828	router.add("GET", "/metrics", (ctx) => {
		sentinelcrew 24d ago High: this admin-gated route has no test. Removing `accountAdmin` would leak every app's repository name and destinations, and no test would fail. Test it for anonymous, non-admin, agent-without-scope and admin.
	4829	const denied = accountAdmin(ctx, "ro");
	4830	if (denied) return denied;
	4831	return new Response(egressMetrics(egressSnapshot()), {
	4832	headers: {
	4833	"content-type": "text/plain; version=0.0.4; charset=utf-8",
	4834	"cache-control": "no-store",
	4835	vary: "authorization, cookie",
	4836	},
	4837	});
	4838	});
	4839
4824	4840	// Explore (#607): the rows /explore shows, for the same one line.
4825	4841	router.add("GET", "/api/v1/explore", async (ctx) => {
4826	4842	const query = parseExploreQuery(ctx.url.searchParams.get("q") ?? "");
		@@ -2425,7 +2425,8 @@ async function answerGet(
2425	2425	// generic as `/:owner/:name` can never claim it — a route added there
2426	2426	// next year still can't bring the HTML 404 back for an API caller.
2427	2427	const isApiPath = url.pathname === "/api" \|\| url.pathname.startsWith("/api/");
2428		const matched = isApiPath
	2428	// `/metrics` is the API router's too: a scraper, not a page.
	2429	const matched = isApiPath \|\| url.pathname === "/metrics"
2429	2430	? apiRouter.match(routeMethod, url.pathname)
2430	2431	: router.match(routeMethod, url.pathname);
2431	2432	if (!matched) {
		@@ -32,7 +32,7 @@ import { truncate, wrap } from "../term/width.ts";
32	32	import { badge, card } from "../ui/widgets.ts";
33	33	import type { Session } from "../session.ts";
34	34	import { PIN_MARK, ago, drawFacts, plural } from "./common.ts";
35		import { egressScreen } from "./egress.ts";
	35	import { connectionsScreen, egressScreen } from "./egress.ts";
36	36	import { forkOf, hostOf, liveOrigin, localRepo, sourceName } from "./where.ts";
37	37
38	38	/**
		@@ -98,7 +98,7 @@ export function aboutScreen(
98	98	: []),
99	99	// A site admin also reaches the repository's private egress grants.
100	100	...(f.user?.isAdmin === true && repo !== undefined
101		? [binding(["e"], "egress")]
	101	? [binding(["e"], "egress"), binding(["c"], "connections")]
102	102	: []),
103	103	binding(["esc"], "close"),
104	104	];
		@@ -209,6 +209,10 @@ export function aboutScreen(
209	209	ctx.push(egressScreen(f, ctx, repo));
210	210	return true;
211	211	}
	212	if (e.key === "c" && f.user?.isAdmin === true) {
	213	ctx.push(connectionsScreen(f, ctx));
	214	return true;
	215	}
212	216	if (e.key === "esc" \|\| e.key === "q" \|\| e.key === "I") {
213	217	ctx.pop();
214	218	return true;
		@@ -10,6 +10,8 @@ import {
10	10	revokeAppEgress,
11	11	type EgressRow,
12	12	} from "../../actions/egress_grants.ts";
	13	import { egressSnapshot } from "../../apps/egress.ts";
	14	import { bytes } from "../../system/format.ts";
13	15	import type { Repo } from "../../core/repos.ts";
14	16	import type { Ctx, Screen } from "../app/app.ts";
15	17	import { binding } from "../app/keys.ts";
		@@ -74,3 +76,38 @@ export function egressScreen(f: Session, ctx: Ctx, repo: Repo): Screen {
74	76	};
75	77	return screen;
76	78	}
	79
	80	/**
	81	* The egress connections open right now, per app and destination: the
	82	* Capacity page's card, for a site admin. Read again on every draw, so it
	83	* moves as connections do; anyone else is offered nothing.
	84	*/
	85	export function connectionsScreen(f: Session, ctx: Ctx): Screen {
	86	const state = new ListState();
	87	const lines = (): ListRow[] =>
	88	f.user?.isAdmin !== true
	89	? []
	90	: egressSnapshot().map((r) => ({
	91	title: r.label === "" ? "(unlabelled)" : r.label,
	92	meta: `${r.destination} · ${r.open} open · ${bytes(r.bytesIn)} in · ${bytes(r.bytesOut)} out`,
	93	}));
	94	return {
	95	title: "Egress connections",
	96	keys: [binding(["j", "k"], "move cursor")],
	97	render(area, s) {
	98	const c = ctx.theme.c;
	99	const rows = lines();
	100	state.clamp(rows.length);
	101	const [head, list] = split(area, "v", [len(1), fill(1)]);
	102	const note =
	103	rows.length === 0
	104	? "No app holds an egress connection open."
	105	: `${rows.length} destination${rows.length === 1 ? "" : "s"} in use`;
	106	s.text(head!.x, head!.y, truncate(note, head!.w), style(c.mutedFg, c.bg), head!.x + head!.w);
	107	drawList(s, list!, c, state, rows);
	108	},
	109	key(e) {
	110	return state.key(e, lines().length);
	111	},
	112	};
	113	}
		@@ -8,6 +8,8 @@
8	8
9	9	import { changeCapacity, listCapacity, type CapacityRow } from "../../actions/capacity.ts";
10	10	import { CAPACITY, type CapacityKey } from "../../core/capacity.ts";
	11	import { bytes } from "../../system/format.ts";
	12	import { egressSnapshot } from "../../apps/egress.ts";
11	13	import type { Ctx } from "../../http/context.ts";
12	14	import { notePhase, takeSessionTime, timePhase } from "../../core/route_timing.ts";
13	15	import { notFoundPage, signInFirst } from "../chrome.ts";
		@@ -155,7 +157,32 @@ export function capacityPage(ctx: Ctx): Response {
155	157	pad: false,
156	158	},
157	159	);
158		const body = html`${happened}${when(
	160	const open = egressSnapshot();
	161	const connections = card(
	162	open.length === 0
	163	? html`<div class="card-body">
	164	<p class="muted small" id="egress-open">
	165	No app holds an egress connection open.
	166	</p>
	167	</div>`
	168	: html`<ul class="list" id="egress-open">
	169	${open.map((r) =>
	170	row({
	171	lead: icon("activity"),
	172	title: r.label === "" ? "(unlabelled)" : r.label,
	173	meta: html`<span class="mono">${r.destination}</span> ·
	174	${bytes(r.bytesIn)} in · ${bytes(r.bytesOut)} out`,
	175	trail: html`<span class="mono">${r.open} open</span>`,
	176	}),
	177	)}
	178	</ul>`,
	179	{
	180	head: html`${icon("activity")}
	181	<h2>Egress connections</h2>`,
	182	pad: false,
	183	},
	184	);
	185	const body = html`${happened}${connections}${when(
159	186	reported,
160	187	() => html`<p class="muted small mb" id="egress-reported">
161	188	${icon("activity")} Egress is reported, not enforced: every new app,
		@@ -18,6 +18,7 @@ import { addToCrew, createAgent } from "../src/actions/crew.ts";
18	18	import { assignIssue, openIssue } from "../src/actions/issues.ts";
19	19	import { commitToBranch, createRepo } from "../src/actions/repos.ts";
20	20	import { enableApp } from "../src/actions/apps.ts";
	21	import { startEgress } from "../src/apps/egress.ts";
21	22	import { AppsRuntime } from "../src/apps/runtime.ts";
22	23	import { startChecks, type ChecksHandle } from "../src/checks/service.ts";
23	24	import {
		@@ -384,6 +385,39 @@ describe("report mode is loud to admins, and to nobody else", () => {
384	385	expect(shown.replace(/\s+/g, " ")).toContain("may reach any public host on ports 80 and 443");
385	386	});
386	387
	388	test("the page shows the egress connections an app holds open, and none once they close", async () => {
	389	const camera = Bun.listen<undefined>({ hostname: "127.0.0.1", port: 0, socket: { data() {} } });
	390	const front = Bun.listen({ hostname: "127.0.0.1", port: 0, socket: { data() {} } });
	391	const listen = front.port;
	392	front.stop(true);
	393	const egress = startEgress(
	394	{ tcp: listen },
	395	{ hosts: [], origins: [{ host: "127.0.0.1", port: camera.port }], label: "app:cam/main" },
	396	);
	397	try {
	398	let reply = "";
	399	const client = await Bun.connect<undefined>({
	400	hostname: "127.0.0.1",
	401	port: listen,
	402	socket: { data: (_s, c) => void (reply += new TextDecoder().decode(c)) },
	403	});
	404	client.write(`CONNECT 127.0.0.1:${camera.port} HTTP/1.1\r\n\r\n`);
	405	while (!reply.includes("200")) await Bun.sleep(10);
	406	const shown = (await page()).replace(/\s+/g, " ");
	407	expect(shown).toContain("app:cam/main");
	408	expect(shown).toContain(`127.0.0.1:${camera.port}`);
	409	expect(shown).toContain("1 open");
	410	expect(shown).toMatch(/\d+ B in · \d+ B out/);
	411	expect(shown).not.toContain("→ 127.0.0.1");
	412	egress.close();
	413	expect(await page()).toContain("No app holds an egress connection open.");
	414	client.terminate();
	415	} finally {
	416	egress.close();
	417	camera.stop(true);
	418	}
	419	});
	420
387	421	test("reported: the public health check, a member and a stranger hear nothing of it", async () => {
388	422	const health = await fetch(`${servers.url}/healthz`);
389	423	expect(health.status).toBe(200);