|
| 1 | +/* |
| 2 | + * Copyright (c) 2023-2026 - Restate Software, Inc., Restate GmbH |
| 3 | + * |
| 4 | + * This file is part of the Restate SDK for Node.js/TypeScript, |
| 5 | + * which is released under the MIT license. |
| 6 | + * |
| 7 | + * You can find a copy of the license in file LICENSE in the root |
| 8 | + * directory of this repository or package, or at |
| 9 | + * https://github.com/restatedev/sdk-typescript/blob/main/LICENSE |
| 10 | + */ |
| 11 | + |
| 12 | +// The slot supervisor. |
| 13 | +// ============================================================================= |
| 14 | +// |
| 15 | +// Multi-homing — one tunnel connection per resolved tunnel server (like the |
| 16 | +// Rust client; the slot set IS the resolved set, it is not configurable). The |
| 17 | +// supervisor resolves the server set, reconciles the slot map against it |
| 18 | +// (starting connections to servers that appear, tearing down ones that |
| 19 | +// vanish), and re-resolves every `resolveIntervalMs` for SRV discovery. Each |
| 20 | +// slot runs its own reconnect loop with fatal-vs-retryable classification. |
| 21 | +// |
| 22 | +// Invariants: |
| 23 | +// E1. A FATAL outcome (unauthorized / bad-tunnel-name / name mismatch) on |
| 24 | +// ANY slot stops the WHOLE tunnel — the credentials are shared, so every |
| 25 | +// other slot would hit the same wall. It aborts every slot and reports |
| 26 | +// via `hooks.onFatal`; the engine surfaces it on `error`/`ready`. |
| 27 | +// E2. Backoff resets only after a connection held for |
| 28 | +// MIN_UPTIME_FOR_BACKOFF_RESET_MS; a drain only skips the backoff sleep |
| 29 | +// under the same guard (drain-spam must compound). |
| 30 | +// E3. Teardown is prompt: `abortAll()` aborts in-flight dials via per-slot |
| 31 | +// signals and wakes the resolve loop out of any sleep; the (un-abortable) |
| 32 | +// DNS work is raced against the wake signal, never awaited. |
| 33 | + |
| 34 | +import type { ResolvedOptions } from "./options.js"; |
| 35 | +import { resolveTargets, targetKey, type Target } from "./targets.js"; |
| 36 | +import { runConnection, type ConnectionDeps } from "./connection.js"; |
| 37 | +import { Backoff, MIN_UPTIME_FOR_BACKOFF_RESET_MS } from "./backoff.js"; |
| 38 | +import { delay, raceAbortable } from "./util.js"; |
| 39 | + |
| 40 | +/** A running per-server connection loop. */ |
| 41 | +interface Slot { |
| 42 | + ctl: AbortController; |
| 43 | + done: Promise<void>; |
| 44 | +} |
| 45 | + |
| 46 | +export interface SupervisorHooks { |
| 47 | + /** A slot hit a non-retryable failure; the whole tunnel must stop (E1). */ |
| 48 | + onFatal: (err: Error) => void; |
| 49 | +} |
| 50 | + |
| 51 | +export class Supervisor { |
| 52 | + private readonly slots = new Map<string, Slot>(); |
| 53 | + /** Aborting this cascades to every slot (each slot chains its ctl to it). */ |
| 54 | + private readonly stopSignal = new AbortController(); |
| 55 | + /** Wakes the resolve loop out of a sleep / DNS race (E3). */ |
| 56 | + private readonly wake = new AbortController(); |
| 57 | + private stopping = false; |
| 58 | + private fatal: Error | undefined; |
| 59 | + |
| 60 | + /** Resolves when the resolve loop has exited AND every slot has settled. */ |
| 61 | + readonly done: Promise<void>; |
| 62 | + |
| 63 | + constructor( |
| 64 | + private readonly opts: ResolvedOptions, |
| 65 | + private readonly deps: ConnectionDeps, |
| 66 | + private readonly hooks: SupervisorHooks, |
| 67 | + private readonly log: (message: string) => void |
| 68 | + ) { |
| 69 | + this.stopSignal.signal.addEventListener("abort", () => this.wake.abort(), { |
| 70 | + once: true, |
| 71 | + }); |
| 72 | + this.done = this.supervise(); |
| 73 | + } |
| 74 | + |
| 75 | + get fatalError(): Error | undefined { |
| 76 | + return this.fatal; |
| 77 | + } |
| 78 | + |
| 79 | + /** |
| 80 | + * Stop starting/resolving new connections; existing slots keep running so |
| 81 | + * their connections can finish draining in place (client-initiated drain). |
| 82 | + */ |
| 83 | + stopResolving(): void { |
| 84 | + this.stopping = true; |
| 85 | + this.wake.abort(); |
| 86 | + } |
| 87 | + |
| 88 | + /** Abort every slot and the resolve loop (engine teardown). */ |
| 89 | + abortAll(): void { |
| 90 | + this.stopping = true; |
| 91 | + this.stopSignal.abort(); |
| 92 | + } |
| 93 | + |
| 94 | + private startSlot(key: string, target: Target): void { |
| 95 | + const ctl = new AbortController(); |
| 96 | + // Chain to the global stop so abortAll() cascades; self-detaching. |
| 97 | + this.stopSignal.signal.addEventListener("abort", () => ctl.abort(), { |
| 98 | + once: true, |
| 99 | + signal: ctl.signal, |
| 100 | + }); |
| 101 | + const slot: Slot = { ctl, done: Promise.resolve() }; |
| 102 | + slot.done = this.runSlot(target, ctl).finally(() => { |
| 103 | + // Guarded: this key may have vanished and re-appeared, in which case a |
| 104 | + // NEWER slot owns it — don't delete someone else's registration. |
| 105 | + if (this.slots.get(key) === slot) this.slots.delete(key); |
| 106 | + }); |
| 107 | + this.slots.set(key, slot); |
| 108 | + } |
| 109 | + |
| 110 | + /** The per-server loop: dial → serve → classify outcome → backoff → redial. */ |
| 111 | + private async runSlot(target: Target, ctl: AbortController): Promise<void> { |
| 112 | + const backoff = new Backoff( |
| 113 | + this.opts.reconnectInitialMs, |
| 114 | + this.opts.reconnectFactor, |
| 115 | + this.opts.reconnectMaxMs |
| 116 | + ); |
| 117 | + |
| 118 | + while (!this.stopping && !ctl.signal.aborted && this.fatal === undefined) { |
| 119 | + const outcome = await runConnection(target, ctl.signal, this.deps); |
| 120 | + if (this.stopping || ctl.signal.aborted) break; |
| 121 | + if (outcome.kind === "fatal") { |
| 122 | + // E1: shared credentials — stop everything. |
| 123 | + this.fatal = new Error(`tunnel: ${outcome.reason}`); |
| 124 | + this.log(`tunnel: FATAL — ${outcome.reason}; stopping all connections`); |
| 125 | + this.hooks.onFatal(this.fatal); |
| 126 | + this.stopSignal.abort(); |
| 127 | + break; |
| 128 | + } |
| 129 | + if (outcome.kind === "served" || outcome.kind === "drained") { |
| 130 | + // E2: only a connection that actually held resets the backoff. |
| 131 | + const heldLongEnough = |
| 132 | + outcome.uptimeMs >= MIN_UPTIME_FOR_BACKOFF_RESET_MS; |
| 133 | + if (heldLongEnough) backoff.reset(); |
| 134 | + if (outcome.kind === "drained" && heldLongEnough) { |
| 135 | + // A stable connection was asked to rotate and the server is holding |
| 136 | + // the old one open for us — replace it NOW. |
| 137 | + this.log("tunnel: draining — reconnecting immediately"); |
| 138 | + continue; |
| 139 | + } |
| 140 | + this.log( |
| 141 | + outcome.kind === "drained" |
| 142 | + ? "tunnel: drained shortly after connecting — reconnecting with backoff" |
| 143 | + : "tunnel: connection ended — reconnecting" |
| 144 | + ); |
| 145 | + } else { |
| 146 | + this.log(`tunnel: ${outcome.reason} — reconnecting`); |
| 147 | + } |
| 148 | + await delay(backoff.next(), ctl.signal); |
| 149 | + } |
| 150 | + } |
| 151 | + |
| 152 | + /** Resolve the server set, reconcile slots, repeat. For SRV discovery the |
| 153 | + * set is re-resolved every resolveIntervalMs; an explicit set is fixed. */ |
| 154 | + private async supervise(): Promise<void> { |
| 155 | + while (!this.stopping && this.fatal === undefined) { |
| 156 | + let targets: Target[]; |
| 157 | + try { |
| 158 | + // E3: race the (un-abortable) DNS work against the wake signal so |
| 159 | + // teardown/fatal don't block on a slow resolver — a late result is |
| 160 | + // discarded by the stopping/fatal check below. |
| 161 | + const resolution = resolveTargets(this.opts); |
| 162 | + resolution.catch(() => {}); // a late rejection must not be unhandled |
| 163 | + const raced = await raceAbortable(resolution, this.wake.signal); |
| 164 | + if (raced === null) break; // woken: stopping or fatal |
| 165 | + targets = raced; |
| 166 | + } catch (err) { |
| 167 | + // Keep whatever slots exist serving; retry the resolution later |
| 168 | + // (the Rust client does the same on SRV failures). |
| 169 | + this.log( |
| 170 | + `tunnel: target resolution failed: ${err instanceof Error ? err.message : String(err)} — retrying` |
| 171 | + ); |
| 172 | + await delay( |
| 173 | + Math.min(5_000, this.opts.resolveIntervalMs), |
| 174 | + this.wake.signal |
| 175 | + ); |
| 176 | + continue; |
| 177 | + } |
| 178 | + if (this.stopping || this.fatal !== undefined) break; |
| 179 | + |
| 180 | + const desired = new Map(targets.map((t) => [targetKey(t), t] as const)); |
| 181 | + for (const [key, target] of desired) { |
| 182 | + if (!this.slots.has(key)) { |
| 183 | + this.log(`tunnel: starting connection to ${key}`); |
| 184 | + this.startSlot(key, target); |
| 185 | + } |
| 186 | + } |
| 187 | + for (const [key, slot] of this.slots) { |
| 188 | + if (!desired.has(key)) { |
| 189 | + this.log(`tunnel: ${key} no longer resolves — tearing down`); |
| 190 | + slot.ctl.abort(); |
| 191 | + } |
| 192 | + } |
| 193 | + |
| 194 | + if (this.opts.srvName === undefined) break; // explicit servers: fixed set |
| 195 | + await delay(this.opts.resolveIntervalMs, this.wake.signal); |
| 196 | + } |
| 197 | + // Slots still in the map are live; evicted ones have already settled. No |
| 198 | + // slot can start after the loop exits (stopping/fatal both gate startSlot). |
| 199 | + await Promise.all([...this.slots.values()].map((s) => s.done)); |
| 200 | + } |
| 201 | +} |
0 commit comments