From 1274b5cd755d8c2223b45206be29680ebdbc4003 Mon Sep 17 00:00:00 2001 From: t3-turbo-bot Date: Tue, 1 Sep 2026 16:49:33 -0400 Subject: [PATCH] fix(client-runtime): re-subscribe server config instead of killing the session Upstream b883fc066 (#8367, "halve server config bootstrap traffic") made the server-config subscription a liveness dependency of the whole WebSocket session: a clean end of that stream failed `configSubscriptionClosed`, which `closed` races, so the supervisor tore the socket down -- and the RPC protocol is built with `retryTransientErrors: false` / `Schedule.recurs(0)`, so nothing retried. On a loaded server the config stream ends or stalls while the socket is perfectly healthy, and every teardown re-established all subscriptions on reconnect, which raised the load that caused the next teardown. Observed on a real environment: six whole-connection kills in 24 minutes at shrinking intervals (245s, 52s, 15s), with 13/13 subscribeServerConfig spans ending interrupted. A clean end of the config stream is no longer session-fatal. The session re-subscribes on the live socket under a jittered exponential backoff (500 ms, capped at 10 s), forked into the session scope so a real disconnect still interrupts it. Recoverable server-side config errors take the same path. Defects, EnvironmentAuthorizationError, and transport failures still fail serverConfigExit and configSubscriptionClosed exactly as before, so closed keeps its race shape and durable consumers keep their transport-shaped failure. Replay state survives the re-subscribe, so consumers see one fresh snapshot rather than a torn projection. Registered as the durable-config-subscription-resilience seam; upstream pingdotgg/t3code#7233 (issue #7231) is the candidate that would retire it. Also bumps the fork version line to 0.0.47. Model: Claude Opus 5 (1M) in Claude Code. Co-Authored-By: Claude Fable 5 --- .t3-turbo/customizations.json | 23 +++ SEAM.md | 27 ++++ apps/desktop/package.json | 2 +- apps/server/package.json | 2 +- apps/web/package.json | 2 +- .../client-runtime/src/rpc/session.test.ts | 153 ++++++++++++++++++ packages/client-runtime/src/rpc/session.ts | 41 ++++- packages/contracts/package.json | 2 +- scripts/turbo-customization-manifest.test.ts | 1 + 9 files changed, 244 insertions(+), 9 deletions(-) diff --git a/.t3-turbo/customizations.json b/.t3-turbo/customizations.json index 43220eedfabd..9c860e05971a 100644 --- a/.t3-turbo/customizations.json +++ b/.t3-turbo/customizations.json @@ -1358,6 +1358,29 @@ ] } ] + }, + { + "id": "durable-config-subscription-resilience", + "status": "implemented", + "summary": "Upstream #8367 (perf: halve server config bootstrap traffic) made the server-config subscription a liveness dependency of the WebSocket session: a clean end of that stream failed configSubscriptionClosed, which races closed and tears the socket down, and the protocol is configured with no retry. On a loaded server the config stream ends or stalls while the socket is healthy, so one slow snapshot caused a full reconnect that re-established every subscription and raised the load that caused the next one. The fork re-subscribes on the live socket with a jittered exponential backoff capped at 10s instead, keeps first-snapshot semantics, and only fails the session on a genuinely non-transient config failure.", + "checks": [ + { + "path": "packages/client-runtime/src/rpc/session.ts", + "markers": [ + "SERVER_CONFIG_RESUBSCRIBE_SCHEDULE", + "environment.serverConfig.resubscribe", + "durable-config-subscription-resilience" + ] + }, + { + "path": "packages/client-runtime/src/rpc/session.test.ts", + "markers": [ + "re-subscribes server config when the stream ends cleanly", + "closes the session when the config stream fails with a non-transient error", + "closes the session when the websocket disconnects" + ] + } + ] } ] } diff --git a/SEAM.md b/SEAM.md index abd138ec6cc2..2b9dc39ba7f8 100644 --- a/SEAM.md +++ b/SEAM.md @@ -437,6 +437,33 @@ version line never advanced there. On a nightly-sync conflict: keep upstream's job graph and re-add the two `env` entries plus the five `ref:`/push lines. If upstream ever gains its own release-branch input, prefer it. +## Durable server-config subscription (fork fix) + +Upstream `b883fc066` (#8367, "halve server config bootstrap traffic") turned the server-config +subscription into a liveness dependency of the whole WebSocket session. In +`packages/client-runtime/src/rpc/session.ts`, a **clean** end of that stream failed +`configSubscriptionClosed`, which is raced by `closed`, so the supervisor tore the socket down; the +RPC protocol is built with `retryTransientErrors: false` and `Schedule.recurs(0)`, so nothing +retried. A busy server ends or stalls the config stream while the socket is perfectly healthy, and +each teardown re-established every subscription on reconnect, which raised the load that caused the +next teardown (observed: six whole-connection kills in 24 minutes, at shrinking intervals). + +- A clean end of the config stream is no longer session-fatal; the session re-subscribes on the + live socket under `SERVER_CONFIG_RESUBSCRIBE_SCHEDULE` (jittered exponential from 500 ms, capped + at 10 s, unbounded while the session scope is open). The loop is forked into the session scope, so + a real disconnect interrupts it. +- Server-side config errors the server can recover from (`KeybindingsConfigParseError`, + `ServerSettingsError`) are recovered into a clean end and take the same loop. +- Genuinely non-transient failures (defects, `EnvironmentAuthorizationError`) and transport failures + still fail `serverConfigExit` and `configSubscriptionClosed` exactly as upstream does, so `closed` + keeps its race shape and consumers still get a transport-shaped failure. +- Replay state is preserved across a re-subscribe: `applyServerConfigProjection` folds the new + snapshot onto the existing projection, so consumers see one snapshot event, not a torn projection. + +On a nightly-sync conflict: keep upstream's `onExit` shape and re-apply the success-branch change, +the `Effect.catchTags` recovery, and the `Effect.repeat` wrapper. Retire this seam if upstream merges +its own durable config subscription — PR pingdotgg/t3code#7233 (issue #7231) is the candidate. + ## Nightly sync conflicts Resolve against the new upstream file first, then reapply only the behavior above; never take the diff --git a/apps/desktop/package.json b/apps/desktop/package.json index 9c0349a02537..fe09cf95e724 100644 --- a/apps/desktop/package.json +++ b/apps/desktop/package.json @@ -1,6 +1,6 @@ { "name": "@t3tools/desktop", - "version": "0.0.46", + "version": "0.0.47", "private": true, "type": "module", "main": "dist-electron/main.cjs", diff --git a/apps/server/package.json b/apps/server/package.json index ea50dd7b2eee..ff2778b941c7 100644 --- a/apps/server/package.json +++ b/apps/server/package.json @@ -1,6 +1,6 @@ { "name": "t3", - "version": "0.0.46", + "version": "0.0.47", "license": "MIT", "repository": { "type": "git", diff --git a/apps/web/package.json b/apps/web/package.json index ea8fd3f4de1c..8427d2253a41 100644 --- a/apps/web/package.json +++ b/apps/web/package.json @@ -1,6 +1,6 @@ { "name": "@t3tools/web", - "version": "0.0.46", + "version": "0.0.47", "private": true, "type": "module", "scripts": { diff --git a/packages/client-runtime/src/rpc/session.test.ts b/packages/client-runtime/src/rpc/session.test.ts index aedd85c5de47..13e53e70d484 100644 --- a/packages/client-runtime/src/rpc/session.test.ts +++ b/packages/client-runtime/src/rpc/session.test.ts @@ -260,6 +260,39 @@ const publishConfigEvents = Effect.fn("TestRpcSessionFactory.publishConfigEvents ); }); +const endConfigStream = Effect.fn("TestRpcSessionFactory.endConfigStream")(function* ( + socket: TestWebSocket, + index = 0, +) { + const request = yield* awaitRequest(socket, index); + socket.serverMessage( + encodeJson({ + _tag: "Exit", + requestId: request.id, + exit: { _tag: "Success", value: null }, + }), + ); +}); + +/** + * Steps the test clock in small increments until the socket has sent `index + 1` + * requests. The re-subscribe backoff is a real sleep, and virtual time only + * moves when something advances it; the increments stay well under the ping + * window so waiting here cannot itself close the session. + */ +const awaitRequestWithBackoff = Effect.fn("TestRpcSessionFactory.awaitRequestWithBackoff")( + function* (socket: TestWebSocket, index: number) { + for (let attempt = 0; attempt < 40; attempt += 1) { + const request = socket.sent.map((message) => decodeJson(message)).filter(isRpcRequest)[index]; + if (request) { + return request; + } + yield* TestClock.adjust("100 millis"); + } + return yield* Effect.die(new Error("Expected the session to re-subscribe to server config.")); + }, +); + describe("RpcSessionFactory", () => { it.effect("owns one scoped websocket attempt and exposes readiness and closure", () => Effect.gen(function* () { @@ -678,6 +711,126 @@ describe("RpcSessionFactory", () => { ), ); + it.effect("re-subscribes server config when the stream ends cleanly", () => + Effect.gen(function* () { + const { factory, sockets } = yield* makeFactory(); + const session = yield* factory.connect(PREPARED); + const readyFiber = yield* Effect.forkChild(session.ready); + const socket = yield* awaitSocket(sockets); + socket.open(); + yield* completeInitialConfig(socket); + yield* Fiber.join(readyFiber); + + const closedFiber = yield* Effect.forkChild(Effect.exit(session.closed)); + const observed = yield* Queue.unbounded(); + yield* session.subscribeServerConfig({}).pipe( + Stream.runForEach((event) => Queue.offer(observed, event)), + Effect.forkChild, + ); + expect((yield* Queue.take(observed)).type).toBe("snapshot"); + + yield* endConfigStream(socket); + + const resubscribe = yield* awaitRequestWithBackoff(socket, 1); + expect(resubscribe).toMatchObject({ + _tag: "Request", + tag: WS_METHODS.subscribeServerConfig, + }); + expect(sockets).toHaveLength(1); + expect(closedFiber.pollUnsafe()).toBeUndefined(); + + socket.serverMessage( + encodeJson({ + _tag: "Chunk", + requestId: resubscribe.id, + values: [ + { + version: 1, + type: "snapshot", + config: { + ...ENCODED_SERVER_CONFIG, + environment: { + ...ENCODED_SERVER_CONFIG.environment, + label: "Re-subscribed environment", + }, + }, + }, + ], + }), + ); + + const replacement = yield* Queue.take(observed); + expect(replacement).toMatchObject({ + type: "snapshot", + config: { environment: { label: "Re-subscribed environment" } }, + }); + expect(closedFiber.pollUnsafe()).toBeUndefined(); + }).pipe(Effect.scoped, Effect.provide(TestClock.layer())), + ); + + it.effect("closes the session when the config stream fails with a non-transient error", () => + Effect.gen(function* () { + const { factory, sockets } = yield* makeFactory(); + const session = yield* factory.connect(PREPARED); + const readyFiber = yield* Effect.forkChild(session.ready); + const socket = yield* awaitSocket(sockets); + socket.open(); + yield* completeInitialConfig(socket); + yield* Fiber.join(readyFiber); + + const closedFiber = yield* Effect.forkChild(Effect.flip(session.closed)); + const request = yield* awaitRequest(socket); + socket.serverMessage( + encodeJson({ + _tag: "Exit", + requestId: request.id, + exit: { + _tag: "Failure", + cause: [ + { + _tag: "Fail", + error: { + _tag: "EnvironmentAuthorizationError", + message: "config subscription rejected", + requiredScope: "orchestration:read", + }, + }, + ], + }, + }), + ); + + const error = yield* Fiber.join(closedFiber); + expect(error).toBeInstanceOf(ConnectionBlockedError); + expect(error).toMatchObject({ reason: "permission" }); + + yield* TestClock.adjust("30 seconds"); + expect(socket.sent.map((message) => decodeJson(message)).filter(isRpcRequest)).toHaveLength( + 1, + ); + }).pipe(Effect.scoped, Effect.provide(TestClock.layer())), + ); + + it.effect("closes the session when the websocket disconnects", () => + Effect.gen(function* () { + const { factory, sockets } = yield* makeFactory(); + const session = yield* factory.connect(PREPARED); + const readyFiber = yield* Effect.forkChild(session.ready); + const socket = yield* awaitSocket(sockets); + socket.open(); + yield* completeInitialConfig(socket); + yield* Fiber.join(readyFiber); + + socket.close(1006, "network lost"); + + const error = yield* Effect.flip(session.closed); + expect(error).toBeInstanceOf(ConnectionTransientError); + expect(error).toMatchObject({ reason: "transport" }); + yield* Effect.yieldNow; + expect(sockets).toHaveLength(1); + }).pipe(Effect.scoped), + ); + it.effect.each([{ failure: "defect" as const }, { failure: "typed" as const }])( "keeps durable config state alive after an owned $failure failure", ({ failure }) => diff --git a/packages/client-runtime/src/rpc/session.ts b/packages/client-runtime/src/rpc/session.ts index 7d975be5c9d3..9d2d377f2b9a 100644 --- a/packages/client-runtime/src/rpc/session.ts +++ b/packages/client-runtime/src/rpc/session.ts @@ -41,6 +41,26 @@ import { const SOCKET_OPEN_TIMEOUT = "15 seconds"; +// T3 Turbo seam: durable-config-subscription-resilience. +// +// Upstream #8367 made the server-config subscription a liveness dependency of +// the whole WebSocket session: a *clean* end of that stream failed +// `configSubscriptionClosed`, which races `closed`, which tears the socket down. +// A loaded server ends or stalls the config stream while the socket itself is +// perfectly healthy, so one slow snapshot became a full reconnect -- and the +// reconnect re-established every subscription, which raised the load that caused +// the next one. We re-subscribe on the live socket instead. +// +// The backoff is bounded rather than unbounded-exponential: the cap is the worst +// config staleness a user can see, and 10s of stale config is cheaper than a +// reconnect. The loop is forked into the session scope, so a real disconnect +// interrupts it. +const SERVER_CONFIG_RESUBSCRIBE_MAX_DELAY = "10 seconds"; +const SERVER_CONFIG_RESUBSCRIBE_SCHEDULE = Schedule.min([ + Schedule.exponential("500 millis"), + Schedule.spaced(SERVER_CONFIG_RESUBSCRIBE_MAX_DELAY), +]).pipe(Schedule.jittered); + export interface RpcSession { readonly client: WsRpcProtocolClient; readonly initialConfig: Effect.Effect; @@ -232,12 +252,19 @@ export const make = Effect.fn("RpcSessionFactory.make")(function* ( } }), ), + // T3 Turbo (durable-config-subscription-resilience): a config error the + // server can recover from is the environment's problem, not the socket's. + // Recovering it into a clean end hands it to the re-subscribe loop below + // instead of killing a healthy session. + Effect.catchTags({ + KeybindingsConfigParseError: () => Effect.void, + ServerSettingsError: () => Effect.void, + }), Effect.onExit((exit) => { if (Exit.isSuccess(exit)) { - return Effect.all([ - Deferred.succeed(serverConfigExit, undefined), - Deferred.fail(configSubscriptionClosed, configSubscriptionEndedError), - ]).pipe(Effect.asVoid); + // T3 Turbo (durable-config-subscription-resilience): a clean end is + // NOT session-fatal. Upstream failed `configSubscriptionClosed` here. + return Effect.void; } if (Cause.hasInterruptsOnly(exit.cause)) { return Effect.void; @@ -248,7 +275,11 @@ export const make = Effect.fn("RpcSessionFactory.make")(function* ( ]).pipe(Effect.asVoid); }), ); - yield* serverConfigSource.pipe(Effect.forkScoped); + yield* serverConfigSource.pipe( + Effect.repeat(SERVER_CONFIG_RESUBSCRIBE_SCHEDULE), + Effect.withSpan("environment.serverConfig.resubscribe"), + Effect.forkScoped, + ); const initialConfig = Effect.raceFirst( Deferred.await(initialConfigDeferred), Deferred.await(serverConfigExit).pipe( diff --git a/packages/contracts/package.json b/packages/contracts/package.json index 88446ca7c052..5eae3c69b823 100644 --- a/packages/contracts/package.json +++ b/packages/contracts/package.json @@ -1,6 +1,6 @@ { "name": "@t3tools/contracts", - "version": "0.0.46", + "version": "0.0.47", "private": true, "files": [ "dist" diff --git a/scripts/turbo-customization-manifest.test.ts b/scripts/turbo-customization-manifest.test.ts index f80162f7c7da..34269090287f 100644 --- a/scripts/turbo-customization-manifest.test.ts +++ b/scripts/turbo-customization-manifest.test.ts @@ -147,6 +147,7 @@ it("verifies the checked-in Turbo manifest and tracks the implemented multi-chat "cheap-message-unpacking", "cheap-timestamp-and-sort-keys", "deferred-streaming-code-blocks", + "durable-config-subscription-resilience", "file-explorer", "markdown-preview-preference", "multi-chat-pane-workspace",