fix(connect): refresh relay credentials before expiry

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

@@ -30,7 +30,10 @@ import * as Ref from "effect/Ref";
 import * as Result from "effect/Result";
 import * as HttpClient from "effect/unstable/http/HttpClient";
 
-import type { PreparedHttpAuthorization } from "../connection/model.ts";
+import {
+  DPOP_ACCESS_TOKEN_REFRESH_SKEW_MS,
+  type PreparedHttpAuthorization,
+} from "../connection/model.ts";
 
 export interface RelayEnvironmentAuthorization {
   readonly environmentId: EnvironmentId;
@@ -66,7 +69,6 @@ export class RemoteEnvironmentAuthorization extends Context.Service<
   }
 >()("@t3tools/client-runtime/authorization/service/RemoteEnvironmentAuthorization") {}
 
-const TOKEN_EXPIRY_SAFETY_MARGIN_MS = 60_000;
 const CACHED_ENDPOINT_SOCKET_TIMEOUT_MS = 3_000;
 const BEARER_DESCRIPTOR_CACHE_TTL_MS = 10_000;
 
@@ -216,7 +218,7 @@ export const make = Effect.gen(function* () {
         Option.isSome(cached) &&
         cached.value.environmentId === input.expectedEnvironmentId &&
         cached.value.dpopThumbprint === thumbprint &&
-        cached.value.expiresAtEpochMs > now + TOKEN_EXPIRY_SAFETY_MARGIN_MS
+        cached.value.expiresAtEpochMs > now + DPOP_ACCESS_TOKEN_REFRESH_SKEW_MS
       ) {
         yield* Effect.annotateCurrentSpan({
           "connection.remote_token_cache": "hit",
@@ -234,6 +236,7 @@ export const make = Effect.gen(function* () {
             httpAuthorization: {
               _tag: "Dpop" as const,
               accessToken: cached.value.accessToken,
+              expiresAtEpochMs: cached.value.expiresAtEpochMs,
             },
           };
         }
@@ -305,6 +308,7 @@ export const make = Effect.gen(function* () {
         httpAuthorization: {
           _tag: "Dpop" as const,
           accessToken: token.accessToken,
+          expiresAtEpochMs: token.expiresAtEpochMs,
         },
       };
     },

packages/client-runtime/src/connection/model.ts

@@ -103,6 +103,8 @@ export class ConnectionBlockedError extends Schema.TaggedErrorClass<ConnectionBl
 
 export type ConnectionAttemptError = ConnectionTransientError | ConnectionBlockedError;
 
+export const DPOP_ACCESS_TOKEN_REFRESH_SKEW_MS = 60_000;
+
 export type PreparedHttpAuthorization =
   | {
       readonly _tag: "Bearer";
@@ -111,6 +113,7 @@ export type PreparedHttpAuthorization =
   | {
       readonly _tag: "Dpop";
       readonly accessToken: string;
+      readonly expiresAtEpochMs: number;
     };
 
 export interface PreparedConnection {

packages/client-runtime/src/connection/resolver.test.ts

@@ -140,6 +140,7 @@ const makeDependencies = Effect.fn("TestConnectionResolver.makeDependencies")((o
             httpAuthorization: {
               _tag: "Dpop" as const,
               accessToken: "dpop-access-token",
+              expiresAtEpochMs: Number.MAX_SAFE_INTEGER,
             },
           }),
         )),
@@ -357,6 +358,7 @@ describe("ConnectionResolver", () => {
               httpAuthorization: {
                 _tag: "Dpop" as const,
                 accessToken: "dpop-access-token",
+                expiresAtEpochMs: Number.MAX_SAFE_INTEGER,
               },
             }),
           ),
@@ -394,6 +396,7 @@ describe("ConnectionResolver", () => {
               httpAuthorization: {
                 _tag: "Dpop" as const,
                 accessToken: "dpop-access-token",
+                expiresAtEpochMs: Number.MAX_SAFE_INTEGER,
               },
             }),
             Effect.withSpan("test.remote.authorizeDpop"),

packages/client-runtime/src/connection/supervisor.test.ts

@@ -16,6 +16,7 @@ import type { ConnectionCatalogEntry } from "./catalog.ts";
 import * as Connectivity from "./connectivity.ts";
 import * as ConnectionDriver from "./driver.ts";
 import {
+  DPOP_ACCESS_TOKEN_REFRESH_SKEW_MS,
   ConnectionBlockedError,
   ConnectionTransientError,
   PrimaryConnectionTarget,
@@ -1096,6 +1097,43 @@ describe("EnvironmentSupervisor", () => {
     }),
   );
 
+  it.effect("renews a relay connection before its DPoP access token expires", () =>
+    Effect.gen(function* () {
+      const tokenLifetimeMs = DPOP_ACCESS_TOKEN_REFRESH_SKEW_MS * 2;
+      const harness = yield* makeHarness({
+        prepare: (attempt) =>
+          Effect.succeed({
+            ...PREPARED_CONNECTION,
+            target: RELAY_TARGET,
+            httpAuthorization: {
+              _tag: "Dpop",
+              accessToken: `access-token-${attempt}`,
+              expiresAtEpochMs: tokenLifetimeMs * attempt,
+            },
+          }),
+      });
+      const supervisor = yield* EnvironmentSupervisor.make(RELAY_ENTRY, {
+        initiallyDesired: true,
+      }).pipe(Effect.provide(harness.dependencies));
+
+      yield* awaitState(supervisor.state, (state) => state.phase === "connected");
+      yield* TestClock.adjust(DPOP_ACCESS_TOKEN_REFRESH_SKEW_MS - 1);
+      expect(yield* Ref.get(harness.sessionCount)).toBe(1);
+
+      yield* TestClock.adjust(1);
+      yield* awaitState(
+        supervisor.state,
+        (state) => state.phase === "connected" && state.generation === 2,
+      );
+
+      expect(yield* Ref.get(harness.sessionCount)).toBe(2);
+      expect(yield* Ref.get(harness.releaseCount)).toBe(1);
+      expect(
+        Option.getOrThrow(yield* SubscriptionRef.get(supervisor.prepared)).httpAuthorization,
+      ).toMatchObject({ accessToken: "access-token-2" });
+    }).pipe(Effect.provide(TestClock.layer())),
+  );
+
   it.effect("interrupts relay setup when credentials change", () =>
     Effect.gen(function* () {
       const firstAttemptStarted = yield* Deferred.make<void>();

packages/client-runtime/src/connection/supervisor.ts

@@ -18,6 +18,7 @@ import type { ConnectionCatalogEntry } from "./catalog.ts";
 import * as Connectivity from "./connectivity.ts";
 import * as ConnectionDriver from "./driver.ts";
 import {
+  DPOP_ACCESS_TOKEN_REFRESH_SKEW_MS,
   type ConnectionAttemptError,
   type ConnectionTarget,
   ConnectionTransientError,
@@ -316,12 +317,10 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (
         }),
       });
       const lease = yield* effect.pipe(
-        Effect.mapError(
-          (error): TracedAttemptFailure => ({
-            error,
-            attemptSpan: Option.some(attemptSpan),
-          }),
-        ),
+        Effect.mapError((error): TracedAttemptFailure => ({
+          error,
+          attemptSpan: Option.some(attemptSpan),
+        })),
       );
       return { attemptSpan: Option.some(attemptSpan), lease };
     }).pipe(Effect.withSpan("relay.connection.attempt", { root: true }));
@@ -357,12 +356,10 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (
         attemptSpan: Option.none<Tracer.Span>(),
         lease,
       })),
-      Effect.mapError(
-        (error): TracedAttemptFailure => ({
-          error,
-          attemptSpan: Option.none(),
-        }),
-      ),
+      Effect.mapError((error): TracedAttemptFailure => ({
+        error,
+        attemptSpan: Option.none(),
+      })),
     );
   });
 
@@ -487,6 +484,21 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (
     }
   });
 
+  const waitForAuthorizationRefresh = Effect.fnUntraced(function* (
+    preparedConnection: PreparedConnection,
+  ) {
+    const authorization = preparedConnection.httpAuthorization;
+    if (authorization?._tag !== "Dpop") {
+      return yield* Effect.never;
+    }
+    const now = yield* Clock.currentTimeMillis;
+    yield* Effect.sleep(
+      Math.max(0, authorization.expiresAtEpochMs - now - DPOP_ACCESS_TOKEN_REFRESH_SKEW_MS),
+    );
+    yield* Effect.logDebug("Refreshing the environment connection before its DPoP token expires.");
+    return true;
+  });
+
   const runAttempt = Effect.fnUntraced(function* (
     attempt: number,
     generation: number,
@@ -498,20 +510,16 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (
       exitUnlessInterrupted(
         establishTracedConnection(attempt, generation, lastFailure, pendingRetry),
       ).pipe(
-        Effect.map(
-          (exit): EstablishmentEvent => ({
-            _tag: "Completed",
-            exit,
-          }),
-        ),
+        Effect.map((exit): EstablishmentEvent => ({
+          _tag: "Completed",
+          exit,
+        })),
       ),
       waitForEstablishmentInterrupt().pipe(
-        Effect.map(
-          (resetRetry): EstablishmentEvent => ({
-            _tag: "Interrupted",
-            resetRetry,
-          }),
-        ),
+        Effect.map((resetRetry): EstablishmentEvent => ({
+          _tag: "Interrupted",
+          resetRetry,
+        })),
       ),
       Effect.sleep(CONNECTION_ESTABLISHMENT_TIMEOUT).pipe(
         Effect.as<EstablishmentEvent>({ _tag: "TimedOut" }),
@@ -584,24 +592,21 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (
       retryAt: null,
     });
 
-    const connectedExit = yield* Effect.raceFirst(
+    const connectedExit = yield* Effect.raceAllFirst([
       active.lease.session.closed.pipe(
-        Effect.mapError(
-          (error): TracedAttemptFailure => ({
-            error,
-            attemptSpan: active.attemptSpan,
-          }),
-        ),
+        Effect.mapError((error): TracedAttemptFailure => ({
+          error,
+          attemptSpan: active.attemptSpan,
+        })),
       ),
       monitorConnectedLease(active.lease).pipe(
-        Effect.mapError(
-          (error): TracedAttemptFailure => ({
-            error,
-            attemptSpan: active.attemptSpan,
-          }),
-        ),
+        Effect.mapError((error): TracedAttemptFailure => ({
+          error,
+          attemptSpan: active.attemptSpan,
+        })),
       ),
-    ).pipe(exitUnlessInterrupted);
+      waitForAuthorizationRefresh(active.lease.prepared),
+    ]).pipe(exitUnlessInterrupted);
     const connectedForMs = (yield* Clock.currentTimeMillis) - connectedAt;
     if (Exit.isSuccess(connectedExit)) {
       return {

packages/contracts/src/environmentHttp.test.ts

@@ -59,4 +59,16 @@ describe("environment HTTP errors", () => {
       expect(error.message).toContain(details[index]);
     });
   });
+
+  it("gives rejected sessions a recovery action", () => {
+    const error = new EnvironmentAuthInvalidError({
+      code: "auth_invalid",
+      reason: "invalid_credential",
+      traceId,
+    });
+
+    expect(error.message).toBe(
+      "This environment session is no longer valid (invalid_credential). Refresh the page or quit and reopen T3 Code.",
+    );
+  });
 });

packages/contracts/src/environmentHttp.ts

@@ -133,6 +133,9 @@ export class EnvironmentAuthInvalidError extends Schema.TaggedErrorClass<Environ
   }
 
   override get message(): string {
+    if (this.reason === "invalid_credential") {
+      return "This environment session is no longer valid (invalid_credential). Refresh the page or quit and reopen T3 Code.";
+    }
     return `The environment rejected this client's credentials (${this.reason}).`;
   }
 }