| ... | ... | @@ -39,8 +39,15 @@ import { |
| 39 | 39 | |
| 40 | 40 | const RETRY_BASE_DELAY_MS = 1_000; |
| 41 | 41 | const RETRY_MAX_DELAY_MS = 300_000; |
| 42 | // Per-route budget to resolve credentials and create the session, which ends |
| 43 | // when the driver reports the synchronizing stage. Each route that gets that |
| 44 | // far adds its own budget for the socket to open (bounded at 15 seconds by the |
| 45 | // RPC session) and the server to deliver the first config snapshot. A cold |
| 46 | // server with a large database can take longer than 15 seconds to answer, and |
| 47 | // that is not a reason to drop a working connection. |
| 42 | 48 | const CONNECTION_ESTABLISHMENT_TIMEOUT = "15 seconds"; |
| 43 | 49 | const establishmentTimeout = Duration.fromInputUnsafe(CONNECTION_ESTABLISHMENT_TIMEOUT); |
| 50 | const synchronizationTimeout = Duration.fromInputUnsafe("75 seconds"); |
| 44 | 51 | const CONNECTION_PROBE_TIMEOUT = "15 seconds"; |
| 45 | 52 | // Mobile resumes, explicit retries, and offline events want a fast answer: |
| 46 | 53 | // the user is waiting, or the network may be gone. |
| ... | ... | @@ -428,9 +435,13 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( |
| 428 | 435 | attempt: number, |
| 429 | 436 | generation: number, |
| 430 | 437 | lastFailure: ConnectionAttemptError | null, |
| 438 | synchronizedRoutes: Ref.Ref<number>, |
| 431 | 439 | ) { |
| 432 | 440 | return yield* driver.connect(yield* attemptEntry, (progress) => |
| 433 | | reportProgress(attempt, generation, lastFailure, progress), |
| 441 | (progress.stage === "synchronizing" |
| 442 | ? Ref.update(synchronizedRoutes, (count) => count + 1) |
| 443 | : Effect.void |
| 444 | ).pipe(Effect.andThen(reportProgress(attempt, generation, lastFailure, progress))), |
| 434 | 445 | ); |
| 435 | 446 | }); |
| 436 | 447 | |
| ... | ... | @@ -481,16 +492,17 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( |
| 481 | 492 | generation: number, |
| 482 | 493 | lastFailure: ConnectionAttemptError | null, |
| 483 | 494 | pendingRetry: Option.Option<PendingRetryTrace>, |
| 495 | synchronizedRoutes: Ref.Ref<number>, |
| 484 | 496 | ) { |
| 485 | 497 | if (usesRelay) { |
| 486 | 498 | return yield* traceRelayEstablishment( |
| 487 | | establishConnection(attempt, generation, lastFailure), |
| 499 | establishConnection(attempt, generation, lastFailure, synchronizedRoutes), |
| 488 | 500 | attempt, |
| 489 | 501 | generation, |
| 490 | 502 | pendingRetry, |
| 491 | 503 | ); |
| 492 | 504 | } |
| 493 | | return yield* establishConnection(attempt, generation, lastFailure).pipe( |
| 505 | return yield* establishConnection(attempt, generation, lastFailure, synchronizedRoutes).pipe( |
| 494 | 506 | Effect.map((lease) => ({ |
| 495 | 507 | attemptSpan: Option.none<Tracer.Span>(), |
| 496 | 508 | lease, |
| ... | ... | @@ -684,9 +696,16 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( |
| 684 | 696 | ) { |
| 685 | 697 | const switchingTo = yield* Ref.get(preferredRouteId); |
| 686 | 698 | yield* SubscriptionRef.set(prepared, Option.none()); |
| 699 | const synchronizedRoutes = yield* Ref.make(0); |
| 687 | 700 | const establishment = yield* Effect.raceAllFirst([ |
| 688 | 701 | exitUnlessInterrupted( |
| 689 | | establishTracedConnection(attempt, generation, lastFailure, pendingRetry), |
| 702 | establishTracedConnection( |
| 703 | attempt, |
| 704 | generation, |
| 705 | lastFailure, |
| 706 | pendingRetry, |
| 707 | synchronizedRoutes, |
| 708 | ), |
| 690 | 709 | ).pipe( |
| 691 | 710 | Effect.map((exit): EstablishmentEvent => ({ |
| 692 | 711 | _tag: "Completed", |
| ... | ... | @@ -699,10 +718,24 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( |
| 699 | 718 | resetRetry, |
| 700 | 719 | })), |
| 701 | 720 | ), |
| 702 | | // Each route may use the full setup time before the next is tried. |
| 703 | | Effect.sleep( |
| 704 | | Duration.times(establishmentTimeout, connectionRoutes(yield* Ref.get(currentEntry)).length), |
| 705 | | ).pipe(Effect.as<EstablishmentEvent>({ _tag: "TimedOut" })), |
| 721 | // Each route may use the full setup time before the next is tried, and |
| 722 | // each route that reaches synchronizing adds its own snapshot budget, so |
| 723 | // a route that fails late does not eat the next route's time. |
| 724 | Effect.gen(function* () { |
| 725 | const routeCount = connectionRoutes(yield* Ref.get(currentEntry)).length; |
| 726 | const startedAt = yield* Clock.monotonicTimeNanos; |
| 727 | const setup = Duration.toNanosUnsafe(Duration.times(establishmentTimeout, routeCount)); |
| 728 | while (true) { |
| 729 | const synchronized = BigInt(yield* Ref.get(synchronizedRoutes)); |
| 730 | const deadline = |
| 731 | startedAt + setup + synchronized * Duration.toNanosUnsafe(synchronizationTimeout); |
| 732 | const remaining = deadline - (yield* Clock.monotonicTimeNanos); |
| 733 | if (remaining <= 0n) { |
| 734 | return { _tag: "TimedOut" } as const satisfies EstablishmentEvent; |
| 735 | } |
| 736 | yield* Effect.sleep(Duration.nanos(remaining)); |
| 737 | } |
| 738 | }), |
| 706 | 739 | ]); |
| 707 | 740 | |
| 708 | 741 | if (establishment._tag === "Interrupted") { |