Make the ingress wait configurable and add an autoscaling guide

.changeset/ingress-wait.md

@@ -0,0 +1,5 @@
+---
+"sql-fs-api": patch
+---
+
+Add `SQL_FS_MAX_INGRESS_WAIT_MS`, the time a request or MCP listener waits for ingress capacity before it gets a 503 `EWORKER_BUSY`. The default stays 250 ms, the previous fixed value. Like the other ingress settings, an empty value uses the default. With workers on, any other invalid value, including a wait above 2147483647 ms, stops startup. `SQL_FS_WORKER_QUEUE_WAIT_MS` now rejects values above 2147483647 ms too, because Node would run such a timer after 1 ms.

.changeset/mcp-shutdown.md

@@ -0,0 +1,5 @@
+---
+"sql-fs-api": patch
+---
+
+Shut down promptly when MCP clients hold a `GET /mcp` listener stream open. The server waited for those streams before it drained workers, so every such shutdown ran into the 60 s force exit and exited with code 1. It now ends listener streams first and lets in-flight MCP tool calls finish.

.github/workflows/ci.yml

@@ -2,7 +2,7 @@ name: CI
 
 on:
   push:
-    branches: [main, "ralph/**", "feature/**"]
+    branches: [main]
   pull_request:
 
 concurrency:

.github/workflows/python-sdk-ci.yml

@@ -2,7 +2,7 @@ name: Python SDK CI
 
 on:
   push:
-    branches: [main, "ci/**", "feat/**", "fix/**", "feature/**", "ralph/**"]
+    branches: [main]
   pull_request:
 
 permissions:

AGENTS.md

@@ -19,7 +19,7 @@ pnpm verify:architecture # also worker failure, stream finalization, input budge
 
 Use `pnpm check:typescript` or `pnpm check:python` for SDK-only changes. Run a targeted core test with `pnpm exec vitest run src/sql-fs/tests/unit/sql-fs.cache.test.ts`. A nonexistent test selection must fail. `pnpm test:integration` is the lower-level command for an already migrated disposable database with `DATABASE_URL` and `REDIS_URL`; it can skip without those variables. Prefer `pnpm verify:local` for completion evidence.
 
-See [READINESS.md](docs/READINESS.md) for verified scope and hosted follow-up. Read [CONTRIBUTING.md](CONTRIBUTING.md) for setup and contribution steps, [DEVELOPER.md](docs/DEVELOPER.md) before changing consistency or session ownership, and [README.md](README.md) for API behavior, and [CONFIGURATION.md](docs/CONFIGURATION.md) for runtime settings.
+See [READINESS.md](docs/READINESS.md) for verified scope and hosted follow-up. Read [CONTRIBUTING.md](CONTRIBUTING.md) for setup and contribution steps, [DEVELOPER.md](docs/DEVELOPER.md) before changing consistency or session ownership, [README.md](README.md) for API behavior, [CONFIGURATION.md](docs/CONFIGURATION.md) for runtime settings, and [AUTOSCALING.md](docs/AUTOSCALING.md) for probes, metrics and replica scaling.
 
 Python type checking targets the oldest supported interpreter. `pnpm check:python:types` uses uv to provision Python 3.9 in `clients/python/.venv-py39`, separate from the Python 3.12 development environment. This keeps the dependency syntax aligned with the declared target. Mypy stays below version 2 while Python 3.9 is supported.
 

README.md

@@ -133,7 +133,7 @@ docker run -p 8080:8080 \
 	sql-fs-api
 ```
 
-To run several replicas, point them all at one Postgres database and one control Redis in `REDIS_URL`. Put the blob cache on a separate Redis in `REDIS_DATA_URL` with `allkeys-lru`, and keep the control Redis on `noeviction`. [Redis eviction policy](docs/CONFIGURATION.md#redis-eviction-policy) explains why.
+To run several replicas, point them all at one Postgres database and one control Redis in `REDIS_URL`. Put the blob cache on a separate Redis in `REDIS_DATA_URL` with `allkeys-lru`, and keep the control Redis on `noeviction`. [Redis eviction policy](docs/CONFIGURATION.md#redis-eviction-policy) explains why. [AUTOSCALING.md](docs/AUTOSCALING.md) covers health probes, metrics and scaling the replica count.
 
 ## Develop
 

docs/CONFIGURATION.md

@@ -171,7 +171,9 @@ To upgrade replicas that run with workers off, roll out in two stages. A rolling
 1. Deploy the new version with `SQL_FS_WORKERS_ENABLED=false` on every replica.
 2. When no replica of the older version remains, remove the setting and roll out again.
 
-**Admission.** Each tenant has a queue, and the server takes requests from the queues in turn. On one sandbox, a waiting writer goes before new readers. With the defaults, one tenant can hold one of the two workers and four of the eight ingress permits. Long-lived `GET /mcp` streams count against a separate listener quota. If a request waits more than 250 ms, it fails with `EWORKER_BUSY` (503) and `Retry-After: 1`.
+**Admission.** Each tenant has a queue, and the server takes requests from the queues in turn. On one sandbox, a waiting writer goes before new readers. With the defaults, one tenant can hold one of the two workers and four of the eight ingress permits. Long-lived `GET /mcp` streams count against a separate listener quota. A request that waits longer than `SQL_FS_MAX_INGRESS_WAIT_MS` for an ingress permit, or an exec that waits longer than `SQL_FS_WORKER_QUEUE_WAIT_MS` for a worker, fails with `EWORKER_BUSY` (503) and `Retry-After: 1`. With the defaults, an exec can queue for up to 0.5 s before it fails.
+
+A longer ingress wait does not absorb bursts, because the ingress queue holds no more requests than ingress admits at once. [Wait limits](AUTOSCALING.md#wait-limits) has the measurements.
 
 **Unknown outcomes.** If a script loses its worker, lease or response after it starts, the request fails with `EEXEC_OUTCOME_UNKNOWN` (503, `retryable: false`). The server and the SDKs do not replay it. That includes read-only scripts, because they can still have network side effects. Read the sandbox state before the next action. Earlier items in a batch can already be committed.
 
@@ -199,6 +201,7 @@ Every worker setting must be a positive integer. If a global limit is above 1, i
 | `SQL_FS_MAX_INGRESS_TENANT_BYTES` | Three quarters of `SQL_FS_MAX_INGRESS_BYTES`, at least 1 |
 | `SQL_FS_MAX_INGRESS_MCP_LISTENERS` | 8 listeners |
 | `SQL_FS_MAX_INGRESS_TENANT_MCP_LISTENERS` | 4 listeners |
+| `SQL_FS_MAX_INGRESS_WAIT_MS` | 250 ms |
 | `SQL_FS_WORKER_MAX_RSS_BYTES` | 1.5 GiB |
 | `SQL_FS_WORKER_MAX_EXTERNAL_BYTES` | 512 MiB |
 | `SQL_FS_WORKER_SOURCE_BYTES` | 1 MiB |
@@ -212,7 +215,7 @@ Every worker setting must be a positive integer. If a global limit is above 1, i
 | `SQL_FS_WORKER_SCRIPT_BUFFER_OPS` | 50000 changes |
 | `SQL_FS_WORKER_SCRIPT_BUFFER_BYTES` | 32 MiB |
 
-A replica does not add workers by itself. Scale the number of replicas on CPU and RSS, together with the worker queue, admission delay and refusal metrics from `GET /metrics`. Waits on one sandbox, on distributed locks and on recovery have their own gauges. More replicas do not shorten those waits.
+A replica does not add workers by itself. To handle more load, add replicas. [AUTOSCALING.md](AUTOSCALING.md) describes the metrics to scale on and the waits that more replicas do not shorten.
 
 ### Sizing
 
@@ -232,13 +235,7 @@ Ten concurrent 64 MiB `PUT` uploads, with `MAX_FILE_WRITE_BYTES` raised to 64 Mi
 
 One replica served 100 agents with separate sandboxes and 5 s think time, about 20 execs per second, with 0 errors and p99 latency under 50 ms. At about 100 execs per second, one tenant saturated at about 71 per second, because its active limit is half the worker count, 1 per replica. Ten tenants reached 97 per second with 0 errors.
 
-Add replicas instead of raising caps, and keep the caps fixed for each box size. Autoscale on these signals:
-
-- Queue wait, from the `sql_fs_worker_queue_wait_milliseconds` histogram.
-- Queue depth, from `sql_fs_worker_queued` and `sql_fs_ingress_queued_requests`.
-- Refusals and timeouts, from `sql_fs_worker_refused_total`, `sql_fs_worker_queue_timeouts_total`, `sql_fs_ingress_refused_total` and `sql_fs_ingress_queue_timeouts_total`.
-- Memory, from `sql_fs_worker_process_rss_bytes`.
-- CPU.
+Add replicas instead of raising caps, and keep the caps fixed for each box size. [AUTOSCALING.md](AUTOSCALING.md) lists the metrics to scale on, a replica-count estimate and an example scale rule.
 
 ### Large files
 

scripts/verify-local.sh

@@ -50,6 +50,7 @@ pnpm build
 pnpm --dir clients/typescript build
 "${verify_env[@]}" node scripts/verify/contracts.mjs
 "${verify_env[@]}" node scripts/verify/output-cap.mjs
+"${verify_env[@]}" node scripts/verify/shutdown.mjs
 "${verify_env[@]}" pnpm test:integration --reporter=default --reporter=json --outputFile="$evidence/integration.json"
 node --input-type=module - "$evidence/integration.json" <<'JS'
 import assert from "node:assert/strict";

scripts/verify/shutdown.mjs

@@ -0,0 +1,65 @@
+#!/usr/bin/env node
+import { once } from "node:events";
+import { join } from "node:path";
+import { performance } from "node:perf_hooks";
+import { LocalApiRun, repo } from "./local-api.mjs";
+
+const run = new LocalApiRun("shutdown");
+let failure;
+try {
+	const server = await run.start({ SQL_FS_WORKERS_ENABLED: "true" });
+	// No run.owner(): its cleanup needs a live server after this one exits.
+	const { signToken } = await import(join(repo, "dist/api/lib/jwt.js"));
+	const token = await signToken({ sub: `${run.runId}-mcp`, expiresIn: "1h", secret: process.env.AUTH_SECRET });
+	const headers = {
+		Authorization: `Bearer ${token}`,
+		Accept: "application/json, text/event-stream",
+		"Content-Type": "application/json",
+	};
+	const initialize = await fetch(`${server.base}/mcp`, {
+		method: "POST",
+		headers,
+		body: JSON.stringify({
+			jsonrpc: "2.0",
+			id: 1,
+			method: "initialize",
+			params: { protocolVersion: "2025-03-26", capabilities: {}, clientInfo: { name: "shutdown-verifier", version: "1" } },
+		}),
+	});
+	await initialize.text();
+	const sessionId = initialize.headers.get("mcp-session-id");
+	run.equal(typeof sessionId === "string" && sessionId.length > 0, true, "MCP session opened");
+	const sessionHeaders = { ...headers, "mcp-session-id": sessionId };
+	const initialized = await fetch(`${server.base}/mcp`, {
+		method: "POST",
+		headers: sessionHeaders,
+		body: JSON.stringify({ jsonrpc: "2.0", method: "notifications/initialized" }),
+	});
+	await initialized.text();
+	const listener = await fetch(`${server.base}/mcp`, {
+		method: "GET",
+		headers: { ...sessionHeaders, Accept: "text/event-stream" },
+	});
+	run.equal(listener.status, 200, "MCP listener stream is open before SIGTERM");
+	const drained = listener.body
+		.pipeTo(new WritableStream())
+		.then(() => "ended")
+		.catch(() => "aborted");
+
+	const started = performance.now();
+	const exited = once(server.child, "close");
+	server.child.kill("SIGTERM");
+	const [code] = await exited;
+	const seconds = (performance.now() - started) / 1000;
+	const listenerEnd = await drained;
+	run.record({ case: "sigterm-with-mcp-listener", exitCode: code, seconds, listener: listenerEnd });
+	run.equal(listenerEnd, "ended", "Shutdown ends the MCP listener stream cleanly");
+	run.equal(code, 0, "SIGTERM with an open MCP listener exits cleanly");
+	run.equal(seconds < 10, true, "SIGTERM with an open MCP listener does not wait for the force-exit timer");
+	run.equal(server.log.includes('"event":"shutdown_complete"'), true, "Shutdown logs shutdown_complete");
+	run.equal(server.log.includes('"event":"shutdown_force_exit"'), false, "Shutdown does not force-exit");
+} catch (error) {
+	failure = error;
+} finally {
+	await run.finish(failure);
+}

src/api/ingress-admission.ts

@@ -2,6 +2,7 @@ import { AsyncLocalStorage } from "node:async_hooks";
 import { ServerResponse } from "node:http";
 import { Http2ServerResponse } from "node:http2";
 import type { Context, MiddlewareHandler, Next } from "hono";
+import { assertTimerMs } from "./lib/env.js";
 import { FairAdmission } from "./worker-owner/admission.js";
 import { WorkerBusyError } from "./worker-owner/protocol.js";
 
@@ -32,6 +33,7 @@ export interface IngressOptions {
 	readonly maxBytes: number;
 	readonly maxTenantBytes: number;
 	readonly maxBodyBytes: number;
+	readonly waitMs: number;
 }
 
 export function loadIngressOptions(env: NodeJS.ProcessEnv = process.env): IngressOptions {
@@ -67,6 +69,7 @@ export function loadIngressOptions(env: NodeJS.ProcessEnv = process.env): Ingres
 		maxBytes,
 		maxTenantBytes,
 		maxBodyBytes: read("MAX_REQUEST_BODY_BYTES", 256 * 1024 * 1024),
+		waitMs: assertTimerMs("SQL_FS_MAX_INGRESS_WAIT_MS", read("SQL_FS_MAX_INGRESS_WAIT_MS", 250)),
 	};
 }
 
@@ -89,7 +92,7 @@ export class IngressAdmission {
 			maxQueuedPerTenant: options.maxTenantRequests,
 			maxQueuedBytes: 0,
 			maxQueuedBytesPerTenant: 0,
-			waitMs: 250,
+			waitMs: options.waitMs,
 		});
 		this.mcpListenerAdmission = new FairAdmission({
 			capacity: options.maxMcpListeners ?? options.maxRequests,
@@ -98,7 +101,7 @@ export class IngressAdmission {
 			maxQueuedPerTenant: options.maxTenantMcpListeners ?? options.maxTenantRequests,
 			maxQueuedBytes: 0,
 			maxQueuedBytesPerTenant: 0,
-			waitMs: 250,
+			waitMs: options.waitMs,
 		});
 	}
 

src/api/lib/env.ts

@@ -30,6 +30,14 @@ export function configuredInteger(env: NodeJS.ProcessEnv, name: string, fallback
 	return parsed;
 }
 
+/** Node runs any longer `setTimeout` delay after 1 ms, so a larger wait would refuse requests at once. */
+export const MAX_TIMER_MS = 2_147_483_647;
+
+export function assertTimerMs(name: string, value: number): number {
+	if (value > MAX_TIMER_MS) throw new Error(`${name} must be at most ${MAX_TIMER_MS}`);
+	return value;
+}
+
 export function configuredBoolean(env: NodeJS.ProcessEnv, name: string, fallback: boolean): boolean {
 	const value = env[name];
 	if (value === undefined) return fallback;

src/api/mcp/server.ts

@@ -78,6 +78,20 @@ export function startMcpSessionSweeper(intervalMs: number = MCP_SWEEP_INTERVAL_M
 	if (typeof sweeperTimer.unref === "function") sweeperTimer.unref();
 }
 
+/**
+ * End every session's standalone `GET /mcp` stream, leaving in-flight tool calls to finish. Shutdown calls this
+ * before `server.close()`, which otherwise waits on those long-lived connections until the force-exit timer.
+ */
+export function closeMcpListeners(): void {
+	for (const entry of sessions.values()) {
+		try {
+			entry.transport.closeStandaloneSSEStream();
+		} catch {
+			// best-effort
+		}
+	}
+}
+
 /** Stop the sweeper and close every active MCP transport. Used during shutdown. */
 export async function shutdownMcp(): Promise<void> {
 	if (sweeperTimer !== undefined) {

src/api/server.ts

@@ -29,7 +29,7 @@ import {
 import { loadExecLockOptions } from "./exec-lock-config.js";
 import { IngressAdmission, loadIngressOptions } from "./ingress-admission.js";
 import { configuredBoolean, loadExecOutputBytes } from "./lib/env.js";
-import { handleMcpRequest, shutdownMcp, startMcpSessionSweeper } from "./mcp/server.js";
+import { closeMcpListeners, handleMcpRequest, shutdownMcp, startMcpSessionSweeper } from "./mcp/server.js";
 import { mcpOptionsResponse, withMcpCors } from "./mcp-cors.js";
 import { runMigrations } from "./migrations.js";
 import { openapiSpec } from "./openapi-spec.js";
@@ -376,6 +376,7 @@ if (isMain) {
 			shuttingDown = true;
 			console.log(JSON.stringify({ event: "shutdown_begin" }));
 			ingressAdmission?.close();
+			closeMcpListeners();
 			stopEventLoopMonitor();
 			// Force-exit guard so a hung Postgres or Redis cleanup cannot keep
 			// the process alive past the orchestrator's grace period.

src/api/tests/unit/ingress-admission.test.ts

@@ -30,6 +30,7 @@ function setup(maxRequests = 8, maxBytes = 128, maxBodyBytes = 256) {
 		maxBytes,
 		maxTenantBytes: Math.max(1, maxBytes - 1),
 		maxBodyBytes,
+		waitMs: 250,
 	});
 	const app = new Hono<{ Variables: AuthVariables }>();
 	let tenantSequence = 0;
@@ -59,6 +60,18 @@ afterEach(() => {
 });
 
 describe("ingress configuration", () => {
+	it("waits 250 ms at ingress by default, independent of the worker queue wait", () => {
+		expect(loadIngressOptions({ SQL_FS_WORKER_QUEUE_WAIT_MS: "900" }).waitMs).toBe(250);
+		expect(loadIngressOptions({ SQL_FS_MAX_INGRESS_WAIT_MS: "40" }).waitMs).toBe(40);
+	});
+
+	it("rejects an ingress wait that Node's timer would cut to 1 ms", () => {
+		expect(loadIngressOptions({ SQL_FS_MAX_INGRESS_WAIT_MS: "2147483647" }).waitMs).toBe(2147483647);
+		expect(() => loadIngressOptions({ SQL_FS_MAX_INGRESS_WAIT_MS: "2147483648" })).toThrow(
+			"SQL_FS_MAX_INGRESS_WAIT_MS must be at most 2147483647",
+		);
+	});
+
 	it("reserves explicit raw-body headroom with bounded concurrent requests", () => {
 		expect(loadIngressOptions({})).toEqual({
 			maxRequests: 8,
@@ -68,6 +81,7 @@ describe("ingress configuration", () => {
 			maxBytes: 128 * 1024 * 1024,
 			maxTenantBytes: 96 * 1024 * 1024,
 			maxBodyBytes: 256 * 1024 * 1024,
+			waitMs: 250,
 		});
 	});
 
@@ -79,6 +93,7 @@ describe("ingress configuration", () => {
 		"SQL_FS_MAX_INGRESS_BYTES",
 		"SQL_FS_MAX_INGRESS_TENANT_BYTES",
 		"MAX_REQUEST_BODY_BYTES",
+		"SQL_FS_MAX_INGRESS_WAIT_MS",
 	])("rejects invalid %s instead of disabling its budget", (name) => {
 		for (const value of ["0", "-1", "1.5", "NaN", "Infinity", "9007199254740992"]) {
 			expect(() => loadIngressOptions({ [name]: value })).toThrow(name);
@@ -264,6 +279,7 @@ describe("HTTP body admission", () => {
 			maxBytes: 100,
 			maxTenantBytes: 99,
 			maxBodyBytes: 256,
+			waitMs: 250,
 		});
 		const app = new Hono<{ Variables: AuthVariables }>();
 		app.onError((_error, c) => c.text("body rejected", 400));
@@ -575,6 +591,7 @@ describe("HTTP body admission", () => {
 			maxBytes: 150,
 			maxTenantBytes: 149,
 			maxBodyBytes: 256,
+			waitMs: 250,
 		});
 		const app = new Hono<{ Variables: AuthVariables }>();
 		app.use("/v1/*", authMiddleware);

src/api/tests/unit/ingress-wait.test.ts

@@ -11,19 +11,20 @@ function deferred() {
 	return { promise, resolve };
 }
 
-function setup(maxRequests = 8, maxTenantRequests = 4) {
+function setup(maxRequests = 8, maxTenantRequests = 4, env: NodeJS.ProcessEnv = {}) {
 	const ingress = new IngressAdmission(
 		loadIngressOptions({
 			SQL_FS_MAX_INGRESS_REQUESTS: String(maxRequests),
 			SQL_FS_MAX_INGRESS_TENANT_REQUESTS: String(maxTenantRequests),
+			...env,
 		}),
 	);
 	const app = new Hono<{ Variables: AuthVariables }>();
 	app.onError((_error, c) => c.text("failed", 500));
 	app.use("*", (c, next) => {
 		if (c.req.path.startsWith("/bootstrap/")) return ingress.bootstrapMiddleware(c, next);
 		c.set("tenant", c.req.header("X-Tenant") ?? "observer");
-		return ingress.middleware(c, next);
+		return c.req.path.startsWith("/mcp") ? ingress.mcpMiddleware(c, next) : ingress.middleware(c, next);
 	});
 	app.post("/work", async (c) => c.text(await c.req.text()));
 	return { ingress, app };
@@ -196,6 +197,40 @@ describe("bounded ingress wait", () => {
 		}
 	});
 
+	it.each([
+		{ gate: "request", method: "POST", path: "/hold" },
+		{ gate: "MCP listener", method: "GET", path: "/mcp/listen" },
+	])("refuses a queued $gate only after SQL_FS_MAX_INGRESS_WAIT_MS", async ({ method, path }) => {
+		const { app, ingress } = setup(2, 1, { SQL_FS_MAX_INGRESS_WAIT_MS: "40" });
+		const hold = deferred();
+		app.on(["GET", "POST"], path, async (c) => {
+			await hold.promise;
+			return c.text("held");
+		});
+		const active = app.request(path, { method });
+		let waiting: Promise<Response> | undefined;
+		try {
+			await flush();
+			waiting = Promise.resolve(app.request(path, { method }));
+			await flush();
+			await vi.advanceTimersByTimeAsync(39);
+			const gate = () =>
+				method === "GET"
+					? { queued: ingress.metrics.queuedMcpListeners, timeouts: ingress.metrics.mcpListenerQueueTimeouts }
+					: { queued: ingress.metrics.queuedRequests, timeouts: ingress.metrics.queueTimeouts };
+			expect(gate()).toEqual({ queued: 1, timeouts: 0 });
+			await vi.advanceTimersByTimeAsync(1);
+			expect(gate()).toEqual({ queued: 0, timeouts: 1 });
+			const response = await waiting;
+			expect(response.status).toBe(503);
+			expect(response.headers.get("Retry-After")).toBe("1");
+			expect(await response.json()).toMatchObject({ code: "EWORKER_BUSY", retryable: true });
+		} finally {
+			hold.resolve();
+			await Promise.allSettled([active, ...(waiting ? [waiting] : [])].map(consume));
+		}
+	});
+
 	it("cancels a queued request without pulling its body and reuses its pending slot", async () => {
 		const { app, ingress } = setup(2, 1);
 		const hold = deferred();

src/api/tests/unit/sandboxes.test.ts

@@ -398,6 +398,7 @@ describe("sandbox creation body admission", () => {
 			maxBytes: 128,
 			maxTenantBytes: 96,
 			maxBodyBytes: 64,
+			waitMs: 250,
 		});
 		const app = new Hono<{ Variables: AuthVariables }>();
 		app.onError((_error, c) => c.json({ error: "body rejected" }, 500));