diff --git a/resources/DESIGN.md b/resources/DESIGN.md index 16e5beb982..98cd6b8e0a 100644 --- a/resources/DESIGN.md +++ b/resources/DESIGN.md @@ -392,7 +392,7 @@ The mitigations live in three places: When adding a new commit-handler early-return path, reset `write.skipped = false` per round and set it to `true` only when the round stages neither a record nor an audit entry. If this round stages an audit-only entry via `auditRecordToStore`, leave `skipped` unset: `save()` uses it to record `appendedTo` and the replication-confirmation target. An earlier attempt's committed or durable audit entry can still own a skipped retry's blobs; RocksDB cleanup checks `blobsAuditReferenced` and the appending handle's durability. Unsubmitted staging owns no files. `cleanupOrphans` is the periodic safety net; don't rely on it for transactional correctness. -**A removed record's blob files go only once its removal commits.** `removeEntry` (`RecordEncoder.ts`) decides from what `store.remove` returns: a promise gates the unlink on its resolution, anything else is taken as already durable. A RocksDB removal staged in a transaction (`{ transaction }`) returns nothing, so through `removeEntry` its blobs are queued for reclamation at staging time, and an abort leaves the record pointing at files reclaimed `storage.blobRetention` later. `getFromSource()`'s source-miss removal therefore stages `primaryStore.remove(id, { transaction })` itself and deletes the removed record's blobs from the fill's commit completion, keeping any file the record now stored at that key references (queueing a file for reclamation clears its re-reference signal); `_writeDelete`, RocksDB `evict()` and `updateRecord`'s superseded-blob deletion still go through the staging-time path ([#2861](https://github.com/HarperFast/harper/issues/2861)). +**A removed record's blob files go only once its removal commits.** `removeEntry` (`RecordEncoder.ts`) decides from what `store.remove` returns: a promise gates the unlink on its resolution, anything else is taken as already durable. A RocksDB removal staged in a transaction (`{ transaction }`) returns nothing, so through `removeEntry` its blobs are queued for reclamation at staging time, and an abort leaves the record pointing at files reclaimed `storage.blobRetention` later. `getFromSource()`'s source-miss removal and RocksDB `evict()` therefore stage the removal themselves and delete the removed record's blobs from their own commit completion, keeping any file the record now stored at that key references (queueing a file for reclamation clears its re-reference signal), so an `ERR_BUSY`-abandoned eviction reclaims nothing; `_writeDelete` and `updateRecord`'s superseded-blob deletion still go through the staging-time path ([#2861](https://github.com/HarperFast/harper/issues/2861), [#3367](https://github.com/HarperFast/harper/issues/3367)). `evict()` reads the entry in its own transaction and skips a version other than the one its caller read on every table, not only audited or sourced ones: the read path and the cleanup scan's blob bypass hand it a record read earlier (`evictionBlobs.test.js`). **Source-unavailable blobs must not abort the commit.** `startPreCommitBlobsForRecord().complete()` awaits each blob's `saving` promise; a rejection there propagates up and aborts the record's apply (the replication subscription loop catches and logs it as `error in subscription handler`). For a blob the replication source can no longer provide — evicted/expired at the origin, the receiver having flagged the rejection `sourceBlobUnavailable` (harper-pro#403) — that abort permanently wedged a replication copy stream on an expiration cache table whose TTL-evicted blobs are gone everywhere: every orphaned record's apply re-threw, the copy never advanced, and backpressure pinned at ~100%. `complete()` therefore tolerates a `sourceBlobUnavailable` rejection (`isSourceBlobUnavailable`): the record commits with a diverged blob reference, left for proactive backfill (harper-pro#388). Local/transient save faults stay unmarked and still reject, so the write aborts and a reconnect retries it — no silent loss. This is the apply/commit-side complement to the replication receiver's resume-cursor advance (harper-pro#403/#405), which handles the durability-watermark side of the same missing blob. diff --git a/resources/Table.ts b/resources/Table.ts index 703b8486de..e17674294f 100644 --- a/resources/Table.ts +++ b/resources/Table.ts @@ -4073,6 +4073,7 @@ export function makeTable(options): TableResourceClass { static evict(id, existingRecord, existingVersion) { if (maintenanceClosed) return Promise.resolve(); let entry; + let removedRecordWithBlobs; // Raw on RocksDB: this method commits or aborts the handle itself, and its catch expects a conflict // to reject with ERR_BUSY (resources/DESIGN.md, over-time transactions). const lmdbTransaction = primaryStore.ifVersion @@ -4082,12 +4083,9 @@ export function makeTable(options): TableResourceClass { let options = { transaction }; let committed = false; try { - if (hasSourceGet || audit) { - if (!existingRecord) return; - entry = primaryStore.getEntry(id, options); - if (!entry || !existingRecord) return; - if (entry.version !== existingVersion) return; - } + if ((hasSourceGet || audit) && !existingRecord) return; + entry = primaryStore.getEntry(id, options); + if (!entry || entry.version !== existingVersion) return; if (hasSourceGet) { // if there is a resolution in-progress, abandon the eviction if (primaryStore.hasLock(id, entry.version)) return; @@ -4100,16 +4098,19 @@ export function makeTable(options): TableResourceClass { // Capture both promises so a real write failure on either resolves through evict()'s catch // below rather than escaping as an unhandled rejection from the fire-and-forget callers. const indexCleanup = primaryStore.ifVersion(id, existingVersion, () => { - updateIndices(id, existingRecord, null); + updateIndices(id, entry.value, null); }); - const removal = removeEntry(primaryStore, entry ?? primaryStore.getEntry(id), existingVersion); + const removal = removeEntry(primaryStore, entry, existingVersion); lmdbCompletion = Promise.all([indexCleanup, removal]); } else { if (hasDerivedIndexRegistration(auditStore, tableId)) reserveLocalKey(primaryStore.rootStore, transaction as RocksTransaction); - updateIndices(id, existingRecord, null, options); + // the entry read in this transaction, not the caller's copy: a resequenced write can reuse its version + updateIndices(id, entry.value, null, options); stageDerivedIndexEviction(transaction as RocksTransaction, id, existingVersion); - removeEntry(primaryStore, entry ?? primaryStore.getEntry(id), options); + // not removeEntry: it queues blob reclamation at staging, and this commit can still conflict + primaryStore.removeSync(id, options); + if (entry.value && entry.metadataFlags & HAS_BLOBS) removedRecordWithBlobs = entry.value; } committed = true; // Eviction is best-effort cleanup, run fire-and-forget from the record-expiration sweep and the @@ -4133,16 +4134,26 @@ export function makeTable(options): TableResourceClass { // RocksDB: eviction writes went directly into the raw transaction via options; commit it directly. // The raw commit has no retry, so a concurrent-write conflict rejects here — swallow it // (abandon the eviction) and log anything unexpected, rather than letting it crash the process. + const onCommitFailure = (error) => { + // The commit failed, so the read-snapshot/transaction handle is still open — release it, as the + // batched-eviction path does on its own commit failures. committed===true skips the finally abort. + try { + (transaction as any).abort(); + } catch {} + if (error?.code === 'ERR_BUSY') logger.trace?.('Abandoned eviction of busy record', id); + else logger.warn?.('Error evicting record', id, error); + }; + const commit = commitTrackedRocksTransaction(transaction as RocksTransaction, primaryStore); + if (!removedRecordWithBlobs) return trackMaintenanceCommit(commit.catch(onCommitFailure)); return trackMaintenanceCommit( - commitTrackedRocksTransaction(transaction as RocksTransaction, primaryStore).catch((error) => { - // The commit failed, so the read-snapshot/transaction handle is still open — release it, as the - // batched-eviction path does on its own commit failures. committed===true skips the finally abort. + commit.then(() => { try { - (transaction as any).abort(); - } catch {} - if (error?.code === 'ERR_BUSY') logger.trace?.('Abandoned eviction of busy record', id); - else logger.warn?.('Error evicting record', id, error); - }) + // a write committed since the eviction may reference the same files again + deleteBlobsInObject(removedRecordWithBlobs, collectRetainedFileIds(primaryStore.getEntry(id)?.value)); + } catch (error) { + logger.warn?.('Error deleting blobs of evicted record', id, error); + } + }, onCommitFailure) ); } finally { if (!committed) { @@ -10160,8 +10171,8 @@ export function makeTable(options): TableResourceClass { count++; } if (action) { - // Blob-bearing records delete their blob files as a non-transactional side effect, so - // they stay on the per-record evict() path that preserves the existing blob/commit ordering. + // Blob-bearing records stay on the per-record evict() path, which reclaims their blob + // files only after its own commit; the batcher's removeEntry would queue them at staging. if (batcher && !(action === 'evict' && metadataFlags & HAS_BLOBS)) { await batcher.add(action, key, version); } else { diff --git a/unitTests/resources/evictionBlobs.test.js b/unitTests/resources/evictionBlobs.test.js new file mode 100644 index 0000000000..534adac67d --- /dev/null +++ b/unitTests/resources/evictionBlobs.test.js @@ -0,0 +1,174 @@ +require('../testUtils'); +const assert = require('node:assert'); +const { existsSync } = require('node:fs'); +const { randomBytes } = require('node:crypto'); +const { setTimeout: delay } = require('node:timers/promises'); +const { setupTestDBPath } = require('../testUtils'); +const { waitFor } = require('../waitFor.js'); +const { table } = require('#src/resources/databases'); +const { + createBlob, + getFileId, + getFilePathForBlob, + getBlobHoldStateForTesting, + setDeletionDelay, +} = require('#src/resources/blob'); +const { setMainIsWorker } = require('#js/server/threads/manageThreads'); + +// Long enough for a zero-delay reclamation queued by a wrong eviction to have unlinked its file, unless +// entries left by earlier tests hold the shared queue's tail back. +const RECLAMATION_SETTLE_MS = 200; +const REREFERENCED = 1; // hold-state slot offset, resources/blob.ts + +describe('evict() leaves a renewed record and its blobs intact', function () { + this.timeout(10000); + let UnauditedBlobs; + + before(function () { + setupTestDBPath(); + setMainIsWorker(true); + setDeletionDelay(0); + UnauditedBlobs = table({ + table: 'EvictUnauditedBlobs', + database: 'test', + audit: false, + attributes: [ + { name: 'id', isPrimaryKey: true }, + { name: 'tag', indexed: true }, + { name: 'blob', type: 'Blob' }, + ], + }); + }); + + after(function () { + setDeletionDelay(undefined); + }); + + async function putWithBlob(id, tag, blob) { + blob ??= await createBlob(randomBytes(20000)); // over the file-storage threshold + await UnauditedBlobs.put({ id, tag, blob }); + const filePath = getFilePathForBlob(blob); + assert.ok(filePath && existsSync(filePath), 'precondition: the blob is file-backed'); + return { entry: UnauditedBlobs.primaryStore.getEntry(id), filePath }; + } + + it('skips a version renewed after the caller read it, without a source or audit log', async function () { + const read = await putWithBlob('renewed', 'v1'); + const renewed = await putWithBlob('renewed', 'v2'); + await UnauditedBlobs.evict('renewed', read.entry.value, read.entry.version); + await delay(RECLAMATION_SETTLE_MS); + + assert.equal(UnauditedBlobs.primaryStore.getEntry('renewed')?.value.tag, 'v2', 'the renewed record must survive'); + assert.equal(UnauditedBlobs.indices.tag.getValuesCount('v2'), 1, "the renewed record's index entry must survive"); + assert.ok(existsSync(renewed.filePath), "the renewed record's blob file must survive"); + + await UnauditedBlobs.evict('renewed', renewed.entry.value, renewed.entry.version); + assert.equal(UnauditedBlobs.primaryStore.getEntry('renewed'), undefined); + assert.equal(UnauditedBlobs.indices.tag.getValuesCount('v2'), 0); + await waitFor(() => !existsSync(renewed.filePath), { + message: 'an eviction of the live version reclaims its blob', + }); + await UnauditedBlobs.evict('renewed', renewed.entry.value, renewed.entry.version); + }); + + it('keeps the blob file when its eviction loses a commit conflict', async function () { + if (process.env.HARPER_STORAGE_ENGINE === 'lmdb') return this.skip(); // LMDB writes have no commit conflict + const { Transaction } = require('@harperfast/rocksdb-js'); + const originalCommit = Transaction.prototype.commit; + const { entry, filePath } = await putWithBlob('contended', 'resident'); + // A competing write keeps the same blob. It has encoded (re-referencing the file) and is committing + // when the eviction reads and stages; the eviction's commit then lands after it and conflicts. + let eviction; + let evictionCommit; + Transaction.prototype.commit = function (...args) { + let competingCommitted; + const competing = new Promise((resolve) => (competingCommitted = resolve)); + Transaction.prototype.commit = function (...evictionArgs) { + Transaction.prototype.commit = originalCommit; + evictionCommit = competing.then(() => originalCommit.apply(this, evictionArgs)); + return evictionCommit; + }; + eviction = UnauditedBlobs.evict('contended', entry.value, entry.version); + const committing = originalCommit.apply(this, args); + committing.then(competingCommitted, competingCommitted); + return committing; + }; + try { + await UnauditedBlobs.put({ ...entry.value, tag: 'competing' }); + assert.ok(eviction, 'the eviction must run inside the competing commit'); + await eviction; + } finally { + Transaction.prototype.commit = originalCommit; + } + await delay(RECLAMATION_SETTLE_MS); + + // Queue-independent: queueing a reclamation clears the re-reference the competing write recorded. + const { table: holdState, slot } = getBlobHoldStateForTesting( + UnauditedBlobs.primaryStore.rootStore, + getFileId(entry.value.blob) + ); + assert.equal( + Atomics.load(holdState, slot + REREFERENCED), + 1, + "an abandoned eviction must not queue the competing write's blob for reclamation" + ); + await assert.rejects( + evictionCommit, + { code: 'ERR_BUSY' }, + 'the eviction commit must conflict with the competing write' + ); + assert.equal(UnauditedBlobs.primaryStore.getEntry('contended')?.value.tag, 'competing'); + assert.ok( + existsSync(filePath), + 'an abandoned eviction must not reclaim the blob the competing write still references' + ); + }); + + it('keeps a blob file that a write committed after the eviction references again', async function () { + if (process.env.HARPER_STORAGE_ENGINE === 'lmdb') return this.skip(); // exercises the RocksDB post-commit step + const { Transaction } = require('@harperfast/rocksdb-js'); + const originalCommit = Transaction.prototype.commit; + const { entry, filePath } = await putWithBlob('refilled', 'resident'); + Transaction.prototype.commit = function (...args) { + Transaction.prototype.commit = originalCommit; + // lands between the eviction's commit and its blob cleanup + return originalCommit + .apply(this, args) + .then((result) => UnauditedBlobs.put({ ...entry.value, tag: 'refilled' }).then(() => result)); + }; + try { + await UnauditedBlobs.evict('refilled', entry.value, entry.version); + } finally { + Transaction.prototype.commit = originalCommit; + } + await delay(RECLAMATION_SETTLE_MS); + + const { table: holdState, slot } = getBlobHoldStateForTesting( + UnauditedBlobs.primaryStore.rootStore, + getFileId(entry.value.blob) + ); + assert.equal(UnauditedBlobs.primaryStore.getEntry('refilled')?.value.tag, 'refilled'); + assert.equal(Atomics.load(holdState, slot + REREFERENCED), 1, 'the refilled record keeps its blob file'); + assert.ok(existsSync(filePath)); + }); + + it('removes the index keys of the record it removes when a resequenced write reused the version', async function () { + if (process.env.HARPER_STORAGE_ENGINE === 'lmdb') return this.skip(); // only RocksDB resequences under a reused version + await UnauditedBlobs.put({ id: 'resequenced', tag: 'read' }); + const { version: baseVersion } = UnauditedBlobs.primaryStore.getEntry('resequenced'); + await UnauditedBlobs.patch('resequenced', { blob: null }, { timestamp: baseVersion + 100 }); + const read = UnauditedBlobs.primaryStore.getEntry('resequenced'); + await UnauditedBlobs.patch('resequenced', { tag: 'merged' }, { timestamp: baseVersion + 50 }); + const merged = UnauditedBlobs.primaryStore.getEntry('resequenced'); + assert.equal(merged.version, read.version, 'precondition: the resequenced write kept the version the caller read'); + assert.equal(merged.value.tag, 'merged'); + + await UnauditedBlobs.evict('resequenced', read.value, read.version); + const present = UnauditedBlobs.primaryStore.getEntry('resequenced') ? 1 : 0; + assert.equal( + UnauditedBlobs.indices.tag.getValuesCount('merged'), + present, + 'the index must match the stored record' + ); + }); +});