diff --git a/packages/client-runtime/src/rpc/client.ts b/packages/client-runtime/src/rpc/client.ts index dd435d820d64..8ae3ea79c56f 100644 --- a/packages/client-runtime/src/rpc/client.ts +++ b/packages/client-runtime/src/rpc/client.ts @@ -114,6 +114,11 @@ export const isAuthorizationFailure = (cause: Cause.Cause) => 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 = @@ -378,12 +383,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-sync.test.ts b/packages/client-runtime/src/state/threads-sync.test.ts index ae372ea938cd..d3677a6fc72b 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. { @@ -869,7 +893,7 @@ describe("EnvironmentThreads", () => { let configReads = 0; const harness = yield* makeHarness({ httpSnapshot: { _tag: "missing" }, - // The retry's config read dies too, so a stale flag would resubscribe forever. + // The retry's config read dies too, so a stale flag would resubscribe without backing off. initialConfig: Effect.suspend(() => ++configReads === 1 ? Effect.succeed({ threadSnapshotPagination: true }) @@ -2382,6 +2406,43 @@ 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, TEST_DEFECT); + 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 }); diff --git a/packages/client-runtime/src/state/threads.ts b/packages/client-runtime/src/state/threads.ts index b67245e735cc..6bbb9f32ee69 100644 --- a/packages/client-runtime/src/state/threads.ts +++ b/packages/client-runtime/src/state/threads.ts @@ -32,7 +32,12 @@ 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 { isAuthorizationFailure, request, subscribeDynamic } from "../rpc/client.ts"; +import { + isAuthorizationFailure, + 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 +60,8 @@ import { type EnvironmentThreadStatus, } from "./threadState.ts"; +const THREAD_SUBSCRIPTION_RETRY_DELAY = "250 millis"; + function statusWithoutLiveData( data: Option.Option, ): EnvironmentThreadStatus { @@ -967,8 +974,8 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make (yield* Ref.get(socketOnlyOpen)) && Option.isNone((yield* SubscriptionRef.get(state)).data) ); }); - // A defect parks the subscription, so a socket-only open resubscribes once. - const httpRetries = yield* Queue.unbounded(); + // A stream defect parks the subscription until something resubscribes it. + const resubscribes = yield* Queue.unbounded(); const markSynchronizing = Effect.gen(function* () { if (yield* Ref.get(resumingLive)) return; yield* SubscriptionRef.update(state, (current) => @@ -979,15 +986,27 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make }); yield* markSynchronizing; - yield* Effect.forkScoped( + // 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); + defectRetries += 1; + return setStreamError("Could not synchronize the thread.").pipe( + Effect.andThen(Effect.sleep(retryDelay)), + ); + }); + const followThread = (): Effect.Effect< + void, + never, + EnvironmentSupervisor.EnvironmentSupervisor + > => subscribeDynamic( 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; } @@ -1024,6 +1043,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); } @@ -1052,9 +1072,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": { @@ -1085,22 +1109,49 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make { // A socket-only open that fails retries over HTTP, which tells a // missing thread apart from a failure, before showing an error. - onDefect: Effect.fn("EnvironmentThreadState.onDefect")(function* () { + onDefect: Effect.fn("EnvironmentThreadState.onDefect")(function* (cause) { if (yield* failedSocketOnlyOpen) { yield* Ref.set(socketOnlyOpen, false); - return yield* Queue.offer(httpRetries, undefined); + return yield* Queue.offer(resubscribes, undefined); + } + if (!Cause.hasDies(cause)) { + return yield* setStreamError("Could not synchronize the thread."); } - yield* setStreamError("Could not synchronize the thread."); + yield* backOffAfterDefect; + yield* Queue.offer(resubscribes, undefined); }), onExpectedFailure: Effect.fn("EnvironmentThreadState.onExpectedFailure")(function* (cause) { const retryOverHttp = !isAuthorizationFailure(cause) && (yield* failedSocketOnlyOpen); if (!retryOverHttp) yield* setStreamError(formatThreadError(cause)); }), - retryExpectedFailureAfter: "250 millis", - resubscribe: Stream.merge(foregroundResubscriptions, Stream.fromQueue(httpRetries)), + retryExpectedFailureAfter: THREAD_SUBSCRIPTION_RETRY_DELAY, + resubscribe: Stream.merge(foregroundResubscriptions, Stream.fromQueue(resubscribes)), }, - ).pipe(Stream.runForEachArray(applyItems)), - ); + ).pipe( + Stream.runForEachArray((items) => + applyItems(items).pipe( + Effect.andThen( + Effect.sync(() => { + defectRetries = 0; + }), + ), + ), + ), + // A defect while applying items ends the stream; resubscribe for a fresh snapshot. + Effect.catchCause((cause) => + Cause.hasInterrupts(cause) + ? Effect.interrupt + : Effect.logWarning("Thread subscription failed; resubscribing.", { + cause: Cause.pretty(cause), + environmentId, + threadId, + }).pipe( + Effect.andThen(backOffAfterDefect), + Effect.andThen(Effect.suspend(followThread)), + ), + ), + ); + yield* Effect.forkScoped(followThread()); return state; });