From 6e353ff68511580219bf0e33f5c5eb74f1d5370a Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 27 Oct 2026 11:13:26 +0000 Subject: [PATCH 1/3] test(client-runtime): reproduce #23041 thread subscription never recovers after a defect A snapshot decode defect ends the thread's subscribeThread fiber; the state stays empty with "Could not synchronize the thread." and no new subscription is issued within the same scope and session. Co-Authored-By: Claude Opus 5.5 --- .../src/state/threads-sync.test.ts | 49 ++++++++++++++++++- 1 file changed, 47 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 3c1c153621e2..fe08d6094485 100644 --- a/packages/client-runtime/src/state/threads-sync.test.ts +++ b/packages/client-runtime/src/state/threads-sync.test.ts @@ -13,6 +13,7 @@ import { type OrchestrationV2TurnItem, } from "@t3tools/contracts"; import { describe, expect, it } from "@effect/vitest"; +import * as Cause from "effect/Cause"; import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; @@ -78,7 +79,7 @@ const BASE_PROJECTION: OrchestrationV2ThreadProjection = { thread: { ...v2Projection.thread, title: "Cached thread" }, }; -type TestThreadInput = OrchestrationV2ThreadStreamItem | Error; +type TestThreadInput = OrchestrationV2ThreadStreamItem | Error | Cause.Cause; function testSession( client: WsRpcProtocolClient, @@ -147,7 +148,11 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o const streamFrom = (queue: Queue.Queue) => Stream.fromQueue(queue).pipe( Stream.mapEffect((input) => - input instanceof Error ? Effect.fail(input) : Effect.succeed(input), + Cause.isCause(input) + ? Effect.failCause(input) + : input instanceof Error + ? Effect.fail(input) + : Effect.succeed(input), ), ); const client = { @@ -1803,6 +1808,46 @@ describe("EnvironmentThreads", () => { }), ); + it.effect( + "resubscribes after a snapshot decode defect ends the thread subscription (#23041)", + () => + Effect.gen(function* () { + const harness = yield* makeHarness({ httpSnapshot: { _tag: "unavailable" } }); + yield* SubscriptionRef.changes(harness.subscriptionCount).pipe( + Stream.filter((count) => count >= 1), + Stream.runHead, + ); + yield* Queue.offer( + harness.inputs, + Cause.die(new Error("Could not decode thread snapshot")), + ); + const failed = yield* awaitThreadState(harness.observed, (value) => + Option.isSome(value.error), + ); + expect(failed.status).toBe("empty"); + expect(Option.getOrThrow(failed.error)).toBe("Could not synchronize the thread."); + + yield* TestClock.adjust("60 seconds"); + expect(yield* SubscriptionRef.get(harness.subscriptionCount)).toBe(2); + + yield* Queue.offer( + harness.inputs, + snapshot({ + ...BASE_PROJECTION, + thread: { ...BASE_PROJECTION.thread, title: "Recovered thread" }, + }), + ); + const recovered = yield* awaitThreadState( + harness.observed, + (value) => + value.status === "live" && + Option.isSome(value.data) && + value.data.value.thread.title === "Recovered thread", + ); + expect(Option.isNone(recovered.error)).toBe(true); + }), + ); + it.effect("keeps replayed updates synchronizing until the completion marker arrives", () => Effect.gen(function* () { const harness = yield* makeHarness({ cached: BASE_PROJECTION, completionMarker: true }); From 16e2e5c6561e8f1892a2322628c9258d99bd384d Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 27 Oct 2026 11:19:11 +0000 Subject: [PATCH 2/3] fix(client-runtime): resubscribe a thread after its subscription dies A defect on the thread detail subscription (for example one socket snapshot that fails to decode) ended the forked fiber, leaving the thread empty with "Could not synchronize the thread." until the user reselected it. The thread state now resubscribes in the same scope after a defect, with the same capped exponential backoff expected failures use (250 ms doubling to 30 s), reset once a batch applies. Interruptions are not retried, and a defect from applying items takes the same path so a dead subscription never sits behind a live status. Fixes pingdotgg/t3code#23041 Co-Authored-By: Claude Opus 5.5 --- packages/client-runtime/src/rpc/client.ts | 14 ++++--- packages/client-runtime/src/state/threads.ts | 43 +++++++++++++++++--- 2 files changed, 46 insertions(+), 11 deletions(-) diff --git a/packages/client-runtime/src/rpc/client.ts b/packages/client-runtime/src/rpc/client.ts index c364b3448d03..743e2b5f4816 100644 --- a/packages/client-runtime/src/rpc/client.ts +++ b/packages/client-runtime/src/rpc/client.ts @@ -108,6 +108,11 @@ const isEnvironmentAuthorizationError = Schema.is(EnvironmentAuthorizationError) const MAX_EXPECTED_FAILURE_RETRY_DELAY_MS = 30_000; +export const subscriptionRetryDelay = (base: Duration.Input, retries: number) => + Duration.millis( + Math.min(Duration.toMillis(base) * 2 ** retries, MAX_EXPECTED_FAILURE_RETRY_DELAY_MS), + ); + export type EnvironmentRpcInput = Parameters>[0]; export type EnvironmentRpcSuccess = @@ -358,12 +363,9 @@ function subscribeDynamicMapped( ) { return handled; } - const retryDelay = Duration.millis( - Math.min( - Duration.toMillis(options.retryExpectedFailureAfter) * - 2 ** expectedFailureRetries, - MAX_EXPECTED_FAILURE_RETRY_DELAY_MS, - ), + const retryDelay = subscriptionRetryDelay( + options.retryExpectedFailureAfter, + expectedFailureRetries, ); expectedFailureRetries += 1; return handled.pipe( diff --git a/packages/client-runtime/src/state/threads.ts b/packages/client-runtime/src/state/threads.ts index 693af5329fe1..a53cca16134b 100644 --- a/packages/client-runtime/src/state/threads.ts +++ b/packages/client-runtime/src/state/threads.ts @@ -32,7 +32,7 @@ import * as ConnectionWakeups from "../connection/wakeups.ts"; import * as Persistence from "../platform/persistence.ts"; import { runCachePersistence } from "./cachePersistence.ts"; import * as ManagedRelay from "../relay/managedRelay.ts"; -import { request, subscribeDynamic } from "../rpc/client.ts"; +import { request, subscribeDynamic, subscriptionRetryDelay } from "../rpc/client.ts"; import { parseThreadKey, threadKey } from "./entities.ts"; import { applyOrchestrationV2ProjectionEvent } from "./orchestrationV2Projection.ts"; import { THREAD_SNAPSHOT_IDLE_TTL_MS } from "./threadRetention.ts"; @@ -55,6 +55,8 @@ import { type EnvironmentThreadStatus, } from "./threadState.ts"; +const THREAD_SUBSCRIPTION_RETRY_DELAY = "250 millis"; + function statusWithoutLiveData( data: Option.Option, ): EnvironmentThreadStatus { @@ -878,7 +880,12 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make }); yield* markSynchronizing; - yield* Effect.forkScoped( + let defectRetries = 0; + const followThread = (): Effect.Effect< + void, + never, + EnvironmentSupervisor.EnvironmentSupervisor + > => subscribeDynamic( ORCHESTRATION_V2_WS_METHODS.subscribeThread, Effect.fn("EnvironmentThreadState.makeSubscribeInput")(function* (session) { @@ -982,11 +989,37 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make { onDefect: () => setStreamError("Could not synchronize the thread."), onExpectedFailure: (cause) => setStreamError(formatThreadError(cause)), - retryExpectedFailureAfter: "250 millis", + retryExpectedFailureAfter: THREAD_SUBSCRIPTION_RETRY_DELAY, resubscribe: foregroundResubscriptions, }, - ).pipe(Stream.runForEachArray(applyItems)), - ); + ).pipe( + Stream.runForEachArray((items) => + applyItems(items).pipe( + Effect.andThen( + Effect.sync(() => { + defectRetries = 0; + }), + ), + ), + ), + // A defect (an undecodable snapshot, or one applyItems cannot handle) ends the + // subscription for good; resubscribe the way reselecting the thread would. + Effect.catchCause((cause) => { + if (Cause.hasInterrupts(cause)) return Effect.interrupt; + const retryDelay = subscriptionRetryDelay(THREAD_SUBSCRIPTION_RETRY_DELAY, defectRetries); + defectRetries += 1; + return Effect.logWarning("Thread subscription failed; resubscribing.", { + cause: Cause.pretty(cause), + environmentId, + threadId, + }).pipe( + Effect.andThen(setStreamError("Could not synchronize the thread.")), + Effect.andThen(Effect.sleep(retryDelay)), + Effect.andThen(Effect.suspend(followThread)), + ); + }), + ); + yield* Effect.forkScoped(followThread()); return state; }); From b8323327d019e34fecfd796834f9afc1d33c4658 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 11 Nov 2026 10:33:19 +0000 Subject: [PATCH 3/3] fix(client): a mounted thread keeps asking after an early 404 until it exists A thread detail load could get thread_not_found before the thread was created. The miss marked the thread deleted and parked the subscription, so the open thread stayed blank until the user navigated away and back. A miss now asks HTTP again with the existing capped backoff while the thread stays mounted; a delete event still parks it for good. AI-assisted; verified with the tests and tools listed above. --- .../src/state/threads-sync.test.ts | 32 ++++++++++++++++--- packages/client-runtime/src/state/threads.ts | 15 ++++++--- 2 files changed, 38 insertions(+), 9 deletions(-) diff --git a/packages/client-runtime/src/state/threads-sync.test.ts b/packages/client-runtime/src/state/threads-sync.test.ts index 750f4b18c82a..5eea69881e86 100644 --- a/packages/client-runtime/src/state/threads-sync.test.ts +++ b/packages/client-runtime/src/state/threads-sync.test.ts @@ -118,6 +118,8 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o readonly latestLocalTurnOrdinal?: number | null; }; readonly httpSnapshot?: ThreadSnapshotLoadResult; + /** Answers per load call; the last one repeats. Overrides `httpSnapshot`. */ + readonly httpSnapshots?: readonly ThreadSnapshotLoadResult[]; /** The test sends the completion marker itself instead of the fake server sending it first. */ readonly completionMarker?: boolean; readonly resumeCache?: NonNullable[1]>; @@ -193,11 +195,12 @@ 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( + Ref.updateAndGet(loaderCalls, (count) => count + 1).pipe( + Effect.map((count) => threadId === THREAD_ID - ? (options?.httpSnapshot ?? - ({ _tag: "unavailable" } satisfies ThreadSnapshotLoadResult)) + ? (options?.httpSnapshots?.[Math.min(count, options.httpSnapshots.length) - 1] ?? + options?.httpSnapshot ?? + ({ _tag: "unavailable" } satisfies ThreadSnapshotLoadResult)) : ({ _tag: "unavailable" } satisfies ThreadSnapshotLoadResult), ), ), @@ -798,6 +801,27 @@ describe("EnvironmentThreads", () => { }), ); + it.effect("keeps asking a mounted thread over HTTP after a miss until it exists", () => + Effect.gen(function* () { + const harness = yield* makeHarness({ + httpSnapshots: [ + { _tag: "missing" }, + { _tag: "missing" }, + { + _tag: "present", + snapshot: { snapshotSequence: CACHED_SNAPSHOT_SEQUENCE, projection: BASE_PROJECTION }, + }, + ], + }); + yield* awaitThreadState(harness.observed, (value) => value.status === "deleted"); + yield* TestClock.adjust("250 millis"); + yield* TestClock.adjust("500 millis"); + const state = yield* awaitThreadState(harness.observed, (value) => value.status === "live"); + expect(Option.getOrThrow(state.data)).toEqual(BASE_PROJECTION); + expect(yield* Ref.get(harness.loaderCalls)).toBe(3); + }), + ); + it.effect.each([ // The HTTP retry decides, so a missing thread never flashes an error. { diff --git a/packages/client-runtime/src/state/threads.ts b/packages/client-runtime/src/state/threads.ts index 2ff20827132a..359befa33d0f 100644 --- a/packages/client-runtime/src/state/threads.ts +++ b/packages/client-runtime/src/state/threads.ts @@ -977,6 +977,8 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make }); yield* markSynchronizing; + // A 404 can arrive before the thread exists, so a miss asks HTTP again while mounted. + let httpMisses = 0; let defectRetries = 0; const backOffAfterDefect = Effect.suspend(() => { const retryDelay = subscriptionRetryDelay(THREAD_SUBSCRIPTION_RETRY_DELAY, defectRetries); @@ -994,10 +996,8 @@ 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. - // Park the subscription attempt without opening the socket so we do not - // retry forever against a known-missing id. - if (current.status === "deleted") { + // A delete event already cleared this thread. Park without opening the socket. + if (current.status === "deleted" && httpMisses === 0) { return yield* Effect.never; } @@ -1034,6 +1034,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make yield* snapshotLoader.load(prepared, threadId); switch (httpResult._tag) { case "present": { + httpMisses = 0; if (canLoadHistory && httpResult.history !== undefined) { yield* Ref.set(acceptsBoundedSocketSnapshots, true); } @@ -1062,9 +1063,13 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make break; } case "missing": { - yield* setDeleted(); + if (current.status !== "deleted") yield* setDeleted(); // A 404 also means "not created yet". Only a delete event may outlive this mount. if (resumeCache?.owner === owner) resumeCache.snapshot = undefined; + yield* Effect.sleep( + subscriptionRetryDelay(THREAD_SUBSCRIPTION_RETRY_DELAY, httpMisses++), + ); + yield* Queue.offer(resubscribes, undefined); return yield* Effect.never; } case "unavailable": {