From cc36b743f40fb69fc8eed782d2b3109d9e8f5fdd Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 24 Oct 2026 10:30:38 +0000 Subject: [PATCH 1/4] test(client-runtime): reproduce issue 4589 durable subscription stuck after transport error A durable subscription whose stream fails with an RpcClientError on a still-healthy session drains forever and never asks the supervisor for a replacement session. This test is expected to fail until that is fixed. Co-Authored-By: Claude Opus 5.5 --- .../client-runtime/src/rpc/client.test.ts | 53 +++++++++++++++++++ 1 file changed, 53 insertions(+) diff --git a/packages/client-runtime/src/rpc/client.test.ts b/packages/client-runtime/src/rpc/client.test.ts index bbd86f2f79112..2dad4dfcbfbe7 100644 --- a/packages/client-runtime/src/rpc/client.test.ts +++ b/packages/client-runtime/src/rpc/client.test.ts @@ -15,6 +15,7 @@ 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 { RpcClientError } from "effect/rpc"; import { @@ -265,6 +266,58 @@ describe("environment RPC", () => { }), ); + it.effect( + "asks the supervisor for a new session when a durable subscription loses its transport", + () => + Effect.gen(function* () { + const subscribed = yield* Deferred.make(); + const recoveryRequested = yield* Deferred.make(); + // The session itself stays healthy (`closed` never resolves), as when the RPC client fails + // to decode one stream chunk, so nothing else will replace it. + const client = { + [WS_METHODS.subscribeTerminalEvents]: () => + Stream.fromEffect(Deferred.succeed(subscribed, undefined)).pipe( + Stream.drain, + Stream.concat( + Stream.fail( + new RpcClientError.RpcClientError({ + reason: new RpcClientError.RpcClientDefect({ + message: "Failed to decode stream chunk", + cause: new Error("invalid stream item"), + }), + }), + ), + ), + ), + } as unknown as WsRpcProtocolClient; + const harness = yield* makeHarness(); + const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({ + ...harness.supervisor, + retryNow: Deferred.succeed(recoveryRequested, undefined).pipe(Effect.asVoid), + }); + + yield* SubscriptionRef.set(harness.activeSession, Option.some(session(client))); + const subscriptionFiber = yield* subscribe(WS_METHODS.subscribeTerminalEvents, {}).pipe( + Stream.runDrain, + Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), + Effect.forkChild, + ); + yield* Deferred.await(subscribed); + const recoveryFiber = yield* Deferred.await(recoveryRequested).pipe( + Effect.timeoutOption("1 second"), + Effect.forkChild, + ); + yield* TestClock.adjust("1 second"); + const recovery = yield* Fiber.join(recoveryFiber); + yield* Fiber.interrupt(subscriptionFiber); + + expect( + Option.isSome(recovery), + "durable subscription drained after a transport RpcClientError without requesting a replacement session", + ).toBe(true); + }), + ); + it.effect("surfaces domain subscription failures without reconnecting", () => Effect.gen(function* () { const domainError = new Error("terminal subscription rejected"); From 383a473f2e5f6c10d81ec297589b3f36e18df3b4 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 24 Oct 2026 10:36:17 +0000 Subject: [PATCH 2/4] fix(client-runtime): resubscribe durable streams that end on a healthy session A durable subscription whose stream ended with a transport RpcClientError drained forever, and one whose chunk failed to decode (RpcClient dies with a SchemaError) failed the whole subscription. Both end only that stream while the socket stays open, so the supervisor never replaces the session and, for example, the shell stopped updating until reload. Both now resubscribe to the same session with the existing bounded exponential backoff. A replaced session still interrupts the pending retry through switchMap. Co-Authored-By: Claude Opus 5.5 --- .../client-runtime/src/rpc/client.test.ts | 131 +++++++++++------- packages/client-runtime/src/rpc/client.ts | 50 +++++-- 2 files changed, 116 insertions(+), 65 deletions(-) diff --git a/packages/client-runtime/src/rpc/client.test.ts b/packages/client-runtime/src/rpc/client.test.ts index 2dad4dfcbfbe7..cbc5e860c09c6 100644 --- a/packages/client-runtime/src/rpc/client.test.ts +++ b/packages/client-runtime/src/rpc/client.test.ts @@ -13,6 +13,7 @@ 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 Schema from "effect/Schema"; import * as Stream from "effect/Stream"; import * as SubscriptionRef from "effect/SubscriptionRef"; import * as TestClock from "effect/testing/TestClock"; @@ -45,6 +46,8 @@ const INSTALL_DOWNLOADING: RelayClientInstallProgressEvent = { stage: "downloading", }; +const decodeNumber = Schema.decodeUnknownEffect(Schema.Number); + function session(client: WsRpcProtocolClient): RpcSession.RpcSession { return { client, @@ -259,6 +262,8 @@ describe("environment RPC", () => { for (let attempt = 0; attempt < 100 && subscriptions.length < 2; attempt += 1) { yield* Effect.yieldNow; } + // The pending same-session retry must not resubscribe the replaced session. + yield* TestClock.adjust("1 minute"); yield* Fiber.interrupt(subscriptionFiber); expect(subscriptions).toEqual(["first", "second"]); @@ -266,56 +271,82 @@ describe("environment RPC", () => { }), ); - it.effect( - "asks the supervisor for a new session when a durable subscription loses its transport", - () => - Effect.gen(function* () { - const subscribed = yield* Deferred.make(); - const recoveryRequested = yield* Deferred.make(); - // The session itself stays healthy (`closed` never resolves), as when the RPC client fails - // to decode one stream chunk, so nothing else will replace it. - const client = { - [WS_METHODS.subscribeTerminalEvents]: () => - Stream.fromEffect(Deferred.succeed(subscribed, undefined)).pipe( - Stream.drain, - Stream.concat( - Stream.fail( - new RpcClientError.RpcClientError({ - reason: new RpcClientError.RpcClientDefect({ - message: "Failed to decode stream chunk", - cause: new Error("invalid stream item"), - }), - }), - ), - ), - ), - } as unknown as WsRpcProtocolClient; - const harness = yield* makeHarness(); - const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({ - ...harness.supervisor, - retryNow: Deferred.succeed(recoveryRequested, undefined).pipe(Effect.asVoid), - }); - - yield* SubscriptionRef.set(harness.activeSession, Option.some(session(client))); - const subscriptionFiber = yield* subscribe(WS_METHODS.subscribeTerminalEvents, {}).pipe( - Stream.runDrain, - Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), - Effect.forkChild, - ); - yield* Deferred.await(subscribed); - const recoveryFiber = yield* Deferred.await(recoveryRequested).pipe( - Effect.timeoutOption("1 second"), - Effect.forkChild, - ); - yield* TestClock.adjust("1 second"); - const recovery = yield* Fiber.join(recoveryFiber); - yield* Fiber.interrupt(subscriptionFiber); - - expect( - Option.isSome(recovery), - "durable subscription drained after a transport RpcClientError without requesting a replacement session", - ).toBe(true); - }), + // The session stays healthy (`closed` never resolves), as when one stream chunk fails to decode, + // so the supervisor will not replace it and only resubscribing restores delivery. + const resumesOnTheSameSession = Effect.fn("TestEnvironmentRpc.resumesOnTheSameSession")( + function* (streamFailure: Stream.Stream) { + const resumedValue = { source: "resumed" } as unknown as ServerLifecycleStreamEvent; + const firstSubscribed = yield* Deferred.make(); + let subscriptions = 0; + const observedDefects: unknown[] = []; + const client = { + [WS_METHODS.subscribeServerLifecycle]: () => { + subscriptions += 1; + return subscriptions === 1 + ? Stream.fromEffect(Deferred.succeed(firstSubscribed, undefined)).pipe( + Stream.drain, + Stream.concat(streamFailure), + ) + : Stream.make(resumedValue).pipe(Stream.concat(Stream.never)); + }, + } as unknown as WsRpcProtocolClient; + const { activeSession, retryCount, supervisor } = yield* makeHarness(); + + yield* SubscriptionRef.set(activeSession, Option.some(session(client))); + const resultFiber = yield* subscribe( + WS_METHODS.subscribeServerLifecycle, + {}, + { + onDefect: (cause) => + Effect.sync(() => { + observedDefects.push(Cause.squash(cause)); + }), + }, + ).pipe( + Stream.runHead, + Effect.timeoutOption("2 seconds"), + Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), + Effect.forkChild, + ); + yield* Deferred.await(firstSubscribed); + yield* TestClock.adjust("2 seconds"); + + expect( + yield* Fiber.join(resultFiber), + "durable subscription stopped delivering after its stream ended on a healthy session", + ).toEqual(Option.some(Option.some(resumedValue))); + expect(subscriptions).toBe(2); + expect(yield* Ref.get(retryCount)).toBe(0); + return observedDefects; + }, + ); + + it.effect("resubscribes on the same session after a transport failure", () => + Effect.gen(function* () { + yield* resumesOnTheSameSession( + Stream.fail( + new RpcClientError.RpcClientError({ + reason: new RpcClientError.RpcClientDefect({ + message: "stream failed", + cause: new Error("stream failed"), + }), + }), + ), + ); + }), + ); + + it.effect("resubscribes on the same session after a stream chunk fails to decode", () => + Effect.gen(function* () { + // Mirrors RpcClient, which decodes each chunk with `Effect.orDie`. + const decodeDefect = decodeNumber("not a number").pipe(Effect.orDie); + const observedDefects = yield* resumesOnTheSameSession( + Stream.fromEffect(decodeDefect).pipe(Stream.drain), + ); + + expect(observedDefects).toHaveLength(1); + expect(Schema.isSchemaError(observedDefects[0])).toBe(true); + }), ); it.effect("surfaces domain subscription failures without reconnecting", () => diff --git a/packages/client-runtime/src/rpc/client.ts b/packages/client-runtime/src/rpc/client.ts index c364b3448d03d..c41476d3aae53 100644 --- a/packages/client-runtime/src/rpc/client.ts +++ b/packages/client-runtime/src/rpc/client.ts @@ -107,6 +107,7 @@ export const isRpcClientError = Schema.is(RpcClientError.RpcClientError); const isEnvironmentAuthorizationError = Schema.is(EnvironmentAuthorizationError); const MAX_EXPECTED_FAILURE_RETRY_DELAY_MS = 30_000; +const STREAM_FAILURE_RETRY_DELAY = Duration.seconds(1); export type EnvironmentRpcInput = Parameters>[0]; @@ -324,6 +325,19 @@ function subscribeDynamicMapped( : Effect.void, ), Stream.catchCause((cause) => { + const retryAfter = (base: Duration.Input) => { + const retryDelay = Duration.millis( + Math.min( + Duration.toMillis(base) * 2 ** expectedFailureRetries, + MAX_EXPECTED_FAILURE_RETRY_DELAY_MS, + ), + ); + expectedFailureRetries += 1; + return Stream.fromEffect(Effect.sleep(retryDelay)).pipe( + Stream.drain, + Stream.concat(subscribeToSession()), + ); + }; const hasOnlyExpectedFailures = cause.reasons.length > 0 && cause.reasons.every((reason) => reason._tag === "Fail"); @@ -332,17 +346,34 @@ function subscribeDynamicMapped( cause.reasons.every( (reason) => reason._tag === "Fail" && isRpcClientError(reason.error), ); - if (isTransportFailure) { + // The RPC client dies with a SchemaError when one stream chunk fails to + // decode, ending only that request while the socket stays open. + const isChunkDecodeDefect = + cause.reasons.length > 0 && + cause.reasons.every( + (reason) => reason._tag === "Die" && Schema.isSchemaError(reason.defect), + ); + // Both end one stream while the session can stay healthy, so nothing else + // would replace it. A dead session is replaced by the supervisor, which + // interrupts this retry through switchMap. + if (isTransportFailure || isChunkDecodeDefect) { return Stream.fromEffect( Effect.logWarning( - "Durable RPC subscription lost its transport; waiting for the next session.", + "Durable RPC subscription stream ended; resubscribing.", { cause: Cause.pretty(cause), method: tag, environmentId: supervisor.target.environmentId, }, ), - ).pipe(Stream.drain); + ).pipe( + Stream.drain, + Stream.concat( + retryAfter( + options?.retryExpectedFailureAfter ?? STREAM_FAILURE_RETRY_DELAY, + ), + ), + ); } if (hasOnlyExpectedFailures && options?.onExpectedFailure !== undefined) { const handled = Stream.fromEffect(options.onExpectedFailure(cause)).pipe( @@ -358,19 +389,8 @@ function subscribeDynamicMapped( ) { return handled; } - const retryDelay = Duration.millis( - Math.min( - Duration.toMillis(options.retryExpectedFailureAfter) * - 2 ** expectedFailureRetries, - MAX_EXPECTED_FAILURE_RETRY_DELAY_MS, - ), - ); - expectedFailureRetries += 1; return handled.pipe( - Stream.concat( - Stream.fromEffect(Effect.sleep(retryDelay)).pipe(Stream.drain), - ), - Stream.concat(subscribeToSession()), + Stream.concat(retryAfter(options.retryExpectedFailureAfter)), ); } return Stream.failCause(cause); From 2b15e0ca0de61c384dc17fda8f69479c58e97a0c Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 24 Oct 2026 11:21:17 +0000 Subject: [PATCH 3/4] test(client-runtime): reproduce finite decode retries and reset stream backoff Two regressions from resubscribing durable streams on a healthy session: - A finite searchThreadStream subscription whose chunk fails to decode now resubscribes on the same session instead of failing its consumer, so the progressive thread find never settles. Before that change the SchemaError defect reached the atom. - A shell-like stream that emits a valid value (metadata) and then fails to decode on every attempt retries every 250ms forever, because the first value resets the retry counter. Delays should grow to the 30s cap on one session and restart from the base delay on a replacement session. Co-Authored-By: Claude Opus 5.5 --- .../client-runtime/src/rpc/client.test.ts | 72 ++++++++++++ .../client-runtime/src/state/runtime.test.ts | 105 ++++++++++++++++++ 2 files changed, 177 insertions(+) diff --git a/packages/client-runtime/src/rpc/client.test.ts b/packages/client-runtime/src/rpc/client.test.ts index cbc5e860c09c6..aaccb42b47c66 100644 --- a/packages/client-runtime/src/rpc/client.test.ts +++ b/packages/client-runtime/src/rpc/client.test.ts @@ -6,6 +6,7 @@ import { } from "@t3tools/contracts"; import { describe, expect, it } from "@effect/vitest"; import * as Cause from "effect/Cause"; +import * as Clock from "effect/Clock"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; @@ -349,6 +350,77 @@ describe("environment RPC", () => { }), ); + it.effect( + "backs off a same-session stream that keeps failing to decode after its first value", + () => + Effect.gen(function* () { + // Like the shell stream: a valid value (metadata) arrives, then a chunk fails to decode. + const value = { source: "metadata" } as unknown as ServerLifecycleStreamEvent; + const decodeDefect = decodeNumber("not a number").pipe(Effect.orDie); + const subscriptions = yield* Queue.unbounded<{ + readonly session: "first" | "second"; + readonly at: number; + }>(); + let secondSubscriptions = 0; + const failingClient = (name: "first" | "second") => + ({ + [WS_METHODS.subscribeServerLifecycle]: () => + Stream.unwrap( + Effect.gen(function* () { + yield* Queue.offer(subscriptions, { + session: name, + at: yield* Clock.currentTimeMillis, + }); + if (name === "second" && ++secondSubscriptions > 2) return Stream.never; + return Stream.make(value).pipe( + Stream.concat(Stream.fromEffect(decodeDefect).pipe(Stream.drain)), + ); + }), + ), + }) as unknown as WsRpcProtocolClient; + const { activeSession, retryCount, supervisor } = yield* makeHarness(); + + yield* SubscriptionRef.set(activeSession, Option.some(session(failingClient("first")))); + const subscriptionFiber = yield* subscribe( + WS_METHODS.subscribeServerLifecycle, + {}, + { + onExpectedFailure: () => Effect.void, + retryExpectedFailureAfter: "250 millis", + }, + ).pipe( + Stream.runDrain, + Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), + Effect.forkChild, + ); + + const expectedFirstSessionDelays = [250, 500, 1_000, 2_000, 4_000, 8_000, 16_000, 30_000]; + const firstSessionDelays: number[] = []; + let previous = yield* Queue.take(subscriptions); + for (const delay of expectedFirstSessionDelays) { + yield* TestClock.adjust(delay); + const next = yield* Queue.take(subscriptions); + firstSessionDelays.push(next.at - previous.at); + previous = next; + } + + // A replacement session starts a fresh backoff. + yield* SubscriptionRef.set(activeSession, Option.some(session(failingClient("second")))); + const replaced = yield* Queue.take(subscriptions); + yield* TestClock.adjust(250); + const retried = yield* Queue.take(subscriptions); + yield* Fiber.interrupt(subscriptionFiber); + + expect( + firstSessionDelays, + "same-session decode retries reset to the base delay after every first value", + ).toEqual(expectedFirstSessionDelays); + expect([replaced.session, retried.session]).toEqual(["second", "second"]); + expect(retried.at - replaced.at).toBe(250); + expect(yield* Ref.get(retryCount)).toBe(0); + }), + ); + it.effect("surfaces domain subscription failures without reconnecting", () => Effect.gen(function* () { const domainError = new Error("terminal subscription rejected"); diff --git a/packages/client-runtime/src/state/runtime.test.ts b/packages/client-runtime/src/state/runtime.test.ts index b563bd8a04af8..3d6d498ca09f1 100644 --- a/packages/client-runtime/src/state/runtime.test.ts +++ b/packages/client-runtime/src/state/runtime.test.ts @@ -1,6 +1,7 @@ import { describe, expect, it } from "@effect/vitest"; import { EnvironmentId, + ORCHESTRATION_V2_WS_METHODS, type PreviewEvent, type TerminalAttachInput, type TerminalAttachStreamEvent, @@ -9,6 +10,7 @@ import { WS_METHODS, } from "@t3tools/contracts"; import * as Cause from "effect/Cause"; +import * as Clock from "effect/Clock"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; @@ -19,6 +21,7 @@ import * as Queue from "effect/Queue"; import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; import * as SubscriptionRef from "effect/SubscriptionRef"; +import * as TestClock from "effect/testing/TestClock"; import { AsyncResult, Atom, AtomRegistry } from "effect/reactivity"; import { @@ -32,6 +35,7 @@ import { import * as EnvironmentRegistry from "../connection/registry.ts"; import * as EnvironmentSupervisor from "../connection/supervisor.ts"; import { EnvironmentRpcUnavailableError } from "../rpc/client.ts"; +import type { WsRpcProtocolClient } from "../rpc/protocol.ts"; import type * as RpcSession from "../rpc/session.ts"; import { createPreviewEnvironmentAtoms } from "./preview.ts"; import type { QueryPersistence } from "./queryPersistence.ts"; @@ -39,6 +43,7 @@ import { createTerminalEnvironmentAtoms } from "./terminal.ts"; import { createAtomCommandScheduler, createEnvironmentQueryAtomFamily, + createEnvironmentRpcSubscriptionAtomFamily, selectEnvironmentQueryData, createEnvironmentSubscriptionAtomFamily, createRuntimeCommand, @@ -53,6 +58,8 @@ const QUERY_ENVIRONMENT = new PrimaryConnectionTarget({ wsBaseUrl: "wss://query.example.test", }); +const decodeNumber = Schema.decodeUnknownEffect(Schema.Number); + const QUERY_RPC_SESSION = {} as RpcSession.RpcSession; class TestQueryError extends Schema.TaggedError()("TestQueryError", { @@ -321,6 +328,104 @@ describe("finite environment subscription lifecycle", () => { }), ), ); + it.effect( + "surfaces a decode defect from a finite RPC subscription instead of resubscribing", + () => + Effect.scoped( + Effect.gen(function* () { + const firstSubscribed = yield* Deferred.make(); + const resubscribed = yield* Deferred.make(); + const surfaced = yield* Deferred.make>(); + let subscriptions = 0; + // Mirrors RpcClient, which decodes each chunk with `Effect.orDie`. + const decodeDefect = decodeNumber("not a number").pipe(Effect.orDie); + const client = { + [ORCHESTRATION_V2_WS_METHODS.searchThreadStream]: () => { + subscriptions += 1; + Deferred.doneUnsafe(subscriptions === 1 ? firstSubscribed : resubscribed, Exit.void); + return Stream.fromEffect(decodeDefect).pipe(Stream.drain); + }, + } as unknown as WsRpcProtocolClient; + const rpcSession = { + client, + initialConfig: Effect.never, + subscribeServerConfig: () => Stream.never, + ready: Effect.void, + probe: Effect.void, + closed: Effect.never, + } as unknown as RpcSession.RpcSession; + const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({ + target: QUERY_ENVIRONMENT, + state: yield* SubscriptionRef.make(queryConnectionState()), + session: yield* SubscriptionRef.make(Option.some(rpcSession)), + prepared: yield* SubscriptionRef.make>(Option.none()), + connect: Effect.void, + disconnect: Effect.void, + retryNow: Effect.void, + } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]); + const environmentRegistry = EnvironmentRegistry.EnvironmentRegistry.of({ + followStream: (_environmentId: EnvironmentId, stream: Stream.Stream) => + Stream.provideService( + stream, + EnvironmentSupervisor.EnvironmentSupervisor, + supervisor, + ), + } as unknown as EnvironmentRegistry.EnvironmentRegistry["Service"]); + // Atoms run their own fibers; share the test clock so a pending retry is driven by it. + const runtime = Atom.runtime( + Layer.mergeAll( + Layer.succeed(EnvironmentRegistry.EnvironmentRegistry, environmentRegistry), + Layer.succeed(Clock.Clock, yield* Clock.Clock), + ), + ); + // Same options as the progressive thread find in orchestration.ts. + const family = createEnvironmentRpcSubscriptionAtomFamily(runtime, { + label: "test.finite-rpc-subscription", + tag: ORCHESTRATION_V2_WS_METHODS.searchThreadStream, + completeWhen: (result) => result.complete !== false, + idleTtlMs: 0, + }); + const atom = family({ + environmentId: QUERY_ENVIRONMENT.environmentId, + input: {} as never, + }); + const registry = AtomRegistry.make(); + const unsubscribe = registry.subscribe( + atom, + (result) => { + if (AsyncResult.isFailure(result)) + Deferred.doneUnsafe(surfaced, Exit.succeed(result.cause)); + }, + { immediate: true }, + ); + yield* Effect.addFinalizer(() => + Effect.sync(() => { + unsubscribe(); + registry.dispose(); + }), + ); + yield* Deferred.await(firstSubscribed); + // Lets any same-session retry run, so its resubscription can win the race below. + yield* TestClock.adjust("1 minute"); + const outcome = yield* Effect.raceFirst( + Deferred.await(surfaced).pipe( + Effect.map((cause) => ({ _tag: "surfaced", cause }) as const), + ), + Deferred.await(resubscribed).pipe(Effect.as({ _tag: "resubscribed" } as const)), + ); + + expect( + outcome._tag, + "finite subscription resubscribed after a decode defect instead of failing its consumer", + ).toBe("surfaced"); + if (outcome._tag === "surfaced") { + expect(Cause.hasDies(outcome.cause)).toBe(true); + expect(Schema.isSchemaError(Cause.squash(outcome.cause))).toBe(true); + } + expect(subscriptions).toBe(1); + }), + ), + ); }); describe("environment query lifecycle", () => { From 86f313103db553087566fef071a1835a54c3ceb5 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 24 Oct 2026 11:22:49 +0000 Subject: [PATCH 4/4] fix(client-runtime): settle finite searches and back off failing durable streams searchThreadStream is a finite subscription, so it no longer resubscribes on the same session: a transport failure waits for the next session again and a chunk decode defect reaches its consumer, as before same-session recovery. Same-session transport and decode retries now use their own per-session counter. A first valid value no longer resets it, so a stream that emits and then fails every time backs off to the 30s cap instead of retrying every base delay. A replacement session starts a fresh counter. Expected domain failures keep their counter and its reset on the first value. Co-Authored-By: Claude Opus 5.5 --- packages/client-runtime/src/rpc/client.ts | 34 +++++++++++++++++++---- 1 file changed, 28 insertions(+), 6 deletions(-) diff --git a/packages/client-runtime/src/rpc/client.ts b/packages/client-runtime/src/rpc/client.ts index c41476d3aae53..c325fc0de3cf5 100644 --- a/packages/client-runtime/src/rpc/client.ts +++ b/packages/client-runtime/src/rpc/client.ts @@ -292,6 +292,8 @@ function subscribeDynamicMapped( EnvironmentRpcStreamFailure >; let expectedFailureRetries = 0; + // Not reset by a first value: a stream can emit before every chunk failure. + let streamFailureRetries = 0; const subscribeToSession = (): Stream.Stream> => Stream.suspend(() => Stream.unwrap( @@ -325,14 +327,13 @@ function subscribeDynamicMapped( : Effect.void, ), Stream.catchCause((cause) => { - const retryAfter = (base: Duration.Input) => { + const retryAfter = (base: Duration.Input, retries: number) => { const retryDelay = Duration.millis( Math.min( - Duration.toMillis(base) * 2 ** expectedFailureRetries, + Duration.toMillis(base) * 2 ** retries, MAX_EXPECTED_FAILURE_RETRY_DELAY_MS, ), ); - expectedFailureRetries += 1; return Stream.fromEffect(Effect.sleep(retryDelay)).pipe( Stream.drain, Stream.concat(subscribeToSession()), @@ -355,8 +356,26 @@ function subscribeDynamicMapped( ); // Both end one stream while the session can stay healthy, so nothing else // would replace it. A dead session is replaced by the supervisor, which - // interrupts this retry through switchMap. - if (isTransportFailure || isChunkDecodeDefect) { + // interrupts this retry through switchMap. A finite search must settle + // rather than restart, so it keeps waiting for the next session. + const resubscribesOnSameSession = + tag !== ORCHESTRATION_V2_WS_METHODS.searchThreadStream; + if (isTransportFailure && !resubscribesOnSameSession) { + return Stream.fromEffect( + Effect.logWarning( + "Durable RPC subscription lost its transport; waiting for the next session.", + { + cause: Cause.pretty(cause), + method: tag, + environmentId: supervisor.target.environmentId, + }, + ), + ).pipe(Stream.drain); + } + if ( + resubscribesOnSameSession && + (isTransportFailure || isChunkDecodeDefect) + ) { return Stream.fromEffect( Effect.logWarning( "Durable RPC subscription stream ended; resubscribing.", @@ -371,6 +390,7 @@ function subscribeDynamicMapped( Stream.concat( retryAfter( options?.retryExpectedFailureAfter ?? STREAM_FAILURE_RETRY_DELAY, + streamFailureRetries++, ), ), ); @@ -390,7 +410,9 @@ function subscribeDynamicMapped( return handled; } return handled.pipe( - Stream.concat(retryAfter(options.retryExpectedFailureAfter)), + Stream.concat( + retryAfter(options.retryExpectedFailureAfter, expectedFailureRetries++), + ), ); } return Stream.failCause(cause);