diff --git a/packages/client-runtime/src/state/shell-sync.test.ts b/packages/client-runtime/src/state/shell-sync.test.ts index 5ce6d7dbaa290..c18a5a94299c2 100644 --- a/packages/client-runtime/src/state/shell-sync.test.ts +++ b/packages/client-runtime/src/state/shell-sync.test.ts @@ -17,6 +17,7 @@ import * as Ref from "effect/Ref"; import * as Scope from "effect/Scope"; import * as Stream from "effect/Stream"; import * as SubscriptionRef from "effect/SubscriptionRef"; +import * as TestClock from "effect/testing/TestClock"; import { AVAILABLE_CONNECTION_STATE, @@ -416,6 +417,69 @@ describe("environment shell synchronization", () => { }), ); + it.effect("resubscribes after a defect ends the shell subscription", () => + Effect.gen(function* () { + const defect = new Error("shell item could not be applied"); + const events = yield* Queue.unbounded(); + let subscriptions = 0; + const client = { + [ORCHESTRATION_V2_WS_METHODS.subscribeShell]: () => { + subscriptions += 1; + return subscriptions === 1 ? Stream.die(defect) : Stream.fromQueue(events); + }, + } as unknown as WsRpcProtocolClient; + const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({ + target: TARGET, + state: yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE), + session: yield* SubscriptionRef.make(Option.some(session(client))), + 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, + loadServerConfig: () => Effect.succeed(Option.none()), + saveServerConfig: () => Effect.void, + loadVcsRefs: () => Effect.succeed(Option.none()), + saveVcsRefs: () => Effect.void, + removeVcsRefs: () => Effect.void, + clearVcsRefs: () => Effect.void, + clear: () => Effect.void, + }); + const snapshotLoader = ShellSnapshotLoader.ShellSnapshotLoader.of({ + load: () => Effect.succeed(Option.some(LIVE_SHELL_SNAPSHOT)), + }); + const shellState = yield* makeEnvironmentShellState().pipe( + Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), + Effect.provideService(Persistence.EnvironmentCacheStore, cache), + Effect.provideService(ShellSnapshotLoader.ShellSnapshotLoader, snapshotLoader), + ); + + const failed = yield* SubscriptionRef.changes(shellState).pipe( + Stream.filter((state) => Option.isSome(state.error)), + Stream.runHead, + ); + expect(Option.isSome(failed)).toBe(true); + yield* TestClock.adjust("1 minute"); + yield* Queue.offer(events, { + kind: "snapshot", + snapshot: { ...LIVE_SHELL_SNAPSHOT, snapshotSequence: 5 }, + } as OrchestrationV2ShellStreamItem); + yield* SubscriptionRef.changes(shellState).pipe( + Stream.filter( + (state) => Option.isSome(state.snapshot) && state.snapshot.value.snapshotSequence === 5, + ), + Stream.runHead, + ); + expect(subscriptions).toBe(2); + }), + ); + it.effect("applies authoritative lower-sequence HTTP resets over client-ahead cache", () => Effect.gen(function* () { const cachedSnapshot: OrchestrationV2ShellSnapshot = { diff --git a/packages/client-runtime/src/state/shell.ts b/packages/client-runtime/src/state/shell.ts index 0a7e2aa7c6664..423f5eb7b1072 100644 --- a/packages/client-runtime/src/state/shell.ts +++ b/packages/client-runtime/src/state/shell.ts @@ -9,11 +9,13 @@ import { type ThreadPullRequestLink, } from "@t3tools/contracts"; import * as Cause from "effect/Cause"; +import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Fiber from "effect/Fiber"; import * as Option from "effect/Option"; import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; +import * as Schedule from "effect/Schedule"; import * as Semaphore from "effect/Semaphore"; import * as Stream from "effect/Stream"; import * as SubscriptionRef from "effect/SubscriptionRef"; @@ -66,6 +68,11 @@ function withoutDeferredPullRequests(value: DeferredShellSnapshot): Orchestratio } const SHELL_SYNCHRONIZATION_ERROR_MESSAGE = "Could not synchronize environment data."; +const SHELL_RESUBSCRIBE_SCHEDULE = Schedule.exponential("1 second").pipe( + Schedule.modifyDelay(({ duration }) => + Effect.succeed(Duration.min(duration, Duration.seconds(30))), + ), +); export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make")(function* () { const supervisor = yield* EnvironmentSupervisor.EnvironmentSupervisor; @@ -391,7 +398,19 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") retryExpectedFailureAfter: "250 millis", resubscribe: foregroundResubscriptions, }, - ).pipe(Stream.runForEachArray((items) => applyItems(items))), + ).pipe( + Stream.runForEachArray((items) => applyItems(items)), + // A defect ends the only subscription feeding the thread list, which + // would otherwise stay on screen, frozen, until the app restarts. + Effect.tapCause((cause) => + Cause.hasInterruptsOnly(cause) ? Effect.void : setStreamError(Cause.squash(cause)), + ), + Effect.sandbox, + Effect.retry({ + schedule: SHELL_RESUBSCRIBE_SCHEDULE, + while: (cause) => !Cause.hasInterruptsOnly(cause), + }), + ), ); yield* SubscriptionRef.changes(supervisor.state).pipe( Stream.runForEach((connectionState) => {