fix(client-runtime): re-subscribe server config instead of killing the session

.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"
+          ]
+        }
+      ]
     }
   ]
 }

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

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

apps/server/package.json

@@ -1,6 +1,6 @@
 {
   "name": "t3",
-  "version": "0.0.46",
+  "version": "0.0.47",
   "license": "MIT",
   "repository": {
     "type": "git",

apps/web/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@t3tools/web",
-  "version": "0.0.46",
+  "version": "0.0.47",
   "private": true,
   "type": "module",
   "scripts": {

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<ServerConfigStreamEventType>();
+      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 }) =>

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<ServerConfig, ConnectionAttemptError>;
@@ -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(

packages/contracts/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@t3tools/contracts",
-  "version": "0.0.46",
+  "version": "0.0.47",
   "private": true,
   "files": [
     "dist"

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