diff --git a/cloneNode/cloneNode.ts b/cloneNode/cloneNode.ts index 6758bd743..bba537291 100644 --- a/cloneNode/cloneNode.ts +++ b/cloneNode/cloneNode.ts @@ -40,6 +40,7 @@ import { import { fetchJWTKeyWithRetry } from './jwtKeyClone.ts'; import { monitorSyncLoop } from './syncMonitor.ts'; import { cloneAttemptPath as cloneAttemptFilePath } from './cloneAttempt.ts'; +import { createBackoff } from '../replication/backoff.ts'; import { isExplicitDatabaseSubscription, isReplicatedDatabase as isReplicatedDatabaseUnder, @@ -98,6 +99,8 @@ import { const DEFAULT_SYNC_TIMEOUT_MS = 300000; const DEFAULT_SYNC_CHECK_INTERVAL_MS = 3000; +const VERSION_PROBE_RETRY_DELAY_MS = 1000; +const VERSION_PROBE_MAX_DELAY_MS = 4000; // Floor for the size-derived sync-wait ceiling, which guarantees the wait terminates even when data // keeps arriving without ever converging (arrivals slide the stall deadline). const MIN_MAX_CLONE_DURATION_MS = 3600000; @@ -707,6 +710,11 @@ async function monitorSync( if (!systemSocketRequired) { log(`'${SYSTEM_SCHEMA_NAME}' is not in this node's replication.databases; not requiring its socket`, 'debug'); } + const versionProbeBackoff = createBackoff({ + initialMs: VERSION_PROBE_RETRY_DELAY_MS, + maxMs: VERSION_PROBE_MAX_DELAY_MS, + minMs: VERSION_PROBE_RETRY_DELAY_MS, + }); for (let attempt = 1; systemSocketRequired && attempt <= 3; attempt++) { try { const registration: any = await leaderRequest({ operation: 'registration_info' }); @@ -716,7 +724,10 @@ async function monitorSync( break; } catch (err) { log(`Leader version probe failed (attempt ${attempt}/3): ${err}`); - if (attempt < 3) await sleep(1000); + if (attempt < 3) { + const delay = versionProbeBackoff.nextDelay(); + if (delay !== undefined) await sleep(delay); + } } } diff --git a/cloneNode/jwtKeyClone.ts b/cloneNode/jwtKeyClone.ts index 71e3c22fc..79a72beb6 100644 --- a/cloneNode/jwtKeyClone.ts +++ b/cloneNode/jwtKeyClone.ts @@ -1,4 +1,5 @@ import { setTimeout as sleep } from 'node:timers/promises'; +import { createBackoff } from '../replication/backoff.ts'; /** * Helpers for cloning the leader's JWT signing keys onto a new node. All nodes in a cluster must @@ -10,6 +11,7 @@ import { setTimeout as sleep } from 'node:timers/promises'; export const JWT_KEY_CLONE_RETRIES = 3; export const JWT_KEY_CLONE_RETRY_DELAY_MS = 250; +export const JWT_KEY_CLONE_MAX_DELAY_MS = 1000; /** * Extracts JWT key material from a `get_key` response. The operations layer wraps a bare string return @@ -35,6 +37,11 @@ export async function fetchJWTKeyWithRetry( delayMs: number = JWT_KEY_CLONE_RETRY_DELAY_MS ): Promise { let lastError: unknown; + const backoff = createBackoff({ + initialMs: delayMs, + maxMs: Math.max(delayMs, JWT_KEY_CLONE_MAX_DELAY_MS), + minMs: delayMs, + }); for (let attempt = 1; attempt <= retries; attempt++) { try { const key = extractKeyMaterial(await requestKey()); @@ -43,7 +50,10 @@ export async function fetchJWTKeyWithRetry( } catch (err) { lastError = err; } - if (attempt < retries) await sleep(delayMs); + if (attempt < retries) { + const delay = backoff.nextDelay(); + if (delay !== undefined) await sleep(delay); + } } throw new Error(`Unable to clone JWT key '${keyName}' from leader after ${retries} attempts`, { cause: lastError }); } diff --git a/replication/DESIGN.md b/replication/DESIGN.md index 23248766b..59bfe4634 100644 --- a/replication/DESIGN.md +++ b/replication/DESIGN.md @@ -10,10 +10,11 @@ Real-time, peer-to-peer replication of table data across cluster nodes via persi --- -## Files (6 total, ~4200 lines) +## Files (7 total, ~4200 lines) | File | Purpose | | -------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `backoff.ts` | The one retry schedule (`createBackoff`): exponential ceiling, full jitter, floor, wall-clock budget. See "Backoff discipline". | | `replicationConnection.ts` | The protocol engine. Defines `NodeReplicationConnection`, encodes/decodes the binary frame format, drives audit-record forwarding, manages blobs, and writes shared latency/back-pressure counters. **The big file.** | | `replicator.ts` | Setup module: `start()`, per-database/per-table `Replicator` resource class, retrieval-connection pool, operation forwarding, mTLS config. | | `subscriptionManager.ts` | Main-thread orchestration. Delegates subscription work to worker threads; routes around disconnects. | @@ -96,6 +97,87 @@ Schema (defined in that function): `name` (PK), `subscriptions[]`, `system_info` **Node discovery & TLS** — `hdb_nodes` subscriptions, `setNode.ts` for member ops, `buildReplicationMtlsConfig()` (`replicator.ts`), `monitorNodeCAs()` (`replicator.ts`). +**Retry pacing** — `createBackoff` (`backoff.ts`) is the one schedule every retry site uses; see below. + +--- + +## Backoff discipline (harper-pro#327) + +Every retry/reconnect initiation site paces attempts through **one** utility, `createBackoff` in +`backoff.ts`: an exponential ceiling with **full jitter** (uniform draw across the window) by default, +an optional fixed floor, an optional wall-clock/attempt budget, and injectable RNG/clock so every bound is +deterministically testable. An exhausted schedule returns no delay rather than `0`, so a missed exhaustion +check cannot create a busy loop. Before this there was no jitter anywhere in production code — a fleet +reacting to one event retried in lockstep — and the delays themselves ranged from flat 200 ms to +jitterless doubling. + +The invariant the discipline enforces: **at most one pending attempt per target, on a bounded, capped, +decorrelated schedule.** Pacing alone is not enough; the storm surface below needed the dedup too. + +| Site | Schedule | Reset signal | +| ------------------------------------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------- | +| `createSubscribeSetupScheduler` (`subscriptionManager.ts`) — subscription setup | floor `NODE_SUBSCRIBE_DELAY`, ceiling `2 × NODE_SUBSCRIBE_DELAY` → 30 s, plus the caller's `RECONNECT_STAGGER_MS` stagger | `connectedToNode` (reset only; the armed setup still fires) | +| `NodeReplicationConnection.scheduleReconnect` | fixed 500 ms floor, ceiling `INITIAL_RETRY_TIME` 500 ms → 30 s, full jitter; the floor preserves the hard minimum from the TLS-state incident while the remaining window decorrelates fleet redials (harper-pro#339) | `onFrameSent` — first frame actually sent, **not** socket open | +| `reconcileWorkers` wedge / receive-stall re-drives | one fixed-window draw per sweep, used as a common base under the existing `RECONNECT_STAGGER_MS` spacing (decorrelation only; the re-drives are already throttled by the `disconnectedAt` / `receiveStallReconnectAt` re-stamps), one owned `entry.reDriveTimer` per entry | n/a — disarmed on unsubscribe, delete, worker exit, and entry replacement | +| `runNodeUpdateWatcher` (`knownNodes.ts`) — hdb_nodes watcher restart | full-jitter ceiling 1 s → 30 s | an iteration that survived `NODE_WATCHER_HEALTHY_UPTIME_MS` | +| `shouldCloseSendAuthWatch` reprobe (`replicationConnection.ts`) | 500 ms → 5 s under a 30 s wall-clock budget, then fails closed | n/a (one-shot loop) | +| `fetchJWTKeyWithRetry`, the clone version probe (`cloneNode/`) | 250 ms → 1 s / 1 s → 4 s, attempt count unchanged | n/a | +| `repairBlobs` (`blobRepair.ts`) — per-record | 50 ms → 1 s, only on a record no peer could repair, under a 60 s per-failure-run pacing budget; once that is spent the sweep keeps scanning unpaced (with a sampled warn) rather than stopping, so an unrepairable prefix cannot hide the records behind it | a repaired record | +| copy-cursor flush retry (`replicationConnection.ts`) — `onPersistFailure` | fixed 250 ms floor, ceiling 250 ms → 30 s; the floor is what the `copyFlushBackoffUntil` guard depends on | a flush that persisted the cursor | +| worker readiness re-attempt (`subscriptionManager.ts`) | floor `NODE_SUBSCRIBE_DELAY`, ceiling `2 × NODE_SUBSCRIBE_DELAY` → 30 s, one armed re-attempt at a time | components loaded | + +**The subscription-setup scheduler is the one with dedup.** `onDatabase` used to turn every qualifying +node update straight into a retained (not unref'd) 200 ms `setTimeout` plus a `subscribe-to-node` +message, so whatever re-drove `onNodeUpdate` amplified 1:1 into main-thread timers, worker-side +WebSocket/TLS setup, and `Setting up subscription with leader` warns — ~1,400 lines/s/node in the field, +ending in an OOM kill. The scheduler holds one armed setup per **(peer URL, database)** in its own map +(_not_ on the `connectionReplicationMap` entry, which the stale-worker path deletes and recreates), and +the armed setup carries the newest level-state payload — reading `entry.nodes` at fire time is _not_ +safe, because `onDatabase`'s early-return path replaces that array without running the leader/url +enrichment (that path calls `refreshPending` instead). Self-catchup is separate one-shot state: it is +attached on a fresh array immediately before dispatch and consumed only after the worker message is +accepted, so timer cancellation, stale dispatch, or a synchronous `postMessage` throw cannot lose it. +A setup is cancelled on +unsubscribe, on node deletion, and on a same-name URL migration, all of which became reachable once a +pending timer could live 30 s instead of 200 ms. The wedge/stall recovery kicks are owned the same way — +one `entry.reDriveTimer` per entry, so a staggered sweep that outruns the reconcile window that started it +replaces its predecessor instead of stacking another wave, and disarmed when the owning worker exits or the +entry is replaced. + +**A connect report cancels nothing; it only resets the escalated delay.** The main thread cannot attribute a +report to the entry that armed the work — `connectToNextWorker` subscribes a failover peer on a worker that +is not `entry.worker`, and a superseded worker's hung-but-open connection reports for the same pair as its +replacement — so cancelling on it strands a just-recreated entry, and gating the _reset_ on that attribution +is what let a chaos-restart peer's setup delay escalate past its reconvergence budget. What makes leaving +the timers armed safe is that each re-checks live state when it fires: the setup re-reads the entry and its +`unsubscribed` flag, the wedge kick claims its entry through the `disconnectedAt` stamp a connect clears, +and the stall kick claims it through `connectGeneration` (a stalled connection is `connected: true` with no +`disconnectedAt`, so the stamp cannot discriminate for it) plus the receive watermark. `shouldFireStallKick` +holds that decision, because the stamp it checks is also the re-detection throttle: a kick skipped because +the leg reconnected has to hand the stamp back, or a fresh socket that stalls too is never detected — its +`lastReceivedTime` can never move past a stamp it never advanced. + +Their jitter is drawn once +per sweep rather than per entry: a per-entry draw would vary consecutive delays by up to ±200 ms and let +several dials share a 50 ms instant, which is the concurrency the `RECONNECT_STAGGER_MS` spacing exists to +bound (#446). The warn moved inside the "actually armed" branch: it now describes an attempt, not an +event. + +The worker boundary has one additional admission point for its asynchronous startup window. Before +components are ready, one insertion-ordered map retains only the latest subscribe/unsubscribe message +per actual connection key and database, with one continuation on the readiness promise — and, if that +readiness rejects, exactly one armed re-attempt on the same schedule, so the retained actions apply without +waiting for another message or the wedge reconcile. After readiness, +the handlers run inline and allocate no queue state; the existing connection map is then the single-flight +owner. Subscribe, unsubscribe, force-reconnect, and startup admission all derive the connection key through +`getSubscriptionConnectionKey`, including the missing nested-URL fallback. + +**What is deliberately NOT on this schedule:** the receive/copy watchdogs and their thresholds, and the +doubling copy-finalize _timeout bound_ alongside them (these _detect_ stalls or bound a wait; this +discipline paces _retries_, and jittering a durability deadline would be actively wrong), +`blobGapReconnectTimer`, the in-place `BLOB_SEND_RETRY_DELAYS_MS` 503 retries, +`PING_INTERVAL`/`PING_TIMEOUT`, and `RECONCILE_INTERVAL_MS`. + --- ## Non-obvious behaviors @@ -172,7 +254,9 @@ Most replication behavior is exercised via integration tests that spin up multi- | ----------------------------------------- | ----------------------------------------------------------------------------------------------------------- | | Where does a remote message get decoded? | `replicationConnection.ts → replicateOverWS` | | Where do cache-miss fetches pick a peer? | `replicator.ts → Replicator.load` (declared inside `setReplicator`) | -| Where is the connection retry loop? | `replicationConnection.ts → NodeReplicationConnection` (uses `INITIAL_RETRY_TIME`) | +| Where is the connection retry loop? | `replicationConnection.ts → NodeReplicationConnection.scheduleReconnect` (uses `INITIAL_RETRY_TIME`) | +| Where is the retry/backoff schedule? | `backoff.ts → createBackoff`; adopting sites listed under "Backoff discipline" | +| Why is a subscribe setup not firing? | `subscriptionManager.ts → createSubscribeSetupScheduler` — one armed setup per (url, database) | | Where is mTLS configured? | `replicator.ts → buildReplicationMtlsConfig` | | Where is a new cluster member added? | `setNode.ts` (the whole file is one operation) | | Where are protocol message types defined? | `replicationConnection.ts` — top-level consts (`SUBSCRIPTION_REQUEST` … `SUBSCRIPTION_UPDATE`) | diff --git a/replication/backoff.ts b/replication/backoff.ts new file mode 100644 index 000000000..071c43dfa --- /dev/null +++ b/replication/backoff.ts @@ -0,0 +1,81 @@ +/** + * The one backoff schedule replication retries use; the adopting sites and their parameters are in + * `DESIGN.md` under "Backoff discipline". + * + * Full jitter (uniform over the whole window) is the default because decorrelating a fleet's retries + * matters more than a tight worst-case delay on these cold error paths. `minMs` preserves a site's + * independent lower bound while jittering the rest of the window. + * + * `budgetMs` is a deadline read off an injected monotonic clock, not a sum of requested sleeps: a + * resolver hang or an event-loop stall must not extend a bounded grace period past what it advertises. + * `maxAttempts` is the bound that still holds when the clock does not advance. + */ + +export interface BackoffOptions { + initialMs: number; + maxMs: number; + minMs?: number; + factor?: number; + jitter?: 'full' | 'none'; + budgetMs?: number; + maxAttempts?: number; + random?: () => number; + now?: () => number; +} + +export interface Backoff { + nextDelay(): number | undefined; + reset(): void; + readonly attempts: number; + readonly ceiling: number; + readonly exhausted: boolean; +} + +export function createBackoff(options: BackoffOptions): Backoff { + const { + initialMs, + maxMs, + minMs = 0, + factor = 2, + jitter = 'full', + budgetMs, + maxAttempts, + random = Math.random, + now = () => performance.now(), + } = options; + let attempts = 0; + let deadline = budgetMs === undefined ? undefined : now() + budgetMs; + + function ceilingFor(attempt: number): number { + return Math.max(minMs, Math.min(initialMs * factor ** attempt, maxMs)); + } + function isExhausted(currentTime: number): boolean { + if (maxAttempts !== undefined && attempts >= maxAttempts) return true; + return deadline !== undefined && currentTime >= deadline; + } + + return { + get attempts() { + return attempts; + }, + get ceiling() { + return ceilingFor(attempts); + }, + get exhausted() { + return isExhausted(now()); + }, + nextDelay() { + const currentTime = now(); + if (isExhausted(currentTime)) return undefined; + const ceiling = ceilingFor(attempts); + attempts++; + let delay = jitter === 'none' ? ceiling : minMs + Math.floor(random() * (ceiling - minMs)); + if (deadline !== undefined) delay = Math.min(delay, deadline - currentTime); + return delay; + }, + reset() { + attempts = 0; + if (budgetMs !== undefined) deadline = now() + budgetMs; + }, + }; +} diff --git a/replication/blobRepair.ts b/replication/blobRepair.ts index 9de91998a..d2a607c94 100644 --- a/replication/blobRepair.ts +++ b/replication/blobRepair.ts @@ -4,9 +4,24 @@ import { databases } from '../core/resources/databases.ts'; import { server } from '../core/server/Server.ts'; import harperLogger from '../core/utility/logging/harper_logger.js'; import type { Logger } from '../core/utility/logging/logger.ts'; +import { setTimeout as sleep } from 'node:timers/promises'; +import { createBackoff } from './backoff.ts'; const logger = harperLogger.forComponent('blob-repair').conditional as Logger; +// Paces a sweep whose peers cannot serve anything: without it the per-record loop spins as fast as the +// incomplete-blob cursor yields. Capped low (not at the 30s replication cap) because the delay is paid +// per unrepairable record while the cursor stays open. +const REPAIR_RETRY_INITIAL_MS = 50; +const REPAIR_RETRY_MAX_MS = 1000; +// Wall-clock ceiling on how long one unbroken failure run may spend *pausing*. Once spent the sweep keeps +// scanning at full speed: pacing exists to stop a hot loop, and inferring the rest of the cursor from an +// unrepairable prefix would silently skip records a later peer can still serve. A repair restarts it. +const REPAIR_PACING_BUDGET_MS = 60_000; +// Once unpaced the per-record warn would fire at peer-RTT rate for the rest of the sweep — hundreds of +// thousands of lines during exactly the incident where the log is the diagnostic channel. Sample it. +const REPAIR_UNPACED_WARN_EVERY = 100; + export async function allBlobsAreComplete( blobs: any[], checkBlob: (blob: any) => Promise = isBlobComplete @@ -15,15 +30,24 @@ export async function allBlobsAreComplete( } export async function repairBlobs( - dbName: string + dbName: string, + deps: { sleep?: (ms: number) => Promise; now?: () => number } = {} ): Promise<{ checked: number; repaired: number; failed: number; noConnection: number }> { const database = (databases as any)[dbName]; if (!database) throw new Error(`Unknown database '${dbName}'`); + const pause = deps.sleep ?? sleep; let checked = 0; let repaired = 0; let failed = 0; let noConnection = 0; + const backoff = createBackoff({ + initialMs: REPAIR_RETRY_INITIAL_MS, + maxMs: REPAIR_RETRY_MAX_MS, + budgetMs: REPAIR_PACING_BUDGET_MS, + now: deps.now, + }); + let pacingSpent = false; for await (const { tableName, table, recordId } of findIncompleteBlobRefs(database, dbName)) { checked++; @@ -70,9 +94,31 @@ export async function repairBlobs( } } - if (!peerRepaired) { + if (peerRepaired) { + backoff.reset(); + pacingSpent = false; + } else { failed++; - logger.warn?.('Could not repair blob for record', recordId, 'in', tableName, '— no peer had a complete copy'); + if (!pacingSpent || failed % REPAIR_UNPACED_WARN_EVERY === 0) + logger.warn?.( + 'Could not repair blob for record', + recordId, + 'in', + tableName, + '— no peer had a complete copy', + pacingSpent ? `(${failed} failed so far; sampling 1 in ${REPAIR_UNPACED_WARN_EVERY})` : '' + ); + const delay = backoff.nextDelay(); + if (delay === undefined) { + if (!pacingSpent) { + pacingSpent = true; + logger.warn?.( + 'Blob repair pacing budget spent for', + dbName, + `after ${REPAIR_PACING_BUDGET_MS}ms of unrepairable records; continuing the sweep unpaced` + ); + } + } else await pause(delay); } } diff --git a/replication/knownNodes.ts b/replication/knownNodes.ts index 907c298ea..7692726d0 100644 --- a/replication/knownNodes.ts +++ b/replication/knownNodes.ts @@ -12,6 +12,7 @@ import * as env from '../core/utility/environment/environmentManager.js'; import { CONFIG_PARAMS } from '../core/utility/hdbTerms.ts'; import { logger } from '../core/utility/logging/logger.ts'; import { isExplicitDatabaseSubscription, isReplicatedDatabase } from './replicatedDatabases.ts'; +import { createBackoff } from './backoff.ts'; type MaybePromise = T | Promise; @@ -122,12 +123,20 @@ const NODE_WATCHER_RESTART_DELAY_MS = 1000; // Cap the exponential backoff so a persistent failure (subscribe throws every time) // doesn't run a tight 1s log+retry loop forever — back off up to 30s instead. const NODE_WATCHER_MAX_DELAY_MS = 30_000; +// How long an iteration has to survive before it counts as a healthy run that clears the backoff. +// Resolving `subscribe()` is not progress, and neither is receiving an event: a subscription that +// replays one row and immediately throws does both on every cycle, which is exactly the shape that +// used to pin the restart delay at 1s forever (harper-pro#327). +const NODE_WATCHER_HEALTHY_UPTIME_MS = 10_000; type WatcherOptions = { subscribe?: () => Promise> | AsyncIterable; processEvent?: (event: any, listener: (node: any, id: string) => void) => Promise | void; restartDelayMs?: number; maxDelayMs?: number; maxRestarts?: number; + healthyUptimeMs?: number; + now?: () => number; + random?: () => number; }; // Generation tokens guarding against duplicate node-update watchers (harper-pro#460). A @@ -168,20 +177,28 @@ export async function runNodeUpdateWatcher( const restartDelayMs = options.restartDelayMs ?? NODE_WATCHER_RESTART_DELAY_MS; const maxDelayMs = options.maxDelayMs ?? NODE_WATCHER_MAX_DELAY_MS; const maxRestarts = options.maxRestarts ?? Infinity; + const healthyUptimeMs = options.healthyUptimeMs ?? NODE_WATCHER_HEALTHY_UPTIME_MS; + const now = options.now ?? (() => performance.now()); const key = options.key ?? DEFAULT_WATCHER_KEY; // Supersede any watcher already running for this key so a reload doesn't stack a second loop // (harper-pro#460). Distinct keys (subscription vs confirmation) run concurrently and untouched. stopNodeUpdateWatcher(key); const generation = watcherGenerations.get(key) ?? 0; let restarts = 0; - let consecutiveFailures = 0; + const backoff = createBackoff({ + initialMs: restartDelayMs, + maxMs: maxDelayMs, + random: options.random, + }); const isCurrent = () => generation === (watcherGenerations.get(key) ?? 0); while (restarts < maxRestarts && isCurrent()) { - let iteratedSuccessfully = false; + // Stamped only once the subscription is live: time spent acquiring (or failing) one is not uptime, + // so a subscribe() that blocks past the threshold and then throws must not read as a healthy run. + let liveSince: number | undefined; try { const events = await subscribe(); if (!isCurrent()) break; // superseded while awaiting subscribe - iteratedSuccessfully = true; // we got past subscribe — any later throw is a fresh failure + liveSince = now(); const iterator = events[Symbol.asyncIterator](); watcherIterators.set(key, iterator); try { @@ -207,12 +224,14 @@ export async function runNodeUpdateWatcher( if (!isCurrent()) break; // superseded watcher; iterator.return rejected logger.error?.('hdb_nodes watcher failed; restarting', error); } - // Successful subscribe → reset backoff so a fresh failure restarts quickly. - consecutiveFailures = iteratedSuccessfully ? 0 : consecutiveFailures + 1; + // A watcher that stayed up long enough to have done real work restarts quickly; anything shorter + // is a failure cycle and keeps escalating toward the cap. + if (liveSince !== undefined && now() - liveSince >= healthyUptimeMs) backoff.reset(); restarts++; if (restarts >= maxRestarts || !isCurrent()) return; - const delay = Math.min(restartDelayMs * Math.pow(2, Math.min(consecutiveFailures, 5)), maxDelayMs); - await new Promise((resolve) => setTimeout(resolve, delay)); + const delay = backoff.nextDelay(); + if (delay === undefined) return; + await new Promise((resolve) => setTimeout(resolve, delay).unref()); } } /** diff --git a/replication/replicationConnection.ts b/replication/replicationConnection.ts index 0cb3401e1..e81062564 100644 --- a/replication/replicationConnection.ts +++ b/replication/replicationConnection.ts @@ -101,6 +101,7 @@ import { PassThrough } from 'node:stream'; import { getLastVersion } from 'lmdb'; import { FrameWriter } from './frameWriter.ts'; import { cloneAttemptSource } from '../cloneNode/cloneAttempt.ts'; +import { createBackoff, type Backoff } from './backoff.ts'; const logger = forComponent('replication').conditional as Logger; // msgpackr v2 removed the built-in `randomAccessStructure` option; that random-access @@ -534,8 +535,13 @@ const PAUSE_STALL_THRESHOLD_MS = Math.max( // base-copy resync re-encodes the row against local structures), so a decode blip never de-authorizes // a healthy peer, while still bounding how long a revocation carried by an undecodable write can go // unenforced. Only reached on a genuine decode failure — a decodable row is evaluated on the first probe. -const SEND_AUTH_REPROBE_INTERVAL_MS = 500; -const SEND_AUTH_REPROBE_ATTEMPTS = 60; +const SEND_AUTH_REPROBE_INITIAL_MS = 500; +const SEND_AUTH_REPROBE_MAX_MS = 5_000; +// The grace period is a wall-clock deadline, not a sum of sleeps: a resolver hang or event-loop stall +// must not extend how long a revocation carried by an undecodable write can go unenforced. +const SEND_AUTH_REPROBE_BUDGET_MS = 30_000; +// The bound that still holds when the clock does not advance. +const SEND_AUTH_REPROBE_ATTEMPTS = 120; /** * Decide whether the dynamic send-authorization watch (the per-subscriber `getHDBNodeTable().subscribe` @@ -573,18 +579,41 @@ export async function shouldCloseSendAuthWatch( isClosed: () => boolean; onReprobeTimeout?: () => void; resolve?: (name: string) => any; - sleep?: () => Promise; + sleep?: (ms: number) => Promise; reprobeAttempts?: number; + reprobeBudgetMs?: number; + now?: () => number; } ): Promise { const resolve = deps.resolve ?? resolveNodeForSendAuth; - const sleep = deps.sleep ?? (() => new Promise((r) => setTimeout(r, SEND_AUTH_REPROBE_INTERVAL_MS).unref())); - const reprobeAttempts = deps.reprobeAttempts ?? SEND_AUTH_REPROBE_ATTEMPTS; + const sleep = deps.sleep ?? ((ms: number) => new Promise((r) => setTimeout(r, ms).unref())); let node = isGenuineNodeDeletion(event.type) ? undefined : resolve(name); - for (let attempt = 0; node === SEND_AUTH_UNCHANGED && attempt < reprobeAttempts && !deps.isClosed(); attempt++) { - await sleep(); + // Built only once a row comes back undecodable: the ordinary decodable event never reads the clock. + let backoff; + while (node === SEND_AUTH_UNCHANGED && !deps.isClosed()) { + backoff ??= createBackoff({ + initialMs: SEND_AUTH_REPROBE_INITIAL_MS, + maxMs: SEND_AUTH_REPROBE_MAX_MS, + budgetMs: deps.reprobeBudgetMs ?? SEND_AUTH_REPROBE_BUDGET_MS, + maxAttempts: deps.reprobeAttempts ?? SEND_AUTH_REPROBE_ATTEMPTS, + now: deps.now, + }); + if (backoff.exhausted) break; + const delay = backoff.nextDelay(); + if (delay === undefined) break; + await sleep(delay); + // Re-check before trusting the next read, not just at the top of the loop: the grace period is + // advertised as a deadline, so a row that only becomes decodable after an event-loop stall + // pushed us past it must not authorize. Fail closed on the elapsed time, not on the row. + if (backoff.exhausted) break; node = resolve(name); + // resolveNodeForSendAuth reads the store synchronously, so the read itself can carry the clock + // past the deadline. Same rule applies to what it returned. + if (backoff.exhausted) { + node = SEND_AUTH_UNCHANGED; + break; + } } if (node === SEND_AUTH_UNCHANGED) { if (deps.isClosed()) return false; @@ -2441,6 +2470,9 @@ export async function createWebSocket( } const INITIAL_RETRY_TIME = 500; +const MAX_RETRY_TIME = 30_000; +const COPY_FLUSH_RETRY_INITIAL_MS = 250; +const COPY_FLUSH_RETRY_MAX_MS = 30_000; /** * This represents a persistent connection to a node for replication, which handles * sockets that may be disconnected and reconnected @@ -2448,7 +2480,9 @@ const INITIAL_RETRY_TIME = 500; export class NodeReplicationConnection extends EventEmitter { socket: WebSocket; startTime: number; - retryTime = INITIAL_RETRY_TIME; + retryBackoff: Backoff; // created on the first failure + + random = Math.random; // injectable so the jittered reconnect schedule is deterministically testable retries = 0; isConnected = true; // we start out assuming we will be connected isFinished = false; @@ -2670,23 +2704,36 @@ export class NodeReplicationConnection extends EventEmitter { scheduleReconnect() { this.reconnectScheduled = true; this.resetSession(); - setTimeout(() => { - this.connect(); - }, this.retryTime).unref(); - // Double the interval each retry, capped at 30 s. The previous ~0.4%/retry + // Double the ceiling each retry, capped at 30 s. The previous ~0.4%/retry // growth took >1000 retries to reach any meaningful delay, so rapid // reconnects to a dead peer (symphony accepts the TLS handshake then drops // it) would still accumulate unreleased native TLS state faster than V8 can // GC under CPU-saturated bulk-write conditions, leading to OOM (#339). - // Doubling reaches 30 s in ~6 retries (~62 s total) and resets on success. - this.retryTime = Math.min(this.retryTime << 1, 30_000); + // Doubling reaches 30 s in ~6 retries and resets on success. The delay is drawn + // with full jitter so a fleet reacting to one outage does not redial in lockstep. Keep + // the original 500 ms lower bound from #339 while jittering the rest of the window. + this.retryBackoff ??= createBackoff({ + initialMs: INITIAL_RETRY_TIME, + maxMs: MAX_RETRY_TIME, + minMs: INITIAL_RETRY_TIME, + random: this.random, + }); + const delay = this.retryBackoff.nextDelay(); + if (delay === undefined) return; + setTimeout(() => { + this.connect(); + }, delay).unref(); + } + /** The ceiling the next reconnect will be drawn under; the initial value means "not backed off". */ + get retryTime(): number { + return this.retryBackoff?.ceiling ?? INITIAL_RETRY_TIME; } // Called by replicateOverWS after a frame is actually sent: real progress, so it is safe to reset the // backoff. Gated on a non-zero retries so the healthy hot path (already reset) does nothing. onFrameSent() { - if (this.retries !== 0 || this.retryTime !== INITIAL_RETRY_TIME) { + if (this.retries !== 0 || this.retryBackoff?.attempts) { this.retries = 0; - this.retryTime = INITIAL_RETRY_TIME; + this.retryBackoff?.reset(); } } // Retire the live replicateOverWS instance: the single enforcement point for "at most one live session @@ -2876,7 +2923,7 @@ export function replicateOverWS(ws: WebSocket, options: any, authorization: any) // at ~100% CPU and flooding logs on a persistent failure. Escalating backoff + a scheduled retry paces it: // transient errors self-heal, a persistent one idles until the operator (or a higher-level watchdog) acts. let copyFlushBackoffUntil = 0; - let copyFlushRetryMs = 0; + let copyFlushBackoff: Backoff | undefined; let copyFlushRetryTimer; const COPY_CURSOR_FLUSH_BYTES = env.get('replication_copyCursorFlushBytes') ?? 64 * 1024 * 1024; const COPY_CURSOR_FLUSH_INTERVAL_MS = Math.max(env.get('replication_copyCursorFlushIntervalMs') ?? 5000, 1); @@ -3064,7 +3111,14 @@ export function replicateOverWS(ws: WebSocket, options: any, authorization: any) if (passAtPersist === copyWatermark.currentPass) pendingCopyCursor ??= cursor; // hold the staged cursor // Escalating backoff (250ms → 30s cap) instead of an immediate re-flush, so a persistent flush // failure idles rather than busy-looping; a scheduled retry drives progress without another event. - copyFlushRetryMs = Math.min(copyFlushRetryMs ? copyFlushRetryMs * 2 : 250, 30000); + // Jittered on the shared schedule so a cluster-wide I/O episode does not re-flush in lockstep, + // keeping 250ms as the hard floor the backoff guard below depends on. + copyFlushBackoff ??= createBackoff({ + initialMs: COPY_FLUSH_RETRY_INITIAL_MS, + maxMs: COPY_FLUSH_RETRY_MAX_MS, + minMs: COPY_FLUSH_RETRY_INITIAL_MS, + }); + const copyFlushRetryMs = copyFlushBackoff.nextDelay() ?? COPY_FLUSH_RETRY_INITIAL_MS; copyFlushBackoffUntil = performance.now() + copyFlushRetryMs; clearTimeout(copyFlushRetryTimer); copyFlushRetryTimer = setTimeout(() => { @@ -3123,7 +3177,7 @@ export function replicateOverWS(ws: WebSocket, options: any, authorization: any) if (copyFromNodeId !== undefined) getDatabaseStores().dbisDB?.put([Symbol.for('copyCursor'), copyFromNodeId], cursorAtFlush); copyFlushBackoffUntil = 0; - copyFlushRetryMs = 0; // flush succeeded; reset backoff + copyFlushBackoff?.reset(); // flush succeeded if (copyCompleteReceived) noteCopyFinalizeProgress(); logger.trace?.(connectionId, 'copy cursor advanced (rows flushed durable)'); // A persist under a drained barrier is the connection's LAST possible progress: every @@ -4678,7 +4732,7 @@ export function replicateOverWS(ws: WebSocket, options: any, authorization: any) onReprobeTimeout: () => logger.warn?.( connectionId, - `hdb_nodes row for ${authorization.name} did not decode within ${SEND_AUTH_REPROBE_ATTEMPTS * SEND_AUTH_REPROBE_INTERVAL_MS}ms; failing closed so a revocation carried by an undecodable write cannot be missed` + `hdb_nodes row for ${authorization.name} did not decode within ${SEND_AUTH_REPROBE_BUDGET_MS}ms; failing closed so a revocation carried by an undecodable write cannot be missed` ), }); if (shouldClose) { diff --git a/replication/replicator.ts b/replication/replicator.ts index 27c1cff7a..992cdefff 100644 --- a/replication/replicator.ts +++ b/replication/replicator.ts @@ -547,7 +547,7 @@ function getSubscriptionConnection( authorization?: string, status?: { reused: boolean } ) { - const connectionKey = connectingUrl + '-' + subscriptionUrl; + const connectionKey = getSubscriptionConnectionKey(connectingUrl, subscriptionUrl); let dbConnections = connections.get(connectionKey); if (!dbConnections) { dbConnections = new Map(); @@ -570,6 +570,15 @@ function getSubscriptionConnection( return connection; } } + +// The `connections` key, shared by subscribe, unsubscribe, force-reconnect and the worker's pre-readiness +// admission so all four derive identical identity including the missing-nested-URL fallback. The teardown +// sites pass the pair in the opposite order to the subscribe path; the two URLs are the same string for +// every ordinary subscription, and differ only under failover (connectToNextWorker subscribes node B over +// peer A's URL), where the teardown lookup has always missed. +export function getSubscriptionConnectionKey(url: string, peerUrl?: string): string { + return url + '-' + (peerUrl ?? url); +} const nodeNameToRetrievalConnections = new Map>(); /** * Get connection by node name, using caching @@ -693,7 +702,7 @@ export function subscribeToNode(request: any) { logger.error('Error in subscription to node', request.nodes[0]?.url, error); } } -export async function unsubscribeFromNode({ url, nodes, database }) { +export function unsubscribeFromNode({ url, nodes, database }) { logger.trace( 'Unsubscribing from node', url, @@ -701,7 +710,7 @@ export async function unsubscribeFromNode({ url, nodes, database }) { 'nodes', Array.from(getHDBNodeTable().primaryStore.getRange({})) ); - const connectionKey = url + '-' + (nodes[0]?.url ?? url); + const connectionKey = getSubscriptionConnectionKey(url, nodes[0]?.url); const dbConnections = connections.get(connectionKey); if (dbConnections) { const connection = dbConnections.get(database); @@ -721,7 +730,7 @@ export async function unsubscribeFromNode({ url, nodes, database }) { // the worker-local copy-progress watchdog (harper-pro#453) did not recover it. The connection-key lookup // mirrors unsubscribeFromNode so it resolves the same connection the subscribe path created. export function forceReconnectToNode({ url, nodes, database }) { - const connectionKey = url + '-' + (nodes?.[0]?.url ?? url); + const connectionKey = getSubscriptionConnectionKey(url, nodes?.[0]?.url); const connection = connections.get(connectionKey)?.get(database); if (connection) connection.forceReconnect(); } diff --git a/replication/subscriptionManager.ts b/replication/subscriptionManager.ts index 935f217b8..09c3726cc 100644 --- a/replication/subscriptionManager.ts +++ b/replication/subscriptionManager.ts @@ -13,6 +13,7 @@ import { forEachReplicatedDatabase, unsubscribeFromNode, forceReconnectToNode, + getSubscriptionConnectionKey, } from './replicator.ts'; import { getThisNodeName, getThisNodeUrl } from '../core/server/nodeName.ts'; import { parentPort } from 'worker_threads'; @@ -35,6 +36,7 @@ import { type ConnectionTruth, } from './replicationConnection.ts'; import * as logger from '../core/utility/logging/harper_logger.js'; +import { createBackoff, type Backoff } from './backoff.ts'; import lodash from 'lodash'; const { cloneDeep } = lodash; import * as env from '../core/utility/environment/environmentManager.js'; @@ -65,6 +67,16 @@ type ConnectedWorkerStatus = { // last shared-memory truth correction applied to it. Logged with each subsequent fire so the // watchdog-demotion soak can tell "sole detector" fires from ones where another layer (or the // truth-driven path) had already engaged. Telemetry only — never consulted for recovery decisions. + // The single recovery timer the reconcile has armed for this entry (wedge re-drive or stall kick). + // Owned rather than fire-and-forget: with many databases a staggered sweep can outrun the reconcile + // window that started it, so each new decision replaces the entry's pending one instead of stacking + // another wave, and connect/unsubscribe/delete can disarm it. + reDriveTimer?: ReturnType; + // Bumped on every connect report. The wedge kick claims its entry through `disconnectedAt`, which a + // connect clears; a stalled connection is connected:true with no `disconnectedAt`, so the stall kick + // needs its own claim or a leg that drops and reconnects inside its stagger window gets force-reconnected + // on the strength of the old socket's watermark. + connectGeneration?: number; lastRecovery?: { mechanism: string; at: number }; lastTruthCorrection?: { direction: 'down' | 'up'; at: number }; }; @@ -94,6 +106,11 @@ const NODE_SUBSCRIBE_DELAY = 200; // delay before sending node subscribe to othe // replicateOverWS instance + TLS buffer). Stagger them so at most ~1 new connection starts // per RECONNECT_STAGGER_MS, keeping peak concurrent connection setup bounded. const RECONNECT_STAGGER_MS = 50; +// Ceiling the subscription-setup backoff starts at, and the cap it grows to. The floor stays +// NODE_SUBSCRIBE_DELAY: that delay exists to let operations complete first, which is unrelated to retry +// pacing, so jitter is drawn above it rather than through it. See createSubscribeSetupScheduler. +const NODE_SUBSCRIBE_INITIAL_CEILING_MS = 2 * NODE_SUBSCRIBE_DELAY; +const NODE_SUBSCRIBE_MAX_DELAY_MS = 30_000; // Cadence of the per-process safety-net reconcile that rebinds subscriptions whose // worker no longer exists. Pure read-side filter against `workers` and // `connectionReplicationMap` on each tick when nothing is wrong, so a short interval @@ -126,6 +143,179 @@ const RECEIVE_STALL_THRESHOLD_MS = 15 * 60_000; const workersWithExitHandler = new WeakSet(); const connectionReplicationMap = new Map(); +interface SubscribeSchedule { + timer?: ReturnType; + backoff: Backoff; + nodes?: any[]; +} + +export interface SubscribeSetupScheduler { + /** + * Arm a setup for this pair with `nodes` as its payload, or return undefined when one is already + * pending (deduped) — the payload is refreshed either way, so the newest one is what fires. + */ + schedule(url: string, database: string, nodes: any[], staggerMs?: number): number | undefined; + /** Replace the payload of an already-armed setup without arming one. */ + refreshPending(url: string, database: string, nodes: any[]): void; + /** The pair reached 'open': real progress, so drop the escalated delay. */ + noteConnected(url: string, database: string): void; + cancel(url: string, database: string): void; + cancelUrl(url: string): void; + pendingCount(): number; +} + +/** + * Admission control for subscription setup: **at most one pending setup per (peer URL, database)**, + * on an escalating jittered schedule that resets when the pair connects. harper-pro#327 — see + * DESIGN.md, "Backoff discipline", for the incident and the rest of the sites. + * + * Keyed in its own map rather than on the `connectionReplicationMap` entry, because the stale-worker + * path in `onDatabase` deletes and recreates that entry — per-entry state would be wiped on exactly + * the path that most needs the dedup. The payload rides on the schedule rather than being read back + * off the entry at fire time: `onDatabase` replaces `entry.nodes` on its early-return path *without* + * running the leader/url enrichment, so the entry's array is not necessarily the one a setup should be + * sent with. That path calls `refreshPending` instead, which is what keeps "newest payload wins" true + * without arming anything. + */ +export function createSubscribeSetupScheduler(deps: { + dispatch: (url: string, database: string, nodes: any[]) => void; + random?: () => number; + initialMs?: number; + maxMs?: number; + minMs?: number; +}): SubscribeSetupScheduler { + const { + dispatch, + random, + initialMs = NODE_SUBSCRIBE_INITIAL_CEILING_MS, + maxMs = NODE_SUBSCRIBE_MAX_DELAY_MS, + minMs = NODE_SUBSCRIBE_DELAY, + } = deps; + const schedules = new Map>(); + + function drop(url: string, database: string) { + const forUrl = schedules.get(url); + const schedule = forUrl?.get(database); + if (!schedule) return; + clearTimeout(schedule.timer); + forUrl!.delete(database); + if (forUrl!.size === 0) schedules.delete(url); + } + + return { + schedule(url, database, nodes, staggerMs = 0) { + let forUrl = schedules.get(url); + if (!forUrl) schedules.set(url, (forUrl = new Map())); + let schedule = forUrl.get(database); + if (!schedule) { + schedule = { backoff: createBackoff({ initialMs, maxMs, minMs, random }) }; + forUrl.set(database, schedule); + } + schedule.nodes = nodes; + if (schedule.timer) return undefined; + const backoffDelay = schedule.backoff.nextDelay(); + if (backoffDelay === undefined) return undefined; + const delay = backoffDelay + staggerMs; + schedule.timer = setTimeout(() => { + schedule.timer = undefined; + const pending = schedule.nodes; + schedule.nodes = undefined; + // A synchronous throw here (postMessage on an uncloneable payload) would take the process + // down, and this is the recovery path. + try { + if (pending) dispatch(url, database, pending); + } catch (error) { + logger.error('Error dispatching subscription setup for', database, url, error); + } + }, delay); + schedule.timer.unref?.(); + return delay; + }, + refreshPending(url, database, nodes) { + const schedule = schedules.get(url)?.get(database); + if (schedule?.timer) schedule.nodes = nodes; + }, + noteConnected(url, database) { + // Reset only — the armed setup is deliberately left to fire. The main thread cannot attribute a + // connect report to the entry that armed the setup: `connectToNextWorker` subscribes a failover + // peer on a worker that is not `entry.worker`, and a superseded worker's hung-but-open connection + // reports for the same (url, database) as its replacement. Cancelling on a report we cannot + // attribute strands a just-recreated entry unsubscribed; letting a redundant subscribe reach a + // live connection is what this path always did. + schedules.get(url)?.get(database)?.backoff.reset(); + }, + cancel(url, database) { + drop(url, database); + }, + cancelUrl(url) { + const forUrl = schedules.get(url); + if (!forUrl) return; + for (const schedule of forUrl.values()) clearTimeout(schedule.timer); + schedules.delete(url); + }, + pendingCount() { + let count = 0; + for (const forUrl of schedules.values()) for (const schedule of forUrl.values()) if (schedule.timer) count++; + return count; + }, + }; +} + +export function dispatchSubscriptionNodes( + nodes: any[], + deps: { + startTime?: number; + nodeName?: string; + now?: () => number; + dispatch: (nodes: any[]) => void; + consume: () => void; + } +) { + if (deps.startTime === undefined) { + deps.dispatch(nodes); + return; + } + const dispatchNodes = [ + ...nodes, + { + replicateByDefault: nodes[0]?.replicateByDefault, + name: deps.nodeName, + startTime: deps.startTime, + endTime: (deps.now ?? Date.now)(), + replicates: true, + }, + ]; + deps.dispatch(dispatchNodes); + deps.consume(); +} + +/** + * Send the subscribe-to-node the scheduler armed. The entry can have been unsubscribed, deleted, or + * reassigned to another worker during the (now up to 30s) wait, so ownership is re-checked against + * live state and the message goes to the entry's current worker. `connected` is deliberately NOT + * re-checked: a re-subscribe after an unsubscribe is legitimately scheduled while the closing connection + * still reads connected:true, and a redundant subscribe on a live connection is a no-op reuse. + */ +function dispatchSubscribeSetup(url: string, database: string, nodes: any[]) { + const entry = connectionReplicationMap.get(url)?.get(database); + if (!entry || entry.unsubscribed || !nodes[0]) return; + const startTime = env.get(CONFIG_PARAMS.REPLICATION_FAILOVER) ? selfCatchupOfDatabase.get(database) : undefined; + dispatchSubscriptionNodes(nodes, { + startTime, + nodeName: getThisNodeName(), + dispatch(dispatchNodes) { + const request = { ...dispatchNodes[0], type: 'subscribe-to-node', database, nodes: dispatchNodes }; + if (entry.worker) entry.worker.postMessage(request); + else subscribeToNode(request); + }, + consume() { + selfCatchupOfDatabase.delete(database); + }, + }); +} + +const subscribeSetupScheduler = createSubscribeSetupScheduler({ dispatch: dispatchSubscribeSetup }); + // Resolve an auditStore for a database (any table's will do — the per-(db, peer) shared-memory status // buffer is keyed by database, not table) so the main thread can read the authoritative connection truth // written by the owning worker. See W1 (harper-pro#431). @@ -199,6 +389,10 @@ export function clearWorkerFromEntries(connectionMap: Map stalledAtWatermark) + return { fire: false, releaseThrottle: false }; + return { fire: true, releaseThrottle: false }; +} export function findStaleNodeUrls(connectionMap: Map, httpWorkers: any[]): Set { const staleNodeUrls = new Set(); // No live workers to reassign to — flagging here would cause endless no-op reassignments. @@ -675,13 +891,16 @@ export async function startOnMainThread(options) { } } if (!dbReplicationWorkers || !url) return; - for (const [database, { worker, nodes }] of dbReplicationWorkers) { + for (const [database, entry] of dbReplicationWorkers) { + const { worker, nodes } = entry; + clearTimeout(entry.reDriveTimer); dbReplicationWorkers.delete(database); logger.warn('Node was deleted, unsubscribing from node', hostname, database, url); worker?.postMessage({ type: 'unsubscribe-from-node', node: hostname, nodes, database, url }); } dbReplicationWorkers.iterator?.remove(); connectionReplicationMap.delete(url); + subscribeSetupScheduler.cancelUrl(url); return; } if (isSelf) return; @@ -712,6 +931,11 @@ export async function startOnMainThread(options) { break; } } + // A node that moved to a new URL leaves its old URL's entry behind (pre-existing); at least do + // not let a setup armed for the address it left fire against it. + const previousNode = nodeMap.get(node.name); + const previousUrl = previousNode && getNodeURL(previousNode); + if (previousUrl && previousUrl !== getNodeURL(node)) subscribeSetupScheduler.cancelUrl(previousUrl); nodeMap.set(node.name, node); } const databases = getDatabases(); @@ -774,22 +998,13 @@ export async function startOnMainThread(options) { // does for table-exclusion. undefined when no config route matches this peer. harper-pro#498. const configRouteReplicates = matchingRoute ? matchingRoute.replicates : undefined; const nodes = [{ replicateByDefault: tablesReplicateByDefault, ...node, routeReplicates, configRouteReplicates }]; - // Self catchup is done in case we have replicated any records that weren't actually written to our storage - // before a crash. - if (selfCatchupOfDatabase.has(databaseName) && env.get(CONFIG_PARAMS.REPLICATION_FAILOVER)) { - // if we have a self catchup (only do if we have failover enabled), we need to add this node to the list of nodes that need to catch up - // and then we will remove it when it is done - nodes.push({ - replicateByDefault: tablesReplicateByDefault, - name: getThisNodeName(), - startTime: selfCatchupOfDatabase.get(databaseName), - endTime: Date.now(), - replicates: true, - }); - selfCatchupOfDatabase.delete(databaseName); - } // Use the enriched payload (nodes[0]) so the receive gate sees configRouteReplicates. const shouldSubscribe = shouldReplicateFromNode(nodes[0] as any, databaseName); + // Resolve the URL here rather than at the subscribe-scheduling site below: this array becomes + // `entry.nodes`, and the early-return path replaces it without ever reaching that site — which + // left the wedge re-drive posting a request with no url ("Failed to create web socket to + // undefined"). Placed after shouldReplicateFromNode so its verdict is unchanged. + nodes[0].url ??= getNodeURL(nodes[0] as any); const httpWorkers = workers.filter((worker) => worker.name === 'http'); // Defensively detect entries that point at a worker no longer in the http pool. // This happens when the worker.on('exit') handler below never fired (hung WebSocket @@ -800,11 +1015,15 @@ export async function startOnMainThread(options) { // entry stuck and the subscription never recovers. if (existingEntry && httpWorkers.length > 0 && !httpWorkers.includes(existingEntry.worker as any)) { logger.warn(`Subscription for ${databaseName} on node ${node.name} has no live worker; reassigning`); + // The armed recovery closes over the worker being replaced; its fire-time guard would no-op, + // but until then it retains an exited Worker and its request per entry. + clearTimeout(existingEntry.reDriveTimer); dbReplicationWorkers.delete(databaseName); existingEntry = undefined; } if (existingEntry) { worker = existingEntry.worker; + nodes[0].isLeader = nodes[0].isLeader || existingEntry.nodes?.[0]?.isLeader; existingEntry.nodes = nodes; // Normally an existing subscribed entry is left alone. Only the wedge reconcile passes // forceResubscribe for a connection that has been connected:false past the threshold: that @@ -817,6 +1036,8 @@ export async function startOnMainThread(options) { !existingEntry.unsubscribed && !(forceResubscribe && existingEntry.connected === false) ) { + // An armed setup would otherwise fire with the payload from before this update. + subscribeSetupScheduler.refreshPending(getNodeURL(node), databaseName, nodes); return; } if (shouldSubscribe && existingEntry.unsubscribed) { @@ -861,7 +1082,6 @@ export async function startOnMainThread(options) { .filter((nodeName) => nodeName !== getThisNodeName()) // find the first node that is not this one )[0]; // try to find the first node const nodeName = nodes[0].name ?? (nodes[0].url && new URL(nodes[0].url).hostname); - logger.warn(`Setting up subscription with leader ${leaderName} for node ${nodeName}`); // isLeader is true only if: // 1. it was explicitly persisted (e.g. by add_node { isLeader: true }), OR // 2. there is no leader candidate at all, OR @@ -869,23 +1089,15 @@ export async function startOnMainThread(options) { // We deliberately do NOT honour nodeName === leaderName when leaderName came // from the "first other node in hdb_nodes" fallback — that's just a guess. nodes[0].isLeader = nodes[0].isLeader || !leaderName || (hasExplicitLeader && nodeName === leaderName); - nodes[0].url ??= getNodeURL(nodes[0]); // Stagger the subscribe when reassigning (subscribeStagger set) so N databases on one peer - // don't dial N catchup connections simultaneously; otherwise use the flat delay. See #446. - const subscribeDelay = subscribeStagger - ? NODE_SUBSCRIBE_DELAY + subscribeStagger.count++ * RECONNECT_STAGGER_MS - : NODE_SUBSCRIBE_DELAY; - setTimeout(() => { - const request = { - ...nodes[0], - type: 'subscribe-to-node', - database: databaseName, - nodes, - }; - if (worker) { - worker.postMessage(request); - } else subscribeToNode(request); - }, subscribeDelay); + // don't dial N catchup connections simultaneously. See #446. + const staggerMs = subscribeStagger ? subscribeStagger.count++ * RECONNECT_STAGGER_MS : 0; + const subscribeDelay = subscribeSetupScheduler.schedule(getNodeURL(node), databaseName, nodes, staggerMs); + // The warn belongs to an armed setup, not to an event, or a re-drive storm reproduces the + // 165k-line logs of harper-pro#327 even though the work itself is now bounded. + if (subscribeDelay !== undefined) { + logger.warn(`Setting up subscription with leader ${leaderName} for node ${nodeName} in ${subscribeDelay}ms`); + } } else { logger.info('Node no longer should be used, unsubscribing from node', { replicates: node.replicates, @@ -924,6 +1136,11 @@ export async function startOnMainThread(options) { // Keep the entry for URL/iterator cleanup, but bypass the reuse fast path after an // explicit unsubscribe so restoring membership can schedule subscribe-to-node again. if (existingEntry) existingEntry.unsubscribed = true; + subscribeSetupScheduler.cancel(getNodeURL(node), databaseName); + if (existingEntry) { + clearTimeout(existingEntry.reDriveTimer); + existingEntry.reDriveTimer = undefined; + } const request = { type: 'unsubscribe-from-node', database: databaseName, @@ -970,15 +1187,13 @@ export async function startOnMainThread(options) { return; } const mainNode: any = existingWorkerEntry.nodes[0]; - if ( - !( - mainNode.replicates === true || - mainNode.replicates?.sends || - mainNode.replicates?.sendsTo?.length || - mainNode.replicates?.receivesFrom?.length || - mainNode.subscriptions?.length - ) - ) { + if (!( + mainNode.replicates === true || + mainNode.replicates?.sends || + mainNode.replicates?.sendsTo?.length || + mainNode.replicates?.receivesFrom?.length || + mainNode.subscriptions?.length + )) { // no replication, so just return return; } @@ -1037,6 +1252,8 @@ export async function startOnMainThread(options) { return; } mainWorkerEntry.connected = true; + mainWorkerEntry.connectGeneration = (mainWorkerEntry.connectGeneration ?? 0) + 1; + subscribeSetupScheduler.noteConnected(connection.url, connection.database); mainWorkerEntry.disconnectedAt = undefined; mainWorkerEntry.latency = connection.latency; const restoredNode = mainWorkerEntry.nodes[0]; @@ -1181,6 +1398,32 @@ export async function startOnMainThread(options) { getReceiveStatus ); if (staleNodeUrls.size === 0 && wedgedNodeUrls.size === 0 && stalledByUrl.size === 0) return; + // One armed recovery per entry: a sweep staggered across many databases can still be firing when the + // next reconcile decides again, and a superseded wave's timers would otherwise be retained until + // they no-op. + const armReDrive = (entry: any, delay: number, url: string, database: string, fire: () => void) => { + clearTimeout(entry.reDriveTimer); + const timer = setTimeout(() => { + entry.reDriveTimer = undefined; + try { + fire(); + } catch (error) { + logger.error('Error dispatching replication recovery for', url, database, error); + } + }, delay); + timer.unref(); + return timer; + }; + // Decorrelation only, no escalation: these re-drives are already throttled by the disconnectedAt / + // receiveStallReconnectAt re-stamps, so the ceiling is fixed. ONE draw for the whole sweep, used as a + // common base offset: drawing per entry would make consecutive delays differ by up to +/-200ms and let + // ~4 dials share any 50ms instant, which is precisely the concurrency #446's stagger bounds. + const reDriveBaseDelay = + createBackoff({ + initialMs: NODE_SUBSCRIBE_INITIAL_CEILING_MS, + maxMs: NODE_SUBSCRIBE_INITIAL_CEILING_MS, + minMs: NODE_SUBSCRIBE_DELAY, + }).nextDelay() ?? NODE_SUBSCRIBE_DELAY; if (staleNodeUrls.size > 0) logger.warn( 'Reconciling replication subscriptions for nodes pointing at exited workers:', @@ -1259,10 +1502,18 @@ export async function startOnMainThread(options) { ); entry.lastRecovery = { mechanism: 'wedge-reconcile', at: reconcileNow }; // Stagger reconnects (RECONNECT_STAGGER_MS apart) so opening N TLS connections - // simultaneously does not spike memory when there are many databases. - const delay = NODE_SUBSCRIBE_DELAY + reconnectCount * RECONNECT_STAGGER_MS; + // simultaneously does not spike memory when there are many databases; jitter the base so + // every node in a fleet reacting to the same peer outage doesn't fire on the same tick. + const delay = reDriveBaseDelay + reconnectCount * RECONNECT_STAGGER_MS; reconnectCount++; - setTimeout(() => worker.postMessage(request), delay).unref(); + entry.reDriveTimer = armReDrive(entry, delay, url, databaseName, () => { + // The stamp is our claim on this entry: connectedToNode clears disconnectedAt and a later + // reconcile re-stamps it, either of which means this re-drive is stale and would interrupt + // a connection that has already recovered or been re-driven. + if (entries.get(databaseName) !== entry || entry.disconnectedAt !== reconcileNow) return; + if (entry.unsubscribed) return; // forceReconnect would reopen work we just told the worker to drop + worker.postMessage(request); + }); } if (reconnectCount > 0) logger.warn( @@ -1292,15 +1543,30 @@ export async function startOnMainThread(options) { entry.lastRecovery = { mechanism: 'receive-stall-net', at: now }; // Throttle clock so this entry is not re-kicked until the threshold elapses again. entry.receiveStallReconnectAt = now; + // Watermark this decision was made against, so progress arriving during the delay cancels it, + // and the connect generation so a reconnect inside the delay does too. + const stalledAtWatermark = getReceiveStatus(databaseName, nodes[0]?.name)?.lastReceivedTime; + const stalledAtGeneration = entry.connectGeneration ?? 0; const request = { ...nodes[0], type: 'force-reconnect-node', database: databaseName, nodes, }; - const delay = NODE_SUBSCRIBE_DELAY + reconnectCount * RECONNECT_STAGGER_MS; + const delay = reDriveBaseDelay + reconnectCount * RECONNECT_STAGGER_MS; reconnectCount++; - setTimeout(() => worker.postMessage(request), delay).unref(); + entry.reDriveTimer = armReDrive(entry, delay, url, databaseName, () => { + const verdict = shouldFireStallKick({ + current: entries?.get(databaseName), + armed: entry, + armedAt: now, + armedGeneration: stalledAtGeneration, + stalledAtWatermark, + currentWatermark: getReceiveStatus(databaseName, nodes[0]?.name)?.lastReceivedTime, + }); + if (verdict.releaseThrottle) entry.receiveStallReconnectAt = undefined; + if (verdict.fire) worker.postMessage(request); + }); } if (reconnectCount > 0) logger.warn( @@ -1385,12 +1651,111 @@ export function requestClusterStatus(message?, port?) { // the dynamic import resolves to the cached module with no side effect. Cached after first use. let componentsLoadedPromise: Promise | undefined; function whenWorkerComponentsLoaded(): Promise { - return (componentsLoadedPromise ??= import('../core/server/threads/threadServer.js').then( - (threadServer) => threadServer.whenComponentsLoaded - )); + return (componentsLoadedPromise ??= import('../core/server/threads/threadServer.js') + .then((threadServer) => threadServer.whenComponentsLoaded) + .catch((error) => { + componentsLoadedPromise = undefined; + throw error; + })); +} + +export function createWorkerSubscriptionAdmission(deps: { + whenReady: () => Promise; + key: (message: any) => string; + dispatch: (message: any) => void; + onError: (message: any | undefined, error: unknown) => void; + retry?: (delayMs: number, attempt: () => void) => void; + random?: () => number; +}) { + const retry = deps.retry ?? ((delayMs: number, attempt: () => void) => void setTimeout(attempt, delayMs).unref()); + let ready = false; + let flushScheduled = false; + let retryArmed = false; + let retryBackoff: Backoff | undefined; + const pending = new Map(); + + function dispatch(message: any) { + try { + deps.dispatch(message); + } catch (error) { + deps.onError(message, error); + } + } + function scheduleFlush() { + // One readiness attempt and one armed re-attempt at a time: without the second guard, every message + // arriving during a boot-time load failure starts its own attempt and arms its own timer, and the + // retained continuations grow with input again — the shape this gate exists to bound. + if (flushScheduled || retryArmed) return; + flushScheduled = true; + deps + .whenReady() + .then(() => { + ready = true; + retryBackoff?.reset(); + for (const [key, message] of pending) { + pending.delete(key); + dispatch(message); + } + }) + .catch((error) => deps.onError(undefined, error)) + .finally(() => { + flushScheduled = false; + // A readiness failure would otherwise strand the retained actions until another message + // arrives or the wedge reconcile notices ~30s later — the empty-subscription window this + // gate exists to close (harper-pro#289 / #233). Re-attempt on the shared schedule instead. + if (!ready && pending.size > 0) { + retryBackoff ??= createBackoff({ + initialMs: NODE_SUBSCRIBE_INITIAL_CEILING_MS, + maxMs: NODE_SUBSCRIBE_MAX_DELAY_MS, + minMs: NODE_SUBSCRIBE_DELAY, + random: deps.random, + }); + const delay = retryBackoff.nextDelay(); + if (delay !== undefined) { + retryArmed = true; + retry(delay, () => { + retryArmed = false; + scheduleFlush(); + }); + } + } + }); + } + + return { + submit(message: any) { + if (ready && pending.size === 0) { + dispatch(message); + return; + } + pending.set(deps.key(message), message); + scheduleFlush(); + }, + pendingCount() { + return pending.size; + }, + }; } if (parentPort) { + const subscriptionAdmission = createWorkerSubscriptionAdmission({ + whenReady: whenWorkerComponentsLoaded, + key: (message) => getSubscriptionConnectionKey(message.url, message.nodes?.[0]?.url) + '\0' + message.database, + dispatch(message) { + if (message.type === 'subscribe-to-node') subscribeToNode(message); + else unsubscribeFromNode(message); + }, + onError(message, error) { + if (message) + logger.error( + 'Error applying deferred replication subscription action for', + message.url, + message.database, + error + ); + else logger.error('Error waiting for worker components before replication subscription setup', error); + }, + }); disconnectedFromNode = (connection) => { parentPort.postMessage({ type: 'disconnected-from-node', ...connection }); }; @@ -1404,13 +1769,12 @@ if (parentPort) { // "no subscriptions" close, wedging the (peer, db) until restart (harper-pro#289 / #233). Once // components are loaded the predicate is authoritative. In steady state the promise is already // resolved, so this is effectively synchronous. - whenWorkerComponentsLoaded().then(() => subscribeToNode(message)); + subscriptionAdmission.submit(message); }); onMessageByType('unsubscribe-from-node', (message) => { - // Defer through the same gate as subscribe-to-node so the two stay ordered: a pre-load - // subscribe followed by an unsubscribe must apply in that order (else the deferred subscribe - // would run after the unsubscribe and re-open a connection the main thread already removed). - whenWorkerComponentsLoaded().then(() => unsubscribeFromNode(message)); + // Before readiness no connection can exist, so the latest action for the pair is authoritative; + // after readiness both handlers run inline in parentPort delivery order. + subscriptionAdmission.submit(message); }); onMessageByType('force-reconnect-node', (message) => { // Reconcile-driven recovery for a connected:true / Receiving / no-progress stall. Acts on an diff --git a/unitTests/replication/backoff.test.mjs b/unitTests/replication/backoff.test.mjs new file mode 100644 index 000000000..7338a9738 --- /dev/null +++ b/unitTests/replication/backoff.test.mjs @@ -0,0 +1,110 @@ +/** Coverage for the shared replication backoff schedule (harper-pro#327). */ + +import assert from 'node:assert'; +import { createBackoff } from '#src/replication/backoff'; + +describe('createBackoff', () => { + it('doubles the ceiling per attempt and caps it', () => { + const backoff = createBackoff({ initialMs: 500, maxMs: 4000, jitter: 'none' }); + assert.deepEqual( + [backoff.nextDelay(), backoff.nextDelay(), backoff.nextDelay(), backoff.nextDelay(), backoff.nextDelay()], + [500, 1000, 2000, 4000, 4000] + ); + }); + + it('honours a non-default factor', () => { + const backoff = createBackoff({ initialMs: 100, maxMs: 10_000, factor: 3, jitter: 'none' }); + assert.deepEqual([backoff.nextDelay(), backoff.nextDelay(), backoff.nextDelay()], [100, 300, 900]); + }); + + it('draws full jitter across the whole window, never reaching the ceiling', () => { + const draws = [0, 0.5, 0.999999]; + let i = 0; + const backoff = createBackoff({ initialMs: 1000, maxMs: 1000, random: () => draws[i++] }); + assert.deepEqual([backoff.nextDelay(), backoff.nextDelay(), backoff.nextDelay()], [0, 500, 999]); + }); + + it('floors the jitter window at minMs', () => { + const backoff = createBackoff({ initialMs: 1000, maxMs: 1000, minMs: 400, random: () => 0 }); + assert.equal(backoff.nextDelay(), 400, 'a zero draw still waits the floor'); + assert.equal(createBackoff({ initialMs: 1000, maxMs: 1000, minMs: 400, random: () => 0.5 }).nextDelay(), 700); + }); + + it('keeps the ceiling above minMs even when maxMs is lower', () => { + const backoff = createBackoff({ initialMs: 10, maxMs: 10, minMs: 250, random: () => 0.9 }); + assert.equal(backoff.nextDelay(), 250); + }); + + it('decorrelates two schedules with identical failure timing', () => { + const a = createBackoff({ initialMs: 500, maxMs: 30_000, random: () => 0.1 }); + const b = createBackoff({ initialMs: 500, maxMs: 30_000, random: () => 0.9 }); + const aDelays = [a.nextDelay(), a.nextDelay(), a.nextDelay()]; + const bDelays = [b.nextDelay(), b.nextDelay(), b.nextDelay()]; + assert.notDeepEqual(aDelays, bDelays); + for (let i = 0; i < aDelays.length; i++) assert.ok(aDelays[i] < bDelays[i]); + }); + + it('reset() returns to the first ceiling', () => { + const backoff = createBackoff({ initialMs: 500, maxMs: 30_000, jitter: 'none' }); + backoff.nextDelay(); + backoff.nextDelay(); + assert.equal(backoff.attempts, 2); + assert.equal(backoff.ceiling, 2000); + backoff.reset(); + assert.equal(backoff.attempts, 0); + assert.equal(backoff.nextDelay(), 500); + }); + + it('is never exhausted without a budget or attempt cap', () => { + const backoff = createBackoff({ initialMs: 1, maxMs: 2 }); + for (let i = 0; i < 1000; i++) backoff.nextDelay(); + assert.equal(backoff.exhausted, false); + }); + + it('exhausts on the wall-clock deadline, not on the sum of requested sleeps', () => { + // The delays are never actually awaited here: what exhausts the budget is the clock passing the + // deadline. This is the send-auth reprobe's guarantee — an event-loop stall or a slow resolver + // cannot extend the grace period past what it advertises. + let clock = 0; + const backoff = createBackoff({ initialMs: 500, maxMs: 5000, budgetMs: 30_000, now: () => clock }); + assert.equal(backoff.exhausted, false); + backoff.nextDelay(); + clock = 29_999; + assert.equal(backoff.exhausted, false); + clock = 30_000; + assert.equal(backoff.exhausted, true); + assert.equal(backoff.nextDelay(), undefined); + }); + + it('clamps the last delay so it cannot overshoot the deadline', () => { + let clock = 0; + const backoff = createBackoff({ + initialMs: 5000, + maxMs: 5000, + budgetMs: 1000, + random: () => 0.999999, + now: () => clock, + }); + clock = 700; + assert.equal(backoff.nextDelay(), 300); + }); + + it('restarts the budget clock on reset()', () => { + let clock = 0; + const backoff = createBackoff({ initialMs: 10, maxMs: 10, budgetMs: 100, now: () => clock }); + clock = 150; + assert.equal(backoff.exhausted, true); + backoff.reset(); + assert.equal(backoff.exhausted, false); + }); + + it('exhausts on maxAttempts even when the clock never advances', () => { + const backoff = createBackoff({ initialMs: 10, maxMs: 10, budgetMs: 30_000, maxAttempts: 3, now: () => 0 }); + backoff.nextDelay(); + backoff.nextDelay(); + assert.equal(backoff.exhausted, false); + backoff.nextDelay(); + assert.equal(backoff.exhausted, true); + assert.equal(backoff.nextDelay(), undefined); + }); +}); diff --git a/unitTests/replication/connectReschedulesOnRejection.test.mjs b/unitTests/replication/connectReschedulesOnRejection.test.mjs index c975d4c0c..b101b8434 100644 --- a/unitTests/replication/connectReschedulesOnRejection.test.mjs +++ b/unitTests/replication/connectReschedulesOnRejection.test.mjs @@ -38,7 +38,11 @@ describe('NodeReplicationConnection connect() reschedules when createWebSocket r // url = null makes createWebSocket reject (TypeError: Invalid URL) before any socket exists or any // listener is attached — the exact pre-'open' rejection shape of the #466 wedge. function makeRejectingConnection() { - return new NodeReplicationConnection(null, null, 'db', 'peer'); + const connection = new NodeReplicationConnection(null, null, 'db', 'peer'); + // The retry delay is drawn uniformly under the ceiling (harper-pro#327). Pin the draw to the top of + // the window so a single tick advances past exactly one retry. + connection.random = () => 0.999999; + return connection; } it('a createWebSocket rejection leaves reconnectScheduled=true with a pending retry, not a permanent stuck state', async () => { diff --git a/unitTests/replication/forceReconnect.test.mjs b/unitTests/replication/forceReconnect.test.mjs index df2856b15..5f1682447 100644 --- a/unitTests/replication/forceReconnect.test.mjs +++ b/unitTests/replication/forceReconnect.test.mjs @@ -94,16 +94,19 @@ describe('NodeReplicationConnection.forceReconnect', () => { expect(conn.socket.terminate.callCount).to.equal(0); }); - it('backs off the retry interval on repeated wedges (mirrors the close-handler backoff)', () => { + it('backs off the retry ceiling on repeated wedges (mirrors the close-handler backoff)', () => { const conn = makeConnection(); + conn.random = () => 0.999999; // pin the draw so the doubled ceiling is observable as a wait - conn.forceReconnect(); // retryTime 500 -> 1000 + conn.forceReconnect(); // ceiling 500 -> 1000, draws 499 + expect(conn.retryTime).to.equal(1000); clock.tick(500); expect(conn.connect.callCount).to.equal(1); - conn.forceReconnect(); // retryTime 1000 -> 2000 - clock.tick(999); - expect(conn.connect.callCount, 'second reconnect waits the doubled interval').to.equal(1); + conn.forceReconnect(); // ceiling 1000 -> 2000, draws 999 + expect(conn.retryTime).to.equal(2000); + clock.tick(998); + expect(conn.connect.callCount, 'second reconnect waits the doubled ceiling').to.equal(1); clock.tick(1); expect(conn.connect.callCount).to.equal(2); }); diff --git a/unitTests/replication/nodeUpdateWatcher.test.mjs b/unitTests/replication/nodeUpdateWatcher.test.mjs index 5c2759cc2..90b5257d0 100644 --- a/unitTests/replication/nodeUpdateWatcher.test.mjs +++ b/unitTests/replication/nodeUpdateWatcher.test.mjs @@ -11,11 +11,39 @@ * - the loop restarts after the events iterable ends normally * - a per-event processor throw does NOT tear down the loop (continues consuming) * - the optional `maxRestarts` knob is observed (used here to bound the test) + * - the restart backoff escalates on a failure cycle and only resets on a healthy run (harper-pro#327) */ import { expect } from 'chai'; import { runNodeUpdateWatcher } from '#src/replication/knownNodes'; +// The restart delay is awaited through a bare setTimeout inside the watcher, so the delays it asks for +// are the only observable of its backoff. +function captureTimerDelays() { + const realSetTimeout = globalThis.setTimeout; + const values = []; + let unrefCalls = 0; + globalThis.setTimeout = (fn, ms) => { + values.push(ms); + const timer = realSetTimeout(fn, ms); + const unref = timer.unref.bind(timer); + timer.unref = () => { + unrefCalls++; + return unref(); + }; + return timer; + }; + return { + values, + get unrefCalls() { + return unrefCalls; + }, + restore() { + globalThis.setTimeout = realSetTimeout; + }, + }; +} + function makeAsyncIterableFromArray(items) { return { [Symbol.asyncIterator]() { @@ -95,6 +123,91 @@ describe('runNodeUpdateWatcher restart loop', () => { expect(subscribeCalls).to.equal(2); }); + // harper-pro#327: the marker used to be set the instant subscribe() resolved, so a subscription that + // resolved and then immediately threw counted as a success on every pass, reset the backoff, and + // pinned the restart at restartDelayMs forever. Against that code the delays below are [4, 4, 4, 4]. + it('escalates when subscribe() resolves and iteration immediately throws', async () => { + const delays = captureTimerDelays(); + try { + await runNodeUpdateWatcher(() => {}, { + subscribe: async () => ({ + [Symbol.asyncIterator]: () => ({ + next: async () => { + throw new Error('stream died immediately'); + }, + }), + }), + restartDelayMs: 4, + maxDelayMs: 256, + random: () => 0.999999, + maxRestarts: 5, + }); + } finally { + delays.restore(); + } + + expect(delays.values).to.deep.equal([3, 7, 15, 31]); + expect(delays.unrefCalls).to.equal(4); + }); + + it('resets the backoff once an iteration survives the healthy-uptime threshold', async () => { + let subscribeCalls = 0; + let fakeNow = 0; + const delays = captureTimerDelays(); + try { + await runNodeUpdateWatcher(() => {}, { + subscribe: async () => { + subscribeCalls++; + return { + // The clock advances while ITERATING, which is the only thing that counts as uptime. + [Symbol.asyncIterator]: () => ({ + next: async () => { + fakeNow += 60_000; + return { done: true }; + }, + }), + }; + }, + restartDelayMs: 4, + maxDelayMs: 256, + healthyUptimeMs: 10_000, + now: () => fakeNow, + random: () => 0.999999, + maxRestarts: 4, + }); + } finally { + delays.restore(); + } + + expect(subscribeCalls).to.equal(4); + expect(delays.values, 'a healthy run never escalates').to.deep.equal([3, 3, 3]); + }); + + // The health clock must start when the subscription goes live, not when the attempt begins: a + // subscribe() that blocks past the threshold and then throws was never a live watcher. + it('does not count time spent acquiring a subscription as uptime', async () => { + let fakeNow = 0; + const delays = captureTimerDelays(); + try { + await runNodeUpdateWatcher(() => {}, { + subscribe: async () => { + fakeNow += 60_000; // a slow acquire, then a failure + throw new Error('subscribe timed out'); + }, + restartDelayMs: 4, + maxDelayMs: 256, + healthyUptimeMs: 10_000, + now: () => fakeNow, + random: () => 0.999999, + maxRestarts: 5, + }); + } finally { + delays.restore(); + } + + expect(delays.values, 'a slow failed acquire still escalates').to.deep.equal([3, 7, 15, 31]); + }); + it('forwards events to the listener via the default processEvent path', async () => { // Smoke test the default processor at the parameter-passing level using an // injected subscribe that yields a single put event for a foreign node. diff --git a/unitTests/replication/reconnectJitter.test.mjs b/unitTests/replication/reconnectJitter.test.mjs new file mode 100644 index 000000000..7045a171c --- /dev/null +++ b/unitTests/replication/reconnectJitter.test.mjs @@ -0,0 +1,130 @@ +/** + * Coverage for the jittered reconnect schedule (harper-pro#327). `scheduleReconnect` used to wait + * exactly `retryTime`, so every node reacting to the same peer outage re-dialed on the same instant. + * The ceiling schedule is unchanged; full jitter decorrelates the fleet, while the fixed 500 ms floor + * retains the hard minimum interval from the native TLS-state incident (harper-pro#339). + * + * `null` for the url makes createWebSocket reject before any socket or listener exists, which is the + * cheapest way to drive the real scheduleReconnect path without a peer. + */ + +import assert from 'node:assert'; +import sinon from 'sinon'; +import { NodeReplicationConnection } from '#src/replication/replicationConnection'; + +const INITIAL_RETRY_TIME = 500; +const MAX_RETRY_TIME = 30_000; + +function captureTimerDelays() { + const realSetTimeout = globalThis.setTimeout; + const values = []; + globalThis.setTimeout = (fn, ms) => { + values.push(ms); + return realSetTimeout(fn, ms); + }; + return { + values, + restore() { + globalThis.setTimeout = realSetTimeout; + }, + }; +} + +function scheduleDelays(connection, attempts) { + const timers = captureTimerDelays(); + try { + for (let i = 0; i < attempts; i++) { + connection.reconnectScheduled = false; // the real clear happens when connect() installs a socket + connection.scheduleReconnect(); + } + } finally { + timers.restore(); + } + return timers.values; +} + +describe('NodeReplicationConnection reconnect jitter (harper-pro#327)', () => { + let clock; + + beforeEach(() => { + clock = sinon.useFakeTimers(); + }); + + afterEach(() => { + // clock.restore() only — a sandbox-wide sinon.restore() here re-restores the stale + // globalThis.setTimeout that receiveWatchdog.test.mjs's manually-restored spy left registered, + // which silently breaks real timers for every file that runs after this one. + clock.restore(); + }); + + function makeConnection(random) { + const connection = new NodeReplicationConnection(null, null, 'db', 'peer'); + connection.random = random; + return connection; + } + + it('two connections failing on identical timing get decorrelated delays', () => { + const early = scheduleDelays( + makeConnection(() => 0.1), + 6 + ); + const late = scheduleDelays( + makeConnection(() => 0.9), + 6 + ); + + assert.notDeepEqual(early, late); + assert.equal(early[0], INITIAL_RETRY_TIME); + for (let i = 1; i < early.length; i++) assert.ok(early[i] < late[i]); + }); + + it('keeps the unchanged 500ms → 30s ceiling schedule, drawing inside it', () => { + const connection = makeConnection(() => 0.999999); + const ceilings = []; + const timers = captureTimerDelays(); + try { + for (let i = 0; i < 8; i++) { + connection.reconnectScheduled = false; + connection.scheduleReconnect(); + ceilings.push(connection.retryTime); + } + } finally { + timers.restore(); + } + + assert.deepEqual(ceilings, [1000, 2000, 4000, 8000, 16_000, 30_000, 30_000, 30_000]); + const ceilingFor = (i) => Math.min(INITIAL_RETRY_TIME * 2 ** i, MAX_RETRY_TIME); + timers.values.forEach((delay, i) => { + assert.ok(delay >= INITIAL_RETRY_TIME); + if (ceilingFor(i) === INITIAL_RETRY_TIME) assert.equal(delay, INITIAL_RETRY_TIME); + else assert.ok(delay < ceilingFor(i)); + }); + }); + + it('a zero draw still waits the fixed TLS-safety floor', () => { + assert.deepEqual( + scheduleDelays( + makeConnection(() => 0), + 4 + ), + [500, 500, 500, 500] + ); + }); + + it('retryTime reads as the initial interval before any failure', () => { + assert.equal(makeConnection(Math.random).retryTime, INITIAL_RETRY_TIME); + }); + + it('onFrameSent resets the ceiling and the retry counter', () => { + const connection = makeConnection(() => 0.5); + scheduleDelays(connection, 3); + connection.retries = 7; + assert.equal(connection.retryTime, 4000); + + connection.onFrameSent(); + + assert.equal(connection.retries, 0); + assert.equal(connection.retryTime, INITIAL_RETRY_TIME); + assert.equal(scheduleDelays(connection, 1)[0], 500, 'drawing under the initial ceiling again'); + }); +}); diff --git a/unitTests/replication/shouldCloseSendAuthWatch.test.mjs b/unitTests/replication/shouldCloseSendAuthWatch.test.mjs index 6e92734c8..22aa0e3c9 100644 --- a/unitTests/replication/shouldCloseSendAuthWatch.test.mjs +++ b/unitTests/replication/shouldCloseSendAuthWatch.test.mjs @@ -88,6 +88,68 @@ describe('shouldCloseSendAuthWatch', () => { expect(timedOut).to.equal(true); }); + // harper-pro#327: the grace period is advertised as a 30s deadline, so a row that only becomes + // decodable after the loop was suspended past it must not authorize. Checking `exhausted` only at the + // top of the loop was not enough — the loop exits on a non-UNCHANGED row without ever consulting it. + it('fails closed on a row that only resolves after the budget deadline passed', async () => { + let clock = 0; + let timedOut = false; + let resolved = 0; + const shouldClose = await shouldCloseSendAuthWatch({ type: 'put' }, 'node-a', 'data', { + isClosed: neverClosed, + // An event-loop stall: the sleep returns long after the delay it was asked for. + sleep: async () => { + clock += 45_000; + }, + now: () => clock, + reprobeBudgetMs: 30_000, + resolve: () => (resolved++ === 0 ? SEND_AUTH_UNCHANGED : { name: 'node-a', replicates: true }), + onReprobeTimeout: () => { + timedOut = true; + }, + }); + expect(shouldClose, 'the late authorizing row is not consulted').to.equal(true); + expect(timedOut).to.equal(true); + expect(resolved, 'no probe after the deadline').to.equal(1); + }); + + // The store read is synchronous, so it can carry the clock past the deadline by itself — the sleep is + // not the only thing that can overrun the advertised grace period. + it('fails closed when the row read itself crosses the deadline', async () => { + let clock = 0; + let timedOut = false; + let resolved = 0; + const shouldClose = await shouldCloseSendAuthWatch({ type: 'put' }, 'node-a', 'data', { + isClosed: neverClosed, + sleep: async () => { + clock += 29_000; + }, + now: () => clock, + reprobeBudgetMs: 30_000, + resolve: () => { + if (resolved++ === 0) return SEND_AUTH_UNCHANGED; + clock += 5000; // the point read blocks past the deadline, then returns an authorizing row + return { name: 'node-a', replicates: true }; + }, + onReprobeTimeout: () => { + timedOut = true; + }, + }); + expect(shouldClose, 'a row that arrived after the deadline does not authorize').to.equal(true); + expect(timedOut).to.equal(true); + }); + + it('does not read the clock at all for an immediately decodable row', async () => { + const shouldClose = await shouldCloseSendAuthWatch({ type: 'put' }, 'node-a', 'data', { + isClosed: neverClosed, + now: () => { + throw new Error('the backoff must not be built for a decodable row'); + }, + resolve: () => ({ name: 'node-a', replicates: true }), + }); + expect(shouldClose).to.equal(false); + }); + it('does not close if the connection closes while the reprobe loop is still suspended', async () => { let closed = false; let timedOut = false; diff --git a/unitTests/replication/shouldFireStallKick.test.mjs b/unitTests/replication/shouldFireStallKick.test.mjs new file mode 100644 index 000000000..a86fe7b55 --- /dev/null +++ b/unitTests/replication/shouldFireStallKick.test.mjs @@ -0,0 +1,62 @@ +import assert from 'node:assert'; +import { shouldFireStallKick } from '#src/replication/subscriptionManager'; + +const ARMED_AT = 1_000; + +function makeEntry(overrides = {}) { + return { receiveStallReconnectAt: ARMED_AT, connectGeneration: 3, ...overrides }; +} + +function verdict(entry, overrides = {}) { + return shouldFireStallKick({ + current: entry, + armed: entry, + armedAt: ARMED_AT, + armedGeneration: 3, + stalledAtWatermark: 500, + currentWatermark: 500, + ...overrides, + }); +} + +describe('shouldFireStallKick', () => { + it('fires when the entry is unchanged and no data has arrived', () => { + assert.deepEqual(verdict(makeEntry()), { fire: true, releaseThrottle: false }); + }); + + it('does not fire for a superseded entry, and does not release a stamp it does not own', () => { + const entry = makeEntry(); + assert.deepEqual(verdict(entry, { current: makeEntry() }), { fire: false, releaseThrottle: false }); + }); + + it('does not fire once a newer reconcile re-stamped the throttle', () => { + assert.deepEqual(verdict(makeEntry({ receiveStallReconnectAt: 2000 })), { fire: false, releaseThrottle: false }); + }); + + it('does not fire for an entry that has since unsubscribed', () => { + assert.deepEqual(verdict(makeEntry({ unsubscribed: true })), { fire: false, releaseThrottle: false }); + }); + + // The regression: skipping on a reconnect used to keep the stamp, and a fresh socket that also stalls + // never moves lastReceivedTime past it — so the net never re-armed for that pair again. + it('releases the throttle when the leg reconnected inside the stagger window', () => { + assert.deepEqual(verdict(makeEntry({ connectGeneration: 4 })), { fire: false, releaseThrottle: true }); + }); + + it('treats a missing connectGeneration as generation 0', () => { + const entry = { receiveStallReconnectAt: ARMED_AT }; + assert.deepEqual(verdict(entry, { armedGeneration: 0 }), { fire: true, releaseThrottle: false }); + assert.deepEqual(verdict(entry, { armedGeneration: 1 }), { fire: false, releaseThrottle: true }); + }); + + // Progress means the stall resolved on its own: no kick is owed and the stamp correctly records the + // epoch, so it must NOT be released. + it('does not fire, or release, when the watermark advanced', () => { + assert.deepEqual(verdict(makeEntry(), { currentWatermark: 900 }), { fire: false, releaseThrottle: false }); + }); + + it('fires when either watermark is unavailable', () => { + assert.deepEqual(verdict(makeEntry(), { currentWatermark: undefined }), { fire: true, releaseThrottle: false }); + assert.deepEqual(verdict(makeEntry(), { stalledAtWatermark: undefined }), { fire: true, releaseThrottle: false }); + }); +}); diff --git a/unitTests/replication/subscribeSetupScheduler.test.mjs b/unitTests/replication/subscribeSetupScheduler.test.mjs new file mode 100644 index 000000000..10d6b019d --- /dev/null +++ b/unitTests/replication/subscribeSetupScheduler.test.mjs @@ -0,0 +1,285 @@ +/** + * Regression coverage for harper-pro#327: subscription setup used to be scheduled with a flat 200ms + * `setTimeout` per qualifying node update — no dedup, no cap, not unref'd — so whatever re-drove + * `onNodeUpdate` amplified 1:1 into main-thread timers, worker-side WebSocket/TLS setup, and warn + * lines, ending in an OOM kill. + * + * The storm test drives input continuously *across* timer firings, which is the shape of the incident; + * a one-shot burst would only prove coalescing. Against the flat-200ms behavior the same input + * produces one dispatch per event and one live timer per event, so both assertions go red. + */ + +import assert from 'node:assert'; +import sinon from 'sinon'; +import { createSubscribeSetupScheduler, dispatchSubscriptionNodes } from '#src/replication/subscriptionManager'; + +const URL_A = 'wss://peer-a:9933'; +const URL_B = 'wss://peer-b:9933'; +// Mirrors the production constants: floor NODE_SUBSCRIBE_DELAY, first ceiling 2x that, cap 30s. +const MIN_DELAY = 200; +const MAX_DELAY = 30_000; + +const NODES = [{ name: 'peer-a', url: URL_A }]; + +function makeScheduler(random) { + const dispatches = []; + const scheduler = createSubscribeSetupScheduler({ + dispatch: (url, database, nodes) => dispatches.push({ url, database, nodes, at: Date.now() }), + random, + }); + return { scheduler, dispatches }; +} + +describe('subscription-setup scheduler (harper-pro#327)', () => { + let clock; + + beforeEach(() => { + clock = sinon.useFakeTimers(); + }); + + afterEach(() => { + // clock.restore() only — a sandbox-wide sinon.restore() here re-restores the stale + // globalThis.setTimeout that receiveWatchdog.test.mjs's manually-restored spy left registered, + // which silently breaks real timers for every file that runs after this one. + clock.restore(); + }); + + it('bounds a 60s re-drive storm to a handful of setups with one pending timer throughout', () => { + const { scheduler, dispatches } = makeScheduler(() => 0.5); + let maxPending = 0; + + // 60,000 qualifying updates spread over 60s of simulated time (~1,000/s, the observed storm rate). + for (let i = 0; i < 60_000; i++) { + scheduler.schedule(URL_A, 'data', NODES); + maxPending = Math.max(maxPending, scheduler.pendingCount()); + clock.tick(1); + } + + // Ceilings double 400 → 30,000; each delay is 200 + 0.5 * (ceiling - 200). + assert.deepEqual( + dispatches.map((d) => d.at), + [300, 800, 1700, 3400, 6700, 13_200, 26_100, 41_200, 56_300] + ); + assert.equal(maxPending, 1, 'never more than one pending setup for the pair'); + assert.equal(scheduler.pendingCount(), 1, 'exactly one still armed at the end'); + }); + + it('keeps every delay inside the floor and the cap', () => { + const draws = [0, 0.999999, 0.25, 0, 0.999999, 0.5, 0.75, 0, 0.999999, 0.999999, 0.999999, 0.999999]; + let i = 0; + const { scheduler } = makeScheduler(() => draws[i++ % draws.length]); + for (let attempt = 0; attempt < 40; attempt++) { + const delay = scheduler.schedule(URL_A, 'data', NODES); + assert.ok(delay >= MIN_DELAY); + assert.ok(delay < MAX_DELAY); + clock.tick(delay); + } + }); + + it('returns undefined instead of arming a second timer for the same pair', () => { + const { scheduler, dispatches } = makeScheduler(() => 0.5); + assert.equal(scheduler.schedule(URL_A, 'data', NODES), 300); + assert.equal(scheduler.schedule(URL_A, 'data', NODES), undefined, 'deduped'); + assert.equal(scheduler.schedule(URL_A, 'data', NODES), undefined, 'still deduped'); + clock.tick(60_000); + assert.equal(dispatches.length, 1); + }); + + it('tracks (url, database) pairs independently', () => { + const { scheduler, dispatches } = makeScheduler(() => 0.5); + assert.equal(scheduler.schedule(URL_A, 'data', NODES), 300); + assert.equal(scheduler.schedule(URL_A, 'other', NODES), 300); + assert.equal(scheduler.schedule(URL_B, 'data', NODES), 300); + assert.equal(scheduler.pendingCount(), 3); + clock.tick(300); + assert.deepEqual( + dispatches.map(({ url, database, at }) => ({ url, database, at })), + [ + { url: URL_A, database: 'data', at: 300 }, + { url: URL_A, database: 'other', at: 300 }, + { url: URL_B, database: 'data', at: 300 }, + ] + ); + }); + + it('decorrelates two peers failing on identical timing', () => { + const a = createSubscribeSetupScheduler({ dispatch: () => {}, random: () => 0.1 }); + const b = createSubscribeSetupScheduler({ dispatch: () => {}, random: () => 0.9 }); + const aDelays = []; + const bDelays = []; + for (let attempt = 0; attempt < 5; attempt++) { + aDelays.push(a.schedule(URL_A, 'data', NODES)); + bDelays.push(b.schedule(URL_A, 'data', NODES)); + clock.tick(MAX_DELAY + MIN_DELAY); + } + assert.notDeepEqual(aDelays, bDelays); + for (let i = 0; i < aDelays.length; i++) assert.ok(aDelays[i] < bDelays[i]); + }); + + // The regression that made the deduped path lose the enriched payload: onDatabase replaces + // entry.nodes on its early-return path without running the leader/url enrichment, so the setup has + // to carry the payload of the call that armed or last refreshed it, not whatever the entry holds. + it('fires with the newest payload a deduped call supplied', () => { + const { scheduler, dispatches } = makeScheduler(() => 0.5); + const first = [{ name: 'peer-a', url: URL_A }]; + const second = [{ name: 'peer-a', url: URL_A, isLeader: true }]; + scheduler.schedule(URL_A, 'data', first); + assert.equal(scheduler.schedule(URL_A, 'data', second), undefined); + clock.tick(300); + assert.equal(dispatches.length, 1); + assert.equal(dispatches[0].nodes, second); + }); + + // onDatabase's early-return path (an already-subscribed, still-desired entry) never reaches + // schedule(), but it does build a fresh payload — so the armed setup has to be told about it or it + // dispatches routing/exclusion state from before the update. + it('refreshPending replaces the payload of an armed setup without arming one', () => { + const { scheduler, dispatches } = makeScheduler(() => 0.5); + const armed = [{ name: 'peer-a', url: URL_A }]; + const refreshed = [{ name: 'peer-a', url: URL_A, routeReplicates: { receives: true } }]; + scheduler.schedule(URL_A, 'data', armed); + scheduler.refreshPending(URL_A, 'data', refreshed); + assert.equal(scheduler.pendingCount(), 1, 'no second timer'); + clock.tick(300); + assert.equal(dispatches.length, 1); + assert.equal(dispatches[0].nodes, refreshed); + }); + + it('refreshPending on a pair with nothing armed does not arm one', () => { + const { scheduler, dispatches } = makeScheduler(() => 0.5); + scheduler.refreshPending(URL_A, 'data', NODES); + assert.equal(scheduler.pendingCount(), 0); + clock.tick(60_000); + assert.deepEqual(dispatches, []); + }); + + it('adds the caller-supplied stagger on top of the backoff', () => { + const { scheduler } = makeScheduler(() => 0.5); + assert.equal(scheduler.schedule(URL_A, 'data', NODES, 150), 450); + }); + + // A connect report cannot be attributed to the entry that armed the setup (failover subscribes on a + // worker that is not entry.worker, and a superseded worker still reports for the same pair), so the + // armed setup is left to fire and only the escalated delay is dropped. Escalating across reconnect + // cycles instead is what pushed a chaos-restart peer past its 25s reconvergence budget. + it('noteConnected resets the escalated delay and leaves the armed setup to fire', () => { + const { scheduler, dispatches } = makeScheduler(() => 0.5); + scheduler.schedule(URL_A, 'data', NODES); + clock.tick(300); + scheduler.schedule(URL_A, 'data', NODES); // second attempt: escalated to 500 + assert.equal(scheduler.pendingCount(), 1); + + scheduler.noteConnected(URL_A, 'data'); + assert.equal(scheduler.pendingCount(), 1, 'still armed'); + clock.tick(60_000); + assert.equal(dispatches.length, 2, 'the armed setup fired'); + + assert.equal(scheduler.schedule(URL_A, 'data', NODES), 300, 'back to the first ceiling after success'); + }); + + it('a pair that reconnects every cycle never escalates', () => { + const { scheduler } = makeScheduler(() => 0.5); + const delays = []; + for (let cycle = 0; cycle < 5; cycle++) { + delays.push(scheduler.schedule(URL_A, 'data', NODES)); + clock.tick(1000); + scheduler.noteConnected(URL_A, 'data'); + } + assert.deepEqual(delays, [300, 300, 300, 300, 300]); + }); + + it('cancel() and cancelUrl() disarm pending setups', () => { + const { scheduler, dispatches } = makeScheduler(() => 0.5); + scheduler.schedule(URL_A, 'data', NODES); + scheduler.schedule(URL_A, 'other', NODES); + scheduler.schedule(URL_B, 'data', NODES); + + scheduler.cancel(URL_A, 'data'); + assert.equal(scheduler.pendingCount(), 2); + scheduler.cancelUrl(URL_A); + assert.equal(scheduler.pendingCount(), 1); + + clock.tick(60_000); + assert.deepEqual( + dispatches.map(({ url, database, at }) => ({ url, database, at })), + [{ url: URL_B, database: 'data', at: 300 }] + ); + }); + + it('a throwing dispatch is contained instead of taking the process down', () => { + const scheduler = createSubscribeSetupScheduler({ + dispatch: () => { + throw new Error('uncloneable payload'); + }, + random: () => 0.5, + }); + scheduler.schedule(URL_A, 'data', NODES); + assert.doesNotThrow(() => clock.tick(300)); + assert.equal(scheduler.pendingCount(), 0, 'ownership released so the pair can be re-armed'); + assert.equal(scheduler.schedule(URL_A, 'data', NODES), 500); + }); +}); + +// Real Node timers, deliberately: hasRef() is what proves the process can still exit with a setup +// pending, and a faked timer has no such thing. +describe('subscription-setup scheduler timer refs', () => { + it("unref's the setup timer so a pending retry cannot hold the process open", () => { + const realSetTimeout = globalThis.setTimeout; + let armed; + globalThis.setTimeout = (fn, ms) => (armed = realSetTimeout(fn, ms)); + try { + createSubscribeSetupScheduler({ dispatch: () => {}, random: () => 0.5 }).schedule(URL_A, 'data', NODES); + } finally { + globalThis.setTimeout = realSetTimeout; + } + assert.equal(armed.hasRef(), false); + clearTimeout(armed); + }); +}); + +describe('self-catchup dispatch', () => { + it('attaches the rider on a fresh payload and consumes it only after dispatch', () => { + const nodes = [{ name: 'peer-a', url: URL_A, replicateByDefault: true }]; + let consumed = 0; + let dispatched; + dispatchSubscriptionNodes(nodes, { + startTime: 123, + nodeName: 'self', + now: () => 456, + dispatch: (value) => (dispatched = value), + consume: () => consumed++, + }); + + assert.equal(nodes.length, 1); + assert.deepEqual(dispatched, [ + nodes[0], + { + replicateByDefault: true, + name: 'self', + startTime: 123, + endTime: 456, + replicates: true, + }, + ]); + assert.equal(consumed, 1); + }); + + it('retains the rider when dispatch throws', () => { + const nodes = [{ name: 'peer-a', url: URL_A }]; + let consumed = 0; + assert.throws( + () => + dispatchSubscriptionNodes(nodes, { + startTime: 123, + nodeName: 'self', + dispatch: () => { + throw new Error('postMessage failed'); + }, + consume: () => consumed++, + }), + /postMessage failed/ + ); + assert.equal(nodes.length, 1); + assert.equal(consumed, 0); + }); +}); diff --git a/unitTests/replication/workerSubscriptionAdmission.test.mjs b/unitTests/replication/workerSubscriptionAdmission.test.mjs new file mode 100644 index 000000000..10506c9b6 --- /dev/null +++ b/unitTests/replication/workerSubscriptionAdmission.test.mjs @@ -0,0 +1,203 @@ +import assert from 'node:assert'; +import { setImmediate as waitForTurn } from 'node:timers/promises'; +import { createWorkerSubscriptionAdmission } from '#src/replication/subscriptionManager'; +import { getSubscriptionConnectionKey } from '#src/replication/replicator'; + +function deferred() { + let resolve; + let reject; + const promise = new Promise((resolvePromise, rejectPromise) => { + resolve = resolvePromise; + reject = rejectPromise; + }); + return { promise, resolve, reject }; +} + +describe('worker subscription admission', () => { + it('derives one connection key for setup and teardown when the nested URL is missing', () => { + const connectingUrl = 'wss://peer:9933'; + assert.equal( + getSubscriptionConnectionKey(connectingUrl, undefined), + getSubscriptionConnectionKey(connectingUrl, connectingUrl) + ); + }); + + it('retains one latest pre-readiness action per target', async () => { + const readiness = deferred(); + const actions = []; + const admission = createWorkerSubscriptionAdmission({ + whenReady: () => readiness.promise, + key: (message) => message.key, + dispatch: (message) => actions.push(message), + onError: assert.fail, + }); + const first = { key: 'peer\0data', type: 'subscribe-to-node', generation: 1 }; + const latest = { key: 'peer\0data', type: 'subscribe-to-node', generation: 60_000 }; + admission.submit(first); + for (let generation = 2; generation < latest.generation; generation++) admission.submit({ ...first, generation }); + admission.submit(latest); + + assert.equal(admission.pendingCount(), 1); + assert.deepEqual(actions, []); + readiness.resolve(); + await waitForTurn(); + assert.deepEqual(actions, [latest]); + assert.equal(admission.pendingCount(), 0); + }); + + it('uses the final pre-readiness subscribe or unsubscribe state', async () => { + const readiness = deferred(); + const actions = []; + const admission = createWorkerSubscriptionAdmission({ + whenReady: () => readiness.promise, + key: (message) => message.key, + dispatch: (message) => actions.push(message.type), + onError: assert.fail, + }); + admission.submit({ key: 'peer\0data', type: 'subscribe-to-node' }); + admission.submit({ key: 'peer\0data', type: 'unsubscribe-from-node' }); + readiness.resolve(); + await waitForTurn(); + assert.deepEqual(actions, ['unsubscribe-from-node']); + }); + + it('runs post-readiness actions inline without deriving a key', async () => { + const readiness = deferred(); + const actions = []; + let keyCalls = 0; + const admission = createWorkerSubscriptionAdmission({ + whenReady: () => readiness.promise, + key: (message) => { + keyCalls++; + return message.key; + }, + dispatch: (message) => actions.push(message.generation), + onError: assert.fail, + }); + admission.submit({ key: 'peer\0data', generation: 1 }); + readiness.resolve(); + await waitForTurn(); + admission.submit({ key: 'peer\0data', generation: 2 }); + + assert.deepEqual(actions, [1, 2]); + assert.equal(keyCalls, 1); + }); + + // A rejection used to leave the retained action waiting for another inbound message or the ~30s wedge + // reconcile — the empty-subscription window the gate exists to close. It now re-attempts itself. + it('re-attempts a rejected readiness on an escalating schedule without a new message', async () => { + let attempts = 0; + const actions = []; + const errors = []; + const retries = []; + const admission = createWorkerSubscriptionAdmission({ + whenReady: () => (++attempts < 3 ? Promise.reject(new Error(`load failed ${attempts}`)) : Promise.resolve()), + key: (message) => message.key, + dispatch: (message) => actions.push(message.generation), + onError: (_message, error) => errors.push(error.message), + retry: (delayMs, attempt) => { + retries.push(delayMs); + attempt(); + }, + random: () => 0.999999, + }); + admission.submit({ key: 'peer\0data', generation: 1 }); + await waitForTurn(); + await waitForTurn(); + + assert.deepEqual(errors, ['load failed 1', 'load failed 2']); + assert.deepEqual(retries, [399, 799], 'the re-attempt delay escalates under the shared schedule'); + assert.deepEqual(actions, [1], 'the retained action is applied once readiness succeeds'); + assert.equal(admission.pendingCount(), 0); + }); + + it('does not add a second attempt or timer while a retry is armed', async () => { + let readinessCalls = 0; + const retries = []; + let fire; + const admission = createWorkerSubscriptionAdmission({ + whenReady: () => { + readinessCalls++; + return Promise.reject(new Error('load failed')); + }, + key: (message) => message.key, + dispatch: () => assert.fail('nothing should dispatch'), + onError: () => {}, + retry: (delayMs, attempt) => { + retries.push(delayMs); + fire = attempt; + }, + }); + admission.submit({ key: 'peer\0data' }); + await waitForTurn(); + assert.equal(retries.length, 1); + + for (let i = 0; i < 50; i++) admission.submit({ key: `peer\0db-${i}` }); + await waitForTurn(); + assert.equal(readinessCalls, 1, 'no second readiness attempt while one is armed'); + assert.equal(retries.length, 1, 'no second timer'); + + fire(); + await waitForTurn(); + assert.equal(readinessCalls, 2, 'the armed retry is the one that re-attempts'); + }); + + // Full jitter with no floor would draw near 0 and re-attempt on the next macrotask, turning a boot-time + // component-load failure into a spin of imports and warn lines — the shape this gate exists to bound. + it('never re-attempts below the floor, even on a zero draw', async () => { + const retries = []; + const admission = createWorkerSubscriptionAdmission({ + whenReady: () => Promise.reject(new Error('load failed')), + key: (message) => message.key, + dispatch: () => assert.fail('nothing should dispatch'), + onError: () => {}, + retry: (delayMs) => retries.push(delayMs), + random: () => 0, + }); + admission.submit({ key: 'peer\0data' }); + await waitForTurn(); + assert.deepEqual(retries, [200]); + }); + + it('stops re-attempting once nothing is pending', async () => { + const retries = []; + const admission = createWorkerSubscriptionAdmission({ + whenReady: () => Promise.reject(new Error('load failed')), + key: (message) => message.key, + dispatch: () => assert.fail('nothing should dispatch'), + onError: () => {}, + retry: (delayMs) => retries.push(delayMs), + }); + admission.submit({ key: 'peer\0data' }); + await waitForTurn(); + assert.equal(retries.length, 1); + }); + + it('retains bounded state after a readiness rejection and applies the newest action on the retry', async () => { + const firstReadiness = deferred(); + let attempts = 0; + const actions = []; + const errors = []; + let fire; + const admission = createWorkerSubscriptionAdmission({ + whenReady: () => (++attempts === 1 ? firstReadiness.promise : Promise.resolve()), + key: (message) => message.key, + dispatch: (message) => actions.push(message.generation), + onError: (_message, error) => errors.push(error.message), + retry: (_delayMs, attempt) => (fire = attempt), + }); + admission.submit({ key: 'peer\0data', generation: 1 }); + firstReadiness.reject(new Error('component load failed')); + await waitForTurn(); + assert.equal(admission.pendingCount(), 1); + // A message arriving before the retry fires replaces the retained one rather than adding work. + admission.submit({ key: 'peer\0data', generation: 2 }); + assert.equal(admission.pendingCount(), 1); + fire(); + await waitForTurn(); + + assert.deepEqual(errors, ['component load failed']); + assert.deepEqual(actions, [2]); + assert.equal(admission.pendingCount(), 0); + }); +});