diff --git a/packages/client-runtime/src/state/shell-sync.test.ts b/packages/client-runtime/src/state/shell-sync.test.ts index 96a5b59c5d743..04c165ee72fed 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 { type ThreadPullRequestLink, } from "@t3tools/contracts"; import { describe, expect, it } from "@effect/vitest"; +import * as Cause from "effect/Cause"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; @@ -1580,6 +1581,73 @@ describe("environment shell cold start", () => { }), ); + it.effect("keeps syncing row status after a shell stream defect", () => + Effect.gen(function* () { + const h = yield* makeColdStart(RECENT_ORIGIN); + const rowUpdate = (sequence: number, status: "completed" | "running") => ({ + kind: "thread.updated" as const, + sequence, + location: "active" as const, + thread: { ...v2ThreadShell, status }, + }); + const liveRow = (sequence: number, status: "completed" | "running") => + SubscriptionRef.changes(h.shellState).pipe( + Stream.filter( + (state) => + state.status === "live" && + Option.exists( + state.snapshot, + (snapshot) => + snapshot.snapshotSequence === sequence && + snapshot.threads.some( + (thread) => thread.id === v2ThreadShell.id && thread.status === status, + ), + ), + ), + Stream.runHead, + Effect.map(Option.getOrThrow), + ); + + const first = yield* Queue.take(h.subscriptions); + yield* Queue.offerAll(first.events, [ + rowUpdate(CACHED_SEQUENCE + 1, "completed"), + { kind: "synchronized" }, + ]); + yield* liveRow(CACHED_SEQUENCE + 1, "completed"); + + const failWithDefect = (events: typeof first.events) => + Queue.failCause(events, Cause.die(new Error("shell invariant failed"))).pipe( + Effect.andThen( + SubscriptionRef.changes(h.shellState).pipe( + Stream.filter((state) => + Option.exists( + state.error, + (error) => error === "Could not synchronize environment data.", + ), + ), + Stream.runHead, + ), + ), + ); + + yield* failWithDefect(first.events); + yield* h.reconnect; + const resumed = yield* Queue.take(h.subscriptions); + expect(resumed.input).toMatchObject({ afterSequence: CACHED_SEQUENCE + 1 }); + yield* Queue.offerAll(resumed.events, [ + rowUpdate(CACHED_SEQUENCE + 2, "running"), + { kind: "synchronized" }, + ]); + expect((yield* liveRow(CACHED_SEQUENCE + 2, "running")).error).toEqual(Option.none()); + + // Without a reconnect or wakeup, the stream resubscribes once the backoff passes. + yield* failWithDefect(resumed.events); + yield* TestClock.adjust("250 millis"); + const retried = yield* Queue.take(h.subscriptions); + expect(retried.input).toMatchObject({ afterSequence: CACHED_SEQUENCE + 2 }); + }), + ); + it.effect("reloads over HTTP when the server answers a resume with snapshot-required", () => Effect.gen(function* () { const h = yield* makeColdStart(RECENT_ORIGIN); diff --git a/packages/client-runtime/src/state/shell.ts b/packages/client-runtime/src/state/shell.ts index 86584c4a03faa..cf5462d3f3e2f 100644 --- a/packages/client-runtime/src/state/shell.ts +++ b/packages/client-runtime/src/state/shell.ts @@ -30,7 +30,7 @@ import * as ConnectionWakeups from "../connection/wakeups.ts"; import { safeErrorLogAttributes } from "../errors/safeLog.ts"; import * as Persistence from "../platform/persistence.ts"; import { runCachePersistence } from "./cachePersistence.ts"; -import { subscribeDynamicWithSession } from "../rpc/client.ts"; +import { subscribeDynamicWithSession, subscriptionRetryDelay } from "../rpc/client.ts"; import type { RpcSession } from "../rpc/session.ts"; import * as ShellSnapshotLoader from "./shellSnapshotHttp.ts"; import type { DeferredShellSnapshot } from "./shellPullRequests.ts"; @@ -203,6 +203,7 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") // Only a server snapshot renews it; `snapshot-required` drops it. const origin = yield* Ref.make(cachedOrigin); const reloadRequests = yield* Queue.unbounded(); + let defectRetries = 0; const latestLive = yield* Ref.make>(Option.none()); const persistence = yield* Queue.sliding(1); @@ -702,11 +703,27 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") }; }), { + // A defect parks the stream, so resubscribe after a backoff rather than wait for a wakeup. + onDefect: (cause) => + setStreamError(Cause.squash(cause)).pipe( + Effect.andThen(Effect.sleep(subscriptionRetryDelay("250 millis", defectRetries++))), + Effect.andThen(Queue.offer(reloadRequests, undefined)), + ), onExpectedFailure: (cause) => setStreamError(Cause.squash(cause)), retryExpectedFailureAfter: "250 millis", resubscribe: resubscriptions, }, - ).pipe(Stream.runForEachArray((entries) => applyItems(entries))), + ).pipe( + Stream.runForEachArray((entries) => + applyItems(entries).pipe( + Effect.andThen( + Effect.sync(() => { + defectRetries = 0; + }), + ), + ), + ), + ), ); yield* SubscriptionRef.changes(supervisor.state).pipe( Stream.runForEach((connectionState) => {