Settle cache-fill record locks and waiting requests when a database closes mid-fill

DESIGN.md

@@ -34,6 +34,7 @@ A note lives in the `DESIGN.md` of the directory that owns the code it describes
 - [Non-`initStore` structure saves read inside their own transaction (`PrimaryRocksDatabase`/`RecordEncoder`)](resources/DESIGN.md#non-initstore-structure-saves-read-inside-their-own-transaction-primaryrocksdatabaserecordencoder) — A non-`initStore` RocksDB store reads `sharedStructuresKey` inside its `saveStructures` transaction so concurrent encoders conflict-check against the read snapshot; `isRocksDB` stays unset on `__dbis__` so `0x42` records are not stripped as version headers.
 - [getFromSource() timing: promise resolves before commit runs](resources/DESIGN.md#getfromsource-timing-promise-resolves-before-commit-runs) — Stamp timestamps before resolving; the commit never reassigns the record and decides residency before indexing · [getFromSource() keeps source versions separate from fill ordering](resources/DESIGN.md#getfromsource-keeps-source-versions-separate-from-fill-ordering) — A reserved fallback timestamp orders competing first fills; the source `lastModified`, capped at local time, is the version.
 - [getFromSource() waiters share a fill that stored nothing](resources/DESIGN.md#getfromsource-waiters-share-a-fill-that-stored-nothing) — A same-thread waiter serves the leader's committed, successful fill when the store holds nothing valid; failures are not shared.
+- [getFromSource() record locks outlive neither their handle nor a close](resources/DESIGN.md#getfromsource-record-locks-outlive-neither-their-handle-nor-a-close-tablets) — Closing the owning handle releases a fill's lock, so a late unlock is a no-op; waiters are only woken, by the lock callback or the timeout.
 - [Blob orphan cleanup: pre-saved files outlive cancelled commits](resources/DESIGN.md#blob-orphan-cleanup-pre-saved-files-outlive-cancelled-commits) — Pre-commit blob files are deleted by the write that skipped or aborted; a deleted blob instance is tombstoned by its `fileId`.
 - [Blob compression: policy at `saveBlob`, completeness via the writer lock, codec is per-file truth](resources/DESIGN.md#blob-compression-policy-at-saveblob-completeness-via-the-writer-lock-codec-is-per-file-truth) — Compression policy resolves in `saveBlob`, completeness comes from the writer lock, and the per-file codec is the truth on read.
 - [Over-time transactions are aborted, not force-committed (`DatabaseTransaction`/`LMDBTransaction`)](resources/DESIGN.md#over-time-transactions-are-aborted-not-force-committed-databasetransactionlmdbtransaction) — Write-bearing request transactions past the open-time limit are aborted and poisoned; one parked in its commit phase is spared; `timeoutBudget` extends a known-long write.

resources/DESIGN.md

@@ -353,11 +353,7 @@ the record it replaces (`recordUpdater` in `RecordEncoder.ts`), not just this so
 ordinary resequenced (out-of-order CRDT-merged) write gets it too. Those records remain ineligible
 for VerificationTable fast-path confirmation until a later write advances their version.
 
-The native layer reads the stored bytes itself: the big-endian version at offset 0, then a big-endian
-metadata word at offset 8 tagged `0x0E` (`ACTION_32_BIT`), with the flag at literal bit `0x10000`. The
-bit is persisted, so it cannot follow a renumbered rocksdb-js constant without a migration.
-`unitTests/resources/caching-rocks-database.test.js` ('marks a record whose version was reused…') pins
-the layout against raw stored values.
+The native layer reads the stored bytes itself: the big-endian version at offset 0, then a big-endian metadata word at offset 8 tagged `0x0E` (`ACTION_32_BIT`), with the flag at literal bit `0x10000`. The bit is persisted, so it cannot follow a renumbered rocksdb-js constant without a migration. `unitTests/resources/caching-rocks-database.test.js` ('marks a record whose version was reused…') pins the layout against raw stored values.
 
 A fill's puts carry no condition of their own: on LMDB the `LMDBTransaction` `ifVersion` block holds
 the condition with the primary, index and audit puts in one native write transaction (a per-put
@@ -371,6 +367,10 @@ invalidated record keeps its local time only while its audit entry exists to be
 (`recordUpdater`), since an appended entry is keyed by the last assigned time.
 `unitTests/resources/sourceFillConditionalWrite.test.js` pins both engines.
 
+## getFromSource() record locks outlive neither their handle nor a close (`Table.ts`)
+
+A fill holds a native record lock (`primaryStore.tryLock`) while it awaits the source, and a database close does not wait for it: the close drains only submitted commits. Closing the handle that owns the lock releases it and wakes its waiters: rocksdb-js `DBHandle::close` → `lockReleaseByOwner`, lmdb-js `EnvWrap::closeEnv` (an LMDB close closes the environment, never the dbis). So an unlock that throws on a closed store has nothing to release, and `releaseRecordLock` treats it as a no-op. Two consequences for waiters: the `tryLock` callback only wakes the request (it can fire during the close, where reading the store throws inside a native callback), and the `LOCK_TIMEOUT` timer wakes it too, because closing this thread's handle never notifies a wait on another thread's lock. `unitTests/resources/cacheFillClose.test.js` covers both engines.
+
 ## Blob orphan cleanup: pre-saved files outlive cancelled commits
 
 Blobs flagged with `saveBeforeCommit` (or `saveInRecord`) are written to disk in the `beforeIntermediate` phase of a `TransactionWrite`, _before_ the LMDB/RocksDB write commits. The write's commit callback can still skip the actual record write — for older versions, supersedence by future updates, residency mismatches, or full transaction abort. In every such path the file is on disk but no record references it.

resources/Table.ts

@@ -11870,17 +11870,28 @@ export function makeTable(options): TableResourceClass {
 		const sourceTimestamp =
 			inheritedTimestamp ||
 			(isRocksDB ? (primaryStore as RocksDatabase).getMonotonicTimestamp() : getNextMonotonicTime());
-		let whenResolved, timer;
+		let wake: () => void;
 		const fillKey = writeKeyId(id);
 		const heldFill = sourceFills?.get(fillKey);
 		// We start by locking the record so that there is only one resolution happening at once;
 		// if there is already a resolution in process, we want to use the results of that resolution
 		// tryLock() will return true if we got the lock, and the callback won't be called.
 		// If another thread has the lock it returns false and then the callback is called once
 		// the other thread releases the lock.
-		const callback = () => {
-			// This is called when another thread releases the lock on resolution. Hopefully
-			// it should be resolved now and we can use the value it saved.
+		// The callback only wakes this request: it can fire during a database close, where reading
+		// the store would throw inside a native callback and leave the request unsettled.
+		const lockAcquired = primaryStore.tryLock(id, () => wake());
+
+		if (!lockAcquired) {
+			let timer: ReturnType<typeof setTimeout>;
+			await new Promise<void>((resolve) => {
+				wake = resolve;
+				// closing this thread's handle does not notify a wait on another thread's lock
+				timer = setTimeout(() => {
+					releaseRecordLock(id);
+					resolve();
+				}, LOCK_TIMEOUT);
+			});
 			clearTimeout(timer);
 			const entry = primaryStore.getEntry(id);
 			if (
@@ -11896,33 +11907,23 @@ export function makeTable(options): TableResourceClass {
 					(outcome.value == null || (heldFill.sharable ??= isSharableRecord(outcome.value)))
 				) {
 					setLoadedFromSource(target, false);
-					whenResolved(outcome);
+					return outcome;
 				}
 				// try again — entry still not valid, need to actually fetch from source
-				else whenResolved(getFromSource(source, id, primaryStore.getEntry(id), context, target));
+				else return getFromSource(source, id, primaryStore.getEntry(id), context, target);
 			} else {
 				// served from cache after waiting for another request to resolve
 				setLoadedFromSource(target, false);
-				whenResolved(entry);
+				return entry;
 			}
-		};
-		const lockAcquired = primaryStore.tryLock(id, callback);
-
-		if (!lockAcquired) {
-			return new Promise((resolve) => {
-				whenResolved = resolve;
-				timer = setTimeout(() => {
-					primaryStore.unlock(id);
-				}, LOCK_TIMEOUT);
-			});
 		}
 		// lock acquired — this request will actually load from source
 		setLoadedFromSource(target, true);
 		const fill: SourceFill = { source };
 		(sourceFills ??= new Map()).set(fillKey, fill);
 		const releaseFill = () => {
 			if (sourceFills.get(fillKey) === fill) sourceFills.delete(fillKey);
-			primaryStore.unlock(id);
+			releaseRecordLock(id);
 		};
 
 		// it is important to remember that this is _NOT_ part of the current transaction; nothing is changing
@@ -12369,6 +12370,19 @@ export function makeTable(options): TableResourceClass {
 		});
 	}
 
+	/**
+	 * Closing the handle that owns a record lock releases it (the RocksDB handle, or the LMDB
+	 * environment), so an unlock that fails after close had nothing left to release.
+	 */
+	function releaseRecordLock(id: Id): void {
+		try {
+			primaryStore.unlock(id);
+		} catch (error) {
+			if (primaryStore.status === 'open' && tableRootStore.status === 'open')
+				logger.error?.(`Error releasing the cache resolution lock for ${id} in ${tableName}`, error);
+		}
+	}
+
 	function trackMaintenanceCommit<T>(commit: Promise<T>): Promise<T> {
 		const tracked = commit.finally(() => maintenanceCommits.delete(tracked));
 		maintenanceCommits.add(tracked);

unitTests/resources/cacheFillClose.test.js

@@ -0,0 +1,173 @@
+'use strict';
+
+// A worker restart closes the thread's databases while a cache fill may still be waiting on its
+// source. Closing the handle that owns a record lock releases it (rocksdb-js DBHandle::close ->
+// lockReleaseByOwner; lmdb-js EnvWrap::closeEnv), so the fill's later unlock must be a no-op rather
+// than a throw from its commit callback, and requests waiting on the lock must still settle.
+
+require('../testUtils');
+const assert = require('node:assert');
+const { setTimeout: delay } = require('node:timers/promises');
+const { setupTestDBPath } = require('../testUtils');
+const { waitFor } = require('../waitFor.js');
+const { table, getDatabases, closeDatabase } = require('#src/resources/databases');
+const { setMainIsWorker } = require('#js/server/threads/manageThreads');
+const { getRegistryStatus, RocksDatabase } = require('@harperfast/rocksdb-js');
+
+describe('Cache fill across database close', function () {
+	const isLmdb = process.env.HARPER_STORAGE_ENGINE === 'lmdb';
+	this.timeout(30_000);
+	let databaseIndex = 0;
+	let captured, savedRejectionListeners, savedExceptionListeners;
+	const captureRejection = (reason) => captured.push(reason);
+	const captureException = (error) => captured.push(error);
+
+	before(function () {
+		setupTestDBPath();
+		setMainIsWorker(true);
+	});
+
+	beforeEach(function () {
+		captured = [];
+		savedRejectionListeners = process.listeners('unhandledRejection');
+		savedExceptionListeners = process.listeners('uncaughtException');
+		process.removeAllListeners('unhandledRejection');
+		process.removeAllListeners('uncaughtException');
+		process.on('unhandledRejection', captureRejection);
+		process.on('uncaughtException', captureException);
+	});
+
+	afterEach(function () {
+		process.removeListener('unhandledRejection', captureRejection);
+		process.removeListener('uncaughtException', captureException);
+		for (const listener of savedRejectionListeners ?? []) process.on('unhandledRejection', listener);
+		for (const listener of savedExceptionListeners ?? []) process.on('uncaughtException', listener);
+	});
+
+	// On RocksDB a second handle on the same path stands in for another thread: it keeps the shared
+	// native descriptor (where record locks live) alive after this thread's handles close.
+	function setUpSlowSourceTable() {
+		const databaseName = `cache_fill_close_${databaseIndex++}`;
+		const SlowTable = table({
+			table: 'SlowSourced',
+			database: databaseName,
+			attributes: [{ name: 'id', isPrimaryKey: true }, { name: 'name' }],
+		});
+		getDatabases();
+		let releaseSource, failSource;
+		const sourceGate = new Promise((resolve, reject) => {
+			releaseSource = resolve;
+			failSource = reject;
+		});
+		let sourceCalls = 0;
+		SlowTable.sourcedFrom({
+			async get(id) {
+				sourceCalls++;
+				await sourceGate;
+				return { id, name: 'from source' };
+			},
+		});
+		const path = SlowTable.primaryStore.rootStore.path;
+		const otherThreadHandle = isLmdb ? undefined : new RocksDatabase(path).open();
+		return {
+			SlowTable,
+			databaseName,
+			releaseSource,
+			failSource,
+			otherThreadHandle,
+			sourceCalls: () => sourceCalls,
+			lockHeld: (id) => otherThreadHandle?.hasLock(id),
+			registry: () => getRegistryStatus().find((entry) => entry.path === path),
+		};
+	}
+
+	async function settleCallbacks() {
+		for (let i = 0; i < 5; i++) await delay(10);
+	}
+
+	function capturedMessages() {
+		return captured.map((error) => error?.message ?? String(error));
+	}
+
+	function assertRocksState(fixture, id, held) {
+		if (isLmdb) return;
+		assert.equal(fixture.lockHeld(id), held);
+		assert.equal(fixture.registry().transactions, 0, 'no transaction handle is left open');
+	}
+
+	for (const sourceOutcome of ['resolves', 'rejects']) {
+		it(`releases the record lock without throwing when the source ${sourceOutcome} after the database closes`, async function () {
+			const fixture = setUpSlowSourceTable();
+			const { SlowTable, databaseName, releaseSource, failSource, sourceCalls } = fixture;
+			try {
+				const fill = SlowTable.get('k1');
+				await waitFor(() => sourceCalls() > 0, { timeout: 5000, message: 'the source call did not start' });
+				if (!isLmdb) assert.equal(fixture.lockHeld('k1'), true, 'the fill holds its lock while the source is pending');
+
+				await closeDatabase(databaseName);
+				if (!isLmdb) assert.equal(fixture.lockHeld('k1'), false, 'closing the handle releases its locks');
+
+				if (sourceOutcome === 'resolves') {
+					releaseSource();
+					assert.equal((await fill)?.name, 'from source');
+				} else {
+					failSource(new Error('origin unavailable'));
+					await assert.rejects(fill, /origin unavailable/);
+				}
+				await settleCallbacks();
+
+				assert.deepEqual(capturedMessages(), []);
+				assertRocksState(fixture, 'k1', false);
+			} finally {
+				fixture.otherThreadHandle?.close();
+			}
+		});
+	}
+
+	it('settles a request waiting on the fill lock when the database closes', async function () {
+		const fixture = setUpSlowSourceTable();
+		const { SlowTable, databaseName, releaseSource, sourceCalls } = fixture;
+		try {
+			const fill = SlowTable.get('k2');
+			await waitFor(() => sourceCalls() > 0, { timeout: 5000, message: 'the source call did not start' });
+			const waiter = SlowTable.get('k2');
+			await delay(5);
+			assert.equal(sourceCalls(), 1);
+
+			await closeDatabase(databaseName);
+			await assert.rejects(waiter, /Database not open|closed database/);
+			releaseSource();
+			assert.equal((await fill)?.name, 'from source');
+			await settleCallbacks();
+
+			assert.deepEqual(capturedMessages(), []);
+			assertRocksState(fixture, 'k2', false);
+		} finally {
+			fixture.otherThreadHandle?.close();
+		}
+	});
+
+	it('settles a request waiting on another thread’s lock after its own database closes', async function () {
+		if (isLmdb) this.skip();
+		const { SlowTable, databaseName, otherThreadHandle, sourceCalls } = setUpSlowSourceTable();
+		try {
+			assert.equal(otherThreadHandle.tryLock('k3'), true);
+			const waiter = SlowTable.get('k3');
+			await delay(5);
+			assert.equal(sourceCalls(), 0, 'the request waits on the other handle’s lock');
+
+			// the other handle's lock survives this close, so only the wait timeout can settle the request
+			await closeDatabase(databaseName);
+			await assert.rejects(waiter, /Database not open/);
+			await settleCallbacks();
+
+			assert.deepEqual(capturedMessages(), []);
+			assert.equal(otherThreadHandle.hasLock('k3'), true, 'a closed handle cannot release another handle’s lock');
+			otherThreadHandle.unlock('k3');
+			await settleCallbacks();
+			assert.deepEqual(capturedMessages(), [], 'a late unlock notification is harmless');
+		} finally {
+			otherThreadHandle.close();
+		}
+	});
+});