| author | |
| committer | |
| log | f36df1852423fef7f7855f86946c9f9a95c3429f |
| tree | 17ab412d7b615d0df9dc472c458199e858862785 |
| parent | 8bd7745f0af918582807326d90e58d59324bb2a6 |
| signature | Signed by PGP key B5690EEEBB952194 |
Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>4 files changed, 334 insertions(+), 13 deletions(-)
apps/web/src/connection/platform.test.ts+70| ... | @@ -7,10 +7,17 @@ import { | ... | @@ -7,10 +7,17 @@ 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 Effect from "effect/Effect"; | 9 | import * as Effect from "effect/Effect"; |
| 10 | import { | ||
| 11 | ConnectionBlockedError, | ||
| 12 | ConnectionTransientError, | ||
| 13 | } from "@t3tools/client-runtime/connection"; | ||
| 10 | 14 | ||
| 11 | import { | 15 | import { |
| 12 | canRetainCachedPlatformRegistrationAfterRefreshFailure, | 16 | canRetainCachedPlatformRegistrationAfterRefreshFailure, |
| 13 | canReuseCachedPlatformRegistration, | 17 | canReuseCachedPlatformRegistration, |
| 18 | isRejectedBootstrapCredentialError, | ||
| 19 | isRejectedSecondaryBootstrap, | ||
| 20 | nextRejectedSecondaryBootstrap, | ||
| 14 | primaryRegistrationToRetainAfterTopologyRead, | 21 | primaryRegistrationToRetainAfterTopologyRead, |
| 15 | provisionDesktopSshEnvironment, | 22 | provisionDesktopSshEnvironment, |
| 16 | readPrimaryEnvironmentTargetResult, | 23 | readPrimaryEnvironmentTargetResult, |
| ... | @@ -202,6 +209,69 @@ describe("desktop-local bearer cache", () => { | ... | @@ -202,6 +209,69 @@ describe("desktop-local bearer cache", () => { |
| 202 | }); | 209 | }); |
| 203 | }); | 210 | }); |
| 204 | 211 | ||
| 212 | describe("rejected desktop-local bootstrap tokens", () => { | ||
| 213 | it("backs off on a rejected signature and retries it on a growing, capped delay", () => { | ||
| 214 | const signature = "http://a|ws://a|old-token"; | ||
| 215 | const first = nextRejectedSecondaryBootstrap(undefined, signature, 0); | ||
| 216 | |||
| 217 | expect(isRejectedSecondaryBootstrap(first, signature, 59_999)).toBe(true); | ||
| 218 | // A backend restarted on the same port accepts the same token again, so it is retried. | ||
| 219 | expect(isRejectedSecondaryBootstrap(first, signature, 60_000)).toBe(false); | ||
| 220 | |||
| 221 | const second = nextRejectedSecondaryBootstrap(first, signature, 60_000); | ||
| 222 | expect(second.retryAtEpochMs).toBe(180_000); | ||
| 223 | |||
| 224 | let backoff = second; | ||
| 225 | for (let attempt = 0; attempt < 10; attempt += 1) { | ||
| 226 | backoff = nextRejectedSecondaryBootstrap(backoff, signature, 0); | ||
| 227 | } | ||
| 228 | expect(backoff.delayMs).toBe(30 * 60_000); | ||
| 229 | }); | ||
| 230 | |||
| 231 | it("retries at once when the token or endpoint changes", () => { | ||
| 232 | const rejected = nextRejectedSecondaryBootstrap(undefined, "http://a|ws://a|old-token", 0); | ||
| 233 | |||
| 234 | expect(isRejectedSecondaryBootstrap(rejected, "http://a|ws://a|new-token", 1)).toBe(false); | ||
| 235 | expect(isRejectedSecondaryBootstrap(rejected, "http://b|ws://b|old-token", 1)).toBe(false); | ||
| 236 | expect(isRejectedSecondaryBootstrap(undefined, "http://a|ws://a|old-token", 1)).toBe(false); | ||
| 237 | expect(nextRejectedSecondaryBootstrap(rejected, "http://a|ws://a|new-token", 1).delayMs).toBe( | ||
| 238 | 60_000, | ||
| 239 | ); | ||
| 240 | }); | ||
| 241 | |||
| 242 | it("treats only authentication rejections as a dead credential", () => { | ||
| 243 | expect( | ||
| 244 | isRejectedBootstrapCredentialError( | ||
| 245 | new ConnectionBlockedError({ reason: "authentication", detail: "invalid" }), | ||
| 246 | ), | ||
| 247 | ).toBe(true); | ||
| 248 | expect( | ||
| 249 | isRejectedBootstrapCredentialError( | ||
| 250 | new ConnectionTransientError({ reason: "endpoint-unavailable", detail: "booting" }), | ||
| 251 | ), | ||
| 252 | ).toBe(false); | ||
| 253 | }); | ||
| 254 | }); | ||
| 255 | |||
| 256 | describe("desktop-local bearer cache across token changes", () => { | ||
| 257 | it("keeps a live bearer when only the bootstrap token rotates", () => { | ||
| 258 | // Cached by endpoint: a rotated token reuses the bearer instead of | ||
| 259 | // dropping the environment (and its drafts) when the new token is rejected. | ||
| 260 | const cached = { | ||
| 261 | signature: "http://a|ws://a", | ||
| 262 | registration: {} as never, | ||
| 263 | expiresAtEpochMs: 20_000, | ||
| 264 | refreshAtEpochMs: 15_000, | ||
| 265 | }; | ||
| 266 | |||
| 267 | expect(canReuseCachedPlatformRegistration(cached, "http://a|ws://a", 10_000)).toBe(true); | ||
| 268 | expect( | ||
| 269 | canRetainCachedPlatformRegistrationAfterRefreshFailure(cached, "http://a|ws://a", 16_000), | ||
| 270 | ).toBe(true); | ||
| 271 | expect(canReuseCachedPlatformRegistration(cached, "http://b|ws://b", 10_000)).toBe(false); | ||
| 272 | }); | ||
| 273 | }); | ||
| 274 | |||
| 205 | describe("primary topology cache", () => { | 275 | describe("primary topology cache", () => { |
| 206 | const registration = {} as never; | 276 | const registration = {} as never; |
| 207 | const cached = { | 277 | const cached = { |
apps/web/src/connection/platform.ts+112-5| ... | @@ -9,6 +9,7 @@ import { | ... | @@ -9,6 +9,7 @@ import { |
| 9 | BearerConnectionRegistration, | 9 | BearerConnectionRegistration, |
| 10 | BearerConnectionTarget, | 10 | BearerConnectionTarget, |
| 11 | ConnectionBlockedError, | 11 | ConnectionBlockedError, |
| 12 | type ConnectionAttemptError, | ||
| 12 | ConnectionTransientError, | 13 | ConnectionTransientError, |
| 13 | Connectivity, | 14 | Connectivity, |
| 14 | mapRemoteEnvironmentError, | 15 | mapRemoteEnvironmentError, |
| ... | @@ -420,7 +421,7 @@ const loadSecondaryConnectionRegistration = Effect.fn( | ... | @@ -420,7 +421,7 @@ const loadSecondaryConnectionRegistration = Effect.fn( |
| 420 | 421 | ||
| 421 | // Poll cadence for the desktop bootstrap topology. There is no change event on | 422 | // Poll cadence for the desktop bootstrap topology. There is no change event on |
| 422 | // the bridge, so the renderer polls; successful registrations are cached by a | 423 | // the bridge, so the renderer polls; successful registrations are cached by a |
| 423 | // signature of their endpoint + token until bearer credentials approach expiry. | 424 | // signature of their endpoint until bearer credentials approach expiry. |
| 424 | const PLATFORM_POLL_INTERVAL = "3 seconds"; | 425 | const PLATFORM_POLL_INTERVAL = "3 seconds"; |
| 425 | const SECONDARY_BEARER_REFRESH_SKEW_MS = 5_000; | 426 | const SECONDARY_BEARER_REFRESH_SKEW_MS = 5_000; |
| 426 | 427 | ||
| ... | @@ -447,6 +448,8 @@ interface CachedPlatformRegistration { | ... | @@ -447,6 +448,8 @@ interface CachedPlatformRegistration { |
| 447 | readonly registration: PlatformConnectionRegistration; | 448 | readonly registration: PlatformConnectionRegistration; |
| 448 | readonly expiresAtEpochMs?: number; | 449 | readonly expiresAtEpochMs?: number; |
| 449 | readonly refreshAtEpochMs?: number; | 450 | readonly refreshAtEpochMs?: number; |
| 451 | /** The bootstrap token a desktop-local bearer was exchanged for. */ | ||
| 452 | readonly bootstrapToken?: string; | ||
| 450 | } | 453 | } |
| 451 | 454 | ||
| 452 | export type PrimaryEnvironmentTargetRead = | 455 | export type PrimaryEnvironmentTargetRead = |
| ... | @@ -499,6 +502,52 @@ export function canRetainCachedPlatformRegistrationAfterRefreshFailure( | ... | @@ -499,6 +502,52 @@ export function canRetainCachedPlatformRegistrationAfterRefreshFailure( |
| 499 | ); | 502 | ); |
| 500 | } | 503 | } |
| 501 | 504 | ||
| 505 | const REJECTED_BOOTSTRAP_RETRY_INITIAL_MS = 60_000; | ||
| 506 | const REJECTED_BOOTSTRAP_RETRY_MAX_MS = 30 * 60_000; | ||
| 507 | |||
| 508 | /** A bootstrap token a backend rejected, and when the poll may try it again. */ | ||
| 509 | export interface RejectedSecondaryBootstrap { | ||
| 510 | readonly signature: string; | ||
| 511 | readonly retryAtEpochMs: number; | ||
| 512 | readonly delayMs: number; | ||
| 513 | } | ||
| 514 | |||
| 515 | /** | ||
| 516 | * A backend that rejected a bootstrap token will usually keep rejecting it, so | ||
| 517 | * the poll backs off on that exact signature instead of re-presenting a dead | ||
| 518 | * credential every few seconds. It still retries on a capped backoff: a | ||
| 519 | * backend that restarts on the same port seeds a fresh grant for the same | ||
| 520 | * token, and nothing in the topology says it restarted. A new token or | ||
| 521 | * endpoint retries at once. | ||
| 522 | */ | ||
| 523 | export function isRejectedSecondaryBootstrap( | ||
| 524 | rejected: RejectedSecondaryBootstrap | undefined, | ||
| 525 | signature: string, | ||
| 526 | nowEpochMs: number, | ||
| 527 | ): rejected is RejectedSecondaryBootstrap { | ||
| 528 | return ( | ||
| 529 | rejected !== undefined && | ||
| 530 | rejected.signature === signature && | ||
| 531 | nowEpochMs < rejected.retryAtEpochMs | ||
| 532 | ); | ||
| 533 | } | ||
| 534 | |||
| 535 | export function nextRejectedSecondaryBootstrap( | ||
| 536 | previous: RejectedSecondaryBootstrap | undefined, | ||
| 537 | signature: string, | ||
| 538 | nowEpochMs: number, | ||
| 539 | ): RejectedSecondaryBootstrap { | ||
| 540 | const delayMs = | ||
| 541 | previous?.signature === signature | ||
| 542 | ? Math.min(previous.delayMs * 2, REJECTED_BOOTSTRAP_RETRY_MAX_MS) | ||
| 543 | : REJECTED_BOOTSTRAP_RETRY_INITIAL_MS; | ||
| 544 | return { signature, retryAtEpochMs: nowEpochMs + delayMs, delayMs }; | ||
| 545 | } | ||
| 546 | |||
| 547 | export function isRejectedBootstrapCredentialError(error: ConnectionAttemptError): boolean { | ||
| 548 | return error._tag === "ConnectionBlockedError" && error.reason === "authentication"; | ||
| 549 | } | ||
| 550 | |||
| 502 | export function secondaryRegistrationsToRetainAfterTopologyRead( | 551 | export function secondaryRegistrationsToRetainAfterTopologyRead( |
| 503 | previous: ReadonlyMap<string, CachedPlatformRegistration>, | 552 | previous: ReadonlyMap<string, CachedPlatformRegistration>, |
| 504 | topologyRead: DesktopSecondaryBootstrapsRead, | 553 | topologyRead: DesktopSecondaryBootstrapsRead, |
| ... | @@ -523,6 +572,7 @@ const layerPlatformConnectionSource = Layer.effect( | ... | @@ -523,6 +572,7 @@ const layerPlatformConnectionSource = Layer.effect( |
| 523 | }); | 572 | }); |
| 524 | } | 573 | } |
| 525 | const cacheRef = yield* Ref.make(new Map<string, CachedPlatformRegistration>()); | 574 | const cacheRef = yield* Ref.make(new Map<string, CachedPlatformRegistration>()); |
| 575 | const rejectedRef = yield* Ref.make(new Map<string, RejectedSecondaryBootstrap>()); | ||
| 526 | 576 | ||
| 527 | // Resolve the full set of platform-managed environments the host currently | 577 | // Resolve the full set of platform-managed environments the host currently |
| 528 | // reports: the primary (same-origin cookie auth) plus any desktop-local | 578 | // reports: the primary (same-origin cookie auth) plus any desktop-local |
| ... | @@ -588,17 +638,45 @@ const layerPlatformConnectionSource = Layer.effect( | ... | @@ -588,17 +638,45 @@ const layerPlatformConnectionSource = Layer.effect( |
| 588 | cause: topologyRead.cause, | 638 | cause: topologyRead.cause, |
| 589 | }); | 639 | }); |
| 590 | } else { | 640 | } else { |
| 641 | const rejected = yield* Ref.get(rejectedRef); | ||
| 642 | const nextRejected = new Map<string, RejectedSecondaryBootstrap>(); | ||
| 591 | for (const bootstrap of topologyRead.bootstraps) { | 643 | for (const bootstrap of topologyRead.bootstraps) { |
| 592 | const signature = `${bootstrap.httpBaseUrl}|${bootstrap.wsBaseUrl}|${bootstrap.bootstrapToken ?? ""}`; | 644 | // The cached bearer belongs to the endpoint, not to the bootstrap |
| 645 | // token it was exchanged for: a new token must not drop a live | ||
| 646 | // session (its removal also clears the environment's drafts). The | ||
| 647 | // token only decides whether a rejected exchange is retried. | ||
| 648 | const endpointSignature = `${bootstrap.httpBaseUrl}|${bootstrap.wsBaseUrl}`; | ||
| 649 | const signature = `${endpointSignature}|${bootstrap.bootstrapToken ?? ""}`; | ||
| 593 | const cached = previous.get(bootstrap.id); | 650 | const cached = previous.get(bootstrap.id); |
| 651 | // A new token from the desktop means something changed (rotation, a | ||
| 652 | // restarted backend): exchange it rather than reusing a bearer that | ||
| 653 | // may be dead. The old bearer stays registered if that exchange fails. | ||
| 594 | if ( | 654 | if ( |
| 595 | cached !== undefined && | 655 | cached !== undefined && |
| 596 | canReuseCachedPlatformRegistration(cached, signature, nowEpochMs) | 656 | cached.bootstrapToken === bootstrap.bootstrapToken && |
| 657 | canReuseCachedPlatformRegistration(cached, endpointSignature, nowEpochMs) | ||
| 597 | ) { | 658 | ) { |
| 598 | next.set(bootstrap.id, cached); | 659 | next.set(bootstrap.id, cached); |
| 599 | registrations.push(cached.registration); | 660 | registrations.push(cached.registration); |
| 600 | continue; | 661 | continue; |
| 601 | } | 662 | } |
| 663 | const previouslyRejected = rejected.get(bootstrap.id); | ||
| 664 | if (isRejectedSecondaryBootstrap(previouslyRejected, signature, nowEpochMs)) { | ||
| 665 | nextRejected.set(bootstrap.id, previouslyRejected); | ||
| 666 | // The bearer minted before the token died is still good until it expires. | ||
| 667 | if ( | ||
| 668 | cached !== undefined && | ||
| 669 | canRetainCachedPlatformRegistrationAfterRefreshFailure( | ||
| 670 | cached, | ||
| 671 | endpointSignature, | ||
| 672 | nowEpochMs, | ||
| 673 | ) | ||
| 674 | ) { | ||
| 675 | next.set(bootstrap.id, cached); | ||
| 676 | registrations.push(cached.registration); | ||
| 677 | } | ||
| 678 | continue; | ||
| 679 | } | ||
| 602 | const built = yield* loadSecondaryConnectionRegistration(bootstrap).pipe( | 680 | const built = yield* loadSecondaryConnectionRegistration(bootstrap).pipe( |
| 603 | Effect.tapError((error) => | 681 | Effect.tapError((error) => |
| 604 | Effect.logWarning("Could not connect a desktop-local backend.", { | 682 | Effect.logWarning("Could not connect a desktop-local backend.", { |
| ... | @@ -606,20 +684,49 @@ const layerPlatformConnectionSource = Layer.effect( | ... | @@ -606,20 +684,49 @@ const layerPlatformConnectionSource = Layer.effect( |
| 606 | error, | 684 | error, |
| 607 | }), | 685 | }), |
| 608 | ), | 686 | ), |
| 687 | Effect.tapError((error) => | ||
| 688 | isRejectedBootstrapCredentialError(error) | ||
| 689 | ? // Back off from when the rejection arrived; the exchanges in | ||
| 690 | // this poll can take longer than the first backoff step. | ||
| 691 | Clock.currentTimeMillis.pipe( | ||
| 692 | Effect.map((rejectedAtEpochMs) => | ||
| 693 | nextRejected.set( | ||
| 694 | bootstrap.id, | ||
| 695 | nextRejectedSecondaryBootstrap( | ||
| 696 | previouslyRejected, | ||
| 697 | signature, | ||
| 698 | rejectedAtEpochMs, | ||
| 699 | ), | ||
| 700 | ), | ||
| 701 | ), | ||
| 702 | ) | ||
| 703 | : Effect.void, | ||
| 704 | ), | ||
| 609 | Effect.option, | 705 | Effect.option, |
| 610 | ); | 706 | ); |
| 611 | if (Option.isSome(built)) { | 707 | if (Option.isSome(built)) { |
| 612 | const cacheEntry = { signature, ...built.value }; | 708 | const cacheEntry = { |
| 709 | signature: endpointSignature, | ||
| 710 | ...(bootstrap.bootstrapToken === undefined | ||
| 711 | ? {} | ||
| 712 | : { bootstrapToken: bootstrap.bootstrapToken }), | ||
| 713 | ...built.value, | ||
| 714 | }; | ||
| 613 | next.set(bootstrap.id, cacheEntry); | 715 | next.set(bootstrap.id, cacheEntry); |
| 614 | registrations.push(built.value.registration); | 716 | registrations.push(built.value.registration); |
| 615 | } else if ( | 717 | } else if ( |
| 616 | cached !== undefined && | 718 | cached !== undefined && |
| 617 | canRetainCachedPlatformRegistrationAfterRefreshFailure(cached, signature, nowEpochMs) | 719 | canRetainCachedPlatformRegistrationAfterRefreshFailure( |
| 720 | cached, | ||
| 721 | endpointSignature, | ||
| 722 | nowEpochMs, | ||
| 723 | ) | ||
| 618 | ) { | 724 | ) { |
| 619 | next.set(bootstrap.id, cached); | 725 | next.set(bootstrap.id, cached); |
| 620 | registrations.push(cached.registration); | 726 | registrations.push(cached.registration); |
| 621 | } | 727 | } |
| 622 | } | 728 | } |
| 729 | yield* Ref.set(rejectedRef, nextRejected); | ||
| 623 | } | 730 | } |
| 624 | 731 | ||
| 625 | yield* Ref.set(cacheRef, next); | 732 | yield* Ref.set(cacheRef, next); |
packages/client-runtime/src/connection/registry.test.ts+111| ... | @@ -158,6 +158,10 @@ const makeHarness = Effect.fn("TestEnvironmentRegistry.makeHarness")(function* ( | ... | @@ -158,6 +158,10 @@ const makeHarness = Effect.fn("TestEnvironmentRegistry.makeHarness")(function* ( |
| 158 | ) => Effect.Effect<void, Persistence.ConnectionPersistenceError>; | 158 | ) => Effect.Effect<void, Persistence.ConnectionPersistenceError>; |
| 159 | readonly checkRoute?: (route: ConnectionRoute) => RouteCheck; | 159 | readonly checkRoute?: (route: ConnectionRoute) => RouteCheck; |
| 160 | readonly prepareRoute?: (target: ConnectionTarget) => ConnectionBlockedError | undefined; | 160 | readonly prepareRoute?: (target: ConnectionTarget) => ConnectionBlockedError | undefined; |
| 161 | /** Runs before preparing a route; may suspend to hold an attempt in flight. */ | ||
| 162 | readonly beforePrepareRoute?: ( | ||
| 163 | target: ConnectionTarget, | ||
| 164 | ) => Effect.Effect<ConnectionBlockedError | undefined>; | ||
| 161 | readonly initialDisabled?: ReadonlyArray<EnvironmentId>; | 165 | readonly initialDisabled?: ReadonlyArray<EnvironmentId>; |
| 162 | }, | 166 | }, |
| 163 | ) { | 167 | ) { |
| ... | @@ -385,6 +389,10 @@ const makeHarness = Effect.fn("TestEnvironmentRegistry.makeHarness")(function* ( | ... | @@ -385,6 +389,10 @@ const makeHarness = Effect.fn("TestEnvironmentRegistry.makeHarness")(function* ( |
| 385 | target, | 389 | target, |
| 386 | }; | 390 | }; |
| 387 | if (options?.prepareError) return yield* options.prepareError; | 391 | if (options?.prepareError) return yield* options.prepareError; |
| 392 | const heldError = options?.beforePrepareRoute | ||
| 393 | ? yield* options.beforePrepareRoute(target) | ||
| 394 | : undefined; | ||
| 395 | if (heldError !== undefined) return yield* heldError; | ||
| 388 | const routeError = options?.prepareRoute?.(target); | 396 | const routeError = options?.prepareRoute?.(target); |
| 389 | if (routeError !== undefined) return yield* routeError; | 397 | if (routeError !== undefined) return yield* routeError; |
| 390 | yield* Ref.update(connectedRoutes, (current) => [...current, connectionRouteId(target)]); | 398 | yield* Ref.update(connectedRoutes, (current) => [...current, connectionRouteId(target)]); |
| ... | @@ -1270,6 +1278,109 @@ describe("EnvironmentRegistry", () => { | ... | @@ -1270,6 +1278,109 @@ describe("EnvironmentRegistry", () => { |
| 1270 | }), | 1278 | }), |
| 1271 | ); | 1279 | ); |
| 1272 | 1280 | ||
| 1281 | it.effect("a replaced platform bearer wakes a supervisor blocked on the old one", () => | ||
| 1282 | Effect.gen(function* () { | ||
| 1283 | let storedCredentials: Ref.Ref<Map<string, ConnectionCredential>> | undefined; | ||
| 1284 | const harness = yield* makeHarness([], [], [], { | ||
| 1285 | // Authentication fails while the old bearer is stored, as the server | ||
| 1286 | // would answer once that session was revoked. | ||
| 1287 | beforePrepareRoute: (target) => | ||
| 1288 | target._tag !== "BearerConnectionTarget" || storedCredentials === undefined | ||
| 1289 | ? Effect.succeed(undefined) | ||
| 1290 | : Ref.get(storedCredentials).pipe( | ||
| 1291 | Effect.map((current) => | ||
| 1292 | current.get(target.connectionId)?.token === "old-bearer" | ||
| 1293 | ? new ConnectionBlockedError({ reason: "authentication", detail: "revoked" }) | ||
| 1294 | : undefined, | ||
| 1295 | ), | ||
| 1296 | ), | ||
| 1297 | }); | ||
| 1298 | storedCredentials = harness.storedCredentials; | ||
| 1299 | |||
| 1300 | yield* Effect.gen(function* () { | ||
| 1301 | const registry = yield* EnvironmentRegistry.EnvironmentRegistry; | ||
| 1302 | const registration = (token: string) => | ||
| 1303 | new BearerConnectionRegistration({ | ||
| 1304 | target: BEARER_TARGET, | ||
| 1305 | profile: BEARER_PROFILE, | ||
| 1306 | credential: new BearerConnectionCredential({ token }), | ||
| 1307 | }); | ||
| 1308 | yield* registry.reconcilePlatform([registration("old-bearer")]); | ||
| 1309 | yield* awaitConnectionState( | ||
| 1310 | registry, | ||
| 1311 | BEARER_TARGET.environmentId, | ||
| 1312 | (state) => state.phase === "blocked", | ||
| 1313 | ); | ||
| 1314 | |||
| 1315 | yield* registry.reconcilePlatform([registration("new-bearer")]); | ||
| 1316 | |||
| 1317 | yield* awaitConnectionState( | ||
| 1318 | registry, | ||
| 1319 | BEARER_TARGET.environmentId, | ||
| 1320 | (state) => state.phase === "connected", | ||
| 1321 | ); | ||
| 1322 | }).pipe(Effect.provide(harness.layer), Effect.scoped); | ||
| 1323 | }), | ||
| 1324 | ); | ||
| 1325 | |||
| 1326 | it.effect("a bearer replaced while an attempt is in flight retries with the new one", () => | ||
| 1327 | Effect.gen(function* () { | ||
| 1328 | const attemptStarted = yield* Deferred.make<void>(); | ||
| 1329 | const releaseFirstAttempt = yield* Deferred.make<void>(); | ||
| 1330 | const attempts = yield* Ref.make(0); | ||
| 1331 | const harness = yield* makeHarness([], [], [], { | ||
| 1332 | // The first attempt read the old bearer; it stalls, then fails | ||
| 1333 | // authentication after the bearer has already been replaced. | ||
| 1334 | beforePrepareRoute: (target) => | ||
| 1335 | target._tag !== "BearerConnectionTarget" | ||
| 1336 | ? Effect.succeed(undefined) | ||
| 1337 | : Ref.updateAndGet(attempts, (count) => count + 1).pipe( | ||
| 1338 | Effect.flatMap((count) => | ||
| 1339 | count === 1 | ||
| 1340 | ? Deferred.succeed(attemptStarted, undefined).pipe( | ||
| 1341 | Effect.andThen(Deferred.await(releaseFirstAttempt)), | ||
| 1342 | Effect.as( | ||
| 1343 | new ConnectionBlockedError({ | ||
| 1344 | reason: "authentication", | ||
| 1345 | detail: "revoked", | ||
| 1346 | }), | ||
| 1347 | ), | ||
| 1348 | ) | ||
| 1349 | : Effect.succeed(undefined), | ||
| 1350 | ), | ||
| 1351 | ), | ||
| 1352 | }); | ||
| 1353 | |||
| 1354 | yield* Effect.gen(function* () { | ||
| 1355 | const registry = yield* EnvironmentRegistry.EnvironmentRegistry; | ||
| 1356 | const registration = (token: string) => | ||
| 1357 | new BearerConnectionRegistration({ | ||
| 1358 | target: BEARER_TARGET, | ||
| 1359 | profile: BEARER_PROFILE, | ||
| 1360 | credential: new BearerConnectionCredential({ token }), | ||
| 1361 | }); | ||
| 1362 | yield* registry.reconcilePlatform([registration("old-bearer")]); | ||
| 1363 | yield* Deferred.await(attemptStarted).pipe(Effect.timeout("1 second")); | ||
| 1364 | |||
| 1365 | // A live state subscription, as the UI holds, keeps the runtime in place | ||
| 1366 | // so the stalled attempt's failure lands on the same supervisor. | ||
| 1367 | yield* registry | ||
| 1368 | .stateChanges(BEARER_TARGET.environmentId) | ||
| 1369 | .pipe(Stream.runDrain, Effect.forkScoped({ startImmediately: true })); | ||
| 1370 | yield* Effect.yieldNow; | ||
| 1371 | yield* registry.reconcilePlatform([registration("new-bearer")]); | ||
| 1372 | yield* Deferred.succeed(releaseFirstAttempt, undefined); | ||
| 1373 | |||
| 1374 | yield* awaitConnectionState( | ||
| 1375 | registry, | ||
| 1376 | BEARER_TARGET.environmentId, | ||
| 1377 | (state) => state.phase === "connected", | ||
| 1378 | ); | ||
| 1379 | expect(yield* Ref.get(attempts)).toBeGreaterThanOrEqual(2); | ||
| 1380 | }).pipe(Effect.provide(harness.layer), Effect.scoped); | ||
| 1381 | }), | ||
| 1382 | ); | ||
| 1383 | |||
| 1273 | it.effect("starts a newly paired bearer environment without re-reading its profile", () => | 1384 | it.effect("starts a newly paired bearer environment without re-reading its profile", () => |
| 1274 | Effect.gen(function* () { | 1385 | Effect.gen(function* () { |
| 1275 | const harness = yield* makeHarness([]); | 1386 | const harness = yield* makeHarness([]); |
packages/client-runtime/src/connection/registry.ts+41-8| ... | @@ -15,6 +15,7 @@ import * as SubscriptionRef from "effect/SubscriptionRef"; | ... | @@ -15,6 +15,7 @@ import * as SubscriptionRef from "effect/SubscriptionRef"; |
| 15 | import * as ClientCapabilities from "../platform/capabilities.ts"; | 15 | import * as ClientCapabilities from "../platform/capabilities.ts"; |
| 16 | import { | 16 | import { |
| 17 | type ConnectionCatalogEntry, | 17 | type ConnectionCatalogEntry, |
| 18 | type ConnectionCredential, | ||
| 18 | type ConnectionProfile, | 19 | type ConnectionProfile, |
| 19 | type ConnectionRegistration, | 20 | type ConnectionRegistration, |
| 20 | type ConnectionRoute, | 21 | type ConnectionRoute, |
| ... | @@ -654,15 +655,26 @@ export const make = Effect.gen(function* () { | ... | @@ -654,15 +655,26 @@ export const make = Effect.gen(function* () { |
| 654 | // on their own loopback origin, so they authenticate with a bearer | 655 | // on their own loopback origin, so they authenticate with a bearer |
| 655 | // token instead of the primary's same-origin cookie. Stash it where | 656 | // token instead of the primary's same-origin cookie. Stash it where |
| 656 | // the resolver's bearer broker looks it up. | 657 | // the resolver's bearer broker looks it up. |
| 658 | let bearerReplaced = false; | ||
| 657 | if (registration._tag === "BearerConnectionRegistration") { | 659 | if (registration._tag === "BearerConnectionRegistration") { |
| 658 | yield* credentials.put(registration.target.connectionId, registration.credential).pipe( | 660 | const stored = yield* credentials |
| 659 | Effect.catch((error) => | 661 | .get(registration.target.connectionId) |
| 660 | Effect.logWarning("Could not store the platform bearer credential.", { | 662 | .pipe(Effect.orElseSucceed(() => Option.none<ConnectionCredential>())); |
| 661 | environmentId: target.environmentId, | 663 | const changed = |
| 662 | error, | 664 | Option.isSome(stored) && !Equal.equals(stored.value, registration.credential); |
| 663 | }), | 665 | // Only a bearer that was actually stored can be retried with; a |
| 664 | ), | 666 | // failed write leaves the rejected one in place. |
| 665 | ); | 667 | bearerReplaced = yield* credentials |
| 668 | .put(registration.target.connectionId, registration.credential) | ||
| 669 | .pipe( | ||
| 670 | Effect.as(changed), | ||
| 671 | Effect.catch((error) => | ||
| 672 | Effect.logWarning("Could not store the platform bearer credential.", { | ||
| 673 | environmentId: target.environmentId, | ||
| 674 | error, | ||
| 675 | }).pipe(Effect.as(false)), | ||
| 676 | ), | ||
| 677 | ); | ||
| 666 | } | 678 | } |
| 667 | 679 | ||
| 668 | if (persisted) { | 680 | if (persisted) { |
| ... | @@ -687,6 +699,27 @@ export const make = Effect.gen(function* () { | ... | @@ -687,6 +699,27 @@ export const make = Effect.gen(function* () { |
| 687 | } | 699 | } |
| 688 | 700 | ||
| 689 | yield* installEntryLocked(entry, { retainEquivalentRuntime: true }); | 701 | yield* installEntryLocked(entry, { retainEquivalentRuntime: true }); |
| 702 | |||
| 703 | // The catalog entry carries no credential, so a fresh bearer keeps | ||
| 704 | // the equivalent runtime. Anything not yet connected may be using the | ||
| 705 | // old bearer: an attempt in flight would fail with it and then wait | ||
| 706 | // blocked for a manual retry. Retry with the new one instead. A | ||
| 707 | // connected session keeps its socket. | ||
| 708 | if (bearerReplaced) { | ||
| 709 | const scope = (yield* SubscriptionRef.get(serviceScopes)).get(target.environmentId); | ||
| 710 | if (scope !== undefined) { | ||
| 711 | const state = yield* SubscriptionRef.get(scope.supervisor.state); | ||
| 712 | if ( | ||
| 713 | state.phase === "connecting" || | ||
| 714 | state.phase === "backoff" || | ||
| 715 | (state.phase === "blocked" && | ||
| 716 | state.lastFailure?._tag === "ConnectionBlockedError" && | ||
| 717 | state.lastFailure.reason === "authentication") | ||
| 718 | ) { | ||
| 719 | yield* scope.supervisor.retryNow; | ||
| 720 | } | ||
| 721 | } | ||
| 722 | } | ||
| 690 | }), | 723 | }), |
| 691 | ); | 724 | ); |
| 692 | }, | 725 | }, |