Cluster record locks: operator-agreed home map transport and successor-freshness barriers (harper-pro#825, harper#2542 inside #822)

.oxlintrc.json

@@ -33,6 +33,7 @@
 		"server": "readonly",
 		"databases": "readonly",
 		"tables": "readonly",
+		"transaction": "readonly",
 		"createBlob": "readonly",
 		"lockdown": "readonly",
 		"threads": "readonly",

benchmarks/recordLockExitCommit/doomedWorker.mjs

@@ -0,0 +1,18 @@
+/**
+ * The departed off-owner worker of harper-pro#865 ledger item 5: it staged writes under a relayed
+ * admission, unlocked, handed the commits to the engine, and is about to be terminated. Nothing is
+ * awaited — the state under test is a commit whose calling thread abandons it.
+ */
+import { parentPort, workerData } from 'node:worker_threads';
+import { RocksDatabase } from '@harperfast/rocksdb-js';
+
+const db = new RocksDatabase(workerData.path);
+db.open();
+
+const value = 'x'.repeat(workerData.valueBytes);
+for (let i = 0; i < workerData.count; i++) {
+	if (workerData.mode === 'transaction') db.transaction(() => db.put(`contested-${i}`, { writer: 'doomed', i, value }));
+	else db.put(`contested-${i}`, { writer: 'doomed', i, value });
+}
+
+parentPort.postMessage({ handedOff: workerData.count, stalledAtHandoff: db.isWriteStalled?.() ?? null });

benchmarks/recordLockExitCommit/exitCommit.probe.mjs

@@ -0,0 +1,113 @@
+/**
+ * harper-pro#865 ledger item 5 / rocksdb-js#865: a worker's EXIT is counted as a completed fence on
+ * the ownerless-handoff path, and the argument for that is the process-wide native key lock. But a
+ * handle is revoked after `unlock()` has already returned the key (`core/resources/recordLock.ts`
+ * `revokeLease()`), so the key lock does not cover a staged write still in flight. The open question
+ * has been whether a commit already handed to the engine can outlive the thread that issued it, and
+ * land after a successor was admitted.
+ *
+ *   npm run probe:record-lock-exit-commit
+ *
+ * Two things are measured, because they have different answers:
+ *
+ *   1. SURVIVAL — does an abandoned commit still land? The doomed worker hands `count` writes to the
+ *      engine, is terminated, and the parent counts how many became visible.
+ *   2. ORDERING — can the doomed worker's write land AFTER a successor's write to the same key?
+ *      That, not survival, is the lost update exit-as-fenced would allow. The parent writes every
+ *      contested key as the successor and reports who won each one.
+ *
+ * `AWAIT_TERMINATE=0` starts the successor without awaiting `worker.terminate()`, because awaiting it
+ * would let the runtime drain the thread first — which is the assumption under test.
+ *
+ * A measurement, not a gate: `.probe.` keeps it out of the `*.test.*` globs, and nothing here asserts.
+ */
+import { Worker } from 'node:worker_threads';
+import { mkdtempSync, writeFileSync } from 'node:fs';
+import { cpus, tmpdir, totalmem } from 'node:os';
+import { performance } from 'node:perf_hooks';
+import { join } from 'node:path';
+import { RocksDatabase } from '@harperfast/rocksdb-js';
+
+const COUNT = Number(process.env.RECORD_LOCK_EXIT_COUNT) || 4_000;
+const VALUE_BYTES = Number(process.env.RECORD_LOCK_EXIT_VALUE_BYTES) || 131_072;
+const TRIALS = Number(process.env.RECORD_LOCK_EXIT_TRIALS) || 5;
+const AWAIT_TERMINATE = process.env.AWAIT_TERMINATE !== '0';
+const MODES = (process.env.RECORD_LOCK_EXIT_MODES || 'put,transaction').split(',');
+const OUT = process.env.RECORD_LOCK_EXIT_OUT || join(tmpdir(), `record-lock-exit-commit-${Date.now()}.json`);
+
+async function trial(mode) {
+	const db = new RocksDatabase(join(mkdtempSync(join(tmpdir(), 'record-lock-exit-')), 'db'));
+	db.open();
+	const worker = new Worker(new URL('./doomedWorker.mjs', import.meta.url), {
+		workerData: { path: db.path, count: COUNT, valueBytes: VALUE_BYTES, mode },
+	});
+	const handed = await new Promise((resolve) => worker.once('message', resolve));
+
+	const terminating = worker.terminate();
+	if (AWAIT_TERMINATE) await terminating;
+	const terminatedAt = performance.now();
+
+	let survived = 0;
+	for (let i = 0; i < COUNT; i++) if (db.get(`contested-${i}`) !== undefined) survived++;
+
+	// The successor: admitted only because the exit was counted as a fence, writing the same keys.
+	const successorWrites = [];
+	for (let i = 0; i < COUNT; i++) successorWrites.push(db.put(`contested-${i}`, { writer: 'successor', i }));
+	await Promise.all(successorWrites);
+	const successorCommitMs = performance.now() - terminatedAt;
+
+	let doomedWon = 0;
+	for (let i = 0; i < COUNT; i++) if (db.get(`contested-${i}`)?.writer === 'doomed') doomedWon++;
+	db.close();
+
+	return {
+		mode,
+		handedOff: handed.handedOff,
+		stalledAtHandoff: handed.stalledAtHandoff,
+		survivedTerminate: survived,
+		doomedWonTheKey: doomedWon,
+		successorWon: COUNT - doomedWon,
+		successorCommitMs: Number(successorCommitMs.toFixed(3)),
+	};
+}
+
+const trials = [];
+for (const mode of MODES) for (let i = 0; i < TRIALS; i++) trials.push(await trial(mode));
+
+console.log(
+	`\n${COUNT} contested keys x ${VALUE_BYTES} B, ${TRIALS} trial(s) per mode, awaitTerminate=${AWAIT_TERMINATE}, Node ${process.version}\n`
+);
+console.log('| mode        | survived terminate | doomed won | successor won | successor commit ms |');
+console.log('| ----------- | ------------------ | ---------- | ------------- | ------------------- |');
+for (const t of trials)
+	console.log(
+		`| ${t.mode.padEnd(11)} | ${`${t.survivedTerminate} / ${COUNT}`.padEnd(18)} | ${String(t.doomedWonTheKey).padEnd(10)} | ${String(t.successorWon).padEnd(13)} | ${String(t.successorCommitMs).padEnd(19)} |`
+	);
+const inversions = trials.reduce((total, t) => total + t.doomedWonTheKey, 0);
+console.log(
+	`\nabandoned commits that still landed: ${trials.reduce((total, t) => total + t.survivedTerminate, 0)} of ${trials.length * COUNT}` +
+		`\nordering inversions (a dead worker's write over a successor's): ${inversions} of ${trials.length * COUNT}\n`
+);
+
+writeFileSync(
+	OUT,
+	`${JSON.stringify(
+		{
+			machine: {
+				cpu: cpus()[0]?.model,
+				cores: cpus().length,
+				memoryGiB: Math.round(totalmem() / 1024 ** 3),
+				platform: `${process.platform}-${process.arch}`,
+				node: process.version,
+			},
+			ranAt: new Date().toISOString(),
+			keysPerTrial: COUNT,
+			valueBytes: VALUE_BYTES,
+			awaitTerminate: AWAIT_TERMINATE,
+			trials,
+		},
+		undefined,
+		'\t'
+	)}\n`
+);
+console.log(`raw: ${OUT}`);

benchmarks/recordLockRelay/coordinatorWorker.mjs

@@ -0,0 +1,50 @@
+/**
+ * The coordinating worker's side of `transport.bench.mjs`: answers an acquire over the direct
+ * worker-to-worker port and over `notify()`, so the two transports are measured against the same
+ * handler doing the same work.
+ *
+ * `busyMs` is the point of the file. A relayed acquire is answered from the owner's event loop, so
+ * it waits out whatever turn that worker is already in — the spin loop is a stand-in for the owner
+ * serving its own requests, which is the normal state of a worker at `threads.count > 1`.
+ */
+import { parentPort, workerData } from 'node:worker_threads';
+import { performance } from 'node:perf_hooks';
+import { RocksDatabase } from '@harperfast/rocksdb-js';
+
+const db = new RocksDatabase(workerData.path);
+db.open();
+
+const peer = workerData.port;
+let admissionId = 1;
+let busyMs = 0;
+
+const mint = (message) => ({
+	requestId: message.requestId,
+	database: message.database,
+	table: message.table,
+	key: message.key,
+	round: { tsR: Date.now(), mintedMono: performance.now(), admissionId: admissionId++ },
+	session: workerData.session,
+});
+
+const spin = () => {
+	if (busyMs > 0) {
+		const until = performance.now() + busyMs;
+		while (performance.now() < until);
+	}
+	setImmediate(spin);
+};
+setImmediate(spin);
+
+peer.on('message', (message) => {
+	if (message.type === 'load') {
+		busyMs = message.busyMs;
+		peer.postMessage({ type: 'load-ack', requestId: message.requestId });
+	} else if (message.type === 'acquire') {
+		peer.postMessage({ type: 'acquire-reply', ...mint(message) });
+	}
+});
+
+db.on('lock-acquire', (message) => db.notify('lock-acquire-reply', mint(message)));
+
+parentPort.postMessage({ ready: true });