fix(client-runtime): stop waiting 3 s on a socket iOS killed in the background

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

@@ -98,6 +98,11 @@ const LAN_THEN_RELAY_ENTRY: ConnectionCatalogEntry = {
   enabled: true,
 };
 
+const PROBE_CAPABLE_CONFIG = {
+  environment: { capabilities: { connectionProbe: true } },
+} as ServerConfig;
+const LEGACY_PROBE_CONFIG = { environment: { capabilities: {} } } as ServerConfig;
+
 function preparedFor(target: ConnectionTarget): PreparedConnection {
   return { ...PREPARED_CONNECTION, target };
 }
@@ -154,6 +159,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 inboundFrames?: (attempt: number) => Effect.Effect<number>;
   readonly initialConfig?: (attempt: number) => Effect.Effect<ServerConfig, ConnectionAttemptError>;
   readonly checkRoute?: (route: ConnectionRoute) => Effect.Effect<ConnectionDriver.RouteCheck>;
 }) {
@@ -210,6 +216,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,
+        inboundFrames: options?.inboundFrames?.(attempt) ?? Effect.succeed(0),
         closed: Deferred.await(closed),
       } satisfies RpcSession.RpcSession),
       () => Ref.update(releaseCount, (count) => count + 1),
@@ -719,6 +726,165 @@ describe("EnvironmentSupervisor", () => {
     }).pipe(Effect.provide(TestClock.layer())),
   );
 
+  it.effect("reconnects without waiting out a resume probe when the host answers afresh", () =>
+    Effect.gen(function* () {
+      const routeChecks = yield* Ref.make(0);
+      const harness = yield* makeHarness({
+        initialConfig: () => Effect.succeed(PROBE_CAPABLE_CONFIG),
+        probe: (attempt) => (attempt === 1 ? Effect.never : Effect.void),
+        checkRoute: () => Ref.update(routeChecks, (count) => count + 1).pipe(Effect.as("answered")),
+      });
+      const supervisor = yield* EnvironmentSupervisor.make(TARGET_ENTRY, {
+        initiallyDesired: true,
+      }).pipe(Effect.provide(harness.dependencies));
+
+      yield* awaitState(
+        supervisor.state,
+        (state) => state.phase === "connected" && state.generation === 1,
+      );
+      expect(yield* Ref.get(routeChecks)).toBe(0);
+      yield* harness.wake("application-active-probe");
+      yield* TestClock.adjust("500 millis");
+
+      yield* eventuallyState(
+        supervisor.state,
+        (state) => state.phase === "connected" && state.generation === 2,
+      );
+      expect(yield* Ref.get(routeChecks)).toBe(1);
+      expect(yield* Ref.get(harness.sessionCount)).toBe(2);
+      expect(yield* Ref.get(harness.releaseCount)).toBe(1);
+    }).pipe(Effect.provide(TestClock.layer())),
+  );
+
+  it.effect.each([
+    { name: "before the route check starts", probeMs: 100, checkMs: 0 },
+    { name: "while the route check is in flight", probeMs: 700, checkMs: 1_000 },
+  ])("keeps a socket that answers its resume probe $name", ({ probeMs, checkMs }) =>
+    Effect.gen(function* () {
+      const routeChecks = yield* Ref.make(0);
+      const probeAnswered = yield* Deferred.make<void>();
+      const harness = yield* makeHarness({
+        initialConfig: () => Effect.succeed(PROBE_CAPABLE_CONFIG),
+        probe: () =>
+          Effect.sleep(probeMs).pipe(Effect.andThen(Deferred.succeed(probeAnswered, undefined))),
+        checkRoute: () =>
+          Ref.update(routeChecks, (count) => count + 1).pipe(
+            Effect.andThen(Effect.sleep(checkMs)),
+            Effect.as("answered"),
+          ),
+      });
+      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(probeMs);
+      yield* Deferred.await(probeAnswered);
+      yield* TestClock.adjust("5 seconds");
+
+      expect(yield* Ref.get(routeChecks)).toBe(probeMs > 500 ? 1 : 0);
+      expect(yield* Ref.get(harness.sessionCount)).toBe(1);
+      expect(yield* Ref.get(harness.releaseCount)).toBe(0);
+      expect(yield* SubscriptionRef.get(supervisor.state)).toMatchObject({
+        phase: "connected",
+        generation: 1,
+      });
+    }).pipe(Effect.provide(TestClock.layer())),
+  );
+
+  it.effect(
+    "keeps waiting for the probe when the socket delivers a frame before the check answers",
+    () =>
+      Effect.gen(function* () {
+        const routeChecks = yield* Ref.make(0);
+        const frames = yield* Ref.make(0);
+        const harness = yield* makeHarness({
+          initialConfig: () => Effect.succeed(PROBE_CAPABLE_CONFIG),
+          probe: (attempt) => (attempt === 1 ? Effect.never : Effect.void),
+          inboundFrames: () => Ref.get(frames),
+          checkRoute: () =>
+            Ref.update(routeChecks, (count) => count + 1).pipe(Effect.as("answered")),
+        });
+        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("50 millis");
+        // A frame the kernel buffered before the path died still drains after resume.
+        yield* Ref.update(frames, (count) => count + 1);
+        yield* TestClock.adjust("2850 millis");
+        expect(yield* Ref.get(routeChecks)).toBe(1);
+        expect(yield* Ref.get(harness.sessionCount)).toBe(1);
+        expect((yield* SubscriptionRef.get(supervisor.state)).phase).toBe("connected");
+
+        yield* TestClock.adjust("100 millis");
+        yield* awaitState(
+          supervisor.state,
+          (state) => state.phase === "connected" && state.generation === 2,
+        );
+      }).pipe(Effect.provide(TestClock.layer())),
+  );
+
+  it.effect.each([
+    { name: "the host does not answer afresh either", check: "silent", capable: true },
+    { name: "the route cannot be checked, as over a relay", check: "unchecked", capable: true },
+    {
+      name: "an older host answers probes with its full config",
+      check: "answered",
+      capable: false,
+    },
+  ] as const)("waits out a resume probe when $name", ({ check, capable }) =>
+    Effect.gen(function* () {
+      const harness = yield* makeHarness({
+        initialConfig: () => Effect.succeed(capable ? PROBE_CAPABLE_CONFIG : LEGACY_PROBE_CONFIG),
+        probe: (attempt) => (attempt === 1 ? Effect.never : Effect.void),
+        checkRoute: () => Effect.succeed(check),
+      });
+      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("2900 millis");
+      expect(yield* Ref.get(harness.sessionCount)).toBe(1);
+      expect((yield* SubscriptionRef.get(supervisor.state)).phase).toBe("connected");
+
+      yield* TestClock.adjust("100 millis");
+      yield* awaitState(
+        supervisor.state,
+        (state) => state.phase === "connected" && state.generation === 2,
+      );
+    }).pipe(Effect.provide(TestClock.layer())),
+  );
+
+  it.effect.each(["network-changed", "retry"] as const)(
+    "does not check the route during a %s probe",
+    (trigger) =>
+      Effect.gen(function* () {
+        const routeChecks = yield* Ref.make(0);
+        const harness = yield* makeHarness({
+          initialConfig: () => Effect.succeed(PROBE_CAPABLE_CONFIG),
+          probe: (attempt) => (attempt === 1 ? Effect.never : Effect.void),
+          checkRoute: () =>
+            Ref.update(routeChecks, (count) => count + 1).pipe(Effect.as("answered")),
+        });
+        const supervisor = yield* EnvironmentSupervisor.make(TARGET_ENTRY, {
+          initiallyDesired: true,
+        }).pipe(Effect.provide(harness.dependencies));
+
+        yield* awaitState(supervisor.state, (state) => state.phase === "connected");
+        if (trigger === "retry") yield* supervisor.retryNow;
+        else yield* harness.wake("network-changed");
+        yield* TestClock.adjust("2900 millis");
+        expect(yield* Ref.get(routeChecks)).toBe(0);
+        expect(yield* Ref.get(harness.sessionCount)).toBe(1);
+      }).pipe(Effect.provide(TestClock.layer())),
+  );
+
   it.effect("reconnects immediately when the foreground liveness probe fails", () =>
     Effect.gen(function* () {
       const allowReconnect = yield* Deferred.make<void>();

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

@@ -47,6 +47,10 @@ const establishmentTimeout = Duration.fromInputUnsafe(CONNECTION_ESTABLISHMENT_T
 const synchronizationTimeout = Duration.fromInputUnsafe("75 seconds");
 const CONNECTION_PROBE_TIMEOUT = "15 seconds";
 const QUICK_CONNECTION_PROBE_TIMEOUT = "3 seconds";
+// A socket that has delivered nothing for this long after a resume probe,
+// while a fresh request to the same address answers, has died, typically
+// suspended by iOS without a close event.
+const RESUME_PROBE_ROUTE_CHECK_AFTER = "500 millis";
 const BACKOFF_RESET_AFTER_MS = 30_000;
 const BETTER_ROUTE_CHECK_INTERVAL = "60 seconds";
 // A better route that answered the check but then failed to connect is not
@@ -584,6 +588,33 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (
     }
   };
 
+  const checkLeaseRoute = Effect.fnUntraced(function* (
+    lease: ConnectionDriver.EnvironmentConnectionLease,
+    framesAtProbe: number | undefined,
+  ) {
+    const inboundFrames = lease.session.inboundFrames;
+    if (inboundFrames === undefined || framesAtProbe === undefined) return "unchecked" as const;
+    // An older host answers probes with its full config, which can be slow
+    // enough to lose to a fresh descriptor request on a healthy socket.
+    const answersCheapProbe = yield* lease.session.initialConfig.pipe(
+      Effect.map((config) => config.environment.capabilities.connectionProbe === true),
+      Effect.catchCause(() => Effect.succeed(false)),
+    );
+    if (!answersCheapProbe) return "unchecked" as const;
+    const latest = yield* Ref.get(currentEntry);
+    const route = connectionRoutes(latest).find(
+      (candidate) =>
+        connectionRouteId(candidate.target) === connectionRouteId(lease.prepared.target),
+    );
+    if (route === undefined) return "unchecked" as const;
+    const check = yield* driver.checkRoute(latest, route);
+    // Any frame since the probe started, even one buffered before the path
+    // died, vetoes the shortcut; the probe's own reply or timeout decides.
+    return check === "answered" && (yield* inboundFrames) === framesAtProbe
+      ? ("answered" as const)
+      : ("unchecked" as const);
+  });
+
   // Returning to the app, an explicit retry, and the network reporting offline
   // all probe the live session instead of replacing it, so a healthy socket is
   // not torn down (the offline report is often wrong, for example for a
@@ -622,7 +653,23 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (
         continue;
       }
       yield* Ref.set(probeUnanswered, true);
+      const framesAtProbe =
+        lease.session.inboundFrames === undefined ? undefined : yield* lease.session.inboundFrames;
       const probe = yield* Effect.forkChild(lease.session.probe);
+      // iOS can kill a suspended socket without a close event, so a resume
+      // probe would otherwise wait out its whole timeout before reconnecting.
+      // Only the signal that starts a probe can start this check: a resume
+      // that arrives during another probe just shortens its deadline below.
+      const routeCheck = yield* Effect.forkChild(
+        next._tag === "Wakeup" && next.reason === "application-active-probe"
+          ? Effect.sleep(RESUME_PROBE_ROUTE_CHECK_AFTER).pipe(
+              Effect.andThen(checkLeaseRoute(lease, framesAtProbe)),
+            )
+          : Effect.never,
+      );
+      const stopProbe = Effect.all([Fiber.interrupt(probe), Fiber.interrupt(routeCheck)], {
+        discard: true,
+      });
       let deadline = (yield* Clock.monotonicTimeNanos) + Duration.toNanosUnsafe(probeTimeout);
       for (;;) {
         const remaining = deadline - (yield* Clock.monotonicTimeNanos);
@@ -634,15 +681,28 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (
           Effect.sleep(Duration.nanos(remaining > 0n ? remaining : 0n)).pipe(
             Effect.as({ _tag: "TimedOut" as const }),
           ),
+          // A check that answers after the deadline changes nothing: the
+          // probe has already failed with the timeout.
+          Fiber.join(routeCheck).pipe(
+            Effect.flatMap((check) =>
+              check === "answered"
+                ? Effect.succeed({ _tag: "SocketSilent" as const })
+                : Effect.never,
+            ),
+          ),
         ]);
-        if (probeEvent._tag === "TimedOut") {
-          yield* Fiber.interrupt(probe);
+        if (probeEvent._tag === "TimedOut" || probeEvent._tag === "SocketSilent") {
+          yield* stopProbe;
           return yield* new ConnectionTransientError({
             reason: "timeout",
-            detail: `${target.label} did not respond to a connection health check.`,
+            detail:
+              probeEvent._tag === "SocketSilent"
+                ? `${target.label} answered a fresh request but not over its connection.`
+                : `${target.label} did not respond to a connection health check.`,
           });
         }
         if (probeEvent._tag === "ProbeCompleted") {
+          yield* Fiber.interrupt(routeCheck);
           if (Exit.isSuccess(probeEvent.exit)) {
             yield* Ref.set(probeUnanswered, false);
           }
@@ -651,7 +711,7 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (
         }
         const endDuringProbe = yield* connectedLeaseEnd(probeEvent.signal, lease);
         if (endDuringProbe !== undefined) {
-          yield* Fiber.interrupt(probe);
+          yield* stopProbe;
           return endDuringProbe === "reset";
         }
         const signalTimeout = probeTimeoutFor(probeEvent.signal);

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

@@ -21,6 +21,7 @@ import * as Schema from "effect/Schema";
 import * as Stream from "effect/Stream";
 import * as SubscriptionRef from "effect/SubscriptionRef";
 import * as Socket from "effect/socket/Socket";
+import * as TestClock from "effect/testing/TestClock";
 
 import {
   AVAILABLE_CONNECTION_STATE,
@@ -343,6 +344,31 @@ describe("RpcSessionFactory", () => {
     }),
   );
 
+  it.effect("counts inbound frames without letting them answer a probe", () =>
+    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 inboundFrames = session.inboundFrames ?? Effect.die("missing inbound frame count");
+
+      const probeFiber = yield* Effect.forkChild(session.probe);
+      yield* awaitRequest(socket, 1);
+      const before = yield* inboundFrames;
+      socket.serverMessage(encodeJson({ _tag: "Pong" }));
+      expect(yield* inboundFrames).toBe(before + 1);
+
+      const outcome = yield* Effect.forkChild(
+        Fiber.join(probeFiber).pipe(Effect.timeoutOption("3 seconds")),
+      );
+      yield* TestClock.adjust("3 seconds");
+      expect(Option.isNone(yield* Fiber.join(outcome))).toBe(true);
+    }),
+  );
+
   it.effect("closes the websocket when the session scope is released", () =>
     Effect.gen(function* () {
       const { factory, sockets } = yield* makeFactory();

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

@@ -53,6 +53,8 @@ export interface RpcSession {
   ) => ServerConfigSubscription;
   readonly ready: Effect.Effect<void, ConnectionAttemptError>;
   readonly probe: Effect.Effect<void, ConnectionAttemptError>;
+  /** Frames received so far. Proof of life that needs no reply. */
+  readonly inboundFrames?: Effect.Effect<number>;
   readonly closed: Effect.Effect<never, ConnectionAttemptError>;
 }
 
@@ -206,9 +208,22 @@ export const make = Effect.fn("RpcSessionFactory.make")(function* (
         Effect.asVoid,
       ),
     });
+    // The protocol never reopens its socket (no retries), so every counted
+    // frame came from this session's one socket.
+    let inboundFrames = 0;
     const layerSocket = Socket.layerWebSocket(connection.socketUrl, {
       openTimeout: SOCKET_OPEN_TIMEOUT,
-    }).pipe(Layer.provide(Layer.succeed(Socket.WebSocketConstructor, webSocketConstructor)));
+    }).pipe(
+      Layer.provide(
+        Layer.succeed(Socket.WebSocketConstructor, (url, options) => {
+          const socket = webSocketConstructor(url, options);
+          socket.addEventListener("message", () => {
+            inboundFrames += 1;
+          });
+          return socket;
+        }),
+      ),
+    );
     const layerProtocol = Layer.effect(
       RpcClient.Protocol,
       RpcClient.makeProtocolSocket({
@@ -412,6 +427,7 @@ export const make = Effect.fn("RpcSessionFactory.make")(function* (
         Effect.raceFirst(Deferred.await(disconnected)),
       ),
       probe,
+      inboundFrames: Effect.sync(() => inboundFrames),
       closed: Effect.raceFirst(
         Deferred.await(disconnected),
         Deferred.await(configSubscriptionClosed),