diff --git a/apps/desktop/src/backend/DesktopBackendConfiguration.test.ts b/apps/desktop/src/backend/DesktopBackendConfiguration.test.ts index 24c90943ec3..86f58fede19 100644 --- a/apps/desktop/src/backend/DesktopBackendConfiguration.test.ts +++ b/apps/desktop/src/backend/DesktopBackendConfiguration.test.ts @@ -135,6 +135,9 @@ describe("DesktopBackendConfiguration", () => { assert.equal(first.cwd, environment.backendCwd); assert.equal(first.captureOutput, true); assert.equal(first.env.ELECTRON_RUN_AS_NODE, "1"); + // Heap headroom for the primary backend (issue #21). extendEnv:true means + // this config value wins over an inherited process.env NODE_OPTIONS. + assert.include(first.env.NODE_OPTIONS ?? "", "--max-old-space-size=4096"); assert.isUndefined(first.env.T3CODE_PORT); assert.isUndefined(first.env.T3CODE_MODE); assert.isUndefined(first.env.T3CODE_DESKTOP_LAN_HOST); @@ -152,6 +155,43 @@ describe("DesktopBackendConfiguration", () => { ), ); + it.effect("resolvePrimary sets a bare heap flag when NODE_OPTIONS is unset", () => { + const previous = process.env.NODE_OPTIONS; + delete process.env.NODE_OPTIONS; + return withHarness( + Effect.gen(function* () { + const configuration = yield* DesktopBackendConfiguration.DesktopBackendConfiguration; + const config = yield* configuration.resolvePrimary; + assert.equal(config.env.NODE_OPTIONS, "--max-old-space-size=4096"); + }), + ).pipe(Effect.ensuring(Effect.sync(() => restoreEnv("NODE_OPTIONS", previous)))); + }); + + it.effect("resolvePrimary appends heap headroom to an existing NODE_OPTIONS", () => { + const previous = process.env.NODE_OPTIONS; + process.env.NODE_OPTIONS = "--enable-source-maps"; + return withHarness( + Effect.gen(function* () { + const configuration = yield* DesktopBackendConfiguration.DesktopBackendConfiguration; + const config = yield* configuration.resolvePrimary; + assert.equal(config.env.NODE_OPTIONS, "--enable-source-maps --max-old-space-size=4096"); + }), + ).pipe(Effect.ensuring(Effect.sync(() => restoreEnv("NODE_OPTIONS", previous)))); + }); + + it.effect("resolvePrimary preserves an explicit NODE_OPTIONS heap setting", () => { + const previous = process.env.NODE_OPTIONS; + process.env.NODE_OPTIONS = "--max-old-space-size=1024"; + return withHarness( + Effect.gen(function* () { + const configuration = yield* DesktopBackendConfiguration.DesktopBackendConfiguration; + const config = yield* configuration.resolvePrimary; + // An operator override wins; we do not append a second, conflicting flag. + assert.equal(config.env.NODE_OPTIONS, "--max-old-space-size=1024"); + }), + ).pipe(Effect.ensuring(Effect.sync(() => restoreEnv("NODE_OPTIONS", previous)))); + }); + it.effect("resolveWsl reuses the primary's bootstrap token", () => withHarness( Effect.gen(function* () { @@ -279,6 +319,8 @@ describe("DesktopBackendConfiguration", () => { "env", "PATH=/home/test user's/.nvm/versions/node/v22.0.0/bin:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin:/home/test user/bin:/opt/test's tools/bin:/usr/bin:/bin", nodePath, + // Heap headroom, passed as a V8 execArg before the entry script (issue #21). + "--max-old-space-size=4096", linuxEntryPath, "--bootstrap-fd", "0", diff --git a/apps/desktop/src/backend/DesktopBackendConfiguration.ts b/apps/desktop/src/backend/DesktopBackendConfiguration.ts index 7ae3ad6c912..6bc4bf0ca18 100644 --- a/apps/desktop/src/backend/DesktopBackendConfiguration.ts +++ b/apps/desktop/src/backend/DesktopBackendConfiguration.ts @@ -94,6 +94,30 @@ const WSL_FORWARDED_ENV_NAMES = ["OPENAI_API_KEY", "ANTHROPIC_API_KEY"] as const const WSL_SERVER_SYSTEM_PATH = "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin"; +// Give the backend a predictable V8 old-space ceiling instead of inheriting the +// default. This is headroom for transient heap spikes (large diffs, big file +// reads) so a memory-constrained host is less likely to push the backend into a +// GC-thrash / stall while it is busy — part of the robustness work for issue +// #21. It is a ceiling, not a reservation, so it does not raise steady-state +// memory use (the backend's normal RSS is well under this). +const DESKTOP_BACKEND_MAX_OLD_SPACE_MB = 4096; +const DESKTOP_BACKEND_MAX_OLD_SPACE_FLAG = `--max-old-space-size=${DESKTOP_BACKEND_MAX_OLD_SPACE_MB}`; + +// Append our heap-headroom flag to any NODE_OPTIONS already exported (by the +// user or the dev-runner) rather than replacing it. If an explicit +// --max-old-space-size is already present we leave it untouched so an operator +// can override the ceiling. Electron-run-as-node honours the NODE_OPTIONS +// safelist, which includes --max-old-space-size. +const backendNodeOptions = (existing: string | undefined): string => { + const trimmed = existing?.trim() ?? ""; + if (trimmed.includes("--max-old-space-size")) { + return trimmed; + } + return trimmed.length > 0 + ? `${trimmed} ${DESKTOP_BACKEND_MAX_OLD_SPACE_FLAG}` + : DESKTOP_BACKEND_MAX_OLD_SPACE_FLAG; +}; + const backendChildEnvPatch = (): Record => Object.fromEntries(DESKTOP_BACKEND_ENV_NAMES.map((name) => [name, undefined])); @@ -356,6 +380,7 @@ const resolvePrimaryStartConfig = Effect.fn("desktop.backendConfiguration.resolv env: { ...backendChildEnvPatch(), ELECTRON_RUN_AS_NODE: "1", + NODE_OPTIONS: backendNodeOptions(process.env.NODE_OPTIONS), }, // Primary wants process.env (PATH, dev-runner's T3CODE_HOME, etc.). extendEnv: true, @@ -546,6 +571,10 @@ const resolveWslStartConfig = Effect.fn("desktop.backendConfiguration.resolveWsl "env", `PATH=${launchPath}`, preflight.nodePath, + // Match the primary backend's heap headroom. WSL runs plain node, so the + // flag is passed as a V8 execArg (before the entry script) rather than via + // NODE_OPTIONS. + DESKTOP_BACKEND_MAX_OLD_SPACE_FLAG, preflight.linuxEntryPath, "--bootstrap-fd", "0", diff --git a/packages/client-runtime/src/connection/supervisor.test.ts b/packages/client-runtime/src/connection/supervisor.test.ts index f3901e42251..fcdbf7be3fa 100644 --- a/packages/client-runtime/src/connection/supervisor.test.ts +++ b/packages/client-runtime/src/connection/supervisor.test.ts @@ -113,6 +113,10 @@ const makeHarness = Effect.fn("TestConnectionHarness.make")(function* (options?: attempt: number, target: ConnectionTarget, ) => Effect.Effect; + // Runs at the "opening" stage, after the socket URL is prepared but before a + // session is acquired — lets a test stall/fail an attempt at "opening" + // specifically (distinct from "preparing" and "synchronizing"). + readonly opening?: (attempt: number) => Effect.Effect; readonly ready?: (attempt: number) => Effect.Effect; readonly probe?: (attempt: number) => Effect.Effect; }) { @@ -154,6 +158,9 @@ const makeHarness = Effect.fn("TestConnectionHarness.make")(function* (options?: yield* reportProgress({ stage: "preparing" }); const prepared = yield* prepare(target); yield* reportProgress({ stage: "opening", prepared }); + if (options?.opening) { + yield* options.opening(yield* Ref.get(prepareCount)); + } const attempt = yield* Ref.updateAndGet(sessionCount, (count) => count + 1); const closed = yield* Deferred.make(); @@ -725,6 +732,325 @@ describe("EnvironmentSupervisor", () => { }).pipe(Effect.provide(TestClock.layer())), ); + it.effect( + "extends the establishment budget and retries quickly after a slow (synchronizing) timeout", + () => + Effect.gen(function* () { + // The backend accepts the socket (reaches "synchronizing") but is always + // slow to finish readiness. Attempt 1 uses the normal 15s budget and times + // out; subsequent attempts must get the enlarged budget and a fast retry so + // a briefly-overloaded-but-alive backend is not abandoned. See issue #21. + const harness = yield* makeHarness({ + ready: () => Effect.never, + }); + const supervisor = yield* EnvironmentSupervisor.make(TARGET_ENTRY, { + initiallyDesired: true, + }).pipe(Effect.provide(harness.dependencies)); + + // Attempt 1: normal 15s budget. + yield* awaitState( + supervisor.state, + (state) => + state.phase === "connecting" && state.stage === "synchronizing" && state.attempt === 1, + ); + yield* TestClock.adjust("15 seconds"); + yield* eventuallyState( + supervisor.state, + (state) => + state.phase === "backoff" && + state.attempt === 1 && + state.lastFailure?.reason === "timeout", + ); + + // Fast retry (1s), not the exponential ladder. + yield* TestClock.adjust("1 second"); + yield* eventuallyState( + supervisor.state, + (state) => + state.phase === "connecting" && state.stage === "synchronizing" && state.attempt === 2, + ); + + // Attempt 2 must survive past the old 15s budget... + yield* TestClock.adjust("15 seconds"); + expect((yield* SubscriptionRef.get(supervisor.state)).phase).toBe("connecting"); + // ...and only time out once the enlarged 45s budget elapses. + yield* TestClock.adjust("30 seconds"); + yield* eventuallyState( + supervisor.state, + (state) => state.phase === "backoff" && state.attempt === 2, + ); + + // Still a fast 1s retry (not 2s), proving the slow path does not escalate. + yield* TestClock.adjust("1 second"); + yield* eventuallyState( + supervisor.state, + (state) => state.phase === "connecting" && state.attempt === 3, + ); + }).pipe(Effect.provide(TestClock.layer())), + ); + + it.effect("connects once a slow backend answers within the enlarged budget", () => + Effect.gen(function* () { + // Attempt 1 stalls at readiness (slow timeout); attempt 2's readiness takes + // 30s — longer than the normal 15s budget but within the enlarged one — and + // must succeed instead of being abandoned. + const harness = yield* makeHarness({ + ready: (attempt) => (attempt === 1 ? Effect.never : Effect.sleep("30 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("15 seconds"); + yield* eventuallyState(supervisor.state, (state) => state.phase === "backoff"); + + yield* TestClock.adjust("1 second"); + yield* eventuallyState( + supervisor.state, + (state) => + state.phase === "connecting" && state.stage === "synchronizing" && state.attempt === 2, + ); + // 30s of readiness would time out under the old 15s budget; the enlarged + // budget lets it complete. (Generation counts established connections, and + // attempt 1 never established, so the successful second attempt is + // generation 1.) + yield* TestClock.adjust("30 seconds"); + yield* eventuallyState( + supervisor.state, + (state) => state.phase === "connected" && state.attempt === 2, + ); + expect(yield* Ref.get(harness.sessionCount)).toBe(2); + }).pipe(Effect.provide(TestClock.layer())), + ); + + it.effect("runs a periodic background liveness probe while connected", () => + Effect.gen(function* () { + const probeCount = yield* Ref.make(0); + const harness = yield* makeHarness({ + probe: () => Ref.update(probeCount, (count) => count + 1), + }); + const supervisor = yield* EnvironmentSupervisor.make(TARGET_ENTRY, { + initiallyDesired: true, + }).pipe(Effect.provide(harness.dependencies)); + + yield* awaitState(supervisor.state, (state) => state.phase === "connected"); + expect(yield* Ref.get(probeCount)).toBe(0); + + // No foreground activation — the heartbeat alone drives the probe. + yield* TestClock.adjust("30 seconds"); + yield* eventuallyState(supervisor.state, () => true); + expect(yield* Ref.get(probeCount)).toBe(1); + expect((yield* SubscriptionRef.get(supervisor.state)).phase).toBe("connected"); + + yield* TestClock.adjust("30 seconds"); + yield* eventuallyState(supervisor.state, () => true); + expect(yield* Ref.get(probeCount)).toBe(2); + expect((yield* SubscriptionRef.get(supervisor.state)).phase).toBe("connected"); + expect(yield* Ref.get(harness.sessionCount)).toBe(1); + expect(yield* Ref.get(harness.releaseCount)).toBe(0); + }).pipe(Effect.provide(TestClock.layer())), + ); + + it.effect("reconnects when a background heartbeat probe times out", () => + Effect.gen(function* () { + const harness = yield* makeHarness({ + probe: (attempt) => (attempt === 1 ? Effect.never : Effect.void), + }); + const supervisor = yield* EnvironmentSupervisor.make(TARGET_ENTRY, { + initiallyDesired: true, + }).pipe(Effect.provide(harness.dependencies)); + + yield* awaitState( + supervisor.state, + (state) => state.phase === "connected" && state.generation === 1, + ); + // Heartbeat fires... + yield* TestClock.adjust("30 seconds"); + // ...the probe stalls and times out after the probe budget. + yield* TestClock.adjust("15 seconds"); + yield* eventuallyState( + supervisor.state, + (state) => state.phase === "backoff" && state.lastFailure?.reason === "timeout", + ); + yield* TestClock.adjust("1 second"); + yield* eventuallyState( + supervisor.state, + (state) => state.phase === "connected" && state.generation === 2, + ); + expect(yield* Ref.get(harness.sessionCount)).toBe(2); + expect(yield* Ref.get(harness.releaseCount)).toBe(1); + }).pipe(Effect.provide(TestClock.layer())), + ); + + it.effect( + "keeps the normal budget and exponential backoff for a timeout that never reached the backend", + () => + Effect.gen(function* () { + // prepare hangs, so every attempt times out at the "preparing" stage + // without ever reaching the backend — this must NOT get the tolerant path. + const harness = yield* makeHarness({ prepare: () => Effect.never }); + const supervisor = yield* EnvironmentSupervisor.make(TARGET_ENTRY, { + initiallyDesired: true, + }).pipe(Effect.provide(harness.dependencies)); + + yield* awaitState( + supervisor.state, + (state) => + state.phase === "connecting" && state.stage === "preparing" && state.attempt === 1, + ); + yield* TestClock.adjust("15 seconds"); + yield* eventuallyState( + supervisor.state, + (state) => state.phase === "backoff" && state.attempt === 1, + ); + + // First retry is 1s either way (RETRY_DELAYS_MS[0] === SLOW_RETRY_DELAY_MS). + yield* TestClock.adjust("1 second"); + yield* eventuallyState( + supervisor.state, + (state) => + state.phase === "connecting" && state.stage === "preparing" && state.attempt === 2, + ); + // Attempt 2 must use the NORMAL 15s budget, not the tolerant 45s one: + // it times out at 15s (a tolerant attempt would still be connecting). + yield* TestClock.adjust("15 seconds"); + yield* eventuallyState( + supervisor.state, + (state) => state.phase === "backoff" && state.attempt === 2, + ); + // ...and the retry is the exponential 2s, not the tolerant 1s. + yield* TestClock.adjust("1 second"); + expect((yield* SubscriptionRef.get(supervisor.state)).phase).toBe("backoff"); + yield* TestClock.adjust("1 second"); + yield* eventuallyState( + supervisor.state, + (state) => state.phase === "connecting" && state.attempt === 3, + ); + }).pipe(Effect.provide(TestClock.layer())), + ); + + it.effect("treats a timeout at the 'opening' stage as slow-but-reachable", () => + Effect.gen(function* () { + // Stall at "opening" (socket URL prepared, session not yet acquired). This + // is the lower boundary of reachedBackend, so it must take the tolerant path. + const harness = yield* makeHarness({ opening: () => Effect.never }); + const supervisor = yield* EnvironmentSupervisor.make(TARGET_ENTRY, { + initiallyDesired: true, + }).pipe(Effect.provide(harness.dependencies)); + + yield* awaitState( + supervisor.state, + (state) => state.phase === "connecting" && state.stage === "opening" && state.attempt === 1, + ); + yield* TestClock.adjust("15 seconds"); + yield* eventuallyState( + supervisor.state, + (state) => state.phase === "backoff" && state.attempt === 1, + ); + + yield* TestClock.adjust("1 second"); + yield* eventuallyState( + supervisor.state, + (state) => state.phase === "connecting" && state.stage === "opening" && state.attempt === 2, + ); + // Attempt 2 gets the enlarged budget: still connecting past the old 15s... + yield* TestClock.adjust("15 seconds"); + expect((yield* SubscriptionRef.get(supervisor.state)).phase).toBe("connecting"); + // ...timing out only once the 45s budget elapses. + yield* TestClock.adjust("30 seconds"); + yield* eventuallyState( + supervisor.state, + (state) => state.phase === "backoff" && state.attempt === 2, + ); + }).pipe(Effect.provide(TestClock.layer())), + ); + + it.effect("falls back to the normal budget after too many consecutive slow timeouts", () => + Effect.gen(function* () { + // Always slow at synchronizing. The tolerant path is bounded, so after + // MAX_TOLERANT_TIMEOUTS (3) consecutive slow timeouts it must revert to the + // normal 15s budget — otherwise a misclassified dead endpoint would loop on + // 45s attempts forever. + const harness = yield* makeHarness({ ready: () => Effect.never }); + const supervisor = yield* EnvironmentSupervisor.make(TARGET_ENTRY, { + initiallyDesired: true, + }).pipe(Effect.provide(harness.dependencies)); + + // Attempt 1: normal 15s budget, then three tolerant 45s attempts (2, 3, 4). + yield* awaitState( + supervisor.state, + (state) => state.phase === "connecting" && state.attempt === 1, + ); + yield* TestClock.adjust("15 seconds"); + yield* eventuallyState( + supervisor.state, + (state) => state.phase === "backoff" && state.attempt === 1, + ); + for (const attempt of [2, 3, 4]) { + yield* TestClock.adjust("1 second"); + yield* eventuallyState( + supervisor.state, + (state) => state.phase === "connecting" && state.attempt === attempt, + ); + // Each of these is a tolerant attempt: still connecting past 15s. + yield* TestClock.adjust("15 seconds"); + expect((yield* SubscriptionRef.get(supervisor.state)).phase).toBe("connecting"); + yield* TestClock.adjust("30 seconds"); + yield* eventuallyState( + supervisor.state, + (state) => state.phase === "backoff" && state.attempt === attempt, + ); + } + + // The 4th slow timeout tripped the cap, so attempt 5 uses the exponential + // ladder (retryDelayMs(3) === 8s) and the normal 15s budget again. + yield* TestClock.adjust("8 seconds"); + yield* eventuallyState( + supervisor.state, + (state) => state.phase === "connecting" && state.attempt === 5, + ); + yield* TestClock.adjust("15 seconds"); + yield* eventuallyState( + supervisor.state, + (state) => state.phase === "backoff" && state.attempt === 5, + ); + }).pipe(Effect.provide(TestClock.layer())), + ); + + it.effect("reconnects a relay session on a credentials change during a background probe", () => + Effect.gen(function* () { + // The first session's heartbeat probe stalls; a credentials change that + // lands while it is in flight must still tear down and reconnect rather + // than wait for the probe to time out. + const harness = yield* makeHarness({ + probe: (attempt) => (attempt === 1 ? Effect.never : Effect.void), + }); + const supervisor = yield* EnvironmentSupervisor.make(RELAY_ENTRY, { + initiallyDesired: true, + }).pipe(Effect.provide(harness.dependencies)); + + yield* awaitState( + supervisor.state, + (state) => state.phase === "connected" && state.generation === 1, + ); + // Heartbeat fires and the probe stalls. + yield* TestClock.adjust("30 seconds"); + // Credentials change mid-probe — reconnect must happen without waiting for + // the 15s probe timeout. + yield* harness.wake("credentials-changed"); + yield* awaitState( + supervisor.state, + (state) => state.phase === "connected" && state.generation === 2, + ); + expect(yield* Ref.get(harness.sessionCount)).toBe(2); + expect(yield* Ref.get(harness.releaseCount)).toBe(1); + }).pipe(Effect.provide(TestClock.layer())), + ); + it.effect("honors an explicit disconnect while a foreground probe is stalled", () => Effect.gen(function* () { const probeStarted = yield* Deferred.make(); diff --git a/packages/client-runtime/src/connection/supervisor.ts b/packages/client-runtime/src/connection/supervisor.ts index d9efcd4263a..fac4660586c 100644 --- a/packages/client-runtime/src/connection/supervisor.ts +++ b/packages/client-runtime/src/connection/supervisor.ts @@ -19,6 +19,7 @@ import * as Connectivity from "./connectivity.ts"; import * as ConnectionDriver from "./driver.ts"; import { type ConnectionAttemptError, + type ConnectionAttemptStage, type ConnectionTarget, ConnectionTransientError, type NetworkStatus, @@ -31,7 +32,30 @@ import * as ConnectionWakeups from "./wakeups.ts"; const RETRY_DELAYS_MS = [1_000, 2_000, 4_000, 8_000, 16_000] as const; const CONNECTION_ESTABLISHMENT_TIMEOUT = "15 seconds"; +// A reconnect attempt that reached the backend (the socket opened / we began +// synchronizing) but then timed out is treated as "slow, not dead": the next +// attempt gets this enlarged establishment budget so a briefly-overloaded — but +// still alive — backend is not repeatedly abandoned before it can answer. This +// is the client half of issue #21 (host memory pressure makes the local backend +// slow, not gone). A timeout that never reached the backend keeps the normal +// budget + exponential backoff below. +const SLOW_CONNECTION_ESTABLISHMENT_TIMEOUT = "45 seconds"; const CONNECTION_PROBE_TIMEOUT = "15 seconds"; +// After a slow timeout we retry promptly instead of climbing the exponential +// ladder — the enlarged establishment budget already paces the attempts, and the +// backend is known to be reachable. +const SLOW_RETRY_DELAY_MS = 1_000; +// The stages we key "slow, not dead" off (opening/synchronizing) are reported +// before the socket is confirmed open, so a genuinely unreachable endpoint can +// be misclassified as slow. Bound the tolerant path: after this many consecutive +// slow timeouts we fall back to the normal budget + exponential backoff, so a +// dead endpoint still surfaces backoff and a truly slow-but-alive backend has +// had ample time (a few 45s attempts) to answer. +const MAX_TOLERANT_TIMEOUTS = 3; +// While connected, actively probe the backend on this cadence so a wedged — but +// still open — socket is detected in the background, not only when the app is +// brought to the foreground. +const CONNECTION_HEARTBEAT_INTERVAL = "30 seconds"; const BACKOFF_RESET_AFTER_MS = 30_000; interface SupervisorIntent { @@ -103,6 +127,26 @@ function retryDelayMs(failureCount: number): number { return RETRY_DELAYS_MS[Math.min(failureCount, RETRY_DELAYS_MS.length - 1)] ?? 16_000; } +const STAGE_RANK: Record = { + preparing: 1, + opening: 2, + synchronizing: 3, +}; + +function furthestStage( + current: ConnectionAttemptStage | null, + next: ConnectionAttemptStage, +): ConnectionAttemptStage { + return current === null || STAGE_RANK[next] > STAGE_RANK[current] ? next : current; +} + +// True once an attempt has begun talking to the backend (socket opening or +// later). A timeout at or past this point means the backend is reachable but +// slow, so we should be patient rather than escalate backoff. +function reachedBackend(stage: ConnectionAttemptStage | null): boolean { + return stage !== null && STAGE_RANK[stage] >= STAGE_RANK.opening; +} + function annotateTarget(target: ConnectionTarget) { return Effect.annotateCurrentSpan({ "environment.id": target.environmentId, @@ -238,6 +282,10 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( ); const session = yield* SubscriptionRef.make>(Option.none()); const prepared = yield* SubscriptionRef.make>(Option.none()); + // Tracks how far the current attempt progressed. Read after an attempt fails + // to distinguish a slow-but-reachable backend from an unreachable one. The + // supervisor runs a single attempt at a time, so a shared Ref is safe. + const attemptFurthestStage = yield* Ref.make(null); const clearLease = Effect.all( [SubscriptionRef.set(session, Option.none()), SubscriptionRef.set(prepared, Option.none())], @@ -272,6 +320,7 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( if ("prepared" in progress) { yield* SubscriptionRef.set(prepared, Option.some(progress.prepared)); } + yield* Ref.update(attemptFurthestStage, (current) => furthestStage(current, progress.stage)); yield* setState( connectingState(yield* Ref.get(intent), generation, attempt, lastFailure, progress.stage), ); @@ -383,11 +432,86 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( } }); + // Runs a single liveness probe against the active lease while staying + // responsive to supervisor signals. Returns "continue" when the probe succeeds + // (stay connected) or "return" when a signal (disconnect / retry / offline) + // asks us to stop monitoring. Fails with a ConnectionTransientError when the + // probe fails or times out, which the caller surfaces as a reconnect. + const runLivenessProbe = Effect.fnUntraced(function* ( + lease: ConnectionDriver.EnvironmentConnectionLease, + ) { + const probe = yield* lease.session.probe.pipe( + Effect.timeoutOrElse({ + duration: CONNECTION_PROBE_TIMEOUT, + orElse: () => + Effect.fail( + new ConnectionTransientError({ + reason: "timeout", + detail: `${target.label} did not respond to a connection health check.`, + }), + ), + }), + Effect.forkChild, + ); + for (;;) { + const probeEvent = yield* Effect.raceFirst( + Fiber.await(probe).pipe(Effect.map((exit) => ({ _tag: "ProbeCompleted" as const, exit }))), + Queue.take(signals).pipe(Effect.map((signal) => ({ _tag: "Signal" as const, signal }))), + ); + if (probeEvent._tag === "ProbeCompleted") { + yield* probeEvent.exit; + return "continue" as const; + } + switch (probeEvent.signal._tag) { + case "DisconnectRequested": + case "RetryRequested": + yield* Fiber.interrupt(probe); + return "return" as const; + case "NetworkChanged": + if (probeEvent.signal.network === "offline") { + yield* Fiber.interrupt(probe); + return "return" as const; + } + break; + case "Wakeup": + // A relay credentials change must still tear down and reconnect even + // if it lands while a probe is in flight — otherwise the enclosing + // monitor would only see it on the next signal, leaving the session on + // stale credentials. Mirror monitorConnectedLease's handling. + if ( + probeEvent.signal.reason === "credentials-changed" && + target._tag === "RelayConnectionTarget" + ) { + yield* Fiber.interrupt(probe); + yield* logManagedRelayAccountChange; + return "return" as const; + } + break; + case "ConnectRequested": + break; + } + } + }); + const monitorConnectedLease = Effect.fnUntraced(function* ( lease: ConnectionDriver.EnvironmentConnectionLease, ) { for (;;) { - const next = yield* Queue.take(signals); + // Wait for the next signal, but wake on a heartbeat interval too so a + // wedged-but-open socket is probed in the background rather than only when + // the app is activated. An interrupted Queue.take does not consume a + // signal, so the loser of this race is not lost. + const event = yield* Effect.raceFirst( + Queue.take(signals).pipe(Effect.map((signal) => ({ _tag: "Signal" as const, signal }))), + Effect.sleep(CONNECTION_HEARTBEAT_INTERVAL).pipe(Effect.as({ _tag: "Heartbeat" as const })), + ); + if (event._tag === "Heartbeat") { + if ((yield* runLivenessProbe(lease)) === "return") { + return; + } + continue; + } + const next = event.signal; switch (next._tag) { case "DisconnectRequested": case "RetryRequested": @@ -403,47 +527,8 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( return; } if (next.reason === "application-active") { - const probe = yield* lease.session.probe.pipe( - Effect.timeoutOrElse({ - duration: CONNECTION_PROBE_TIMEOUT, - orElse: () => - Effect.fail( - new ConnectionTransientError({ - reason: "timeout", - detail: `${target.label} did not respond to a connection health check.`, - }), - ), - }), - Effect.forkChild, - ); - for (;;) { - const probeEvent = yield* Effect.raceFirst( - Fiber.await(probe).pipe( - Effect.map((exit) => ({ _tag: "ProbeCompleted" as const, exit })), - ), - Queue.take(signals).pipe( - Effect.map((signal) => ({ _tag: "Signal" as const, signal })), - ), - ); - if (probeEvent._tag === "ProbeCompleted") { - yield* probeEvent.exit; - break; - } - switch (probeEvent.signal._tag) { - case "DisconnectRequested": - case "RetryRequested": - yield* Fiber.interrupt(probe); - return; - case "NetworkChanged": - if (probeEvent.signal.network === "offline") { - yield* Fiber.interrupt(probe); - return; - } - break; - case "ConnectRequested": - case "Wakeup": - break; - } + if ((yield* runLivenessProbe(lease)) === "return") { + return; } } break; @@ -458,8 +543,13 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( generation: number, lastFailure: ConnectionAttemptError | null, pendingRetry: Option.Option, + tolerant: boolean, ) { + yield* Ref.set(attemptFurthestStage, null); yield* SubscriptionRef.set(prepared, Option.none()); + const establishmentTimeout = tolerant + ? SLOW_CONNECTION_ESTABLISHMENT_TIMEOUT + : CONNECTION_ESTABLISHMENT_TIMEOUT; const establishment = yield* Effect.raceAllFirst([ exitUnlessInterrupted( establishTracedConnection(attempt, generation, lastFailure, pendingRetry), @@ -472,9 +562,7 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( ), ), waitForEstablishmentInterrupt().pipe(Effect.as({ _tag: "Interrupted" })), - Effect.sleep(CONNECTION_ESTABLISHMENT_TIMEOUT).pipe( - Effect.as({ _tag: "TimedOut" }), - ), + Effect.sleep(establishmentTimeout).pipe(Effect.as({ _tag: "TimedOut" })), ]); if (establishment._tag === "Interrupted") { @@ -589,6 +677,12 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( let generation = 0; let latestFailure: ConnectionAttemptError | null = null; let pendingRetry = Option.none(); + // When true, the previous attempt timed out after reaching the backend, so + // the next attempt gets the enlarged establishment budget. Bounded by + // consecutiveSlowTimeouts so a misclassified dead endpoint reverts to + // exponential backoff. + let tolerant = false; + let consecutiveSlowTimeouts = 0; for (;;) { const currentIntent = yield* Ref.get(intent); @@ -596,6 +690,8 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( failureCount = 0; latestFailure = null; pendingRetry = Option.none(); + tolerant = false; + consecutiveSlowTimeouts = 0; yield* clearLease; yield* setState(availableState(currentIntent, generation)); yield* waitForSignal; @@ -611,7 +707,7 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( const attempt = failureCount + 1; const nextGeneration = generation + 1; const outcome: AttemptOutcome = yield* Effect.scoped( - runAttempt(attempt, nextGeneration, latestFailure, pendingRetry), + runAttempt(attempt, nextGeneration, latestFailure, pendingRetry, tolerant), ); if (outcome.established) { generation = nextGeneration; @@ -619,6 +715,8 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( failureCount = 0; latestFailure = null; pendingRetry = Option.none(); + tolerant = false; + consecutiveSlowTimeouts = 0; } } if (outcome._tag === "Interrupted") { @@ -629,6 +727,8 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( const error: ConnectionAttemptError = outcome.failure.error; latestFailure = error; if (error._tag === "ConnectionBlockedError") { + tolerant = false; + consecutiveSlowTimeouts = 0; const blockedIntent = yield* Ref.get(intent); yield* setState({ desired: blockedIntent.desired, @@ -645,7 +745,18 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( } failureCount += 1; - const delayMs = retryDelayMs(failureCount - 1); + // A timeout that reached the backend means it is (probably) alive but slow: + // retry promptly and grant the next attempt the enlarged budget instead of + // climbing the exponential ladder — but only up to MAX_TOLERANT_TIMEOUTS in + // a row, after which we assume the endpoint is really unreachable and fall + // back to normal backoff. Every other failure keeps the existing behaviour. + const furthest = yield* Ref.get(attemptFurthestStage); + const reachedBackendTimeout: boolean = error.reason === "timeout" && reachedBackend(furthest); + const slowTimeout: boolean = + reachedBackendTimeout && consecutiveSlowTimeouts < MAX_TOLERANT_TIMEOUTS; + consecutiveSlowTimeouts = slowTimeout ? consecutiveSlowTimeouts + 1 : 0; + tolerant = slowTimeout; + const delayMs = slowTimeout ? SLOW_RETRY_DELAY_MS : retryDelayMs(failureCount - 1); pendingRetry = Option.map(attemptSpan, (previousAttempt) => ({ previousAttempt, failureCount,