fix(client): slow networks no longer drop working connections

apps/mobile/src/connection/app-state-wakeups.test.ts

@@ -6,8 +6,11 @@ import {
 } from "./app-state-wakeups";
 
 describe("mobileApplicationActiveWakeup", () => {
-  it("uses a fast probe after a short interruption", () => {
-    expect(mobileApplicationActiveWakeup(null, 20_000)).toBe("application-active-probe");
+  it("skips the probe when the app never left the foreground", () => {
+    expect(mobileApplicationActiveWakeup(null, 20_000)).toBe("application-refocused");
+  });
+
+  it("uses a fast probe after a short background visit", () => {
     expect(
       mobileApplicationActiveWakeup(20_000, 20_000 + MOBILE_BACKGROUND_RECONNECT_AFTER_MS - 1),
     ).toBe("application-active-probe");

apps/mobile/src/connection/app-state-wakeups.ts

@@ -4,15 +4,20 @@ export const MOBILE_BACKGROUND_RECONNECT_AFTER_MS = 10_000;
 
 export type MobileApplicationActiveWakeup = Extract<
   Wakeups.ConnectionWakeup,
-  "application-active-probe" | "application-active-reconnect"
+  "application-active-probe" | "application-active-reconnect" | "application-refocused"
 >;
 
+/**
+ * The wakeup for returning to the foreground. Without a background visit the
+ * app only went inactive, such as for Control Center or Face ID, and its
+ * socket was never suspended, so it skips the probe.
+ */
 export function mobileApplicationActiveWakeup(
   backgroundedAtMs: number | null,
   activeAtMs: number,
 ): MobileApplicationActiveWakeup {
-  return backgroundedAtMs !== null &&
-    activeAtMs - backgroundedAtMs >= MOBILE_BACKGROUND_RECONNECT_AFTER_MS
+  if (backgroundedAtMs === null) return "application-refocused";
+  return activeAtMs - backgroundedAtMs >= MOBILE_BACKGROUND_RECONNECT_AFTER_MS
     ? "application-active-reconnect"
     : "application-active-probe";
 }

apps/mobile/src/connection/platform.ts

@@ -27,7 +27,10 @@ import { appAtomRegistry } from "../state/atom-registry";
 import { clearThreadOutboxEnvironment } from "../state/thread-outbox-removal";
 import { clearComposerDraftsEnvironment } from "../state/use-composer-drafts";
 import { clearThreadComposerErrorsForEnvironment } from "../state/thread-composer-error";
-import { mobileApplicationActiveWakeup } from "./app-state-wakeups";
+import {
+  mobileApplicationActiveWakeup,
+  type MobileApplicationActiveWakeup,
+} from "./app-state-wakeups";
 import * as ConnectionStorage from "./storage";
 
 function networkStatus(state: Network.NetworkState): "unknown" | "offline" | "online" {
@@ -124,7 +127,7 @@ const networkPathChanges = Stream.callback<"network-changed">((queue) =>
 const layerWakeups = Wakeups.layer({
   changes: Stream.mergeAll(
     [
-      Stream.callback<"application-active-probe" | "application-active-reconnect">((queue) =>
+      Stream.callback<MobileApplicationActiveWakeup>((queue) =>
         Effect.acquireRelease(
           Effect.sync(() => {
             let backgroundedAtMs = AppState.currentState === "background" ? Date.now() : null;

apps/web/src/cloud/linkEnvironment.test.ts

@@ -91,6 +91,7 @@ function layerRegistry(options?: {
         subscribeServerConfig: (input) => client.subscribeServerConfig(input),
         ready: Effect.void,
         probe: Effect.void,
+        receivedReads: Effect.succeed(0),
         closed: Effect.never,
       };
       const target = new PrimaryConnectionTarget({

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

@@ -86,7 +86,9 @@ export class RemoteEnvironmentAuthorization extends Context.Service<
   }
 >()("@t3tools/client-runtime/authorization/service/RemoteEnvironmentAuthorization") {}
 
-const CACHED_ENDPOINT_SOCKET_TIMEOUT_MS = 3_000;
+// A cached token whose ticket fails on the T3 Connect route is replaced, so
+// this must outlast a slow but working link, not just a healthy one.
+const CACHED_ENDPOINT_SOCKET_TIMEOUT_MS = 6_000;
 const BEARER_DESCRIPTOR_CACHE_TTL_MS = 10_000;
 const DPOP_AUTHORIZATION_TIMEOUT_MS = 30_000;
 

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

@@ -420,6 +420,7 @@ const makeHarness = Effect.fn("TestEnvironmentRegistry.makeHarness")(function* (
             Stream.die(new Error("Config is not used by registry tests.")),
           ready: Effect.void,
           probe: Effect.void,
+          receivedReads: Effect.succeed(0),
           closed: Deferred.await(closed),
         } satisfies RpcSession.RpcSession),
         () => Ref.update(releasedSessions, (count) => count + 1),

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

@@ -157,6 +157,7 @@ const makeHarness = Effect.fn("TestConnectionHarness.make")(function* (options?:
   ) => Effect.Effect<PreparedConnection, ConnectionAttemptError>;
   readonly ready?: (attempt: number) => Effect.Effect<void, ConnectionAttemptError>;
   readonly probe?: (attempt: number) => Effect.Effect<void, ConnectionAttemptError>;
+  readonly receivedReads?: Effect.Effect<number>;
   readonly initialConfig?: (attempt: number) => Effect.Effect<ServerConfig, ConnectionAttemptError>;
   readonly checkRoute?: (route: ConnectionRoute) => Effect.Effect<ConnectionDriver.RouteCheck>;
 }) {
@@ -213,6 +214,7 @@ const makeHarness = Effect.fn("TestConnectionHarness.make")(function* (options?:
         subscribeServerConfig: (input) => TEST_RPC_CLIENT.subscribeServerConfig(input),
         ready: options?.ready?.(attempt) ?? Effect.void,
         probe: options?.probe?.(attempt) ?? Effect.void,
+        receivedReads: options?.receivedReads ?? Effect.succeed(0),
         closed: Deferred.await(closed),
       } satisfies RpcSession.RpcSession),
       () => Ref.update(releaseCount, (count) => count + 1),
@@ -626,6 +628,42 @@ describe("EnvironmentSupervisor", () => {
     }),
   );
 
+  it.effect("a mobile refocus restarts the retry ladder but never probes a live session", () =>
+    Effect.gen(function* () {
+      const probeCount = yield* Ref.make(0);
+      const harness = yield* makeHarness({
+        prepare: (attempt) =>
+          attempt <= 2 ? Effect.fail(transient()) : Effect.succeed(PREPARED_CONNECTION),
+        probe: () => Ref.update(probeCount, (count) => count + 1),
+      });
+      const supervisor = yield* EnvironmentSupervisor.make(TARGET_ENTRY, {
+        initiallyDesired: true,
+      }).pipe(Effect.provide(harness.dependencies));
+
+      yield* awaitState(
+        supervisor.state,
+        (state) => state.phase === "backoff" && state.attempt === 1,
+      );
+      yield* TestClock.adjust("3 seconds");
+      yield* eventuallyState(
+        supervisor.state,
+        (state) => state.phase === "backoff" && state.attempt === 2,
+      );
+      yield* harness.wake("application-refocused");
+      yield* eventuallyState(
+        supervisor.state,
+        (state) => state.phase === "connected" && state.attempt === 1,
+      );
+
+      yield* harness.wake("application-refocused");
+      for (let attempt = 0; attempt < 100; attempt += 1) {
+        yield* Effect.yieldNow;
+      }
+      expect(yield* Ref.get(probeCount)).toBe(0);
+      expect(yield* Ref.get(harness.sessionCount)).toBe(1);
+    }).pipe(Effect.provide(TestClock.layer())),
+  );
+
   it.effect("explicit retry starts a fresh backoff sequence", () =>
     Effect.gen(function* () {
       const harness = yield* makeHarness({
@@ -800,7 +838,7 @@ describe("EnvironmentSupervisor", () => {
           (state) => state.phase === "connected" && state.generation === 1,
         );
         yield* harness.setNetworkStatus("offline");
-        yield* TestClock.adjust("3 seconds");
+        yield* TestClock.adjust("6 seconds");
         yield* awaitState(supervisor.state, (state) => state.phase === "offline");
 
         expect(yield* Ref.get(harness.releaseCount)).toBe(1);
@@ -1313,8 +1351,8 @@ describe("EnvironmentSupervisor", () => {
       yield* harness.wake("application-active");
       yield* TestClock.adjust("5 seconds");
       yield* supervisor.retryNow;
-      // The retry's 3 second limit applies, not the 10 seconds left of the 15.
-      yield* TestClock.adjust("2999 millis");
+      // The retry's 6 second limit applies, not the 10 seconds left of the 15.
+      yield* TestClock.adjust("5999 millis");
       expect(yield* Ref.get(harness.sessionCount)).toBe(1);
       yield* TestClock.adjust("1 milli");
       yield* awaitState(
@@ -1326,6 +1364,35 @@ describe("EnvironmentSupervisor", () => {
     }).pipe(Effect.provide(TestClock.layer())),
   );
 
+  it.effect("extends a retry-shortened probe by the retry's window when data arrives", () =>
+    Effect.gen(function* () {
+      const frames = yield* Ref.make(0);
+      const harness = yield* makeHarness({
+        probe: (attempt) => (attempt === 1 ? Effect.never : Effect.void),
+        receivedReads: Ref.get(frames),
+      });
+      const supervisor = yield* EnvironmentSupervisor.make(TARGET_ENTRY, {
+        initiallyDesired: true,
+      }).pipe(Effect.provide(harness.dependencies));
+
+      yield* awaitState(supervisor.state, (state) => state.phase === "connected");
+      yield* harness.wake("application-active");
+      yield* TestClock.adjust("1 second");
+      yield* supervisor.retryNow;
+      yield* TestClock.adjust("1 second");
+      yield* Ref.set(frames, 1);
+      // The retry's deadline lands at 7 seconds and the data extends it by the
+      // retry's 6 seconds to 13, not by the original 15.
+      yield* TestClock.adjust("10999 millis");
+      expect(yield* Ref.get(harness.sessionCount)).toBe(1);
+      yield* TestClock.adjust("1 milli");
+      yield* awaitState(
+        supervisor.state,
+        (state) => state.phase === "connected" && state.generation === 2,
+      );
+    }).pipe(Effect.provide(TestClock.layer())),
+  );
+
   it.effect("quickly times out a stalled mobile foreground liveness probe", () =>
     Effect.gen(function* () {
       const harness = yield* makeHarness({
@@ -1337,7 +1404,7 @@ describe("EnvironmentSupervisor", () => {
 
       yield* awaitState(supervisor.state, (state) => state.phase === "connected");
       yield* harness.wake("application-active-probe");
-      yield* TestClock.adjust("3 seconds");
+      yield* TestClock.adjust("6 seconds");
       // The timed-out wake probe reconnects immediately without a backoff
       // sleep: no further clock advance is needed.
       yield* awaitState(
@@ -1349,6 +1416,62 @@ describe("EnvironmentSupervisor", () => {
     }).pipe(Effect.provide(TestClock.layer())),
   );
 
+  it.effect("keeps waiting on a quick probe while data still arrives, up to the full timeout", () =>
+    Effect.gen(function* () {
+      const frames = yield* Ref.make(0);
+      const harness = yield* makeHarness({
+        probe: (attempt) => (attempt === 1 ? Effect.never : Effect.void),
+        receivedReads: Ref.get(frames),
+      });
+      const supervisor = yield* EnvironmentSupervisor.make(TARGET_ENTRY, {
+        initiallyDesired: true,
+      }).pipe(Effect.provide(harness.dependencies));
+
+      yield* awaitState(supervisor.state, (state) => state.phase === "connected");
+      yield* harness.wake("application-active-probe");
+      yield* TestClock.adjust("5 seconds");
+      yield* Ref.set(frames, 1);
+      yield* TestClock.adjust("6 seconds");
+      expect(yield* Ref.get(harness.sessionCount)).toBe(1);
+
+      // Frames keep arriving, but the probe still gets no more than 15 seconds.
+      yield* Ref.set(frames, 2);
+      yield* TestClock.adjust("3999 millis");
+      expect(yield* Ref.get(harness.sessionCount)).toBe(1);
+      yield* Ref.set(frames, 3);
+      yield* TestClock.adjust("1 milli");
+      yield* awaitState(
+        supervisor.state,
+        (state) => state.phase === "connected" && state.generation === 2,
+      );
+    }).pipe(Effect.provide(TestClock.layer())),
+  );
+
+  it.effect("times out a quick probe once data stops arriving", () =>
+    Effect.gen(function* () {
+      const frames = yield* Ref.make(0);
+      const harness = yield* makeHarness({
+        probe: (attempt) => (attempt === 1 ? Effect.never : Effect.void),
+        receivedReads: Ref.get(frames),
+      });
+      const supervisor = yield* EnvironmentSupervisor.make(TARGET_ENTRY, {
+        initiallyDesired: true,
+      }).pipe(Effect.provide(harness.dependencies));
+
+      yield* awaitState(supervisor.state, (state) => state.phase === "connected");
+      yield* harness.wake("application-active-probe");
+      yield* TestClock.adjust("1 second");
+      yield* Ref.set(frames, 1);
+      yield* TestClock.adjust("5 seconds");
+      expect(yield* Ref.get(harness.sessionCount)).toBe(1);
+      yield* TestClock.adjust("6 seconds");
+      yield* awaitState(
+        supervisor.state,
+        (state) => state.phase === "connected" && state.generation === 2,
+      );
+    }).pipe(Effect.provide(TestClock.layer())),
+  );
+
   it.effect("honors an explicit disconnect while a foreground probe is stalled", () =>
     Effect.gen(function* () {
       const probeStarted = yield* Deferred.make<void>();
@@ -1953,7 +2076,7 @@ describe("EnvironmentSupervisor routes", () => {
 
       yield* Ref.set(lanReachable, false);
       yield* harness.wake("network-changed");
-      yield* TestClock.adjust("3 seconds");
+      yield* TestClock.adjust("6 seconds");
 
       yield* awaitState(
         supervisor.state,

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

@@ -43,8 +43,12 @@ const CONNECTION_ESTABLISHMENT_TIMEOUT = "15 seconds";
 const establishmentTimeout = Duration.fromInputUnsafe(CONNECTION_ESTABLISHMENT_TIMEOUT);
 const CONNECTION_PROBE_TIMEOUT = "15 seconds";
 // Mobile resumes, explicit retries, and offline events want a fast answer:
-// the user is waiting, or the network may be gone.
-const QUICK_CONNECTION_PROBE_TIMEOUT = "3 seconds";
+// the user is waiting, or the network may be gone. It still has to cover a
+// lost packet and its retransmit on a 500ms round trip, or a slow link reads
+// as a dead socket and every resume tears down a working session. It must
+// stay longer than PING_INTERVAL plus a round trip, so a working link always
+// delivers a frame that extends it.
+const QUICK_CONNECTION_PROBE_TIMEOUT = "6 seconds";
 const BACKOFF_RESET_AFTER_MS = 30_000;
 // While connected over a fallback route, how often to look for a better one.
 // Network changes and returning to the app also trigger a check.
@@ -632,7 +636,11 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (
       yield* Ref.set(probeUnanswered, true);
       const probe = yield* Effect.forkChild(lease.session.probe);
       // Monotonic nanoseconds, so a wall-clock correction cannot move the deadline.
-      let deadline = (yield* Clock.monotonicTimeNanos) + Duration.toNanosUnsafe(probeTimeout);
+      const probeStartedAt = yield* Clock.monotonicTimeNanos;
+      const latestDeadline = probeStartedAt + Duration.toNanosUnsafe(CONNECTION_PROBE_TIMEOUT);
+      let window = probeTimeout;
+      let deadline = probeStartedAt + Duration.toNanosUnsafe(window);
+      let receivedReads = yield* lease.session.receivedReads;
       for (;;) {
         const remaining = deadline - (yield* Clock.monotonicTimeNanos);
         const probeEvent = yield* Effect.raceAllFirst([
@@ -645,6 +653,17 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (
           ),
         ]);
         if (probeEvent._tag === "TimedOut") {
+          // Data still arriving means a slow link, not a dead one: the answer
+          // may be queued behind it. Wait another window, up to the full
+          // probe timeout.
+          const received = yield* lease.session.receivedReads;
+          const now = yield* Clock.monotonicTimeNanos;
+          if (received > receivedReads && now < latestDeadline) {
+            receivedReads = received;
+            const extended = now + Duration.toNanosUnsafe(window);
+            deadline = extended < latestDeadline ? extended : latestDeadline;
+            continue;
+          }
           yield* Fiber.interrupt(probe);
           return yield* new ConnectionTransientError({
             reason: "timeout",
@@ -669,7 +688,10 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (
         if (signalTimeout !== undefined) {
           const signalDeadline =
             (yield* Clock.monotonicTimeNanos) + Duration.toNanosUnsafe(signalTimeout);
-          if (signalDeadline < deadline) deadline = signalDeadline;
+          if (signalDeadline < deadline) {
+            deadline = signalDeadline;
+            window = signalTimeout;
+          }
         }
       }
     }

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

@@ -6,6 +6,10 @@ export type ConnectionWakeup =
   | "application-active"
   | "application-active-probe"
   | "application-active-reconnect"
+  // The app came back from a brief inactive state, such as Control Center or
+  // Face ID, without being suspended. Its socket is fine, but a pending retry
+  // should still run now.
+  | "application-refocused"
   | "credentials-changed"
   // The device moved to a different network (another Wi-Fi, Wi-Fi to
   // cellular) while staying online. Saved routes may have changed reach.
@@ -15,7 +19,8 @@ function isApplicationActiveWakeup(reason: ConnectionWakeup): boolean {
   return (
     reason === "application-active" ||
     reason === "application-active-probe" ||
-    reason === "application-active-reconnect"
+    reason === "application-active-reconnect" ||
+    reason === "application-refocused"
   );
 }
 

packages/client-runtime/src/operations/commands.performance.test.ts

@@ -186,6 +186,7 @@ const makeSupervisor = Effect.fn("CommandPerformance.makeSupervisor")(function*
     subscribeServerConfig: (requestInput) => client.subscribeServerConfig(requestInput),
     ready: Effect.void,
     probe: Effect.void,
+    receivedReads: Effect.succeed(0),
     closed: Effect.never,
   };
   return EnvironmentSupervisor.EnvironmentSupervisor.of({

packages/client-runtime/src/operations/commands.test.ts

@@ -135,6 +135,7 @@ const makeSupervisor = Effect.fn("TestEnvironmentCommands.makeSupervisor")(funct
     subscribeServerConfig: (input) => client.subscribeServerConfig(input),
     ready: Effect.void,
     probe: Effect.void,
+    receivedReads: Effect.succeed(0),
     closed: Effect.never,
   };
   return EnvironmentSupervisor.EnvironmentSupervisor.of({

packages/client-runtime/src/rpc/client.test.ts

@@ -61,6 +61,7 @@ function session(client: WsRpcProtocolClient): RpcSession.RpcSession {
     subscribeServerConfig: (input) => client.subscribeServerConfig(input),
     ready: Effect.void,
     probe: Effect.void,
+    receivedReads: Effect.succeed(0),
     closed: Effect.never,
   };
 }

packages/client-runtime/src/rpc/protocol.ts

@@ -7,6 +7,9 @@ type RpcClientFactory = typeof makeWsRpcProtocolClient;
 export type WsRpcProtocolClient =
   RpcClientFactory extends Effect.Effect<infer Client, any, any> ? Client : never;
 
-// Pings go out every 5 seconds, and a server busy for a few seconds stops
-// answering them. Allow 15 seconds without any frame before dropping the socket.
+// Each pong is a frame, so an idle but working socket shows a received
+// frame inside every quick connection probe window. A server busy for a few
+// seconds stops answering pings, so allow 15 seconds without any frame before
+// dropping the socket.
+export const PING_INTERVAL = "3 seconds";
 export const PING_TIMEOUT = "15 seconds";