| ... | @@ -60,16 +60,24 @@ interface TracedAttemptFailure { | ... | @@ -60,16 +60,24 @@ interface TracedAttemptFailure { |
| 60 | readonly attemptSpan: Option.Option<Tracer.Span>; | 60 | readonly attemptSpan: Option.Option<Tracer.Span>; |
| 61 | } | 61 | } |
| 62 | | 62 | |
| | 63 | interface ScopedConnection { |
| | 64 | readonly attemptSpan: Option.Option<Tracer.Span>; |
| | 65 | readonly lease: ConnectionDriver.EnvironmentConnectionLease; |
| | 66 | readonly scope: Scope.Closeable; |
| | 67 | } |
| | 68 | |
| 63 | type AttemptOutcome = | 69 | type AttemptOutcome = |
| 64 | | { | 70 | | { |
| 65 | readonly _tag: "Interrupted"; | 71 | readonly _tag: "Interrupted"; |
| 66 | readonly established: boolean; | 72 | readonly established: boolean; |
| | 73 | readonly generation: number; |
| 67 | readonly stable: boolean; | 74 | readonly stable: boolean; |
| 68 | readonly resetRetry: boolean; | 75 | readonly resetRetry: boolean; |
| 69 | } | 76 | } |
| 70 | | { | 77 | | { |
| 71 | readonly _tag: "Failure"; | 78 | readonly _tag: "Failure"; |
| 72 | readonly established: boolean; | 79 | readonly established: boolean; |
| | 80 | readonly generation: number; |
| 73 | readonly stable: boolean; | 81 | readonly stable: boolean; |
| 74 | readonly failure: TracedAttemptFailure; | 82 | readonly failure: TracedAttemptFailure; |
| 75 | }; | 83 | }; |
| ... | @@ -77,17 +85,26 @@ type AttemptOutcome = | ... | @@ -77,17 +85,26 @@ type AttemptOutcome = |
| 77 | type EstablishmentEvent = | 85 | type EstablishmentEvent = |
| 78 | | { | 86 | | { |
| 79 | readonly _tag: "Completed"; | 87 | readonly _tag: "Completed"; |
| 80 | readonly exit: Exit.Exit< | 88 | readonly exit: Exit.Exit<ScopedConnection, TracedAttemptFailure>; |
| 81 | { | | |
| 82 | readonly attemptSpan: Option.Option<Tracer.Span>; | | |
| 83 | readonly lease: ConnectionDriver.EnvironmentConnectionLease; | | |
| 84 | }, | | |
| 85 | TracedAttemptFailure | | |
| 86 | >; | | |
| 87 | } | 89 | } |
| 88 | | { readonly _tag: "Interrupted"; readonly resetRetry: boolean } | 90 | | { readonly _tag: "Interrupted"; readonly resetRetry: boolean } |
| 89 | | { readonly _tag: "TimedOut" }; | 91 | | { readonly _tag: "TimedOut" }; |
| 90 | | 92 | |
| | 93 | type ReplacementPreparationEvent = |
| | 94 | | { readonly _tag: "Completed"; readonly exit: Exit.Exit<ScopedConnection, TracedAttemptFailure> } |
| | 95 | | { readonly _tag: "TimedOut" } |
| | 96 | | { readonly _tag: "AuthorizationExpired" }; |
| | 97 | |
| | 98 | type ConnectedLeaseEvent = |
| | 99 | | { |
| | 100 | readonly _tag: "ActiveCompleted"; |
| | 101 | readonly exit: Exit.Exit<boolean, TracedAttemptFailure>; |
| | 102 | } |
| | 103 | | { |
| | 104 | readonly _tag: "ReplacementCompleted"; |
| | 105 | readonly exit: Exit.Exit<Option.Option<ScopedConnection>, TracedAttemptFailure>; |
| | 106 | }; |
| | 107 | |
| 91 | function exitUnlessInterrupted<A, E, R>( | 108 | function exitUnlessInterrupted<A, E, R>( |
| 92 | effect: Effect.Effect<A, E, R>, | 109 | effect: Effect.Effect<A, E, R>, |
| 93 | ): Effect.Effect<Exit.Exit<A, E>, never, R> { | 110 | ): Effect.Effect<Exit.Exit<A, E>, never, R> { |
| ... | @@ -168,16 +185,18 @@ function failureFromExit<A>( | ... | @@ -168,16 +185,18 @@ function failureFromExit<A>( |
| 168 | target: ConnectionTarget, | 185 | target: ConnectionTarget, |
| 169 | exit: Exit.Exit<A, TracedAttemptFailure>, | 186 | exit: Exit.Exit<A, TracedAttemptFailure>, |
| 170 | established: boolean, | 187 | established: boolean, |
| | 188 | generation: number, |
| 171 | stable: boolean, | 189 | stable: boolean, |
| 172 | ): AttemptOutcome { | 190 | ): AttemptOutcome { |
| 173 | if (Exit.isSuccess(exit) || Cause.hasInterruptsOnly(exit.cause)) { | 191 | if (Exit.isSuccess(exit) || Cause.hasInterruptsOnly(exit.cause)) { |
| 174 | return { _tag: "Interrupted", established, stable, resetRetry: false }; | 192 | return { _tag: "Interrupted", established, generation, stable, resetRetry: false }; |
| 175 | } | 193 | } |
| 176 | const typedFailure = exit.cause.reasons.find(Cause.isFailReason); | 194 | const typedFailure = exit.cause.reasons.find(Cause.isFailReason); |
| 177 | if (typedFailure) { | 195 | if (typedFailure) { |
| 178 | return { | 196 | return { |
| 179 | _tag: "Failure", | 197 | _tag: "Failure", |
| 180 | established, | 198 | established, |
| | 199 | generation, |
| 181 | stable, | 200 | stable, |
| 182 | failure: typedFailure.error, | 201 | failure: typedFailure.error, |
| 183 | }; | 202 | }; |
| ... | @@ -185,6 +204,7 @@ function failureFromExit<A>( | ... | @@ -185,6 +204,7 @@ function failureFromExit<A>( |
| 185 | return { | 204 | return { |
| 186 | _tag: "Failure", | 205 | _tag: "Failure", |
| 187 | established, | 206 | established, |
| | 207 | generation, |
| 188 | stable, | 208 | stable, |
| 189 | failure: { | 209 | failure: { |
| 190 | error: new ConnectionTransientError({ | 210 | error: new ConnectionTransientError({ |
| ... | @@ -289,9 +309,10 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( | ... | @@ -289,9 +309,10 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( |
| 289 | attempt: number, | 309 | attempt: number, |
| 290 | generation: number, | 310 | generation: number, |
| 291 | lastFailure: ConnectionAttemptError | null, | 311 | lastFailure: ConnectionAttemptError | null, |
| | 312 | publishProgress: boolean, |
| 292 | ) { | 313 | ) { |
| 293 | return yield* driver.connect(entry, (progress) => | 314 | return yield* driver.connect(entry, (progress) => |
| 294 | reportProgress(attempt, generation, lastFailure, progress), | 315 | publishProgress ? reportProgress(attempt, generation, lastFailure, progress) : Effect.void, |
| 295 | ); | 316 | ); |
| 296 | }); | 317 | }); |
| 297 | | 318 | |
| ... | @@ -342,16 +363,17 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( | ... | @@ -342,16 +363,17 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( |
| 342 | generation: number, | 363 | generation: number, |
| 343 | lastFailure: ConnectionAttemptError | null, | 364 | lastFailure: ConnectionAttemptError | null, |
| 344 | pendingRetry: Option.Option<PendingRetryTrace>, | 365 | pendingRetry: Option.Option<PendingRetryTrace>, |
| | 366 | publishProgress: boolean, |
| 345 | ) { | 367 | ) { |
| 346 | if (target._tag === "RelayConnectionTarget") { | 368 | if (target._tag === "RelayConnectionTarget") { |
| 347 | return yield* traceRelayEstablishment( | 369 | return yield* traceRelayEstablishment( |
| 348 | establishConnection(attempt, generation, lastFailure), | 370 | establishConnection(attempt, generation, lastFailure, publishProgress), |
| 349 | attempt, | 371 | attempt, |
| 350 | generation, | 372 | generation, |
| 351 | pendingRetry, | 373 | pendingRetry, |
| 352 | ); | 374 | ); |
| 353 | } | 375 | } |
| 354 | return yield* establishConnection(attempt, generation, lastFailure).pipe( | 376 | return yield* establishConnection(attempt, generation, lastFailure, publishProgress).pipe( |
| 355 | Effect.map((lease) => ({ | 377 | Effect.map((lease) => ({ |
| 356 | attemptSpan: Option.none<Tracer.Span>(), | 378 | attemptSpan: Option.none<Tracer.Span>(), |
| 357 | lease, | 379 | lease, |
| ... | @@ -363,6 +385,37 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( | ... | @@ -363,6 +385,37 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( |
| 363 | ); | 385 | ); |
| 364 | }); | 386 | }); |
| 365 | | 387 | |
| | 388 | const forkScopedTracedConnection = Effect.fnUntraced(function* ( |
| | 389 | attempt: number, |
| | 390 | generation: number, |
| | 391 | lastFailure: ConnectionAttemptError | null, |
| | 392 | pendingRetry: Option.Option<PendingRetryTrace>, |
| | 393 | publishProgress: boolean, |
| | 394 | ) { |
| | 395 | const parentScope = yield* Scope.Scope; |
| | 396 | return yield* Effect.uninterruptibleMask((restore) => |
| | 397 | Effect.gen(function* () { |
| | 398 | const connectionScope = yield* Scope.fork(parentScope, "sequential"); |
| | 399 | const fiber = yield* restore( |
| | 400 | establishTracedConnection( |
| | 401 | attempt, |
| | 402 | generation, |
| | 403 | lastFailure, |
| | 404 | pendingRetry, |
| | 405 | publishProgress, |
| | 406 | ).pipe( |
| | 407 | Scope.provide(connectionScope), |
| | 408 | Effect.map( |
| | 409 | (established) => |
| | 410 | ({ ...established, scope: connectionScope }) satisfies ScopedConnection, |
| | 411 | ), |
| | 412 | ), |
| | 413 | ).pipe(Effect.forkChild); |
| | 414 | return { fiber, scope: connectionScope }; |
| | 415 | }), |
| | 416 | ); |
| | 417 | }); |
| | 418 | |
| 366 | const waitForEstablishmentInterrupt = Effect.fnUntraced(function* () { | 419 | const waitForEstablishmentInterrupt = Effect.fnUntraced(function* () { |
| 367 | for (;;) { | 420 | for (;;) { |
| 368 | const next = yield* Queue.take(signals); | 421 | const next = yield* Queue.take(signals); |
| ... | @@ -484,32 +537,136 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( | ... | @@ -484,32 +537,136 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( |
| 484 | } | 537 | } |
| 485 | }); | 538 | }); |
| 486 | | 539 | |
| 487 | const waitForAuthorizationRefresh = Effect.fnUntraced(function* ( | 540 | const waitForAuthorizationDeadline = Effect.fnUntraced(function* ( |
| 488 | preparedConnection: PreparedConnection, | 541 | preparedConnection: PreparedConnection, |
| | 542 | skewMs: number, |
| 489 | ) { | 543 | ) { |
| 490 | const authorization = preparedConnection.httpAuthorization; | 544 | const authorization = preparedConnection.httpAuthorization; |
| 491 | if (authorization?._tag !== "Dpop") { | 545 | if (authorization?._tag !== "Dpop") { |
| 492 | return yield* Effect.never; | 546 | return yield* Effect.never; |
| 493 | } | 547 | } |
| 494 | const now = yield* Clock.currentTimeMillis; | 548 | const now = yield* Clock.currentTimeMillis; |
| 495 | yield* Effect.sleep( | 549 | yield* Effect.sleep(Math.max(0, authorization.expiresAtEpochMs - now - skewMs)); |
| 496 | Math.max(0, authorization.expiresAtEpochMs - now - DPOP_ACCESS_TOKEN_REFRESH_SKEW_MS), | 550 | }); |
| | 551 | |
| | 552 | const waitForActiveCompletion = Effect.fnUntraced(function* (active: ScopedConnection) { |
| | 553 | const exit = yield* exitUnlessInterrupted( |
| | 554 | Effect.raceAllFirst([ |
| | 555 | active.lease.session.closed.pipe( |
| | 556 | Effect.mapError((error): TracedAttemptFailure => ({ |
| | 557 | error, |
| | 558 | attemptSpan: active.attemptSpan, |
| | 559 | })), |
| | 560 | ), |
| | 561 | monitorConnectedLease(active.lease).pipe( |
| | 562 | Effect.mapError((error): TracedAttemptFailure => ({ |
| | 563 | error, |
| | 564 | attemptSpan: active.attemptSpan, |
| | 565 | })), |
| | 566 | ), |
| | 567 | ]), |
| 497 | ); | 568 | ); |
| 498 | yield* Effect.logDebug("Refreshing the environment connection before its DPoP token expires."); | 569 | return { _tag: "ActiveCompleted", exit } satisfies ConnectedLeaseEvent; |
| 499 | return true; | 570 | }); |
| | 571 | |
| | 572 | const prepareReplacement = Effect.fnUntraced(function* ( |
| | 573 | active: ScopedConnection, |
| | 574 | generation: number, |
| | 575 | ) { |
| | 576 | yield* waitForAuthorizationDeadline(active.lease.prepared, DPOP_ACCESS_TOKEN_REFRESH_SKEW_MS); |
| | 577 | yield* Effect.logDebug( |
| | 578 | "Preparing a replacement environment connection before its DPoP token expires.", |
| | 579 | ); |
| | 580 | |
| | 581 | let failureCount = 0; |
| | 582 | for (;;) { |
| | 583 | const candidate = yield* forkScopedTracedConnection( |
| | 584 | failureCount + 1, |
| | 585 | generation, |
| | 586 | null, |
| | 587 | Option.none(), |
| | 588 | false, |
| | 589 | ); |
| | 590 | const replacement = yield* Effect.raceAllFirst([ |
| | 591 | Fiber.await(candidate.fiber).pipe( |
| | 592 | Effect.map((exit): ReplacementPreparationEvent => ({ _tag: "Completed", exit })), |
| | 593 | ), |
| | 594 | waitForAuthorizationDeadline(active.lease.prepared, 0).pipe( |
| | 595 | Effect.as<ReplacementPreparationEvent>({ _tag: "AuthorizationExpired" }), |
| | 596 | ), |
| | 597 | Effect.sleep(CONNECTION_ESTABLISHMENT_TIMEOUT).pipe( |
| | 598 | Effect.as<ReplacementPreparationEvent>({ _tag: "TimedOut" }), |
| | 599 | ), |
| | 600 | ]); |
| | 601 | |
| | 602 | if (replacement._tag !== "Completed") { |
| | 603 | yield* Fiber.interrupt(candidate.fiber); |
| | 604 | yield* Fiber.await(candidate.fiber); |
| | 605 | yield* Scope.close(candidate.scope, Exit.void).pipe(Effect.ignore); |
| | 606 | } else if (Exit.isFailure(replacement.exit)) { |
| | 607 | yield* Scope.close(candidate.scope, Exit.void).pipe(Effect.ignore); |
| | 608 | } |
| | 609 | if (replacement._tag === "AuthorizationExpired") { |
| | 610 | return Option.none<ScopedConnection>(); |
| | 611 | } |
| | 612 | if (replacement._tag === "Completed" && Exit.isSuccess(replacement.exit)) { |
| | 613 | return Option.some(replacement.exit.value); |
| | 614 | } |
| | 615 | |
| | 616 | let replacementError: ConnectionTransientError; |
| | 617 | if (replacement._tag === "Completed") { |
| | 618 | if (Exit.isSuccess(replacement.exit)) { |
| | 619 | return yield* Effect.die("A successful replacement was not installed."); |
| | 620 | } |
| | 621 | const failure = Cause.findErrorOption(replacement.exit.cause); |
| | 622 | if (Option.isNone(failure) || failure.value.error._tag === "ConnectionBlockedError") { |
| | 623 | return yield* Effect.failCause(replacement.exit.cause); |
| | 624 | } |
| | 625 | replacementError = failure.value.error; |
| | 626 | } else { |
| | 627 | replacementError = new ConnectionTransientError({ |
| | 628 | reason: "timeout", |
| | 629 | detail: `${target.label} did not respond during connection setup.`, |
| | 630 | }); |
| | 631 | } |
| | 632 | |
| | 633 | const retryDelay = retryDelayMs(failureCount); |
| | 634 | failureCount += 1; |
| | 635 | yield* Effect.logWarning( |
| | 636 | "Could not prepare a replacement environment connection; keeping the active connection.", |
| | 637 | ).pipe( |
| | 638 | Effect.annotateLogs({ |
| | 639 | "authorization.refresh.retry_delay_ms": retryDelay, |
| | 640 | ...safeErrorLogAttributes(replacementError), |
| | 641 | }), |
| | 642 | ); |
| | 643 | const retryBeforeExpiry = yield* Effect.raceFirst( |
| | 644 | Effect.sleep(retryDelay).pipe(Effect.as(true)), |
| | 645 | waitForAuthorizationDeadline(active.lease.prepared, 0).pipe(Effect.as(false)), |
| | 646 | ); |
| | 647 | if (!retryBeforeExpiry) { |
| | 648 | return Option.none<ScopedConnection>(); |
| | 649 | } |
| | 650 | } |
| 500 | }); | 651 | }); |
| 501 | | 652 | |
| 502 | const runAttempt = Effect.fnUntraced(function* ( | 653 | const runAttempt = Effect.fnUntraced(function* ( |
| 503 | attempt: number, | 654 | attempt: number, |
| 504 | generation: number, | 655 | previousGeneration: number, |
| 505 | lastFailure: ConnectionAttemptError | null, | 656 | lastFailure: ConnectionAttemptError | null, |
| 506 | pendingRetry: Option.Option<PendingRetryTrace>, | 657 | pendingRetry: Option.Option<PendingRetryTrace>, |
| 507 | ) { | 658 | ) { |
| | 659 | const initialGeneration = previousGeneration + 1; |
| 508 | yield* SubscriptionRef.set(prepared, Option.none()); | 660 | yield* SubscriptionRef.set(prepared, Option.none()); |
| | 661 | const initial = yield* forkScopedTracedConnection( |
| | 662 | attempt, |
| | 663 | initialGeneration, |
| | 664 | lastFailure, |
| | 665 | pendingRetry, |
| | 666 | true, |
| | 667 | ); |
| 509 | const establishment = yield* Effect.raceAllFirst([ | 668 | const establishment = yield* Effect.raceAllFirst([ |
| 510 | exitUnlessInterrupted( | 669 | Fiber.await(initial.fiber).pipe( |
| 511 | establishTracedConnection(attempt, generation, lastFailure, pendingRetry), | | |
| 512 | ).pipe( | | |
| 513 | Effect.map((exit): EstablishmentEvent => ({ | 670 | Effect.map((exit): EstablishmentEvent => ({ |
| 514 | _tag: "Completed", | 671 | _tag: "Completed", |
| 515 | exit, | 672 | exit, |
| ... | @@ -526,10 +683,18 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( | ... | @@ -526,10 +683,18 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( |
| 526 | ), | 683 | ), |
| 527 | ]); | 684 | ]); |
| 528 | | 685 | |
| | 686 | if (establishment._tag !== "Completed") { |
| | 687 | yield* Fiber.interrupt(initial.fiber); |
| | 688 | yield* Fiber.await(initial.fiber); |
| | 689 | yield* Scope.close(initial.scope, Exit.void).pipe(Effect.ignore); |
| | 690 | } else if (Exit.isFailure(establishment.exit)) { |
| | 691 | yield* Scope.close(initial.scope, Exit.void).pipe(Effect.ignore); |
| | 692 | } |
| 529 | if (establishment._tag === "Interrupted") { | 693 | if (establishment._tag === "Interrupted") { |
| 530 | return { | 694 | return { |
| 531 | _tag: "Interrupted", | 695 | _tag: "Interrupted", |
| 532 | established: false, | 696 | established: false, |
| | 697 | generation: previousGeneration, |
| 533 | stable: false, | 698 | stable: false, |
| 534 | resetRetry: establishment.resetRetry, | 699 | resetRetry: establishment.resetRetry, |
| 535 | } satisfies AttemptOutcome; | 700 | } satisfies AttemptOutcome; |
| ... | @@ -538,6 +703,7 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( | ... | @@ -538,6 +703,7 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( |
| 538 | return { | 703 | return { |
| 539 | _tag: "Failure", | 704 | _tag: "Failure", |
| 540 | established: false, | 705 | established: false, |
| | 706 | generation: previousGeneration, |
| 541 | stable: false, | 707 | stable: false, |
| 542 | failure: { | 708 | failure: { |
| 543 | error: new ConnectionTransientError({ | 709 | error: new ConnectionTransientError({ |
| ... | @@ -552,7 +718,7 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( | ... | @@ -552,7 +718,7 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( |
| 552 | const isUnexpectedDefect = | 718 | const isUnexpectedDefect = |
| 553 | !Cause.hasInterruptsOnly(establishment.exit.cause) && | 719 | !Cause.hasInterruptsOnly(establishment.exit.cause) && |
| 554 | !establishment.exit.cause.reasons.some(Cause.isFailReason); | 720 | !establishment.exit.cause.reasons.some(Cause.isFailReason); |
| 555 | const outcome = failureFromExit(target, establishment.exit, false, false); | 721 | const outcome = failureFromExit(target, establishment.exit, false, previousGeneration, false); |
| 556 | if (isUnexpectedDefect) { | 722 | if (isUnexpectedDefect) { |
| 557 | const defect = establishment.exit.cause.reasons.find(Cause.isDieReason)?.defect; | 723 | const defect = establishment.exit.cause.reasons.find(Cause.isDieReason)?.defect; |
| 558 | yield* Effect.logError("Connection attempt failed with an unexpected defect.").pipe( | 724 | yield* Effect.logError("Connection attempt failed with an unexpected defect.").pipe( |
| ... | @@ -567,18 +733,20 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( | ... | @@ -567,18 +733,20 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( |
| 567 | return outcome; | 733 | return outcome; |
| 568 | } | 734 | } |
| 569 | | 735 | |
| 570 | const active = establishment.exit.value; | | |
| 571 | const currentIntent = yield* Ref.get(intent); | 736 | const currentIntent = yield* Ref.get(intent); |
| 572 | if (!currentIntent.desired || currentIntent.network === "offline") { | 737 | if (!currentIntent.desired || currentIntent.network === "offline") { |
| 573 | return { | 738 | return { |
| 574 | _tag: "Interrupted", | 739 | _tag: "Interrupted", |
| 575 | established: false, | 740 | established: false, |
| | 741 | generation: previousGeneration, |
| 576 | stable: false, | 742 | stable: false, |
| 577 | resetRetry: false, | 743 | resetRetry: false, |
| 578 | } satisfies AttemptOutcome; | 744 | } satisfies AttemptOutcome; |
| 579 | } | 745 | } |
| 580 | | 746 | |
| 581 | const connectedAt = yield* Clock.currentTimeMillis; | 747 | const connectedAt = yield* Clock.currentTimeMillis; |
| | 748 | let active = establishment.exit.value; |
| | 749 | let activeGeneration = initialGeneration; |
| 582 | yield* SubscriptionRef.set(prepared, Option.some(active.lease.prepared)); | 750 | yield* SubscriptionRef.set(prepared, Option.some(active.lease.prepared)); |
| 583 | yield* SubscriptionRef.set(session, Option.some(active.lease.session)); | 751 | yield* SubscriptionRef.set(session, Option.some(active.lease.session)); |
| 584 | yield* setState({ | 752 | yield* setState({ |
| ... | @@ -587,36 +755,78 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( | ... | @@ -587,36 +755,78 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( |
| 587 | phase: "connected", | 755 | phase: "connected", |
| 588 | stage: null, | 756 | stage: null, |
| 589 | attempt, | 757 | attempt, |
| 590 | generation, | 758 | generation: activeGeneration, |
| 591 | lastFailure: null, | 759 | lastFailure: null, |
| 592 | retryAt: null, | 760 | retryAt: null, |
| 593 | }); | 761 | }); |
| 594 | | 762 | |
| 595 | const connectedExit = yield* Effect.raceAllFirst([ | 763 | for (;;) { |
| 596 | active.lease.session.closed.pipe( | 764 | const connectedEvent = yield* Effect.raceAllFirst([ |
| 597 | Effect.mapError((error): TracedAttemptFailure => ({ | 765 | waitForActiveCompletion(active), |
| 598 | error, | 766 | exitUnlessInterrupted(prepareReplacement(active, activeGeneration + 1)).pipe( |
| 599 | attemptSpan: active.attemptSpan, | 767 | Effect.map((exit): ConnectedLeaseEvent => ({ _tag: "ReplacementCompleted", exit })), |
| 600 | })), | 768 | ), |
| 601 | ), | 769 | ]); |
| 602 | monitorConnectedLease(active.lease).pipe( | 770 | const stable = (yield* Clock.currentTimeMillis) - connectedAt >= BACKOFF_RESET_AFTER_MS; |
| 603 | Effect.mapError((error): TracedAttemptFailure => ({ | 771 | if (connectedEvent._tag === "ActiveCompleted") { |
| 604 | error, | 772 | if (Exit.isSuccess(connectedEvent.exit)) { |
| 605 | attemptSpan: active.attemptSpan, | 773 | return { |
| 606 | })), | 774 | _tag: "Interrupted", |
| 607 | ), | 775 | established: true, |
| 608 | waitForAuthorizationRefresh(active.lease.prepared), | 776 | generation: activeGeneration, |
| 609 | ]).pipe(exitUnlessInterrupted); | 777 | stable, |
| 610 | const connectedForMs = (yield* Clock.currentTimeMillis) - connectedAt; | 778 | resetRetry: connectedEvent.exit.value, |
| 611 | if (Exit.isSuccess(connectedExit)) { | 779 | } satisfies AttemptOutcome; |
| 612 | return { | 780 | } |
| 613 | _tag: "Interrupted", | 781 | return failureFromExit(target, connectedEvent.exit, true, activeGeneration, stable); |
| 614 | established: true, | 782 | } |
| 615 | stable: connectedForMs >= BACKOFF_RESET_AFTER_MS, | 783 | if (Exit.isFailure(connectedEvent.exit)) { |
| 616 | resetRetry: connectedExit.value, | 784 | return failureFromExit(target, connectedEvent.exit, true, activeGeneration, stable); |
| 617 | } satisfies AttemptOutcome; | 785 | } |
| | 786 | if (Option.isNone(connectedEvent.exit.value)) { |
| | 787 | return { |
| | 788 | _tag: "Interrupted", |
| | 789 | established: true, |
| | 790 | generation: activeGeneration, |
| | 791 | stable, |
| | 792 | resetRetry: true, |
| | 793 | } satisfies AttemptOutcome; |
| | 794 | } |
| | 795 | |
| | 796 | const candidate = connectedEvent.exit.value.value; |
| | 797 | const replacementIntent = yield* Ref.get(intent); |
| | 798 | if (!replacementIntent.desired || replacementIntent.network === "offline") { |
| | 799 | yield* Scope.close(candidate.scope, Exit.void).pipe(Effect.ignore); |
| | 800 | return { |
| | 801 | _tag: "Interrupted", |
| | 802 | established: true, |
| | 803 | generation: activeGeneration, |
| | 804 | stable, |
| | 805 | resetRetry: false, |
| | 806 | } satisfies AttemptOutcome; |
| | 807 | } |
| | 808 | |
| | 809 | const previous = active; |
| | 810 | yield* Effect.uninterruptible( |
| | 811 | Effect.gen(function* () { |
| | 812 | active = candidate; |
| | 813 | activeGeneration += 1; |
| | 814 | yield* SubscriptionRef.set(prepared, Option.some(active.lease.prepared)); |
| | 815 | yield* SubscriptionRef.set(session, Option.some(active.lease.session)); |
| | 816 | yield* setState({ |
| | 817 | desired: true, |
| | 818 | network: replacementIntent.network, |
| | 819 | phase: "connected", |
| | 820 | stage: null, |
| | 821 | attempt: 1, |
| | 822 | generation: activeGeneration, |
| | 823 | lastFailure: null, |
| | 824 | retryAt: null, |
| | 825 | }); |
| | 826 | yield* Scope.close(previous.scope, Exit.void).pipe(Effect.ignore); |
| | 827 | }), |
| | 828 | ); |
| 618 | } | 829 | } |
| 619 | return failureFromExit(target, connectedExit, true, connectedForMs >= BACKOFF_RESET_AFTER_MS); | | |
| 620 | }, Effect.ensuring(clearLease)); | 830 | }, Effect.ensuring(clearLease)); |
| 621 | | 831 | |
| 622 | const waitForRetrySignal = Effect.fnUntraced(function* (delayMs: number) { | 832 | const waitForRetrySignal = Effect.fnUntraced(function* (delayMs: number) { |
| ... | @@ -681,15 +891,14 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( | ... | @@ -681,15 +891,14 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( |
| 681 | } | 891 | } |
| 682 | | 892 | |
| 683 | const attempt = failureCount + 1; | 893 | const attempt = failureCount + 1; |
| 684 | const nextGeneration = generation + 1; | | |
| 685 | const outcome: AttemptOutcome = yield* Effect.scoped( | 894 | const outcome: AttemptOutcome = yield* Effect.scoped( |
| 686 | runAttempt(attempt, nextGeneration, latestFailure, pendingRetry), | 895 | runAttempt(attempt, generation, latestFailure, pendingRetry), |
| 687 | ); | 896 | ); |
| 688 | // Consumed on every iteration so a stale marker can never leak into a | 897 | // Consumed on every iteration so a stale marker can never leak into a |
| 689 | // later, unrelated failure. | 898 | // later, unrelated failure. |
| 690 | const failedWakeProbe = yield* Ref.getAndSet(wakeProbeFailed, false); | 899 | const failedWakeProbe = yield* Ref.getAndSet(wakeProbeFailed, false); |
| | 900 | generation = outcome.generation; |
| 691 | if (outcome.established) { | 901 | if (outcome.established) { |
| 692 | generation = nextGeneration; | | |
| 693 | if (outcome.stable) { | 902 | if (outcome.stable) { |
| 694 | resetRetryLadder(); | 903 | resetRetryLadder(); |
| 695 | latestFailure = null; | 904 | latestFailure = null; |
maria <maria@kuuro.net> 2026-09-04 01:07:49-04:00
GitHub <noreply@github.com> 2026-09-04 01:07:49-04:00