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",