| author | |
| committer | |
| log | 990bb0b689b23a33f8e53f12962fe2542b3f4005 |
| tree | 6467139147b298037087d3b9545cc827d80d4c37 |
| parent | 7251f1a1f14afdf626e7c7f8664da423594732a3 |
| signature | Signed by PGP key B5690EEEBB952194 |
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>3 files changed, 98 insertions(+), 14 deletions(-)
apps/server/src/server.ts+7-2| ... | @@ -583,8 +583,13 @@ export const makeServerLayer = Layer.unwrap( | ... | @@ -583,8 +583,13 @@ export const makeServerLayer = Layer.unwrap( |
| 583 | const server = yield* HttpServer.HttpServer; | 583 | const server = yield* HttpServer.HttpServer; |
| 584 | const address = server.address; | 584 | const address = server.address; |
| 585 | if (typeof address === "string" || !("port" in address)) return; | 585 | if (typeof address === "string" || !("port" in address)) return; |
| 586 | yield* Effect.sleep("250 millis").pipe( | 586 | // No settling delay before the first attempt: routes are already |
| 587 | Effect.andThen(reconcileDesiredCloudLink(`http://127.0.0.1:${address.port}`)), | 587 | // serving by the time activation opens this gate (the startup |
| 588 | // sequence awaits routesReady), and the retry schedule below | ||
| 589 | // covers anything this sleep used to hedge against. Every | ||
| 590 | // millisecond here is dead time on the path to remote | ||
| 591 | // reachability after a restart. | ||
| 592 | yield* reconcileDesiredCloudLink(`http://127.0.0.1:${address.port}`).pipe( | ||
| 588 | Effect.retry({ | 593 | Effect.retry({ |
| 589 | while: (error) => | 594 | while: (error) => |
| 590 | error._tag !== "EnvironmentHttpBadRequestError" && | 595 | error._tag !== "EnvironmentHttpBadRequestError" && |
packages/client-runtime/src/state/server.test.ts+55| ... | @@ -7,12 +7,16 @@ import { | ... | @@ -7,12 +7,16 @@ import { |
| 7 | } from "@t3tools/contracts"; | 7 | } from "@t3tools/contracts"; |
| 8 | import { describe, expect, it } from "@effect/vitest"; | 8 | import { describe, expect, it } from "@effect/vitest"; |
| 9 | import * as Cause from "effect/Cause"; | 9 | import * as Cause from "effect/Cause"; |
| 10 | import * as Duration from "effect/Duration"; | ||
| 10 | import * as Effect from "effect/Effect"; | 11 | import * as Effect from "effect/Effect"; |
| 11 | import * as Exit from "effect/Exit"; | 12 | import * as Exit from "effect/Exit"; |
| 13 | import * as Fiber from "effect/Fiber"; | ||
| 12 | import * as Option from "effect/Option"; | 14 | import * as Option from "effect/Option"; |
| 13 | import * as Queue from "effect/Queue"; | 15 | import * as Queue from "effect/Queue"; |
| 16 | import * as Ref from "effect/Ref"; | ||
| 14 | import * as Stream from "effect/Stream"; | 17 | import * as Stream from "effect/Stream"; |
| 15 | import * as SubscriptionRef from "effect/SubscriptionRef"; | 18 | import * as SubscriptionRef from "effect/SubscriptionRef"; |
| 19 | import * as TestClock from "effect/testing/TestClock"; | ||
| 16 | import { RpcClientError } from "effect/unstable/rpc"; | 20 | import { RpcClientError } from "effect/unstable/rpc"; |
| 17 | import * as Socket from "effect/unstable/socket/Socket"; | 21 | import * as Socket from "effect/unstable/socket/Socket"; |
| 18 | 22 | ||
| ... | @@ -30,6 +34,7 @@ import { | ... | @@ -30,6 +34,7 @@ import { |
| 30 | makeEnvironmentServerConfigState, | 34 | makeEnvironmentServerConfigState, |
| 31 | isLegacyUpdateHandoffLoss, | 35 | isLegacyUpdateHandoffLoss, |
| 32 | matchesServerUpdateReadyEvent, | 36 | matchesServerUpdateReadyEvent, |
| 37 | nudgeReconnectDuringUpdateRestart, | ||
| 33 | projectServerWelcome, | 38 | projectServerWelcome, |
| 34 | resolveServerConfigValue, | 39 | resolveServerConfigValue, |
| 35 | resolveServerUpdateProgressResult, | 40 | resolveServerUpdateProgressResult, |
| ... | @@ -71,6 +76,56 @@ function session(client: WsRpcProtocolClient): RpcSession { | ... | @@ -71,6 +76,56 @@ function session(client: WsRpcProtocolClient): RpcSession { |
| 71 | }; | 76 | }; |
| 72 | } | 77 | } |
| 73 | 78 | ||
| 79 | describe("update restart reconnect nudges", () => { | ||
| 80 | it.effect("retries once per backoff entry instead of only the first", () => | ||
| 81 | Effect.gen(function* () { | ||
| 82 | const retries = yield* Ref.make(0); | ||
| 83 | const states = [ | ||
| 84 | { phase: "backoff" }, | ||
| 85 | { phase: "connecting" }, | ||
| 86 | { phase: "backoff" }, | ||
| 87 | { phase: "backoff" }, | ||
| 88 | ]; | ||
| 89 | |||
| 90 | yield* nudgeReconnectDuringUpdateRestart({ | ||
| 91 | stateChanges: Stream.fromIterable(states), | ||
| 92 | retryNow: Ref.update(retries, (count) => count + 1), | ||
| 93 | interval: Duration.zero, | ||
| 94 | }); | ||
| 95 | |||
| 96 | // Three backoff entries, three nudges. The old one-shot behavior fired | ||
| 97 | // once and then let the supervisor's ladder stretch to 16-second gaps. | ||
| 98 | expect(yield* Ref.get(retries)).toBe(3); | ||
| 99 | }), | ||
| 100 | ); | ||
| 101 | |||
| 102 | it.effect("paces nudges so a fast-failing connection cannot spin", () => | ||
| 103 | Effect.gen(function* () { | ||
| 104 | const retries = yield* Ref.make(0); | ||
| 105 | |||
| 106 | const fiber = yield* Effect.forkChild( | ||
| 107 | nudgeReconnectDuringUpdateRestart({ | ||
| 108 | stateChanges: Stream.fromIterable([{ phase: "backoff" }, { phase: "backoff" }]), | ||
| 109 | retryNow: Ref.update(retries, (count) => count + 1), | ||
| 110 | }), | ||
| 111 | { startImmediately: true }, | ||
| 112 | ); | ||
| 113 | |||
| 114 | // Each nudge waits out the interval first, so nothing fires immediately. | ||
| 115 | yield* TestClock.adjust(Duration.millis(999)); | ||
| 116 | expect(yield* Ref.get(retries)).toBe(0); | ||
| 117 | |||
| 118 | yield* TestClock.adjust(Duration.millis(1)); | ||
| 119 | expect(yield* Ref.get(retries)).toBe(1); | ||
| 120 | |||
| 121 | yield* TestClock.adjust(Duration.seconds(1)); | ||
| 122 | expect(yield* Ref.get(retries)).toBe(2); | ||
| 123 | |||
| 124 | yield* Fiber.join(fiber); | ||
| 125 | }).pipe(Effect.provide(TestClock.layer())), | ||
| 126 | ); | ||
| 127 | }); | ||
| 128 | |||
| 74 | describe("server state projection", () => { | 129 | describe("server state projection", () => { |
| 75 | it("only treats a legacy transport interruption as an unacknowledged handoff", () => { | 130 | it("only treats a legacy transport interruption as an unacknowledged handoff", () => { |
| 76 | expect(isLegacyUpdateHandoffLoss(Cause.interrupt(1))).toBe(true); | 131 | expect(isLegacyUpdateHandoffLoss(Cause.interrupt(1))).toBe(true); |
packages/client-runtime/src/state/server.ts+36-12| ... | @@ -149,6 +149,35 @@ export function validateServerUpdateReadyEvent( | ... | @@ -149,6 +149,35 @@ export function validateServerUpdateReadyEvent( |
| 149 | ); | 149 | ); |
| 150 | } | 150 | } |
| 151 | 151 | ||
| 152 | /** | ||
| 153 | * Keeps reconnect attempts ~1s apart for the whole update restart. | ||
| 154 | * | ||
| 155 | * A restart takes the server down for ~15 seconds, but the supervisor's normal | ||
| 156 | * backoff ladder (1/2/4/8/16s) assumes an unexpected failure and lands attempts | ||
| 157 | * at ~3, 5, 9, 17 and 33 seconds — so a 15-second restart is observed as a | ||
| 158 | * 33-second "Resuming". Nudging on every backoff entry (not just the first) | ||
| 159 | * holds the retry cadence flat until the server answers again. The sleep before | ||
| 160 | * each nudge is the pacer: a connection that fails instantly re-enters backoff | ||
| 161 | * immediately and would otherwise spin a tight retry loop. | ||
| 162 | * | ||
| 163 | * Callers fork this as a child of the update command so it is interrupted as | ||
| 164 | * soon as the update settles, whether it succeeds, fails, or times out. | ||
| 165 | */ | ||
| 166 | export function nudgeReconnectDuringUpdateRestart(input: { | ||
| 167 | readonly stateChanges: Stream.Stream<{ readonly phase: string }, unknown>; | ||
| 168 | readonly retryNow: Effect.Effect<void>; | ||
| 169 | readonly interval?: Duration.Duration; | ||
| 170 | }): Effect.Effect<void> { | ||
| 171 | return input.stateChanges.pipe( | ||
| 172 | Stream.filter((state) => state.phase === "backoff"), | ||
| 173 | Stream.runForEach(() => | ||
| 174 | Effect.sleep(input.interval ?? Duration.seconds(1)).pipe(Effect.andThen(input.retryNow)), | ||
| 175 | ), | ||
| 176 | Effect.timeoutOption(SERVER_UPDATE_RESUME_TIMEOUT), | ||
| 177 | Effect.ignore, | ||
| 178 | ); | ||
| 179 | } | ||
| 180 | |||
| 152 | export function serverUpdateStateForProgressEvent( | 181 | export function serverUpdateStateForProgressEvent( |
| 153 | fromVersion: string, | 182 | fromVersion: string, |
| 154 | targetVersion: string, | 183 | targetVersion: string, |
| ... | @@ -589,18 +618,13 @@ export function createServerEnvironmentAtoms<R, E>( | ... | @@ -589,18 +618,13 @@ export function createServerEnvironmentAtoms<R, E>( |
| 589 | }), | 618 | }), |
| 590 | ); | 619 | ); |
| 591 | 620 | ||
| 592 | // The update restart is intentional. As soon as the supervisor sees | 621 | // The update restart is intentional and the server stays unreachable |
| 593 | // that first failed connection, discard any prior backoff debt and | 622 | // for the whole restart, so hold the retry cadence flat instead of |
| 594 | // retry immediately instead of carrying an old 16-second delay. | 623 | // letting the supervisor climb its backoff ladder. |
| 595 | yield* environmentRegistry.stateChanges(target.environmentId).pipe( | 624 | yield* nudgeReconnectDuringUpdateRestart({ |
| 596 | Stream.filter((state) => state.phase === "backoff"), | 625 | stateChanges: environmentRegistry.stateChanges(target.environmentId), |
| 597 | Stream.take(1), | 626 | retryNow: environmentRegistry.retryNow(target.environmentId), |
| 598 | Stream.runDrain, | 627 | }).pipe(Effect.forkChild); |
| 599 | Effect.andThen(environmentRegistry.retryNow(target.environmentId)), | ||
| 600 | Effect.timeoutOption(Duration.seconds(30)), | ||
| 601 | Effect.ignore, | ||
| 602 | Effect.forkChild, | ||
| 603 | ); | ||
| 604 | 628 | ||
| 605 | const resumed = yield* environmentRegistry | 629 | const resumed = yield* environmentRegistry |
| 606 | .followStream(target.environmentId, subscribe(WS_METHODS.subscribeServerLifecycle, {})) | 630 | .followStream(target.environmentId, subscribe(WS_METHODS.subscribeServerLifecycle, {})) |