From 2f4d9167e1ff0453c22bf52e4d608db06ba03cc7 Mon Sep 17 00:00:00 2001 From: Pixel Perfect Date: Fri, 10 Jul 2026 11:43:52 -0700 Subject: [PATCH 1/2] fix(client): fall back when HTTP snapshots fail --- .../src/state/shell-sync.test.ts | 68 +++++++++++++++++++ packages/client-runtime/src/state/shell.ts | 4 +- .../src/state/shellSnapshotHttp.ts | 8 ++- .../src/state/threadSnapshotHttp.ts | 5 +- .../src/state/threads-sync.test.ts | 46 +++++++++++-- packages/client-runtime/src/state/threads.ts | 6 +- 6 files changed, 128 insertions(+), 9 deletions(-) diff --git a/packages/client-runtime/src/state/shell-sync.test.ts b/packages/client-runtime/src/state/shell-sync.test.ts index fadf4736fb7..357d3684a85 100644 --- a/packages/client-runtime/src/state/shell-sync.test.ts +++ b/packages/client-runtime/src/state/shell-sync.test.ts @@ -8,6 +8,7 @@ import { describe, expect, it } from "@effect/vitest"; import * as Effect from "effect/Effect"; import * as Option from "effect/Option"; import * as Queue from "effect/Queue"; +import * as Ref from "effect/Ref"; import * as Stream from "effect/Stream"; import * as SubscriptionRef from "effect/SubscriptionRef"; @@ -18,6 +19,7 @@ import { } from "../connection/model.ts"; import * as EnvironmentSupervisor from "../connection/supervisor.ts"; import * as Persistence from "../platform/persistence.ts"; +import { RemoteEnvironmentAuthFetchError } from "../rpc/http.ts"; import * as RpcSession from "../rpc/session.ts"; import type { WsRpcProtocolClient } from "../rpc/protocol.ts"; import { makeEnvironmentShellState, ShellSnapshotLoader } from "./shell.ts"; @@ -55,6 +57,72 @@ function session(client: WsRpcProtocolClient): RpcSession.RpcSession { } describe("environment shell synchronization", () => { + it.effect("uses the socket snapshot when the HTTP snapshot load fails", () => + Effect.gen(function* () { + const events = yield* Queue.unbounded(); + const subscribeInput = yield* Ref.make<{ readonly afterSequence?: number } | null>(null); + const loaderCalls = yield* Ref.make(0); + const client = { + [ORCHESTRATION_V2_WS_METHODS.subscribeShell]: (input: { + readonly afterSequence?: number; + }) => + Stream.unwrap(Ref.set(subscribeInput, input).pipe(Effect.as(Stream.fromQueue(events)))), + } as unknown as WsRpcProtocolClient; + const supervisorState = yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE); + const activeSession = yield* SubscriptionRef.make>( + Option.some(session(client)), + ); + const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({ + target: TARGET, + state: supervisorState, + session: activeSession, + prepared: yield* SubscriptionRef.make(Option.some(PREPARED)), + connect: Effect.void, + disconnect: Effect.void, + retryNow: Effect.void, + } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]); + const cache = Persistence.EnvironmentCacheStore.of({ + loadShell: () => Effect.succeed(Option.none()), + saveShell: () => Effect.void, + loadThread: () => Effect.succeed(Option.none()), + saveThread: () => Effect.void, + removeThread: () => Effect.void, + clear: () => Effect.void, + }); + const snapshotLoader = ShellSnapshotLoader.of({ + load: () => + Ref.update(loaderCalls, (count) => count + 1).pipe( + Effect.andThen( + Effect.fail( + new RemoteEnvironmentAuthFetchError({ + message: "HTTP snapshot failed", + cause: null, + }), + ), + ), + ), + }); + const shellState = yield* makeEnvironmentShellState().pipe( + Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), + Effect.provideService(Persistence.EnvironmentCacheStore, cache), + Effect.provideService(ShellSnapshotLoader, snapshotLoader), + ); + + yield* Queue.offer(events, { + kind: "snapshot", + snapshot: LIVE_SHELL_SNAPSHOT, + }); + const state = yield* SubscriptionRef.changes(shellState).pipe( + Stream.filter((current) => current.status === "live"), + Stream.runHead, + ); + + expect(yield* Ref.get(loaderCalls)).toBe(1); + expect(yield* Ref.get(subscribeInput)).toEqual({}); + expect(Option.getOrThrow(Option.getOrThrow(state).snapshot)).toEqual(LIVE_SHELL_SNAPSHOT); + }), + ); + it.effect("publishes live state before persistence and preserves it when ready", () => Effect.gen(function* () { const events = yield* Queue.unbounded(); diff --git a/packages/client-runtime/src/state/shell.ts b/packages/client-runtime/src/state/shell.ts index aed33f9e500..d4455e7cfe9 100644 --- a/packages/client-runtime/src/state/shell.ts +++ b/packages/client-runtime/src/state/shell.ts @@ -171,7 +171,9 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") Stream.runHead, ); return Option.isSome(prepared) - ? yield* snapshotLoader.load(prepared.value) + ? yield* snapshotLoader + .load(prepared.value) + .pipe(Effect.orElseSucceed(() => Option.none())) : Option.none(); }); diff --git a/packages/client-runtime/src/state/shellSnapshotHttp.ts b/packages/client-runtime/src/state/shellSnapshotHttp.ts index 2cb154aba08..173741a297e 100644 --- a/packages/client-runtime/src/state/shellSnapshotHttp.ts +++ b/packages/client-runtime/src/state/shellSnapshotHttp.ts @@ -9,7 +9,11 @@ import { HttpClient } from "effect/unstable/http"; import type { PreparedConnection } from "../connection/model.ts"; import { environmentEndpointUrl } from "../environment/endpoint.ts"; import { ManagedRelayDpopSigner } from "../relay/managedRelay.ts"; -import { executeEnvironmentHttpRequest, makeEnvironmentHttpApiClient } from "../rpc/http.ts"; +import { + executeEnvironmentHttpRequest, + makeEnvironmentHttpApiClient, + type RemoteEnvironmentRequestError, +} from "../rpc/http.ts"; import { buildEnvironmentAuthHeaders, withEnvironmentCredentials } from "./environmentHttpAuth.ts"; // Bounded so a pathologically slow endpoint cannot block the (cheaper) socket @@ -58,7 +62,7 @@ export class ShellSnapshotLoader extends Context.Service< { readonly load: ( prepared: PreparedConnection, - ) => Effect.Effect>; + ) => Effect.Effect, RemoteEnvironmentRequestError>; } >()("@t3tools/client-runtime/state/shellSnapshotHttp/ShellSnapshotLoader") {} diff --git a/packages/client-runtime/src/state/threadSnapshotHttp.ts b/packages/client-runtime/src/state/threadSnapshotHttp.ts index e539617c4fb..ed1a021799d 100644 --- a/packages/client-runtime/src/state/threadSnapshotHttp.ts +++ b/packages/client-runtime/src/state/threadSnapshotHttp.ts @@ -72,7 +72,10 @@ export class ThreadSnapshotLoader extends Context.Service< readonly load: ( prepared: PreparedConnection, threadId: ThreadId, - ) => Effect.Effect>; + ) => Effect.Effect< + Option.Option, + RemoteEnvironmentRequestError + >; } >()("@t3tools/client-runtime/state/threadSnapshotHttp/ThreadSnapshotLoader") {} diff --git a/packages/client-runtime/src/state/threads-sync.test.ts b/packages/client-runtime/src/state/threads-sync.test.ts index 469b7150dfc..80abc3107e1 100644 --- a/packages/client-runtime/src/state/threads-sync.test.ts +++ b/packages/client-runtime/src/state/threads-sync.test.ts @@ -26,6 +26,10 @@ import { } from "../connection/model.ts"; import * as EnvironmentSupervisor from "../connection/supervisor.ts"; import * as Persistence from "../platform/persistence.ts"; +import { + RemoteEnvironmentAuthFetchError, + type RemoteEnvironmentRequestError, +} from "../rpc/http.ts"; import * as RpcSession from "../rpc/session.ts"; import { v2Projection, v2ThreadId } from "./orchestrationV2TestFixtures.ts"; import { @@ -82,6 +86,7 @@ function awaitThreadState( const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (options?: { readonly cached?: OrchestrationV2ThreadProjection; readonly httpSnapshot?: Option.Option; + readonly httpSnapshotFailure?: RemoteEnvironmentRequestError; }) { const inputs = yield* Queue.unbounded(); const observed = yield* Queue.unbounded(); @@ -119,10 +124,14 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o const snapshotLoader = ThreadSnapshotLoader.of({ load: (_prepared, threadId) => Ref.update(loaderCalls, (count) => count + 1).pipe( - Effect.as( - threadId === THREAD_ID - ? (options?.httpSnapshot ?? Option.none()) - : Option.none(), + Effect.andThen( + options?.httpSnapshotFailure === undefined + ? Effect.succeed( + threadId === THREAD_ID + ? (options?.httpSnapshot ?? Option.none()) + : Option.none(), + ) + : Effect.fail(options.httpSnapshotFailure), ), ), }); @@ -306,6 +315,35 @@ describe("EnvironmentThreads", () => { }), ); + it.effect("uses the socket snapshot when the HTTP snapshot load fails", () => + Effect.gen(function* () { + const socketProjection: OrchestrationV2ThreadProjection = { + ...BASE_PROJECTION, + thread: { ...BASE_PROJECTION.thread, title: "Socket title" }, + }; + const harness = yield* makeHarness({ + httpSnapshotFailure: new RemoteEnvironmentAuthFetchError({ + message: "HTTP snapshot failed", + cause: null, + }), + }); + yield* Queue.offer(harness.inputs, snapshot(socketProjection, 1)); + + const state = yield* awaitThreadState( + harness.observed, + (value) => + value.status === "live" && + Option.isSome(value.data) && + value.data.value.thread.title === "Socket title", + ); + + expect(Option.getOrThrow(state.data)).toEqual(socketProjection); + expect(yield* Ref.get(harness.loaderCalls)).toBe(1); + expect(yield* Ref.get(harness.subscriptionCount)).toBe(1); + expect(yield* Ref.get(harness.lastSubscribeAfterSequence)).toBeUndefined(); + }), + ); + it.effect("ignores replayed thread events at or below the snapshot sequence", () => 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 ce9227be532..c39c40d95d6 100644 --- a/packages/client-runtime/src/state/threads.ts +++ b/packages/client-runtime/src/state/threads.ts @@ -225,7 +225,11 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make Stream.runHead, ); return Option.isSome(prepared) - ? yield* snapshotLoader.load(prepared.value, threadId) + ? yield* snapshotLoader + .load(prepared.value, threadId) + .pipe( + Effect.orElseSucceed(() => Option.none()), + ) : Option.none(); }); From bdcfd6ab6c6358fbacdfc8cd5c2d9a87230ebf97 Mon Sep 17 00:00:00 2001 From: Pixel Perfect Date: Fri, 10 Jul 2026 11:54:11 -0700 Subject: [PATCH 2/2] fix(client): log HTTP snapshot fallback --- packages/client-runtime/src/state/shell.ts | 16 +++++++++++++--- packages/client-runtime/src/state/threads.ts | 20 +++++++++++++++----- 2 files changed, 28 insertions(+), 8 deletions(-) diff --git a/packages/client-runtime/src/state/shell.ts b/packages/client-runtime/src/state/shell.ts index d4455e7cfe9..b9cb985bc00 100644 --- a/packages/client-runtime/src/state/shell.ts +++ b/packages/client-runtime/src/state/shell.ts @@ -171,9 +171,19 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") Stream.runHead, ); return Option.isSome(prepared) - ? yield* snapshotLoader - .load(prepared.value) - .pipe(Effect.orElseSucceed(() => Option.none())) + ? yield* snapshotLoader.load(prepared.value).pipe( + Effect.catch((error) => + Effect.logWarning( + "Could not load the environment shell snapshot over HTTP; using the socket snapshot instead.", + ).pipe( + Effect.annotateLogs({ + environmentId, + ...safeErrorLogAttributes(error), + }), + Effect.as(Option.none()), + ), + ), + ) : Option.none(); }); diff --git a/packages/client-runtime/src/state/threads.ts b/packages/client-runtime/src/state/threads.ts index c39c40d95d6..cc6f660fd79 100644 --- a/packages/client-runtime/src/state/threads.ts +++ b/packages/client-runtime/src/state/threads.ts @@ -17,6 +17,7 @@ import { Atom } from "effect/unstable/reactivity"; import { EnvironmentRegistry } from "../connection/registry.ts"; import { connectionProjectionPhase } from "../connection/model.ts"; import { EnvironmentSupervisor } from "../connection/supervisor.ts"; +import { safeErrorLogAttributes } from "../errors/safeLog.ts"; import { EnvironmentCacheStore } from "../platform/persistence.ts"; import { subscribe } from "../rpc/client.ts"; import { ThreadSnapshotLoader } from "./threadSnapshotHttp.ts"; @@ -225,11 +226,20 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make Stream.runHead, ); return Option.isSome(prepared) - ? yield* snapshotLoader - .load(prepared.value, threadId) - .pipe( - Effect.orElseSucceed(() => Option.none()), - ) + ? yield* snapshotLoader.load(prepared.value, threadId).pipe( + Effect.catch((error) => + Effect.logWarning( + "Could not load the thread snapshot over HTTP; using the socket snapshot instead.", + ).pipe( + Effect.annotateLogs({ + environmentId, + threadId, + ...safeErrorLogAttributes(error), + }), + Effect.as(Option.none()), + ), + ), + ) : Option.none(); });