fix(replication): recover ping-alive setup stalls before DB_SCHEMA (#642)

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'
+		);
+	});
+});

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 `<x>.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