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/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/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/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({ 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, }, 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)