cherry-pick: Place replication subscriptions round-robin with record locks on, relaying control entries to the coordinating worker (#987 → v5.3)

.github/glossary

@@ -23,4 +23,4 @@ home map: operator-agreed list of nodes participating in record locks, published
 home generation: version number of the home map, which staging refuses unless above every generation the node has seen
 quiesce: post-staging drain in which a node surrenders every delegation it holds and recalls every grant it issued
 lock barrier: record-lock control entry whose commit, matched on origin, position and nonce, proves that origin's prior writes applied
-coordinating worker: the one worker thread per database that owns its record-lock coordination and, with recordLocks on, its peer subscriptions
+coordinating worker: the one worker thread per database that owns its record-lock coordination; peer subscriptions are placed independently and relay control entries to it

integrationTests/cluster/fixture-record-lock-bench/resources.js

@@ -89,6 +89,41 @@ export class BenchWrite extends Resource {
 	}
 }
 
+export class BenchWriteBatched extends Resource {
+	async post(data) {
+		const started = performance.now();
+		const batch = data.batch ?? 50;
+		for (let i = 0; i < data.count; i += batch) {
+			const context = {};
+			const end = Math.min(i + batch, data.count);
+			await transaction(context, async () => {
+				const puts = [];
+				let thrown;
+				for (let j = i; j < end; j++) {
+					try {
+						puts.push(tables.Counter.put({ id: `${data.prefix}-${j}`, n: j }, context));
+					} catch (error) {
+						thrown ??= error;
+						break;
+					}
+				}
+				const settled = await Promise.allSettled(puts);
+				const failed = settled.find((result) => result.status === 'rejected');
+				if (thrown) throw thrown;
+				if (failed) throw failed.reason;
+			});
+		}
+		return { elapsedMs: performance.now() - started };
+	}
+}
+
+export class BenchCount extends Resource {
+	async get() {
+		const { recordCount } = await tables.Counter.getRecordCount({ exactCount: true });
+		return { recordCount };
+	}
+}
+
 /** What a cluster-scoped lock() answers here: the observable for which enablement arm a node is in. */
 export class LockProbe extends Resource {
 	async post(data) {

integrationTests/cluster/recordLockCluster.test.mjs

@@ -75,7 +75,7 @@ suite('cluster record locks: three-node full mesh', { timeout: 420_000 }, (ctx)
 			assert.equal(typeof locks.ownerThreadId, 'number', 'the single http worker coordinates the database');
 			assert.equal(locks.granted, 0);
 			assert.equal(locks.admitted, 0);
-			assert.equal(locks.droppedOffOwner, 0, 'nothing is applied off the coordinating worker');
+			assert.equal(locks.droppedOffOwner, 0, 'no control entry failed to reach the coordinating worker');
 			rings.push(JSON.stringify(locks.members));
 		}
 		// The whole point of deriving the home from an agreed set: every node computes the same ring.
@@ -454,3 +454,109 @@ suite('cluster record locks at threads.count > 1 (harper-pro#852)', { timeout: 4
 		// land off the owner is routing-dependent. Exclusion across the node boundary is the real proof.
 	});
 });
+
+suite(
+	'cluster record locks: a release applied off the coordinating worker (harper-pro#977)',
+	{ timeout: 420_000 },
+	(ctx) => {
+		const contexts = [];
+		let multi; // the home: several http workers, subscriptions placed round-robin across them
+		let peers; // two single-worker delegates, so one of their streams provably lands off multi's owner
+		let nodes;
+
+		before(async () => {
+			const started = await Promise.allSettled([
+				startNode(ctx.name, {}, 3),
+				startNode(ctx.name, {}, 1),
+				startNode(ctx.name, {}, 1),
+			]);
+			for (const result of started) if (result.status === 'fulfilled') contexts.push(result.value);
+			const failed = started.find((result) => result.status === 'rejected');
+			if (failed) throw failed.reason;
+			[multi, ...peers] = contexts.map((c) => c.harper);
+			nodes = [multi, ...peers];
+			await connectMesh(nodes);
+			await bootstrapHomeMap(nodes);
+			await waitForRing(nodes, nodes.length);
+		});
+
+		after(async () => {
+			for (const c of contexts) await stopNode(c);
+		});
+
+		async function offOwnerDelegate() {
+			const status = await clusterStatusOf(multi);
+			const ownerThreadId = status.recordLocks?.[DB]?.ownerThreadId;
+			const placement = status.connections.map((connection) => ({
+				peer: connection.name,
+				threadId: connection.database_sockets.find((socket) => socket.database === DB)?.threadId,
+			}));
+			const off = placement.find((entry) => typeof entry.threadId === 'number' && entry.threadId !== ownerThreadId);
+			assert.ok(
+				off,
+				`no peer's ${DB} subscription is applied off multi's coordinating worker ${ownerThreadId}: ${JSON.stringify(placement)}`
+			);
+			const delegate = peers.find((peer) => peer.hostname === off.peer || off.peer.startsWith(peer.hostname));
+			assert.ok(delegate, `peer ${off.peer} is not one of ${peers.map((peer) => peer.hostname)}`);
+			return { delegate, contender: peers.find((peer) => peer !== delegate), ownerThreadId, placement };
+		}
+
+		/** multi's `granted` gauge moves only for its own grants. */
+		async function holdKeyHomedOnMulti(delegate) {
+			for (let i = 0; i < 40; i++) {
+				const id = `relay-${Date.now()}-${i}`;
+				const before = (await clusterStatusOf(multi)).recordLocks[DB].granted;
+				const held = await call(delegate, 'LockHold/', { id, lease: 30_000, timeout: 5_000 });
+				assert.equal(held.status, 200, `hold ${id} on ${delegate.hostname}: ${JSON.stringify(held.body)}`);
+				if ((await clusterStatusOf(multi)).recordLocks[DB].granted > before) return { id, token: held.body.token };
+				await call(delegate, 'LockRelease/', { token: held.body.token });
+			}
+			assert.fail('no key homed on multi in 40 tries');
+		}
+
+		test("a delegate's release reaches the home from a non-owner worker, so a contender is admitted fresh", async () => {
+			const { delegate, contender, ownerThreadId, placement } = await offOwnerDelegate();
+			const { id, token } = await holdKeyHomedOnMulti(delegate);
+			const written = await call(delegate, 'LockWrite/', { token, n: 7 });
+			assert.equal(written.status, 200, JSON.stringify(written.body));
+			const released = await call(delegate, 'LockRelease/', { token });
+			assert.equal(released.status, 200);
+			// The delegation stays with `delegate` after unlock. A contender makes multi recall it; the
+			// delegate's lockRelease then replicates to multi over the stream placed OFF its coordinating
+			// worker, and only the relay lets the home clear the grant before the delegation's own lease
+			// (minutes) — inside this timeout, a dropped release answers 423.
+			const before = (await clusterStatusOf(multi)).recordLocks[DB];
+			const contended = await call(contender, 'LockedIncrement/', { id, timeout: 30_000 });
+			assert.equal(
+				contended.status,
+				200,
+				`${contender.hostname} was not admitted after ${delegate.hostname} released (placement ${JSON.stringify(placement)}, owner ${ownerThreadId}): ${JSON.stringify(contended.body)}`
+			);
+			assert.equal(contended.body.n, 8, "the successor read the predecessor's write (successor freshness)");
+			// And the home itself as the next successor: its own lock() relays to the owner worker, and the
+			// contender's release in turn reaches it over whichever worker applies that stream.
+			const own = await call(multi, 'LockedIncrement/', { id, timeout: 30_000 });
+			assert.equal(own.status, 200, JSON.stringify(own.body));
+			assert.equal(own.body.n, 9);
+			const after = (await clusterStatusOf(multi)).recordLocks[DB];
+			assert.ok(
+				(after.relayedControlEntries ?? 0) > (before.relayedControlEntries ?? 0),
+				`no control entry was relayed to multi's coordinating worker: ${JSON.stringify(after)}`
+			);
+			assert.equal(
+				after.droppedOffOwner,
+				0,
+				`a release failed to reach the coordinating worker: ${JSON.stringify(after)}`
+			);
+			assert.equal(after.controlEntryRelayDrops, 0, `releases could not be relayed: ${JSON.stringify(after)}`);
+			assert.ok(after.outbound, 'the owner reports how its lock operations reached the peers');
+			assert.equal(
+				after.outbound.fresh,
+				0,
+				`the owner opened a connection per call: ${JSON.stringify(after.outbound)}`
+			);
+			const finalValues = await waitForCounter(nodes, id, 9);
+			assert.deepEqual(finalValues, [9, 9, 9]);
+		});
+	}
+);

integrationTests/cluster/recordLockCost.bench.mjs

@@ -58,6 +58,9 @@ const REACQUISITION_LEASES_MS = process.env.RECORD_LOCK_BENCH_REACQ_LEASES_MS
 const REACQUISITION_CADENCE_MS = Number(process.env.RECORD_LOCK_BENCH_REACQ_CADENCE_MS) || 5_000;
 const REACQUISITION_MAX_RUN_MS = Number(process.env.RECORD_LOCK_BENCH_REACQ_RUN_MS) || 260_000;
 const OUT = process.env.RECORD_LOCK_BENCH_OUT || join(tmpdir(), `record-lock-cost-${Date.now()}.json`);
+/** http workers per node. Above 1, an off-owner lock() relays its admission and the owner's outbound
+ * lock operations are forwarded to the worker holding the session (harper-pro#852, #977). */
+const THREADS = Number(process.env.RECORD_LOCK_BENCH_THREADS) || 1;
 
 const results = {
 	machine: {
@@ -67,6 +70,7 @@ const results = {
 		platform: `${process.platform}-${process.arch}`,
 		node: process.version,
 	},
+	threads: THREADS,
 	ranAt: new Date().toISOString(),
 };
 
@@ -85,7 +89,7 @@ function optionsFor(hostname, replication) {
 		config: {
 			analytics: { aggregatePeriod: -1 },
 			logging: { colors: false, stdStreams: true, console: true, level: 'warn' },
-			threads: { count: 1 },
+			threads: { count: THREADS },
 			replication,
 		},
 		// `bootstrapHomeMap` stages and activates back to back, so the drain backstop has to go, as it
@@ -370,6 +374,36 @@ function logDelta(before, after, acquisitions) {
 	});
 }
 
+/** The owner's `granted` gauge; the fixture's in-process gauge is the serving worker's, not the owner's. */
+async function grantedOn(node) {
+	const status = await sendOperation(node, { operation: 'cluster_status' });
+	const granted = status.recordLocks?.[DB]?.granted;
+	if (typeof granted !== 'number') throw new Error(`${node.hostname}: the owner's granted gauge is unavailable`);
+	return granted;
+}
+
+/** `BenchLock` over `ids` with each key's home classified; above one worker the serving worker's own
+ * gauge cannot see the owner's grants, so one request per id is bracketed by the owner's gauge. */
+async function lockEachClassified(node, ids, options) {
+	if (THREADS === 1) return call(node, 'BenchLock/', { ...options, ids, classifyHome: true });
+	// Delegated before the first baseline so its own grant never lands inside a measured interval.
+	const warmupId = `warm-${ids[0]}`;
+	await call(node, 'BenchLock/', { ids: [warmupId] });
+	const acquireMs = [];
+	const releaseMs = [];
+	const atMs = [];
+	const homeLocal = [];
+	for (const id of ids) {
+		const before = await grantedOn(node);
+		const one = await call(node, 'BenchLock/', { ...options, ids: [id], warmupId });
+		homeLocal.push((await grantedOn(node)) > before);
+		acquireMs.push(one.acquireMs[0]);
+		releaseMs.push(one.releaseMs[0]);
+		atMs.push(one.atMs[0]);
+	}
+	return { acquireMs, releaseMs, atMs, homeLocal };
+}
+
 /**
  * Lock fresh ids from `node` until every side of the ring in `wanted` has one, keeping the answer:
  * the probe IS that key's first lock, so a later batch on the same id measures only repeats.
@@ -380,7 +414,7 @@ async function probeKeys(node, prefix, wanted, options) {
 	const found = {};
 	for (let attempt = 0; attempt < 40 && !wanted.every((side) => found[side]); attempt++) {
 		const id = `${prefix}-${attempt}-${Date.now()}`;
-		const first = await call(node, 'BenchLock/', { ...options, ids: [id], classifyHome: true });
+		const first = await lockEachClassified(node, [id], options);
 		found[first.homeLocal[0] ? 'local' : 'remote'] ??= { id, first };
 	}
 	for (const side of wanted) assert.ok(found[side], `${prefix}: no ${side}-home key found`);
@@ -446,7 +480,7 @@ suite('record lock cost: 3-node full mesh, replication.recordLocks on', { timeou
 		for (const [i, node] of nodes.entries()) {
 			const ids = Array.from({ length: UNCONTENDED_PER_NODE }, (_, k) => `uncontended-${i}-${k}-${Date.now()}`);
 			const before = await logSnapshot(nodes);
-			const { acquireMs, releaseMs, homeLocal } = await call(node, 'BenchLock/', { ids, classifyHome: true });
+			const { acquireMs, releaseMs, homeLocal } = await lockEachClassified(node, ids);
 			const after = await logSnapshotAfter(nodes);
 			pooledAcquire.push(...acquireMs);
 			pooledRelease.push(...releaseMs);