fix(connect): T3 Connect sockets no longer drop every hour

apps/server/src/auth/SessionStore.test.ts

@@ -98,10 +98,10 @@ it.layer(NodeServices.layer)("SessionStore.layer", (it) => {
             Effect.forkScoped({ startImmediately: true }),
           );
         const first = yield* sessions
-          .awaitInvalidation(client.sessionId)
+          .awaitInvalidation(client.sessionId, relaySessionInput.proofKeyThumbprint)
           .pipe(Effect.forkScoped({ startImmediately: true }));
         const second = yield* sessions
-          .awaitInvalidation(client.sessionId)
+          .awaitInvalidation(client.sessionId, relaySessionInput.proofKeyThumbprint)
           .pipe(Effect.forkScoped({ startImmediately: true }));
         if (operation === "revoke") yield* sessions.revoke(client.sessionId);
         else if (operation === "revokeAllExcept") yield* sessions.revokeAllExcept(owner.sessionId);
@@ -120,7 +120,7 @@ it.layer(NodeServices.layer)("SessionStore.layer", (it) => {
       yield* sessions.markConnected(client.sessionId);
       const invalidated = yield* Deferred.make<void>();
       const watcher = yield* sessions
-        .awaitInvalidation(client.sessionId)
+        .awaitInvalidation(client.sessionId, relaySessionInput.proofKeyThumbprint)
         .pipe(
           Effect.andThen(Deferred.succeed(invalidated, undefined)),
           Effect.forkScoped({ startImmediately: true }),
@@ -133,6 +133,102 @@ it.layer(NodeServices.layer)("SessionStore.layer", (it) => {
     }).pipe(Effect.scoped, Effect.provide(Layer.merge(layerSessionStore(), TestClock.layer()))),
   );
 
+  it.effect.each([
+    { name: "a renewal for the same key", renewal: {}, closesAtMinute: 119 },
+    {
+      name: "a renewal for another key",
+      renewal: { proofKeyThumbprint: "other-proof-key" },
+      closesAtMinute: 60,
+    },
+    {
+      name: "a renewal with fewer scopes",
+      renewal: { scopes: ["orchestration:read"] as const },
+      closesAtMinute: 60,
+    },
+    {
+      name: "a renewal, until the original is revoked",
+      renewal: {},
+      closesAtMinute: 90,
+      revoke: "original",
+    },
+    {
+      name: "a renewal, until the renewal is revoked",
+      renewal: {},
+      closesAtMinute: 90,
+      revoke: "renewal",
+    },
+  ] as const)("closes a DPoP socket at minute $closesAtMinute after $name", (testCase) =>
+    Effect.gen(function* () {
+      const sessions = yield* SessionStore.SessionStore;
+      const client = yield* sessions.issue(relaySessionInput);
+      const invalidated = yield* Deferred.make<void>();
+      const watcher = yield* sessions
+        .awaitInvalidation(client.sessionId, relaySessionInput.proofKeyThumbprint)
+        .pipe(
+          Effect.andThen(Deferred.succeed(invalidated, undefined)),
+          Effect.forkScoped({ startImmediately: true }),
+        );
+      yield* TestClock.adjust(Duration.minutes(59));
+      const renewal = yield* sessions.issue({ ...relaySessionInput, ...testCase.renewal });
+      yield* TestClock.adjust(Duration.millis((testCase.closesAtMinute - 59) * 60_000 - 1));
+      expect(yield* Deferred.isDone(invalidated)).toBe(false);
+      yield* TestClock.adjust(Duration.millis(1));
+      if ("revoke" in testCase) {
+        yield* sessions.revoke(
+          testCase.revoke === "original" ? client.sessionId : renewal.sessionId,
+        );
+      }
+      yield* Fiber.join(watcher);
+    }).pipe(Effect.scoped, Effect.provide(Layer.merge(layerSessionStore(), TestClock.layer()))),
+  );
+
+  it.effect("closes a DPoP socket when an earlier renewal in its chain is revoked", () =>
+    Effect.gen(function* () {
+      const sessions = yield* SessionStore.SessionStore;
+      const client = yield* sessions.issue(relaySessionInput);
+      const invalidated = yield* Deferred.make<void>();
+      const watcher = yield* sessions
+        .awaitInvalidation(client.sessionId, relaySessionInput.proofKeyThumbprint)
+        .pipe(
+          Effect.andThen(Deferred.succeed(invalidated, undefined)),
+          Effect.forkScoped({ startImmediately: true }),
+        );
+      yield* TestClock.adjust(Duration.minutes(59));
+      const first = yield* sessions.issue(relaySessionInput);
+      yield* TestClock.adjust(Duration.minutes(59));
+      yield* sessions.issue(relaySessionInput);
+      // Past the first renewal's expiry, so the socket now follows the second.
+      yield* TestClock.adjust(Duration.minutes(2));
+      expect(yield* Deferred.isDone(invalidated)).toBe(false);
+      yield* sessions.revoke(first.sessionId);
+      yield* Fiber.join(watcher);
+    }).pipe(Effect.scoped, Effect.provide(Layer.merge(layerSessionStore(), TestClock.layer()))),
+  );
+
+  it.effect("keeps a renewed DPoP socket when another socket on its expired session closes", () =>
+    Effect.gen(function* () {
+      const sessions = yield* SessionStore.SessionStore;
+      const client = yield* sessions.issue(relaySessionInput);
+      yield* sessions.markConnected(client.sessionId);
+      const invalidated = yield* Deferred.make<void>();
+      const watcher = yield* sessions
+        .awaitInvalidation(client.sessionId, relaySessionInput.proofKeyThumbprint)
+        .pipe(
+          Effect.andThen(Deferred.succeed(invalidated, undefined)),
+          Effect.forkScoped({ startImmediately: true }),
+        );
+      yield* TestClock.adjust(Duration.minutes(59));
+      yield* sessions.issue(relaySessionInput);
+      yield* TestClock.adjust(Duration.minutes(2));
+      // An expired session that loses its last socket is reported as removed, not revoked.
+      yield* sessions.markDisconnected(client.sessionId);
+      yield* TestClock.adjust(Duration.millis(58 * 60_000 - 1));
+      expect(yield* Deferred.isDone(invalidated)).toBe(false);
+      yield* TestClock.adjust(Duration.millis(1));
+      yield* Fiber.join(watcher);
+    }).pipe(Effect.scoped, Effect.provide(Layer.merge(layerSessionStore(), TestClock.layer()))),
+  );
+
   it.effect("keys remote cookies by environment identity instead of state directory", () =>
     Effect.gen(function* () {
       const cookieName = (stateDir: string, environmentId: EnvironmentId) =>

apps/server/src/auth/SessionStore.ts

@@ -409,8 +409,10 @@ export class SessionStore extends Context.Service<
       SessionCredentialInternalError
     >;
     readonly streamChanges: Stream.Stream<SessionCredentialChange>;
+    /** Completes when a live socket's session is revoked or expires without a renewal for its DPoP key. */
     readonly awaitInvalidation: (
       sessionId: AuthSessionId,
+      proofKeyThumbprint?: string,
     ) => Effect.Effect<void, SessionCredentialVerificationError>;
     readonly revoke: (
       sessionId: AuthSessionId,
@@ -497,6 +499,9 @@ const make = Effect.gen(function* () {
   const signingSecret = yield* secretStore.getOrCreateRandom(SIGNING_SECRET_NAME, 32);
   const connectedSessionsRef = yield* Ref.make(new Map<AuthSessionId, number>());
   const changesPubSub = yield* PubSub.unbounded<SessionCredentialChange>();
+  // Newest session per DPoP key. A live socket moves to it at expiry, so a
+  // client that renews its short-lived token keeps its socket.
+  const latestSessionByProofKey = new Map<string, AuthSessionId>();
   const cookieInput = {
     mode: serverConfig.mode,
     port: serverConfig.port,
@@ -703,6 +708,7 @@ const make = Effect.gen(function* () {
             })
           : authSessions.create(sessionRecord).pipe(Effect.as([] as ReadonlyArray<AuthSessionId>))
       ).pipe(Effect.mapError((cause) => new SessionCredentialIssueError({ sessionId, cause })));
+      if (claims.jkt) latestSessionByProofKey.set(claims.jkt, sessionId);
       if (replacedSessionIds.length > 0) {
         yield* Ref.update(connectedSessionsRef, (current) => {
           const next = new Map(current);
@@ -1034,31 +1040,61 @@ const make = Effect.gen(function* () {
     return revokedSessionIds.length;
   });
 
-  const awaitInvalidation = Effect.fn(function* (sessionId: AuthSessionId) {
+  const awaitInvalidation = Effect.fn(function* (
+    sessionId: AuthSessionId,
+    proofKeyThumbprint?: string,
+  ) {
     return yield* Effect.scoped(
       Effect.gen(function* () {
         // Subscribe before reading: revocation can race with the WebSocket upgrade.
         const subscription = yield* PubSub.subscribe(changesPubSub);
-        const row = yield* authSessions
-          .getById({ sessionId })
-          .pipe(
-            Effect.mapError(
-              (cause) => new SessionCredentialVerificationError({ sessionId, cause }),
-            ),
-          );
-        if (Option.isNone(row) || row.value.revokedAt !== null) return;
-        const now = yield* DateTime.now;
-        const remaining = row.value.expiresAt.epochMilliseconds - now.epochMilliseconds;
-        if (remaining <= 0) return;
-        yield* Effect.raceFirst(
-          Stream.fromSubscription(subscription).pipe(
-            Stream.filter(
-              (change) => change.type === "clientRemoved" && change.sessionId === sessionId,
-            ),
-            Stream.runHead,
-          ),
-          Effect.sleep(Duration.millis(remaining)),
-        );
+        const readRow = (id: AuthSessionId) =>
+          authSessions
+            .getById({ sessionId: id })
+            .pipe(
+              Effect.mapError(
+                (cause) => new SessionCredentialVerificationError({ sessionId: id, cause }),
+              ),
+            );
+        const original = yield* readRow(sessionId);
+        if (Option.isNone(original)) return;
+        // Revoking any session the socket has used closes it. Events only wake this loop
+        // and the rows decide: revocations are written before they are published, and
+        // clientRemoved also reports an expired session losing its last socket.
+        const chain = [sessionId];
+        const nextRemoval = Effect.gen(function* () {
+          while (true) {
+            const change = yield* PubSub.take(subscription);
+            if (change.type === "clientRemoved" && chain.includes(change.sessionId)) return;
+          }
+        });
+        let current = original.value;
+        while (true) {
+          const rows = yield* Effect.forEach(chain, readRow);
+          if (rows.some((row) => Option.isNone(row) || row.value.revokedAt !== null)) return;
+          const now = yield* DateTime.now;
+          const remaining = current.expiresAt.epochMilliseconds - now.epochMilliseconds;
+          if (remaining > 0) {
+            yield* Effect.raceFirst(nextRemoval, Effect.sleep(Duration.millis(remaining)));
+            continue;
+          }
+          const successorId =
+            proofKeyThumbprint === undefined
+              ? undefined
+              : latestSessionByProofKey.get(proofKeyThumbprint);
+          if (successorId === undefined || successorId === current.sessionId) return;
+          const successor = yield* readRow(successorId);
+          // The socket keeps its original scopes, so the renewal must still grant all of them.
+          if (
+            Option.isNone(successor) ||
+            successor.value.subject !== original.value.subject ||
+            !original.value.scopes.every((scope) => successor.value.scopes.includes(scope))
+          ) {
+            return;
+          }
+          chain.push(successorId);
+          current = successor.value;
+        }
       }),
     );
   });

apps/server/src/ws.ts

@@ -3385,7 +3385,7 @@ export const layer = Layer.unwrap(
           () =>
             Effect.raceFirst(
               rpcWebSocketHttpEffect,
-              sessions.awaitInvalidation(session.sessionId).pipe(
+              sessions.awaitInvalidation(session.sessionId, session.proofKeyThumbprint).pipe(
                 Effect.as(HttpServerResponse.empty()),
                 Effect.catchTags({
                   SessionCredentialVerificationError: (error) =>

docs/internals/connection-runtime.md

@@ -46,8 +46,15 @@ RPC sessions authenticate at socket upgrade, while HTTP requests need current
 credentials from the
 [authorization service](../../packages/client-runtime/src/authorization/service.ts).
 Replacing a healthy socket for HTTP renewal would interrupt conversations and
-change the transport generation without a transport failure. Credential expiry
-does not close the socket, and refresh failure belongs to the HTTP operation.
+change the transport generation without a transport failure. Refresh failure
+belongs to the HTTP operation.
+
+The server closes a socket when its session is revoked or expires. T3 Connect
+access tokens last one hour, so the driver renews a live socket's token in its
+last minute. The server then moves the socket to the newer session for the same
+DPoP key, subject, and scopes
+([`SessionStore.awaitInvalidation`](../../apps/server/src/auth/SessionStore.ts)).
+If renewal fails, the socket closes at expiry and the supervisor reconnects.
 
 Session listings must retain unrevoked connected sessions after credential expiry
 so an open connection does not disappear from connection management. This does

packages/client-runtime/src/authorization/layer.test.ts

@@ -602,6 +602,59 @@ describe("RemoteEnvironmentAuthorization", () => {
       }),
   );
 
+  it.effect("renews a live socket's token in the minute before each expiry", () =>
+    Effect.gen(function* () {
+      const puts = yield* Queue.unbounded<void>();
+      const relayBlips = yield* Ref.make(1);
+      const harness = yield* makeHarness({
+        beforeBootstrap: Ref.getAndUpdate(relayBlips, (count) => count - 1).pipe(
+          Effect.flatMap((count) =>
+            count > 0
+              ? Effect.fail(
+                  new ManagedRelay.ManagedRelayRequestTimeoutError({
+                    activity: "Relay environment connection",
+                    timeoutMs: 30_000,
+                    traceId: null,
+                  }),
+                )
+              : Effect.void,
+          ),
+        ),
+        initialToken: persistedToken({ expiresAtEpochMs: 3_600_000 }),
+        responses: [
+          Response.json(DESCRIPTOR),
+          accessToken("renewed-1"),
+          Response.json(DESCRIPTOR),
+          accessToken("renewed-2"),
+        ],
+        beforePut: Queue.offer(puts, undefined).pipe(Effect.asVoid),
+      });
+
+      yield* Effect.gen(function* () {
+        const remote = yield* RemoteEnvironmentAuthorization.RemoteEnvironmentAuthorization;
+        yield* remote
+          .renewDpopBeforeExpiry({
+            expectedEnvironmentId: ENVIRONMENT_ID,
+            expiresAtEpochMs: 3_600_000,
+          })
+          .pipe(Effect.forkScoped);
+        yield* TestClock.adjust(3_540_000 - 1);
+        expect(yield* Ref.get(harness.bootstrapCalls)).toBe(0);
+        // The first try hits a relay blip, and the retry 5 s later succeeds.
+        yield* TestClock.adjust(1);
+        expect(yield* Ref.get(harness.bootstrapCalls)).toBe(1);
+        yield* TestClock.adjust(5_000);
+        yield* Queue.take(puts);
+        // Renewed at 0:59:05, so the next renewal is at 1:58:05.
+        yield* TestClock.adjust(7_085_000 - 3_545_000 - 1);
+        expect(yield* Ref.get(harness.bootstrapCalls)).toBe(2);
+        yield* TestClock.adjust(1);
+        yield* Queue.take(puts);
+        expect(yield* Ref.get(harness.bootstrapCalls)).toBe(3);
+      }).pipe(Effect.scoped, Effect.provide(Layer.merge(harness.layer, TestClock.layer())));
+    }),
+  );
+
   it.effect("evicts an auth-invalid cached token and obtains a fresh bootstrap", () =>
     Effect.gen(function* () {
       const cached = new TokenStore.RemoteDpopAccessToken({

packages/client-runtime/src/authorization/service.ts

@@ -82,6 +82,14 @@ export class RemoteEnvironmentAuthorization extends Context.Service<
       readonly expectedEnvironmentId: EnvironmentId;
       readonly rejectedAccessToken?: string;
     }) => Effect.Effect<AuthorizedRemoteHttpEnvironment, ConnectionAttemptError>;
+    /**
+     * Renews a live socket's token before it expires, so the server moves the
+     * socket to the new session instead of closing it. Runs until interrupted.
+     */
+    readonly renewDpopBeforeExpiry: (input: {
+      readonly expectedEnvironmentId: EnvironmentId;
+      readonly expiresAtEpochMs: number;
+    }) => Effect.Effect<void, ConnectionBlockedError>;
   }
 >()("@t3tools/client-runtime/authorization/service/RemoteEnvironmentAuthorization") {}
 
@@ -530,6 +538,30 @@ export const make = Effect.gen(function* () {
         withRelayClientTracing,
       );
     }),
+    // Untraced: a span here would stay open for the socket's lifetime. Each renewal has its own.
+    renewDpopBeforeExpiry: Effect.fnUntraced(function* (input) {
+      let expiresAtEpochMs = input.expiresAtEpochMs;
+      let now = yield* Clock.currentTimeMillis;
+      // Past expiry the server has closed the socket.
+      while (now < expiresAtEpochMs) {
+        // getDpopToken reuses a cached token until it is within the skew of expiry.
+        // The floor spaces out retries and keeps a wall-clock jump from spinning.
+        yield* Effect.sleep(
+          Math.max(5_000, expiresAtEpochMs - DPOP_ACCESS_TOKEN_REFRESH_SKEW_MS - now),
+        );
+        const renewed = yield* getDpopToken({
+          expectedEnvironmentId: input.expectedEnvironmentId,
+        }).pipe(
+          Effect.map(({ token }) => token.expiresAtEpochMs),
+          Effect.withSpan("environment.authorization.renewSocket"),
+          withRelayClientTracing,
+          // A relay blip retries until expiry; a blocked credential ends the loop.
+          Effect.catchTags({ ConnectionTransientError: () => Effect.succeed(expiresAtEpochMs) }),
+        );
+        expiresAtEpochMs = Math.max(expiresAtEpochMs, renewed);
+        now = yield* Clock.currentTimeMillis;
+      }
+    }),
   });
 });