From 60cda175e93d3819fbac52a5aba4870274512b40 Mon Sep 17 00:00:00 2001 From: Bil0000 <62337003+Bil0000@users.noreply.github.com> Date: Mon, 9 Nov 2026 03:20:22 +0100 Subject: [PATCH 1/3] perf(mobile): deflate large socket frames for clients without permessage-deflate iOS WebSockets never negotiate permessage-deflate, so the connect config (about 78 KB) and every snapshot crossed the wire uncompressed. Clients now ask with frameDeflate=1; when the upgrade offers no permessage-deflate, the server sends string frames of 1 KB or more as raw DEFLATE binary parts (node zlib, level 1) and client-runtime inflates them with fflate. Old servers ignore the flag and old clients never send it. --- apps/server/src/socketFrameDeflate.ts | 5 ++ apps/server/src/ws.ts | 14 +++- .../client-runtime/src/rpc/session.test.ts | 78 ++++++++++++------- packages/client-runtime/src/rpc/session.ts | 2 + packages/shared/package.json | 1 + packages/shared/src/socketFrameParts.ts | 62 +++++++++++++-- pnpm-lock.yaml | 3 + 7 files changed, 128 insertions(+), 37 deletions(-) create mode 100644 apps/server/src/socketFrameDeflate.ts diff --git a/apps/server/src/socketFrameDeflate.ts b/apps/server/src/socketFrameDeflate.ts new file mode 100644 index 000000000000..293abf1b707c --- /dev/null +++ b/apps/server/src/socketFrameDeflate.ts @@ -0,0 +1,5 @@ +// @effect-diagnostics nodeBuiltinImport:off -- zlib deflates 6-8x faster than a JS deflate on the event loop. +import * as NodeZlib from "node:zlib"; + +export const deflateSocketFrame = (frame: string): Uint8Array => + NodeZlib.deflateRawSync(frame, { level: 1 }); diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 1b6835ef993f..eb9fb554d928 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -265,6 +265,7 @@ import { failEnvironmentAuthInvalid, failEnvironmentInternal } from "./auth/http import { isNetworkPath } from "@t3tools/shared/path"; import * as RelayClient from "@t3tools/shared/relayClient"; import { + SOCKET_FRAME_DEFLATE_QUERY_PARAM, SOCKET_FRAME_PARTS_QUERY_PARAM, splitLargeSocketFrames, } from "@t3tools/shared/socketFrameParts"; @@ -274,6 +275,7 @@ import { withUsageLimitsCommands, } from "@t3tools/shared/usageLimits"; import * as AgentSessionScanner from "./project/AgentSessionScanner.ts"; +import { deflateSocketFrame } from "./socketFrameDeflate.ts"; import * as AgentSessionImporter from "./project/AgentSessionImporter.ts"; import * as UsageLimitSources from "./usage/UsageLimitSources.ts"; import { requestUrl } from "./requestUrl.ts"; @@ -3632,6 +3634,12 @@ export const layer = Layer.unwrap( ), ), ); + // A socket with permessage-deflate is already compressed on the wire. + const deflateFrames = + url.value.searchParams.has(SOCKET_FRAME_DEFLATE_QUERY_PARAM) && + !request.headers["sec-websocket-extensions"]?.includes("permessage-deflate") + ? deflateSocketFrame + : undefined; const framedRpcWebSocketHttpEffect = url.value.searchParams.has( SOCKET_FRAME_PARTS_QUERY_PARAM, ) @@ -3641,7 +3649,11 @@ export const layer = Layer.unwrap( // The RPC server upgrades the request itself, so its socket is // wrapped through the request it reads. Object.create(request, { - upgrade: { value: Effect.map(request.upgrade, splitLargeSocketFrames) }, + upgrade: { + value: Effect.map(request.upgrade, (socket) => + splitLargeSocketFrames(socket, deflateFrames), + ), + }, }), ), ) diff --git a/packages/client-runtime/src/rpc/session.test.ts b/packages/client-runtime/src/rpc/session.test.ts index 8f4bee5dc16c..6ae1fdcb0117 100644 --- a/packages/client-runtime/src/rpc/session.test.ts +++ b/packages/client-runtime/src/rpc/session.test.ts @@ -8,7 +8,7 @@ import { WS_METHODS, UsageLimitSourceId, } from "@t3tools/contracts"; -import { splitSocketFrame } from "@t3tools/shared/socketFrameParts"; +import { splitDeflatedSocketFrame, splitSocketFrame } from "@t3tools/shared/socketFrameParts"; import { describe, expect, it } from "@effect/vitest"; import * as Cause from "effect/Cause"; import * as Deferred from "effect/Deferred"; @@ -91,7 +91,7 @@ class TestWebSocket { this.emit("open", { type: "open" }); } - serverMessage(data: string) { + serverMessage(data: string | ArrayBuffer) { this.emit("message", { data, type: "message" }); } @@ -277,34 +277,52 @@ const publishConfigEvents = Effect.fn("TestRpcSessionFactory.publishConfigEvents }); describe("RpcSessionFactory", () => { - it.effect("connects while a large config snapshot downloads at 4000 bytes per second", () => - 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(); - const request = yield* awaitRequest(socket); - // Multi-byte text, so parts must be sized in UTF-8 bytes and never split a surrogate pair. - const config = { ...SERVER_CONFIG, cwd: "/" + "é€😀x".repeat(32 * 1024) }; - const frame = encodeJson({ - _tag: "Chunk", - requestId: request.id, - values: [{ version: 1, type: "snapshot", config: encodeServerConfig(config) }], - }); - const parts = new URL(socket.url).searchParams.has("frameParts") - ? splitSocketFrame(frame) - : [frame]; - - for (const part of parts) { - const bytes = new TextEncoder().encode(part); - yield* TestClock.adjust((bytes.length / 4000) * 1000); - socket.serverMessage(new TextDecoder().decode(bytes)); - } - yield* Fiber.join(readyFiber); + it.effect.each([ + { label: "text parts", deflated: false }, + { label: "deflated binary parts", deflated: true }, + ])( + "connects while a large config snapshot downloads at 4000 bytes per second as $label", + ({ deflated }) => + 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(); + const request = yield* awaitRequest(socket); + // Multi-byte text, so parts must be sized in UTF-8 bytes and never split a surrogate pair. + const config = { ...SERVER_CONFIG, cwd: "/" + "é€😀x".repeat(32 * 1024) }; + const frame = encodeJson({ + _tag: "Chunk", + requestId: request.id, + values: [{ version: 1, type: "snapshot", config: encodeServerConfig(config) }], + }); + const url = new URL(socket.url).searchParams; + expect(url.has("frameParts") && url.has("frameDeflate")).toBe(true); + const parts = deflated + ? splitDeflatedSocketFrame( + yield* Effect.promise(() => + new Response( + new Blob([frame]).stream().pipeThrough(new CompressionStream("deflate-raw")), + ) + .arrayBuffer() + .then((buffer) => new Uint8Array(buffer)), + ), + // Repeated text deflates small, so a small part size forces several parts. + 1024, + ) + : splitSocketFrame(frame); + expect(parts.length).toBeGreaterThan(1); + + for (const part of parts) { + const bytes = typeof part === "string" ? new TextEncoder().encode(part) : part; + yield* TestClock.adjust((bytes.length / 4000) * 1000); + socket.serverMessage(typeof part === "string" ? part : bytes.slice().buffer); + } + yield* Fiber.join(readyFiber); - expect(yield* session.initialConfig).toEqual(config); - }), + expect(yield* session.initialConfig).toEqual(config); + }), ); it.effect("owns one scoped websocket attempt and exposes readiness and closure", () => @@ -314,7 +332,7 @@ describe("RpcSessionFactory", () => { const readyFiber = yield* Effect.forkChild(session.ready); const socket = yield* awaitSocket(sockets); - expect(socket.url).toBe(`${PREPARED.socketUrl}&frameParts=1`); + expect(socket.url).toBe(`${PREPARED.socketUrl}&frameParts=1&frameDeflate=1`); socket.open(); yield* completeInitialConfig(socket); yield* Fiber.join(readyFiber); diff --git a/packages/client-runtime/src/rpc/session.ts b/packages/client-runtime/src/rpc/session.ts index 3ea4b73bf6bd..3c0d8b76c2f4 100644 --- a/packages/client-runtime/src/rpc/session.ts +++ b/packages/client-runtime/src/rpc/session.ts @@ -6,6 +6,7 @@ import { } from "@t3tools/contracts"; import { joinSocketFrameParts, + SOCKET_FRAME_DEFLATE_QUERY_PARAM, SOCKET_FRAME_PARTS_QUERY_PARAM, } from "@t3tools/shared/socketFrameParts"; import * as Cause from "effect/Cause"; @@ -231,6 +232,7 @@ export const make = Effect.fn("RpcSessionFactory.make")(function* ( }; const socketUrl = new URL(connection.socketUrl); socketUrl.searchParams.set(SOCKET_FRAME_PARTS_QUERY_PARAM, "1"); + socketUrl.searchParams.set(SOCKET_FRAME_DEFLATE_QUERY_PARAM, "1"); const layerSocket = Layer.effect( Socket.Socket, Effect.map(Effect.service(Socket.Socket), joinSocketFrameParts), diff --git a/packages/shared/package.json b/packages/shared/package.json index 5d9a87ac7719..793a68eebb00 100644 --- a/packages/shared/package.json +++ b/packages/shared/package.json @@ -554,6 +554,7 @@ "@t3tools/contracts": "workspace:*", "effect": "catalog:", "ffi-rs": "1.3.2", + "fflate": "0.8.3", "mdast-util-directive": "catalog:", "mdast-util-to-hast": "^13.2.1", "micromark-extension-directive": "catalog:", diff --git a/packages/shared/src/socketFrameParts.ts b/packages/shared/src/socketFrameParts.ts index e0c82a87de3b..564f8e492b18 100644 --- a/packages/shared/src/socketFrameParts.ts +++ b/packages/shared/src/socketFrameParts.ts @@ -1,9 +1,17 @@ import * as Arr from "effect/Array"; import * as Effect from "effect/Effect"; import * as Socket from "effect/socket/Socket"; +import { inflateSync, strFromU8 } from "fflate"; // Old clients omit this; old servers ignore it. export const SOCKET_FRAME_PARTS_QUERY_PARAM = "frameParts"; +// The client can inflate frames the server deflated. The server only does so +// when the socket negotiated no permessage-deflate, as iOS sockets never do. +export const SOCKET_FRAME_DEFLATE_QUERY_PARAM = "frameDeflate"; +// Smaller frames gain little from a deflate without a shared window. +const SOCKET_FRAME_DEFLATE_MIN_CHARS = 1024; +const BINARY_PART_MORE = 1; +const BINARY_PART_LAST = 0; // WebSocket messages arrive whole. At 4,000 bytes/s each part takes about // 12 seconds, leaving margin before the client's 15-second liveness timeout. @@ -41,27 +49,69 @@ export function splitSocketFrame( return Arr.append(parts, frame.slice(start)); } -/** Server side: sends string frames longer than one part as consecutive parts. */ -export function splitLargeSocketFrames(socket: Socket.Socket): Socket.Socket { +/** Raw DEFLATE bytes as binary parts, each led by a byte that says whether more follow. */ +export function splitDeflatedSocketFrame( + deflated: Uint8Array, + partBytes = SOCKET_FRAME_PART_BYTES, +): Arr.NonEmptyReadonlyArray { + const size = partBytes - 1; + const parts: Array = []; + for (let start = 0; start === 0 || start < deflated.length; start += size) { + const body = deflated.subarray(start, start + size); + const part = new Uint8Array(body.length + 1); + part[0] = start + size < deflated.length ? BINARY_PART_MORE : BINARY_PART_LAST; + part.set(body, 1); + parts.push(part); + } + return parts as unknown as Arr.NonEmptyReadonlyArray; +} + +/** + * Server side: sends string frames longer than one part as consecutive parts. + * With `deflate` (raw DEFLATE), larger string frames go out as binary parts. + */ +export function splitLargeSocketFrames( + socket: Socket.Socket, + deflate?: (frame: string) => Uint8Array, +): Socket.Socket { const split = (chunk: string | Uint8Array): Arr.NonEmptyReadonlyArray => - typeof chunk === "string" ? splitSocketFrame(chunk) : [chunk]; + typeof chunk !== "string" + ? [chunk] + : deflate !== undefined && chunk.length >= SOCKET_FRAME_DEFLATE_MIN_CHARS + ? splitDeflatedSocketFrame(deflate(chunk)) + : splitSocketFrame(chunk); return Socket.make({ reader: socket.reader, writer: Effect.map(socket.writer, (writer) => ({ write: (chunk) => - typeof chunk === "string" ? writer.writeAll(splitSocketFrame(chunk)) : writer.write(chunk), + typeof chunk === "string" ? writer.writeAll(split(chunk)) : writer.write(chunk), writeAll: (chunks) => writer.writeAll(Arr.flatMap(chunks, split)), })), }); } -/** Client side: joins parts back into whole frames. */ +/** Client side: joins parts back into whole frames, inflating deflated ones. */ export function joinSocketFrameParts(socket: Socket.Socket): Socket.Socket { return Socket.make({ reader: Effect.map(socket.reader, (reader) => { let pending = ""; + let pendingBinary: Array = []; + const joinBinary = (part: Uint8Array) => { + pendingBinary.push(part.subarray(1)); + if (part[0] === BINARY_PART_MORE) return PART_RECEIVED_FRAME; + const deflated = new Uint8Array( + pendingBinary.reduce((size, body) => size + body.length, 0), + ); + let offset = 0; + for (const body of pendingBinary) { + deflated.set(body, offset); + offset += body.length; + } + pendingBinary = []; + return strFromU8(inflateSync(deflated)); + }; const join = (frame: string | Uint8Array) => { - if (typeof frame !== "string") return frame; + if (typeof frame !== "string") return joinBinary(frame); if (frame.startsWith(PART_PREFIX)) { pending += frame.slice(PART_PREFIX.length); return PART_RECEIVED_FRAME; diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index fa8574ccfee7..2cf3f217194e 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -2057,6 +2057,9 @@ importers: ffi-rs: specifier: 1.3.2 version: 1.3.2 + fflate: + specifier: 0.8.3 + version: 0.8.3 mdast-util-directive: specifier: 'catalog:' version: 3.1.0(supports-color@8.1.1) From a005905a5e63b0f442ae425e6d1c2dad3b7886a5 Mon Sep 17 00:00:00 2001 From: Bil0000 <62337003+Bil0000@users.noreply.github.com> Date: Mon, 9 Nov 2026 03:21:41 +0100 Subject: [PATCH 2/3] perf(mobile): open a thread from its socket snapshot, without the HTTP round trip A cold thread open fetched the bounded snapshot over HTTP, then subscribed over the socket, which sends the same bounded window. On servers that advertise threadSnapshotPagination the open now subscribes straight away. If that subscribe fails, the retry asks HTTP first, so a missing thread still ends as deleted instead of flashing an error. --- .../src/state/threads-atoms.test.ts | 163 +++++++++--------- .../src/state/threads-sync.test.ts | 36 +++- packages/client-runtime/src/state/threads.ts | 22 ++- 3 files changed, 133 insertions(+), 88 deletions(-) diff --git a/packages/client-runtime/src/state/threads-atoms.test.ts b/packages/client-runtime/src/state/threads-atoms.test.ts index af2ab6d0575a..460f7e593990 100644 --- a/packages/client-runtime/src/state/threads-atoms.test.ts +++ b/packages/client-runtime/src/state/threads-atoms.test.ts @@ -53,7 +53,7 @@ const SNAPSHOT: OrchestrationV2ThreadDetailSnapshot = { snapshotSequence: 7, pro const makeHarness = Effect.fn("TestThreadAtoms.makeHarness")(function* (options?: { readonly snapshot?: OrchestrationV2ThreadDetailSnapshot; - readonly snapshotUnavailable?: boolean; + readonly socketSnapshot?: false; }) { const subscriptions = yield* Queue.unbounded<{ readonly afterSequence: number | undefined; @@ -87,6 +87,10 @@ const makeHarness = Effect.fn("TestThreadAtoms.makeHarness")(function* (options? }).pipe(Effect.andThen(Deferred.succeed(closed, undefined))), ); yield* Queue.offer(subscriptions, { afterSequence: input.afterSequence, events, closed }); + // Like a paging server, a fresh subscribe opens with the bounded snapshot. + if (input.afterSequence === undefined && options?.socketSnapshot !== false) { + yield* Queue.offer(events, { kind: "snapshot", ...snapshot }); + } return Stream.fromQueue(events); }), ), @@ -191,9 +195,6 @@ const makeHarness = Effect.fn("TestThreadAtoms.makeHarness")(function* (options? load: () => Effect.sync(() => { httpLoads += 1; - if (options?.snapshotUnavailable === true) { - return { _tag: "unavailable" as const }; - } return { _tag: "present" as const, snapshot, @@ -286,7 +287,7 @@ describe("createEnvironmentThreadStateAtoms", () => { expect(currentThread(h.registry, h.stateAtom)).toBe(before); const next = yield* Queue.take(h.subscriptions); expect(next.afterSequence).toBe(8); - expect(h.counts()).toEqual({ httpLoads: 1, diskLoads: 1, opened: 2, active: 1 }); + expect(h.counts()).toEqual({ httpLoads: 0, diskLoads: 1, opened: 2, active: 1 }); remount(); yield* Deferred.await(next.closed); }), @@ -316,7 +317,7 @@ describe("createEnvironmentThreadStateAtoms", () => { const retrying = yield* h.loadEarlier().pipe(Effect.forkScoped); const retried = yield* Queue.take(h.olderLoads); expect(retried.cursor).toBe("older-1"); - expect(h.counts().httpLoads).toBe(1); + expect(h.counts().httpLoads).toBe(0); remount(); yield* Deferred.await(next.closed); yield* Deferred.await(retried.closed); @@ -324,86 +325,84 @@ describe("createEnvironmentThreadStateAtoms", () => { }), ); - it.effect( - "merges older history after an HTTP failure falls back to a bounded socket snapshot", - () => - Effect.gen(function* () { - const h = yield* makeHarness({ snapshotUnavailable: true }); - const unmount = h.registry.mount(h.stateAtom); - const subscription = yield* Queue.take(h.subscriptions); - expect(subscription.afterSequence).toBeUndefined(); + it.effect("opens from the bounded socket snapshot without HTTP and merges older history", () => + Effect.gen(function* () { + const h = yield* makeHarness({ socketSnapshot: false }); + const unmount = h.registry.mount(h.stateAtom); + const subscription = yield* Queue.take(h.subscriptions); + expect(subscription.afterSequence).toBeUndefined(); - yield* Queue.offer(subscription.events, { - kind: "snapshot", - snapshotSequence: 12, - projection: THREAD, - historyCursor: "socket-older-1", - hasMoreHistory: true, - latestLocalTurnOrdinal: 4, - }); - yield* Queue.offer(subscription.events, { kind: "synchronized" }); - yield* observeState( - h.registry, - h.stateAtom, - (state) => state.status === "live" && state.history.historyCursor === "socket-older-1", - ); + yield* Queue.offer(subscription.events, { + kind: "snapshot", + snapshotSequence: 12, + projection: THREAD, + historyCursor: "socket-older-1", + hasMoreHistory: true, + latestLocalTurnOrdinal: 4, + }); + yield* Queue.offer(subscription.events, { kind: "synchronized" }); + yield* observeState( + h.registry, + h.stateAtom, + (state) => state.status === "live" && state.history.historyCursor === "socket-older-1", + ); - const loading = yield* h.loadEarlier().pipe(Effect.forkScoped); - const request = yield* Queue.take(h.olderLoads); - expect(request.cursor).toBe("socket-older-1"); - const olderItem = { - id: TurnItemId.make("socket-fallback-older-item"), - type: "command_execution" as const, - threadId: THREAD_ID, - runId: null, - nodeId: null, - providerThreadId: null, - providerTurnId: null, - nativeItemRef: null, - parentItemId: null, - ordinal: 1, - status: "completed" as const, - title: "Earlier command", - input: "pwd", - output: "/workspace", - exitCode: 0, - startedAt: THREAD.thread.createdAt, - completedAt: THREAD.thread.createdAt, - updatedAt: THREAD.thread.createdAt, - }; - yield* Deferred.succeed(request.response, { - snapshotSequence: 12, - items: [ - { - position: 0, - visibility: "local", - sourceThreadId: THREAD_ID, - sourceItemId: olderItem.id, - item: olderItem, - }, - ], - nextCursor: null, - hasMoreHistory: false, - }); - expect(yield* Fiber.join(loading)).toEqual({ _tag: "loaded" }); + const loading = yield* h.loadEarlier().pipe(Effect.forkScoped); + const request = yield* Queue.take(h.olderLoads); + expect(request.cursor).toBe("socket-older-1"); + const olderItem = { + id: TurnItemId.make("socket-fallback-older-item"), + type: "command_execution" as const, + threadId: THREAD_ID, + runId: null, + nodeId: null, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 1, + status: "completed" as const, + title: "Earlier command", + input: "pwd", + output: "/workspace", + exitCode: 0, + startedAt: THREAD.thread.createdAt, + completedAt: THREAD.thread.createdAt, + updatedAt: THREAD.thread.createdAt, + }; + yield* Deferred.succeed(request.response, { + snapshotSequence: 12, + items: [ + { + position: 0, + visibility: "local", + sourceThreadId: THREAD_ID, + sourceItemId: olderItem.id, + item: olderItem, + }, + ], + nextCursor: null, + hasMoreHistory: false, + }); + expect(yield* Fiber.join(loading)).toEqual({ _tag: "loaded" }); - const state = h.registry.get(h.stateAtom); - expect( - Option.getOrThrow(state.data).visibleTurnItems.map((row) => row.sourceItemId), - ).toEqual([olderItem.id]); - expect(state.history).toMatchObject({ - historyCursor: null, - hasMoreHistory: false, - loading: false, - error: null, - expanded: true, - latestLocalTurnOrdinal: 4, - }); - expect(h.counts().httpLoads).toBe(1); + const state = h.registry.get(h.stateAtom); + expect(Option.getOrThrow(state.data).visibleTurnItems.map((row) => row.sourceItemId)).toEqual( + [olderItem.id], + ); + expect(state.history).toMatchObject({ + historyCursor: null, + hasMoreHistory: false, + loading: false, + error: null, + expanded: true, + latestLocalTurnOrdinal: 4, + }); + expect(h.counts().httpLoads).toBe(0); - unmount(); - yield* Deferred.await(subscription.closed); - }), + unmount(); + yield* Deferred.await(subscription.closed); + }), ); }); diff --git a/packages/client-runtime/src/state/threads-sync.test.ts b/packages/client-runtime/src/state/threads-sync.test.ts index 20adc0e05fa0..ea37770d24d3 100644 --- a/packages/client-runtime/src/state/threads-sync.test.ts +++ b/packages/client-runtime/src/state/threads-sync.test.ts @@ -81,10 +81,10 @@ const BASE_PROJECTION: OrchestrationV2ThreadProjection = { type TestThreadInput = OrchestrationV2ThreadStreamItem | Error; -function testSession(client: WsRpcProtocolClient): RpcSession.RpcSession { +function testSession(client: WsRpcProtocolClient, config: object = {}): RpcSession.RpcSession { return { client, - initialConfig: Effect.succeed({} as never), + initialConfig: Effect.succeed(config as never), subscribeServerConfig: (input = {}) => client.subscribeServerConfig(input), ready: Effect.void, probe: Effect.void, @@ -118,6 +118,7 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o readonly loadCached?: ReturnType; readonly saveThread?: Persistence.EnvironmentCacheStore["Service"]["saveThread"]; readonly historyPaging?: "enabled" | "no-http" | "no-controller"; + readonly socketSnapshots?: true; readonly historyHttpClient?: HttpClient.HttpClient; readonly getTurnItem?: (input: { readonly itemId: TurnItemId; @@ -171,7 +172,9 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o options?.getTurnItem?.(input) ?? Effect.die("Unexpected getTurnItem request"), } as unknown as WsRpcProtocolClient; const supervisorSession = yield* SubscriptionRef.make>( - Option.some(testSession(client)), + Option.some( + testSession(client, options?.socketSnapshots ? { threadSnapshotPagination: true } : {}), + ), ); const prepared = yield* SubscriptionRef.make>( Option.some(PREPARED), @@ -780,6 +783,33 @@ describe("EnvironmentThreads", () => { }), ); + it.effect( + "opens over the socket alone, and asks HTTP whether a thread exists after it fails", + () => + Effect.gen(function* () { + const harness = yield* makeHarness({ + httpSnapshot: { _tag: "missing" }, + socketSnapshots: true, + }); + yield* SubscriptionRef.changes(harness.subscriptionCount).pipe( + Stream.filter((count) => count === 1), + Stream.runHead, + ); + expect(yield* Ref.get(harness.lastAcceptBoundedSnapshot)).toBe(true); + expect(yield* Ref.get(harness.loaderCalls)).toBe(0); + + yield* Queue.offer(harness.inputs, new Error("Thread not found")); + yield* TestClock.adjust("250 millis"); + const settled = yield* awaitThreadState( + harness.observed, + (value) => value.status === "deleted" || Option.isSome(value.error), + ); + // The HTTP retry decides, so a missing thread never flashes an error. + expect(settled.status).toBe("deleted"); + expect(yield* Ref.get(harness.loaderCalls)).toBe(1); + }), + ); + it.effect("resumes a warm cache via afterSequence without an HTTP fetch", () => Effect.gen(function* () { const harness = yield* makeHarness({ cached: BASE_PROJECTION }); diff --git a/packages/client-runtime/src/state/threads.ts b/packages/client-runtime/src/state/threads.ts index 872863f93f77..ea6147f15dc6 100644 --- a/packages/client-runtime/src/state/threads.ts +++ b/packages/client-runtime/src/state/threads.ts @@ -948,6 +948,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make // status. A replacement session or foreground resubscribe on the same scope // may have missed events, so those show sync progress until confirmed. const resumingLive = yield* Ref.make(initialState.status === "live"); + const socketSnapshotFailed = yield* Ref.make(false); const markSynchronizing = Effect.gen(function* () { if (yield* Ref.get(resumingLive)) return; yield* SubscriptionRef.update(state, (current) => @@ -970,12 +971,20 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make return yield* Effect.never; } - yield* Effect.ignore(session.initialConfig); + const config = yield* Effect.option(session.initialConfig); yield* Ref.set(awaitingCompletion, true); yield* markSynchronizing; yield* Ref.set(resumingLive, false); - if (Option.isNone(current.data)) { + // A server that pages snapshots sends the same bounded window over the + // socket, so a cold open skips the HTTP round trip. After a socket + // failure, HTTP tells a missing thread apart from a failed load. + const socketSnapshot = + canLoadHistory && + !(yield* Ref.get(socketSnapshotFailed)) && + Option.exists(config, (value) => value.threadSnapshotPagination === true); + + if (Option.isNone(current.data) && !socketSnapshot) { const prepared = yield* SubscriptionRef.get(supervisor.prepared).pipe( Effect.flatMap( Option.match({ @@ -1054,7 +1063,14 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make }), { onDefect: () => setStreamError("Could not synchronize the thread."), - onExpectedFailure: (cause) => setStreamError(formatThreadError(cause)), + // A socket-only open that fails retries over HTTP, which tells a + // missing thread apart from a failure, before showing an error. + onExpectedFailure: Effect.fnUntraced(function* (cause) { + const retryOverHttp = + !(yield* Ref.getAndSet(socketSnapshotFailed, true)) && + Option.isNone((yield* SubscriptionRef.get(state)).data); + if (!retryOverHttp) yield* setStreamError(formatThreadError(cause)); + }), retryExpectedFailureAfter: "250 millis", resubscribe: foregroundResubscriptions, }, From c0036527b7b1797b661e3380aac66cca31f6a3f3 Mon Sep 17 00:00:00 2001 From: Bil0000 <62337003+Bil0000@users.noreply.github.com> Date: Mon, 9 Nov 2026 03:22:57 +0100 Subject: [PATCH 3/3] perf(mobile): reconnect reuses the descriptor its host check just read Resuming from the background checks the host with a descriptor GET, then the reconnect read the same descriptor again before asking for a ticket. The host check now hands its answer to the next direct authorization of that address, once and only within 5 seconds. --- .../src/authorization/layer.test.ts | 43 +++++++++++++++++++ .../src/authorization/service.ts | 29 +++++++++++-- .../client-runtime/src/connection/driver.ts | 14 ++++-- .../src/connection/resolver.test.ts | 1 + .../src/state/environmentHttpAuth.test.ts | 1 + 5 files changed, 81 insertions(+), 7 deletions(-) diff --git a/packages/client-runtime/src/authorization/layer.test.ts b/packages/client-runtime/src/authorization/layer.test.ts index 9ed7b9a63e9d..1b5b00319def 100644 --- a/packages/client-runtime/src/authorization/layer.test.ts +++ b/packages/client-runtime/src/authorization/layer.test.ts @@ -317,6 +317,49 @@ describe("RemoteEnvironmentAuthorization", () => { }), ); + it.effect("reuses a descriptor a host check just read, once and only while fresh", () => + Effect.gen(function* () { + const descriptor = { + ...DESCRIPTOR, + orchestrationProtocolVersion: ORCHESTRATION_PROTOCOL_VERSION, + }; + const harness = yield* makeHarness({ + responses: [ + websocketTicket("first-ticket"), + Response.json(descriptor), + websocketTicket("second-ticket"), + Response.json(descriptor), + websocketTicket("third-ticket"), + ], + }); + + yield* Effect.gen(function* () { + const remote = yield* RemoteEnvironmentAuthorization.RemoteEnvironmentAuthorization; + const authorize = remote.authorizeBearer({ + expectedEnvironmentId: ENVIRONMENT_ID, + httpBaseUrl: ENDPOINT.httpBaseUrl, + wsBaseUrl: ENDPOINT.wsBaseUrl, + bearerToken: "bearer-token", + connectionMethod: "direct", + }); + yield* remote.rememberCheckedDescriptor(ENDPOINT.httpBaseUrl, descriptor as never); + yield* authorize; + yield* authorize; + yield* remote.rememberCheckedDescriptor(ENDPOINT.httpBaseUrl, descriptor as never); + yield* TestClock.adjust("6 seconds"); + yield* authorize; + }).pipe(Effect.provide(harness.layer)); + + expect(harness.fetch.calls.map(([url]) => new URL(String(url)).pathname)).toEqual([ + "/api/auth/websocket-ticket/rpc", + "/.well-known/t3/environment", + "/api/auth/websocket-ticket/rpc", + "/.well-known/t3/environment", + "/api/auth/websocket-ticket/rpc", + ]); + }), + ); + it.effect("blocks a warm direct bearer prepare when the environment changes identity", () => Effect.gen(function* () { const harness = yield* makeHarness({ diff --git a/packages/client-runtime/src/authorization/service.ts b/packages/client-runtime/src/authorization/service.ts index 9b4d9c2f1f89..a93743c02215 100644 --- a/packages/client-runtime/src/authorization/service.ts +++ b/packages/client-runtime/src/authorization/service.ts @@ -82,6 +82,11 @@ export class RemoteEnvironmentAuthorization extends Context.Service< readonly expectedEnvironmentId: EnvironmentId; readonly rejectedAccessToken?: string; }) => Effect.Effect; + /** Lets the next direct authorization of this address skip its own descriptor read. */ + readonly rememberCheckedDescriptor: ( + httpBaseUrl: string, + descriptor: ExecutionEnvironmentDescriptor, + ) => Effect.Effect; /** * Renews a live socket's token before it expires, so the server moves the * socket to the new session instead of closing it. Runs until interrupted. @@ -95,6 +100,7 @@ export class RemoteEnvironmentAuthorization extends Context.Service< const CACHED_ENDPOINT_SOCKET_TIMEOUT_MS = 3_000; const DPOP_AUTHORIZATION_TIMEOUT_MS = 30_000; +const CHECKED_DESCRIPTOR_TTL_MS = 5_000; function mapDpopSocketError(error: RemoteEnvironmentAuthError | ConnectionAttemptError) { return error._tag === "ConnectionTransientError" || error._tag === "ConnectionBlockedError" @@ -135,6 +141,13 @@ export const make = Effect.gen(function* () { } >(); + // A host check that just answered hands its descriptor to the reconnect it + // triggers, once, so the reconnect skips a round trip. + const checkedDescriptors = new Map< + string, + { readonly descriptor: ExecutionEnvironmentDescriptor; readonly at: number } + >(); + /** * A saved LAN address can belong to another machine on a different network. */ @@ -144,9 +157,15 @@ export const make = Effect.gen(function* () { readonly httpBaseUrl: string; readonly connectionMethod: ClientConnectionMethod; }) { - const descriptor = yield* fetchDescriptor(input.httpBaseUrl, input.connectionMethod).pipe( - Effect.provideService(HttpClient.HttpClient, httpClient), - ); + const checked = checkedDescriptors.get(input.httpBaseUrl); + checkedDescriptors.delete(input.httpBaseUrl); + const descriptor = + checked !== undefined && + (yield* Clock.currentTimeMillis) - checked.at <= CHECKED_DESCRIPTOR_TTL_MS + ? checked.descriptor + : yield* fetchDescriptor(input.httpBaseUrl, input.connectionMethod).pipe( + Effect.provideService(HttpClient.HttpClient, httpClient), + ); if (descriptor.environmentId !== input.expectedEnvironmentId) { return yield* environmentMismatchError({ expected: input.expectedEnvironmentId, @@ -527,6 +546,10 @@ export const make = Effect.gen(function* () { return RemoteEnvironmentAuthorization.of({ authorizeBearer, + // Untraced: it runs on every live session's host check. + rememberCheckedDescriptor: Effect.fnUntraced(function* (httpBaseUrl, descriptor) { + checkedDescriptors.set(httpBaseUrl, { descriptor, at: yield* Clock.currentTimeMillis }); + }), authorizeDpop: Effect.fn("environment.authorization")(function (input) { return authorizeDpop(input); }), diff --git a/packages/client-runtime/src/connection/driver.ts b/packages/client-runtime/src/connection/driver.ts index 6039f25fa99d..421d64c369a6 100644 --- a/packages/client-runtime/src/connection/driver.ts +++ b/packages/client-runtime/src/connection/driver.ts @@ -147,13 +147,18 @@ export const make = Effect.gen(function* () { // The descriptor is public, so this sends no credential to whatever // answers at a saved LAN address on a different network. - const checkDescriptor = (httpBaseUrl: string, environmentId: string) => + const checkDescriptor = (httpBaseUrl: string, environmentId: string, remember = false) => fetchRemoteEnvironmentDescriptor({ httpBaseUrl, timeoutMs: ROUTE_CHECK_TIMEOUT_MS, }).pipe( - Effect.map((descriptor): RouteCheck => - descriptor.environmentId === environmentId ? "answered" : "silent", + Effect.flatMap((descriptor) => + descriptor.environmentId !== environmentId + ? Effect.succeed("silent") + : (remember + ? remote.rememberCheckedDescriptor(httpBaseUrl, descriptor) + : Effect.void + ).pipe(Effect.as("answered")), ), Effect.orElseSucceed((): RouteCheck => "silent"), // A cached descriptor would answer for a host that cannot be reached. @@ -171,8 +176,9 @@ export const make = Effect.gen(function* () { ); }; + // Only a live session's host check feeds the reconnect it may trigger. const checkHost = (prepared: PreparedConnection) => - checkDescriptor(prepared.httpBaseUrl, prepared.environmentId).pipe( + checkDescriptor(prepared.httpBaseUrl, prepared.environmentId, true).pipe( Effect.withSpan("ConnectionDriver.checkHost", { attributes: { "connection.target.kind": prepared.target._tag }, }), diff --git a/packages/client-runtime/src/connection/resolver.test.ts b/packages/client-runtime/src/connection/resolver.test.ts index 82a2aafa0014..62d43bbbe641 100644 --- a/packages/client-runtime/src/connection/resolver.test.ts +++ b/packages/client-runtime/src/connection/resolver.test.ts @@ -108,6 +108,7 @@ const makeDependencies = Effect.fn("TestConnectionResolver.makeDependencies")((o }, })), authorizeDpopHttp: () => Effect.die("unused"), + rememberCheckedDescriptor: () => Effect.void, renewDpopBeforeExpiry: () => Effect.die("unused"), }); const ssh = ClientCapabilities.SshEnvironmentGateway.of({ diff --git a/packages/client-runtime/src/state/environmentHttpAuth.test.ts b/packages/client-runtime/src/state/environmentHttpAuth.test.ts index 62bfc4f229ad..faac1a87ff6a 100644 --- a/packages/client-runtime/src/state/environmentHttpAuth.test.ts +++ b/packages/client-runtime/src/state/environmentHttpAuth.test.ts @@ -146,6 +146,7 @@ function makeHarness(reply: (requestNumber: number) => Response | Promise Effect.void, renewDpopBeforeExpiry: () => Effect.die("HTTP requests must not renew a socket's token."), }); const signer = ManagedRelay.ManagedRelayDpopSigner.of({