# Files changed · PR \#19426 · pingdotgg/t3code [View on GitCafe](https://git.cafe/pingdotgg/t3code/pull/19426/diffs) Repository: [pingdotgg/t3code](https://git.cafe/pingdotgg/t3code) Visibility: public [fix\(client\): slow networks no longer drop working connections](https://git.cafe/pingdotgg/t3code/pull/19426.md) Version: 1 Head commit: aa7bb7087eb8390248c6fd8ea06329a21bc105ca Base commit: 1451e7cb23034d29774338608368a15d8845df36 Comparison base: 1451e7cb23034d29774338608368a15d8845df36 Showing up to 20 entries from this revision. If the pull request changes, reopen the first page before continuing. ## apps/mobile/src/connection/app-state-wakeups.test.ts Status: modified; +2; −2 ```diff @@ -6,8 +6,8 @@ } 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("only probes after a real background visit", () => { + expect(mobileApplicationActiveWakeup(null, 20_000)).toBe("application-refocused"); expect( mobileApplicationActiveWakeup(20_000, 20_000 + MOBILE_BACKGROUND_RECONNECT_AFTER_MS - 1), ).toBe("application-active-probe"); ``` ## apps/mobile/src/connection/app-state-wakeups.ts Status: modified; +3; −3 ```diff @@ -4,15 +4,15 @@ export type MobileApplicationActiveWakeup = Extract< Wakeups.ConnectionWakeup, - "application-active-probe" | "application-active-reconnect" + "application-active-probe" | "application-active-reconnect" | "application-refocused" >; export function mobileApplicationActiveWakeup( backgroundedAtMs: number | null, activeAtMs: number, ): MobileApplicationActiveWakeup { - return backgroundedAtMs !== null && + if (backgroundedAtMs === null) return "application-refocused"; + return activeAtMs - backgroundedAtMs >= MOBILE_BACKGROUND_RECONNECT_AFTER_MS - activeAtMs - backgroundedAtMs >= MOBILE_BACKGROUND_RECONNECT_AFTER_MS ? "application-active-reconnect" : "application-active-probe"; } ``` ## apps/mobile/src/connection/platform.ts Status: modified; +5; −2 ```diff @@ -27,7 +27,10 @@ import { clearThreadOutboxEnvironment } from "../state/thread-outbox-removal"; import { clearComposerDraftsEnvironment } from "../state/use-composer-drafts"; import { clearThreadComposerErrorsForEnvironment } from "../state/thread-composer-error"; +import { + mobileApplicationActiveWakeup, + type MobileApplicationActiveWakeup, +} from "./app-state-wakeups"; -import { mobileApplicationActiveWakeup } from "./app-state-wakeups"; import * as ConnectionStorage from "./storage"; function networkStatus(state: Network.NetworkState): "unknown" | "offline" | "online" { @@ -124,7 +127,7 @@ const layerWakeups = Wakeups.layer({ changes: Stream.mergeAll( [ + Stream.callback((queue) => - Stream.callback<"application-active-probe" | "application-active-reconnect">((queue) => Effect.acquireRelease( Effect.sync(() => { let backgroundedAtMs = AppState.currentState === "background" ? Date.now() : null; ``` ## apps/web/src/cloud/linkEnvironment.test.ts Status: modified; +1; −0 ```diff @@ -91,6 +91,7 @@ subscribeServerConfig: (input) => client.subscribeServerConfig(input), ready: Effect.void, probe: Effect.void, + receivedFrames: Effect.succeed(0), closed: Effect.never, }; const target = new PrimaryConnectionTarget({ ``` ## docs/internals/connection-runtime.md Status: modified; +12; −6 ```diff @@ -18,12 +18,18 @@ Foregrounding, an explicit retry, and an offline report probe the established session, and only a failed probe reconnects. Offline reports are often wrong, for +example for a loopback server. Probe replies share the RPC read loop with +subscriptions, so a full stream buffer can delay the reply while WebSocket +messages still arrive. Measure progress before that read loop and allow it to -example for a loopback server. A long mobile background suspension is the one -exception: it replaces the session at once, because the OS can kill a socket -without reporting closure, and a probe would hold a dead socket in "Resuming" -until it times out. That fresh attempt runs even while the network reports -offline. Foregrounding also wakes a pending retry immediately and -leaves an ordinary in-flight attempt alone. +extend a quick probe, with an absolute deadline so unrelated traffic cannot +keep a stuck probe alive forever. A long mobile background suspension replaces +the session at once instead, because the OS can kill a socket without reporting +closure, and a probe would hold a dead socket in "Resuming" until it times out. +That fresh attempt runs even while the network reports offline. An inactive-only +mobile return, such as from Control Center, must stay distinct from a background +resume: its socket and subscriptions were not suspended. It wakes retries and +route checks without probing or resubscribing. Foregrounding also wakes a +pending retry immediately and leaves an ordinary in-flight attempt alone. The [registry](../../packages/client-runtime/src/connection/registry.ts) scopes connections by environment. An involuntary disconnect retains the registration ``` ## packages/client-runtime/src/authorization/layer.test.ts Status: modified; +78; −30 ```diff @@ -46,7 +46,7 @@ expiresAt: "2026-06-06T01:00:00.000Z", }; -type RecordedResponse = Response | ((init: RequestInit) => Response); +type RecordedResponse = Response | ((init: RequestInit) => Response | Promise); function recordedFetch(responses: ReadonlyArray) { const calls: Array = []; @@ -308,37 +308,52 @@ }), ); - it.effect("reuses a same-account persisted environment token without contacting the relay", () => - Effect.gen(function* () { - const cached = new TokenStore.RemoteDpopAccessToken({ - environmentId: ENVIRONMENT_ID, - accountId: "account-1", + it.effect( + "reuses the persisted token through a slow ticket response without a relay request", + () => + Effect.gen(function* () { - label: DESCRIPTOR.label, + const started = yield* Deferred.make(); - endpoint: ENDPOINT, - accessToken: "cached-access-token", - expiresAtEpochMs: Number.MAX_SAFE_INTEGER, - dpopThumbprint: "thumbprint-1", - }); - const harness = yield* makeHarness({ - initialToken: cached, - responses: [websocketTicket("cached-ticket")], - }); - - const authorized = yield* Effect.gen(function* () { - const remote = yield* RemoteEnvironmentAuthorization.RemoteEnvironmentAuthorization; - return yield* remote.authorizeDpop({ - expectedEnvironmentId: ENVIRONMENT_ID, + const response = Promise.withResolvers(); + const cached = new TokenStore.RemoteDpopAccessToken({ + environmentId: ENVIRONMENT_ID, + accountId: "account-1", + label: DESCRIPTOR.label, + endpoint: ENDPOINT, + accessToken: "cached-access-token", + expiresAtEpochMs: Number.MAX_SAFE_INTEGER, + dpopThumbprint: "thumbprint-1", }); + const harness = yield* makeHarness({ + initialToken: cached, + responses: [ + () => { + Deferred.doneUnsafe(started, Effect.void); + return response.promise; + }, + ], + }); - }).pipe(Effect.provide(harness.layer)); - expect(authorized.socketUrl).toContain("wsTicket=cached-ticket"); - expect(authorized.socketUrl).toContain("connectionMethod=relay"); - expect(yield* Ref.get(harness.bootstrapCalls)).toBe(0); - expect(harness.fetch.calls).toHaveLength(1); - expect(String(harness.fetch.calls[0]?.[0])).toBe( - "https://environment.example.test/api/auth/websocket-ticket", - ); - }), + const authorizing = yield* Effect.gen(function* () { + const remote = yield* RemoteEnvironmentAuthorization.RemoteEnvironmentAuthorization; + return yield* remote.authorizeDpop({ + expectedEnvironmentId: ENVIRONMENT_ID, + }); + }).pipe(Effect.provide(harness.layer), Effect.forkChild); + + yield* Deferred.await(started); + yield* TestClock.adjust("5999 millis"); + expect(yield* Ref.get(harness.bootstrapCalls)).toBe(0); + response.resolve(websocketTicket("cached-ticket")); + const authorized = yield* Fiber.join(authorizing); + + expect(authorized.socketUrl).toContain("wsTicket=cached-ticket"); + expect(authorized.socketUrl).toContain("connectionMethod=relay"); + expect(yield* Ref.get(harness.bootstrapCalls)).toBe(0); + expect(harness.fetch.calls).toHaveLength(1); + expect(String(harness.fetch.calls[0]?.[0])).toBe( + "https://environment.example.test/api/auth/websocket-ticket", + ); + }).pipe(Effect.provide(TestClock.layer())), ); it.effect("uses the T3 Connect token on a learned direct address without the relay", () => @@ -666,6 +681,39 @@ }), ); + it.effect("refreshes a silent cached endpoint after six seconds", () => + Effect.gen(function* () { + const started = yield* Deferred.make(); + const harness = yield* makeHarness({ + initialToken: persistedToken(), + responses: [ + () => { + Deferred.doneUnsafe(started, Effect.void); + return new Promise(() => {}); + }, + Response.json(DESCRIPTOR), + accessToken("replacement-access-token"), + websocketTicket("replacement-ticket"), + ], + }); + const authorizing = yield* Effect.gen(function* () { + const remote = yield* RemoteEnvironmentAuthorization.RemoteEnvironmentAuthorization; + return yield* remote.authorizeDpop({ expectedEnvironmentId: ENVIRONMENT_ID }); + }).pipe(Effect.provide(harness.layer), Effect.forkChild); + + yield* Deferred.await(started); + yield* TestClock.adjust("5999 millis"); + expect(yield* Ref.get(harness.bootstrapCalls)).toBe(0); + yield* TestClock.adjust("1 milli"); + const authorized = yield* Fiber.join(authorizing); + expect(authorized.socketUrl).toContain("wsTicket=replacement-ticket"); + expect(yield* Ref.get(harness.bootstrapCalls)).toBe(1); + expect((yield* Ref.get(harness.tokens)).get(ENVIRONMENT_ID)?.accessToken).toBe( + "replacement-access-token", + ); + }).pipe(Effect.provide(TestClock.layer())), + ); + it.effect("removes a newly refreshed token when websocket ticket authorization fails", () => Effect.gen(function* () { const harness = yield* makeHarness({ ``` ## packages/client-runtime/src/authorization/service.ts Status: modified; +2; −1 ```diff @@ -85,7 +85,8 @@ } >()("@t3tools/client-runtime/authorization/service/RemoteEnvironmentAuthorization") {} +// A premature ticket timeout on T3 Connect needlessly replaces a valid cached token. +const CACHED_ENDPOINT_SOCKET_TIMEOUT_MS = 6_000; -const CACHED_ENDPOINT_SOCKET_TIMEOUT_MS = 3_000; const BEARER_DESCRIPTOR_CACHE_TTL_MS = 10_000; const DPOP_AUTHORIZATION_TIMEOUT_MS = 30_000; ``` ## packages/client-runtime/src/connection/registry.test.ts Status: modified; +1; −0 ```diff @@ -413,6 +413,7 @@ Stream.die(new Error("Config is not used by registry tests.")), ready: Effect.void, probe: Effect.void, + receivedFrames: Effect.succeed(0), closed: Deferred.await(closed), } satisfies RpcSession.RpcSession), () => Ref.update(releasedSessions, (count) => count + 1), ``` ## packages/client-runtime/src/connection/supervisor.test.ts Status: modified; +125; −4 ```diff @@ -157,6 +157,7 @@ ) => Effect.Effect; readonly ready?: (attempt: number) => Effect.Effect; readonly probe?: (attempt: number) => Effect.Effect; + readonly receivedFrames?: Effect.Effect; readonly initialConfig?: (attempt: number) => Effect.Effect; readonly checkRoute?: (route: ConnectionRoute) => Effect.Effect; }) { @@ -213,6 +214,7 @@ subscribeServerConfig: (input) => TEST_RPC_CLIENT.subscribeServerConfig(input), ready: options?.ready?.(attempt) ?? Effect.void, probe: options?.probe?.(attempt) ?? Effect.void, + receivedFrames: options?.receivedFrames ?? Effect.succeed(0), closed: Deferred.await(closed), } satisfies RpcSession.RpcSession), () => Ref.update(releaseCount, (count) => count + 1), @@ -648,6 +650,58 @@ }), ); + it.effect("a mobile refocus resets retries and checks routes without probing the session", () => + Effect.gen(function* () { + const probeCount = yield* Ref.make(0); + const lanReachable = yield* Ref.make(false); + const harness = yield* makeHarness({ + checkRoute: (route) => + route.target._tag === "BearerConnectionTarget" + ? Ref.get(lanReachable).pipe( + Effect.map((reachable) => (reachable ? "answered" : "silent")), + ) + : Effect.succeed("unchecked"), + prepare: (attempt, target) => + attempt <= 4 ? Effect.fail(transient()) : Effect.succeed(preparedFor(target)), + probe: () => Ref.update(probeCount, (count) => count + 1), + }); + const supervisor = yield* EnvironmentSupervisor.make(LAN_THEN_RELAY_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* awaitState( + supervisor.state, + (state) => state.phase === "backoff" && state.attempt === 2, + ); + yield* harness.wake("application-refocused"); + yield* awaitState( + supervisor.state, + (state) => state.phase === "connected" && state.attempt === 1, + ); + + yield* harness.wake("application-refocused"); + yield* TestClock.adjust("0 millis"); + expect(yield* Ref.get(probeCount)).toBe(0); + expect(yield* Ref.get(harness.sessionCount)).toBe(1); + + yield* Ref.set(lanReachable, true); + yield* harness.wake("application-refocused"); + yield* awaitState( + supervisor.state, + (state) => state.phase === "connected" && state.generation === 2, + ); + expect(Option.getOrThrow(yield* SubscriptionRef.get(supervisor.prepared)).target._tag).toBe( + "BearerConnectionTarget", + ); + expect(yield* Ref.get(probeCount)).toBe(0); + }).pipe(Effect.provide(TestClock.layer())), + ); + it.effect("explicit retry starts a fresh backoff sequence", () => Effect.gen(function* () { const harness = yield* makeHarness({ @@ -819,7 +873,7 @@ (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); @@ -1393,7 +1447,8 @@ yield* harness.wake("application-active"); yield* TestClock.adjust("5 seconds"); yield* supervisor.retryNow; - 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( @@ -1405,6 +1460,36 @@ }).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), + receivedFrames: 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"); + expect(yield* Ref.get(harness.sessionCount)).toBe(2); + 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({ @@ -1416,7 +1501,11 @@ yield* awaitState(supervisor.state, (state) => state.phase === "connected"); yield* harness.wake("application-active-probe"); - yield* TestClock.adjust("3 seconds"); + yield* TestClock.adjust("5999 millis"); + expect(yield* Ref.get(harness.sessionCount)).toBe(1); + yield* TestClock.adjust("1 milli"); + // The timed-out wake probe reconnects immediately without a backoff + // sleep: no further clock advance is needed. yield* awaitState( supervisor.state, (state) => state.phase === "connected" && state.generation === 2 && state.attempt === 1, @@ -1426,6 +1515,38 @@ }).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), + receivedFrames: 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"); + expect(yield* Ref.get(harness.sessionCount)).toBe(2); + 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(); @@ -2052,7 +2173,7 @@ 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 Status: modified; +20; −3 ```diff @@ -45,7 +45,8 @@ const establishmentTimeout = Duration.fromInputUnsafe(CONNECTION_ESTABLISHMENT_TIMEOUT); const synchronizationTimeout = Duration.fromInputUnsafe("75 seconds"); const CONNECTION_PROBE_TIMEOUT = "15 seconds"; +// Leave room for packet retransmission before replacing a slow but working link. +const QUICK_CONNECTION_PROBE_TIMEOUT = "6 seconds"; -const QUICK_CONNECTION_PROBE_TIMEOUT = "3 seconds"; 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 @@ -591,8 +592,12 @@ continue; } yield* Ref.set(probeUnanswered, true); + const probeStartedAt = yield* Clock.monotonicTimeNanos; + const latestDeadline = probeStartedAt + Duration.toNanosUnsafe(CONNECTION_PROBE_TIMEOUT); + let probeWindow = probeTimeout; + let deadline = probeStartedAt + Duration.toNanosUnsafe(probeWindow); + let receivedFrames = yield* lease.session.receivedFrames; const probe = yield* Effect.forkChild(lease.session.probe); - let deadline = (yield* Clock.monotonicTimeNanos) + Duration.toNanosUnsafe(probeTimeout); for (;;) { const remaining = deadline - (yield* Clock.monotonicTimeNanos); const probeEvent = yield* Effect.raceAllFirst([ @@ -605,6 +610,15 @@ ), ]); if (probeEvent._tag === "TimedOut") { + // The reply may be behind other traffic; the absolute cap still bounds a stuck probe. + const received = yield* lease.session.receivedFrames; + const now = yield* Clock.monotonicTimeNanos; + if (received > receivedFrames && now < latestDeadline) { + receivedFrames = received; + const extended = now + Duration.toNanosUnsafe(probeWindow); + deadline = extended < latestDeadline ? extended : latestDeadline; + continue; + } yield* Fiber.interrupt(probe); return yield* new ConnectionTransientError({ reason: "timeout", @@ -627,7 +641,10 @@ if (signalTimeout !== undefined) { const signalDeadline = (yield* Clock.monotonicTimeNanos) + Duration.toNanosUnsafe(signalTimeout); + if (signalDeadline < deadline) { + deadline = signalDeadline; + probeWindow = signalTimeout; - if (signalDeadline < deadline) deadline = signalDeadline; + } } } } ``` ## packages/client-runtime/src/connection/wakeups.ts Status: modified; +5; −1 ```diff @@ -6,6 +6,8 @@ | "application-active" | "application-active-probe" | "application-active-reconnect" + // Inactive-only mobile returns wake retries without probing or resubscribing. + | "application-refocused" | "credentials-changed" | "network-changed"; @@ -13,7 +15,8 @@ return ( reason === "application-active" || reason === "application-active-probe" || - reason === "application-active-reconnect" + reason === "application-active-reconnect" || + reason === "application-refocused" ); } @@ -22,6 +25,7 @@ } // A long resume replaces the session, and the new session subscribes on its own. +// A refocus never suspended the socket, so its streams are still live. export function shouldResubscribeAfterWakeup(reason: ConnectionWakeup): boolean { return reason === "application-active" || reason === "application-active-probe"; } ``` ## packages/client-runtime/src/operations/commands.performance.test.ts Status: modified; +1; −0 ```diff @@ -177,6 +177,7 @@ subscribeServerConfig: (requestInput) => client.subscribeServerConfig(requestInput), ready: Effect.void, probe: Effect.void, + receivedFrames: Effect.succeed(0), closed: Effect.never, }; return EnvironmentSupervisor.EnvironmentSupervisor.of({ ``` ## packages/client-runtime/src/operations/commands.test.ts Status: modified; +1; −0 ```diff @@ -135,6 +135,7 @@ subscribeServerConfig: (input) => client.subscribeServerConfig(input), ready: Effect.void, probe: Effect.void, + receivedFrames: Effect.succeed(0), closed: Effect.never, }; return EnvironmentSupervisor.EnvironmentSupervisor.of({ ``` ## packages/client-runtime/src/rpc/client.test.ts Status: modified; +1; −0 ```diff @@ -61,6 +61,7 @@ subscribeServerConfig: (input) => client.subscribeServerConfig(input), ready: Effect.void, probe: Effect.void, + receivedFrames: Effect.succeed(0), closed: Effect.never, }; } ``` ## packages/client-runtime/src/rpc/session.test.ts Status: modified; +5; −0 ```diff @@ -1141,6 +1141,11 @@ }), ); yield* Deferred.await(consuming); + + // Arrival must be visible even while the full stream blocks RPC reads. + const before = yield* session.receivedFrames; + socket.serverMessage(encodeJson({ _tag: "Pong" })); + expect(yield* session.receivedFrames).toBe(before + 1); yield* Fiber.interrupt(streamFiber); const probeFiber = yield* Effect.forkChild(session.probe); ``` ## packages/client-runtime/src/rpc/session.ts Status: modified; +26; −3 ```diff @@ -52,6 +52,8 @@ ) => ServerConfigSubscription; readonly ready: Effect.Effect; readonly probe: Effect.Effect; + /** Message arrivals, including Effect's application pongs, before RPC backpressure. */ + readonly receivedFrames: Effect.Effect; readonly closed: Effect.Effect; } @@ -191,9 +193,29 @@ Effect.asVoid, ), }); + // RPC reads can stall behind a full subscription buffer while the link still delivers. + let receivedFrames = 0; + const countFrame = () => { + receivedFrames += 1; + }; + const layerSocket = Layer.effect( + Socket.Socket, + Socket.fromWebSocket( + Effect.acquireRelease( + Effect.sync(() => { + const socket = webSocketConstructor(connection.socketUrl); + socket.addEventListener("message", countFrame); + return socket; + }), + (socket) => + Effect.sync(() => { + socket.removeEventListener("message", countFrame); - const layerSocket = Socket.layerWebSocket(connection.socketUrl, { - openTimeout: SOCKET_OPEN_TIMEOUT, - }).pipe(Layer.provide(Layer.succeed(Socket.WebSocketConstructor, webSocketConstructor))); + socket.close(1000); + }), + ), + { openTimeout: SOCKET_OPEN_TIMEOUT }, + ), + ); const layerProtocol = Layer.effect( RpcClient.Protocol, RpcClient.makeProtocolSocket({ @@ -389,6 +411,7 @@ Effect.raceFirst(Deferred.await(disconnected)), ), probe, + receivedFrames: Effect.sync(() => receivedFrames), closed: Effect.raceFirst( Deferred.await(disconnected), Deferred.await(configSubscriptionClosed), ``` ## packages/client-runtime/src/state/pullRequests.test.ts Status: modified; +1; −0 ```diff @@ -602,6 +602,7 @@ subscribeServerConfig: (input) => client.subscribeServerConfig(input), ready: Effect.void, probe: Effect.void, + receivedFrames: Effect.succeed(0), closed: Effect.never, }; } ``` ## packages/client-runtime/src/state/server.test.ts Status: modified; +1; −0 ```diff @@ -82,6 +82,7 @@ subscribeServerConfig: (input) => client.subscribeServerConfig(input), ready: Effect.void, probe: Effect.void, + receivedFrames: Effect.succeed(0), closed: Effect.never, }; } ``` ## packages/client-runtime/src/state/serverUsage.test.ts Status: modified; +1; −0 ```diff @@ -137,6 +137,7 @@ subscribeServerConfig: (input) => client.subscribeServerConfig(input), ready: Effect.void, probe: Effect.void, + receivedFrames: Effect.succeed(0), closed: Effect.never, }; const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({ ``` ## packages/client-runtime/src/state/shell-sync.test.ts Status: modified; +2; −0 ```diff @@ -61,6 +61,7 @@ subscribeServerConfig: (input) => client.subscribeServerConfig(input), ready: Effect.void, probe: Effect.void, + receivedFrames: Effect.succeed(0), closed: Effect.never, }; } @@ -429,6 +430,7 @@ } expect(yield* Ref.get(capturedAfterSequences)).toEqual([10, 40, 40]); + yield* Queue.offer(wakeups, "application-refocused"); yield* Queue.offer(wakeups, "application-active-reconnect"); for (let attempt = 0; attempt < 10; attempt += 1) { yield* Effect.yieldNow; ``` [Next page](https://git.cafe/pingdotgg/t3code/pull/19426/diffs?after=InBhY2thZ2VzL2NsaWVudC1ydW50aW1lL3NyYy9zdGF0ZS9zaGVsbC1zeW5jLnRlc3QudHMi&expectedVersion=1&format=markdown)