diff --git a/packages/client-runtime/src/connection/supervisor.test.ts b/packages/client-runtime/src/connection/supervisor.test.ts index a1acb6f245f51..d2d8da66d1315 100644 --- a/packages/client-runtime/src/connection/supervisor.test.ts +++ b/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; readonly ready?: (attempt: number) => Effect.Effect; readonly probe?: (attempt: number) => Effect.Effect; + readonly inboundFrames?: (attempt: number) => Effect.Effect; readonly initialConfig?: (attempt: number) => Effect.Effect; readonly checkRoute?: (route: ConnectionRoute) => Effect.Effect; }) { @@ -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(); + 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(); diff --git a/packages/client-runtime/src/connection/supervisor.ts b/packages/client-runtime/src/connection/supervisor.ts index d500fe6d8347c..92a0433e45fb0 100644 --- a/packages/client-runtime/src/connection/supervisor.ts +++ b/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); diff --git a/packages/client-runtime/src/rpc/session.test.ts b/packages/client-runtime/src/rpc/session.test.ts index 3d024b9dc75dc..ea4c27154932b 100644 --- a/packages/client-runtime/src/rpc/session.test.ts +++ b/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(); diff --git a/packages/client-runtime/src/rpc/session.ts b/packages/client-runtime/src/rpc/session.ts index a3618ffd1d2ac..f88629491a9bc 100644 --- a/packages/client-runtime/src/rpc/session.ts +++ b/packages/client-runtime/src/rpc/session.ts @@ -53,6 +53,8 @@ export interface RpcSession { ) => ServerConfigSubscription; readonly ready: Effect.Effect; readonly probe: Effect.Effect; + /** Frames received so far. Proof of life that needs no reply. */ + readonly inboundFrames?: Effect.Effect; readonly closed: Effect.Effect; } @@ -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),