Scheduler: in-repo election, failover, and catch-up test coverage

.gitignore

@@ -58,3 +58,7 @@ benchmarks/ycsb/bench-out/
 
 # northwind export_local test artifact
 test_export.json.json
+
+# dev-mode boots (harper dev <fixture>) symlink node_modules/harper into the
+# fixture component dir; keep those out of commits
+integrationTests/fixtures/*/node_modules/

resources/scheduler/engine.ts

@@ -74,6 +74,13 @@ let leaderlessSince: number | undefined;
 // When this node became leader — fallback for the lease row's initializedAt
 // when a heartbeat renews after a failed read
 let leaderInitializedAt: string | undefined;
+// The in-flight (or settled) election kicked off by startSchedulerEngine —
+// held so tests can await the role decision instead of polling
+let electionPromise: Promise<void> | undefined;
+// Bumped by stopSchedulerEngine: async transitions (electRole, becomeLeader)
+// capture it before their awaits and abandon themselves if it moved — an
+// in-flight election must not resurrect timers after a stop (review finding)
+let engineEpoch = 0;
 const jobsByComponent = new Map<string, Map<string, RegisteredJob>>();
 
 let _stateTable: any;
@@ -277,11 +284,31 @@ export function unregisterComponentJobs(componentName: string): void {
 export function startSchedulerEngine(): void {
 	if (engineStarted) return;
 	engineStarted = true;
-	electRole().catch((error) => {
+	electionPromise = electRole().catch((error) => {
 		schedulerLogger.error?.('Scheduler engine failed to start', error);
 	});
 }
 
+/** @internal — testing only: resolves when the initial election has settled */
+export function electionSettledForTests(): Promise<void> {
+	return electionPromise ?? Promise.resolve();
+}
+
+/**
+ * @internal — testing only: run one heartbeat tick immediately (same body the
+ * interval runs), so tests can drive lease renewal / takeover checks without
+ * waiting out HEARTBEAT_INTERVAL_MS. Same seam pattern as
+ * setCoolingFunctionForTests in transactionLogCooling.
+ */
+export function runHeartbeatForTests(): Promise<void> {
+	return heartbeat();
+}
+
+/** @internal — testing only: run one failover-watcher tick immediately */
+export function runFailoverCheckForTests(): Promise<void> {
+	return failoverCheck();
+}
+
 /** Reset all engine state and timers. Intended for tests. */
 export function stopSchedulerEngine(): void {
 	for (const componentName of [...jobsByComponent.keys()]) {
@@ -296,6 +323,8 @@ export function stopSchedulerEngine(): void {
 	catchUpRunning = false;
 	leaderlessSince = undefined;
 	leaderInitializedAt = undefined;
+	electionPromise = undefined;
+	engineEpoch++;
 	_stateTable = undefined;
 }
 
@@ -310,17 +339,20 @@ export function getRegisteredJobNames(componentName: string): string[] {
 
 async function electRole(): Promise<void> {
 	const self = currentNodeName();
+	const epoch = engineEpoch;
 	let leaderRow: any;
 	try {
 		leaderRow = await withStateTimeout(getStateTable().get(LEADER_ROW_ID), 'leader state read');
 	} catch (error) {
+		if (epoch !== engineEpoch) return; // stopped while electing
 		// Fail toward followership: electing ourselves while the state table is
 		// unreadable risks a second leader. The failover watcher keeps checking
 		// and promotes once reads succeed and show a leaderless cluster.
 		schedulerLogger.warn?.(`Could not read scheduler leader state, defaulting to follower: ${safeErrorMessage(error)}`);
 		becomeFollower();
 		return;
 	}
+	if (epoch !== engineEpoch) return; // stopped while electing
 	// Sticky leadership: a node (re)starting while another node is actively
 	// leading defers to it rather than seizing leadership back
 	if (leaderRow && leaderRow.leaderNode !== self && !isHeartbeatStale(leaderRow.lastHeartbeat)) {
@@ -346,6 +378,7 @@ async function electRole(): Promise<void> {
 
 async function becomeLeader(skipInitialCatchUp = false): Promise<void> {
 	const self = currentNodeName();
+	const epoch = engineEpoch;
 	role = 'leader';
 	if (failoverWatcherTimer) {
 		clearInterval(failoverWatcherTimer);
@@ -370,6 +403,10 @@ async function becomeLeader(skipInitialCatchUp = false): Promise<void> {
 	const now = new Date().toISOString();
 	leaderInitializedAt = now;
 	await putStateRow({ id: LEADER_ROW_ID, leaderNode: self, lastHeartbeat: now, initializedAt: now });
+	// A stop while the lease write was in flight must not resurrect the
+	// heartbeat interval (review finding: in-flight transitions surviving
+	// stopSchedulerEngine)
+	if (epoch !== engineEpoch) return;
 	// The heartbeat interval must be beating BEFORE the promotion catch-up
 	// pass: catch-up runs user handlers serially and can exceed the stale
 	// threshold, and a leader that stops renewing mid-catch-up looks dead —
@@ -449,6 +486,7 @@ async function heartbeat(): Promise<void> {
 async function failoverCheck(): Promise<void> {
 	if (role !== 'follower') return;
 	const self = currentNodeName();
+	const epoch = engineEpoch;
 	let leaderRow: any;
 	try {
 		leaderRow = await withStateTimeout(getStateTable().get(LEADER_ROW_ID), 'leader state read');
@@ -464,9 +502,9 @@ async function failoverCheck(): Promise<void> {
 		return;
 	}
 	// Re-check after the await: a concurrent (stalled) check may have promoted
-	// this node already, and a second becomeLeader would duplicate work
-	// (audit finding)
-	if (role !== 'follower') return;
+	// this node already (audit finding), or the engine may have been stopped
+	// while the read was in flight (review finding) — either way, do not promote
+	if (role !== 'follower' || epoch !== engineEpoch) return;
 	const now = Date.now();
 	leaderlessSince ??= now;
 	const roster = nodeRoster();

unitTests/resources/scheduler/engineElection.test.js

@@ -0,0 +1,268 @@
+'use strict';
+
+// #1866 — in-repo coverage of the scheduler's election/heartbeat/failover/
+// catch-up orchestration (electRole, becomeLeader, heartbeat, failoverCheck,
+// runCatchUp, executeJob), which previously had only pure-helper unit tests
+// here and true-network validation only in the harper-pro cluster harness.
+//
+// True multi-node replication is not available in this repo (replication ships
+// as a separate component), so this uses the same-process simulated-peer
+// pattern (cf. deploy-tracking-peer-branch.test.ts): everything the engine
+// knows about other nodes flows through exactly two observables —
+// `server.nodes` (normally populated by the replication component) and rows in
+// the hdb_scheduler_state table (normally replicated). The test supplies both,
+// against the REAL engine module, REAL timers, and a REAL store. Staleness is
+// data-driven (a lease heartbeat stamped 10 minutes ago IS a dead leader), so
+// no test waits out production thresholds; heartbeat/failover ticks are driven
+// through the ForTests seams instead of their intervals.
+
+require('../../testUtils');
+const assert = require('node:assert');
+const { hostname } = require('node:os');
+const { table } = require('#src/resources/databases');
+const { setMainIsWorker } = require('#js/server/threads/manageThreads');
+const {
+	registerComponentJobs,
+	unregisterComponentJobs,
+	startSchedulerEngine,
+	stopSchedulerEngine,
+	getEngineRole,
+	getRegisteredJobNames,
+	electionSettledForTests,
+	runHeartbeatForTests,
+	runFailoverCheckForTests,
+	STALE_THRESHOLD_MS,
+} = require('#src/resources/scheduler/engine');
+const { CronExpression } = require('#src/resources/scheduler/CronExpression');
+const { waitFor } = require('../../waitFor.js');
+
+// server.hostname is a non-configurable getter (nodeName.ts), so instead of
+// stubbing the node's identity we read the same expression the engine uses
+// and pick peer names that deterministically sort around it: '\u0000' sorts
+// before any real hostname, '\uffff' after.
+const SELF = global.server.hostname || hostname();
+const PEER = '\u0000-peer'; // sorts FIRST, so the peer is the preferred leader
+const LAST_PEER = '\uffff-peer'; // sorts LAST, so SELF is preferred
+const LEADER_ROW_ID = 'leader';
+
+function isoAgo(ms) {
+	return new Date(Date.now() - ms).toISOString();
+}
+
+describe('scheduler engine election and failover (simulated peer, #1866)', () => {
+	let stateTable;
+	let originalNodes;
+	// Every handler invocation lands here: [{ jobName, catchUp, scheduledAt }]
+	const handlerRuns = [];
+
+	function makeCronJob(name) {
+		return {
+			name,
+			componentName: 'election-test',
+			cron: new CronExpression('0 2 * * *'),
+			timezone: 'UTC',
+			handler: (context) => {
+				handlerRuns.push(context);
+			},
+		};
+	}
+
+	before(async () => {
+		setMainIsWorker(true);
+		// The engine sees peers through server.nodes, normally supplied by the
+		// replication component. global.server IS the object engine.ts imports
+		// (globals.js + _assignPackageExport), the same stub point the analytics
+		// unit tests use for server.nodes.
+		originalNodes = global.server.nodes;
+		global.server.nodes = [{ name: PEER }];
+		// Same table the engine lazily creates; declaring it here too resolves to
+		// the same underlying store
+		stateTable = table({
+			database: 'system',
+			table: 'hdb_scheduler_state',
+			replicate: true,
+			audit: true,
+			attributes: [
+				{ name: 'id', type: 'string', isPrimaryKey: true },
+				{ name: 'leaderNode', type: 'string' },
+				{ name: 'lastHeartbeat', type: 'string' },
+				{ name: 'initializedAt', type: 'string' },
+				{ name: 'firstSeenAt', type: 'string' },
+				{ name: 'lastRunAt', type: 'string' },
+				{ name: 'lastStatus', type: 'string' },
+				{ name: 'lastError', type: 'string' },
+				{ name: 'lastDurationMs', type: 'number' },
+			],
+		});
+		stopSchedulerEngine();
+	});
+
+	after(async () => {
+		stopSchedulerEngine();
+		global.server.nodes = originalNodes;
+		// Leave the shared system table the way we found it
+		await stateTable.delete(LEADER_ROW_ID).catch(() => {});
+		await stateTable.delete('job:election-test:nightly').catch(() => {});
+	});
+
+	// The tests below are deliberately sequential: they walk one node through
+	// the full lifecycle — defer to a live peer, take over when it dies (with
+	// catch-up), surrender to a takeover, and win a cold-start election.
+
+	it('defers to a peer whose lease heartbeat is fresh (sticky election)', async () => {
+		await stateTable.put({
+			id: LEADER_ROW_ID,
+			leaderNode: PEER,
+			lastHeartbeat: isoAgo(1000),
+			initializedAt: isoAgo(1000),
+		});
+		registerComponentJobs('election-test', [makeCronJob('nightly')]);
+		startSchedulerEngine();
+		await electionSettledForTests();
+		assert.strictEqual(getEngineRole(), 'follower', 'a fresh peer lease must win the election');
+		assert.strictEqual(handlerRuns.length, 0, 'a follower must not run jobs');
+	});
+
+	it('does not promote while the leader keeps beating', async () => {
+		await stateTable.put({
+			id: LEADER_ROW_ID,
+			leaderNode: PEER,
+			lastHeartbeat: isoAgo(1000),
+			initializedAt: isoAgo(60_000),
+		});
+		await runFailoverCheckForTests();
+		assert.strictEqual(getEngineRole(), 'follower');
+		assert.strictEqual(handlerRuns.length, 0);
+	});
+
+	it('promotes itself when the leader lease goes stale, and catch-up fires the missed occurrence', async () => {
+		// The peer "dies": its last heartbeat is now well past the stale
+		// threshold (data-driven — no waiting)
+		await stateTable.put({
+			id: LEADER_ROW_ID,
+			leaderNode: PEER,
+			lastHeartbeat: isoAgo(STALE_THRESHOLD_MS * 2),
+			initializedAt: isoAgo(STALE_THRESHOLD_MS * 3),
+		});
+		// The job has run before (two days ago), so its 02:00 UTC occurrence was
+		// missed while the old leader was dead — promotion catch-up must fire it
+		await stateTable.put({
+			id: 'job:election-test:nightly',
+			firstSeenAt: isoAgo(48 * 60 * 60 * 1000),
+			lastRunAt: isoAgo(48 * 60 * 60 * 1000),
+			lastStatus: 'success',
+		});
+		await runFailoverCheckForTests();
+		assert.strictEqual(getEngineRole(), 'leader', 'stale peer lease must yield the leadership');
+		assert.strictEqual(handlerRuns.length, 1, 'promotion catch-up must fire exactly one missed occurrence');
+		assert.strictEqual(handlerRuns[0].catchUp, true);
+		assert.strictEqual(handlerRuns[0].jobName, 'nightly');
+		const lease = await stateTable.get(LEADER_ROW_ID);
+		assert.strictEqual(lease.leaderNode, SELF, 'the new leader must have claimed the lease row');
+	});
+
+	it('renews its own lease on heartbeat', async () => {
+		const before = await stateTable.get(LEADER_ROW_ID);
+		// Condition-wait rather than a fixed sleep: on coarse clocks (Windows CI)
+		// two ISO timestamps a few ms apart can be equal, so tick until the
+		// renewal is observable (each iteration IS a real heartbeat)
+		await waitFor(async () => {
+			await runHeartbeatForTests();
+			const current = await stateTable.get(LEADER_ROW_ID);
+			return Date.parse(current.lastHeartbeat) > Date.parse(before.lastHeartbeat);
+		});
+		const after = await stateTable.get(LEADER_ROW_ID);
+		assert.strictEqual(after.leaderNode, SELF);
+		assert.ok(
+			Date.parse(after.lastHeartbeat) > Date.parse(before.lastHeartbeat),
+			'heartbeat must advance the lease timestamp'
+		);
+	});
+
+	it('steps down when a heartbeat finds another node took the lease (racing write)', async () => {
+		// The peer recovered and wrote a FRESH takeover lease between our
+		// heartbeats — the committed state a racing write leaves behind, which
+		// the takeover check exists to heal (true concurrent-write races are
+		// exercised in the harper-pro cluster harness)
+		await stateTable.put({
+			id: LEADER_ROW_ID,
+			leaderNode: PEER,
+			lastHeartbeat: new Date().toISOString(),
+			initializedAt: new Date().toISOString(),
+		});
+		const runsBefore = handlerRuns.length;
+		await runHeartbeatForTests();
+		assert.strictEqual(getEngineRole(), 'follower', 'a fresh competing lease must trigger step-down');
+		const lease = await stateTable.get(LEADER_ROW_ID);
+		assert.strictEqual(lease.leaderNode, PEER, 'step-down must not overwrite the winner’s lease');
+		assert.strictEqual(handlerRuns.length, runsBefore, 'stepping down must not run jobs');
+	});
+
+	it('wins a cold-start election when it is the preferred node', async () => {
+		stopSchedulerEngine();
+		handlerRuns.length = 0;
+		await stateTable.delete(LEADER_ROW_ID);
+		await stateTable.delete('job:election-test:nightly');
+		// Roster where SELF sorts first: it should elect itself immediately
+		global.server.nodes = [{ name: LAST_PEER }];
+		registerComponentJobs('election-test', [makeCronJob('nightly')]);
+		startSchedulerEngine();
+		await electionSettledForTests();
+		assert.strictEqual(getEngineRole(), 'leader', 'the alphabetically-first node must win a cold start');
+		const lease = await stateTable.get(LEADER_ROW_ID);
+		assert.strictEqual(lease.leaderNode, SELF);
+		// A brand-new job (no prior state row) must NOT fire immediately: the
+		// first sweep records firstSeenAt as its baseline instead
+		assert.strictEqual(handlerRuns.length, 0, 'a never-run job must not fire on promotion');
+		const jobRow = await stateTable.get('job:election-test:nightly');
+		assert.ok(jobRow?.firstSeenAt, 'promotion must record the catch-up baseline for a new job');
+	});
+
+	it('loses a cold-start election to an alphabetically-preferred roster peer', async () => {
+		// No lease row at all: the decision comes ONLY from the roster
+		// (server.nodes + self) via nodeRoster()/pickNextLeader — this is the
+		// one outcome that cannot pass unless that wiring is correct
+		stopSchedulerEngine();
+		handlerRuns.length = 0;
+		await stateTable.delete(LEADER_ROW_ID);
+		await stateTable.delete('job:election-test:nightly');
+		global.server.nodes = [{ name: PEER }];
+		registerComponentJobs('election-test', [makeCronJob('nightly')]);
+		startSchedulerEngine();
+		await electionSettledForTests();
+		assert.strictEqual(getEngineRole(), 'follower', 'a roster peer that sorts first must win the election');
+		assert.strictEqual(handlerRuns.length, 0, 'the election loser must not run jobs');
+		const lease = await stateTable.get(LEADER_ROW_ID);
+		assert.ok(lease == null, 'the election loser must not claim the lease row');
+	});
+
+	it('a stop during an in-flight election prevents the election from resurrecting the engine', async () => {
+		// Start an election and immediately stop the engine — the election's
+		// state read is still in flight; when it settles it must abandon itself
+		// (epoch guard) instead of assigning a role or arming timers
+		stopSchedulerEngine();
+		await stateTable.delete(LEADER_ROW_ID);
+		global.server.nodes = [{ name: LAST_PEER }]; // SELF would win if allowed
+		registerComponentJobs('election-test', [makeCronJob('nightly')]);
+		startSchedulerEngine();
+		const settled = electionSettledForTests();
+		stopSchedulerEngine(); // stop while electRole is awaiting its read
+		await settled;
+		assert.strictEqual(getEngineRole(), 'inactive', 'a stopped engine must stay inactive after the election settles');
+		const lease = await stateTable.get(LEADER_ROW_ID);
+		assert.ok(lease == null, 'an abandoned election must not claim the lease row');
+	});
+
+	it('replaces the job set on re-registration and forgets it on unregister', async () => {
+		registerComponentJobs('election-test', [makeCronJob('nightly'), makeCronJob('hourly')]);
+		assert.deepStrictEqual(getRegisteredJobNames('election-test').sort(), ['hourly', 'nightly']);
+		registerComponentJobs('election-test', [makeCronJob('hourly')]);
+		assert.deepStrictEqual(
+			getRegisteredJobNames('election-test'),
+			['hourly'],
+			're-registration must replace, not accumulate'
+		);
+		unregisterComponentJobs('election-test');
+		assert.deepStrictEqual(getRegisteredJobNames('election-test'), [], 'unregister must forget the component');
+	});
+});