From fd95723cf6a0ab3b79162f899d7c912c62115271 Mon Sep 17 00:00:00 2001 From: Daniel Gordon Date: Thu, 23 Jul 2026 01:00:37 +0000 Subject: [PATCH] fix(sync): stop the shell thread list from wedging on a stale cache MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The sidebar thread list is a snapshot cached in IndexedDB that resumes over a persisted event cursor. Two gaps let a client pin itself to a stale list that survived reloads, because a reload re-reads the same cache and re-attempts the same resume: - The shell subscription passed no `retryExpectedFailureAfter`, so an expected (non-transport) failure ended the stream for the life of the page. The cached snapshot kept rendering and never updated again. - The resume cursor was captured once at boot, so every reconnect replayed the same catch-up window instead of resuming from where the client actually got to. - The server always honored `afterSequence`. A cursor far behind our head asks for a replay costing far more than the snapshot it stands in for (every thread-aggregate event is a projection lookup plus a full thread shell on the socket), and a client that cannot afford that replay never advances its cursor, so its next attempt is just as expensive. A cursor *ahead* of our head — a rebuilt or restored event log — replayed nothing while every live event was dropped by the client's sequence check. Adds the retry, resolves the resume cursor per subscribe, and falls back to a snapshot when the client's cursor is unusable or too far behind. `subscribe` now accepts a thunk so cursor-carrying inputs are re-read on every attempt. Co-Authored-By: Claude Opus 4.8 --- apps/server/src/server.test.ts | 110 ++++++++++++++++++ apps/server/src/ws.ts | 72 ++++++++++-- packages/client-runtime/src/rpc/client.ts | 9 +- .../src/state/shell-sync.test.ts | 87 ++++++++++++++ packages/client-runtime/src/state/shell.ts | 19 ++- 5 files changed, 279 insertions(+), 18 deletions(-) diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index ee420840be5..f75b7fc8d1a 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -5624,6 +5624,116 @@ it.layer(NodeServices.layer)("server router seam", (it) => { }).pipe(Effect.provide(NodeHttpServer.layerTest)), ); + // A resuming shell client is only replayed when the replay is both possible + // and cheaper than the snapshot it stands in for. See + // SHELL_CATCH_UP_REPLAY_LIMIT in ws.ts. + const subscribeShellResume = (input: { + readonly afterSequence: number; + readonly projectedSequence: number; + readonly onReadEvents: () => void; + }) => + Effect.gen(function* () { + yield* buildAppUnderTest({ + layers: { + orchestrationEngine: { + readEvents: (_fromSequenceExclusive, _limit) => + Stream.suspend(() => { + input.onReadEvents(); + // `project.deleted` maps to a shell event without any further + // projection lookup, so it isolates the replay-vs-snapshot + // decision from the rest of the shell mapping. + return Stream.make({ + sequence: input.afterSequence + 1, + eventId: EventId.make("event-replayed"), + aggregateKind: "project", + aggregateId: defaultProjectId, + occurredAt: "2026-04-05T00:00:00.000Z", + commandId: null, + causationEventId: null, + correlationId: null, + metadata: {}, + type: "project.deleted", + payload: { + projectId: defaultProjectId, + deletedAt: "2026-04-05T00:00:00.000Z", + }, + } satisfies Extract); + }), + }, + projectionSnapshotQuery: { + getSnapshotSequence: () => + Effect.succeed({ snapshotSequence: input.projectedSequence }), + }, + }, + }); + + const wsUrl = yield* getWsServerUrl("/ws"); + const items = yield* Effect.scoped( + withWsRpcClient(wsUrl, (client) => + client[ORCHESTRATION_WS_METHODS.subscribeShell]({ + afterSequence: input.afterSequence, + }).pipe(Stream.take(1), Stream.runCollect), + ), + ); + return items[0]; + }); + + it.effect("replays shell events for a client resuming from a recent sequence", () => + Effect.gen(function* () { + let readEventsCalls = 0; + const first = yield* subscribeShellResume({ + afterSequence: 10, + projectedSequence: 12, + onReadEvents: () => { + readEventsCalls += 1; + }, + }); + + assert.equal(readEventsCalls, 1); + assert.equal(first?.kind, "project-removed"); + }).pipe(Effect.provide(NodeHttpServer.layerTest)), + ); + + it.effect("sends a shell snapshot instead of replaying when the client is far behind", () => + Effect.gen(function* () { + let readEventsCalls = 0; + const first = yield* subscribeShellResume({ + afterSequence: 10, + // Far enough behind that replaying the window would cost more than the + // snapshot, and slow enough to send that the client may never finish it. + projectedSequence: 100_000, + onReadEvents: () => { + readEventsCalls += 1; + }, + }); + + assert.equal(readEventsCalls, 0); + assert.equal(first?.kind, "snapshot"); + }).pipe(Effect.provide(NodeHttpServer.layerTest)), + ); + + it.effect( + "sends a shell snapshot when the client cursor is ahead of the projected sequence", + () => + Effect.gen(function* () { + let readEventsCalls = 0; + // A cursor past our own head means the client cached against an event log + // we no longer have (rebuilt, restored, or a different server). Replaying + // would send nothing and every live event would be dropped by the client's + // sequence check, leaving it stale forever. + const first = yield* subscribeShellResume({ + afterSequence: 500, + projectedSequence: 12, + onReadEvents: () => { + readEventsCalls += 1; + }, + }); + + assert.equal(readEventsCalls, 0); + assert.equal(first?.kind, "snapshot"); + }).pipe(Effect.provide(NodeHttpServer.layerTest)), + ); + it.effect("enriches replayed project events with repository identity metadata", () => Effect.gen(function* () { const repositoryIdentity = { diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 74853995616..f2990cc0e7d 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -282,6 +282,20 @@ function isThreadDetailEvent(event: OrchestrationEvent): event is Extract< const PROVIDER_STATUS_DEBOUNCE_MS = 200; +/** + * How far behind a resuming shell client may be before we send it a fresh + * snapshot instead of replaying its catch-up window. + * + * Replay is only a win while the gap is small. Every thread-aggregate event in + * the window costs a `getThreadShellById` query and emits a full thread shell + * on the socket, so a client that has been away for a few days can ask for tens + * of megabytes to rebuild a list that a single snapshot expresses in a few + * hundred kilobytes. Worse, a client that cannot afford the replay never + * advances its cursor, so its next attempt is just as expensive — it stays + * stale indefinitely. + */ +const SHELL_CATCH_UP_REPLAY_LIMIT = 500; + const RPC_REQUIRED_SCOPE = new Map([ [ORCHESTRATION_WS_METHODS.dispatchCommand, AuthOrchestrationOperateScope], [ORCHESTRATION_WS_METHODS.getTurnDiff, AuthOrchestrationReadScope], @@ -1084,6 +1098,19 @@ const makeWsRpcLayer = ( ), ); + const loadShellSnapshot = projectionSnapshotQuery.getShellSnapshot().pipe( + Effect.tapError((cause) => + Effect.logError("orchestration shell snapshot load failed", { cause }), + ), + Effect.mapError( + (cause) => + new OrchestrationGetSnapshotError({ + message: "Failed to load orchestration shell snapshot", + cause, + }), + ), + ); + // When the client already holds a shell snapshot (cached, or loaded // over HTTP) it passes that snapshot's sequence, and we resume by // replaying shell events after it instead of re-sending the whole @@ -1101,6 +1128,38 @@ const makeWsRpcLayer = ( yield* Effect.forkScoped( liveStream.pipe(Stream.runForEach((item) => Queue.offer(liveBuffer, item))), ); + + // A resume cursor is only usable if we can actually serve it. + // If it sits ahead of our projected sequence the client is + // resuming against an event log that no longer matches its + // cache (rebuilt, restored from backup, or a different + // server), and every live event we send will be discarded by + // its sequence check — it would render its stale cache + // forever. If it sits too far behind, the replay costs far + // more than the snapshot it stands in for. Both cases are + // repaired by sending a snapshot, which the client applies + // wholesale rather than merging by sequence. + const projectedSequence = yield* projectionSnapshotQuery + .getSnapshotSequence() + .pipe( + Effect.map(({ snapshotSequence }) => snapshotSequence), + // If we cannot read our own cursor, prefer the previous + // behaviour (replay) over forcing a full snapshot. + Effect.orElseSucceed(() => afterSequence), + ); + const gap = projectedSequence - afterSequence; + if (gap < 0 || gap > SHELL_CATCH_UP_REPLAY_LIMIT) { + yield* Effect.logInfo( + "orchestration shell resume fell back to a full snapshot", + { afterSequence, projectedSequence, gap }, + ); + const snapshot = yield* loadShellSnapshot; + return Stream.concat( + Stream.make({ kind: "snapshot" as const, snapshot }), + Stream.fromQueue(liveBuffer), + ); + } + const catchUpStream = orchestrationEngine .readEvents(afterSequence, Number.MAX_SAFE_INTEGER) .pipe( @@ -1121,18 +1180,7 @@ const makeWsRpcLayer = ( ); } - const snapshot = yield* projectionSnapshotQuery.getShellSnapshot().pipe( - Effect.tapError((cause) => - Effect.logError("orchestration shell snapshot load failed", { cause }), - ), - Effect.mapError( - (cause) => - new OrchestrationGetSnapshotError({ - message: "Failed to load orchestration shell snapshot", - cause, - }), - ), - ); + const snapshot = yield* loadShellSnapshot; return Stream.concat( Stream.make({ diff --git a/packages/client-runtime/src/rpc/client.ts b/packages/client-runtime/src/rpc/client.ts index 92892431e45..d36f9cfe093 100644 --- a/packages/client-runtime/src/rpc/client.ts +++ b/packages/client-runtime/src/rpc/client.ts @@ -149,7 +149,10 @@ export function runStream( export function subscribe( tag: TTag, - input: EnvironmentRpcInput, + // Pass a thunk when the input carries a resume cursor: it is re-read on every + // (re)subscribe, so a subscription that outlives several sessions resumes + // from where it actually got to rather than from where the page booted. + input: EnvironmentRpcInput | (() => EnvironmentRpcInput), options?: { readonly onExpectedFailure?: ( cause: Cause.Cause>, @@ -161,6 +164,8 @@ export function subscribe( EnvironmentRpcStreamFailure, EnvironmentSupervisor > { + const resolveInput = (): EnvironmentRpcInput => + typeof input === "function" ? (input as () => EnvironmentRpcInput)() : input; return Stream.unwrap( EnvironmentSupervisor.pipe( Effect.map((supervisor) => @@ -180,7 +185,7 @@ export function subscribe( EnvironmentRpcStreamFailure > => Stream.suspend(() => - method(input).pipe( + method(resolveInput()).pipe( Stream.catchCause((cause) => { const hasOnlyExpectedFailures = cause.reasons.length > 0 && diff --git a/packages/client-runtime/src/state/shell-sync.test.ts b/packages/client-runtime/src/state/shell-sync.test.ts index de240d58bbc..d69dd97d432 100644 --- a/packages/client-runtime/src/state/shell-sync.test.ts +++ b/packages/client-runtime/src/state/shell-sync.test.ts @@ -1,6 +1,7 @@ import { EnvironmentId, ORCHESTRATION_WS_METHODS, + ThreadId, type OrchestrationShellSnapshot, type OrchestrationShellStreamItem, } from "@t3tools/contracts"; @@ -8,8 +9,10 @@ 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"; +import * as TestClock from "effect/testing/TestClock"; import { AVAILABLE_CONNECTION_STATE, @@ -193,4 +196,88 @@ describe("environment shell synchronization", () => { expect(yield* SubscriptionRef.get(loaderCalls)).toBe(0); }), ); + + it.effect("retries a failed shell stream and resumes from the latest applied sequence", () => + Effect.gen(function* () { + const cachedSnapshot: OrchestrationShellSnapshot = { + snapshotSequence: 5, + projects: [], + threads: [], + updatedAt: "2026-06-06T00:00:00.000Z", + }; + // Records the resume cursor the client asks for on each subscribe attempt. + const attempts = yield* Ref.make>([]); + const client = { + [ORCHESTRATION_WS_METHODS.subscribeShell]: (input: { readonly afterSequence?: number }) => + Stream.unwrap( + Ref.updateAndGet(attempts, (seen) => [...seen, input.afterSequence]).pipe( + Effect.map((seen) => + // The first attempt advances the shell to sequence 9 and then + // fails with a domain (non-transport) error; without a retry the + // shell would stay on that snapshot for the life of the page. + seen.length === 1 + ? Stream.concat( + Stream.make({ + kind: "thread-removed" as const, + sequence: 9, + threadId: ThreadId.make("thread-1"), + }), + Stream.fail(new Error("shell stream failed")), + ) + : Stream.never, + ), + ), + ), + } 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.some(cachedSnapshot)), + saveShell: () => Effect.void, + loadThread: () => Effect.succeed(Option.none()), + saveThread: () => Effect.void, + removeThread: () => Effect.void, + clear: () => Effect.void, + }); + const snapshotLoader = ShellSnapshotLoader.of({ + load: () => Effect.succeed(Option.none()), + }); + yield* makeEnvironmentShellState().pipe( + Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), + Effect.provideService(Persistence.EnvironmentCacheStore, cache), + Effect.provideService(ShellSnapshotLoader, snapshotLoader), + ); + + for (let attempt = 0; attempt < 100; attempt += 1) { + if ((yield* Ref.get(attempts)).length >= 1) { + break; + } + yield* Effect.yieldNow; + } + expect(yield* Ref.get(attempts)).toEqual([5]); + + yield* TestClock.adjust("250 millis"); + for (let attempt = 0; attempt < 100; attempt += 1) { + if ((yield* Ref.get(attempts)).length >= 2) { + break; + } + yield* Effect.yieldNow; + } + + // Resubscribed (rather than ending the stream), and asked to resume from + // the sequence it actually reached rather than the boot-time cursor. + expect(yield* Ref.get(attempts)).toEqual([5, 9]); + }), + ); }); diff --git a/packages/client-runtime/src/state/shell.ts b/packages/client-runtime/src/state/shell.ts index faa70bc4f3a..562e448e974 100644 --- a/packages/client-runtime/src/state/shell.ts +++ b/packages/client-runtime/src/state/shell.ts @@ -123,6 +123,11 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") ), ); + // The sequence the subscription should resume from. Kept as plain closure + // state (rather than read back out of `state`) because the subscribe input + // thunk below is synchronous. It always tracks the applied snapshot. + let resumeSequence: number | null = null; + const applyItem = Effect.fn("EnvironmentShellState.applyItem")(function* ( item: OrchestrationShellStreamItem, ) { @@ -140,6 +145,7 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") if (nextSnapshot === null) { return; } + resumeSequence = nextSnapshot.snapshotSequence; yield* SubscriptionRef.set(state, { snapshot: Option.some(nextSnapshot), @@ -178,13 +184,18 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") yield* applyItem({ kind: "snapshot", snapshot: base.value }); } - const subscribeInput = Option.match(base, { - onNone: () => ({}), - onSome: (snapshot) => ({ afterSequence: snapshot.snapshotSequence }), - }); + // Resolved per (re)subscribe so reconnects resume from the latest applied + // sequence instead of replaying the boot-time catch-up window again. + const subscribeInput = () => + resumeSequence === null ? {} : { afterSequence: resumeSequence }; + // Without a retry an expected (non-transport) failure ends the stream for + // the life of the page: the shell keeps rendering its cached snapshot and + // silently never updates again, even across reloads, because the cache is + // what the next load resumes from. Mirrors the thread subscription. yield* subscribe(ORCHESTRATION_WS_METHODS.subscribeShell, subscribeInput, { onExpectedFailure: (cause) => setStreamError(Cause.squash(cause)), + retryExpectedFailureAfter: "250 millis", }).pipe(Stream.runForEach(applyItem)); }), );