diff --git a/packages/client-runtime/src/connection/supervisor.test.ts b/packages/client-runtime/src/connection/supervisor.test.ts index 37adf87c1b47c6ad7406eb2b88fac37bbc6f6f20..f83598b8095296fa330e6ef62d9a8a5519382b8e 100644 --- a/packages/client-runtime/src/connection/supervisor.test.ts +++ b/packages/client-runtime/src/connection/supervisor.test.ts @@ -488,6 +488,29 @@ describe("EnvironmentSupervisor", () => { }).pipe(Effect.provide(TestClock.layer())), ); + it.effect("keeps a working socket open while a slow server config snapshot arrives", () => + Effect.gen(function* () { + const harness = yield* makeHarness({ + ready: () => Effect.sleep("40 seconds"), + }); + const supervisor = yield* EnvironmentSupervisor.make(TARGET_ENTRY, { + initiallyDesired: true, + }).pipe(Effect.provide(harness.dependencies)); + + yield* awaitState( + supervisor.state, + (state) => state.phase === "connecting" && state.stage === "synchronizing", + ); + yield* TestClock.adjust("39 seconds"); + expect((yield* SubscriptionRef.get(supervisor.state)).phase).toBe("connecting"); + + yield* TestClock.adjust("1 second"); + yield* awaitState(supervisor.state, (state) => state.phase === "connected"); + expect(yield* Ref.get(harness.prepareCount)).toBe(1); + expect(yield* Ref.get(harness.releaseCount)).toBe(0); + }).pipe(Effect.provide(TestClock.layer())), + ); + it.effect("retries when a session never becomes ready", () => Effect.gen(function* () { const harness = yield* makeHarness({ @@ -501,7 +524,8 @@ describe("EnvironmentSupervisor", () => { supervisor.state, (state) => state.phase === "connecting" && state.stage === "synchronizing", ); - yield* TestClock.adjust("14 seconds"); + // 15 seconds of setup plus 75 for the snapshot; setup finished instantly. + yield* TestClock.adjust("89 seconds"); expect((yield* SubscriptionRef.get(supervisor.state)).stage).toBe("synchronizing"); yield* TestClock.adjust("1 second"); @@ -1742,6 +1766,38 @@ describe("EnvironmentSupervisor routes", () => { }), ); + it.effect( + "gives the next route its own snapshot budget after a route fails while synchronizing", + () => + Effect.gen(function* () { + const harness = yield* makeHarness({ + checkRoute: () => Effect.succeed("answered"), + prepare: (_attempt, target) => Effect.succeed(preparedFor(target)), + // The LAN session fails 50 seconds into its snapshot; T3 Connect needs 40. + ready: (attempt) => + attempt === 1 + ? Effect.sleep("50 seconds").pipe(Effect.andThen(Effect.fail(transient()))) + : Effect.sleep("40 seconds"), + }); + const supervisor = yield* EnvironmentSupervisor.make(LAN_THEN_RELAY_ENTRY, { + initiallyDesired: true, + }).pipe(Effect.provide(harness.dependencies)); + + yield* awaitState( + supervisor.state, + (state) => state.phase === "connecting" && state.stage === "synchronizing", + ); + yield* TestClock.adjust("50 seconds"); + yield* TestClock.adjust("40 seconds"); + yield* awaitState(supervisor.state, (state) => state.phase === "connected"); + expect(Option.getOrThrow(yield* SubscriptionRef.get(supervisor.prepared)).target._tag).toBe( + "RelayConnectionTarget", + ); + expect(yield* Ref.get(harness.prepareCount)).toBe(2); + expect(yield* Ref.get(harness.releaseCount)).toBe(1); + }).pipe(Effect.provide(TestClock.layer())), + ); + it.effect("still tries a silent LAN route after the T3 Connect route fails", () => Effect.gen(function* () { const harness = yield* makeHarness({ diff --git a/packages/client-runtime/src/connection/supervisor.ts b/packages/client-runtime/src/connection/supervisor.ts index 67f8daae691ab5d592e1801326eb98c3ce1ba939..88a53216541da5c330e11c1ba47e99ba80987e34 100644 --- a/packages/client-runtime/src/connection/supervisor.ts +++ b/packages/client-runtime/src/connection/supervisor.ts @@ -39,8 +39,15 @@ import { const RETRY_BASE_DELAY_MS = 1_000; const RETRY_MAX_DELAY_MS = 300_000; +// Per-route budget to resolve credentials and create the session, which ends +// when the driver reports the synchronizing stage. Each route that gets that +// far adds its own budget for the socket to open (bounded at 15 seconds by the +// RPC session) and the server to deliver the first config snapshot. A cold +// server with a large database can take longer than 15 seconds to answer, and +// that is not a reason to drop a working connection. const CONNECTION_ESTABLISHMENT_TIMEOUT = "15 seconds"; const establishmentTimeout = Duration.fromInputUnsafe(CONNECTION_ESTABLISHMENT_TIMEOUT); +const synchronizationTimeout = Duration.fromInputUnsafe("75 seconds"); const CONNECTION_PROBE_TIMEOUT = "15 seconds"; // Mobile resumes, explicit retries, and offline events want a fast answer: // the user is waiting, or the network may be gone. @@ -428,9 +435,13 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( attempt: number, generation: number, lastFailure: ConnectionAttemptError | null, + synchronizedRoutes: Ref.Ref, ) { return yield* driver.connect(yield* attemptEntry, (progress) => - reportProgress(attempt, generation, lastFailure, progress), + (progress.stage === "synchronizing" + ? Ref.update(synchronizedRoutes, (count) => count + 1) + : Effect.void + ).pipe(Effect.andThen(reportProgress(attempt, generation, lastFailure, progress))), ); }); @@ -481,16 +492,17 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( generation: number, lastFailure: ConnectionAttemptError | null, pendingRetry: Option.Option, + synchronizedRoutes: Ref.Ref, ) { if (usesRelay) { return yield* traceRelayEstablishment( - establishConnection(attempt, generation, lastFailure), + establishConnection(attempt, generation, lastFailure, synchronizedRoutes), attempt, generation, pendingRetry, ); } - return yield* establishConnection(attempt, generation, lastFailure).pipe( + return yield* establishConnection(attempt, generation, lastFailure, synchronizedRoutes).pipe( Effect.map((lease) => ({ attemptSpan: Option.none(), lease, @@ -684,9 +696,16 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( ) { const switchingTo = yield* Ref.get(preferredRouteId); yield* SubscriptionRef.set(prepared, Option.none()); + const synchronizedRoutes = yield* Ref.make(0); const establishment = yield* Effect.raceAllFirst([ exitUnlessInterrupted( - establishTracedConnection(attempt, generation, lastFailure, pendingRetry), + establishTracedConnection( + attempt, + generation, + lastFailure, + pendingRetry, + synchronizedRoutes, + ), ).pipe( Effect.map((exit): EstablishmentEvent => ({ _tag: "Completed", @@ -699,10 +718,24 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( resetRetry, })), ), - // Each route may use the full setup time before the next is tried. - Effect.sleep( - Duration.times(establishmentTimeout, connectionRoutes(yield* Ref.get(currentEntry)).length), - ).pipe(Effect.as({ _tag: "TimedOut" })), + // Each route may use the full setup time before the next is tried, and + // each route that reaches synchronizing adds its own snapshot budget, so + // a route that fails late does not eat the next route's time. + Effect.gen(function* () { + const routeCount = connectionRoutes(yield* Ref.get(currentEntry)).length; + const startedAt = yield* Clock.monotonicTimeNanos; + const setup = Duration.toNanosUnsafe(Duration.times(establishmentTimeout, routeCount)); + while (true) { + const synchronized = BigInt(yield* Ref.get(synchronizedRoutes)); + const deadline = + startedAt + setup + synchronized * Duration.toNanosUnsafe(synchronizationTimeout); + const remaining = deadline - (yield* Clock.monotonicTimeNanos); + if (remaining <= 0n) { + return { _tag: "TimedOut" } as const satisfies EstablishmentEvent; + } + yield* Effect.sleep(Duration.nanos(remaining)); + } + }), ]); if (establishment._tag === "Interrupted") {