Stop RocksDB eviction from deleting renewed records or the blobs of records that survive

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.
 

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 {

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'
+		);
+	});
+});