refactor(service): Split bodies, clients, run config, recording files, and the JSON scan into files

README.md

@@ -761,7 +761,11 @@ To see how well the cache works:
   whose body does not come in 300 s too. This is also true in a tunnel
   and with `proxyProtocol`.
 - A `CONNECT` tunnel that sends no byte for 10 s after the `200` closes. A
-  tunnel that sends and gets no byte for 10 minutes closes.
+  tunnel that sends and gets no byte for 10 minutes while a request waits
+  for its response closes. Between two requests, a tunnel closes after
+  about 5 s without a new request (the `keepAliveTimeout` of Node), and
+  the client opens a new one. A gap: in a tunnel without TLS, Node removes
+  the 10-minute limit after the first request.
 - The credentials that a run learns and the values that the sessions of
   a run without the token remember have limits. See [Config](#config) and
   [Changing values](#changing-values).
@@ -1040,9 +1044,16 @@ In `packages/service/src/`:
 - `cli.ts`: the command line, which the image runs.
 - `server.ts`: sockets, HTTPS interception, the allow list, and the control
   API.
+- `bodies.ts`: request bodies, their size limits, and the budget that the
+  bodies of runs without the token share.
+- `clients.ts`: the PROXY protocol header and the address of a client.
+- `run-config.ts`: the check of the run config of a client, with its
+  limits.
 - `errors.ts`: the log line and the client text of an unexpected error.
 - `addresses.ts`: the public addresses that the proxy can send requests to.
 - `recorder.ts`: modes, sessions, replay, and misses.
+- `recording-file.ts`: a recording file from a live response, the response
+  of a replay, and the decoding of compressed bodies.
 - `request-key.ts`: the key of a request.
 - `key-worker.ts`: the worker process that makes the keys of runs without
   the token and redacts the text of their misses, with a time limit and a
@@ -1055,4 +1066,6 @@ In `packages/service/src/`:
 - `streams.ts`: merges the deltas of a recorded model stream.
 - `usage.ts`: the token usage of a model response, for Sentry.
 - `secrets.ts`: redaction of credentials.
+- `json-strings.ts`: the one-pass scan of the JSON strings of a body, for
+  `secrets.ts`.
 - `certificates.ts`: the certificate authority for HTTPS.

TELEMETRY.md

@@ -66,11 +66,11 @@ For metrics, use `-f dataset=tracemetrics`,
 | Pivot                       | Meaning                                    | Found In                          |
 | --------------------------- | ------------------------------------------ | --------------------------------- |
 | `roach.tenant`              | repository of the run                      | metrics, spans, errors            |
-| `roach.mode`                | `auto`, `replay`, `record`                 | metrics, spans, errors            |
+| `roach.mode`                | `auto`, `replay`, `record`, `off`          | metrics, spans, errors            |
 | `roach.rule`                | recording rule, such as `model`            | metrics, spans                    |
 | `roach.result`              | `replayed`, `live`, and others (see Words) | metrics, spans                    |
 | `roach.origin`              | upstream origin                            | metrics, spans                    |
-| `roach.reason`              | why the proxy refused a request            | `roach.request`                   |
+| `roach.reason`              | why the proxy refused a request            | `roach.request` metric and span   |
 | `roach.run.id`              | `name` of the run, else the service id     | spans, errors                     |
 | `roach.ci.run_id`           | `GITHUB_RUN_ID` of the run                 | spans, errors                     |
 | `roach.ci.job`              | `GITHUB_JOB` of the run                    | spans, errors                     |

packages/service/src/bodies.ts

@@ -0,0 +1,169 @@
+/**
+ * Request bodies of the proxy and the control API: their size limits, the
+ * budget of bytes that the bodies of many runs share (`BodyBudget`), and
+ * the time that a body has to come.
+ */
+import type http from "node:http";
+
+/**
+ * A control API body over this size gets HTTP 413. Anyone can call
+ * `POST /runs`, so the service reads no more before it checks the token.
+ */
+const MAX_CONTROL_BODY_BYTES = 1024 * 1024;
+
+/** An error that answers a request with its own status. */
+const httpError = (status: number, message: string) =>
+  Object.assign(new Error(message), { status });
+
+/**
+ * The bytes of request bodies that the requests of many targets can hold
+ * at one time (`ProxyTarget.bodyBudget`). `used` counts the bodies of the
+ * requests that did not finish, and `clients` counts them for each client
+ * (`ProxyTarget.client`). One client can hold at most `perClient` bytes,
+ * so it cannot take all of `max`. A body that takes longer than `readMs`
+ * to come fails with HTTP 408 and gives its bytes back.
+ */
+export interface BodyBudget {
+  max: number;
+  perClient: number;
+  readMs: number;
+  used: number;
+  clients: Map<string, number>;
+}
+
+/** The bytes that one request holds of its `BodyBudget`. */
+export interface Held {
+  budget: BodyBudget | undefined;
+  client: string;
+  bytes: number;
+}
+
+/** Hold `bytes` more of the budget. Returns `false` when they do not fit. */
+function hold(held: Held, bytes: number): boolean {
+  const { budget, client } = held;
+  if (!budget) return true;
+  const ofClient = budget.clients.get(client) ?? 0;
+  if (budget.used + bytes > budget.max || ofClient + bytes > budget.perClient) {
+    return false;
+  }
+  budget.used += bytes;
+  budget.clients.set(client, ofClient + bytes);
+  held.bytes += bytes;
+  return true;
+}
+
+/** Give back all bytes that a request holds. */
+export function release(held: Held): void {
+  const { budget, client, bytes } = held;
+  held.bytes = 0;
+  if (!budget || bytes === 0) return;
+  budget.used -= bytes;
+  const left = (budget.clients.get(client) ?? 0) - bytes;
+  if (left > 0) budget.clients.set(client, left);
+  else budget.clients.delete(client);
+}
+
+const busyError = () =>
+  httpError(
+    503,
+    "the service has too many large requests at this time. Send the request again later.",
+  );
+
+/**
+ * Read a whole body. A body over `maxBytes` fails with HTTP 413. A body
+ * that does not fit in the budget of `held` fails with HTTP 503, and one
+ * that does not come in the `readMs` of the budget fails with HTTP 408.
+ * Only request bodies have a limit: the client sends them, and a large
+ * upstream response is not the fault of the client.
+ */
+export function readBody(
+  stream: http.IncomingMessage,
+  maxBytes = Infinity,
+  held: Held = { budget: undefined, client: "", bytes: 0 },
+): Promise<Buffer> {
+  const declared = Number(stream.headers["content-length"]);
+  if (declared > maxBytes) {
+    // Drop the body without a look at it, so it costs no memory.
+    stream.resume();
+    return Promise.reject(httpError(413, `body is over ${maxBytes} bytes`));
+  }
+  // Hold the bytes when they come, not the declared length. So a client
+  // that declares a large body and then sends nothing holds nothing.
+  return new Promise((resolve, reject) => {
+    const chunks: Buffer[] = [];
+    let size = 0;
+    let failed = false;
+    const fail = (error: Error) => {
+      // Read the rest without keeping it, so the client gets the error.
+      failed = true;
+      chunks.length = 0;
+      clearTimeout(deadline);
+      reject(error);
+    };
+    // Node gives a whole request 300 s (`HTTP_OPTIONS`). That is too long
+    // to hold a part of the budget, so these bodies have a shorter limit.
+    const readMs = held.budget?.readMs;
+    const deadline =
+      readMs === undefined
+        ? undefined
+        : setTimeout(
+            () =>
+              fail(
+                httpError(408, `the body did not come in ${readMs / 1000} s`),
+              ),
+            readMs,
+          );
+    stream.on("data", (chunk: Buffer) => {
+      if (failed) return;
+      size += chunk.length;
+      if (size > maxBytes) {
+        fail(httpError(413, `body is over ${maxBytes} bytes`));
+      } else if (!hold(held, chunk.length)) {
+        fail(busyError());
+      } else {
+        chunks.push(chunk);
+      }
+    });
+    stream.on("end", () => {
+      clearTimeout(deadline);
+      resolve(Buffer.concat(chunks));
+    });
+    stream.on("error", (error: Error & { code?: string }) => {
+      clearTimeout(deadline);
+      // A client that closes its connection before its body came is not a
+      // failure of the proxy, so it does not go to Sentry. `method` is set
+      // only on a request, not on an upstream response.
+      const aborted = stream.method !== null && error.code === "ECONNRESET";
+      reject(aborted ? abortedError() : error);
+    });
+  });
+}
+
+/** The error of a request whose client closed it before its body came. */
+const abortedError = () =>
+  Object.assign(
+    httpError(400, "the client closed the request before its body came"),
+    { aborted: true },
+  );
+
+/**
+ * Read a JSON object body of a control API call. A body over
+ * `MAX_CONTROL_BODY_BYTES` fails with HTTP 413 before it is parsed. Any
+ * other body that is not a JSON object fails with HTTP 400.
+ */
+export async function readJson<T>(incoming: http.IncomingMessage): Promise<T> {
+  const text = (await readBody(incoming, MAX_CONTROL_BODY_BYTES)).toString(
+    "utf8",
+  );
+  let value: unknown;
+  try {
+    value = JSON.parse(text || "{}");
+  } catch {
+    throw httpError(400, "body is not JSON");
+  }
+  // Callers read fields, so `null`, a list, or a single value is a bad body.
+  if (typeof value !== "object" || value === null || Array.isArray(value)) {
+    throw httpError(400, "body is not a JSON object");
+  }
+  return value as T;
+}

packages/service/src/clients.ts

@@ -0,0 +1,127 @@
+/**
+ * The address of the client of a proxy connection, for the limit of runs
+ * for each client (`service.ts`).
+ *
+ * Behind a load balancer, a connection can start with a PROXY protocol v1
+ * header (`proxyProtocol`). It gives the address of the client
+ * (`clientAddress()`). An IPv6 client is grouped by its /64
+ * (`clientGroup()`).
+ */
+import type http from "node:http";
+import { isIP, type Socket } from "node:net";
+
+/**
+ * A PROXY protocol v1 header, such as
+ * `PROXY TCP4 203.0.113.7 10.0.0.2 51234 443`. It is at most 107 bytes.
+ * The first bytes of a connection must come in `PROXY_HEADER_MS`.
+ */
+const PROXY_PREFIX = Buffer.from("PROXY ");
+
+const PROXY_HEADER =
+  /^PROXY (?:TCP[46] ([0-9A-Fa-f:.]+) [0-9A-Fa-f:.]+ \d{1,5} \d{1,5}|UNKNOWN[^\r\n]*)\r\n/;
+
+const MAX_PROXY_HEADER_BYTES = 107;
+
+const PROXY_HEADER_MS = 10_000;
+
+/** The address of the client of each connection with a PROXY header. */
+const proxiedClients = new WeakMap<Socket, string>();
+
+/**
+ * The address of the client of a request. With `proxyProtocol`, it comes
+ * from the PROXY header of the connection, when the connection has one.
+ * Else it is the address of the socket.
+ */
+export function clientAddress(incoming: http.IncomingMessage): string {
+  return (
+    proxiedClients.get(incoming.socket) ?? incoming.socket.remoteAddress ?? ""
+  );
+}
+
+/**
+ * The group of a client address for the limit of runs for each client. An
+ * IPv6 client often has a whole /64, so the group of an IPv6 address is
+ * its /64, such as `2001:db8:0:1::/64`. An IPv4 address is its own group,
+ * also in each IPv6 form, such as `::ffff:192.0.2.1`, `::ffff:c000:201`,
+ * and `0:0:0:0:0:ffff:192.0.2.1`.
+ */
+export function clientGroup(address: string): string {
+  if (isIP(address) !== 6) return address;
+  const groups = ipv6Groups(address);
+  if (
+    groups.slice(0, 5).every((group) => group === 0) &&
+    groups[5] === 0xffff
+  ) {
+    const [high = 0, low = 0] = groups.slice(6);
+    return [high >> 8, high & 0xff, low >> 8, low & 0xff].join(".");
+  }
+  return `${groups
+    .slice(0, 4)
+    .map((group) => group.toString(16))
+    .join(":")}::/64`;
+}
+
+/** The eight 16-bit groups of an IPv6 address. */
+function ipv6Groups(address: string): number[] {
+  // The URL parser writes each IPv6 form with hex groups only, such as
+  // `::ffff:c000:201`. A zone, such as `%eth0`, is not part of it.
+  const [head = "", tail] = new URL(
+    `http://[${address.split("%")[0]}]`,
+  ).hostname
+    .slice(1, -1)
+    .split("::");
+  const front = head ? head.split(":") : [];
+  const back = tail ? tail.split(":") : [];
+  const zeros = Array<string>(8 - front.length - back.length).fill("0");
+  return [...front, ...zeros, ...back].map((group) =>
+    Number.parseInt(group, 16),
+  );
+}
+
+/**
+ * Read the PROXY protocol v1 header of a new connection, when it has one,
+ * and keep the address of its client. Then give the connection to
+ * `server`. A connection without the header keeps the address of its
+ * socket. A header that is not valid closes the connection.
+ *
+ * The load balancer writes its header first, so a client cannot put a
+ * header before it. Without a load balancer, a client can send its own
+ * header. That changes only the address for the limit of runs for each
+ * client.
+ */
+export function acceptProxied(socket: Socket, server: http.Server): void {
+  let read = Buffer.alloc(0);
+  const timer = setTimeout(() => socket.destroy(), PROXY_HEADER_MS);
+  const accept = (rest: Buffer) => {
+    socket.off("data", onData);
+    clearTimeout(timer);
+    socket.pause();
+    if (rest.length > 0) socket.unshift(rest);
+    server.emit("connection", socket);
+    socket.resume();
+  };
+  const onData = (chunk: Buffer) => {
+    read = Buffer.concat([read, chunk]);
+    const prefix = Math.min(read.length, PROXY_PREFIX.length);
+    if (!read.subarray(0, prefix).equals(PROXY_PREFIX.subarray(0, prefix))) {
+      accept(read);
+      return;
+    }
+    const end = read.indexOf("\r\n");
+    if (end < 0 || end + 2 > MAX_PROXY_HEADER_BYTES) {
+      if (end >= 0 || read.length >= MAX_PROXY_HEADER_BYTES) socket.destroy();
+      return;
+    }
+    const match = PROXY_HEADER.exec(read.subarray(0, end + 2).toString());
+    if (!match) {
+      socket.destroy();
+      return;
+    }
+    // `UNKNOWN`, such as from a health check, gives no address.
+    if (match[1] && isIP(match[1])) proxiedClients.set(socket, match[1]);
+    accept(read.subarray(end + 2));
+  };
+  socket.on("error", () => socket.destroy());
+  socket.on("close", () => clearTimeout(timer));
+  socket.on("data", onData);
+}