From 1021076a12b301149c7627ce8106b832b309cb1c Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Thu, 12 Nov 2026 00:34:58 -0800 Subject: [PATCH 01/11] perf(connect): connects skip a repeated descriptor read and reuse HTTP connections Each environment connect did serial requests that slow links multiply. - A route check's descriptor is reused by the connect it leads to over plain HTTP on loopback and the tailnet, not only over HTTPS. A plain LAN address still reads again, because it can belong to another machine after a network change. - T3 Connect: a token mint hands its descriptor to the connect, and a cached token reads the descriptor beside the ticket request instead of after it. - A silent loopback or LAN address fails its descriptor read after 3 seconds instead of 10. SSH forwards keep 10 seconds. - The server keeps idle HTTP connections for 95 seconds instead of Node's 5, so later requests reuse the connection the connect opened. A connection that finishes a response after the server stopped listening is closed, so the longer keep-alive cannot hold shutdown. Co-Authored-By: Claude Fable 5.2 --- .../server/src/httpResponseErrorGuard.test.ts | 33 ++++++ apps/server/src/httpResponseErrorGuard.ts | 19 +++ apps/server/src/server.ts | 3 +- .../src/authorization/layer.test.ts | 100 +++++++++++++--- .../src/authorization/service.ts | 110 ++++++++++++------ .../client-runtime/src/connection/driver.ts | 15 ++- .../src/connection/routes.test.ts | 29 +++++ .../client-runtime/src/connection/routes.ts | 38 +++++- 8 files changed, 285 insertions(+), 62 deletions(-) diff --git a/apps/server/src/httpResponseErrorGuard.test.ts b/apps/server/src/httpResponseErrorGuard.test.ts index ed3c7e65c7974..50451526a241e 100644 --- a/apps/server/src/httpResponseErrorGuard.test.ts +++ b/apps/server/src/httpResponseErrorGuard.test.ts @@ -125,6 +125,39 @@ describe("guardHttpResponseWriteErrors", () => { }); }); +describe("keepIdleConnections", () => { + it("closes a connection whose response finishes after the server stopped listening", async () => { + const server = HttpResponseErrorGuard.keepIdleConnections(NodeHttp.createServer()); + const body = "x".repeat(1_000_000); + const closed = Promise.withResolvers(); + server.on("request", (_request, response) => { + // The server shuts down while this request is in flight. + server.close(() => closed.resolve()); + response.end(body); + }); + const port = await listen(server); + const agent = new NodeHttp.Agent({ keepAlive: true }); + + const received = await new Promise((resolve, reject) => { + const request = NodeHttp.get({ host: "127.0.0.1", port, agent }, (response) => { + let bytes = 0; + response.on("data", (chunk: Buffer) => { + bytes += chunk.length; + }); + response.on("end", () => resolve(bytes)); + response.on("error", reject); + }); + requests.push(request); + request.on("error", reject); + }); + + expect(received).toBe(body.length); + // Left alone, the kept-alive connection holds the server open for the keep-alive time. + await closed.promise; + agent.destroy(); + }); +}); + describe("rejectRequestsUntilServing", () => { it("rejects early HTTP and upgrades, then hands each off to its serving handler", async () => { const writeErrors: unknown[] = []; diff --git a/apps/server/src/httpResponseErrorGuard.ts b/apps/server/src/httpResponseErrorGuard.ts index 0f215ba797f87..1650fd0677742 100644 --- a/apps/server/src/httpResponseErrorGuard.ts +++ b/apps/server/src/httpResponseErrorGuard.ts @@ -20,6 +20,25 @@ export function guardHttpResponseWriteErrors( return server; } +// cloudflared and Tailscale serve keep an idle connection to this server for 90 +// seconds. Node's 5 second default closes it first, so they can send a request +// into a closed connection, and a slow client reconnects in the middle of a connect. +const HTTP_KEEP_ALIVE_TIMEOUT_MS = 95_000; + +/** Install before `rejectRequestsUntilServing`, whose handoff starts at the next request listener. */ +export function keepIdleConnections(server: T): T { + server.keepAliveTimeout = HTTP_KEEP_ALIVE_TIMEOUT_MS; + server.headersTimeout = HTTP_KEEP_ALIVE_TIMEOUT_MS + 5_000; + // `close` drops only the connections idle at that moment. One that finishes + // its response later would hold shutdown for the whole keep-alive time. + server.on("request", (request, response) => { + response.once("finish", () => { + if (!server.listening) request.socket.end(); + }); + }); + return server; +} + /** * NodeHttpServer listens before serve() attaches its handlers, so requests in * between would hang. Install this after the write-error guards so only the diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 293d6670bbf0b..54f0dd2fae815 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -33,6 +33,7 @@ import { withUntracedRequests } from "./http.ts"; import * as ServerHttp from "./http.ts"; import { guardHttpResponseWriteErrors, + keepIdleConnections, rejectRequestsUntilServing, } from "./httpResponseErrorGuard.ts"; import { fixPath } from "./os-jank.ts"; @@ -353,7 +354,7 @@ const layerHttpServer = Layer.unwrap( return NodeHttpServer.layer( () => { const server = rejectRequestsUntilServing( - guardHttpResponseWriteErrors(NodeHttp.createServer()), + keepIdleConnections(guardHttpResponseWriteErrors(NodeHttp.createServer())), ); if (config.privateNetworkOnly) { server.on("connection", (socket) => { diff --git a/packages/client-runtime/src/authorization/layer.test.ts b/packages/client-runtime/src/authorization/layer.test.ts index 8c7cd57502b8b..771ba4b6aafbd 100644 --- a/packages/client-runtime/src/authorization/layer.test.ts +++ b/packages/client-runtime/src/authorization/layer.test.ts @@ -51,13 +51,16 @@ const BOOTSTRAP: RelayEnvironmentConnectResponse = { expiresAt: "2026-06-06T01:00:00.000Z", }; -type RecordedResponse = Response | ((init: RequestInit) => Response); +type RecordedResponse = Response | ((init: RequestInit) => Response | Promise); -function recordedFetch(responses: ReadonlyArray) { +function recordedFetch(responses: ReadonlyArray, descriptor?: object) { const calls: Array = []; let responseIndex = 0; const fetchFn = ((input, init) => { calls.push([input, init ?? {}]); + if (descriptor !== undefined && String(input).endsWith("/.well-known/t3/environment")) { + return Promise.resolve(Response.json(descriptor)); + } const response = responses[responseIndex++]; return response === undefined ? Promise.reject(new Error(`Unexpected fetch call to ${String(input)}`)) @@ -113,6 +116,8 @@ const persistedToken = ( const makeHarness = Effect.fn("TestRemoteAuthorization.makeHarness")(function* (input: { readonly initialToken?: TokenStore.RemoteDpopAccessToken; readonly responses: ReadonlyArray; + /** Answers every descriptor read, so one that runs beside a ticket does not shift `responses`. */ + readonly descriptor?: object; readonly bootstrap?: RelayEnvironmentConnectResponse; readonly beforeBootstrap?: Effect.Effect; readonly beforePut?: Effect.Effect; @@ -141,7 +146,7 @@ const makeHarness = Effect.fn("TestRemoteAuthorization.makeHarness")(function* ( readonly accessToken?: string; }> >([]); - const fetch = recordedFetch(input.responses); + const fetch = recordedFetch(input.responses, input.descriptor); const tokenStore = TokenStore.RemoteDpopAccessTokenStore.of({ get: (environmentId) => @@ -238,6 +243,9 @@ const makeHarness = Effect.fn("TestRemoteAuthorization.makeHarness")(function* ( }; }); +const requestsTo = (harness: Effect.Success>, path: string) => + harness.fetch.calls.filter(([url]) => String(url).endsWith(path)).length; + const prepareDirectBearer = Effect.fn("TestRemoteAuthorization.prepareDirectBearer")(function* ( harness: Effect.Success>, times: number, @@ -318,7 +326,7 @@ describe("RemoteEnvironmentAuthorization", () => { }), ); - it.effect("reuses a matching descriptor a check just read, once, while fresh, over HTTPS", () => + it.effect("reuses a matching descriptor a check just read, once, while fresh", () => Effect.gen(function* () { const descriptor = { ...DESCRIPTOR, @@ -327,10 +335,13 @@ describe("RemoteEnvironmentAuthorization", () => { const harness = yield* makeHarness({ responses: [ websocketTicket("reused"), - ...Array.from({ length: 5 }, () => [ + ...Array.from({ length: 4 }, () => [ Response.json(descriptor), websocketTicket("fresh"), ]).flat(), + websocketTicket("reused over plain HTTP"), + Response.json(descriptor), + websocketTicket("fresh"), ], }); @@ -346,7 +357,7 @@ describe("RemoteEnvironmentAuthorization", () => { }); const authorize = authorizeAt(ENDPOINT.httpBaseUrl); const remember = (value: Option.Option) => - remote.rememberCheckedDescriptor(ENDPOINT.httpBaseUrl, value); + remote.rememberCheckedDescriptor(ENDPOINT.httpBaseUrl, value, true); yield* remember(Option.some(descriptor)); yield* authorize; @@ -363,17 +374,31 @@ describe("RemoteEnvironmentAuthorization", () => { yield* remember(Option.some(descriptor)); yield* remember(Option.none()); yield* authorize; + // Plain HTTP is no reason to read again where the address keeps its host. + yield* remote.rememberCheckedDescriptor( + "http://100.101.102.103:3773", + Option.some(descriptor), + true, + ); + yield* authorizeAt("http://100.101.102.103:3773"); // After a network change, a plain LAN address can belong to another machine. - yield* remote.rememberCheckedDescriptor("http://192.168.1.2:3773", Option.some(descriptor)); + yield* remote.rememberCheckedDescriptor( + "http://192.168.1.2:3773", + Option.some(descriptor), + false, + ); yield* authorizeAt("http://192.168.1.2:3773"); }).pipe(Effect.provide(harness.layer)); expect(harness.fetch.calls.map(([url]) => new URL(String(url)).pathname)).toEqual([ "/api/auth/websocket-ticket/rpc", - ...Array.from({ length: 5 }, () => [ + ...Array.from({ length: 4 }, () => [ "/.well-known/t3/environment", "/api/auth/websocket-ticket/rpc", ]).flat(), + "/api/auth/websocket-ticket/rpc", + "/.well-known/t3/environment", + "/api/auth/websocket-ticket/rpc", ]); }), ); @@ -446,6 +471,44 @@ describe("RemoteEnvironmentAuthorization", () => { }), ); + it.effect("gives up on a silent loopback or LAN address sooner than on a remote one", () => + Effect.gen(function* () { + const silent = () => new Promise(() => {}); + const harness = yield* makeHarness({ responses: [silent, silent, silent] }); + + yield* Effect.gen(function* () { + const remote = yield* RemoteEnvironmentAuthorization.RemoteEnvironmentAuthorization; + const authorizeAt = (httpBaseUrl: string, connectionMethod: "direct" | "ssh" = "direct") => + remote + .authorizeBearer({ + expectedEnvironmentId: ENVIRONMENT_ID, + httpBaseUrl, + wsBaseUrl: httpBaseUrl.replace(/^http/, "ws"), + bearerToken: "bearer-token", + connectionMethod, + }) + .pipe(Effect.flip, Effect.forkChild); + const lan = yield* authorizeAt("http://192.168.1.2:3773"); + const tailnet = yield* authorizeAt("http://100.101.102.103:3773"); + // An SSH forward listens on loopback but reaches a remote host. + const forwarded = yield* authorizeAt("http://127.0.0.1:4010", "ssh"); + + yield* TestClock.adjust("3 seconds"); + expect(lan.pollUnsafe()).toBeDefined(); + expect(tailnet.pollUnsafe()).toBeUndefined(); + expect(forwarded.pollUnsafe()).toBeUndefined(); + expect(yield* Fiber.join(lan)).toMatchObject({ + _tag: "ConnectionTransientError", + reason: "timeout", + }); + + yield* TestClock.adjust("7 seconds"); + expect(tailnet.pollUnsafe()).toBeDefined(); + expect(forwarded.pollUnsafe()).toBeDefined(); + }).pipe(Effect.provide(harness.layer)); + }), + ); + it.effect("reuses a same-account persisted environment token without contacting the relay", () => Effect.gen(function* () { const cached = new TokenStore.RemoteDpopAccessToken({ @@ -459,6 +522,7 @@ describe("RemoteEnvironmentAuthorization", () => { }); const harness = yield* makeHarness({ initialToken: cached, + descriptor: DESCRIPTOR, responses: [websocketTicket("cached-ticket")], }); @@ -472,10 +536,12 @@ describe("RemoteEnvironmentAuthorization", () => { expect(authorized.socketUrl).toContain("wsTicket=cached-ticket"); expect(authorized.socketUrl).toContain("connectionMethod=relay"); expect(yield* Ref.get(harness.bootstrapCalls)).toBe(0); - expect(harness.fetch.calls).toHaveLength(1); - expect(String(harness.fetch.calls[0]?.[0])).toBe( + // The descriptor is read beside the ticket, so the connect needs no later read. + expect(authorized.descriptor).toMatchObject({ environmentId: ENVIRONMENT_ID }); + expect(harness.fetch.calls.map(([url]) => String(url)).toSorted()).toEqual([ + "https://environment.example.test/.well-known/t3/environment", "https://environment.example.test/api/auth/websocket-ticket/rpc", - ); + ]); }), ); @@ -576,6 +642,8 @@ describe("RemoteEnvironmentAuthorization", () => { }).pipe(Effect.provide(harness.layer)); expect(authorized.socketUrl).toContain("wsTicket=fresh-ticket"); + // The mint already read the descriptor. + expect(authorized.descriptor).toMatchObject({ environmentId: ENVIRONMENT_ID }); expect(yield* Ref.get(harness.bootstrapCalls)).toBe(1); expect((yield* Ref.get(harness.tokens)).get(ENVIRONMENT_ID)).toEqual( expect.objectContaining({ @@ -729,9 +797,9 @@ describe("RemoteEnvironmentAuthorization", () => { }); const harness = yield* makeHarness({ initialToken: cached, + descriptor: DESCRIPTOR, responses: [ authInvalid(), - Response.json(DESCRIPTOR), accessToken("replacement-access-token"), websocketTicket("replacement-ticket"), ], @@ -751,7 +819,8 @@ describe("RemoteEnvironmentAuthorization", () => { accessToken: "replacement-access-token", }), ); - expect(harness.fetch.calls).toHaveLength(4); + expect(requestsTo(harness, "/api/auth/websocket-ticket/rpc")).toBe(2); + expect(requestsTo(harness, "/oauth/token")).toBe(1); }), ); @@ -768,9 +837,9 @@ describe("RemoteEnvironmentAuthorization", () => { }); const harness = yield* makeHarness({ initialToken: cached, + descriptor: DESCRIPTOR, responses: [ new Response("endpoint unavailable", { status: 503 }), - Response.json(DESCRIPTOR), accessToken("replacement-access-token"), websocketTicket("replacement-ticket"), ], @@ -790,7 +859,8 @@ describe("RemoteEnvironmentAuthorization", () => { accessToken: "replacement-access-token", }), ); - expect(harness.fetch.calls).toHaveLength(4); + expect(requestsTo(harness, "/api/auth/websocket-ticket/rpc")).toBe(2); + expect(requestsTo(harness, "/oauth/token")).toBe(1); }), ); diff --git a/packages/client-runtime/src/authorization/service.ts b/packages/client-runtime/src/authorization/service.ts index c197c3a9b38d1..822dfcb7c8b63 100644 --- a/packages/client-runtime/src/authorization/service.ts +++ b/packages/client-runtime/src/authorization/service.ts @@ -22,6 +22,7 @@ import { ConnectionTransientError, type ConnectionAttemptError, } from "../connection/model.ts"; +import { isNearbyAddress } from "../connection/routes.ts"; import { fetchRemoteEnvironmentDescriptor } from "../environment/descriptor.ts"; import { environmentEndpointUrl } from "../environment/endpoint.ts"; import * as ClientCapabilities from "../platform/capabilities.ts"; @@ -82,10 +83,14 @@ export class RemoteEnvironmentAuthorization extends Context.Service< readonly expectedEnvironmentId: EnvironmentId; readonly rejectedAccessToken?: string; }) => Effect.Effect; - /** Lets the next direct authorization of this address skip its own descriptor read. */ + /** + * Lets the next direct authorization of this address skip its own descriptor + * read. Kept only where the address reaches the same machine on every network. + */ readonly rememberCheckedDescriptor: ( httpBaseUrl: string, descriptor: Option.Option, + keepsItsHost: boolean, ) => Effect.Effect; /** * Renews a live socket's token before it expires, so the server moves the @@ -101,6 +106,14 @@ export class RemoteEnvironmentAuthorization extends Context.Service< const CACHED_ENDPOINT_SOCKET_TIMEOUT_MS = 3_000; const DPOP_AUTHORIZATION_TIMEOUT_MS = 30_000; const CHECKED_DESCRIPTOR_TTL_NANOS = 5_000_000_000n; +// A loopback or LAN host answers well inside this, so a dead one frees its attempt sooner. +const NEARBY_DESCRIPTOR_TIMEOUT_MS = 3_000; + +/** A token with the descriptor its mint read. A cached token has none. */ +interface IssuedDpopToken { + readonly token: TokenStore.RemoteDpopAccessToken; + readonly descriptor?: ExecutionEnvironmentDescriptor; +} function mapDpopSocketError(error: RemoteEnvironmentAuthError | ConnectionAttemptError) { return error._tag === "ConnectionTransientError" || error._tag === "ConnectionBlockedError" @@ -111,10 +124,12 @@ function mapDpopSocketError(error: RemoteEnvironmentAuthError | ConnectionAttemp const fetchDescriptor = Effect.fn("clientRuntime.connection.remote.fetchDescriptor")(function* ( httpBaseUrl: string, connectionMethod: ClientConnectionMethod, + timeoutMs?: number, ) { - return yield* fetchRemoteEnvironmentDescriptor({ httpBaseUrl }).pipe( - Effect.mapError((error) => mapRemoteEnvironmentError(error, connectionMethod)), - ); + return yield* fetchRemoteEnvironmentDescriptor({ + httpBaseUrl, + ...(timeoutMs === undefined ? {} : { timeoutMs }), + }).pipe(Effect.mapError((error) => mapRemoteEnvironmentError(error, connectionMethod))); }); export const make = Effect.gen(function* () { @@ -137,12 +152,12 @@ export const make = Effect.gen(function* () { readonly id: object; readonly identity: ClientCapabilities.CloudSessionIdentity; readonly thumbprint: string; - readonly token: Effect.Effect; + readonly issued: Effect.Effect; } >(); // A route or host check hands its descriptor to the connect it leads to, once, - // so the connect skips a round trip. Only over HTTPS: a plain LAN address can + // so the connect skips a round trip. Not for a plain LAN address, which can // belong to another machine after a network change. const checkedDescriptors = new Map< string, @@ -167,9 +182,14 @@ export const make = Effect.gen(function* () { checked.descriptor.environmentId === input.expectedEnvironmentId && (yield* Clock.monotonicTimeNanos) - checked.at <= CHECKED_DESCRIPTOR_TTL_NANOS ? checked.descriptor - : yield* fetchDescriptor(input.httpBaseUrl, input.connectionMethod).pipe( - Effect.provideService(HttpClient.HttpClient, httpClient), - ); + : yield* fetchDescriptor( + input.httpBaseUrl, + input.connectionMethod, + // An SSH forward listens on loopback but reaches a remote host. + input.connectionMethod === "direct" && isNearbyAddress(input.httpBaseUrl) + ? NEARBY_DESCRIPTOR_TIMEOUT_MS + : undefined, + ).pipe(Effect.provideService(HttpClient.HttpClient, httpClient)); if (descriptor.environmentId !== input.expectedEnvironmentId) { return yield* environmentMismatchError({ expected: input.expectedEnvironmentId, @@ -352,7 +372,7 @@ export const make = Effect.gen(function* () { Effect.withSpan("environment.authorization.accessToken.exchange"), ); const issuedAt = yield* Clock.currentTimeMillis; - return new TokenStore.RemoteDpopAccessToken({ + const token = new TokenStore.RemoteDpopAccessToken({ environmentId: descriptor.environmentId, accountId: identity.accountId, label: descriptor.label, @@ -361,6 +381,7 @@ export const make = Effect.gen(function* () { expiresAtEpochMs: issuedAt + access.expires_in * 1_000, dpopThumbprint: thumbprint, }); + return { token, descriptor } satisfies IssuedDpopToken; }, ); @@ -403,14 +424,17 @@ export const make = Effect.gen(function* () { accessToken: cached.value.accessToken, identity, }); - return { token: Effect.succeed(cached.value), fromCache: true }; + return { + issued: Effect.succeed({ token: cached.value }), + fromCache: true, + }; } if (input.rejectedAccessToken !== undefined) { yield* removeRejectedToken(input.expectedEnvironmentId, input.rejectedAccessToken); } const pending = pendingTokens.get(input.expectedEnvironmentId); if (pending?.identity === identity && pending.thumbprint === thumbprint) { - return { token: pending.token, fromCache: false }; + return { issued: pending.issued, fromCache: false }; } const id = {}; @@ -419,9 +443,10 @@ export const make = Effect.gen(function* () { thumbprint, identity, ).pipe( - Effect.flatMap((token) => + Effect.flatMap((issued) => tokenLock.withPermits(1)( Effect.gen(function* () { + const { token } = issued; yield* assertSession(identity); yield* tokenStore .put(token) @@ -438,7 +463,7 @@ export const make = Effect.gen(function* () { if (pendingTokens.get(input.expectedEnvironmentId)?.id === id) { pendingTokens.delete(input.expectedEnvironmentId); } - return token; + return issued; }), ), ), @@ -464,17 +489,17 @@ export const make = Effect.gen(function* () { // One caller's timeout must not cancel renewal for other HTTP requests. Effect.forkIn(serviceScope), ); - const token = Fiber.join(fiber); - pendingTokens.set(input.expectedEnvironmentId, { id, identity, thumbprint, token }); - return { token, fromCache: false }; + const issued = Fiber.join(fiber); + pendingTokens.set(input.expectedEnvironmentId, { id, identity, thumbprint, issued }); + return { issued, fromCache: false }; }), ); yield* Effect.annotateCurrentSpan({ "connection.remote_token_cache": selected.fromCache ? "hit" : "miss", }); - const token = yield* selected.token; + const issued = yield* selected.issued; yield* assertSession(identity); - return { token, fromCache: selected.fromCache, identity }; + return { ...issued, fromCache: selected.fromCache, identity }; }); const httpAuthorization = ( @@ -495,7 +520,7 @@ export const make = Effect.gen(function* () { input: Parameters[0], ) { const endpoint = input.directEndpoint; - const descriptor = + const directDescriptor = endpoint === undefined ? undefined : yield* verifyDirectEndpoint({ @@ -503,21 +528,36 @@ export const make = Effect.gen(function* () { httpBaseUrl: endpoint.httpBaseUrl, connectionMethod: "direct", }); - const authorized = (token: TokenStore.RemoteDpopAccessToken, socketUrl: string) => ({ - ...httpAuthorization(token, endpoint?.httpBaseUrl), - socketUrl, - ...(descriptor === undefined ? {} : { descriptor }), + const authorized = ( + token: TokenStore.RemoteDpopAccessToken, + socketUrl: string, + descriptor: ExecutionEnvironmentDescriptor, + ) => ({ ...httpAuthorization(token, endpoint?.httpBaseUrl), socketUrl, descriptor }); + // A cached token comes without a descriptor. That read needs no ticket, so + // it runs beside the ticket request, and a failed ticket abandons it. + const openSocket = Effect.fnUntraced(function* (issued: IssuedDpopToken, timeoutMs?: number) { + const socket = createDpopSocketUrl(issued.token, timeoutMs, endpoint).pipe(Effect.result); + const known = directDescriptor ?? issued.descriptor; + if (known !== undefined) { + return Result.map(yield* socket, (socketUrl) => authorized(issued.token, socketUrl, known)); + } + const reading = yield* fetchDescriptor(issued.token.endpoint.httpBaseUrl, "relay").pipe( + Effect.provideService(HttpClient.HttpClient, httpClient), + Effect.forkChild, + ); + const opened = yield* socket; + if (Result.isFailure(opened)) { + yield* Fiber.interrupt(reading); + return Result.fail(opened.failure); + } + return Result.succeed(authorized(issued.token, opened.success, yield* Fiber.join(reading))); }); let selected = yield* getDpopToken(input); if (selected.fromCache) { - const cachedSocket = yield* createDpopSocketUrl( - selected.token, - CACHED_ENDPOINT_SOCKET_TIMEOUT_MS, - endpoint, - ).pipe(Effect.result); + const cachedSocket = yield* openSocket(selected, CACHED_ENDPOINT_SOCKET_TIMEOUT_MS); if (Result.isSuccess(cachedSocket)) { yield* assertSession(selected.identity); - return authorized(selected.token, cachedSocket.success); + return cachedSocket.success; } if (cachedSocket.failure._tag === "ConnectionBlockedError") { return yield* mapDpopSocketError(cachedSocket.failure); @@ -533,9 +573,7 @@ export const make = Effect.gen(function* () { rejectedAccessToken: selected.token.accessToken, }); } - const socket = yield* createDpopSocketUrl(selected.token, undefined, endpoint).pipe( - Effect.result, - ); + const socket = yield* openSocket(selected); if (Result.isFailure(socket)) { if (endpoint === undefined || socket.failure._tag === "ConnectionBlockedError") { yield* tokenLock.withPermits(1)( @@ -545,14 +583,14 @@ export const make = Effect.gen(function* () { return yield* mapDpopSocketError(socket.failure); } yield* assertSession(selected.identity); - return authorized(selected.token, socket.success); + return socket.success; }); return RemoteEnvironmentAuthorization.of({ authorizeBearer, // Untraced: it runs on every route and host check. - rememberCheckedDescriptor: Effect.fnUntraced(function* (httpBaseUrl, descriptor) { - if (Option.isNone(descriptor) || !httpBaseUrl.startsWith("https:")) { + rememberCheckedDescriptor: Effect.fnUntraced(function* (httpBaseUrl, descriptor, keepsItsHost) { + if (Option.isNone(descriptor) || !keepsItsHost) { checkedDescriptors.delete(httpBaseUrl); return; } diff --git a/packages/client-runtime/src/connection/driver.ts b/packages/client-runtime/src/connection/driver.ts index db253adea5333..512dba21e2d92 100644 --- a/packages/client-runtime/src/connection/driver.ts +++ b/packages/client-runtime/src/connection/driver.ts @@ -20,6 +20,7 @@ import { connectionRoutes, routeEntry, routeHttpBaseUrl, + routeKeepsItsHost, sendsTokenInClear, usableRoutes, } from "./routes.ts"; @@ -148,7 +149,7 @@ export const make = Effect.gen(function* () { // The descriptor is public, so this sends no credential to whatever // answers at a saved LAN address on a different network. - const checkDescriptor = (httpBaseUrl: string, environmentId: string) => + const checkDescriptor = (httpBaseUrl: string, environmentId: string, keepsItsHost: boolean) => fetchRemoteEnvironmentDescriptor({ httpBaseUrl, timeoutMs: ROUTE_CHECK_TIMEOUT_MS, @@ -156,7 +157,9 @@ export const make = Effect.gen(function* () { Effect.option, // The connect this check leads to skips its own read. Every answer, even a // failure, replaces what an older check remembered. - Effect.tap((descriptor) => remote.rememberCheckedDescriptor(httpBaseUrl, descriptor)), + Effect.tap((descriptor) => + remote.rememberCheckedDescriptor(httpBaseUrl, descriptor, keepsItsHost), + ), Effect.map((descriptor): RouteCheck => Option.exists(descriptor, (value) => value.environmentId === environmentId) ? "answered" @@ -170,7 +173,7 @@ export const make = Effect.gen(function* () { const checkRoute = (entry: ConnectionCatalogEntry, route: ConnectionRoute) => { const httpBaseUrl = routeHttpBaseUrl(route); if (httpBaseUrl === null) return Effect.succeed("unchecked"); - return checkDescriptor(httpBaseUrl, entry.target.environmentId).pipe( + return checkDescriptor(httpBaseUrl, entry.target.environmentId, routeKeepsItsHost(route)).pipe( Effect.withSpan("ConnectionDriver.checkRoute", { attributes: { "connection.target.kind": route.target._tag }, }), @@ -178,7 +181,11 @@ export const make = Effect.gen(function* () { }; const checkHost = (prepared: PreparedConnection) => - checkDescriptor(prepared.httpBaseUrl, prepared.environmentId).pipe( + checkDescriptor( + prepared.httpBaseUrl, + prepared.environmentId, + prepared.httpBaseUrl.startsWith("https:"), + ).pipe( Effect.withSpan("ConnectionDriver.checkHost", { attributes: { "connection.target.kind": prepared.target._tag }, }), diff --git a/packages/client-runtime/src/connection/routes.test.ts b/packages/client-runtime/src/connection/routes.test.ts index da9c631638cc4..e11aefe407e03 100644 --- a/packages/client-runtime/src/connection/routes.test.ts +++ b/packages/client-runtime/src/connection/routes.test.ts @@ -16,9 +16,11 @@ import { credentialConnectionId, insertRoute, entryWithRoutes, + isNearbyAddress, mergeLearnedRoutes, pendingRouteOrder, removedWithRelay, + routeKeepsItsHost, routesAfterRemoving, sendsTokenInClear, upsertRoute, @@ -77,6 +79,33 @@ describe("connection routes", () => { expect(connectionRouteLabel(RELAY)).toBe("T3 Connect"); }); + it("knows which addresses reach the same machine on every network", () => { + const tailnetIp = direct("ts-ip", "http://100.101.102.103:3773/"); + const profile = Option.getOrThrow(tailnetIp.profile) as BearerConnectionProfile; + const marked = { + target: tailnetIp.target, + profile: Option.some(new BearerConnectionProfile({ ...profile, network: "tailscale" })), + }; + expect(routeKeepsItsHost(PUBLIC)).toBe(true); + expect(routeKeepsItsHost(direct("lo", "http://127.0.0.1:3773/"))).toBe(true); + expect(routeKeepsItsHost(direct("ts", "http://desk.tail1234.ts.net:3773/"))).toBe(true); + expect(routeKeepsItsHost(marked)).toBe(true); + // 100.64.0.0/10 is shared with other VPNs unless the server reports it as its tailnet. + expect(routeKeepsItsHost(tailnetIp)).toBe(false); + expect(routeKeepsItsHost(LAN)).toBe(false); + expect(routeKeepsItsHost(direct("http", "http://desk.example.com/"))).toBe(false); + expect(routeKeepsItsHost(RELAY)).toBe(false); + }); + + it("treats only loopback and LAN addresses as nearby", () => { + expect(isNearbyAddress("http://127.0.0.1:3773/")).toBe(true); + expect(isNearbyAddress("http://192.168.1.10:3773/")).toBe(true); + expect(isNearbyAddress("http://100.101.102.103:3773/")).toBe(false); + expect(isNearbyAddress("https://desk.tail1234.ts.net/")).toBe(false); + expect(isNearbyAddress("https://desk.example.com/")).toBe(false); + expect(isNearbyAddress("not a url")).toBe(false); + }); + it("places a new route after faster kinds and ahead of T3 Connect", () => { expect(insertRoute([RELAY], LAN)).toEqual([LAN, RELAY]); expect(insertRoute([LAN, RELAY], TAILNET)).toEqual([LAN, TAILNET, RELAY]); diff --git a/packages/client-runtime/src/connection/routes.ts b/packages/client-runtime/src/connection/routes.ts index 9d83550a8c4a2..d856d75d8c5a1 100644 --- a/packages/client-runtime/src/connection/routes.ts +++ b/packages/client-runtime/src/connection/routes.ts @@ -81,6 +81,13 @@ function routeHostname(route: ConnectionRoute): string | null { } } +function directAddressKind(hostname: string, tailscale = false) { + if (isLocalLoopbackHost(hostname)) return "loopback"; + if (tailscale || isTailnetHost(hostname)) return "tailnet"; + if (isSharedAddressSpaceHost(hostname)) return "vpn"; + return isPrivateNetworkHost(hostname) ? "lan" : "public"; +} + export function connectionRouteKind(route: ConnectionRoute): ConnectionRouteKind { switch (route.target._tag) { case "RelayConnectionTarget": @@ -91,17 +98,36 @@ export function connectionRouteKind(route: ConnectionRoute): ConnectionRouteKind case "BearerConnectionTarget": { const hostname = routeHostname(route); if (hostname === null) return "public"; - if (isLocalLoopbackHost(hostname)) return "loopback"; const profile = Option.getOrNull(route.profile); - const tailscale = - profile?._tag === "BearerConnectionProfile" && profile.network === "tailscale"; - if (tailscale || isTailnetHost(hostname)) return "tailnet"; - if (isSharedAddressSpaceHost(hostname)) return "vpn"; - return isPrivateNetworkHost(hostname) ? "lan" : "public"; + return directAddressKind( + hostname, + profile?._tag === "BearerConnectionProfile" && profile.network === "tailscale", + ); } } } +/** + * Whether the route's address reaches the same machine on every network: HTTPS, + * loopback, or the tailnet. A plain LAN address can belong to another machine + * after a network change, so an answer from it is not kept. + */ +export function routeKeepsItsHost(route: ConnectionRoute): boolean { + if (routeHttpBaseUrl(route)?.startsWith("https:") === true) return true; + const kind = connectionRouteKind(route); + return kind === "loopback" || kind === "tailnet"; +} + +/** A loopback or LAN host answers within a few seconds or is not there. */ +export function isNearbyAddress(httpBaseUrl: string): boolean { + try { + const kind = directAddressKind(new URL(httpBaseUrl).hostname); + return kind === "loopback" || kind === "lan"; + } catch { + return false; + } +} + const ROUTE_KIND_RANK: Record = { loopback: 0, lan: 1, From 038aaab05a008761a59e9371ef87913bc969d255 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Thu, 12 Nov 2026 00:48:27 -0800 Subject: [PATCH 02/11] perf(shell): the thread list loads while the socket opens, and is not downloaded twice The shell HTTP snapshot waited for the socket handshake and the full server config. It now starts when the connection is prepared. The prepared connection keeps the public descriptor, which gives the server build and migration, and the descriptor now names the shell flags (shellThreadFamilies, shellOlderSettled). When the live config disagrees with the descriptor, the client reloads with the right flags and replaces the rows. A subscribe at the head sequence got a full socket snapshot when the head was below the pruned boundary (a database pruned by migrate-dev-db). A cursor at the head now always resumes. A client that subscribes from the HTTP snapshot it just loaded names the project identities of that snapshot, and the server does not send them again. The HTTP shell request names the cursor of held rows (resumeFrom). The server answers `resumable` in a few bytes when the socket can replay from it, so a cold start costs one shell request plus the subscribe. Servers whose descriptor does not name the shell flags keep the old flow. Co-Authored-By: Claude Fable 5.2 --- .../src/environment/ServerEnvironment.ts | 2 + .../src/orchestration-v2/ShellStream.ts | 79 ++++++ apps/server/src/orchestration-v2/http.test.ts | 10 + apps/server/src/orchestration-v2/http.ts | 39 ++- apps/server/src/ws.test.ts | 153 ++++++++--- apps/server/src/ws.ts | 55 ++-- docs/internals/connection-runtime.md | 11 + docs/internals/performance-regressions.md | 5 +- .../client-runtime/src/connection/model.ts | 4 +- .../src/connection/resolver.test.ts | 9 + .../client-runtime/src/connection/resolver.ts | 1 + .../src/state/environmentHttpAuth.test.ts | 6 +- .../src/state/shell-sync.test.ts | 127 ++++++++- packages/client-runtime/src/state/shell.ts | 247 +++++++++++++----- .../src/state/shellSnapshotHttp.ts | 68 +++-- packages/contracts/src/environment.ts | 6 + packages/contracts/src/environmentHttp.ts | 10 +- packages/contracts/src/orchestrationV2.ts | 21 ++ 18 files changed, 680 insertions(+), 173 deletions(-) diff --git a/apps/server/src/environment/ServerEnvironment.ts b/apps/server/src/environment/ServerEnvironment.ts index 0bc1387ea7c1b..1f56d49c88b04 100644 --- a/apps/server/src/environment/ServerEnvironment.ts +++ b/apps/server/src/environment/ServerEnvironment.ts @@ -291,6 +291,8 @@ export const make = Effect.gen(function* () { linkFence: true, mcpApprovalCodes: true, mcpHostedApproval: true, + shellThreadFamilies: true, + shellOlderSettled: true, }, }; diff --git a/apps/server/src/orchestration-v2/ShellStream.ts b/apps/server/src/orchestration-v2/ShellStream.ts index 0fa5251a7ef73..9a01294014eea 100644 --- a/apps/server/src/orchestration-v2/ShellStream.ts +++ b/apps/server/src/orchestration-v2/ShellStream.ts @@ -19,6 +19,8 @@ import { OrchestrationV2ThreadShell as ThreadShellSchema, } from "@t3tools/contracts"; import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient"; +// @effect-diagnostics-next-line nodeBuiltinImport:off - a synchronous digest of a few project rows. +import * as NodeCrypto from "node:crypto"; import type * as Arr from "effect/Array"; import type * as Cause from "effect/Cause"; import * as Clock from "effect/Clock"; @@ -31,11 +33,88 @@ import * as Schema from "effect/Schema"; import type * as SqlClient from "effect/sql/SqlClient"; import * as Stream from "effect/Stream"; +import type * as OrchestrationEventStore from "../persistence/OrchestrationEventStore.ts"; import { bufferLatestLiveStream } from "./LiveStreamBudget.ts"; import type { ActiveShellIndexEntry } from "./ProjectionStore.ts"; type ShellDelta = Extract; +// Replaying each intervening event costs a shell refetch; past this gap a +// single O(active-threads) snapshot is cheaper and bounded. +const SHELL_RESUME_MAX_GAP = 1_000; +// Row counts do not bound large tool payloads; sum stored JSON bytes in SQL and snapshot above budget. +const SHELL_REPLAY_PAYLOAD_BUDGET_BYTES = 8 * 1024 * 1024; + +/** + * Whether the events after a shell cursor can be replayed. The socket resume and the HTTP + * snapshot's `resumeFrom` ask the same question, so a client the route tells to resume is not + * refused by the socket. + */ +export const canResumeShellFrom = Effect.fnUntraced(function* ( + events: Pick< + OrchestrationEventStore.OrchestrationEventStore["Service"], + "prunedThroughSequence" | "getReplayStats" + >, + afterSequence: number, + headSequence: number, +) { + // A cursor at the head missed nothing, even where pruning has passed the head. + if (afterSequence === headSequence) return true; + const replayGap = headSequence - afterSequence; + if (replayGap < 0 || replayGap > SHELL_RESUME_MAX_GAP) return false; + if (afterSequence < (yield* events.prunedThroughSequence)) return false; + const stats = yield* events.getReplayStats({ afterSequence, throughSequence: headSequence }); + if (stats.rawPayloadBytes > SHELL_REPLAY_PAYLOAD_BUDGET_BYTES) { + yield* Effect.logDebug("orchestration replay replaced by snapshot", { + afterSequence, + headSequence, + replayGap, + eventCount: stats.eventCount, + payloadBytes: stats.rawPayloadBytes, + payloadBudgetBytes: SHELL_REPLAY_PAYLOAD_BUDGET_BYTES, + }); + return false; + } + return true; +}); + +/** + * Names the repository identities a list of project shells carries. A client sends back the one + * its HTTP snapshot had, and a resume that computes the same one skips its marked snapshot. + */ +export function projectIdentitiesDigest( + projects: ReadonlyArray>, +): string { + const hash = NodeCrypto.createHash("sha256"); + for (const project of projects) { + hash.update(JSON.stringify([project.id, project.repositoryIdentity ?? null])); + } + return hash.digest("base64url"); +} + +/** + * The resolved roots a resume sends. A client that names the current identities holds them + * already, but it keeps a cached identity over an absent one, so a root resolved to no identity + * still goes out. + */ +export function resumeIdentityRoots(input: { + readonly projects: ReadonlyArray< + Pick + >; + readonly resolvedRepositoryIdentityRoots: ReadonlyArray; + readonly knownProjectIdentities: string | undefined; +}): ReadonlyArray { + if (input.knownProjectIdentities !== projectIdentitiesDigest(input.projects)) { + return input.resolvedRepositoryIdentityRoots; + } + const withoutIdentity = new Set( + input.projects.flatMap((project) => + project.repositoryIdentity == null ? [project.workspaceRoot] : [], + ), + ); + return input.resolvedRepositoryIdentityRoots.filter((root) => withoutIdentity.has(root)); +} + /** Snapshots, synchronization, and enrichment markers must bypass full-state coalescing. */ export const bufferShellLiveStream = ( source: Stream.Stream, diff --git a/apps/server/src/orchestration-v2/http.test.ts b/apps/server/src/orchestration-v2/http.test.ts index 570d34468d37a..9d54a38675d84 100644 --- a/apps/server/src/orchestration-v2/http.test.ts +++ b/apps/server/src/orchestration-v2/http.test.ts @@ -279,6 +279,8 @@ it.effect("leaves older settled threads out of the HTTP shell only when asked", }), ); return (await response.json()) as { + readonly snapshotSequence: number; + readonly projectIdentities?: string; readonly threads: ReadonlyArray<{ readonly id: string }>; readonly olderSettled?: { readonly counts: ReadonlyArray }; }; @@ -292,5 +294,13 @@ it.effect("leaves older settled threads out of the HTTP shell only when asked", [thread.id, ...Array.from({ length: 25 }, (_, index) => `settled-${index + 1}`)].toSorted(), ); expect(cut.olderSettled?.counts).toEqual([{ projectId: thread.projectId, count: 1 }]); + + // A client that holds rows the socket can replay from keeps them; any other cursor reloads. + expect(full.projectIdentities).toBeTypeOf("string"); + expect(yield* shell(`?resumeFrom=${full.snapshotSequence}`)).toEqual({ resumable: true }); + expect(yield* shell(`?resumeFrom=${full.snapshotSequence - 1}`)).toEqual({ resumable: true }); + for (const cursor of [full.snapshotSequence + 1, "x"]) { + expect((yield* shell(`?omitOlderSettled=1&resumeFrom=${cursor}`)).threads).toHaveLength(26); + } }), ); diff --git a/apps/server/src/orchestration-v2/http.ts b/apps/server/src/orchestration-v2/http.ts index 3bba05f7db63a..f1c2d01779ad9 100644 --- a/apps/server/src/orchestration-v2/http.ts +++ b/apps/server/src/orchestration-v2/http.ts @@ -31,7 +31,11 @@ import { } from "./threadHistoryPaging.ts"; import * as ThreadManagementService from "./ThreadManagementService.ts"; import * as ProjectStore from "./ProjectStore.ts"; -import { loadSubscriberShellSnapshot } from "./ShellStream.ts"; +import { + canResumeShellFrom, + loadSubscriberShellSnapshot, + projectIdentitiesDigest, +} from "./ShellStream.ts"; import { boundedSnapshotResponseFields } from "./ThreadStream.ts"; import { projectThreadProjectionForWire } from "./WireProjection.ts"; @@ -94,7 +98,33 @@ export const layer = HttpApiBuilder.group( options, }); const projects = yield* enrichProjectShells(snapshot.projects); - return { ...snapshot, projects }; + return { ...snapshot, projects, projectIdentities: projectIdentitiesDigest(projects) }; + }, + ); + + // A client that holds rows names their cursor; a small answer tells it to keep them. + const answerShellSnapshot = Effect.fn("http.orchestration.answerShellSnapshot")( + function* (query: { + readonly omitIdleSubagents?: string; + readonly omitOlderSettled?: string; + readonly resumeFrom?: string; + }) { + const resumeFrom = query.resumeFrom === undefined ? Number.NaN : Number(query.resumeFrom); + if ( + Number.isSafeInteger(resumeFrom) && + resumeFrom >= 0 && + (yield* canResumeShellFrom( + applicationEvents, + resumeFrom, + yield* applicationEvents.latestApplicationSequence, + )) + ) { + return { resumable: true as const }; + } + return yield* loadShellSnapshot({ + omitIdleSubagents: query.omitIdleSubagents === "1", + omitOlderSettled: query.omitOlderSettled === "1", + }); }, ); @@ -166,10 +196,7 @@ export const layer = HttpApiBuilder.group( Effect.fn("environment.orchestration.shellSnapshot")(function* (args) { yield* annotateEnvironmentRequest(args.endpoint.name); yield* requireEnvironmentScope(AuthOrchestrationReadScope); - return yield* loadShellSnapshot({ - omitIdleSubagents: args.query.omitIdleSubagents === "1", - omitOlderSettled: args.query.omitOlderSettled === "1", - }).pipe( + return yield* answerShellSnapshot(args.query).pipe( traceLocalHandlerWork, Effect.catch((cause) => failEnvironmentInternal("orchestration_snapshot_failed", cause), diff --git a/apps/server/src/ws.test.ts b/apps/server/src/ws.test.ts index 2b64947dbc7df..0b731274369f9 100644 --- a/apps/server/src/ws.test.ts +++ b/apps/server/src/ws.test.ts @@ -149,6 +149,7 @@ import * as EventSink from "./orchestration-v2/EventSink.ts"; import * as LegacyV1ThreadImporter from "./orchestration-v2/legacy/LegacyV1ThreadImporter.ts"; import * as ThreadCommandExecutor from "./orchestration-v2/ThreadCommandExecutor.ts"; import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; +import { projectIdentitiesDigest } from "./orchestration-v2/ShellStream.ts"; import * as OrchestrationEventStore from "./persistence/OrchestrationEventStore.ts"; import * as SqlitePersistence from "./persistence/Sqlite.ts"; import * as ExternalLauncher from "./process/externalLauncher.ts"; @@ -1857,41 +1858,123 @@ it.effect.each(["thread", "shell"] as const)( }).pipe(Effect.scoped, Effect.provide(ReplayTestLayer)), ); -// A thread idle since before the boundary missed nothing; resuming must not resend its body. -it.effect("thread resume at its own high-water below the pruned boundary skips the snapshot", () => - Effect.gen(function* () { - const sql = yield* SqlClient.SqlClient; - const now = "2026-10-13T00:00:00.000Z"; - const threadId = ThreadId.make("thread:idle"); - const [last] = yield* sql<{ sequence: number }>` - INSERT INTO orchestration_events ( - event_id, aggregate_kind, stream_id, stream_version, event_type, - occurred_at, actor_kind, payload_json, metadata_json, application_event_version - ) VALUES ( - 'idle:0', 'thread', ${threadId}, 0, 'provider-session.detached', - ${now}, 'server', ${JSON.stringify({ providerSessionId: "session:idle", detachedAt: now })}, - '{}', 2 - ) RETURNING sequence - `; - yield* sql` - INSERT INTO orchestration_v2_projection_metadata ( - projection_name, schema_version, last_sequence, updated_at - ) VALUES (${OrchestrationEventStore.EVENT_LOG_PRUNED_THROUGH}, 2, ${last!.sequence + 10}, ${now}) - `; - const exit = yield* subscribeOrchestrationV2Thread({ - threadId, - afterSequence: last!.sequence, - }).pipe( - Effect.provideService(ThreadManagement.ThreadManagementService, { - ensureLegacyTranscript: () => Effect.void, - getThreadSnapshot: () => Effect.die("snapshot"), - streamStoredEventsFrom: () => Stream.empty, - } as unknown as ThreadManagement.ThreadManagementService["Service"]), - Effect.exit, - ); - // A snapshot load dies with "snapshot", so success means the resume replayed nothing. - assert.isTrue(Exit.isSuccess(exit)); - }).pipe(Effect.scoped, Effect.provide(ReplayTestLayer)), +// A cursor at the high-water missed nothing, even below the boundary; resuming must not resend +// a snapshot. The shell cases are a client that just loaded the HTTP snapshot of a pruned +// database: it gets the marker, and the project identities only when it does not name them. +it.effect.each([ + { scope: "thread", knowsIdentities: false, isRepository: true, kinds: [] }, + { + scope: "shell", + knowsIdentities: false, + isRepository: true, + kinds: ["snapshot", "synchronized"], + }, + { scope: "shell", knowsIdentities: true, isRepository: true, kinds: ["synchronized"] }, + // The client keeps a cached identity over an absent one, so "resolved to none" still goes out. + { + scope: "shell", + knowsIdentities: true, + isRepository: false, + kinds: ["snapshot", "synchronized"], + }, +] as const)( + "$scope resume at its own high-water below the pruned boundary skips the snapshot (knows identities: $knowsIdentities, repository: $isRepository)", + ({ scope, knowsIdentities, isRepository, kinds }) => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const now = "2026-10-13T00:00:00.000Z"; + const threadId = ThreadId.make("thread:idle"); + const [last] = yield* sql<{ sequence: number }>` + INSERT INTO orchestration_events ( + event_id, aggregate_kind, stream_id, stream_version, event_type, + occurred_at, actor_kind, payload_json, metadata_json, application_event_version + ) VALUES ( + 'idle:0', 'thread', ${threadId}, 0, 'provider-session.detached', + ${now}, 'server', ${JSON.stringify({ providerSessionId: "session:idle", detachedAt: now })}, + '{}', 2 + ) RETURNING sequence + `; + yield* sql` + INSERT INTO orchestration_v2_projection_metadata ( + projection_name, schema_version, last_sequence, updated_at + ) VALUES (${OrchestrationEventStore.EVENT_LOG_PRUNED_THROUGH}, 2, ${last!.sequence + 10}, ${now}) + `; + const identity = isRepository + ? ({ + canonicalKey: "github.com/t3/idle", + locator: { + source: "git-remote", + remoteName: "origin", + remoteUrl: "git@github.com:t3/idle", + }, + } as const) + : null; + const project = { + id: ProjectId.make("project:idle"), + title: "Idle", + workspaceRoot: "/tmp/idle", + defaultModelSelection: null, + scripts: [], + createdAt: now, + updatedAt: now, + } satisfies OrchestrationProjectShell; + const subscription = Effect.gen(function* () { + if (scope === "thread") { + return yield* subscribeOrchestrationV2Thread({ + threadId, + afterSequence: last!.sequence, + }).pipe(Effect.as([])); + } + const stream = yield* subscribeOrchestrationV2Shell({ + afterSequence: last!.sequence, + requestCompletionMarker: true, + ...(knowsIdentities + ? { + knownProjectIdentities: projectIdentitiesDigest([ + { ...project, repositoryIdentity: identity }, + ]), + } + : {}), + }); + return yield* stream.pipe( + Stream.takeUntil((item) => item.kind === "synchronized"), + Stream.map((item) => item.kind), + Stream.runCollect, + ); + }); + const exit = yield* subscription.pipe( + Effect.provideService(ThreadManagement.ThreadManagementService, { + ensureLegacyTranscript: () => Effect.void, + getThreadSnapshot: () => Effect.die("snapshot"), + listActiveShellIndex: () => Effect.die("snapshot"), + readShellSnapshot: () => Effect.die("snapshot"), + listActiveSubagentThreadIds: () => Effect.succeed([]), + streamStoredEventsFrom: () => Stream.empty, + } as unknown as ThreadManagement.ThreadManagementService["Service"]), + Effect.provideService(ProjectStore.ProjectStoreV2, { + listShells: () => Effect.succeed([project]), + } as unknown as ProjectStore.ProjectStoreV2["Service"]), + Effect.provideService( + ProjectService.ProjectService, + {} as ProjectService.ProjectService["Service"], + ), + Effect.provideService(ProjectEnrichment.ProjectEnrichmentService, { + subscribeChanges: PubSub.subscribe(yield* PubSub.unbounded()), + getAvailable: () => + Effect.succeed({ + repositoryIdentity: identity, + faviconPath: null, + repositoryIdentityResolved: true, + }), + } as unknown as ProjectEnrichment.ProjectEnrichmentService["Service"]), + Effect.exit, + ); + // A snapshot load dies with "snapshot", so success means the resume replayed nothing. + assert.deepEqual( + Exit.isFailure(exit) ? Cause.squash(exit.cause) : Array.from(exit.value), + kinds, + ); + }).pipe(Effect.scoped, Effect.provide(ReplayTestLayer)), ); const acpDriver = ProviderDriverKind.make("acpRegistry"); diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 5257da599c4f2..55bb03f6b7c80 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -133,6 +133,8 @@ import { shellStreamItemFromThreadShell, shellStreamItemsFromInitialSnapshot, loadSubscriberShellSnapshot, + canResumeShellFrom, + resumeIdentityRoots, shellStreamItemsFromResumeSnapshot, shouldPublishShellApplicationEvent, skipUnchangedThreadShells, @@ -678,12 +680,6 @@ const isOtherWsRpc = (rpc: ServerWsRpc): rpc is Exclude const serverWsRpcs = Array.from(ServerWsRpcGroup.requests.values()); const SplitWsRpcGroup = RpcGroup.make(...serverWsRpcs.filter(isSplitWsRpc)); const OtherWsRpcGroup = RpcGroup.make(...serverWsRpcs.filter(isOtherWsRpc)); -// Replaying each intervening event costs a shell refetch; past this gap a -// single O(active-threads) snapshot is cheaper and bounded. -const SHELL_RESUME_MAX_GAP = 1_000; -// Row counts do not bound large tool payloads; sum stored JSON bytes in SQL and snapshot above budget. -const ORCHESTRATION_REPLAY_PAYLOAD_BUDGET_BYTES = 8 * 1024 * 1024; - function toAuthAccessStreamEvent( change: PairingGrantStore.BootstrapCredentialChange | SessionStore.SessionCredentialChange, revision: number, @@ -820,36 +816,6 @@ function readClientOwners( ); } -const canReplayPersistedRange = Effect.fnUntraced(function* ( - afterSequence: number, - headSequence: number, - maxGap: number, -) { - const applicationEvents = yield* OrchestrationEventStore.OrchestrationEventStore; - - const replayGap = headSequence - afterSequence; - if (replayGap < 0 || replayGap > maxGap) { - return false; - } - if (afterSequence < (yield* applicationEvents.prunedThroughSequence)) return false; - const stats = yield* applicationEvents.getReplayStats({ - afterSequence, - throughSequence: headSequence, - }); - if (stats.rawPayloadBytes > ORCHESTRATION_REPLAY_PAYLOAD_BUDGET_BYTES) { - yield* Effect.logDebug("orchestration replay replaced by snapshot", { - afterSequence, - headSequence, - replayGap, - eventCount: stats.eventCount, - payloadBytes: stats.rawPayloadBytes, - payloadBudgetBytes: ORCHESTRATION_REPLAY_PAYLOAD_BUDGET_BYTES, - }); - return false; - } - return true; -}); - const enrichProjectShells = Effect.fn("ws.orchestrationV2.enrichProjectShells")( (projects: ReadonlyArray) => Effect.flatMap(ProjectEnrichmentService.ProjectEnrichmentService, (projectEnrichment) => @@ -1114,6 +1080,7 @@ export const subscribeOrchestrationV2Shell = Effect.fn("ws.orchestrationV2.subsc function* (input: { readonly afterSequence?: number; readonly httpSnapshotFallback?: true; + readonly knownProjectIdentities?: string; readonly requestCompletionMarker?: boolean; readonly omitIdleSubagents?: boolean; readonly omitOlderSettled?: boolean; @@ -1339,7 +1306,7 @@ export const subscribeOrchestrationV2Shell = Effect.fn("ws.orchestrationV2.subsc }); }); const highWater = yield* applicationEvents.latestApplicationSequence; - if (!(yield* canReplayPersistedRange(input.afterSequence, highWater, SHELL_RESUME_MAX_GAP))) { + if (!(yield* canResumeShellFrom(applicationEvents, input.afterSequence, highWater))) { return yield* fromSnapshot; } @@ -1358,12 +1325,22 @@ export const subscribeOrchestrationV2Shell = Effect.fn("ws.orchestrationV2.subsc skipUnknownEventTypes: true, }) .pipe(Stream.filter(filterSubagentTranscript), Stream.runCollect); - if (input.afterSequence < (yield* applicationEvents.prunedThroughSequence)) { + if ( + input.afterSequence < highWater && + input.afterSequence < (yield* applicationEvents.prunedThroughSequence) + ) { return yield* fromSnapshot; } const replay = toShellStream(Stream.fromIterable(replayed)); return composeShellStreamWithEnrichment({ - initial: initialEnrichmentItems(loaded), + initial: initialEnrichmentItems({ + snapshot: loaded.snapshot, + resolvedRepositoryIdentityRoots: resumeIdentityRoots({ + projects: loaded.snapshot.projects, + resolvedRepositoryIdentityRoots: loaded.resolvedRepositoryIdentityRoots, + knownProjectIdentities: input.knownProjectIdentities, + }), + }), tail: Stream.concat(Stream.concat(replay, completionMarker), liveFrom(highWater)), enrichment: enrichmentRefreshes, }); diff --git a/docs/internals/connection-runtime.md b/docs/internals/connection-runtime.md index 698ddac7a73f9..94a599be81e8d 100644 --- a/docs/internals/connection-runtime.md +++ b/docs/internals/connection-runtime.md @@ -75,6 +75,17 @@ connection nor overwrite newer live data during a reconnect. Loading and resuming snapshots belongs to the shared state services, so every view agrees on which data is current. +The shell snapshot loads over HTTP, which needs only the prepared connection. +When the server's public descriptor names its shell flags, the +[shell state](../../packages/client-runtime/src/state/shell.ts) asks for the +shell as soon as the connection is prepared, while the socket opens and the +config downloads. The request names held rows that may resume (`resumeFrom`), and +the server answers `resumable` instead of a snapshot when the socket can replay +from them, so a cold start costs one shell request plus the subscription. The +live config stays the authority: rows loaded for a build or with omissions it +does not confirm are replaced, never resumed. A server whose descriptor does not +name the flags is asked only after its config arrives. + Shell snapshots can leave rows out when the client asks. Servers that advertise `shellThreadFamilies` leave out idle subagent children: a child stays while it has a live or queued run, a pending request or background work, or a subagent diff --git a/docs/internals/performance-regressions.md b/docs/internals/performance-regressions.md index e641933b313fc..3fb1ff93d8893 100644 --- a/docs/internals/performance-regressions.md +++ b/docs/internals/performance-regressions.md @@ -52,7 +52,10 @@ The v2 checks pin these invariants: - The initial shell contains active navigation rows only. Archived rows use the dedicated archive query, and transcript message bodies stay in thread detail regardless of message size. - Shell resume sends deltas plus compact repository-enrichment metadata, not another full project - and thread snapshot. + and thread snapshot. A cursor at the head sequence always resumes, even below the pruned + boundary, so a client that just loaded the HTTP snapshot never downloads it again over the + socket. That client also names the snapshot's project identities, and a server whose + identities did not change since resends only the roots that resolved to no identity. - Auto, steer, and restart sends resolve delivery from authoritative state inside the server's per-thread dispatch lock. Model selection and identified-checkpoint rollback also dispatch without first fetching a full thread projection when the server advertises support. diff --git a/packages/client-runtime/src/connection/model.ts b/packages/client-runtime/src/connection/model.ts index 73675d431aa7a..feeb0c25a6909 100644 --- a/packages/client-runtime/src/connection/model.ts +++ b/packages/client-runtime/src/connection/model.ts @@ -1,4 +1,4 @@ -import { EnvironmentId } from "@t3tools/contracts"; +import { EnvironmentId, type ExecutionEnvironmentDescriptor } from "@t3tools/contracts"; import { RelayEnvironmentEndpointFailure } from "@t3tools/contracts/relay"; import * as Schema from "effect/Schema"; @@ -154,6 +154,8 @@ export interface PreparedConnection { readonly socketUrl: string; readonly httpAuthorization: PreparedHttpAuthorization | null; readonly target: ConnectionTarget; + /** What the server said about itself while this connection was prepared. */ + readonly descriptor?: ExecutionEnvironmentDescriptor; } export type SupervisorConnectionPhase = diff --git a/packages/client-runtime/src/connection/resolver.test.ts b/packages/client-runtime/src/connection/resolver.test.ts index 62d43bbbe641a..df59003227e1b 100644 --- a/packages/client-runtime/src/connection/resolver.test.ts +++ b/packages/client-runtime/src/connection/resolver.test.ts @@ -206,6 +206,15 @@ describe("ConnectionResolver", () => { "ws://127.0.0.1:3777/ws?clientSurface=web&clientDeviceType=desktop&connectionMethod=direct&orchestrationProtocol=2&clientCapabilities=settingsSecretReferences", httpAuthorization: null, target, + // The shell reads the server's build and shell flags here before the socket opens. + descriptor: { + environmentId: ENVIRONMENT_ID, + label: "Compatible environment", + platform: { os: "linux", arch: "x64" }, + serverVersion: "0.0.0-test", + orchestrationProtocolVersion: ORCHESTRATION_PROTOCOL_VERSION, + capabilities: { repositoryIdentity: true }, + }, }); }), ); diff --git a/packages/client-runtime/src/connection/resolver.ts b/packages/client-runtime/src/connection/resolver.ts index d3b22143d327a..05235ef53603f 100644 --- a/packages/client-runtime/src/connection/resolver.ts +++ b/packages/client-runtime/src/connection/resolver.ts @@ -321,6 +321,7 @@ export const make = Effect.gen(function* () { } return { ...prepared, + descriptor, socketUrl: appendOrchestrationProtocol(prepared.socketUrl), }; }); diff --git a/packages/client-runtime/src/state/environmentHttpAuth.test.ts b/packages/client-runtime/src/state/environmentHttpAuth.test.ts index faac1a87ff6af..3839742694e81 100644 --- a/packages/client-runtime/src/state/environmentHttpAuth.test.ts +++ b/packages/client-runtime/src/state/environmentHttpAuth.test.ts @@ -223,7 +223,11 @@ const LOADERS: ReadonlyArray<{ expected: SHELL, load: (input: HttpInput) => fetchEnvironmentShellSnapshot(input).pipe( - Effect.map(({ loadPullRequests: _links, ...snapshot }) => snapshot), + Effect.map((answer) => { + if ("resumable" in answer) return answer; + const { loadPullRequests: _links, ...snapshot } = answer; + return snapshot; + }), ), }, { diff --git a/packages/client-runtime/src/state/shell-sync.test.ts b/packages/client-runtime/src/state/shell-sync.test.ts index 062db03809cfc..d127c14c7de8f 100644 --- a/packages/client-runtime/src/state/shell-sync.test.ts +++ b/packages/client-runtime/src/state/shell-sync.test.ts @@ -1,5 +1,6 @@ import { EnvironmentId, + type ExecutionEnvironmentDescriptor, ORCHESTRATION_V2_WS_METHODS, type OrchestrationV2ShellSnapshot, type OrchestrationV2ShellStreamItem, @@ -1151,8 +1152,22 @@ describe("environment shell cold start", () => { }; const RECENT_ORIGIN = { ...CURRENT_ORIGIN, snapshotAt: NOW - HOUR }; + // A server that names its shell flags in the descriptor, and the matching live config. + const FAMILY_DESCRIPTOR = { + serverVersion: "1.0.0", + capabilities: { shellThreadFamilies: true, shellOlderSettled: true }, + } as ExecutionEnvironmentDescriptor; + const FAMILY_CONFIG = { + shellThreadFamilies: true, + shellOlderSettled: true, + environment: { serverVersion: "1.0.0" }, + }; + const FAMILY_FLAGS = { omitIdleSubagents: true, omitOlderSettled: true }; + const makeColdStart = Effect.fn("test.makeColdStartShell")(function* ( origin: Persistence.ShellCacheOrigin | undefined, + // Set for a connection that is prepared but has no session yet. + early?: { readonly serverResumes: boolean }, ) { yield* TestClock.setTime(NOW); const subscriptions = yield* Queue.unbounded<{ @@ -1164,6 +1179,7 @@ describe("environment shell cold start", () => { readonly origin: Persistence.ShellCacheOrigin | undefined; }>(); const httpLoads = yield* Ref.make(0); + const loadOptions = yield* Queue.unbounded(); const client = { [ORCHESTRATION_V2_WS_METHODS.subscribeShell]: (input: unknown) => Stream.unwrap( @@ -1174,12 +1190,18 @@ describe("environment shell cold start", () => { }), ), } as unknown as WsRpcProtocolClient; - const activeSession = yield* SubscriptionRef.make(Option.some(session(client))); + const activeSession = yield* SubscriptionRef.make( + early === undefined ? Option.some(session(client)) : Option.none(), + ); const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({ target: TARGET, state: yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE), session: activeSession, - prepared: yield* SubscriptionRef.make(Option.some(PREPARED)), + prepared: yield* SubscriptionRef.make( + Option.some( + early === undefined ? PREPARED : { ...PREPARED, descriptor: FAMILY_DESCRIPTOR }, + ), + ), connect: Effect.void, disconnect: Effect.void, retryNow: Effect.void, @@ -1205,10 +1227,19 @@ describe("environment shell cold start", () => { clear: () => Effect.void, }); const snapshotLoader = ShellSnapshotLoader.ShellSnapshotLoader.of({ - load: () => - Ref.updateAndGet(httpLoads, (count) => count + 1).pipe( + load: (_prepared, options = {}) => + Queue.offer(loadOptions, options).pipe( + Effect.andThen(Ref.updateAndGet(httpLoads, (count) => count + 1)), Effect.map((count) => - Option.some({ ...v2ShellSnapshot, snapshotSequence: HTTP_SEQUENCE + count - 1 }), + Option.some( + early?.serverResumes === true && options.resumeFrom !== undefined + ? { resumable: true } + : { + ...v2ShellSnapshot, + snapshotSequence: HTTP_SEQUENCE + count - 1, + projectIdentities: `identities-${count}`, + }, + ), ), ), }); @@ -1218,9 +1249,16 @@ describe("environment shell cold start", () => { Effect.provideService(ShellSnapshotLoader.ShellSnapshotLoader, snapshotLoader), ); return { + shellState, subscriptions, saved, httpLoads, + loadOptions, + connect: (config: object) => + SubscriptionRef.set( + activeSession, + Option.some({ ...session(client), initialConfig: Effect.succeed(config as never) }), + ), reconnect: SubscriptionRef.set(activeSession, Option.some(session(client))), liveAt: (sequence: number) => SubscriptionRef.changes(shellState).pipe( @@ -1284,6 +1322,85 @@ describe("environment shell cold start", () => { }), ); + it.effect.each([ + { + label: "no rows to resume", + origin: undefined, + serverResumes: false, + config: FAMILY_CONFIG, + loads: [FAMILY_FLAGS], + subscribes: { afterSequence: HTTP_SEQUENCE, knownProjectIdentities: "identities-1" }, + }, + { + label: "rows the server can replay from", + origin: { ...RECENT_ORIGIN, ...FAMILY_FLAGS }, + serverResumes: true, + config: FAMILY_CONFIG, + loads: [{ ...FAMILY_FLAGS, resumeFrom: CACHED_SEQUENCE }], + subscribes: { afterSequence: CACHED_SEQUENCE, httpSnapshotFallback: true }, + }, + { + label: "rows the server cannot replay from", + origin: { ...RECENT_ORIGIN, ...FAMILY_FLAGS }, + serverResumes: false, + config: FAMILY_CONFIG, + loads: [{ ...FAMILY_FLAGS, resumeFrom: CACHED_SEQUENCE }], + subscribes: { afterSequence: HTTP_SEQUENCE, knownProjectIdentities: "identities-1" }, + }, + { + // The rows loaded with the descriptor's omissions are replaced, never kept. + label: "a live config with other omissions", + origin: undefined, + serverResumes: false, + config: { environment: { serverVersion: "1.0.0" } }, + loads: [FAMILY_FLAGS, { omitIdleSubagents: false, omitOlderSettled: false }], + subscribes: { afterSequence: HTTP_SEQUENCE + 1, knownProjectIdentities: "identities-2" }, + }, + ])( + "asks for the shell once the connection is prepared, with $label", + ({ origin, serverResumes, config, loads, subscribes }) => + Effect.gen(function* () { + const h = yield* makeColdStart(origin, { serverResumes }); + expect(yield* Queue.take(h.loadOptions)).toEqual(loads[0]); + if (!serverResumes) { + // The rows are on screen before a socket exists, and are not live yet. + const early = yield* SubscriptionRef.changes(h.shellState).pipe( + Stream.filter((state) => + Option.exists( + state.snapshot, + (snapshot) => snapshot.snapshotSequence === HTTP_SEQUENCE, + ), + ), + Stream.runHead, + Effect.map(Option.getOrThrow), + ); + expect(early.status).not.toBe("live"); + expect(early.generation).toBeGreaterThan(0); + } + expect(yield* Queue.size(h.subscriptions)).toBe(0); + + yield* h.connect(config); + const first = yield* Queue.take(h.subscriptions); + expect(first.input).toMatchObject(subscribes); + for (const key of ["httpSnapshotFallback", "knownProjectIdentities"] as const) { + if (!(key in subscribes)) expect(first.input).not.toHaveProperty(key); + } + expect(yield* Queue.clear(h.loadOptions)).toEqual(loads.slice(1)); + expect(yield* Ref.get(h.httpLoads)).toBe(loads.length); + + yield* Queue.offer(first.events, { kind: "synchronized" }); + yield* h.liveAt(subscribes.afterSequence); + if (loads.length > 1) { + yield* TestClock.adjust("500 millis"); + const saves = yield* Queue.takeAll(h.saved); + expect(saves.at(-1)).toMatchObject({ + snapshot: { snapshotSequence: subscribes.afterSequence }, + origin: { ...CURRENT_ORIGIN, snapshotAt: NOW }, + }); + } + }), + ); + it.effect("reloads over HTTP when the server answers a resume with snapshot-required", () => Effect.gen(function* () { const h = yield* makeColdStart(RECENT_ORIGIN); diff --git a/packages/client-runtime/src/state/shell.ts b/packages/client-runtime/src/state/shell.ts index 808fd9fdece56..9c452642df094 100644 --- a/packages/client-runtime/src/state/shell.ts +++ b/packages/client-runtime/src/state/shell.ts @@ -1,6 +1,7 @@ import { ORCHESTRATION_V2_WS_METHODS, type EnvironmentId, + type ExecutionEnvironmentDescriptor, type OrchestrationV2ShellSnapshot, type OrchestrationV2ShellStreamItem, type OrchestrationV2ThreadShell, @@ -11,6 +12,7 @@ import { import * as Cause from "effect/Cause"; import * as Clock from "effect/Clock"; import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; import * as Fiber from "effect/Fiber"; import * as Option from "effect/Option"; import * as Queue from "effect/Queue"; @@ -18,10 +20,11 @@ import * as Ref from "effect/Ref"; import * as Semaphore from "effect/Semaphore"; import * as Stream from "effect/Stream"; import * as SubscriptionRef from "effect/SubscriptionRef"; +import * as SynchronizedRef from "effect/SynchronizedRef"; import { AsyncResult, Atom } from "effect/reactivity"; import * as EnvironmentRegistry from "../connection/registry.ts"; -import { connectionProjectionPhase } from "../connection/model.ts"; +import { connectionProjectionPhase, type PreparedConnection } from "../connection/model.ts"; import * as EnvironmentSupervisor from "../connection/supervisor.ts"; import * as ConnectionWakeups from "../connection/wakeups.ts"; import { safeErrorLogAttributes } from "../errors/safeLog.ts"; @@ -97,6 +100,37 @@ interface ShellSubscription { readonly identity: ShellOriginIdentity | null; } +function shellIdentity( + environment: Pick, + flags: { readonly shellThreadFamilies?: boolean; readonly shellOlderSettled?: boolean }, +): ShellOriginIdentity { + const omitIdleSubagents = flags.shellThreadFamilies === true; + return { + serverVersion: environment.serverVersion, + // Migrations rewrite projections without events, so a migrated server reloads. + databaseMigration: environment.databaseMigration ?? null, + omitIdleSubagents, + omitOlderSettled: flags.shellOlderSettled === true && omitIdleSubagents, + }; +} + +/** Null for a server whose descriptor does not name its shell flags; only its config does. */ +function descriptorShellIdentity( + descriptor: ExecutionEnvironmentDescriptor | undefined, +): ShellOriginIdentity | null { + if (descriptor?.capabilities.shellThreadFamilies === undefined) return null; + return shellIdentity(descriptor, descriptor.capabilities); +} + +function sameShellIdentity(left: ShellOriginIdentity, right: ShellOriginIdentity): boolean { + return ( + left.serverVersion === right.serverVersion && + left.databaseMigration === right.databaseMigration && + left.omitIdleSubagents === right.omitIdleSubagents && + left.omitOlderSettled === right.omitOlderSettled + ); +} + function canResumeFrom( origin: Persistence.ShellCacheOrigin | undefined, identity: ShellOriginIdentity | null, @@ -104,14 +138,21 @@ function canResumeFrom( ): boolean { if (origin === undefined || identity === null) return false; const age = now - origin.snapshotAt; - return ( - origin.serverVersion === identity.serverVersion && - origin.databaseMigration === identity.databaseMigration && - origin.omitIdleSubagents === identity.omitIdleSubagents && - origin.omitOlderSettled === identity.omitOlderSettled && - age >= 0 && - age < CACHE_RESUME_MAX_AGE_MILLIS - ); + return sameShellIdentity(origin, identity) && age >= 0 && age < CACHE_RESUME_MAX_AGE_MILLIS; +} + +type HttpShellOutcome = + | { readonly kind: "snapshot"; readonly projectIdentities: string | undefined } + /** The server can replay from the rows the client holds. */ + | { readonly kind: "resumable" } + | { readonly kind: "failed" }; + +interface EarlyShellLoad { + readonly prepared: PreparedConnection; + readonly identity: ShellOriginIdentity; + readonly fiber: Fiber.Fiber; + // One subscription uses a load; a later one on the same connection decides again. + readonly claimed: boolean; } interface LiveShell { @@ -249,17 +290,17 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") // time, so neither can overwrite what the other wrote in between. const snapshotWriteLock = yield* Semaphore.make(1); - const applyItems = Effect.fn("EnvironmentShellState.applyItems")(function* ( - entries: ReadonlyArray, + const applyLocked = Effect.fnUntraced(function* ( + items: ReadonlyArray, + /** What an authoritative snapshot among the items descends from. */ + identity: ShellOriginIdentity | null, + /** Set for an HTTP snapshot: the rows that still wait for their pull request links. */ pendingLinkIds?: ReadonlySet, ) { - // Read under the lock: items buffered across a session change must not land on the - // rows the next session loaded. - const subscription = yield* Ref.get(activeSubscription); - const items = entries.flatMap(([session, item]) => - session === subscription?.session ? [item] : [], - ); const initial = yield* SubscriptionRef.get(state); + // No socket has confirmed an HTTP snapshot, so it never makes the shell live. + const fromHttp = pendingLinkIds !== undefined; + const disconnected = initial.status === "empty" || initial.status === "cached"; let waiting = yield* Ref.get(awaitingCompletion); let next = initial; let receivedAuthoritativeSnapshot = false; @@ -330,7 +371,8 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") if (authoritative) receivedAuthoritativeSnapshot = true; next = { snapshot: Option.some(nextSnapshot), - status: waiting ? "synchronizing" : "live", + status: + fromHttp && disconnected ? "cached" : waiting || fromHttp ? "synchronizing" : "live", error: Option.none(), generation: authoritative ? ++lastShellGeneration : next.generation, removedAt: authoritative ? EMPTY_REMOVALS : next.removedAt, @@ -347,9 +389,7 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") yield* Queue.offer(reloadRequests, undefined); } else if (receivedAuthoritativeSnapshot) { nextOrigin = - subscription?.identity == null - ? undefined - : { ...subscription.identity, snapshotAt: yield* Clock.currentTimeMillis }; + identity === null ? undefined : { ...identity, snapshotAt: yield* Clock.currentTimeMillis }; yield* Ref.set(origin, nextOrigin); } if (next === initial) return; @@ -368,6 +408,18 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") if (next.snapshot !== initial.snapshot && Option.isSome(next.snapshot)) { yield* Queue.offer(persistence, next.snapshot.value); } + }); + + const applyItems = Effect.fn("EnvironmentShellState.applyItems")(function* ( + entries: ReadonlyArray, + ) { + // Read under the lock: items buffered across a session change must not land on the + // rows the next session loaded. + const subscription = yield* Ref.get(activeSubscription); + yield* applyLocked( + entries.flatMap(([session, item]) => (session === subscription?.session ? [item] : [])), + subscription?.identity ?? null, + ); }, snapshotWriteLock.withPermit); const stateScope = yield* Effect.scope; @@ -437,6 +489,85 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") yield* fillDeferredPullRequests(cached.value, cachedSnapshot.value); } + const applyHttpSnapshot = Effect.fnUntraced(function* ( + loaded: ShellSnapshotLoader.ShellHttpSnapshot, + identity: ShellOriginIdentity | null, + ) { + const { projectIdentities: _identities, ...deferred } = loaded; + yield* cancelActiveFill; + const previous = yield* SubscriptionRef.get(state); + const snapshot = reuseUnchangedThreadShells( + Option.getOrNull(previous.snapshot), + withoutDeferredPullRequests(deferred), + ); + const pendingLinkIds = new Set( + deferred.loadPullRequests === undefined ? [] : deferred.threads.map((thread) => thread.id), + ); + yield* snapshotWriteLock.withPermit( + applyLocked([{ kind: "snapshot", snapshot }], identity, pendingLinkIds), + ); + const current = yield* SubscriptionRef.get(state); + if (Option.isSome(current.snapshot)) { + yield* fillDeferredPullRequests(deferred, current.snapshot.value); + } + }, Effect.uninterruptible); + + /** + * One request to the HTTP shell route. Held rows that may resume are named, so the server + * answers that they still can, in a few bytes, or sends the snapshot that replaces them. + */ + const loadOverHttp = Effect.fn("EnvironmentShellState.loadOverHttp")(function* ( + prepared: PreparedConnection, + identity: ShellOriginIdentity | null, + ) { + const held = yield* SubscriptionRef.get(state); + const resumeFrom = + Option.isSome(held.snapshot) && + canResumeFrom(yield* Ref.get(origin), identity, yield* Clock.currentTimeMillis) + ? held.snapshot.value.snapshotSequence + : undefined; + // HTTP and the socket must agree, or omitted children would return through the socket. + const answer = yield* snapshotLoader.load(prepared, { + omitIdleSubagents: identity?.omitIdleSubagents === true, + omitOlderSettled: identity?.omitOlderSettled === true, + ...(resumeFrom === undefined ? {} : { resumeFrom }), + }); + if (Option.isNone(answer)) return { kind: "failed" } satisfies HttpShellOutcome; + if ("resumable" in answer.value) return { kind: "resumable" } satisfies HttpShellOutcome; + yield* applyHttpSnapshot(answer.value, identity); + return { + kind: "snapshot", + projectIdentities: answer.value.projectIdentities, + } satisfies HttpShellOutcome; + }); + + // A server whose descriptor names its shell flags is asked for the shell as soon as the + // connection is prepared, while the socket opens and the config downloads. The subscription + // claims that load, and reloads when the live config disagrees with the descriptor. + const earlyLoad = yield* SynchronizedRef.make(null); + const useEarlyLoad = (prepared: PreparedConnection | null, claim: boolean) => + SynchronizedRef.modifyEffect( + earlyLoad, + Effect.fnUntraced(function* (current) { + if (current !== null && current.prepared === prepared) { + const next = claim ? { ...current, claimed: true } : current; + return [current.claimed ? null : current, next] as const; + } + if (current !== null) yield* Fiber.interrupt(current.fiber); + const identity = descriptorShellIdentity(prepared?.descriptor); + if (prepared === null || identity === null) return [null, null] as const; + // Items a closed session still has in flight must not land on the rows this loads. + yield* Ref.set(activeSubscription, null); + const fiber = yield* loadOverHttp(prepared, identity).pipe(Effect.forkIn(stateScope)); + const next: EarlyShellLoad = { prepared, identity, fiber, claimed: claim }; + return [next, next] as const; + }, Effect.uninterruptible), + ); + yield* SubscriptionRef.changes(supervisor.prepared).pipe( + Stream.runForEach((prepared) => useEarlyLoad(Option.getOrNull(prepared), false)), + Effect.forkScoped, + ); + const resubscriptions = Stream.merge( Stream.fromQueue(reloadRequests), Option.match(wakeups, { @@ -452,22 +583,28 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") ORCHESTRATION_V2_WS_METHODS.subscribeShell, Effect.fn("EnvironmentShellState.makeSubscribeInput")(function* (session) { const identity = yield* session.initialConfig.pipe( - Effect.map((config): ShellOriginIdentity => { - const omitIdleSubagents = config.shellThreadFamilies === true; - return { - serverVersion: config.environment.serverVersion, - // Migrations rewrite projections without events, so a migrated server reloads. - databaseMigration: config.environment.databaseMigration ?? null, - omitIdleSubagents, - omitOlderSettled: config.shellOlderSettled === true && omitIdleSubagents, - }; - }), + Effect.map((config) => shellIdentity(config.environment, config)), Effect.orElseSucceed(() => null), ); const { omitIdleSubagents, omitOlderSettled } = identity ?? { omitIdleSubagents: false, omitOlderSettled: false, }; + const early = yield* useEarlyLoad( + Option.getOrNull(yield* SubscriptionRef.get(supervisor.prepared)), + true, + ); + const earlyExit = early === null ? null : yield* Fiber.await(early.fiber); + // Rows the descriptor's flags or build loaded are not kept when the live config differs: + // their origin then cannot resume, so the load below replaces them. + let loaded = + early !== null && + earlyExit !== null && + Exit.isSuccess(earlyExit) && + identity !== null && + sameShellIdentity(early.identity, identity) + ? earlyExit.value + : null; yield* Ref.set(activeSubscription, { session, identity }); yield* Ref.set(awaitingCompletion, true); yield* setSynchronizing; @@ -475,14 +612,12 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") // Rows from a recent server snapshot with the same omissions and server build resume // from their cursor, in memory or from the cache. Anything else reloads the HTTP // snapshot so a valid cursor cannot preserve stale or incomplete data. - let current = yield* SubscriptionRef.get(state); - let canResume = - Option.isSome(current.snapshot) && - canResumeFrom(yield* Ref.get(origin), identity, yield* Clock.currentTimeMillis); - // Only rows the client already held may be refused; a cursor from the snapshot just - // loaded falls back to the socket snapshot, so a refusal cannot loop. - const resumingHeldRows = canResume; - if (!canResume) { + const resumingHeldRows = + loaded?.kind === "resumable" || + (loaded?.kind !== "snapshot" && + Option.isSome((yield* SubscriptionRef.get(state)).snapshot) && + canResumeFrom(yield* Ref.get(origin), identity, yield* Clock.currentTimeMillis)); + if (!resumingHeldRows && loaded === null) { const prepared = yield* SubscriptionRef.get(supervisor.prepared).pipe( Effect.flatMap( Option.match({ @@ -497,42 +632,26 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") }), ), ); - // HTTP and the socket must agree, or omitted children would return through the socket. - const httpSnapshot = yield* snapshotLoader.load(prepared, { - omitIdleSubagents, - omitOlderSettled, - }); - if (Option.isSome(httpSnapshot)) { - yield* cancelActiveFill; - const previous = yield* SubscriptionRef.get(state); - const snapshot = reuseUnchangedThreadShells( - Option.getOrNull(previous.snapshot), - withoutDeferredPullRequests(httpSnapshot.value), - ); - const pendingLinkIds = new Set( - httpSnapshot.value.loadPullRequests === undefined - ? [] - : httpSnapshot.value.threads.map((thread) => thread.id), - ); - yield* applyItems([[session, { kind: "snapshot", snapshot }]], pendingLinkIds); - canResume = true; - current = yield* SubscriptionRef.get(state); - if (Option.isSome(current.snapshot)) { - yield* fillDeferredPullRequests(httpSnapshot.value, current.snapshot.value); - } - } + loaded = yield* loadOverHttp(prepared, identity); } + const current = yield* SubscriptionRef.get(state); const omission = { ...(omitIdleSubagents ? { omitIdleSubagents: true as const } : {}), ...(omitOlderSettled ? { omitOlderSettled: true as const } : {}), }; - if (!canResume || Option.isNone(current.snapshot)) { + if (Option.isNone(current.snapshot) || !(resumingHeldRows || loaded?.kind === "snapshot")) { return { ...omission, requestCompletionMarker: true as const }; } return { afterSequence: current.snapshot.value.snapshotSequence, + // Only rows the client already held may be refused; a cursor from the snapshot just + // loaded falls back to the socket snapshot, so a refusal cannot loop. ...(resumingHeldRows ? { httpSnapshotFallback: true as const } : {}), + // No stream item has touched the snapshot just loaded, so its identities still hold. + ...(loaded?.kind === "snapshot" && loaded.projectIdentities !== undefined + ? { knownProjectIdentities: loaded.projectIdentities } + : {}), ...omission, requestCompletionMarker: true as const, }; diff --git a/packages/client-runtime/src/state/shellSnapshotHttp.ts b/packages/client-runtime/src/state/shellSnapshotHttp.ts index 9ff944ad6b2fe..e98846a0e3c73 100644 --- a/packages/client-runtime/src/state/shellSnapshotHttp.ts +++ b/packages/client-runtime/src/state/shellSnapshotHttp.ts @@ -1,9 +1,10 @@ -import { EnvironmentHttpCommonError } from "@t3tools/contracts"; +import { EnvironmentHttpCommonError, OrchestrationV2ShellResumable } from "@t3tools/contracts"; import * as Cause from "effect/Cause"; import * as Context from "effect/Context"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; +import * as Predicate from "effect/Predicate"; import * as Schema from "effect/Schema"; import { HttpClient, HttpClientError, type HttpClientResponse } from "effect/http"; @@ -19,6 +20,20 @@ import { type DeferredShellSnapshot, } from "./shellPullRequests.ts"; +export interface ShellSnapshotOptions { + readonly omitIdleSubagents?: boolean; + readonly omitOlderSettled?: boolean; + /** The cursor of held rows loaded with these omissions; the server may answer `resumable`. */ + readonly resumeFrom?: number; +} + +/** `projectIdentities` names the repository identities the snapshot carried. */ +export interface ShellHttpSnapshot extends DeferredShellSnapshot { + readonly projectIdentities?: string; +} + +export type ShellHttpAnswer = ShellHttpSnapshot | OrchestrationV2ShellResumable; + // On timeout the socket asks the same server for the same full snapshot, so a // short deadline only throws the first build away. The socket fallback is for // setups where /api fails but /ws works, such as a proxy that blocks /api. @@ -26,20 +41,21 @@ const DEFAULT_SHELL_SNAPSHOT_TIMEOUT_MS = 20_000; export const fetchEnvironmentShellSnapshot = Effect.fn( "clientRuntime.state.fetchEnvironmentShellSnapshot", -)(function* (input: { - readonly prepared: PreparedConnection; - readonly signer: Option.Option; - readonly remoteAuthorization?: Option.Option< - RemoteEnvironmentAuthorization.RemoteEnvironmentAuthorization["Service"] - >; - readonly timeoutMs?: number; - readonly omitIdleSubagents?: boolean; - readonly omitOlderSettled?: boolean; -}) { +)(function* ( + input: { + readonly prepared: PreparedConnection; + readonly signer: Option.Option; + readonly remoteAuthorization?: Option.Option< + RemoteEnvironmentAuthorization.RemoteEnvironmentAuthorization["Service"] + >; + readonly timeoutMs?: number; + } & ShellSnapshotOptions, +) { const endpoint = { query: { ...(input.omitIdleSubagents === true ? { omitIdleSubagents: "1" } : {}), ...(input.omitOlderSettled === true ? { omitOlderSettled: "1" } : {}), + ...(input.resumeFrom === undefined ? {} : { resumeFrom: String(input.resumeFrom) }), }, }; return yield* executeAuthenticatedEnvironmentHttpRequest({ @@ -63,14 +79,29 @@ const decodeEnvironmentHttpCommonError = Schema.decodeUnknownEffect( Schema.toCodecJson(EnvironmentHttpCommonError), ); +const isShellResumable = Schema.is(OrchestrationV2ShellResumable); + const decodeShellSnapshotResponse = ( response: HttpClientResponse.HttpClientResponse, ): Effect.Effect< - DeferredShellSnapshot, + ShellHttpAnswer, HttpClientError.HttpClientError | Schema.SchemaError | EnvironmentHttpCommonError > => { if (response.status === 200) { - return response.json.pipe(Effect.flatMap(decodeShellSnapshotDeferringPullRequests)); + return response.json.pipe( + Effect.flatMap((json): Effect.Effect => { + if (isShellResumable(json)) return Effect.succeed(json); + const projectIdentities = + Predicate.isObject(json) && typeof json.projectIdentities === "string" + ? json.projectIdentities + : undefined; + return decodeShellSnapshotDeferringPullRequests(json).pipe( + Effect.map((snapshot) => + projectIdentities === undefined ? snapshot : { ...snapshot, projectIdentities }, + ), + ); + }), + ); } const statusError = new HttpClientError.HttpClientError({ reason: new HttpClientError.StatusCodeError({ request: response.request, response }), @@ -89,8 +120,8 @@ export class ShellSnapshotLoader extends Context.Service< { readonly load: ( prepared: PreparedConnection, - options?: { readonly omitIdleSubagents?: boolean; readonly omitOlderSettled?: boolean }, - ) => Effect.Effect>; + options?: ShellSnapshotOptions, + ) => Effect.Effect>; } >()("@t3tools/client-runtime/state/shellSnapshotHttp/ShellSnapshotLoader") {} @@ -110,17 +141,16 @@ export const layer: Layer.Layer), + Effect.map(Option.some), Effect.provideService(HttpClient.HttpClient, httpClient), Effect.catchCause((cause) => Effect.logWarning( "Could not load the environment shell snapshot over HTTP; using the socket snapshot instead.", ).pipe( Effect.annotateLogs({ cause: Cause.pretty(cause) }), - Effect.as(Option.none()), + Effect.as(Option.none()), ), ), ); diff --git a/packages/contracts/src/environment.ts b/packages/contracts/src/environment.ts index 5e9cca70752b6..45a8e7119b1e3 100644 --- a/packages/contracts/src/environment.ts +++ b/packages/contracts/src/environment.ts @@ -184,6 +184,12 @@ export const ExecutionEnvironmentCapabilities = Schema.Struct({ /** Server can transcribe 16 kHz mono PCM audio on the environment machine. */ voiceTranscription: Schema.optionalKey(Schema.Boolean), cloudVoiceTranscription: Schema.optionalKey(Schema.Boolean), + /** + * The server config's flags of the same names. A server that names them here lets a client + * load the shell snapshot before the socket opens, and answers its `resumeFrom`. + */ + shellThreadFamilies: Schema.optionalKey(Schema.Boolean), + shellOlderSettled: Schema.optionalKey(Schema.Boolean), }); export type ExecutionEnvironmentCapabilities = typeof ExecutionEnvironmentCapabilities.Type; diff --git a/packages/contracts/src/environmentHttp.ts b/packages/contracts/src/environmentHttp.ts index 4bbae77fb922d..03d1cf34a396e 100644 --- a/packages/contracts/src/environmentHttp.ts +++ b/packages/contracts/src/environmentHttp.ts @@ -67,7 +67,8 @@ import { } from "./speech.ts"; import { SecretReference } from "./secretReference.ts"; import { - OrchestrationV2ShellSnapshot, + OrchestrationV2ShellHttpSnapshot, + OrchestrationV2ShellResumable, OrchestrationV2ThreadBoundedSnapshot, OrchestrationV2ThreadDetailSnapshot, OrchestrationV2ThreadHistoryPage, @@ -685,6 +686,11 @@ const EnvironmentOrchestrationThreadSnapshotParams = Schema.Struct({ const EnvironmentOrchestrationShellSnapshotQuery = Schema.Struct({ omitIdleSubagents: Schema.optionalKey(Schema.String), omitOlderSettled: Schema.optionalKey(Schema.String), + /** + * The cursor of rows the client holds, loaded with the same omissions. The server answers + * `resumable` instead of a snapshot when the socket can replay from it. + */ + resumeFrom: Schema.optionalKey(Schema.String), }); const EnvironmentOrchestrationThreadBoundedSnapshotQuery = Schema.Struct({ @@ -714,7 +720,7 @@ class EnvironmentOrchestrationHttpApi extends HttpApiGroup.make("orchestration") HttpApiEndpoint.get("shellSnapshot", "/api/orchestration/shell", { headers: OrchestrationProtocolHeaders, query: EnvironmentOrchestrationShellSnapshotQuery, - success: OrchestrationV2ShellSnapshot, + success: [OrchestrationV2ShellHttpSnapshot, OrchestrationV2ShellResumable], error: EnvironmentOrchestrationSnapshotErrors, }).middleware(EnvironmentAuthenticatedAuth), ) diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 8d8cc42c235b2..1434a65c37bc3 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -2276,6 +2276,22 @@ export const OrchestrationV2ShellSnapshot = Schema.Struct({ }); export type OrchestrationV2ShellSnapshot = typeof OrchestrationV2ShellSnapshot.Type; +/** The HTTP shell snapshot. `projectIdentities` names its repository identities; see `knownProjectIdentities`. */ +export const OrchestrationV2ShellHttpSnapshot = Schema.Struct({ + ...OrchestrationV2ShellSnapshot.fields, + projectIdentities: Schema.optionalKey(Schema.String), +}); +export type OrchestrationV2ShellHttpSnapshot = typeof OrchestrationV2ShellHttpSnapshot.Type; + +/** + * The HTTP shell route's answer to `resumeFrom` when the socket can replay from that cursor: + * the client keeps its rows and subscribes from the cursor. + */ +export const OrchestrationV2ShellResumable = Schema.Struct({ + resumable: Schema.Literal(true), +}); +export type OrchestrationV2ShellResumable = typeof OrchestrationV2ShellResumable.Type; + export const OrchestrationV2ShellStreamItem = Schema.Union([ Schema.Struct({ kind: Schema.Literal("synchronized"), @@ -3762,6 +3778,11 @@ export const OrchestrationV2SubscribeShellInput = Schema.Struct({ * a socket snapshot, which clients without transport compression download over gzip HTTP. */ httpSnapshotFallback: Schema.optionalKey(Schema.Literal(true)), + /** + * The `projectIdentities` of the HTTP snapshot whose cursor is `afterSequence`, when the client + * changed nothing since. A resume that still has the same identities does not resend them. + */ + knownProjectIdentities: Schema.optionalKey(Schema.String), requestCompletionMarker: Schema.optionalKey(Schema.Boolean), /** * Snapshots leave out idle subagent children; read them with `getThreadFamily`. Send only to From fc640a0e2cc4e312597684ad5789d5851b0c37d3 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Thu, 12 Nov 2026 02:04:03 -0800 Subject: [PATCH 03/11] fix(client): a rotated credential no longer stops thread list updates Review findings: - The registry swaps a rotated bearer into the prepared connection of a live session. The prepare-time shell load took that for a new connection and dropped the session's events. It now keeps the subscription. - At shutdown, a socket with pipelined requests ends after its last response, not its first. - The resume-or-snapshot decision moves from the HTTP handler into ShellStream. Co-Authored-By: Claude Fable 5.2 --- .../server/src/httpResponseErrorGuard.test.ts | 38 +++++++++++++++++++ apps/server/src/httpResponseErrorGuard.ts | 12 ++++-- .../src/orchestration-v2/ShellStream.ts | 25 ++++++++++++ apps/server/src/orchestration-v2/http.ts | 37 +++++++----------- .../src/state/shell-sync.test.ts | 28 +++++++++++--- packages/client-runtime/src/state/shell.ts | 10 ++++- 6 files changed, 117 insertions(+), 33 deletions(-) diff --git a/apps/server/src/httpResponseErrorGuard.test.ts b/apps/server/src/httpResponseErrorGuard.test.ts index 50451526a241e..b5e52aa4de1a7 100644 --- a/apps/server/src/httpResponseErrorGuard.test.ts +++ b/apps/server/src/httpResponseErrorGuard.test.ts @@ -156,6 +156,44 @@ describe("keepIdleConnections", () => { await closed.promise; agent.destroy(); }); + + it("keeps a connection until every response on it has finished", async () => { + const server = HttpResponseErrorGuard.keepIdleConnections(NodeHttp.createServer()); + const closed = Promise.withResolvers(); + let first: NodeHttp.ServerResponse | undefined; + server.on("request", (request, response) => { + if (request.url === "/first") { + first = response; + return; + } + // Both pipelined requests are in flight when the server shuts down. + server.close(() => closed.resolve()); + first?.once("finish", () => setImmediate(() => response.end("second-body"))); + first?.end("first-body"); + }); + const port = await listen(server); + + const received = await new Promise((resolve, reject) => { + let data = ""; + const client = NodeNet.connect(port, "127.0.0.1", () => { + client.write( + ["/first", "/second"] + .map((path) => `GET ${path} HTTP/1.1\r\nHost: 127.0.0.1\r\n\r\n`) + .join(""), + ); + }); + sockets.push(client); + client.on("data", (chunk: Buffer) => { + data += chunk.toString(); + }); + client.on("end", () => resolve(data)); + client.on("error", reject); + }); + + expect(received).toContain("first-body"); + expect(received).toContain("second-body"); + await closed.promise; + }); }); describe("rejectRequestsUntilServing", () => { diff --git a/apps/server/src/httpResponseErrorGuard.ts b/apps/server/src/httpResponseErrorGuard.ts index 1650fd0677742..6f0ac75dbc2a8 100644 --- a/apps/server/src/httpResponseErrorGuard.ts +++ b/apps/server/src/httpResponseErrorGuard.ts @@ -30,10 +30,16 @@ export function keepIdleConnections(server: T): T { server.keepAliveTimeout = HTTP_KEEP_ALIVE_TIMEOUT_MS; server.headersTimeout = HTTP_KEEP_ALIVE_TIMEOUT_MS + 5_000; // `close` drops only the connections idle at that moment. One that finishes - // its response later would hold shutdown for the whole keep-alive time. + // its responses later would hold shutdown for the whole keep-alive time. + const responsesInFlight = new WeakMap(); server.on("request", (request, response) => { - response.once("finish", () => { - if (!server.listening) request.socket.end(); + const socket = request.socket; + responsesInFlight.set(socket, (responsesInFlight.get(socket) ?? 0) + 1); + response.once("close", () => { + const left = (responsesInFlight.get(socket) ?? 1) - 1; + responsesInFlight.set(socket, left); + // Pipelined requests share the socket, so the last response ends it. + if (left === 0 && !server.listening) socket.end(); }); }); return server; diff --git a/apps/server/src/orchestration-v2/ShellStream.ts b/apps/server/src/orchestration-v2/ShellStream.ts index 9a01294014eea..cb25bfe3dab3d 100644 --- a/apps/server/src/orchestration-v2/ShellStream.ts +++ b/apps/server/src/orchestration-v2/ShellStream.ts @@ -78,6 +78,31 @@ export const canResumeShellFrom = Effect.fnUntraced(function* ( return true; }); +/** + * Answers a client that asks for the shell with the cursor of the rows it holds: `resumable` + * when the socket can replay from that cursor, so the client keeps its rows, else the snapshot. + */ +export const loadShellUnlessResumable = Effect.fnUntraced(function* (input: { + readonly events: Pick< + OrchestrationEventStore.OrchestrationEventStore["Service"], + "prunedThroughSequence" | "getReplayStats" | "latestApplicationSequence" + >; + readonly resumeFrom: number | undefined; + readonly loadSnapshot: Effect.Effect; +}) { + if ( + input.resumeFrom !== undefined && + (yield* canResumeShellFrom( + input.events, + input.resumeFrom, + yield* input.events.latestApplicationSequence, + )) + ) { + return { resumable: true as const }; + } + return yield* input.loadSnapshot; +}); + /** * Names the repository identities a list of project shells carries. A client sends back the one * its HTTP snapshot had, and a resume that computes the same one skips its marked snapshot. diff --git a/apps/server/src/orchestration-v2/http.ts b/apps/server/src/orchestration-v2/http.ts index f1c2d01779ad9..b58ce1b143119 100644 --- a/apps/server/src/orchestration-v2/http.ts +++ b/apps/server/src/orchestration-v2/http.ts @@ -32,7 +32,7 @@ import { import * as ThreadManagementService from "./ThreadManagementService.ts"; import * as ProjectStore from "./ProjectStore.ts"; import { - canResumeShellFrom, + loadShellUnlessResumable, loadSubscriberShellSnapshot, projectIdentitiesDigest, } from "./ShellStream.ts"; @@ -103,30 +103,21 @@ export const layer = HttpApiBuilder.group( ); // A client that holds rows names their cursor; a small answer tells it to keep them. - const answerShellSnapshot = Effect.fn("http.orchestration.answerShellSnapshot")( - function* (query: { - readonly omitIdleSubagents?: string; - readonly omitOlderSettled?: string; - readonly resumeFrom?: string; - }) { - const resumeFrom = query.resumeFrom === undefined ? Number.NaN : Number(query.resumeFrom); - if ( - Number.isSafeInteger(resumeFrom) && - resumeFrom >= 0 && - (yield* canResumeShellFrom( - applicationEvents, - resumeFrom, - yield* applicationEvents.latestApplicationSequence, - )) - ) { - return { resumable: true as const }; - } - return yield* loadShellSnapshot({ + const answerShellSnapshot = Effect.fn("http.orchestration.answerShellSnapshot")((query: { + readonly omitIdleSubagents?: string; + readonly omitOlderSettled?: string; + readonly resumeFrom?: string; + }) => { + const resumeFrom = query.resumeFrom === undefined ? Number.NaN : Number(query.resumeFrom); + return loadShellUnlessResumable({ + events: applicationEvents, + resumeFrom: Number.isSafeInteger(resumeFrom) && resumeFrom >= 0 ? resumeFrom : undefined, + loadSnapshot: loadShellSnapshot({ omitIdleSubagents: query.omitIdleSubagents === "1", omitOlderSettled: query.omitOlderSettled === "1", - }); - }, - ); + }), + }); + }); const loadThreadSnapshot = Effect.fn("http.orchestration.loadThreadSnapshot")(function* ( threadId: Parameters[0], diff --git a/packages/client-runtime/src/state/shell-sync.test.ts b/packages/client-runtime/src/state/shell-sync.test.ts index d127c14c7de8f..2c1d57e711b97 100644 --- a/packages/client-runtime/src/state/shell-sync.test.ts +++ b/packages/client-runtime/src/state/shell-sync.test.ts @@ -1193,15 +1193,14 @@ describe("environment shell cold start", () => { const activeSession = yield* SubscriptionRef.make( early === undefined ? Option.some(session(client)) : Option.none(), ); + const prepared = yield* SubscriptionRef.make( + Option.some(early === undefined ? PREPARED : { ...PREPARED, descriptor: FAMILY_DESCRIPTOR }), + ); const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({ target: TARGET, state: yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE), session: activeSession, - prepared: yield* SubscriptionRef.make( - Option.some( - early === undefined ? PREPARED : { ...PREPARED, descriptor: FAMILY_DESCRIPTOR }, - ), - ), + prepared, connect: Effect.void, disconnect: Effect.void, retryNow: Effect.void, @@ -1260,6 +1259,14 @@ describe("environment shell cold start", () => { Option.some({ ...session(client), initialConfig: Effect.succeed(config as never) }), ), reconnect: SubscriptionRef.set(activeSession, Option.some(session(client))), + // The registry swaps a rotated bearer into the prepared connection of a live session. + rotateCredential: SubscriptionRef.update( + prepared, + Option.map((current) => ({ + ...current, + httpAuthorization: { _tag: "Bearer" as const, token: "rotated" }, + })), + ), liveAt: (sequence: number) => SubscriptionRef.changes(shellState).pipe( Stream.filter( @@ -1398,6 +1405,17 @@ describe("environment shell cold start", () => { origin: { ...CURRENT_ORIGIN, snapshotAt: NOW }, }); } + + // A rotated credential keeps the session, so its events still apply and nothing reloads. + yield* h.rotateCredential; + yield* Queue.offer(first.events, { + kind: "thread.removed", + sequence: subscribes.afterSequence + 1, + location: "active", + threadId: ThreadId.make("no-such-thread"), + }); + yield* h.liveAt(subscribes.afterSequence + 1); + expect(yield* Ref.get(h.httpLoads)).toBe(loads.length); }), ); diff --git a/packages/client-runtime/src/state/shell.ts b/packages/client-runtime/src/state/shell.ts index 9c452642df094..d4b9b01aa7953 100644 --- a/packages/client-runtime/src/state/shell.ts +++ b/packages/client-runtime/src/state/shell.ts @@ -115,6 +115,11 @@ function shellIdentity( } /** Null for a server whose descriptor does not name its shell flags; only its config does. */ +function sameConnection(left: PreparedConnection, right: PreparedConnection): boolean { + const keys = Object.keys({ ...left, ...right }) as Array; + return keys.every((key) => key === "httpAuthorization" || left[key] === right[key]); +} + function descriptorShellIdentity( descriptor: ExecutionEnvironmentDescriptor | undefined, ): ShellOriginIdentity | null { @@ -549,8 +554,9 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") SynchronizedRef.modifyEffect( earlyLoad, Effect.fnUntraced(function* (current) { - if (current !== null && current.prepared === prepared) { - const next = claim ? { ...current, claimed: true } : current; + // A rotated credential replaces the prepared connection of a session that stays up. + if (current !== null && prepared !== null && sameConnection(current.prepared, prepared)) { + const next = { ...current, prepared, claimed: current.claimed || claim }; return [current.claimed ? null : current, next] as const; } if (current !== null) yield* Fiber.interrupt(current.fiber); From 866d3ae2189c8fabe15b64d2cc55784515230693 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Thu, 12 Nov 2026 02:19:57 -0800 Subject: [PATCH 04/11] fix(client): a second subscription waits for the pending thread list load A resubscription while the prepare-time load was still running started a second request. If that one answered first, the older answer replaced its rows under a cursor the socket had already passed. A claimed load that still runs is now handed out again, so there is one request. Co-Authored-By: Claude Fable 5.2 --- .../src/state/shell-sync.test.ts | 38 +++++++++++++++++-- packages/client-runtime/src/state/shell.ts | 5 ++- 2 files changed, 39 insertions(+), 4 deletions(-) diff --git a/packages/client-runtime/src/state/shell-sync.test.ts b/packages/client-runtime/src/state/shell-sync.test.ts index 2c1d57e711b97..87da5deef7127 100644 --- a/packages/client-runtime/src/state/shell-sync.test.ts +++ b/packages/client-runtime/src/state/shell-sync.test.ts @@ -1167,7 +1167,7 @@ describe("environment shell cold start", () => { const makeColdStart = Effect.fn("test.makeColdStartShell")(function* ( origin: Persistence.ShellCacheOrigin | undefined, // Set for a connection that is prepared but has no session yet. - early?: { readonly serverResumes: boolean }, + early?: { readonly serverResumes: boolean; readonly holdLoads?: boolean }, ) { yield* TestClock.setTime(NOW); const subscriptions = yield* Queue.unbounded<{ @@ -1180,6 +1180,8 @@ describe("environment shell cold start", () => { }>(); const httpLoads = yield* Ref.make(0); const loadOptions = yield* Queue.unbounded(); + const loadsReleased = yield* Deferred.make(); + if (early?.holdLoads !== true) yield* Deferred.succeed(loadsReleased, undefined); const client = { [ORCHESTRATION_V2_WS_METHODS.subscribeShell]: (input: unknown) => Stream.unwrap( @@ -1228,6 +1230,7 @@ describe("environment shell cold start", () => { const snapshotLoader = ShellSnapshotLoader.ShellSnapshotLoader.of({ load: (_prepared, options = {}) => Queue.offer(loadOptions, options).pipe( + Effect.andThen(Deferred.await(loadsReleased)), Effect.andThen(Ref.updateAndGet(httpLoads, (count) => count + 1)), Effect.map((count) => Option.some( @@ -1253,10 +1256,18 @@ describe("environment shell cold start", () => { saved, httpLoads, loadOptions, - connect: (config: object) => + releaseLoads: Deferred.succeed(loadsReleased, undefined), + // Resolves `configRead` when the subscription reads the config, just before it claims the load. + connect: (config: object, configRead?: Deferred.Deferred) => SubscriptionRef.set( activeSession, - Option.some({ ...session(client), initialConfig: Effect.succeed(config as never) }), + Option.some({ + ...session(client), + initialConfig: (configRead === undefined + ? Effect.void + : Deferred.succeed(configRead, undefined) + ).pipe(Effect.as(config as never)), + }), ), reconnect: SubscriptionRef.set(activeSession, Option.some(session(client))), // The registry swaps a rotated bearer into the prepared connection of a live session. @@ -1419,6 +1430,27 @@ describe("environment shell cold start", () => { }), ); + it.effect("a replaced session waits for the load its connection already started", () => + Effect.gen(function* () { + const h = yield* makeColdStart(undefined, { serverResumes: false, holdLoads: true }); + expect(yield* Queue.take(h.loadOptions)).toEqual(FAMILY_FLAGS); + const claimed = yield* Deferred.make(); + yield* h.connect(FAMILY_CONFIG, claimed); + yield* Deferred.await(claimed); + // Lets the first subscription run on from its config read to its wait for the load. + yield* TestClock.adjust("1 millis"); + // The first subscription waits for the load. A second request could answer first, and + // the older answer would then replace its rows under a cursor the socket has passed. + yield* h.connect(FAMILY_CONFIG); + yield* TestClock.adjust("1 millis"); + yield* h.releaseLoads; + const subscribed = yield* Queue.take(h.subscriptions); + expect(subscribed.input).toMatchObject({ afterSequence: HTTP_SEQUENCE }); + expect(yield* Queue.size(h.loadOptions)).toBe(0); + expect(yield* Ref.get(h.httpLoads)).toBe(1); + }), + ); + it.effect("reloads over HTTP when the server answers a resume with snapshot-required", () => Effect.gen(function* () { const h = yield* makeColdStart(RECENT_ORIGIN); diff --git a/packages/client-runtime/src/state/shell.ts b/packages/client-runtime/src/state/shell.ts index d4b9b01aa7953..baaf4c2068605 100644 --- a/packages/client-runtime/src/state/shell.ts +++ b/packages/client-runtime/src/state/shell.ts @@ -557,7 +557,10 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") // A rotated credential replaces the prepared connection of a session that stays up. if (current !== null && prepared !== null && sameConnection(current.prepared, prepared)) { const next = { ...current, prepared, claimed: current.claimed || claim }; - return [current.claimed ? null : current, next] as const; + // A claimed load that still runs is handed out again. A second request could + // answer first, and this one would then replace its rows under a passed cursor. + const done = current.fiber.pollUnsafe() !== undefined; + return [current.claimed && done ? null : current, next] as const; } if (current !== null) yield* Fiber.interrupt(current.fiber); const identity = descriptorShellIdentity(prepared?.descriptor); From 35d1611417c54dedf956cf503a6d70002ba4ff38 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Thu, 12 Nov 2026 02:21:01 -0800 Subject: [PATCH 05/11] fix(client): a late connection change cannot stop thread list updates Takeover review of the prepare-time shell load and the keep-alive change: - The shell reads the prepared connection of the moment it handles a change. It acted on each queued change in order, so a change handled late could start a load for a connection that ended and drop the items of the session that replaced it. - At shutdown the server closes its idle connections after each response. Node tracks which connections still have a pipelined response to send, so the per-socket counter goes away. - headersTimeout keeps Node's 60 second default. An idle kept-alive connection does not count against it, so the longer keep-alive needs no change there. - Test: a load for a connection that ended is dropped, and the subscription waits for the load of the live connection. Co-Authored-By: Claude Fable 5.2 --- apps/server/src/httpResponseErrorGuard.ts | 16 +++------- .../src/state/shell-sync.test.ts | 32 +++++++++++++++++++ packages/client-runtime/src/state/shell.ts | 10 ++++-- 3 files changed, 45 insertions(+), 13 deletions(-) diff --git a/apps/server/src/httpResponseErrorGuard.ts b/apps/server/src/httpResponseErrorGuard.ts index 6f0ac75dbc2a8..95d0dba318661 100644 --- a/apps/server/src/httpResponseErrorGuard.ts +++ b/apps/server/src/httpResponseErrorGuard.ts @@ -28,18 +28,12 @@ const HTTP_KEEP_ALIVE_TIMEOUT_MS = 95_000; /** Install before `rejectRequestsUntilServing`, whose handoff starts at the next request listener. */ export function keepIdleConnections(server: T): T { server.keepAliveTimeout = HTTP_KEEP_ALIVE_TIMEOUT_MS; - server.headersTimeout = HTTP_KEEP_ALIVE_TIMEOUT_MS + 5_000; // `close` drops only the connections idle at that moment. One that finishes - // its responses later would hold shutdown for the whole keep-alive time. - const responsesInFlight = new WeakMap(); - server.on("request", (request, response) => { - const socket = request.socket; - responsesInFlight.set(socket, (responsesInFlight.get(socket) ?? 0) + 1); - response.once("close", () => { - const left = (responsesInFlight.get(socket) ?? 1) - 1; - responsesInFlight.set(socket, left); - // Pipelined requests share the socket, so the last response ends it. - if (left === 0 && !server.listening) socket.end(); + // its responses later would hold shutdown for the whole keep-alive time. A + // connection with another pipelined response to send is not idle yet. + server.on("request", (_request, response) => { + response.once("finish", () => { + if (!server.listening) server.closeIdleConnections(); }); }); return server; diff --git a/packages/client-runtime/src/state/shell-sync.test.ts b/packages/client-runtime/src/state/shell-sync.test.ts index 87da5deef7127..08b296bae0c16 100644 --- a/packages/client-runtime/src/state/shell-sync.test.ts +++ b/packages/client-runtime/src/state/shell-sync.test.ts @@ -1257,6 +1257,15 @@ describe("environment shell cold start", () => { httpLoads, loadOptions, releaseLoads: Deferred.succeed(loadsReleased, undefined), + // A connection attempt that ended, then the next one. + prepareAgain: SubscriptionRef.set(prepared, Option.none()).pipe( + Effect.andThen( + SubscriptionRef.set( + prepared, + Option.some({ ...PREPARED, descriptor: { ...FAMILY_DESCRIPTOR } }), + ), + ), + ), // Resolves `configRead` when the subscription reads the config, just before it claims the load. connect: (config: object, configRead?: Deferred.Deferred) => SubscriptionRef.set( @@ -1451,6 +1460,29 @@ describe("environment shell cold start", () => { }), ); + it.effect("drops the load of a connection that ended and waits for the next one", () => + Effect.gen(function* () { + const h = yield* makeColdStart(undefined, { serverResumes: false, holdLoads: true }); + yield* Queue.take(h.loadOptions); + yield* h.prepareAgain; + yield* Queue.take(h.loadOptions); + + // The socket is ready while the second load is in flight. + yield* h.connect(FAMILY_CONFIG); + yield* h.releaseLoads; + const first = yield* Queue.take(h.subscriptions); + expect(first.input).toMatchObject({ + afterSequence: HTTP_SEQUENCE, + knownProjectIdentities: "identities-1", + }); + // The first load ended with its connection and never answered. + expect(yield* Ref.get(h.httpLoads)).toBe(1); + + yield* Queue.offer(first.events, { kind: "synchronized" }); + yield* h.liveAt(HTTP_SEQUENCE); + }), + ); + it.effect("reloads over HTTP when the server answers a resume with snapshot-required", () => Effect.gen(function* () { const h = yield* makeColdStart(RECENT_ORIGIN); diff --git a/packages/client-runtime/src/state/shell.ts b/packages/client-runtime/src/state/shell.ts index baaf4c2068605..8a520bdc2dfe3 100644 --- a/packages/client-runtime/src/state/shell.ts +++ b/packages/client-runtime/src/state/shell.ts @@ -114,12 +114,12 @@ function shellIdentity( }; } -/** Null for a server whose descriptor does not name its shell flags; only its config does. */ function sameConnection(left: PreparedConnection, right: PreparedConnection): boolean { const keys = Object.keys({ ...left, ...right }) as Array; return keys.every((key) => key === "httpAuthorization" || left[key] === right[key]); } +/** Null for a server whose descriptor does not name its shell flags; only its config does. */ function descriptorShellIdentity( descriptor: ExecutionEnvironmentDescriptor | undefined, ): ShellOriginIdentity | null { @@ -573,7 +573,13 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") }, Effect.uninterruptible), ); yield* SubscriptionRef.changes(supervisor.prepared).pipe( - Stream.runForEach((prepared) => useEarlyLoad(Option.getOrNull(prepared), false)), + // Reads the connection of this moment: a change handled late must not start a load for a + // connection that ended, which would also drop the items of the session that replaced it. + Stream.runForEach(() => + SubscriptionRef.get(supervisor.prepared).pipe( + Effect.flatMap((prepared) => useEarlyLoad(Option.getOrNull(prepared), false)), + ), + ), Effect.forkScoped, ); From f9d0998176f3d47ee85cefcc438180b3f21391cb Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Thu, 12 Nov 2026 02:29:54 -0800 Subject: [PATCH 06/11] fix(server): shutdown keeps a pipelined response that still has output to send My last commit replaced the per-socket response count with closeIdleConnections. That call destroys a socket whose next pipelined response is ended but not flushed, so a large response was cut off (8 MB arrived as 2.6 MB in a local run). The count is back. The test now sends three pipelined requests. It fails with closeIdleConnections and with a socket end on the first finish. Co-Authored-By: Claude Fable 5.2 --- .../server/src/httpResponseErrorGuard.test.ts | 22 ++++++++++++------- apps/server/src/httpResponseErrorGuard.ts | 17 +++++++++----- 2 files changed, 26 insertions(+), 13 deletions(-) diff --git a/apps/server/src/httpResponseErrorGuard.test.ts b/apps/server/src/httpResponseErrorGuard.test.ts index b5e52aa4de1a7..7227e03dade0a 100644 --- a/apps/server/src/httpResponseErrorGuard.test.ts +++ b/apps/server/src/httpResponseErrorGuard.test.ts @@ -160,15 +160,20 @@ describe("keepIdleConnections", () => { it("keeps a connection until every response on it has finished", async () => { const server = HttpResponseErrorGuard.keepIdleConnections(NodeHttp.createServer()); const closed = Promise.withResolvers(); - let first: NodeHttp.ServerResponse | undefined; + // Larger than a socket buffer, so it still has output to send after `end`. + const thirdBody = "x".repeat(8_000_000); + const held: NodeHttp.ServerResponse[] = []; server.on("request", (request, response) => { - if (request.url === "/first") { - first = response; + if (request.url !== "/third") { + held.push(response); return; } - // Both pipelined requests are in flight when the server shuts down. + const [first, second] = held; + // All three pipelined requests are in flight when the server shuts down. The third + // response is ended before the first two, and the second ends after the first. + response.end(thirdBody); server.close(() => closed.resolve()); - first?.once("finish", () => setImmediate(() => response.end("second-body"))); + first?.once("finish", () => setImmediate(() => second?.end("second-body"))); first?.end("first-body"); }); const port = await listen(server); @@ -177,7 +182,7 @@ describe("keepIdleConnections", () => { let data = ""; const client = NodeNet.connect(port, "127.0.0.1", () => { client.write( - ["/first", "/second"] + ["/first", "/second", "/third"] .map((path) => `GET ${path} HTTP/1.1\r\nHost: 127.0.0.1\r\n\r\n`) .join(""), ); @@ -190,8 +195,9 @@ describe("keepIdleConnections", () => { client.on("error", reject); }); - expect(received).toContain("first-body"); - expect(received).toContain("second-body"); + expect(received.slice(0, 1_000)).toContain("first-body"); + expect(received.slice(0, 1_000)).toContain("second-body"); + expect(received.endsWith(thirdBody)).toBe(true); await closed.promise; }); }); diff --git a/apps/server/src/httpResponseErrorGuard.ts b/apps/server/src/httpResponseErrorGuard.ts index 95d0dba318661..c75ccae660a55 100644 --- a/apps/server/src/httpResponseErrorGuard.ts +++ b/apps/server/src/httpResponseErrorGuard.ts @@ -29,11 +29,18 @@ const HTTP_KEEP_ALIVE_TIMEOUT_MS = 95_000; export function keepIdleConnections(server: T): T { server.keepAliveTimeout = HTTP_KEEP_ALIVE_TIMEOUT_MS; // `close` drops only the connections idle at that moment. One that finishes - // its responses later would hold shutdown for the whole keep-alive time. A - // connection with another pipelined response to send is not idle yet. - server.on("request", (_request, response) => { - response.once("finish", () => { - if (!server.listening) server.closeIdleConnections(); + // its responses later would hold shutdown for the whole keep-alive time. + // `closeIdleConnections` cannot do this: it destroys a socket whose next + // pipelined response is ended but still has output to send. + const responsesInFlight = new WeakMap(); + server.on("request", (request, response) => { + const socket = request.socket; + responsesInFlight.set(socket, (responsesInFlight.get(socket) ?? 0) + 1); + response.once("close", () => { + const left = (responsesInFlight.get(socket) ?? 1) - 1; + responsesInFlight.set(socket, left); + // Pipelined requests share the socket, so the last response ends it. + if (left === 0 && !server.listening) socket.end(); }); }); return server; From f52a48a651b324c08c10756bdade2e13d7cbadc3 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Thu, 12 Nov 2026 02:55:33 -0800 Subject: [PATCH 07/11] fix(client): a plain http tailnet route reads its descriptor again before the token goes out A checked descriptor was reused for 5 seconds on tailnet routes. Over plain http nothing proves who answers: with Tailscale down, a 100.x address can reach another host, and the token went there. The reuse now applies only to HTTPS and loopback routes. Co-Authored-By: Claude Fable 5.2 --- .../client-runtime/src/authorization/layer.test.ts | 4 ++-- packages/client-runtime/src/authorization/service.ts | 4 ++-- packages/client-runtime/src/connection/routes.test.ts | 8 +++++--- packages/client-runtime/src/connection/routes.ts | 10 +++++----- 4 files changed, 14 insertions(+), 12 deletions(-) diff --git a/packages/client-runtime/src/authorization/layer.test.ts b/packages/client-runtime/src/authorization/layer.test.ts index 771ba4b6aafbd..8db36ec6c399f 100644 --- a/packages/client-runtime/src/authorization/layer.test.ts +++ b/packages/client-runtime/src/authorization/layer.test.ts @@ -376,11 +376,11 @@ describe("RemoteEnvironmentAuthorization", () => { yield* authorize; // Plain HTTP is no reason to read again where the address keeps its host. yield* remote.rememberCheckedDescriptor( - "http://100.101.102.103:3773", + "http://127.0.0.1:3773", Option.some(descriptor), true, ); - yield* authorizeAt("http://100.101.102.103:3773"); + yield* authorizeAt("http://127.0.0.1:3773"); // After a network change, a plain LAN address can belong to another machine. yield* remote.rememberCheckedDescriptor( "http://192.168.1.2:3773", diff --git a/packages/client-runtime/src/authorization/service.ts b/packages/client-runtime/src/authorization/service.ts index 822dfcb7c8b63..75b4921f0f85a 100644 --- a/packages/client-runtime/src/authorization/service.ts +++ b/packages/client-runtime/src/authorization/service.ts @@ -157,8 +157,8 @@ export const make = Effect.gen(function* () { >(); // A route or host check hands its descriptor to the connect it leads to, once, - // so the connect skips a round trip. Not for a plain LAN address, which can - // belong to another machine after a network change. + // so the connect skips a round trip. Only over HTTPS or loopback: any other + // plain http address can belong to another machine after a network change. const checkedDescriptors = new Map< string, { readonly descriptor: ExecutionEnvironmentDescriptor; readonly at: bigint } diff --git a/packages/client-runtime/src/connection/routes.test.ts b/packages/client-runtime/src/connection/routes.test.ts index e11aefe407e03..e8dd244e043a6 100644 --- a/packages/client-runtime/src/connection/routes.test.ts +++ b/packages/client-runtime/src/connection/routes.test.ts @@ -87,10 +87,12 @@ describe("connection routes", () => { profile: Option.some(new BearerConnectionProfile({ ...profile, network: "tailscale" })), }; expect(routeKeepsItsHost(PUBLIC)).toBe(true); + expect(routeKeepsItsHost(direct("ts-tls", "https://desk.tail1234.ts.net/"))).toBe(true); expect(routeKeepsItsHost(direct("lo", "http://127.0.0.1:3773/"))).toBe(true); - expect(routeKeepsItsHost(direct("ts", "http://desk.tail1234.ts.net:3773/"))).toBe(true); - expect(routeKeepsItsHost(marked)).toBe(true); - // 100.64.0.0/10 is shared with other VPNs unless the server reports it as its tailnet. + // With Tailscale down, a plain http tailnet address can reach another host: + // carrier networks use 100.64.0.0/10 too, and nothing proves who answers. + expect(routeKeepsItsHost(direct("ts", "http://desk.tail1234.ts.net:3773/"))).toBe(false); + expect(routeKeepsItsHost(marked)).toBe(false); expect(routeKeepsItsHost(tailnetIp)).toBe(false); expect(routeKeepsItsHost(LAN)).toBe(false); expect(routeKeepsItsHost(direct("http", "http://desk.example.com/"))).toBe(false); diff --git a/packages/client-runtime/src/connection/routes.ts b/packages/client-runtime/src/connection/routes.ts index d856d75d8c5a1..7a3b4a11e616c 100644 --- a/packages/client-runtime/src/connection/routes.ts +++ b/packages/client-runtime/src/connection/routes.ts @@ -108,14 +108,14 @@ export function connectionRouteKind(route: ConnectionRoute): ConnectionRouteKind } /** - * Whether the route's address reaches the same machine on every network: HTTPS, - * loopback, or the tailnet. A plain LAN address can belong to another machine - * after a network change, so an answer from it is not kept. + * Whether the route's address reaches the same machine on every network: TLS + * proves who answers, and loopback cannot move. Any other plain http address, + * a tailnet one too once Tailscale is down, can belong to another machine after + * a network change, so an answer from it is not kept. */ export function routeKeepsItsHost(route: ConnectionRoute): boolean { if (routeHttpBaseUrl(route)?.startsWith("https:") === true) return true; - const kind = connectionRouteKind(route); - return kind === "loopback" || kind === "tailnet"; + return connectionRouteKind(route) === "loopback"; } /** A loopback or LAN host answers within a few seconds or is not there. */ From 14ea5c442b7422fcd0cb972603da9e65f483b9dd Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Thu, 12 Nov 2026 02:55:51 -0800 Subject: [PATCH 08/11] fix(client): a ready socket waits at most 3 seconds for the early shell answer A client with rows that can resume waited for the HTTP answer before it subscribed. A proxy that drops /api without an answer held each connect for the 20 second HTTP limit. After 3 seconds the request is now interrupted and the socket resumes the held rows, as on main. The interrupt waits for an answer that is being applied, so a late answer cannot replace rows the socket has moved on. Co-Authored-By: Claude Fable 5.2 --- docs/internals/connection-runtime.md | 6 ++- .../src/state/shell-sync.test.ts | 40 +++++++++++++++++++ packages/client-runtime/src/state/shell.ts | 28 ++++++++++++- 3 files changed, 72 insertions(+), 2 deletions(-) diff --git a/docs/internals/connection-runtime.md b/docs/internals/connection-runtime.md index 94a599be81e8d..e82ff0398e478 100644 --- a/docs/internals/connection-runtime.md +++ b/docs/internals/connection-runtime.md @@ -84,7 +84,11 @@ the server answers `resumable` instead of a snapshot when the socket can replay from them, so a cold start costs one shell request plus the subscription. The live config stays the authority: rows loaded for a build or with omissions it does not confirm are replaced, never resumed. A server whose descriptor does not -name the flags is asked only after its config arrives. +name the flags is asked only after its config arrives. The subscription waits +for that answer, so a snapshot never lands on rows the socket has moved on. When +held rows can resume, the wait ends 3 seconds after the socket is ready: the +request is interrupted and the socket resumes the rows, so a proxy that drops +`/api` without an answer does not hold the connect for the HTTP timeout. Shell snapshots can leave rows out when the client asks. Servers that advertise `shellThreadFamilies` leave out idle subagent children: a child stays while it diff --git a/packages/client-runtime/src/state/shell-sync.test.ts b/packages/client-runtime/src/state/shell-sync.test.ts index 08b296bae0c16..e6d77db7bc0a5 100644 --- a/packages/client-runtime/src/state/shell-sync.test.ts +++ b/packages/client-runtime/src/state/shell-sync.test.ts @@ -1460,6 +1460,46 @@ describe("environment shell cold start", () => { }), ); + it.effect("resumes held rows on the socket when the HTTP answer does not come", () => + Effect.gen(function* () { + const h = yield* makeColdStart( + { ...RECENT_ORIGIN, ...FAMILY_FLAGS }, + { serverResumes: false, holdLoads: true }, + ); + yield* Queue.take(h.loadOptions); + const claimed = yield* Deferred.make(); + yield* h.connect(FAMILY_CONFIG, claimed); + yield* Deferred.await(claimed); + // Lets the subscription run on from its config read to its wait for the load. + yield* TestClock.adjust("1 millis"); + expect(yield* Queue.size(h.subscriptions)).toBe(0); + + yield* TestClock.adjust("3 seconds"); + const first = yield* Queue.take(h.subscriptions); + expect(first.input).toMatchObject({ + afterSequence: CACHED_SEQUENCE, + httpSnapshotFallback: true, + }); + yield* Queue.offer(first.events, { + kind: "thread.removed", + sequence: HTTP_SEQUENCE + 1, + location: "active", + threadId: ThreadId.make("no-such-thread"), + }); + yield* Queue.offer(first.events, { kind: "synchronized" }); + yield* h.liveAt(HTTP_SEQUENCE + 1); + + // The late answer is an older snapshot. It must not replace the rows the socket moved on. + yield* h.releaseLoads; + yield* TestClock.adjust("1 millis"); + expect(yield* Ref.get(h.httpLoads)).toBe(0); + expect(yield* SubscriptionRef.get(h.shellState)).toMatchObject({ + status: "live", + snapshot: Option.some({ snapshotSequence: HTTP_SEQUENCE + 1 }), + }); + }), + ); + it.effect("drops the load of a connection that ended and waits for the next one", () => Effect.gen(function* () { const h = yield* makeColdStart(undefined, { serverResumes: false, holdLoads: true }); diff --git a/packages/client-runtime/src/state/shell.ts b/packages/client-runtime/src/state/shell.ts index 8a520bdc2dfe3..86d4cb0afd967 100644 --- a/packages/client-runtime/src/state/shell.ts +++ b/packages/client-runtime/src/state/shell.ts @@ -92,6 +92,11 @@ const SHELL_SYNCHRONIZATION_ERROR_MESSAGE = "Could not synchronize environment d */ const CACHE_RESUME_MAX_AGE_MILLIS = 12 * 60 * 60 * 1000; +// How long a ready socket waits for the early HTTP answer before it resumes held rows itself. +// The supervisor's quick probe allows a live host the same time. A proxy that drops `/api` +// without an answer would otherwise hold each connect for the full HTTP timeout. +const EARLY_LOAD_GRACE = "3 seconds"; + type ShellOriginIdentity = Omit; interface ShellSubscription { @@ -572,6 +577,27 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") return [next, next] as const; }, Effect.uninterruptible), ); + /** + * The subscription waits for the early load, so its answer is applied before the socket + * moves the rows on. Held rows that can resume wait a limited time only: after that the + * load is ended, so a late answer applies nothing, and the socket resumes them. + */ + const awaitEarlyLoad = Effect.fnUntraced(function* ( + early: EarlyShellLoad, + identity: ShellOriginIdentity | null, + ) { + const resumable = + Option.isSome((yield* SubscriptionRef.get(state)).snapshot) && + canResumeFrom(yield* Ref.get(origin), identity, yield* Clock.currentTimeMillis); + if (!resumable) return yield* Fiber.await(early.fiber); + const answered = yield* Fiber.await(early.fiber).pipe(Effect.timeoutOption(EARLY_LOAD_GRACE)); + if (Option.isSome(answered)) return answered.value; + // Returns after an answer that is being applied has finished, so the cursor below is read + // from rows that no longer change under it. + yield* Fiber.interrupt(early.fiber); + return yield* Fiber.await(early.fiber); + }); + yield* SubscriptionRef.changes(supervisor.prepared).pipe( // Reads the connection of this moment: a change handled late must not start a load for a // connection that ended, which would also drop the items of the session that replaced it. @@ -609,7 +635,7 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") Option.getOrNull(yield* SubscriptionRef.get(supervisor.prepared)), true, ); - const earlyExit = early === null ? null : yield* Fiber.await(early.fiber); + const earlyExit = early === null ? null : yield* awaitEarlyLoad(early, identity); // Rows the descriptor's flags or build loaded are not kept when the live config differs: // their origin then cannot resume, so the load below replaces them. let loaded = From ac76d3d4db54dfa65d8f832581371220ce2ef12d Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Thu, 12 Nov 2026 03:26:13 -0800 Subject: [PATCH 09/11] fix(client): a cancelled subscription keeps the record of the shell load it started Two review findings in the early shell load. - The early load ref was stored after the uninterruptible callback. A subscription cancelled while the callback waited for the previous load started the next load, then took the interrupt before the store. That load ran on with no record, and the next caller started a second one. The whole modification is now uninterruptible. - A `resumable` answer skipped the age check at socket time. Rows that pass the 12 hour limit while the socket opens now ask the socket for a full snapshot. Co-Authored-By: Claude Fable 5.2 --- .../src/state/shell-sync.test.ts | 85 ++++++++++++++++--- packages/client-runtime/src/state/shell.ts | 16 ++-- 2 files changed, 81 insertions(+), 20 deletions(-) diff --git a/packages/client-runtime/src/state/shell-sync.test.ts b/packages/client-runtime/src/state/shell-sync.test.ts index e6d77db7bc0a5..2699c8846f7bc 100644 --- a/packages/client-runtime/src/state/shell-sync.test.ts +++ b/packages/client-runtime/src/state/shell-sync.test.ts @@ -1167,7 +1167,12 @@ describe("environment shell cold start", () => { const makeColdStart = Effect.fn("test.makeColdStartShell")(function* ( origin: Persistence.ShellCacheOrigin | undefined, // Set for a connection that is prepared but has no session yet. - early?: { readonly serverResumes: boolean; readonly holdLoads?: boolean }, + // Each held load waits for one `releaseLoads`. An "uninterruptible" hold stands for a load + // that applies its answer, which an interrupt has to wait for. + early?: { + readonly serverResumes: boolean; + readonly holdLoads?: boolean | "uninterruptible"; + }, ) { yield* TestClock.setTime(NOW); const subscriptions = yield* Queue.unbounded<{ @@ -1180,8 +1185,17 @@ describe("environment shell cold start", () => { }>(); const httpLoads = yield* Ref.make(0); const loadOptions = yield* Queue.unbounded(); - const loadsReleased = yield* Deferred.make(); - if (early?.holdLoads !== true) yield* Deferred.succeed(loadsReleased, undefined); + const releases = yield* Queue.unbounded(); + const scope = yield* Effect.scope; + const holdLoad = + early?.holdLoads === "uninterruptible" + ? // Registered after the load's fiber, so the end of a failed test does not wait for it. + Scope.addFinalizer(scope, Queue.shutdown(releases)).pipe( + Effect.andThen(Effect.uninterruptible(Queue.take(releases))), + ) + : early?.holdLoads === true + ? Queue.take(releases) + : Effect.void; const client = { [ORCHESTRATION_V2_WS_METHODS.subscribeShell]: (input: unknown) => Stream.unwrap( @@ -1230,7 +1244,7 @@ describe("environment shell cold start", () => { const snapshotLoader = ShellSnapshotLoader.ShellSnapshotLoader.of({ load: (_prepared, options = {}) => Queue.offer(loadOptions, options).pipe( - Effect.andThen(Deferred.await(loadsReleased)), + Effect.andThen(holdLoad), Effect.andThen(Ref.updateAndGet(httpLoads, (count) => count + 1)), Effect.map((count) => Option.some( @@ -1250,21 +1264,23 @@ describe("environment shell cold start", () => { Effect.provideService(Persistence.EnvironmentCacheStore, cache), Effect.provideService(ShellSnapshotLoader.ShellSnapshotLoader, snapshotLoader), ); + const nextConnection = Effect.suspend(() => + SubscriptionRef.set( + prepared, + Option.some({ ...PREPARED, descriptor: { ...FAMILY_DESCRIPTOR } }), + ), + ); return { shellState, subscriptions, saved, httpLoads, loadOptions, - releaseLoads: Deferred.succeed(loadsReleased, undefined), + releaseLoads: Queue.offer(releases, undefined), + nextConnection, // A connection attempt that ended, then the next one. prepareAgain: SubscriptionRef.set(prepared, Option.none()).pipe( - Effect.andThen( - SubscriptionRef.set( - prepared, - Option.some({ ...PREPARED, descriptor: { ...FAMILY_DESCRIPTOR } }), - ), - ), + Effect.andThen(nextConnection), ), // Resolves `configRead` when the subscription reads the config, just before it claims the load. connect: (config: object, configRead?: Deferred.Deferred) => @@ -1374,6 +1390,16 @@ describe("environment shell cold start", () => { loads: [{ ...FAMILY_FLAGS, resumeFrom: CACHED_SEQUENCE }], subscribes: { afterSequence: HTTP_SEQUENCE, knownProjectIdentities: "identities-1" }, }, + { + // The server answered that it can replay, then the rows passed their age limit. + label: "rows that expire before the socket is ready", + origin: { ...RECENT_ORIGIN, ...FAMILY_FLAGS, snapshotAt: NOW - 12 * HOUR + 1 }, + serverResumes: true, + config: FAMILY_CONFIG, + loads: [{ ...FAMILY_FLAGS, resumeFrom: CACHED_SEQUENCE }], + subscribes: {}, + expires: true, + }, { // The rows loaded with the descriptor's omissions are replaced, never kept. label: "a live config with other omissions", @@ -1385,7 +1411,7 @@ describe("environment shell cold start", () => { }, ])( "asks for the shell once the connection is prepared, with $label", - ({ origin, serverResumes, config, loads, subscribes }) => + ({ origin, serverResumes, config, loads, subscribes, expires }) => Effect.gen(function* () { const h = yield* makeColdStart(origin, { serverResumes }); expect(yield* Queue.take(h.loadOptions)).toEqual(loads[0]); @@ -1406,14 +1432,21 @@ describe("environment shell cold start", () => { } expect(yield* Queue.size(h.subscriptions)).toBe(0); + if (expires) yield* TestClock.adjust("1 millis"); yield* h.connect(config); const first = yield* Queue.take(h.subscriptions); expect(first.input).toMatchObject(subscribes); - for (const key of ["httpSnapshotFallback", "knownProjectIdentities"] as const) { + for (const key of [ + "afterSequence", + "httpSnapshotFallback", + "knownProjectIdentities", + ] as const) { if (!(key in subscribes)) expect(first.input).not.toHaveProperty(key); } expect(yield* Queue.clear(h.loadOptions)).toEqual(loads.slice(1)); expect(yield* Ref.get(h.httpLoads)).toBe(loads.length); + // The socket sends the full snapshot these rows asked for; the rest needs a cursor. + if (!("afterSequence" in subscribes)) return; yield* Queue.offer(first.events, { kind: "synchronized" }); yield* h.liveAt(subscribes.afterSequence); @@ -1441,7 +1474,10 @@ describe("environment shell cold start", () => { it.effect("a replaced session waits for the load its connection already started", () => Effect.gen(function* () { - const h = yield* makeColdStart(undefined, { serverResumes: false, holdLoads: true }); + const h = yield* makeColdStart(undefined, { + serverResumes: false, + holdLoads: "uninterruptible", + }); expect(yield* Queue.take(h.loadOptions)).toEqual(FAMILY_FLAGS); const claimed = yield* Deferred.make(); yield* h.connect(FAMILY_CONFIG, claimed); @@ -1452,6 +1488,27 @@ describe("environment shell cold start", () => { // the older answer would then replace its rows under a cursor the socket has passed. yield* h.connect(FAMILY_CONFIG); yield* TestClock.adjust("1 millis"); + expect(yield* Queue.size(h.loadOptions)).toBe(0); + + // The connection changes twice while the first load ends. The next session reads the + // last connection before its load exists, so it replaces the load of the one between. + yield* h.nextConnection; + yield* TestClock.adjust("1 millis"); + yield* h.nextConnection; + yield* h.connect(FAMILY_CONFIG); + yield* TestClock.adjust("1 millis"); + yield* h.releaseLoads; + expect(yield* Queue.take(h.loadOptions)).toEqual(FAMILY_FLAGS); + yield* TestClock.adjust("1 millis"); + // That session is replaced while it waits for the load between to end. The load it then + // starts must stay the one load of the last connection. + yield* h.connect(FAMILY_CONFIG); + yield* TestClock.adjust("1 millis"); + yield* h.releaseLoads; + expect(yield* Queue.take(h.loadOptions)).toEqual(FAMILY_FLAGS); + yield* TestClock.adjust("1 millis"); + expect(yield* Queue.size(h.loadOptions)).toBe(0); + yield* h.releaseLoads; const subscribed = yield* Queue.take(h.subscriptions); expect(subscribed.input).toMatchObject({ afterSequence: HTTP_SEQUENCE }); diff --git a/packages/client-runtime/src/state/shell.ts b/packages/client-runtime/src/state/shell.ts index 86d4cb0afd967..52f10e0b2d736 100644 --- a/packages/client-runtime/src/state/shell.ts +++ b/packages/client-runtime/src/state/shell.ts @@ -575,7 +575,11 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") const fiber = yield* loadOverHttp(prepared, identity).pipe(Effect.forkIn(stateScope)); const next: EarlyShellLoad = { prepared, identity, fiber, claimed: claim }; return [next, next] as const; - }, Effect.uninterruptible), + }), + ).pipe( + // Covers the store too. A cancel that lands during the wait above would otherwise take + // effect before the store, and the load just started would run on with no record. + Effect.uninterruptible, ); /** * The subscription waits for the early load, so its answer is applied before the socket @@ -652,12 +656,12 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") // Rows from a recent server snapshot with the same omissions and server build resume // from their cursor, in memory or from the cache. Anything else reloads the HTTP - // snapshot so a valid cursor cannot preserve stale or incomplete data. + // snapshot so a valid cursor cannot preserve stale or incomplete data. A `resumable` + // answer is checked again too: the rows can pass their age limit while the socket opens. const resumingHeldRows = - loaded?.kind === "resumable" || - (loaded?.kind !== "snapshot" && - Option.isSome((yield* SubscriptionRef.get(state)).snapshot) && - canResumeFrom(yield* Ref.get(origin), identity, yield* Clock.currentTimeMillis)); + loaded?.kind !== "snapshot" && + Option.isSome((yield* SubscriptionRef.get(state)).snapshot) && + canResumeFrom(yield* Ref.get(origin), identity, yield* Clock.currentTimeMillis); if (!resumingHeldRows && loaded === null) { const prepared = yield* SubscriptionRef.get(supervisor.prepared).pipe( Effect.flatMap( From 5296c253ffde0f2054657d33a2d659fd10679ee4 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Thu, 12 Nov 2026 04:14:00 -0800 Subject: [PATCH 10/11] docs(internals): the shell load paragraph keeps the constraints and drops the request sequence Co-Authored-By: Claude Fable 5.2 --- docs/internals/connection-runtime.md | 23 +++++++++-------------- 1 file changed, 9 insertions(+), 14 deletions(-) diff --git a/docs/internals/connection-runtime.md b/docs/internals/connection-runtime.md index e22dabcc80f20..4ebad4c7056ef 100644 --- a/docs/internals/connection-runtime.md +++ b/docs/internals/connection-runtime.md @@ -75,20 +75,15 @@ connection nor overwrite newer live data during a reconnect. Loading and resuming snapshots belongs to the shared state services, so every view agrees on which data is current. -The shell snapshot loads over HTTP, which needs only the prepared connection. -When the server's public descriptor names its shell flags, the -[shell state](../../packages/client-runtime/src/state/shell.ts) asks for the -shell as soon as the connection is prepared, while the socket opens and the -config downloads. The request names held rows that may resume (`resumeFrom`), and -the server answers `resumable` instead of a snapshot when the socket can replay -from them, so a cold start costs one shell request plus the subscription. The -live config stays the authority: rows loaded for a build or with omissions it -does not confirm are replaced, never resumed. A server whose descriptor does not -name the flags is asked only after its config arrives. The subscription waits -for that answer, so a snapshot never lands on rows the socket has moved on. When -held rows can resume, the wait ends 3 seconds after the socket is ready: the -request is interrupted and the socket resumes the rows, so a proxy that drops -`/api` without an answer does not hold the connect for the HTTP timeout. +The shell snapshot loads over HTTP, which needs only the prepared connection, so +the [shell state](../../packages/client-runtime/src/state/shell.ts) can ask for +it before the socket and the config are ready. Two constraints follow. The live +config stays the authority: rows loaded early for a build or with omissions that +it does not confirm are replaced, never resumed. And an HTTP answer must never +land on rows the socket has moved on: the subscription waits for the answer, and +when held rows can resume, it ends the request 3 seconds after the socket is +ready. A proxy that drops `/api` without an answer then does not hold the +connect for the HTTP timeout. Each environment synchronizes at its own time, so a thread list that follows every snapshot moves once per environment on each cold start and resume. Mobile lists read From 7055bea7a64a83aefe45641c511217bc2ab7ef25 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Thu, 12 Nov 2026 06:24:43 -0800 Subject: [PATCH 11/11] test(server): wire budget sends the shell resume cursor and project identities The merge of main brought #35225, which types the budget test's opt-in records `Required` against the contracts. This branch adds `resumeFrom` to the shell query and `knownProjectIdentities` to the shell subscribe input, so the server typecheck failed on both records. Both fields differ per request, like the cursors the records already leave out, so each scenario sends its own: - The fleet shell subscribe names the snapshot's project identities. The fixture project is now in a repository; without the digest the subscribe resends it (851 B in 2 messages, over the 200 B and 1 message ceiling). - New row "fleet shell resume check": the HTTP route's answer to a client that holds rows, 18 B. - New row "stale shell reload": the same request from a cursor too old to replay, which gets the snapshot. Co-Authored-By: Claude Fable 5.2 --- .../TransferBudgetReport.integration.ts | 19 +++++- .../TransferBudgetV2Fixture.integration.ts | 40 ++++++++++++ .../transferBudgetV2.integration.test.ts | 61 ++++++++++++++----- docs/internals/performance-regressions.md | 5 +- 4 files changed, 105 insertions(+), 20 deletions(-) diff --git a/apps/server/integration/TransferBudgetReport.integration.ts b/apps/server/integration/TransferBudgetReport.integration.ts index 005c15f8e1669..bcfa2f3896f20 100644 --- a/apps/server/integration/TransferBudgetReport.integration.ts +++ b/apps/server/integration/TransferBudgetReport.integration.ts @@ -74,7 +74,11 @@ export interface TransferBudgetRun { readonly fleetShellLegacy: TimelineHttpMeasurement; /** The shell subscription a cold connect opens from that snapshot's cursor. */ readonly fleetShellSubscribe: SocketOpenMeasurement; - /** A shell resumed from a cursor too old to replay, by a client that reloads over HTTP. */ + /** A client that holds that snapshot's rows asks the HTTP route whether they still resume. */ + readonly fleetShellResume: TimelineHttpMeasurement; + /** The same question from a cursor too old to replay: the route answers with the snapshot. */ + readonly staleShellReload: TimelineHttpMeasurement; + /** A shell resumed over the socket from that old cursor, by a client that reloads over HTTP. */ readonly staleShellResume: SocketOpenMeasurement; /** * The compact bounded snapshot of the long thread whose finished runs left todo lists open, @@ -141,6 +145,7 @@ const SOCKET_BUDGETS = { longTurnColdOpen: { decodedBytes: LONG_TURN_FINISHED_BUDGET.decodedBytes, messages: 2 }, // These three send no rows. A snapshot in their place is 393 KB for the thread, 42 KB for the shell. idleThreadResume: { decodedBytes: 200, messages: 1 }, + // A subscribe that does not name the snapshot's project identities gets the project again: 851 B. fleetShellSubscribe: { decodedBytes: 200, messages: 1 }, staleShellResume: { decodedBytes: 400, messages: 2 }, }; @@ -151,6 +156,12 @@ const FLEET_SHELL_BUDGET = { rows: 27, }; +// The answer is `{"resumable":true}`. A snapshot in its place is 42 KB. +const FLEET_SHELL_RESUME_BUDGET = { + decodedBytes: 100, + rows: 0, +}; + // About 14% above the observed 75 KB. The window holds the working run only: 2 nodes, 2 subagents, // 1 plan, 1 provider turn and the last finished turn's checkpoint. When finished runs ride along // it holds 82 nodes, 82 subagents, 41 plans, 41 provider turns, 40 checkpoints and 740 KB. @@ -367,6 +378,8 @@ function scenarioRows(run: TransferBudgetRun): ScenarioRow[] { run.fleetShellSubscribe, SOCKET_BUDGETS.fleetShellSubscribe, ), + ...httpRows("fleet shell resume check", run.fleetShellResume, FLEET_SHELL_RESUME_BUDGET), + ...httpRows("stale shell reload", run.staleShellReload, FLEET_SHELL_BUDGET), ...socketRows("stale shell resume", run.staleShellResume, SOCKET_BUDGETS.staleShellResume), ...httpRows("stale todo list cold open", run.staleTodoColdOpen, STALE_TODO_BUDGET), ...STALE_TODO_RECORD_KINDS.map((kind): ScenarioRow => ({ @@ -439,8 +452,8 @@ export function formatTransferBudgetReport(runs: ReadonlyArray; +} satisfies Required>; const CLIENT_THREAD_SUBSCRIPTION = { requestCompletionMarker: true, acceptBoundedSnapshot: true, @@ -114,17 +118,22 @@ const CLIENT_THREAD_SUBSCRIPTION = { acceptCompactIds: true, acceptDedupedBodies: true, } satisfies Required>; +type ShellResumeField = "afterSequence" | "httpSnapshotFallback" | "knownProjectIdentities"; const CLIENT_SHELL_SUBSCRIPTION = { requestCompletionMarker: true, omitIdleSubagents: true, omitOlderSettled: true, -} satisfies Required< - Omit ->; +} satisfies Required>; const query = (optIns: Readonly>) => new URLSearchParams(optIns).toString(); const CLIENT_BOUNDED_QUERY = query(CLIENT_BOUNDED_OPT_INS); const CLIENT_HISTORY_QUERY = query(CLIENT_HISTORY_OPT_INS); const CLIENT_SHELL_QUERY = query(CLIENT_SHELL_OPT_INS); +// A client that holds shell rows names their cursor in the same request. +const clientShellResumeQuery = (resumeFrom: number) => + query({ + ...CLIENT_SHELL_OPT_INS, + resumeFrom: String(resumeFrom), + } satisfies Required); const decodeThreadSnapshot = Schema.decodeUnknownEffect( Schema.fromJsonString(Schema.toCodecJson(OrchestrationV2ThreadDetailSnapshot)), @@ -136,7 +145,10 @@ const decodeHistoryPage = Schema.decodeUnknownEffect( Schema.fromJsonString(Schema.toCodecJson(OrchestrationV2ThreadHistoryPage)), ); const decodeShellSnapshot = Schema.decodeUnknownEffect( - Schema.fromJsonString(Schema.toCodecJson(OrchestrationV2ShellSnapshot)), + Schema.fromJsonString(Schema.toCodecJson(OrchestrationV2ShellHttpSnapshot)), +); +const decodeShellResumable = Schema.decodeUnknownEffect( + Schema.fromJsonString(Schema.toCodecJson(OrchestrationV2ShellResumable)), ); const layerPersistence = EventSink.layer.pipe( @@ -176,7 +188,7 @@ const layerEnrichment = Layer.unwrap( return Layer.mock(ProjectEnrichmentService.ProjectEnrichmentService)({ getAvailable: () => Effect.succeed({ - repositoryIdentity: null, + repositoryIdentity: FLEET_PROJECT_IDENTITY, faviconPath: null, repositoryIdentityResolved: true, }), @@ -263,10 +275,7 @@ const collectUntil = Effect.fn("transfer.collectUntil")(function* ) - | ({ readonly shell: true } & Pick< - OrchestrationV2SubscribeShellInput, - "afterSequence" | "httpSnapshotFallback" - >); + | ({ readonly shell: true } & Pick); function subscriptionStream( client: MeasuredWsClient, subscription: Subscription, @@ -457,8 +466,10 @@ it.live("reports V2 thread HTTP and WebSocket transfer budgets", () => const readShell = (query: string) => Effect.gen(function* () { const http = yield* get(`/api/orchestration/shell?${query}`); - const { threads, snapshotSequence } = yield* decodeShellSnapshot(decoded(http)); - return { http, rows: threads.length, snapshotSequence }; + const { threads, snapshotSequence, projectIdentities } = yield* decodeShellSnapshot( + decoded(http), + ); + return { http, rows: threads.length, snapshotSequence, projectIdentities }; }); // A new socket, read until the subscription caught up or the server ended it. const open = (subscription: Subscription) => @@ -529,6 +540,8 @@ it.live("reports V2 thread HTTP and WebSocket transfer budgets", () => // The transfer thread plus the long-turn thread. const earlierThreads = decodedShell.threads.length + 1; + // The fleet's project is in a repository. Earlier phases have no project, as on `main`. + yield* (yield* ProjectStore.ProjectStoreV2).apply(fleetProjectCreated); yield* sink.write({ events: shellFleetEvents(provider) }); const fleetShellLegacy = yield* readShell(""); const fleetShell = yield* readShell(CLIENT_SHELL_QUERY); @@ -536,13 +549,27 @@ it.live("reports V2 thread HTTP and WebSocket transfer budgets", () => fleetShellLegacy.rows, earlierThreads + SHELL_SETTLED_THREADS + SHELL_IDLE_SUBAGENTS, ); - // A cold connect subscribes from the snapshot it just loaded. + const { projectIdentities } = fleetShell; + if (projectIdentities === undefined) + return yield* Effect.die("Expected the shell snapshot to name its project identities"); + // A cold connect subscribes from the snapshot it just loaded and names its identities. + // Without them the resume sends the project again, in a marked snapshot. const fleetShellSubscribe = yield* open({ shell: true, afterSequence: fleetShell.snapshotSequence, + knownProjectIdentities: projectIdentities, }); assert.notInclude(fleetShellSubscribe.kinds, "snapshot"); - // A cursor too old to replay is told to reload over HTTP, not sent a socket snapshot. + // A client that holds those rows asks the route whether they still resume, and keeps them. + const fleetShellResumeHttp = yield* get( + `/api/orchestration/shell?${clientShellResumeQuery(fleetShell.snapshotSequence)}`, + ); + yield* decodeShellResumable(decoded(fleetShellResumeHttp)); + const fleetShellResume = { http: fleetShellResumeHttp, rows: 0 }; + // A cursor too old to replay gets the snapshot in the same answer. + const staleShellReload = yield* readShell(clientShellResumeQuery(sequence)); + assert.equal(staleShellReload.rows, fleetShell.rows); + // When that request fails, the socket tells the client to reload, with no snapshot. const staleShellResume = yield* open({ shell: true, afterSequence: sequence, @@ -601,6 +628,8 @@ it.live("reports V2 thread HTTP and WebSocket transfer budgets", () => fleetShellLegacy, fleetShell, fleetShellSubscribe, + fleetShellResume, + staleShellReload, staleShellResume, staleTodoColdOpen, } satisfies TransferBudgetRun; diff --git a/docs/internals/performance-regressions.md b/docs/internals/performance-regressions.md index 2b095859148af..8dbf42fc45f72 100644 --- a/docs/internals/performance-regressions.md +++ b/docs/internals/performance-regressions.md @@ -79,7 +79,10 @@ bounded snapshot and one older page) and after it finishes (the deferred cold op over the socket alone, and the first page of its deferred work). It caps the socket bytes of three resumes that must not send a snapshot: a thread after one ordinary turn, an idle thread, and a shell cursor too old to replay. It caps the cold-connect shell for a sidebar of settled threads and -idle subagents, and the subscription a client opens from that snapshot. It also caps the bytes and +idle subagents, and the subscription a client opens from that snapshot with its project identities. +It caps both answers of the HTTP shell route to a client that holds rows and names their cursor: the +few bytes that keep them, and the snapshot that replaces rows too old to replay. It also caps the +bytes and the node, subagent, plan, provider turn and checkpoint counts of the bounded snapshot for a thread of 40 finished runs behind one working run, so finished runs cannot ride along with a window. The working run has a ready context handoff, so its history cannot ride along either.