From 4076a4f8efa1d4f60eeebc391bc9088cc4af3e61 Mon Sep 17 00:00:00 2001 From: rolfilho Date: Fri, 30 Oct 2026 10:30:26 -0300 Subject: [PATCH 1/2] fix(client): recover snapshots that race thread creation --- .../src/state/threads-sync.test.ts | 84 +++++++++++++++++-- packages/client-runtime/src/state/threads.ts | 12 ++- 2 files changed, 88 insertions(+), 8 deletions(-) diff --git a/packages/client-runtime/src/state/threads-sync.test.ts b/packages/client-runtime/src/state/threads-sync.test.ts index 9956694aa193a..605726d324931 100644 --- a/packages/client-runtime/src/state/threads-sync.test.ts +++ b/packages/client-runtime/src/state/threads-sync.test.ts @@ -114,6 +114,7 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o readonly latestLocalTurnOrdinal?: number | null; }; readonly httpSnapshot?: ThreadSnapshotLoadResult; + readonly loadSnapshot?: () => Effect.Effect; readonly completionMarker?: boolean; readonly resumeCache?: NonNullable[1]>; readonly loadCached?: ReturnType; @@ -176,11 +177,15 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o const snapshotLoader = ThreadSnapshotLoader.ThreadSnapshotLoader.of({ load: (_prepared, threadId) => Ref.update(loaderCalls, (count) => count + 1).pipe( - Effect.as( - threadId === THREAD_ID - ? (options?.httpSnapshot ?? - ({ _tag: "unavailable" } satisfies ThreadSnapshotLoadResult)) - : ({ _tag: "unavailable" } satisfies ThreadSnapshotLoadResult), + Effect.andThen( + options?.loadSnapshot ?? + (() => + Effect.succeed( + threadId === THREAD_ID + ? (options?.httpSnapshot ?? + ({ _tag: "unavailable" } satisfies ThreadSnapshotLoadResult)) + : ({ _tag: "unavailable" } satisfies ThreadSnapshotLoadResult), + )), ), ), }); @@ -608,6 +613,75 @@ describe("EnvironmentThreads", () => { }), ); + it.effect("recovers when the first snapshot precedes thread creation", () => + Effect.gen(function* () { + const attempts = yield* Queue.unbounded(); + let created = false; + const harness = yield* makeHarness({ + loadSnapshot: () => + Queue.offer(attempts, undefined).pipe( + Effect.as( + created + ? { + _tag: "present" as const, + snapshot: { snapshotSequence: 7, projection: BASE_PROJECTION }, + } + : { _tag: "missing" as const }, + ), + ), + }); + yield* Queue.take(attempts); + created = true; + yield* TestClock.adjust("500 millis"); + yield* Queue.take(attempts); + yield* awaitThreadState(harness.observed, (state) => state.status === "live"); + yield* Queue.offer(harness.inputs, titleUpdated("Response arrived", 8)); + const state = yield* awaitThreadState( + harness.observed, + (value) => Option.getOrNull(value.data)?.thread.title === "Response arrived", + ); + expect(state.status).toBe("live"); + expect(yield* Ref.get(harness.removedThreads)).toEqual([]); + }), + ); + + it.effect("bounds missing snapshot retries and recovers on foreground", () => + Effect.gen(function* () { + const attempts = yield* Queue.unbounded(); + let created = false; + const harness = yield* makeHarness({ + loadSnapshot: () => + Queue.offer(attempts, undefined).pipe( + Effect.as( + created + ? { + _tag: "present" as const, + snapshot: { snapshotSequence: 7, projection: BASE_PROJECTION }, + } + : { _tag: "missing" as const }, + ), + ), + }); + yield* Queue.take(attempts); + for (const delay of [500, 1_000, 2_000, 4_000, 8_000, 16_000]) { + yield* TestClock.adjust(delay); + yield* Queue.take(attempts); + } + const missing = yield* awaitThreadState(harness.observed, (value) => + Option.isSome(value.error), + ); + expect(missing.status).not.toBe("deleted"); + yield* TestClock.adjust("1 minute"); + expect(yield* Ref.get(harness.loaderCalls)).toBe(7); + expect(yield* Ref.get(harness.subscriptionCount)).toBe(0); + created = true; + yield* Queue.offer(harness.wakeups, "application-active"); + yield* awaitThreadState(harness.observed, (value) => value.status === "live"); + expect(yield* Ref.get(harness.loaderCalls)).toBe(8); + expect(yield* Ref.get(harness.removedThreads)).toEqual([]); + }), + ); + it.effect("retains a deletion instead of restoring the old disk snapshot", () => Effect.gen(function* () { const resumeCache: NonNullable[1]> = { diff --git a/packages/client-runtime/src/state/threads.ts b/packages/client-runtime/src/state/threads.ts index 3bfa1ffbb4096..f226a5aa3f8d0 100644 --- a/packages/client-runtime/src/state/threads.ts +++ b/packages/client-runtime/src/state/threads.ts @@ -874,7 +874,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make ORCHESTRATION_V2_WS_METHODS.subscribeThread, Effect.fn("EnvironmentThreadState.makeSubscribeInput")(function* (session) { let current = yield* SubscriptionRef.get(state); - // A prior definitive miss (or delete event) already cleared this thread. + // An explicit delete event already cleared this thread. // Park the subscription attempt without opening the socket so we do not // retry forever against a known-missing id. if (current.status === "deleted") { @@ -904,8 +904,14 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make }), ), ); - const httpResult: ThreadSnapshotLoader.ThreadSnapshotLoadResult = + let httpResult: ThreadSnapshotLoader.ThreadSnapshotLoadResult = yield* snapshotLoader.load(prepared, threadId); + // Navigation can precede thread creation; an HTTP miss is not a deletion event. + for (const delay of [500, 1_000, 2_000, 4_000, 8_000, 16_000]) { + if (httpResult._tag !== "missing") break; + yield* Effect.sleep(delay); + httpResult = yield* snapshotLoader.load(prepared, threadId); + } switch (httpResult._tag) { case "present": { if (canLoadHistory && httpResult.history !== undefined) { @@ -936,7 +942,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make break; } case "missing": { - yield* setDeleted(); + yield* setStreamError("Thread not found. Reopen the thread to try again."); return yield* Effect.never; } case "unavailable": { From ceae507ec8daacbfbde0823915c65a8b63eb162c Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Sun, 1 Nov 2026 02:54:29 +0000 Subject: [PATCH 2/2] test(client): await snapshot retries before remounting --- .../client-runtime/src/state/threads-sync.test.ts | 13 +++++++++++-- 1 file changed, 11 insertions(+), 2 deletions(-) diff --git a/packages/client-runtime/src/state/threads-sync.test.ts b/packages/client-runtime/src/state/threads-sync.test.ts index edefcbfe307c6..5d4d35bb5c0a0 100644 --- a/packages/client-runtime/src/state/threads-sync.test.ts +++ b/packages/client-runtime/src/state/threads-sync.test.ts @@ -755,8 +755,17 @@ describe("EnvironmentThreads", () => { }; yield* Effect.scoped( Effect.gen(function* () { - const first = yield* makeHarness({ httpSnapshot: { _tag: "missing" }, resumeCache }); - yield* TestClock.adjust("31500 millis"); + const attempts = yield* Queue.unbounded(); + const first = yield* makeHarness({ + resumeCache, + loadSnapshot: () => + Queue.offer(attempts, undefined).pipe(Effect.as({ _tag: "missing" as const })), + }); + yield* Queue.take(attempts); + for (const delay of [500, 1_000, 2_000, 4_000, 8_000, 16_000]) { + yield* TestClock.adjust(delay); + yield* Queue.take(attempts); + } yield* awaitThreadState(first.observed, (value) => Option.isSome(value.error)); }), );