fix(client): fall back when HTTP snapshots fail

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<OrchestrationV2ShellStreamItem>();
+      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.Option<RpcSession.RpcSession>>(
+        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<OrchestrationV2ShellStreamItem>();

packages/client-runtime/src/state/shell.ts

@@ -171,7 +171,19 @@ 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.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<OrchestrationV2ShellSnapshot>()),
+                    ),
+                  ),
+                )
               : Option.none<OrchestrationV2ShellSnapshot>();
           });
 

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<Option.Option<OrchestrationV2ShellSnapshot>>;
+    ) => Effect.Effect<Option.Option<OrchestrationV2ShellSnapshot>, RemoteEnvironmentRequestError>;
   }
 >()("@t3tools/client-runtime/state/shellSnapshotHttp/ShellSnapshotLoader") {}
 

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<Option.Option<OrchestrationV2ThreadDetailSnapshot>>;
+    ) => Effect.Effect<
+      Option.Option<OrchestrationV2ThreadDetailSnapshot>,
+      RemoteEnvironmentRequestError
+    >;
   }
 >()("@t3tools/client-runtime/state/threadSnapshotHttp/ThreadSnapshotLoader") {}
 

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<OrchestrationV2ThreadDetailSnapshot>;
+  readonly httpSnapshotFailure?: RemoteEnvironmentRequestError;
 }) {
   const inputs = yield* Queue.unbounded<TestThreadInput>();
   const observed = yield* Queue.unbounded<EnvironmentThreadState>();
@@ -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<OrchestrationV2ThreadDetailSnapshot>())
-            : Option.none<OrchestrationV2ThreadDetailSnapshot>(),
+        Effect.andThen(
+          options?.httpSnapshotFailure === undefined
+            ? Effect.succeed(
+                threadId === THREAD_ID
+                  ? (options?.httpSnapshot ?? Option.none<OrchestrationV2ThreadDetailSnapshot>())
+                  : Option.none<OrchestrationV2ThreadDetailSnapshot>(),
+              )
+            : 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 });

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,7 +226,20 @@ 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.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<OrchestrationV2ThreadDetailSnapshot>()),
+                    ),
+                  ),
+                )
               : Option.none<OrchestrationV2ThreadDetailSnapshot>();
           });