diff --git a/packages/client-runtime/src/rpc/client.test.ts b/packages/client-runtime/src/rpc/client.test.ts index 1d0005e35f0a4..de99737a28dd0 100644 --- a/packages/client-runtime/src/rpc/client.test.ts +++ b/packages/client-runtime/src/rpc/client.test.ts @@ -7,15 +7,17 @@ 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 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 TestClock from "effect/testing/TestClock"; import * as SubscriptionRef from "effect/SubscriptionRef"; +import * as TestClock from "effect/testing/TestClock"; import { RpcClientError } from "effect/rpc"; import { @@ -45,6 +47,8 @@ const INSTALL_DOWNLOADING: RelayClientInstallProgressEvent = { stage: "downloading", }; +const decodeNumber = Schema.decodeUnknownEffect(Schema.Number); + function session(client: WsRpcProtocolClient): RpcSession.RpcSession { return { client, @@ -265,6 +269,8 @@ describe("environment RPC", () => { while (subscriptions.length < 2) { yield* Queue.take(subscriptionChanges); } + // 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"]); @@ -272,6 +278,136 @@ describe("environment RPC", () => { }), ); + // 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 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, {}).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); + }, + ); + + it.effect("resubscribes on the same session after a transport failure", () => + 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", () => + // Mirrors RpcClient, which decodes each chunk with `Effect.orDie`. + resumesOnTheSameSession( + Stream.fromEffect(decodeNumber("not a number").pipe(Effect.orDie)).pipe(Stream.drain), + ), + ); + + 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/rpc/client.ts b/packages/client-runtime/src/rpc/client.ts index 4e3dad24e7a12..ee70400ca47a5 100644 --- a/packages/client-runtime/src/rpc/client.ts +++ b/packages/client-runtime/src/rpc/client.ts @@ -114,6 +114,7 @@ export const isAuthorizationFailure = (cause: Cause.Cause) => ); const MAX_EXPECTED_FAILURE_RETRY_DELAY_MS = 30_000; +const STREAM_FAILURE_RETRY_DELAY = Duration.seconds(1); export const subscriptionRetryDelay = (base: Duration.Input, retries: number) => Duration.millis( @@ -318,6 +319,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( @@ -354,6 +357,11 @@ function subscribeDynamicMapped( : Effect.void, ), Stream.catchCause((cause) => { + const retryAfter = (base: Duration.Input, retries: number) => + Stream.fromEffect(Effect.sleep(subscriptionRetryDelay(base, retries))).pipe( + Stream.drain, + Stream.concat(subscribeToSession()), + ); const hasOnlyExpectedFailures = cause.reasons.length > 0 && cause.reasons.every((reason) => reason._tag === "Fail"); @@ -362,7 +370,22 @@ 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. A caller + // with `onDefect` owns its defect recovery. + const isChunkDecodeDefect = + options?.onDefect === undefined && + 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. 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.", @@ -374,6 +397,29 @@ function subscribeDynamicMapped( ), ).pipe(Stream.drain); } + if ( + resubscribesOnSameSession && + (isTransportFailure || isChunkDecodeDefect) + ) { + return Stream.fromEffect( + Effect.logWarning( + "Durable RPC subscription stream ended; resubscribing.", + { + cause: Cause.pretty(cause), + method: tag, + environmentId: supervisor.target.environmentId, + }, + ), + ).pipe( + Stream.drain, + Stream.concat( + retryAfter( + options?.retryExpectedFailureAfter ?? STREAM_FAILURE_RETRY_DELAY, + streamFailureRetries++, + ), + ), + ); + } if (hasOnlyExpectedFailures && options?.onExpectedFailure !== undefined) { const handled = Stream.fromEffect(options.onExpectedFailure(cause)).pipe( Stream.drain, @@ -384,16 +430,10 @@ function subscribeDynamicMapped( ) { return handled; } - const retryDelay = subscriptionRetryDelay( - options.retryExpectedFailureAfter, - expectedFailureRetries, - ); - expectedFailureRetries += 1; return handled.pipe( Stream.concat( - Stream.fromEffect(Effect.sleep(retryDelay)).pipe(Stream.drain), + retryAfter(options.retryExpectedFailureAfter, expectedFailureRetries++), ), - Stream.concat(subscribeToSession()), ); } // A reported defect parks until the next session or resubscribe. Failing would diff --git a/packages/client-runtime/src/state/runtime.test.ts b/packages/client-runtime/src/state/runtime.test.ts index 0d2c1ba56088d..63f00b28fb173 100644 --- a/packages/client-runtime/src/state/runtime.test.ts +++ b/packages/client-runtime/src/state/runtime.test.ts @@ -18,6 +18,7 @@ import { WS_METHODS, } from "@t3tools/contracts"; import * as Cause from "effect/Cause"; +import * as Clock from "effect/Clock"; import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; @@ -29,6 +30,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 type { HttpClient } from "effect/http"; import { AsyncResult, Atom, AtomRegistry } from "effect/reactivity"; @@ -43,6 +45,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 { createOrchestrationEnvironmentAtoms } from "./orchestration.ts"; import { createReviewEnvironmentAtoms } from "./review.ts"; @@ -54,6 +57,7 @@ import { createTerminalEnvironmentAtoms } from "./terminal.ts"; import { createAtomCommandScheduler, createEnvironmentQueryAtomFamily, + createEnvironmentRpcSubscriptionAtomFamily, selectEnvironmentQueryData, createEnvironmentSubscriptionAtomFamily, createRuntimeCommand, @@ -70,6 +74,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", { @@ -393,6 +399,104 @@ describe("finite environment subscription lifecycle", () => { expect(observed.some((result) => AsyncResult.isFailure(result))).toBe(false); }), ); + 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); + }), + ), + ); it.effect.each([ { completedTtlMs: undefined, subscriptions: 2 }, { completedTtlMs: 60_000, subscriptions: 1 },