Record locks: drain by recall so a membership change is not a ~6 minute lock outage

replication/recordLockHomes.ts

@@ -26,7 +26,11 @@ import { createHash } from 'node:crypto';
 import Joi from 'joi';
 import { table } from '../core/resources/databases.ts';
 import { transaction } from '../core/resources/transaction.ts';
-import { DELEGATION_LEASE_MS, LOCK_LEASE_SKEW_MS } from '../core/resources/recordLockCoordinator.ts';
+import {
+	DELEGATION_LEASE_MS,
+	LOCK_LEASE_SKEW_MS,
+	type QuiesceResult,
+} from '../core/resources/recordLockCoordinator.ts';
 import { validateBySchema } from '../core/validation/validationWrapper.js';
 import { handleHDBError, hdbErrors, ClientError } from '../core/utility/errors/hdbError.js';
 import * as logger from '../core/utility/logging/harper_logger.js';
@@ -237,7 +241,32 @@ export function planStage(
 	};
 }
 
-export async function stageGeneration(request: any): Promise<{ staged: RecordLockGenerationState }> {
+/**
+ * Never throws: a stage that durably landed must not report failure because the drain did, or the
+ * operator retries a transition that already happened. A drain that cannot run at all is reported as
+ * unknown, which reads to an orchestrator exactly like outstanding work — fall back to the timer.
+ */
+async function drainForStage(database: string): Promise<QuiesceResult | { error: string }> {
+	try {
+		return await drainReader(database, STAGE_DRAIN_BUDGET_MS);
+	} catch (error) {
+		logger.warn?.(`Record lock quiesce for ${database} could not complete; fall back to the drain interval`, error);
+		return { error: (error as Error)?.message ?? String(error) };
+	}
+}
+
+export async function stageGeneration(
+	request: any
+): Promise<{ staged: RecordLockGenerationState; quiesced: QuiesceResult | { error: string } }> {
+	const staged = await stageRow(request);
+	// Deliberately OUTSIDE `withRow`: the drain waits on live critical sections, and holding the
+	// database's transition queue for that would block every other stage, fence and activate on this
+	// node behind it. The row write already retracted `active`, so nothing new can be granted while
+	// this runs, and a concurrent transition is free to proceed.
+	return { ...staged, quiesced: await drainForStage(request.database) };
+}
+
+async function stageRow(request: any): Promise<{ staged: RecordLockGenerationState }> {
 	const validation = validateBySchema(request, stageSchema);
 	if (validation)
 		throw handleHDBError(validation, validation.message, HTTP_STATUS_CODES.BAD_REQUEST, undefined, undefined, true);
@@ -415,6 +444,34 @@ export function setHomesMembershipReaders(readers: MembershipReaders): void {
 	membership = readers;
 }
 
+/**
+ * Installed by `recordLockTransport.ts`: drain on the thread that coordinates the database, not on
+ * whichever one answered the operation. Defaults to refusing rather than reporting a clean drain it
+ * never performed.
+ */
+let drainReader: (database: string, budgetMs: number) => Promise<QuiesceResult | { error: string }> = async () => ({
+	error: 'no record lock drain is wired on this node',
+});
+/**
+ * An orchestrator may skip the drain interval for a node ONLY on this: a drain that both completed
+ * and found nothing. `complete` is core's statement that the sweep could have seen everything —
+ * coordinators are built lazily, so an empty `outstanding` alone is not a proof (harper-pro#856).
+ */
+export function provesQuiescence(quiesced: QuiesceResult | { error: string } | undefined): boolean {
+	// Every field is checked positively: this value crosses a worker boundary, so a malformed reply must
+	// read as "not proven" rather than slipping through on a missing `length`.
+	return (
+		!!quiesced &&
+		!('error' in quiesced) &&
+		quiesced.complete === true &&
+		Array.isArray(quiesced.outstanding) &&
+		quiesced.outstanding.length === 0
+	);
+}
+export function setHomesDrainReader(reader: typeof drainReader): void {
+	drainReader = reader;
+}
+
 export interface HomesProposal {
 	generation: number;
 	homes: string[];
@@ -506,6 +563,15 @@ const proposeSchema = Joi.object({
 });
 
 export const DELEGATION_DRAIN_MS = DELEGATION_LEASE_MS + LOCK_LEASE_SKEW_MS;
+/**
+ * How long `stage` spends draining before it reports what is left (harper-pro#856). A drain waits on
+ * live critical sections, which are milliseconds in the normal case and bounded by the caller's own
+ * lock lease; this is a reporting bound, not a safety one — whatever is still outstanding is returned,
+ * and the operator falls back to `DELEGATION_DRAIN_MS` for those nodes.
+ */
+export const STAGE_DRAIN_BUDGET_MS = Number.isFinite(Number(process.env.HARPER_TEST_RECORD_LOCK_STAGE_DRAIN_MS))
+	? Number(process.env.HARPER_TEST_RECORD_LOCK_STAGE_DRAIN_MS)
+	: 10_000;
 
 server.registerOperation?.({
 	name: 'record_lock_propose_homes',

replication/recordLockRpc.ts

@@ -27,6 +27,7 @@ import * as logger from '../core/utility/logging/harper_logger.js';
 import {
 	deliverDelegationRecall,
 	deliverDelegationRequest,
+	quiesceDelegations,
 	writeLockBarrier,
 	type DelegationRecall,
 	type DelegationReply,
@@ -215,7 +216,7 @@ async function executeBarrier(request: any): Promise<{ position: number }> {
 
 // ---- relay through the main thread to the owner worker -----------------------------------------
 
-type RelayKind = 'delegate' | 'recall';
+type RelayKind = 'delegate' | 'recall' | 'quiesce';
 let nextRelayId = 1;
 const pendingRelays = new Map<number, (reply: any) => void>();
 
@@ -248,10 +249,28 @@ async function relay(kind: RelayKind, database: string, table: string, payload:
 
 async function executeLocally(kind: RelayKind, database: string, table: string, payload: any): Promise<any> {
 	if (kind === 'delegate') return deliverDelegationRequest(database, table, payload);
+	if (kind === 'quiesce') return quiesceDelegations(database, payload?.deadlineMs);
 	await deliverDelegationRecall(database, table, payload);
 	return RECALLED;
 }
 
+/**
+ * Drain this database's delegations on the thread that actually coordinates them (harper-pro#856).
+ *
+ * The operations API answers on whichever thread took the request, and coordinator state is
+ * per-thread: calling `quiesceDelegations` here would sweep an empty registry and report "nothing
+ * outstanding" while the owner thread still holds every grant — a drain claimed but not performed,
+ * which is the one thing this must never do. A relay that cannot reach the owner reports that as an
+ * error rather than an empty result, so the operator falls back to the drain interval.
+ */
+export async function quiesceOnOwner(database: string, deadlineMs: number): Promise<any> {
+	if (ownership.ownsDatabase(database)) return quiesceDelegations(database, deadlineMs);
+	const answer = await relay('quiesce', database, '', { deadlineMs });
+	if (!answer || typeof answer !== 'object' || !Array.isArray(answer.outstanding))
+		throw new ClientError('the record lock drain did not reach the coordinating worker in time', 503);
+	return answer;
+}
+
 /**
  * Main thread: answer locally when main owns the database, otherwise forward to the owner worker.
  * The forward hop carries an id minted HERE — worker-minted ids are per worker and would collide in

replication/recordLockTransport.ts

@@ -55,13 +55,15 @@ import { CLUSTER_RECORD_LOCKS_ENABLED } from './recordLockConfig.ts';
 import {
 	currentRow,
 	onRecordLockHomesChanged,
+	setHomesDrainReader,
 	setHomesMembershipReaders,
 	type RecordLockGenerationState,
 } from './recordLockHomes.ts';
 import {
 	BARRIER_OPERATION,
 	DELEGATE_OPERATION,
 	RECALL_OPERATION,
+	quiesceOnOwner,
 	sendRecordLockOperation,
 	setRecordLockOwnershipReaders,
 } from './recordLockRpc.ts';
@@ -1043,6 +1045,8 @@ export async function collectRecordLockStatus(
 	return result;
 }
 
+setHomesDrainReader((database, deadlineMs) => quiesceOnOwner(database, deadlineMs));
+
 setHomesMembershipReaders({
 	thisNodeName: getThisNodeName,
 	replicatingPeers(database) {

unitTests/replication/recordLockHomes.test.mjs

@@ -6,11 +6,15 @@
  */
 import assert from 'node:assert';
 import {
+	STAGE_DRAIN_BUDGET_MS,
 	canonicalizeHomes,
+	provesQuiescence,
+	setHomesDrainReader,
 	digestOf,
 	planActivate,
 	planProposal,
 	planStage,
+	stageGeneration,
 	validateGenerationInput,
 } from '#src/replication/recordLockHomes';
 
@@ -314,3 +318,72 @@ describe('planProposal', () => {
 		assert.throws(() => planProposal('a', many, undefined, 'data'), /homes must be/);
 	});
 });
+
+describe('stage drains instead of waiting out the lease (harper-pro#856)', () => {
+	afterEach(() => setHomesDrainReader(async () => ({ error: 'no record lock drain is wired on this node' })));
+
+	it('reports whatever the coordinating thread found, unchanged', async () => {
+		const drained = { surrendered: 2, recalled: 3, outstanding: [] };
+		let asked;
+		setHomesDrainReader(async (database, deadlineMs) => {
+			asked = { database, deadlineMs };
+			return drained;
+		});
+		const { quiesced } = await stageGeneration({
+			database: 'drain1',
+			generation: 1,
+			homes: ['a'],
+			hdb_user: { name: 'op' },
+		});
+		assert.deepStrictEqual(quiesced, drained);
+		assert.deepStrictEqual(asked, { database: 'drain1', deadlineMs: STAGE_DRAIN_BUDGET_MS });
+	});
+
+	it('a drain that throws becomes {error} and never fails a stage that already landed', async () => {
+		setHomesDrainReader(async () => {
+			throw new Error('the drain did not reach the coordinating worker');
+		});
+		const result = await stageGeneration({ database: 'drain2', generation: 1, homes: ['a'], hdb_user: { name: 'op' } });
+		assert.strictEqual(result.staged.generation, 1, 'the stage itself still succeeded');
+		assert.match(result.quiesced.error, /did not reach the coordinating worker/);
+	});
+
+	it('defaults to refusing rather than reporting a clean drain it never performed', async () => {
+		const { quiesced } = await stageGeneration({
+			database: 'drain3',
+			generation: 1,
+			homes: ['a'],
+			hdb_user: { name: 'op' },
+		});
+		assert.ok(quiesced.error, 'an unwired drain must not read as an empty outstanding list');
+		assert.strictEqual(quiesced.outstanding, undefined);
+	});
+
+	it('only an empty AND complete drain lets an orchestrator skip the interval', () => {
+		assert.strictEqual(provesQuiescence({ complete: true, surrendered: 0, recalled: 0, outstanding: [] }), true);
+		// Empty but not complete: a lazily-unbuilt coordinator, or a process that restarted recently.
+		assert.strictEqual(provesQuiescence({ complete: false, surrendered: 0, recalled: 0, outstanding: [] }), false);
+		assert.strictEqual(
+			provesQuiescence({ complete: true, surrendered: 0, recalled: 0, outstanding: [{ table: 'T' }] }),
+			false
+		);
+		assert.strictEqual(provesQuiescence({ error: 'the drain did not reach the coordinating worker' }), false);
+		assert.strictEqual(provesQuiescence(undefined), false);
+		// A malformed reply crossing a worker boundary must read as "not proven", not slip through on a
+		// missing `length`.
+		assert.strictEqual(provesQuiescence({ complete: true, outstanding: {} }), false);
+		assert.strictEqual(provesQuiescence({ complete: true }), false);
+		assert.strictEqual(provesQuiescence({ complete: 'yes', outstanding: [] }), false);
+	});
+
+	it('has a reporting budget far below the interval it replaces', () => {
+		// The budget bounds how long `stage` spends DRAINING before it reports what is left; it is not a
+		// safety interval. It must be well under DELEGATION_DRAIN_MS, or reporting would cost as much as
+		// the wait the drain exists to avoid.
+		assert.ok(STAGE_DRAIN_BUDGET_MS > 0);
+		assert.ok(
+			STAGE_DRAIN_BUDGET_MS < 365_000,
+			`the drain budget (${STAGE_DRAIN_BUDGET_MS}ms) must be shorter than the drain interval it replaces`
+		);
+	});
+});