diff --git a/apps/mobile/src/connection/app-state-wakeups.test.ts b/apps/mobile/src/connection/app-state-wakeups.test.ts index 4e8bdf1edf4a..cf961edaaafc 100644 --- a/apps/mobile/src/connection/app-state-wakeups.test.ts +++ b/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"); diff --git a/apps/mobile/src/connection/app-state-wakeups.ts b/apps/mobile/src/connection/app-state-wakeups.ts index 185ff744958e..e371291a16c1 100644 --- a/apps/mobile/src/connection/app-state-wakeups.ts +++ b/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"; } diff --git a/apps/mobile/src/connection/platform.ts b/apps/mobile/src/connection/platform.ts index e8817de69eeb..41c0cc1b4ddb 100644 --- a/apps/mobile/src/connection/platform.ts +++ b/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((queue) => Effect.acquireRelease( Effect.sync(() => { let backgroundedAtMs = AppState.currentState === "background" ? Date.now() : null; diff --git a/apps/web/src/cloud/linkEnvironment.test.ts b/apps/web/src/cloud/linkEnvironment.test.ts index dbcaacfa25a4..319be620d29c 100644 --- a/apps/web/src/cloud/linkEnvironment.test.ts +++ b/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({ diff --git a/packages/client-runtime/src/authorization/service.ts b/packages/client-runtime/src/authorization/service.ts index 780c3a77fdc0..00aca960f00e 100644 --- a/packages/client-runtime/src/authorization/service.ts +++ b/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; diff --git a/packages/client-runtime/src/connection/registry.test.ts b/packages/client-runtime/src/connection/registry.test.ts index 821e244fd544..6fb6721988c6 100644 --- a/packages/client-runtime/src/connection/registry.test.ts +++ b/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), diff --git a/packages/client-runtime/src/connection/supervisor.test.ts b/packages/client-runtime/src/connection/supervisor.test.ts index 37adf87c1b47..decd0aebf3cf 100644 --- a/packages/client-runtime/src/connection/supervisor.test.ts +++ b/packages/client-runtime/src/connection/supervisor.test.ts @@ -157,6 +157,7 @@ const makeHarness = Effect.fn("TestConnectionHarness.make")(function* (options?: ) => Effect.Effect; readonly ready?: (attempt: number) => Effect.Effect; readonly probe?: (attempt: number) => Effect.Effect; + readonly receivedReads?: Effect.Effect; readonly initialConfig?: (attempt: number) => Effect.Effect; readonly checkRoute?: (route: ConnectionRoute) => Effect.Effect; }) { @@ -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(); @@ -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, diff --git a/packages/client-runtime/src/connection/supervisor.ts b/packages/client-runtime/src/connection/supervisor.ts index 67f8daae691a..798b0a2aa9fa 100644 --- a/packages/client-runtime/src/connection/supervisor.ts +++ b/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; + } } } } diff --git a/packages/client-runtime/src/connection/wakeups.ts b/packages/client-runtime/src/connection/wakeups.ts index 38a45b092bbe..bf6e507faeea 100644 --- a/packages/client-runtime/src/connection/wakeups.ts +++ b/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" ); } diff --git a/packages/client-runtime/src/operations/commands.performance.test.ts b/packages/client-runtime/src/operations/commands.performance.test.ts index 112560bf5f34..a1a6949dbae7 100644 --- a/packages/client-runtime/src/operations/commands.performance.test.ts +++ b/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({ diff --git a/packages/client-runtime/src/operations/commands.test.ts b/packages/client-runtime/src/operations/commands.test.ts index 2c38e18f3e5c..2b1636dd194c 100644 --- a/packages/client-runtime/src/operations/commands.test.ts +++ b/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({ diff --git a/packages/client-runtime/src/rpc/client.test.ts b/packages/client-runtime/src/rpc/client.test.ts index cd3053c4bb25..c1362c4031e8 100644 --- a/packages/client-runtime/src/rpc/client.test.ts +++ b/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, }; } diff --git a/packages/client-runtime/src/rpc/protocol.ts b/packages/client-runtime/src/rpc/protocol.ts index e8aea66d3d89..2d3b1503e9fc 100644 --- a/packages/client-runtime/src/rpc/protocol.ts +++ b/packages/client-runtime/src/rpc/protocol.ts @@ -7,6 +7,9 @@ type RpcClientFactory = typeof makeWsRpcProtocolClient; export type WsRpcProtocolClient = RpcClientFactory extends Effect.Effect ? 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"; diff --git a/packages/client-runtime/src/rpc/session.test.ts b/packages/client-runtime/src/rpc/session.test.ts index 38e1b7f89d31..64638fc022dc 100644 --- a/packages/client-runtime/src/rpc/session.test.ts +++ b/packages/client-runtime/src/rpc/session.test.ts @@ -1099,13 +1099,9 @@ describe("RpcSessionFactory", () => { yield* TestClock.adjust("15 seconds"); expect(closedFiber.pollUnsafe()).toBeUndefined(); - expect(socket.sent.map((message) => decodeJson(message)).filter(isPing)).toEqual([ - { _tag: "Ping" }, - { _tag: "Ping" }, - { _tag: "Ping" }, - ]); + expect(socket.sent.map((message) => decodeJson(message)).filter(isPing)).toHaveLength(5); - yield* TestClock.adjust("5 seconds"); + yield* TestClock.adjust("3 seconds"); const error = yield* Fiber.join(closedFiber); expect(error).toBeInstanceOf(ConnectionTransientError); expect(error).toMatchObject({ @@ -1115,6 +1111,25 @@ describe("RpcSessionFactory", () => { }).pipe(Effect.scoped, Effect.provide(TestClock.layer())), ); + it.effect("counts every socket read, pongs included", () => + 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 before = yield* session.receivedReads; + expect(before).toBeGreaterThan(0); + yield* TestClock.adjust("3 seconds"); + socket.serverMessage(encodeJson({ _tag: "Pong" })); + yield* Effect.yieldNow; + expect(yield* session.receivedReads).toBe(before + 1); + }).pipe(Effect.scoped, Effect.provide(TestClock.layer())), + ); + it.effect("keeps reading replies after closing a stream with a full buffer", () => 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 9f83e7c6019d..cb7f1a93f3b8 100644 --- a/packages/client-runtime/src/rpc/session.ts +++ b/packages/client-runtime/src/rpc/session.ts @@ -24,7 +24,12 @@ import * as RpcClientError from "effect/rpc/RpcClientError"; import * as RpcSerialization from "effect/rpc/RpcSerialization"; import * as Socket from "effect/socket/Socket"; -import { makeWsRpcProtocolClient, PING_TIMEOUT, type WsRpcProtocolClient } from "./protocol.ts"; +import { + makeWsRpcProtocolClient, + PING_INTERVAL, + PING_TIMEOUT, + type WsRpcProtocolClient, +} from "./protocol.ts"; import { NETWORK_BLOCKING_HINT } from "../errors/network.ts"; import type { ConnectionAttemptError, @@ -52,6 +57,8 @@ export interface RpcSession { ) => ServerConfigSubscription; readonly ready: Effect.Effect; readonly probe: Effect.Effect; + /** How many socket reads have delivered data, pongs included. A rising count means the link still carries traffic. */ + readonly receivedReads: Effect.Effect; readonly closed: Effect.Effect; } @@ -195,12 +202,34 @@ export const make = Effect.fn("RpcSessionFactory.make")(function* ( Effect.asVoid, ), }); - const layerSocket = Socket.layerWebSocket(connection.socketUrl, { - openTimeout: SOCKET_OPEN_TIMEOUT, - }).pipe(Layer.provide(Layer.succeed(Socket.WebSocketConstructor, webSocketConstructor))); + let receivedReads = 0; + const layerSocket = Layer.effect( + Socket.Socket, + Effect.gen(function* () { + const socket = yield* Socket.Socket; + return Socket.make({ + reader: Effect.map(socket.reader, (reader) => ({ + ...reader, + pull: Effect.map(reader.pull, (frames) => { + receivedReads += 1; + return frames; + }), + })), + writer: socket.writer, + }); + }), + ).pipe( + Layer.provide( + Socket.layerWebSocket(connection.socketUrl, { + openTimeout: SOCKET_OPEN_TIMEOUT, + }), + ), + Layer.provide(Layer.succeed(Socket.WebSocketConstructor, webSocketConstructor)), + ); const layerProtocol = Layer.effect( RpcClient.Protocol, RpcClient.makeProtocolSocket({ + pingInterval: PING_INTERVAL, pingTimeout: PING_TIMEOUT, retryTransientErrors: false, retryPolicy: Schedule.recurs(0), @@ -392,6 +421,7 @@ export const make = Effect.fn("RpcSessionFactory.make")(function* ( Effect.raceFirst(Deferred.await(disconnected)), ), probe, + receivedReads: Effect.sync(() => receivedReads), closed: Effect.raceFirst( Deferred.await(disconnected), Deferred.await(configSubscriptionClosed), diff --git a/packages/client-runtime/src/state/pullRequests.test.ts b/packages/client-runtime/src/state/pullRequests.test.ts index 0217cfd00078..d8a96edc3c55 100644 --- a/packages/client-runtime/src/state/pullRequests.test.ts +++ b/packages/client-runtime/src/state/pullRequests.test.ts @@ -602,6 +602,7 @@ function session(client: WsRpcProtocolClient): RpcSession { subscribeServerConfig: (input) => client.subscribeServerConfig(input), ready: Effect.void, probe: Effect.void, + receivedReads: Effect.succeed(0), closed: Effect.never, }; } diff --git a/packages/client-runtime/src/state/server.test.ts b/packages/client-runtime/src/state/server.test.ts index 330df548ba0f..32cbb0579e4e 100644 --- a/packages/client-runtime/src/state/server.test.ts +++ b/packages/client-runtime/src/state/server.test.ts @@ -84,6 +84,7 @@ function session(client: WsRpcProtocolClient): RpcSession { subscribeServerConfig: (input) => client.subscribeServerConfig(input), ready: Effect.void, probe: Effect.void, + receivedReads: Effect.succeed(0), closed: Effect.never, }; } diff --git a/packages/client-runtime/src/state/serverUsage.test.ts b/packages/client-runtime/src/state/serverUsage.test.ts index 3410fd3506fe..de11067db954 100644 --- a/packages/client-runtime/src/state/serverUsage.test.ts +++ b/packages/client-runtime/src/state/serverUsage.test.ts @@ -134,6 +134,7 @@ const makeHarness = Effect.fn("ServerUsageTest.makeHarness")(function* ( subscribeServerConfig: (input) => client.subscribeServerConfig(input), ready: Effect.void, probe: Effect.void, + receivedReads: Effect.succeed(0), closed: Effect.never, }; const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({ diff --git a/packages/client-runtime/src/state/shell-sync.test.ts b/packages/client-runtime/src/state/shell-sync.test.ts index b20d3e9cda19..1e46029a9786 100644 --- a/packages/client-runtime/src/state/shell-sync.test.ts +++ b/packages/client-runtime/src/state/shell-sync.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, }; } diff --git a/packages/client-runtime/src/state/sourceControl.test.ts b/packages/client-runtime/src/state/sourceControl.test.ts index ad974ea6841a..6804bc7b457e 100644 --- a/packages/client-runtime/src/state/sourceControl.test.ts +++ b/packages/client-runtime/src/state/sourceControl.test.ts @@ -56,6 +56,7 @@ function session(client: WsRpcProtocolClient): RpcSession { subscribeServerConfig: (input) => client.subscribeServerConfig(input), ready: Effect.void, probe: Effect.void, + receivedReads: Effect.succeed(0), closed: Effect.never, }; } diff --git a/packages/client-runtime/src/state/threads-atoms.test.ts b/packages/client-runtime/src/state/threads-atoms.test.ts index 1eebcd23c489..db15b0977a2c 100644 --- a/packages/client-runtime/src/state/threads-atoms.test.ts +++ b/packages/client-runtime/src/state/threads-atoms.test.ts @@ -102,6 +102,7 @@ const makeHarness = Effect.fn("TestThreadAtoms.makeHarness")(function* (options? subscribeServerConfig: (input) => client.subscribeServerConfig(input), ready: Effect.void, probe: Effect.void, + receivedReads: Effect.succeed(0), closed: Effect.never, }; const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({ diff --git a/packages/client-runtime/src/state/threads-sync.test.ts b/packages/client-runtime/src/state/threads-sync.test.ts index b78377b2b1ec..51603746385b 100644 --- a/packages/client-runtime/src/state/threads-sync.test.ts +++ b/packages/client-runtime/src/state/threads-sync.test.ts @@ -83,6 +83,7 @@ function testSession( subscribeServerConfig: (input) => client.subscribeServerConfig(input), ready: Effect.void, probe: Effect.void, + receivedReads: Effect.succeed(0), closed: Effect.never, }; } diff --git a/packages/client-runtime/src/state/vcs.test.ts b/packages/client-runtime/src/state/vcs.test.ts index b72f804194ca..e1da9a7db72b 100644 --- a/packages/client-runtime/src/state/vcs.test.ts +++ b/packages/client-runtime/src/state/vcs.test.ts @@ -91,6 +91,7 @@ function session(client: WsRpcProtocolClient): RpcSession { subscribeServerConfig: (input) => client.subscribeServerConfig(input), ready: Effect.void, probe: Effect.void, + receivedReads: Effect.succeed(0), closed: Effect.never, }; } diff --git a/packages/client-runtime/src/state/vcsAction.test.ts b/packages/client-runtime/src/state/vcsAction.test.ts index 32dda6f6cba2..e3f8dae0b333 100644 --- a/packages/client-runtime/src/state/vcsAction.test.ts +++ b/packages/client-runtime/src/state/vcsAction.test.ts @@ -92,6 +92,7 @@ function session(client: WsRpcProtocolClient): RpcSession { subscribeServerConfig: (input) => client.subscribeServerConfig(input), ready: Effect.void, probe: Effect.void, + receivedReads: Effect.succeed(0), closed: Effect.never, }; }