diff --git a/DESIGN.md b/DESIGN.md index 566a8acae2..e3b3768a69 100644 --- a/DESIGN.md +++ b/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. diff --git a/resources/DESIGN.md b/resources/DESIGN.md index b25fd9bc5d..93d7621bbb 100644 --- a/resources/DESIGN.md +++ b/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. diff --git a/resources/Table.ts b/resources/Table.ts index 3d69592fd8..9dcca4e527 100644 --- a/resources/Table.ts +++ b/resources/Table.ts @@ -11870,7 +11870,7 @@ 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; @@ -11878,9 +11878,20 @@ export function makeTable(options): TableResourceClass { // 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; + await new Promise((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,25 +11907,15 @@ 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); @@ -11922,7 +11923,7 @@ export function makeTable(options): TableResourceClass { (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(commit: Promise): Promise { const tracked = commit.finally(() => maintenanceCommits.delete(tracked)); maintenanceCommits.add(tracked); diff --git a/unitTests/resources/cacheFillClose.test.js b/unitTests/resources/cacheFillClose.test.js new file mode 100644 index 0000000000..0bd7087d90 --- /dev/null +++ b/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(); + } + }); +});