diff --git a/integrationTests/cluster/subscriptionSetupRecovery.test.mjs b/integrationTests/cluster/subscriptionSetupRecovery.test.mjs new file mode 100644 index 000000000..e45d970a8 --- /dev/null +++ b/integrationTests/cluster/subscriptionSetupRecovery.test.mjs @@ -0,0 +1,293 @@ +// harper-pro#642 end-to-end regression: a stuck sender-side setup gate must not leave the connection +// ping-alive forever; the receiver's watchdog must reconnect from the durable cursor and converge. + +import { suite, test, before, after } from 'node:test'; +import assert from 'node:assert/strict'; +import { setTimeout as delay } from 'node:timers/promises'; +import { join } from 'node:path'; +import { startHarper, teardownHarper, getNextAvailableLoopbackAddress } from '@harperfast/integration-testing'; +import { sendOperation, readLog } from './clusterShared.mjs'; + +process.env.HARPER_INTEGRATION_TEST_INSTALL_SCRIPT = join(import.meta.dirname, '..', '..', 'dist', 'bin', 'harper.js'); + +const DB = 'data'; +const TABLE = 'setup_recovery'; +const SETUP_TIMEOUT_MS = 3000; +const RECOVERY_TIMEOUT_MS = 30000; + +function optionsFor(node, env, databases = [DB, 'system']) { + return { + config: { + analytics: { aggregatePeriod: -1 }, + logging: { colors: false, stdStreams: true, console: true, level: 'warn' }, + threads: { count: 1 }, + replication: { + securePort: node.hostname + ':9933', + databases, + pingInterval: 1000, + pingTimeout: 3000, + }, + }, + env, + }; +} + +async function hasRecord(node, id) { + const result = await sendOperation(node, { + operation: 'search_by_id', + database: DB, + table: TABLE, + ids: [id], + get_attributes: ['id'], + }).catch(() => null); + return Array.isArray(result) && result.some((record) => record?.id === id); +} + +async function waitForRecord(node, id, timeoutMs = RECOVERY_TIMEOUT_MS) { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + if (await hasRecord(node, id)) return true; + await delay(250); + } + return false; +} + +async function waitForLog(node, pattern, timeoutMs = RECOVERY_TIMEOUT_MS) { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + const log = await readLog(node); + if (pattern.test(log)) return log; + await delay(250); + } + return ''; +} + +function countSetupWatchdogWarnings(log, database = DB) { + return log + .split('\n') + .filter((line) => line.includes('Subscription-setup watchdog:') && line.includes(`(db: "${database}")`)).length; +} + +async function socketConnected(node, database) { + const status = await sendOperation(node, { operation: 'cluster_status' }); + return status.connections.some((connection) => + connection.database_sockets?.some((socket) => socket.database === database && socket.connected === true) + ); +} + +async function waitForSocket(node, database, timeoutMs = RECOVERY_TIMEOUT_MS) { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + if (await socketConnected(node, database).catch(() => false)) return true; + await delay(250); + } + return false; +} + +async function waitForRole(node, role, timeoutMs = RECOVERY_TIMEOUT_MS) { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + const roles = await sendOperation(node, { operation: 'list_roles' }).catch(() => null); + if (Array.isArray(roles) && roles.some((entry) => entry?.role === role)) return true; + await delay(250); + } + return false; +} + +suite('subscription setup recovery', { timeout: 120000 }, (ctx) => { + before(async () => { + const sourceCtx = { name: ctx.name, harper: { hostname: await getNextAvailableLoopbackAddress() } }; + const receiverCtx = { name: ctx.name, harper: { hostname: await getNextAvailableLoopbackAddress() } }; + await Promise.all([ + startHarper(sourceCtx, optionsFor(sourceCtx.harper, { HARPER_TEST_SUBSCRIPTION_SETUP_STALL_ONCE_DB: DB })), + startHarper( + receiverCtx, + optionsFor(receiverCtx.harper, { HARPER_TEST_SUBSCRIPTION_SETUP_TIMEOUT_MS: String(SETUP_TIMEOUT_MS) }) + ), + ]); + ctx.source = sourceCtx.harper; + ctx.receiver = receiverCtx.harper; + + await Promise.all( + [ctx.source, ctx.receiver].map((node) => + sendOperation(node, { + operation: 'create_table', + database: DB, + table: TABLE, + primary_key: 'id', + attributes: [{ name: 'id', type: 'ID' }], + }) + ) + ); + }); + + after(async () => { + await Promise.all([ctx.source, ctx.receiver].filter(Boolean).map((node) => teardownHarper({ harper: node }))); + }); + + test('a ping-alive setup hang reconnects and converges without a restart', async () => { + await sendOperation(ctx.receiver, { + operation: 'add_node', + rejectUnauthorized: false, + hostname: ctx.source.hostname, + authorization: ctx.receiver.admin, + }); + + const sourceStallLog = await waitForLog(ctx.source, /\[test\] stalling subscription setup before DB_SCHEMA/); + assert.match(sourceStallLog, /\[test\] stalling subscription setup before DB_SCHEMA/); + const recoveryLog = await waitForLog(ctx.receiver, /Subscription-setup watchdog:.*\(db: "data"\)/); + assert.match( + recoveryLog, + /Subscription-setup watchdog:.*\(db: "data"\)/, + 'the receiver data watchdog must drive recovery' + ); + + const first = `after-setup-watchdog-${Date.now()}`; + await sendOperation(ctx.source, { + operation: 'insert', + database: DB, + table: TABLE, + records: [{ id: first }], + }); + assert.equal( + await waitForRecord(ctx.receiver, first), + true, + 'a record written after the setup hang must arrive over the recovered subscription' + ); + assert.equal(await socketConnected(ctx.receiver, DB), true, 'the recovered data socket must be connected'); + + const warningsBeforeIdle = countSetupWatchdogWarnings(await readLog(ctx.receiver)); + assert.ok(warningsBeforeIdle >= 1, 'at least one setup-watchdog recovery should have occurred'); + await delay(SETUP_TIMEOUT_MS * 3); + const second = `after-idle-${Date.now()}`; + await sendOperation(ctx.source, { + operation: 'insert', + database: DB, + table: TABLE, + records: [{ id: second }], + }); + assert.equal(await waitForRecord(ctx.receiver, second), true, 'healthy idle must not rearm setup recovery'); + assert.equal( + countSetupWatchdogWarnings(await readLog(ctx.receiver)), + warningsBeforeIdle, + 'healthy idle must not cause setup-watchdog reconnect churn' + ); + }); +}); + +suite('sender subscription setup recovery', { timeout: 120000 }, (ctx) => { + before(async () => { + const sourceCtx = { name: ctx.name, harper: { hostname: await getNextAvailableLoopbackAddress() } }; + const receiverCtx = { name: ctx.name, harper: { hostname: await getNextAvailableLoopbackAddress() } }; + await Promise.all([ + startHarper( + sourceCtx, + optionsFor(sourceCtx.harper, { + HARPER_TEST_SUBSCRIPTION_SETUP_STALL_ONCE_DB: DB, + HARPER_TEST_SEND_SUBSCRIPTION_RESOLVE_TIMEOUT_MS: '2000', + }) + ), + startHarper(receiverCtx, optionsFor(receiverCtx.harper, { HARPER_TEST_SUBSCRIPTION_SETUP_TIMEOUT_MS: '20000' })), + ]); + ctx.source = sourceCtx.harper; + ctx.receiver = receiverCtx.harper; + + await Promise.all( + [ctx.source, ctx.receiver].map((node) => + sendOperation(node, { + operation: 'create_table', + database: DB, + table: TABLE, + primary_key: 'id', + attributes: [{ name: 'id', type: 'ID' }], + }) + ) + ); + }); + + after(async () => { + await Promise.all([ctx.source, ctx.receiver].filter(Boolean).map((node) => teardownHarper({ harper: node }))); + }); + + test('the bounded sender gate closes first and the replacement subscription converges', async () => { + await sendOperation(ctx.receiver, { + operation: 'add_node', + rejectUnauthorized: false, + hostname: ctx.source.hostname, + authorization: ctx.receiver.admin, + }); + + const timeoutLog = await waitForLog(ctx.source, /Timed out waiting for authorization subscription setup/); + assert.match(timeoutLog, /Timed out waiting for authorization subscription setup/); + assert.doesNotMatch( + await readLog(ctx.receiver), + /Subscription-setup watchdog:.*\(db: "data"\)/, + 'the longer receiver backstop must not race the sender gate timeout' + ); + + const id = `after-sender-timeout-${Date.now()}`; + await sendOperation(ctx.source, { + operation: 'insert', + database: DB, + table: TABLE, + records: [{ id }], + }); + assert.equal(await waitForRecord(ctx.receiver, id), true, 'the sender-timeout retry must converge'); + assert.doesNotMatch( + await readLog(ctx.receiver), + /Subscription-setup watchdog:.*\(db: "data"\)/, + 'the receiver data watchdog must remain quiet after sender-driven convergence' + ); + }); +}); + +suite('system subscription setup recovery', { timeout: 120000 }, (ctx) => { + before(async () => { + const sourceCtx = { name: ctx.name, harper: { hostname: await getNextAvailableLoopbackAddress() } }; + const receiverCtx = { name: ctx.name, harper: { hostname: await getNextAvailableLoopbackAddress() } }; + await Promise.all([ + startHarper( + sourceCtx, + optionsFor(sourceCtx.harper, { HARPER_TEST_SUBSCRIPTION_SETUP_STALL_ONCE_DB: 'system' }, ['system']) + ), + startHarper( + receiverCtx, + optionsFor(receiverCtx.harper, { HARPER_TEST_SUBSCRIPTION_SETUP_TIMEOUT_MS: '3000' }, ['system']) + ), + ]); + ctx.source = sourceCtx.harper; + ctx.receiver = receiverCtx.harper; + }); + + after(async () => { + await Promise.all([ctx.source, ctx.receiver].filter(Boolean).map((node) => teardownHarper({ harper: node }))); + }); + + test('an unsolicited handshake schema cannot acknowledge the stalled system request', async () => { + await sendOperation(ctx.receiver, { + operation: 'add_node', + rejectUnauthorized: false, + hostname: ctx.source.hostname, + authorization: ctx.receiver.admin, + }); + + assert.match( + await waitForLog(ctx.source, /\[test\] stalling subscription setup before DB_SCHEMA for db "system"/), + /\[test\] stalling subscription setup before DB_SCHEMA for db "system"/ + ); + assert.match( + await waitForLog(ctx.receiver, /Subscription-setup watchdog:.*\(db: "system"\)/), + /Subscription-setup watchdog:.*\(db: "system"\)/, + 'the unsolicited handshake schema must not retire the correlated system request' + ); + assert.equal(await waitForSocket(ctx.receiver, 'system'), true, 'the replacement system socket must connect'); + + const role = `after-system-setup-watchdog-${Date.now()}`; + await sendOperation(ctx.source, { operation: 'add_role', role, permission: { super_user: false } }); + assert.equal(await waitForRole(ctx.receiver, role), true, 'system-table replication must converge after recovery'); + assert.ok( + countSetupWatchdogWarnings(await readLog(ctx.receiver), 'system') >= 1, + 'the correlated system request should trigger recovery' + ); + }); +}); diff --git a/replication/DESIGN.md b/replication/DESIGN.md index 657b78e8c..360f65d8e 100644 --- a/replication/DESIGN.md +++ b/replication/DESIGN.md @@ -132,6 +132,8 @@ Schema (defined in that function): `name` (PK), `subscriptions[]`, `system_info` 13. **A connection's local subscription may be an unresolved placeholder — never read `send`/`auditStore`/`dbisDB` off it.** A replication connection for a database is set up independently of `Replicator.subscribe()`, which registers that database's `IterableEventQueue` in `databaseSubscriptions` when the first of its tables is set up on this thread. Whichever loses the race, the connection is handed the placeholder Promise from `createPendingDatabaseSubscription` instead. Both use sites got this wrong (harper-pro#622): the receive path called `.send()` on it (a `.send is not a function` per inbound message, every record in it dropped — 329k errors / 500MB of `hdb.log` in 8 minutes on a 12-node cluster), and `sendSubscriptionRequestUpdate` read `auditStore`/`dbisDB` off it, so `nodeId` was `undefined`, no `seq` cursor resolved, `startTime` fell back to `1` and the node **requested a full copy of every database on every restart while a current cursor sat on disk**. The two fixes are asymmetric because their requirements are: (a) the record path genuinely needs the resolved queue, so it waits (`awaitPendingSubscription`, bounded by `SUBSCRIPTION_RESOLVE_TIMEOUT` — nothing else watches a wedged `messageProcessing` chain, since the receive watchdog is reset by the very frames not being processed — and pausing socket intake for the wait, because blocking that chain does **not** stop `ws.on('message')` from appending closures that each retain a whole inbound frame, so a peer mid-copy would OOM the worker before the timeout fired; `PAUSE_STALL_THRESHOLD_MS` is floored above the timeout so the paused-liveness watchdog can't pre-empt the wait); (b) the handshake must **not** wait — an empty node bootstrapping a database it does not have locally would deadlock (the peer only sends `DB_SCHEMA` in response to a subscription request, and that schema is what creates the tables that resolve the placeholder) — so `resolveDatabaseStores` reads the stores off a local table instead, both being per-database (`rootStore.auditStore` / `rootStore.dbisDb`) with the subscription queue only a carrier. Empty stores then mean what `startTime === 1` always assumed: no local tables at all, i.e. a genuine bootstrap. Compare `readDbisCursorSync` (#476/#484) — same "`undefined` masquerades as no resume cursor → spurious full copy" failure, different source of the `undefined`. +14. **Transport liveness does not prove subscription setup completed.** On the sender, both the dynamic `hdb_nodes` authorization subscription and the database's internal subscription placeholder must resolve before `DB_SCHEMA` is sent and the replay loop starts. A never-settling promise used to leave that `(peer, db)` socket ping-alive forever with no application frames, no received-version/time, and no cursor movement (harper-pro#642). Both sender gates are now bounded by `SUBSCRIPTION_RESOLVE_TIMEOUT`; expiry logs the exact gate and closes transiently so the subscriber retries from its last durable cursor. This deliberately also retries a peer/database mismatch whose placeholder can never resolve, matching the receive-path bound: the state is indistinguishable from a registration failure and may become valid after deployment. Independently, peers advertise support for a correlated setup acknowledgement and their effective two-gate setup budget in `NODE_NAME`; the outbound receiver then attaches a request id to each non-empty `SUBSCRIPTION_REQUEST` and arms a one-shot **subscription-setup watchdog** for the advertised budget, capped at the larger of four times its local window or ten minutes so a peer cannot disable the receiver's recovery net. A peer configured beyond that cap may therefore see periodic recovery reconnects instead of a suppressed watchdog. The post-gate `DB_SCHEMA` retires the watchdog only when it echoes that exact id. This correlation is required because the `system` handshake sends unsolicited schemas before the subscription request is processed, and a superseded replay can still have frames in flight. A receiver disables this independent watchdog when its sending peer does not advertise the capability, preserving wire compatibility without treating an unsolicited schema as an acknowledgement. In a mixed-version pair the setup-stall fix therefore depends on the **sender** being upgraded so its gates are bounded; upgrading only the receiver cannot safely distinguish an old sender's handshake schema from a setup response. Expiry calls `forceReconnect()` and includes the W1 truth snapshot, because the socket truth is legitimately connected — this is application-progress semantics, not a connection-truth failure. The watchdog is suspended during intentional socket back-pressure, cancelled on close/unsubscribe, and rearmed by a superseding request. Crucially, setup acknowledgement never advances the durable cursor: a zero receive timestamp is ambiguous for a healthy caught-up peer, and `SEQUENCE_ID_UPDATE` is cursor-mutating rather than a harmless ACK. + --- ## Tests diff --git a/replication/replicationConnection.ts b/replication/replicationConnection.ts index afd3845c2..317537780 100644 --- a/replication/replicationConnection.ts +++ b/replication/replicationConnection.ts @@ -115,6 +115,7 @@ const BLOB_CHUNK = 146; const SUBSCRIPTION_UPDATE = 147; const COPY_START = 148; // leader -> follower: a bulk table copy is starting; carries copyStartTime + copy-order version const COPY_COMPLETE = 149; // leader -> follower: the bulk table copy finished; follower clears its resume cursor +const SUBSCRIPTION_SETUP_ACK_CAPABILITY = 1; // Identifies the table ordering the leader copies in (see orderTablesForCopy). The resume skip-loop // trusts that every table before the cursor's currentTable was already copied — only true if the // resume runs under the SAME order that built the cursor. Bump this whenever orderTablesForCopy @@ -248,6 +249,10 @@ const COPY_CHECKPOINT_RECORDS = env.get('replication_copyCheckpointRecords') ?? // — the receive watchdog keeps being reset by the very frames we are not processing — so we close instead // and let reconnect backoff retry, which costs one log line per attempt rather than one per message. const SUBSCRIPTION_RESOLVE_TIMEOUT = positiveMsOr(env.get('replication_subscriptionResolveTimeout'), 60000); +const SEND_SUBSCRIPTION_RESOLVE_TIMEOUT = positiveMsOr( + process.env.HARPER_TEST_SEND_SUBSCRIPTION_RESOLVE_TIMEOUT_MS, + SUBSCRIPTION_RESOLVE_TIMEOUT +); // Wall-clock ceiling on the gap between socket flushes / event-loop yields during a bulk copy. // Reading a large cold table out of RocksDB dominates copy cost (decompress + decode), so a purely @@ -342,6 +347,17 @@ const STORAGE_IS_ROCKSDB = (process.env.HARPER_STORAGE_ENGINE || env.get(CONFIG_ // to 'close', etc.) this timer-based watchdog is the belt-and-suspenders that forces the // reconnect — see harper-pro#233. const RECEIVE_SILENCE_THRESHOLD_MS = PING_TIMEOUT; +// Application-level setup must complete after the transport handshake. A live ping/pong socket is not +// proof that the peer got past its send-authorization/database-subscription awaits and entered the replay +// loop (harper-pro#642). Keep this strictly behind both sequential sender subscription-resolution bounds +// so the sender gets first chance to identify the exact gate and close/retry, with the receiver-side +// watchdog as the independent net. The extra ping interval prevents equal-deadline timer races. +const TEST_SUBSCRIPTION_SETUP_TIMEOUT_MS = Number(process.env.HARPER_TEST_SUBSCRIPTION_SETUP_TIMEOUT_MS); +const SUBSCRIPTION_SETUP_TIMEOUT_MS = positiveMsOr( + TEST_SUBSCRIPTION_SETUP_TIMEOUT_MS, + Math.max(PING_TIMEOUT * 2, SUBSCRIPTION_RESOLVE_TIMEOUT * 2 + PING_INTERVAL) +); +const SEND_SUBSCRIPTION_SETUP_BUDGET_MS = SEND_SUBSCRIPTION_RESOLVE_TIMEOUT * 2 + PING_INTERVAL; // While the receive socket is paused for back-pressure the byte-silence watchdog above is stopped — // `ws.pause()` freezes `bytesRead`, so it can no longer tell a healthy back-pressure pause from a peer // that died mid-pause — and the active sendPing is exempt while `pauseReasons > 0`. That left a paused @@ -789,6 +805,20 @@ export function maybeStallCopyForTest(databaseName?: string): Promise | un return new Promise(() => {}); // never resolves; the sendPing timer keeps pings flowing } +// Test-only fault injection for harper-pro#642. The first dynamic send-authorization setup for the named +// database never resolves, before DB_SCHEMA or replay can start, while the independent ping loop keeps the +// socket transport-live. One-shot per process so a receiver watchdog reconnect reaches the normal setup +// path. Never arms in production: the env var is set only by the regression test. +let subscriptionSetupStallForTestArmed = false; +export function maybeStallSubscriptionSetupForTest(databaseName?: string): Promise | undefined { + if (!process.env.HARPER_TEST_SUBSCRIPTION_SETUP_STALL_ONCE_DB) return; + if (subscriptionSetupStallForTestArmed || process.env.HARPER_TEST_SUBSCRIPTION_SETUP_STALL_ONCE_DB !== databaseName) + return; + subscriptionSetupStallForTestArmed = true; + logger.warn?.(`[test] stalling subscription setup before DB_SCHEMA for db "${databaseName}" (harper-pro#642)`); + return new Promise(() => {}); +} + /** * Mark an error as a *source-reported* blob unavailability: the sender told us (via a BLOB_CHUNK * `error` marker) that it cannot provide this blob — classically `ENOENT` because the blob was @@ -1001,13 +1031,67 @@ export function resolveDatabaseStores( return { auditStore, dbisDB }; } +const SETUP_TIMED_OUT = Symbol('setup timed out'); + +async function awaitWithTimeoutOutcome( + value: T | PromiseLike | undefined, + timeout: number +): Promise { + if (!value || typeof (value as any).then !== 'function') return value as T | undefined; + let timer: NodeJS.Timeout | undefined; + try { + return await Promise.race([ + value, + new Promise((resolve) => { + timer = setTimeout(() => resolve(SETUP_TIMED_OUT), timeout); + timer.unref(); + }), + ]); + } finally { + if (timer) clearTimeout(timer); + } +} + +/** Await a promise-like setup dependency without letting a never-settling promise wedge the connection. */ +export async function awaitWithTimeout( + value: T | PromiseLike | undefined, + timeout: number +): Promise { + const outcome = await awaitWithTimeoutOutcome(value, timeout); + return outcome === SETUP_TIMED_OUT ? undefined : outcome; +} + +/** + * Resolve the two sender-side gates that precede DB_SCHEMA/replay, reporting the exact gate that failed + * to settle. Rejections intentionally propagate to the existing subscription-handler catch; only timeout + * is converted to `undefined` so the caller can close and retry without losing the durable cursor. + */ +export async function resolveSendSubscriptionSetup( + authorizationSubscription: TAuthorization | PromiseLike | undefined, + databaseSubscription: TDatabase | PromiseLike | undefined, + timeout: number, + onFailure: (gate: 'authorization' | 'database', reason: 'timeout' | 'unavailable') => void +): Promise { + if (authorizationSubscription) { + const resolvedAuthorizationSubscription = await awaitWithTimeoutOutcome(authorizationSubscription, timeout); + if (resolvedAuthorizationSubscription === SETUP_TIMED_OUT || !resolvedAuthorizationSubscription) { + onFailure('authorization', resolvedAuthorizationSubscription === SETUP_TIMED_OUT ? 'timeout' : 'unavailable'); + return; + } + } + const resolvedDatabaseSubscription = await awaitWithTimeoutOutcome(databaseSubscription, timeout); + if (resolvedDatabaseSubscription === SETUP_TIMED_OUT || !resolvedDatabaseSubscription) { + onFailure('database', resolvedDatabaseSubscription === SETUP_TIMED_OUT ? 'timeout' : 'unavailable'); + return; + } + return resolvedDatabaseSubscription; +} + /** * Await a database subscription that may still be the pending placeholder, bounded by `timeout` ms. * Resolves to the registered `IterableEventQueue`, or to `undefined` if the registration did not land in * time. A subscription that is already resolved (or absent) is returned as-is without yielding. * - * The timer is unref'd and always cleared, so neither outcome holds the event loop open. - * * `backpressure` (the receive path's `addPauseReason`/`removePauseReason`) is applied around the wait and * ONLY around a wait that actually happens. It is load-bearing rather than defensive: the receive path * awaits this from inside the serialized `messageProcessing` chain, but blocking that chain does not stop @@ -1015,8 +1099,6 @@ export function resolveDatabaseStores( * the socket, a peer mid-copy queues a full timeout's worth of large frames and exhausts the worker heap * before the timeout can close the connection. Paired in a `finally` so the pause is balanced on every * outcome, including a rejection. - * - * Exported for unit coverage; the production caller is `whenSubscriptionResolved` in `replicateOverWS`. */ export async function awaitPendingSubscription( subscription: DatabaseSubscription | undefined, @@ -1024,22 +1106,108 @@ export async function awaitPendingSubscription( backpressure?: { pause: () => void; resume: () => void } ): Promise { if (!subscription?.then) return subscription; - let timer: NodeJS.Timeout; backpressure?.pause(); try { - return await Promise.race([ - subscription as Promise, - new Promise((resolve) => { - timer = setTimeout(() => resolve(undefined), timeout); - timer.unref(); - }), - ]); + return await awaitWithTimeout(subscription, timeout); } finally { - clearTimeout(timer); backpressure?.resume(); } } +export function isSubscriptionSetupProgressFrame( + command: number | undefined, + requestedDatabase: string | undefined, + frameDatabase: string | undefined, + requestId: number | undefined, + frameRequestId: number | undefined +): boolean { + return ( + command === DB_SCHEMA && + requestId !== undefined && + frameDatabase === requestedDatabase && + frameRequestId === requestId + ); +} + +export function resolveSubscriptionSetupCapability( + capabilities: any, + localTimeoutMs: number, + usePeerBudget = true +): { supported: boolean; timeoutMs: number } { + const supported = capabilities?.subscriptionSetupAck >= SUBSCRIPTION_SETUP_ACK_CAPABILITY; + const peerSetupBudgetMs = capabilities?.subscriptionSetupBudgetMs; + const maxPeerSetupBudgetMs = Math.max(localTimeoutMs * 4, 10 * 60_000); + return { + supported, + timeoutMs: + usePeerBudget && supported && Number.isFinite(peerSetupBudgetMs) && peerSetupBudgetMs > 0 + ? Math.max(localTimeoutMs, Math.min(peerSetupBudgetMs, maxPeerSetupBudgetMs)) + : localTimeoutMs, + }; +} + +export function createSubscriptionSetupWatchdog(opts: { timeoutMs: number | (() => number); onTimeout: () => void }): { + arm: () => void; + complete: () => void; + pause: () => void; + resume: () => void; + stop: () => void; +} { + let timer: NodeJS.Timeout | undefined; + let pending = false; + let paused = false; + // Remaining budget (ms) for the current pending window, consumed as time elapses. Only `arm()` + // resets it to a full window (a superseding request); `pause()`/`resume()` preserve whatever was + // left, so a link with recurring, short back-pressure pauses can't keep re-granting a full window + // forever and never trip this independent recovery net. + let remainingMs = 0; + let scheduledAt = 0; + const clearTimer = () => { + if (timer) { + clearTimeout(timer); + timer = undefined; + } + }; + const fire = () => { + timer = undefined; + pending = false; + opts.onTimeout(); + }; + const schedule = () => { + clearTimer(); + if (!pending || paused) return; + scheduledAt = performance.now(); // monotonic — immune to wall-clock adjustments + timer = setTimeout(fire, remainingMs).unref(); + }; + return { + arm() { + pending = true; + remainingMs = typeof opts.timeoutMs === 'function' ? opts.timeoutMs() : opts.timeoutMs; + schedule(); + }, + complete() { + pending = false; + clearTimer(); + }, + pause() { + if (paused) return; + if (pending && timer) remainingMs = Math.max(0, remainingMs - (performance.now() - scheduledAt)); + paused = true; + clearTimer(); + }, + resume() { + if (!paused) return; + paused = false; + schedule(); + }, + stop() { + pending = false; + paused = false; + clearTimer(); + }, + }; +} + /** * Create (and register) the placeholder used for a database whose local subscription queue does not * exist yet: `Replicator.subscribe()` runs when the first table of the database is set up on this @@ -2004,6 +2172,21 @@ export function replicateOverWS(ws: WebSocket, options: any, authorization: any) } let sendPingInterval, lastPingTime, skippedMessageSequenceUpdateTimer; let receiveWatchdog: { reset: () => void; stop: () => void } | undefined; + let peerSupportsSubscriptionSetupAck = false; + let nextSubscriptionSetupRequestId = 0; + let pendingSubscriptionSetupRequestId: number | undefined; + let subscriptionSetupTimeoutMs = SUBSCRIPTION_SETUP_TIMEOUT_MS; + // Outbound-only application setup guard (harper-pro#642). Unlike receiveWatchdog it deliberately + // ignores ping/pong bytes: those prove the socket is alive, not that the peer entered its replay loop. + let subscriptionSetupWatchdog: + | { + arm: () => void; + complete: () => void; + pause: () => void; + resume: () => void; + stop: () => void; + } + | undefined; // Companion to receiveWatchdog that guards the back-pressure-paused window the byte watchdog is // blind to (harper-pro#466). Armed on pause, stopped on resume — see addPauseReason/removePauseReason. let pauseStallWatchdog: { reset: () => void; stop: () => void } | undefined; @@ -2201,6 +2384,19 @@ export function replicateOverWS(ws: WebSocket, options: any, authorization: any) else ws.terminate(); }, }); + subscriptionSetupWatchdog = createSubscriptionSetupWatchdog({ + timeoutMs: () => subscriptionSetupTimeoutMs, + onTimeout: () => { + if (wsClosed) return; + pendingSubscriptionSetupRequestId = undefined; + const dbContext = databaseName ? ` (db: "${databaseName}")` : ''; + logger.warn?.( + `Subscription-setup watchdog: no application response from ${remoteNodeName}${dbContext} for ${subscriptionSetupTimeoutMs}ms while transport remained connected — reconnecting from the durable cursor (harper-pro#642) — ${truthSnapshotForLog()}` + ); + if (options.connection) options.connection.forceReconnect(); + else ws.terminate(); + }, + }); ws._socket?.setMaxListeners(200); // we should allow a lot of drain listeners for concurrent blob streams let ratioOfBackPressureTime = 0; let lastBackPressureCheck = 0; @@ -2257,6 +2453,7 @@ export function replicateOverWS(ws: WebSocket, options: any, authorization: any) if (pauseReasons === 0) { ws.pause(); pauseStartTime = Date.now(); + subscriptionSetupWatchdog?.pause(); // Suspend the receive watchdog while the socket is intentionally paused — `bytesRead` // is frozen by `ws.pause()` so the byte check cannot tell legitimate backpressure // from peer silence, and firing here would terminate a healthy mid-ingest connection. @@ -2275,6 +2472,7 @@ export function replicateOverWS(ws: WebSocket, options: any, authorization: any) pauseReasons--; if (pauseReasons === 0) { ws.resume(); + subscriptionSetupWatchdog?.resume(); // Resuming: the byte watchdog can see the socket again, so retire the pause-stall watchdog and // restart the silence window from the resume point — we deliberately do not penalize the // connection for the time it spent paused. @@ -2534,9 +2732,28 @@ export function replicateOverWS(ws: WebSocket, options: any, authorization: any) // not a transaction, special message const message = decode(body); const [command, data, tableId] = message; + if ( + isSubscriptionSetupProgressFrame( + command, + databaseName, + message[2], + pendingSubscriptionSetupRequestId, + message[3] + ) + ) { + pendingSubscriptionSetupRequestId = undefined; + subscriptionSetupWatchdog?.complete(); + } switch (command) { case NODE_NAME: { if (data) { + const setupCapability = resolveSubscriptionSetupCapability( + message[4], + SUBSCRIPTION_SETUP_TIMEOUT_MS, + !(TEST_SUBSCRIPTION_SETUP_TIMEOUT_MS > 0) + ); + peerSupportsSubscriptionSetupAck = setupCapability.supported; + subscriptionSetupTimeoutMs = setupCapability.timeoutMs; // this is the node name if (remoteNodeName) { if (remoteNodeName !== data) { @@ -3109,6 +3326,7 @@ export function replicateOverWS(ws: WebSocket, options: any, authorization: any) case SUBSCRIPTION_REQUEST: { nodeSubscriptions = data; excludedNodes = message[2]; // use the third argument for exclusion list + const subscriptionSetupRequestId = message[3]; // permission check to make sure that this node is allowed to subscribe to this database, that is that // we have publish permission for this node/database let subscriptionToHdbNodes, whenSubscribedToHdbNodes; @@ -3159,10 +3377,20 @@ export function replicateOverWS(ws: WebSocket, options: any, authorization: any) // so skip the hdb_nodes auth watch entirely. Only when there is no directional config route // (undefined) do we watch the subscriber's hdb_nodes record for dynamic (de)authorization. if (configSendDecision === undefined) { - whenSubscribedToHdbNodes = getHDBNodeTable().subscribe(authorization.name); - whenSubscribedToHdbNodes.then( - async (subscription) => { + whenSubscribedToHdbNodes = + maybeStallSubscriptionSetupForTest(databaseName) ?? getHDBNodeTable().subscribe(authorization.name); + whenSubscribedToHdbNodes + .then(async (subscription) => { subscriptionToHdbNodes = subscription; + // The setup wait below is bounded. If it timed out and closed this socket before the + // subscription promise eventually resolved, retire the late subscription immediately + // instead of leaking a global hdb_nodes listener after the WS close event already fired. + if (closed || wsClosed) { + const lateSubscription = subscriptionToHdbNodes; + subscriptionToHdbNodes = undefined; + lateSubscription?.end(); + return; + } for await (const event of subscriptionToHdbNodes) { const shouldClose = await shouldCloseSendAuthWatch(event, authorization.name, databaseName, { isClosed: () => closed, @@ -3182,11 +3410,10 @@ export function replicateOverWS(ws: WebSocket, options: any, authorization: any) return; } } - }, - (error) => { + }) + .catch((error) => { logger.error?.(connectionId, 'Error subscribing to HDB nodes', error); - } - ); + }); } } else if (!(authorization?.role?.permission?.super_user || authorization.replicates)) { ws.send(encode([DISCONNECT])); @@ -3554,8 +3781,7 @@ export function replicateOverWS(ws: WebSocket, options: any, authorization: any) }); // find the earliest start time of the subscriptions let copyResume: - | { copyStartTime: number; currentTable: string; afterKey: any; copyOrder?: number } - | undefined; + { copyStartTime: number; currentTable: string; afterKey: any; copyOrder?: number } | undefined; for (const subscription of nodeSubscriptions) { if (subscription.startTime < currentSequenceId) currentSequenceId = subscription.startTime; // a follower resuming an interrupted bulk copy sends back where it left off. This keeps the @@ -3565,10 +3791,35 @@ export function replicateOverWS(ws: WebSocket, options: any, authorization: any) if (subscription.copyResume) copyResume = subscription.copyResume; } - // wait for internal subscription, might be waiting for a table to be registered - (whenSubscribedToHdbNodes || Promise.resolve()) + // A promise that never settles here would leave this WS ping-alive forever with the + // receiver seeing no application frames and no cursor progress (harper-pro#642). Bound + // each gate separately so the log identifies which one stuck, then close transiently so + // the peer retries from its last durable cursor. + Promise.resolve() .then(async () => { - tableSubscriptionToReplicator = await tableSubscriptionToReplicator; + const resolvedDatabaseSubscription = await resolveSendSubscriptionSetup( + whenSubscribedToHdbNodes, + tableSubscriptionToReplicator, + SEND_SUBSCRIPTION_RESOLVE_TIMEOUT, + (gate, reason) => { + if (closed || wsClosed) return; + closed = true; + const databaseHint = + gate === 'database' ? '; this can also mean this node does not host the database' : ''; + const failure = + reason === 'timeout' + ? `Timed out waiting for ${gate} subscription setup` + : `${gate} subscription setup resolved without a subscription`; + logger.error?.( + connectionId, + `${failure} for ${databaseName}${databaseHint}; closing so the subscriber retries from its durable cursor (harper-pro#642)` + ); + close(1011, `Replication ${gate} setup ${reason === 'timeout' ? 'timed out' : 'unavailable'}`); + } + ); + if (!resolvedDatabaseSubscription) return; + tableSubscriptionToReplicator = resolvedDatabaseSubscription; + if (closed || wsClosed) return; auditStore = tableSubscriptionToReplicator.auditStore; tableById = tableSubscriptionToReplicator.tableById.map(tableToTableEntry); subscribedNodeIds = []; @@ -3586,7 +3837,7 @@ export function replicateOverWS(ws: WebSocket, options: any, authorization: any) subscribedNodeName = name; } - sendDBSchema(databaseName); + sendDBSchema(databaseName, subscriptionSetupRequestId); if (!schemaUpdateListener) { schemaUpdateListener = onUpdatedTable((table) => { if (table.databaseName === databaseName) { @@ -4353,8 +4604,10 @@ export function replicateOverWS(ws: WebSocket, options: any, authorization: any) ws.on('close', (code, reasonBuffer) => { // cleanup wsClosed = true; + pendingSubscriptionSetupRequestId = undefined; clearInterval(sendPingInterval); receiveWatchdog?.stop(); + subscriptionSetupWatchdog?.stop(); pauseStallWatchdog?.stop(); copyProgressWatchdog?.stop(); clearInterval(blobsTimer); @@ -5077,10 +5330,22 @@ export function replicateOverWS(ws: WebSocket, options: any, authorization: any) logger.debug?.(connectionId, 'sending subscription request', nodeSubscriptions, dbisDB?.path); clearTimeout(delayedClose); if (nodeSubscriptions.length > 0) { - ws.send(encode([SUBSCRIPTION_REQUEST, nodeSubscriptions, excluded])); + const requestId = peerSupportsSubscriptionSetupAck ? ++nextSubscriptionSetupRequestId : undefined; + ws.send(encode([SUBSCRIPTION_REQUEST, nodeSubscriptions, excluded, requestId])); + // Start a fresh request -> first application-response window on every non-empty request. This is + // intentionally after send(): a synchronous send failure must not leave an orphaned timer. + if (requestId === undefined) { + pendingSubscriptionSetupRequestId = undefined; + subscriptionSetupWatchdog?.stop(); + } else { + pendingSubscriptionSetupRequestId = requestId; + subscriptionSetupWatchdog?.arm(); + } // Track the excluded list we just sent lastSentExcludedNodes = excluded ? [...excluded] : []; } else { + pendingSubscriptionSetupRequestId = undefined; + subscriptionSetupWatchdog?.stop(); // no nodes means we are unsubscribing/disconnecting // don't immediately close the connection, but wait a bit to see if we get any messages, since opening new connections is a bit expensive const scheduleClose = () => { @@ -5164,9 +5429,20 @@ export function replicateOverWS(ws: WebSocket, options: any, authorization: any) }); } logger.trace?.('Sending database info for node', thisNodeName, 'database name', databaseName); - ws.send(encode([NODE_NAME, thisNodeName, databaseName, tables])); + ws.send( + encode([ + NODE_NAME, + thisNodeName, + databaseName, + tables, + { + subscriptionSetupAck: SUBSCRIPTION_SETUP_ACK_CAPABILITY, + subscriptionSetupBudgetMs: SEND_SUBSCRIPTION_SETUP_BUDGET_MS, + }, + ]) + ); } - function sendDBSchema(databaseName) { + function sendDBSchema(databaseName, subscriptionSetupRequestId?) { const database = getDatabases()?.[databaseName]; const tables = []; for (const tableName in database) { @@ -5189,7 +5465,7 @@ export function replicateOverWS(ws: WebSocket, options: any, authorization: any) }); } - ws.send(encode([DB_SCHEMA, tables, databaseName])); + ws.send(encode([DB_SCHEMA, tables, databaseName, subscriptionSetupRequestId])); } blobsTimer = setInterval( () => { diff --git a/unitTests/replication/subscriptionSetupWatchdog.test.mjs b/unitTests/replication/subscriptionSetupWatchdog.test.mjs new file mode 100644 index 000000000..47162c17d --- /dev/null +++ b/unitTests/replication/subscriptionSetupWatchdog.test.mjs @@ -0,0 +1,263 @@ +// harper-pro#642: ping/pong must not count as application setup progress. + +import assert from 'node:assert/strict'; +import sinon from 'sinon'; +import { + awaitWithTimeout, + createSubscriptionSetupWatchdog, + isSubscriptionSetupProgressFrame, + resolveSubscriptionSetupCapability, + resolveSendSubscriptionSetup, +} from '#src/replication/replicationConnection'; + +const COPY_START = 148; +const DB_SCHEMA = 145; +const NODE_NAME = 140; +const SEQUENCE_ID_UPDATE = 143; + +describe('awaitWithTimeout', () => { + it('returns an already-resolved value without waiting', async () => { + const value = { ready: true }; + assert.equal(await awaitWithTimeout(value, 5), value); + }); + + it('returns a promise value when it settles inside the bound', async () => { + assert.equal(await awaitWithTimeout(Promise.resolve('ready'), 5), 'ready'); + }); + + it('returns undefined when a setup promise never settles', async () => { + assert.equal(await awaitWithTimeout(new Promise(() => {}), 5), undefined); + }); + + it('preserves a setup rejection for the existing handler catch', async () => { + await assert.rejects(awaitWithTimeout(Promise.reject(new Error('setup failed')), 5), /setup failed/); + }); +}); + +describe('resolveSendSubscriptionSetup', () => { + it('returns the database subscription after both gates resolve', async () => { + const database = { auditStore: {} }; + const timedOut = sinon.spy(); + assert.equal( + await resolveSendSubscriptionSetup(Promise.resolve({ end() {} }), Promise.resolve(database), 5, timedOut), + database + ); + assert.equal(timedOut.callCount, 0); + }); + + it('identifies an authorization gate that never settles', async () => { + const timedOut = sinon.spy(); + assert.equal( + await resolveSendSubscriptionSetup(new Promise(() => {}), Promise.resolve({}), 5, timedOut), + undefined + ); + assert.deepEqual(timedOut.args, [['authorization', 'timeout']]); + }); + + it('identifies a database gate that never settles', async () => { + const timedOut = sinon.spy(); + assert.equal( + await resolveSendSubscriptionSetup(Promise.resolve({ end() {} }), new Promise(() => {}), 5, timedOut), + undefined + ); + assert.deepEqual(timedOut.args, [['database', 'timeout']]); + }); + + it('distinguishes a settled-but-unavailable gate from a timeout', async () => { + const failed = sinon.spy(); + assert.equal(await resolveSendSubscriptionSetup(Promise.resolve(null), Promise.resolve({}), 5, failed), undefined); + assert.deepEqual(failed.args, [['authorization', 'unavailable']]); + }); +}); + +describe('resolveSubscriptionSetupCapability', () => { + it('accepts newer additive versions and honors a longer sender budget', () => { + assert.deepEqual( + resolveSubscriptionSetupCapability({ subscriptionSetupAck: 2, subscriptionSetupBudgetMs: 300 }, 150), + { + supported: true, + timeoutMs: 300, + } + ); + }); + + it('keeps the local timeout for an old peer or an explicit test override', () => { + assert.deepEqual(resolveSubscriptionSetupCapability(undefined, 150), { supported: false, timeoutMs: 150 }); + assert.deepEqual( + resolveSubscriptionSetupCapability({ subscriptionSetupAck: 1, subscriptionSetupBudgetMs: 300 }, 25, false), + { supported: true, timeoutMs: 25 } + ); + }); + + it('caps a peer budget that would disable the local recovery net', () => { + assert.deepEqual( + resolveSubscriptionSetupCapability({ subscriptionSetupAck: 1, subscriptionSetupBudgetMs: 86_400_000 }, 150_000), + { supported: true, timeoutMs: 600_000 } + ); + }); +}); + +describe('isSubscriptionSetupProgressFrame', () => { + it('accepts the requested database schema with the matching request id', () => { + assert.equal(isSubscriptionSetupProgressFrame(DB_SCHEMA, 'flair', 'flair', 7, 7), true); + }); + + it('rejects schema traffic for a sibling database', () => { + assert.equal(isSubscriptionSetupProgressFrame(DB_SCHEMA, 'flair', 'data', 7, 7), false); + }); + + it('rejects an unsolicited handshake schema and a stale response', () => { + assert.equal(isSubscriptionSetupProgressFrame(DB_SCHEMA, 'flair', 'flair', 7, undefined), false); + assert.equal(isSubscriptionSetupProgressFrame(DB_SCHEMA, 'flair', 'flair', 7, 6), false); + }); + + it('does not let uncorrelated copy, sequence, or replication data retire setup', () => { + assert.equal(isSubscriptionSetupProgressFrame(COPY_START, 'flair', undefined, 7, undefined), false); + assert.equal(isSubscriptionSetupProgressFrame(SEQUENCE_ID_UPDATE, 'flair', undefined, 7, undefined), false); + assert.equal(isSubscriptionSetupProgressFrame(undefined, 'flair', undefined, 7, undefined), false); + }); + + it('does not accept transport/identity handshake traffic', () => { + assert.equal(isSubscriptionSetupProgressFrame(NODE_NAME, 'flair', undefined, 7, undefined), false); + }); +}); + +describe('createSubscriptionSetupWatchdog', () => { + let clock; + + beforeEach(() => { + clock = sinon.useFakeTimers(); + }); + + afterEach(() => { + clock.restore(); + }); + + it('fires exactly once when setup never progresses', () => { + const onTimeout = sinon.spy(); + const watchdog = createSubscriptionSetupWatchdog({ timeoutMs: 60_000, onTimeout }); + + watchdog.arm(); + clock.tick(60_000); + clock.tick(60_000); + + assert.equal(onTimeout.callCount, 1); + }); + + it('is cancelled by setup progress', () => { + const onTimeout = sinon.spy(); + const watchdog = createSubscriptionSetupWatchdog({ timeoutMs: 60_000, onTimeout }); + + watchdog.arm(); + clock.tick(30_000); + watchdog.complete(); + clock.tick(60_000); + + assert.equal(onTimeout.callCount, 0); + }); + + it('stop cancels and a later non-empty request can rearm', () => { + const onTimeout = sinon.spy(); + const watchdog = createSubscriptionSetupWatchdog({ timeoutMs: 60_000, onTimeout }); + + watchdog.arm(); + watchdog.stop(); + clock.tick(60_000); + assert.equal(onTimeout.callCount, 0); + + watchdog.arm(); + clock.tick(60_000); + assert.equal(onTimeout.callCount, 1); + }); + + it('a superseding request restarts the full timeout window', () => { + const onTimeout = sinon.spy(); + const watchdog = createSubscriptionSetupWatchdog({ timeoutMs: 60_000, onTimeout }); + + watchdog.arm(); + clock.tick(30_000); + watchdog.arm(); + clock.tick(30_000); + assert.equal(onTimeout.callCount, 0); + clock.tick(30_000); + assert.equal(onTimeout.callCount, 1); + }); + + it('uses the peer-adjusted timeout when a request is armed', () => { + const onTimeout = sinon.spy(); + let timeoutMs = 60_000; + const watchdog = createSubscriptionSetupWatchdog({ timeoutMs: () => timeoutMs, onTimeout }); + + timeoutMs = 120_000; + watchdog.arm(); + clock.tick(60_000); + assert.equal(onTimeout.callCount, 0); + clock.tick(60_000); + assert.equal(onTimeout.callCount, 1); + }); + + it('does not count paused time against a pending setup window, but preserves unpaused progress', () => { + const onTimeout = sinon.spy(); + const watchdog = createSubscriptionSetupWatchdog({ timeoutMs: 60_000, onTimeout }); + + watchdog.arm(); + clock.tick(30_000); + watchdog.pause(); + clock.tick(120_000); + assert.equal(onTimeout.callCount, 0); + + // 30s already elapsed before the pause; only the remaining 30s should be left after resume — + // recurring pause/resume must not keep re-granting a full window forever (harper-pro#642 review). + watchdog.resume(); + clock.tick(29_999); + assert.equal(onTimeout.callCount, 0); + clock.tick(1); + assert.equal(onTimeout.callCount, 1); + }); + + it('recurring short pause/resume cycles still consume down to firing, never re-granting a full window', () => { + const onTimeout = sinon.spy(); + const watchdog = createSubscriptionSetupWatchdog({ timeoutMs: 60_000, onTimeout }); + + watchdog.arm(); + for (let i = 0; i < 5; i++) { + clock.tick(10_000); // active progress + watchdog.pause(); + clock.tick(5_000); // paused; must not count + watchdog.resume(); + } + // 5 * 10s active progress consumed of the 60s budget; 10s remains. + assert.equal(onTimeout.callCount, 0); + clock.tick(9_999); + assert.equal(onTimeout.callCount, 0); + clock.tick(1); + assert.equal(onTimeout.callCount, 1); + }); + + it('does not rearm on resume after setup completed while paused', () => { + const onTimeout = sinon.spy(); + const watchdog = createSubscriptionSetupWatchdog({ timeoutMs: 60_000, onTimeout }); + + watchdog.arm(); + watchdog.pause(); + watchdog.complete(); + watchdog.resume(); + clock.tick(60_000); + + assert.equal(onTimeout.callCount, 0); + }); + + it('defers a request armed during back pressure until the socket resumes', () => { + const onTimeout = sinon.spy(); + const watchdog = createSubscriptionSetupWatchdog({ timeoutMs: 60_000, onTimeout }); + + watchdog.pause(); + watchdog.arm(); + clock.tick(120_000); + assert.equal(onTimeout.callCount, 0); + + watchdog.resume(); + clock.tick(60_000); + assert.equal(onTimeout.callCount, 1); + }); +});