Clone sync verification now detects an in-progress base copy on RocksDB instead of marking the node Available mid-copy

cloneNode/cloneNode.ts

@@ -30,6 +30,7 @@ import {
 } from '../core/utility/hdbTerms.ts';
 import { fetchJWTKeyWithRetry } from './jwtKeyClone.ts';
 import { monitorSyncLoop } from './syncMonitor.ts';
+import { isReplicatedDatabase as isReplicatedDatabaseUnder } from './replicatedDatabases.ts';
 
 /**
  * Environment Variables:
@@ -512,13 +513,38 @@ async function monitorSync(): Promise<SyncOutcome> {
 		`Starting to monitor sync status. Will check every ${DEFAULT_SYNC_CHECK_INTERVAL_MS}ms and fail if no replication data arrives for ${Math.round(stallTimeoutMs / 1000)}s`
 	);
 
+	// Whether the system database's socket is required is a leader-capability question: a legacy
+	// (v4) leader never replicates the system database, so requiring its socket would wedge the
+	// clone, while on a v5+ leader it must be required up front — otherwise a small user database
+	// completing before the system subscription registers could finish the clone with the system
+	// copy unverified. registration_info is the version probe present on every leader version
+	// (see core/bin/cliOperations.ts). Fail CLOSED: only a positively-read legacy major version
+	// exempts system; a missing/unparseable version or a persistently failing probe requires it,
+	// so a transient probe error against a v5 leader cannot reopen the premature-Available race.
+	let systemSocketRequired = true;
+	for (let attempt = 1; attempt <= 3; attempt++) {
+		try {
+			const registration: any = await leaderRequest({ operation: 'registration_info' });
+			// First digit run tolerates prefixed version strings (e.g. "v4.3.7"), which parseInt would NaN.
+			const leaderMajorVersion = Number(String(registration?.version ?? '').match(/\d+/)?.[0] ?? NaN);
+			systemSocketRequired = !(leaderMajorVersion >= 1 && leaderMajorVersion < 5);
+			break;
+		} catch (err) {
+			log(`Leader version probe failed (attempt ${attempt}/3): ${err}`);
+			if (attempt < 3) await sleep(1000);
+		}
+	}
+
 	const outcome = await monitorSyncLoop({
 		targetTimestamps,
 		clusterStatus,
 		leaderReplicationURL,
 		stallTimeoutMs,
 		checkIntervalMs: DEFAULT_SYNC_CHECK_INTERVAL_MS,
 		log,
+		requiredSocketDatabases: Object.keys(targetTimestamps).filter(
+			(database) => database !== 'system' || systemSocketRequired
+		),
 	});
 
 	if (outcome === 'synced') {
@@ -549,16 +575,40 @@ async function monitorSync(): Promise<SyncOutcome> {
  * and record the most recent timestamp for each database in a JSON file.
  * @returns {Promise<void>}
  */
+// A database the clone doesn't subscribe to must not be pre-created (cloneSchemas) or become a
+// sync target (getLastUpdatedRecord: its socket never exists, so a target would wedge the sync
+// monitor). A sharded entry replicates only from a same-shard leader (`shouldReplicateFromNode`);
+// the leader's shard comes from its configuration. Fail closed on an unreadable configuration by
+// treating sharded entries as replicated: a wrong inclusion stalls the clone visibly, a wrong
+// exclusion would skip verifying a database that is being copied.
+function isReplicatedDatabase(dbName: string, shardedReplicates?: (entry: any) => boolean): boolean {
+	return isReplicatedDatabaseUnder(envMgr.get(CONFIG_PARAMS.REPLICATION_DATABASES), dbName, shardedReplicates);
+}
+
+async function leaderShardedReplicates(): Promise<(entry: any) => boolean> {
+	try {
+		const leaderConfiguration: any = await leaderRequest({ operation: 'get_configuration' });
+		const leaderShard = leaderConfiguration?.replication?.shard;
+		const localShard = envMgr.get(CONFIG_PARAMS.REPLICATION_SHARD);
+		return () => leaderShard === localShard;
+	} catch (err) {
+		log(`Could not read the leader configuration for shard matching (${err}); keeping sharded sync targets`);
+		return () => true;
+	}
+}
+
 async function getLastUpdatedRecord(): Promise<Record<string, number>> {
 	log('Getting last updated record timestamp for all database', 'debug');
 	const lastUpdated: Record<string, number> = {};
 	const systemDb: Record<string, any> = await leaderRequest({ operation: 'describe_database', database: 'system' });
 	lastUpdated['system'] = findMostRecentTimestamp(systemDb);
 
+	const shardedReplicates = await leaderShardedReplicates();
 	const allDb: Record<string, any> = await leaderRequest({ operation: 'describe_all' });
 	for (const db in allDb) {
 		// requestId is part of the describe response so we ignore it
 		if (typeof allDb[db] !== 'object') continue;
+		if (!isReplicatedDatabase(db, shardedReplicates)) continue;
 		lastUpdated[db] = findMostRecentTimestamp(allDb[db]);
 	}
 
@@ -912,19 +962,6 @@ async function cloneSchemas(): Promise<void> {
 	const { createSchema, createTable } = await import('../core/dataLayer/schema.js');
 	const { databases } = await import('../core/resources/databases.js');
 
-	// Filter by this node's `replication.databases` so we don't materialize empty databases the
-	// clone isn't even subscribing to. Matches the gating used by `shouldReplicateFromNode` in
-	// `replication/knownNodes.ts`: `undefined` or `'*'` accept everything; an array accepts only
-	// the names it lists (objects with `.name` are sharded-database entries).
-	const databaseReplications = envMgr.get(CONFIG_PARAMS.REPLICATION_DATABASES);
-	const isReplicatedDatabase = (dbName: string): boolean => {
-		if (!databaseReplications || databaseReplications === '*') return true;
-		if (!Array.isArray(databaseReplications)) return true;
-		return databaseReplications.some((entry: any) =>
-			typeof entry === 'string' ? entry === dbName : entry?.name === dbName
-		);
-	};
-
 	for (const dbName of Object.keys(allDb)) {
 		const dbDescribe = allDb[dbName];
 		if (!dbDescribe || typeof dbDescribe !== 'object' || dbName === SYSTEM_SCHEMA_NAME) continue;

cloneNode/replicatedDatabases.ts

@@ -0,0 +1,21 @@
+/**
+ * Whether `dbName` is replicated under a `replication.databases` config value, mirroring the
+ * name/shard gating of `shouldReplicateFromNode` (`replication/knownNodes.ts`): `undefined` or
+ * `'*'` accept everything; an array accepts the names it lists, and a sharded entry only when
+ * `shardedReplicates` accepts it (same-shard leader). Callers that cannot evaluate the shard
+ * predicate must pass a fail-closed (always-true) predicate: a wrong inclusion stalls the clone
+ * visibly, while a wrong exclusion would skip verification of a database that is being copied.
+ */
+export function isReplicatedDatabase(
+	databaseReplications: unknown,
+	dbName: string,
+	shardedReplicates: (entry: any) => boolean = () => true
+): boolean {
+	if (!databaseReplications || databaseReplications === '*') return true;
+	if (!Array.isArray(databaseReplications)) return true;
+	return databaseReplications.some((entry: any) =>
+		typeof entry === 'string'
+			? entry === dbName
+			: entry?.name === dbName && (!entry.sharded || shardedReplicates(entry))
+	);
+}

cloneNode/syncMonitor.ts

@@ -6,11 +6,13 @@ export type SyncCheckResult = {
 	syncComplete: boolean;
 	/** Most recent arrival stamp (ms epoch) among databases still below their target; 0 if none. */
 	latestReceivedMs: number;
+	/** Databases that had a replication socket to the leader in this check. */
+	socketDatabases: Set<string>;
 };
 
 /**
- * Check whether every database with a target timestamp has caught up, and report the freshest
- * arrival stamp among the databases that have not.
+ * Check whether every database has caught up with the leader, and report the freshest arrival
+ * stamp among the databases that have not.
  *
  * Completion and liveness are deliberately separate signals: `lastReceivedVersion` is frozen for the
  * whole bulk copy (it jumps to copyStartTime only on the post-copy end_txn), so it can only answer
@@ -21,42 +23,47 @@ export async function checkSyncStatus(
 	targetTimestamps: Record<string, number>,
 	clusterStatus: () => Promise<any>,
 	leaderReplicationURL: string,
-	log: SyncMonitorLog
+	log: SyncMonitorLog,
+	requiredSocketDatabases: string[] = Object.keys(targetTimestamps)
 ): Promise<SyncCheckResult> {
 	const clusterResponse = await clusterStatus();
 	log(`clone sync check cluster status response: ${JSON.stringify(clusterResponse)}`, 'debug');
 
 	if (!clusterResponse) {
 		log('No cluster status response received for clone, will wait and retry');
-		return { syncComplete: false, latestReceivedMs: 0 };
+		return { syncComplete: false, latestReceivedMs: 0, socketDatabases: new Set() };
 	}
 
 	if (!clusterResponse.connections?.length) {
 		log('No connections found in cluster status response for clone, will wait and retry');
-		return { syncComplete: false, latestReceivedMs: 0 };
+		return { syncComplete: false, latestReceivedMs: 0, socketDatabases: new Set() };
 	}
 
 	const leaderConnection = clusterResponse.connections.find((conn) => conn.url === leaderReplicationURL);
 
 	if (!leaderConnection) {
 		log('No connection found matching leader replication URL, will wait and retry');
-		return { syncComplete: false, latestReceivedMs: 0 };
+		return { syncComplete: false, latestReceivedMs: 0, socketDatabases: new Set() };
 	}
 
 	if (!leaderConnection.database_sockets?.length) {
 		log(`No database sockets found for connection leader ${leaderConnection.name}`, 'debug');
-		return { syncComplete: false, latestReceivedMs: 0 };
+		return { syncComplete: false, latestReceivedMs: 0, socketDatabases: new Set() };
 	}
 
 	let syncComplete = true;
 	let latestReceivedMs = 0;
+	const socketDatabases = new Set<string>();
 	for (const socket of leaderConnection.database_sockets) {
 		const dbName = socket.database;
-		const targetTime = targetTimestamps[dbName];
-		if (!targetTime) {
-			log(`Database ${dbName}: No target timestamp, skipping sync check`, 'debug');
-			continue;
-		}
+		socketDatabases.add(dbName);
+		// A missing target — an empty database, or a leader whose describe cannot report
+		// last_updated_record (RocksDB, harper#2091) — must not skip verification, or the check
+		// passes vacuously when every target is absent (#655). The received-version watermark is
+		// held at 0 for the whole bulk copy and only becomes positive via the final end_txn the
+		// sender emits at copyStartTime, so a positive watermark is the copy's own completion
+		// signal, independent of the leader's describe support.
+		const targetTime = targetTimestamps[dbName] || 1;
 
 		// Raw version (high-precision float64) preserves the sub-millisecond precision needed for
 		// an accurate comparison against the leader's last_updated_record targets.
@@ -83,7 +90,20 @@ export async function checkSyncStatus(
 		if (Number.isFinite(receivedAt) && receivedAt > latestReceivedMs) latestReceivedMs = receivedAt;
 	}
 
-	return { syncComplete, latestReceivedMs };
+	// A required database with no socket yet (its subscription is still registering with the
+	// main thread) is pending, not verified — otherwise a lone early socket (e.g. the system DB,
+	// whose small copy finishes in seconds) could complete the check before the data databases'
+	// sockets even appear. Only databases the clone actually subscribes to are required: a legacy
+	// (v4) leader never replicates the system database, so demanding its socket would wedge the
+	// clone; when the socket does exist it is still verified by the loop above.
+	for (const dbName of requiredSocketDatabases) {
+		if (!socketDatabases.has(dbName)) {
+			log(`Database ${dbName}: no replication socket to the leader yet`, 'debug');
+			syncComplete = false;
+		}
+	}
+
+	return { syncComplete, latestReceivedMs, socketDatabases };
 }
 
 export type MonitorSyncLoopOptions = {
@@ -93,6 +113,8 @@ export type MonitorSyncLoopOptions = {
 	stallTimeoutMs: number;
 	checkIntervalMs: number;
 	log: SyncMonitorLog;
+	/** Databases whose replication socket must exist before sync can complete (default: every target). */
+	requiredSocketDatabases?: string[];
 	/** Test hooks: injectable clock and delay. */
 	now?: () => number;
 	delay?: (ms: number) => Promise<unknown>;
@@ -109,6 +131,12 @@ export async function monitorSyncLoop(options: MonitorSyncLoopOptions): Promise<
 	const delay = options.delay ?? sleep;
 	let lastProgressAt = now();
 	let loopCount = 0;
+	const baseRequired = options.requiredSocketDatabases ?? Object.keys(options.targetTimestamps);
+	// Ratchet: a target database whose socket has been seen once stays required even if the socket
+	// later drops, and a non-required one (e.g. `system`, optional because v4 leaders never
+	// replicate it) becomes required as soon as its socket appears — so on a v5 leader a small user
+	// database finishing first cannot complete the clone while the system copy is still pending.
+	const seenTargetSockets = new Set<string>();
 
 	while (now() - lastProgressAt < options.stallTimeoutMs) {
 		try {
@@ -121,7 +149,8 @@ export async function monitorSyncLoop(options: MonitorSyncLoopOptions): Promise<
 				options.targetTimestamps,
 				options.clusterStatus,
 				options.leaderReplicationURL,
-				options.log
+				options.log,
+				[...new Set([...baseRequired, ...seenTargetSockets])]
 			);
 			checkPromise.catch(() => {});
 			const result = await Promise.race([
@@ -133,7 +162,10 @@ export async function monitorSyncLoop(options: MonitorSyncLoopOptions): Promise<
 				await delay(options.checkIntervalMs);
 				continue;
 			}
-			const { syncComplete, latestReceivedMs } = result;
+			const { syncComplete, latestReceivedMs, socketDatabases } = result;
+			for (const dbName of socketDatabases) {
+				if (dbName in options.targetTimestamps) seenTargetSockets.add(dbName);
+			}
 
 			if (syncComplete) return 'synced';
 

unitTests/cloneNode/replicatedDatabases.test.mjs

@@ -0,0 +1,39 @@
+import assert from 'node:assert/strict';
+import { isReplicatedDatabase } from '#src/cloneNode/replicatedDatabases';
+
+describe('isReplicatedDatabase', () => {
+	it('accepts everything when replication.databases is unset or a wildcard', () => {
+		assert.equal(isReplicatedDatabase(undefined, 'data'), true);
+		assert.equal(isReplicatedDatabase('*', 'data'), true);
+	});
+
+	it('matches string entries by name', () => {
+		assert.equal(isReplicatedDatabase(['data'], 'data'), true);
+		assert.equal(isReplicatedDatabase(['data'], 'other'), false);
+	});
+
+	it('matches unsharded object entries by name regardless of the shard predicate', () => {
+		assert.equal(
+			isReplicatedDatabase([{ name: 'data' }], 'data', () => false),
+			true
+		);
+	});
+
+	it('accepts a sharded entry only when the shard predicate does (same-shard leader)', () => {
+		const entries = [{ name: 'data', sharded: true }];
+		assert.equal(
+			isReplicatedDatabase(entries, 'data', () => true),
+			true
+		);
+		assert.equal(
+			isReplicatedDatabase(entries, 'data', () => false),
+			false
+		);
+	});
+
+	it('fails closed for a sharded entry when no shard predicate is supplied', () => {
+		// Callers that cannot evaluate the leader's shard must keep the database as a sync target:
+		// a wrong inclusion stalls the clone visibly, a wrong exclusion skips verifying a copy.
+		assert.equal(isReplicatedDatabase([{ name: 'data', sharded: true }], 'data'), true);
+	});
+});