Stop RocksDB eviction from deleting records renewed under the same version, or the blobs of records an abandoned batch keeps

resources/DESIGN.md

@@ -360,7 +360,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. Pass the write to `updateRecord` as its `write` option: `recordUpdater` records the entry it appends on `write.stagedAudit`, from which staging derives `appendedTo` and the confirmation checks read eligibility (below). 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 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`).
+**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`). A reused RocksDB version (`VERSION_REUSED`: a resequenced write, a 304 revalidation, a replica store) proves nothing about the value, so eviction also requires the metadata expiry its caller read — the one read-hiding uses — to still be the entry's own: `evict()`'s `observedExpiresAt` (passed by every deferred caller, checked before a TTL demotion; omitted, only the version is checked) and the batch item's scanned `expiresAt`. The `@expiresAt` sweep decides on the field but passes this metadata expiry too; validating the field is left to making that sweep run on RocksDB ([#1481](https://github.com/HarperFast/harper/issues/1481)). The batcher also re-checks the scan's `HAS_BLOBS | REPLICA` routing on the entry it stages: a same-version write can add a blob after the scan sent the record toward `removeEntry`.
 
 **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

@@ -998,8 +998,15 @@ interface TableResourceClass {
 	_storeReplica(existingEntry: Entry, entry: Entry & { originNode?: string }): Promise<boolean>;
 	/**
 	 * Evicting a record will remove it from a caching table. This is not considered a canonical data change, and it is assumed that retrieving this record from the source will still yield the same record, this is only removing the local copy of the record.
+	 * `observedExpiresAt` is the expiry of the entry the caller read (`null` when it had none); omitted, only the version is checked.
 	 */
-	evict(id: any, existingRecord: any, existingVersion: any, observedDeadline?: number): Promise<unknown>;
+	evict(
+		id: any,
+		existingRecord: any,
+		existingVersion: any,
+		observedDeadline?: number,
+		observedExpiresAt?: number | null
+	): Promise<unknown>;
 	/**
 	 * Static entry point: `Table.lock(id, options?, context?)` — creates an instance in the given,
 	 * ambient, or a fresh context and delegates to the instance lock(). This shadows Resource.static
@@ -4356,7 +4363,7 @@ export function makeTable(options): TableResourceClass {
 		/**
 		 * Evicting a record will remove it from a caching table. This is not considered a canonical data change, and it is assumed that retrieving this record from the source will still yield the same record, this is only removing the local copy of the record.
 		 */
-		static evict(id, existingRecord, existingVersion, observedDeadline?: number) {
+		static evict(id, existingRecord, existingVersion, observedDeadline?: number, observedExpiresAt?: number | null) {
 			if (maintenanceClosed || isReadOnlyMode()) return Promise.resolve();
 			let entry;
 			let removedRecordWithBlobs;
@@ -4376,6 +4383,13 @@ export function makeTable(options): TableResourceClass {
 					// if there is a resolution in-progress, abandon the eviction
 					if (primaryStore.hasLock(id, entry.version)) return;
 				}
+				// a replica sweep call is validated by its deadline in demoteReplica instead
+				if (
+					observedDeadline === undefined &&
+					observedExpiresAt !== undefined &&
+					reusedVersionChangedExpiry(entry, observedExpiresAt)
+				)
+					return;
 				if (observedDeadline !== undefined && !(entry?.metadataFlags & REPLICA)) return Promise.resolve();
 				if (isRocksDB && entry?.metadataFlags & REPLICA && entry.version === existingVersion) {
 					if (hasDerivedIndexRegistration(auditStore, tableId))
@@ -9970,7 +9984,7 @@ export function makeTable(options): TableResourceClass {
 				// nulling the shared cached entry would not hide the record from LMDB's weak reference
 				if (isReadOnlyMode()) return ABSENT_ENTRY;
 				// if it has expired and there is no source, we evict it and then return null, using a fake promise to indicate that this is providing the response
-				TableResource.evict(entry.key, entry.value, entry.version);
+				TableResource.evict(entry.key, entry.value, entry.version, undefined, entry.expiresAt);
 				entry.value = null;
 				return {
 					then(callback) {
@@ -10710,7 +10724,7 @@ export function makeTable(options): TableResourceClass {
 	// into a fresh transaction (dropping the now-changed record) and otherwise skip the batch, leaving
 	// those records for the next cleanup cycle.
 	function createEvictionBatcher() {
-		type EvictItem = { type: 'evict' | 'tombstone'; key: any; version: number };
+		type EvictItem = { type: 'evict' | 'tombstone'; key: any; version: number; expiresAt: number | undefined };
 		let pending: EvictItem[] = [];
 		const inFlight = new Set<Promise<void>>();
 
@@ -10722,10 +10736,13 @@ export function makeTable(options): TableResourceClass {
 			for (const item of items) {
 				const entry = primaryStore.getEntry(item.key, options);
 				if (!entry || entry.version !== item.version) continue; // gone or changed since the scan; leave for next cycle
+				if (reusedVersionChangedExpiry(entry, item.expiresAt)) continue;
 				if (item.type === 'tombstone') {
 					if (entry.value != null) continue; // resurrected since the scan
 				} else {
 					if (entry.value == null) continue; // already removed
+					// a same-version write can add these after the scan routed the record here
+					if (entry.metadataFlags & (HAS_BLOBS | REPLICA)) continue;
 					if (hasSourceGet && primaryStore.hasLock(item.key, entry.version)) continue; // resolution in progress
 					updateIndices(item.key, entry.value, null, options);
 					stageDerivedIndexEviction(transaction, item.key, entry.version);
@@ -10788,8 +10805,8 @@ export function makeTable(options): TableResourceClass {
 		}
 
 		return {
-			add(type: 'evict' | 'tombstone', key: any, version: number): Promise<void> | void {
-				pending.push({ type, key, version });
+			add(type: 'evict' | 'tombstone', key: any, version: number, expiresAt: number | undefined): Promise<void> | void {
+				pending.push({ type, key, version, expiresAt });
 				if (pending.length >= EVICTION_BATCH_SIZE) {
 					const items = pending;
 					pending = [];
@@ -10926,12 +10943,12 @@ export function makeTable(options): TableResourceClass {
 											// 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 | REPLICA))) {
-												await batcher.add(action, key, version);
+												await batcher.add(action, key, version, expiresAt);
 											} else {
 												const resolution =
 													action === 'tombstone'
 														? removeEntry(primaryStore, entry, version)
-														: TableResource.evict(key, record, version);
+														: TableResource.evict(key, record, version, undefined, expiresAt);
 												if (resolution) {
 													await outstandingCleanupOperations[cleanupIndex];
 													outstandingCleanupOperations[cleanupIndex] = resolution.catch((error) => {
@@ -11009,7 +11026,13 @@ export function makeTable(options): TableResourceClass {
 								if (backpressure) await backpressure;
 							} else if (recordEntry.value[expiresAtName] < Date.now()) {
 								// make sure the record hasn't changed and won't change while removing
-								const eviction = TableResource.evict(id, recordEntry.value, recordEntry.version);
+								const eviction = TableResource.evict(
+									id,
+									recordEntry.value,
+									recordEntry.version,
+									undefined,
+									recordEntry.expiresAt ?? null
+								);
 								if (eviction) {
 									const backpressure = trackInFlight(eviction);
 									if (backpressure) await backpressure;
@@ -11544,6 +11567,10 @@ export function makeTable(options): TableResourceClass {
 	}
 }
 
+// a reused version names more than one stored value (RecordEncoder.ts VERSION_REUSED)
+function reusedVersionChangedExpiry(entry: Entry, observedExpiresAt: number | null | undefined): boolean {
+	return Boolean(entry.metadataFlags & VERSION_REUSED) && (entry.expiresAt ?? null) !== (observedExpiresAt ?? null);
+}
 function attributesAsObject(attribute_permissions, type) {
 	const attrObject = attribute_permissions.attr_object || (attribute_permissions.attr_object = {});
 	let attrsForType = attrObject[type];

security/certificateVerification/verificationTables.ts

@@ -116,7 +116,7 @@ async function evictVerdictsWithoutExpiry(CertificateCache: any): Promise<void>
 		if (value == null || expiresAt != undefined) continue;
 		const slot = evictions++ % MAX_OUTSTANDING_EVICTIONS;
 		await outstanding[slot];
-		outstanding[slot] = CertificateCache.evict(key, value, version);
+		outstanding[slot] = CertificateCache.evict(key, value, version, undefined, null);
 		// yields between evictions, as the table's own cleanup scan does, so LMDB can renew its read transaction
 		await new Promise(setImmediate);
 	}