Deflake deferred cache and await LMDB reload markers

resources/DatabaseTransaction.ts

@@ -253,11 +253,12 @@ export type TransactionWrite = {
 	entry?: Partial<Entry>;
 	before?: () => void | Promise<void>;
 	beforeIntermediate?: () => void | Promise<void>;
-	commit?: (txnTime: number, existingEntry: Partial<Entry>, retry: boolean, transaction: any) => void;
+	commit?: (txnTime: number, existingEntry: Partial<Entry>, retry: boolean, transaction: any) => MaybePromise<void>;
 	validate?: (txnTime: number) => void;
 	fullUpdate?: boolean;
 	saved?: boolean;
 	deferSave?: boolean;
+	skipReplicationConfirmation?: boolean;
 	nodeName?: string;
 	nodeId?: number;
 	promise?: Promise<any>;
@@ -307,8 +308,8 @@ export function priorStagedWrite(operation: TransactionWrite): TransactionWrite
  * `[-0]` and `[null]` vs `[NaN]` are DIFFERENT stored keys that value-ish encodings (JSON, string
  * coercion) collapse, cross-contaminating unrelated records. So every key is mapped through the
  * same encoder the stores use; latin1 keeps the bytes injective in a string. Symbol keys (internal
- * metadata writes) can't be key-encoded and keep native identity; so does null (topic-less
- * publishes), which never stages a record.
+ * metadata writes) can't be key-encoded and keep native identity. Null is reserved for topic-less
+ * publishes and audit-only markers.
  */
 export function writeKeyId(key: Id): unknown {
 	if (typeof key === 'symbol' || key == null) return key;
@@ -574,6 +575,51 @@ export class DatabaseTransaction implements Transaction {
 		this.overloadChecked = true; // only check this once, don't interrupt ongoing transactions that have already made writes
 	}
 
+	/**
+	 * The stored entry of the last write eligible for replication confirmation. Two kinds of write are
+	 * skipped (rather than ending the search) so a trailing one cannot suppress confirmation for
+	 * replicable writes staged earlier: writes that explicitly opt out (audit-only markers, which stage
+	 * no record), and writes with no stored entry at all — a delete leaves a readable tombstone on any
+	 * audited or delete-tracking table, so only a delete on a table with neither is entry-less. Such a
+	 * `put(A); delete(B)` confirms on A's entry, whose version is this transaction's (every write in a
+	 * transaction is stamped with one version).
+	 */
+	lastConfirmableEntry(): Partial<Entry> | undefined {
+		for (let i = this.writes.length - 1; i >= 0; i--) {
+			const write = this.writes[i];
+			if (!write || write.skipReplicationConfirmation) continue;
+			const entry = write.store.getEntry(write.key);
+			if (entry) return entry;
+		}
+	}
+
+	/**
+	 * Stage an async operation that commit() must wait for. Staging can happen a turn or more before
+	 * commit() attaches its Promise.all, so the no-op rejection handler is attached here: without it a
+	 * rejection in that window is an unhandled rejection (fatal under --unhandled-rejections=strict).
+	 * The rejection still surfaces through commit()'s Promise.all.
+	 */
+	stageCompletion(completion: Promise<void>) {
+		completion.then(undefined, () => {});
+		this.completions.push(completion);
+	}
+
+	/**
+	 * Discard staged completions that no commit() will ever aggregate (abort path). Their rejections
+	 * are already no-op-handled by stageCompletion(), so without this they fail silently; log instead.
+	 * Clearing them also keeps a reused transaction's next commit() from rejecting with the previous
+	 * batch's error.
+	 */
+	drainCompletions(): void {
+		if (this.completions.length === 0) return;
+		const completions = this.completions;
+		this.completions = [];
+		for (const completion of completions)
+			completion.then(undefined, (error) =>
+				harperLogger.warn?.('A staged transaction completion failed after the transaction was aborted', error)
+			);
+	}
+
 	addWrite(operation: TransactionWrite) {
 		if (this.timedOut) throw transactionOpenTooLongError();
 		// A write is activity: it re-arms the idle limit on this link even though the reads it
@@ -640,11 +686,12 @@ export class DatabaseTransaction implements Transaction {
 				return;
 			}
 			let result: Promise<void> = operation.before?.() as Promise<void>;
-			if (result?.then) this.completions.push(result);
+			if (result?.then) this.stageCompletion(result);
 			result = operation.beforeIntermediate?.() as Promise<void>;
-			if (result?.then) this.completions.push(result);
+			if (result?.then) this.stageCompletion(result);
 		}
-		operation.commit(txnTime, operation.entry, this.retries > 0, transaction);
+		const completion = operation.commit(txnTime, operation.entry, this.retries > 0, transaction) as Promise<void>;
+		if (typeof completion?.then === 'function') this.stageCompletion(completion);
 		// Sticky record that THIS write staged with its audit entry appended (log entries batch on the
 		// native transaction and are durably written by its commit attempt — even a failed one — so
 		// they survive the abort-after-failed-commit of the retry paths). isRetry stagings
@@ -857,14 +904,10 @@ export class DatabaseTransaction implements Transaction {
 								// if we want to wait for replication confirmation, we need to track the transaction times
 								// and when replication notifications come in, we count the number of confirms until we reach the desired number
 								const databaseName = this.writes[0].store.rootStore.databaseName;
-								const lastWrite = this.writes[this.writes.length - 1];
-								if (confirmReplication && lastWrite) {
+								const lastEntry = this.lastConfirmableEntry();
+								if (confirmReplication && lastEntry) {
 									completions.push(
-										confirmReplication(
-											databaseName,
-											(lastWrite.store.getEntry(lastWrite.key) as any).version,
-											this.replicatedConfirmation
-										)
+										confirmReplication(databaseName, (lastEntry as any).version, this.replicatedConfirmation)
 									);
 								}
 							}
@@ -1005,7 +1048,13 @@ export class DatabaseTransaction implements Transaction {
 	}
 	abort(): void {
 		while (this.readTxnsUsed > 0) this.doneReadTxn(); // release the read snapshot when we abort, we assume we don't need it
+		// A write-only transaction never took a read reference (getReadTxn was never called), so the loop
+		// above releases nothing even though save() created a native handle; release it here instead of
+		// leaking the handle and its snapshot until GC. abortChainAfterRetries() detaches the handle
+		// before calling abort(), so this is a no-op there rather than a double-abort.
+		if (this.transaction) this.releaseReadTxn();
 		this.open = TRANSACTION_STATE.CLOSED;
+		this.drainCompletions();
 		for (const write of this.writes) {
 			if (write?.savedBlobs)
 				cleanupUnusedBlobs(write.savedBlobs, collectRetainedFileIds(write.store.getEntry(write.key)?.value));

resources/LMDBTransaction.ts

@@ -19,7 +19,6 @@ export const TRANSACTION_STATE = {
 	OPEN: 1, // the transaction is open and can be used for reads and writes
 	LINGERING: 2, // the transaction has completed a read, but can be used for immediate writes
 };
-let outstandingCommit;
 let confirmReplication;
 export function replicationConfirmation(callback) {
 	confirmReplication = callback;
@@ -184,10 +183,18 @@ export class LMDBTransaction extends DatabaseTransaction {
 		this.open = options?.doneWriting ? TRANSACTION_STATE.LINGERING : TRANSACTION_STATE.OPEN;
 		let resolution;
 		const completions = [];
+		let commitCompletions: Promise<void>[];
 		let writeIndex = 0;
 		this.writes = this.writes.filter((write) => write); // filter out removed entries
 		const doWrite = (write) => {
-			write.commit(txnTime, write.entry, retries);
+			const completion = write.commit(txnTime, write.entry, retries);
+			if (typeof completion?.then === 'function') {
+				// the aggregating Promise.all is attached a turn or more later (after the conditional batch
+				// or the exclusive transaction resolves), so handle rejection here to keep the gap from
+				// producing an unhandled rejection; it still surfaces through that Promise.all
+				completion.then(undefined, () => {});
+				(commitCompletions ??= []).push(completion);
+			}
 		};
 		// this uses optimistic locking to submit a transaction, conditioning each write on the expected version
 		const nextCondition = () => {
@@ -222,29 +229,36 @@ export class LMDBTransaction extends DatabaseTransaction {
 			// will fail and retry due to contention. This is used to determine when to give up on optimistic writes and
 			// use a real (async) transaction to get exclusive access to the data
 			if (db?.retryRisk) db.retryRisk *= 0.99; // gradually decay the retry risk
-			if (this.writes.length + (db?.retryRisk || 0) < MAX_OPTIMISTIC_SIZE >> retries) nextCondition();
-			else {
+			if (this.writes.length + (db?.retryRisk || 0) < MAX_OPTIMISTIC_SIZE >> retries) {
+				nextCondition();
+				if (commitCompletions) {
+					if (resolution) {
+						resolution = Promise.resolve(resolution).then((committed) =>
+							Promise.all(commitCompletions).then(() => committed)
+						);
+					} else {
+						// Must resolve truthy or the conflict branch re-runs the writes.
+						resolution = Promise.all(commitCompletions).then(() => true);
+					}
+				}
+			} else {
 				// if it is too big to expect optimistic writes to work, or we have done too many retries we use
 				// a real LMDB transaction to get exclusive access to reading and writing
-				resolution = this.writes[0].store.transaction(() => {
+				const transactionResolution = this.writes[0].store.transaction(() => {
 					for (const write of this.writes) {
 						// we load latest data while in the transaction
 						write.entry = write.store.getEntry(write.key);
 						doWrite(write);
 					}
 					return true; // success. always success
 				});
+				resolution = transactionResolution.then((committed) =>
+					commitCompletions ? Promise.all(commitCompletions).then(() => committed) : committed
+				);
 			}
 		}
 
 		if (resolution) {
-			if (!outstandingCommit) {
-				outstandingCommit = resolution;
-				outstandingCommit.then(() => {
-					outstandingCommit = null;
-				});
-			}
-
 			return resolution.then((resolution) => {
 				if (resolution) {
 					if (this.next) {
@@ -257,14 +271,10 @@ export class LMDBTransaction extends DatabaseTransaction {
 						// if we want to wait for replication confirmation, we need to track the transaction times
 						// and when replication notifications come in, we count the number of confirms until we reach the desired number
 						const databaseName = this.writes[0].store.rootStore.databaseName;
-						const lastWrite = this.writes[this.writes.length - 1];
-						if (confirmReplication && lastWrite)
+						const lastEntry = this.lastConfirmableEntry();
+						if (confirmReplication && lastEntry)
 							completions.push(
-								confirmReplication(
-									databaseName,
-									(lastWrite.store.getEntry(lastWrite.key) as any).localTime,
-									this.replicatedConfirmation
-								)
+								confirmReplication(databaseName, (lastEntry as any).localTime, this.replicatedConfirmation)
 							);
 					}
 					// commit succeeded; clean up files for any writes whose commit-handler took an early-return,
@@ -315,6 +325,7 @@ export class LMDBTransaction extends DatabaseTransaction {
 	abort(): void {
 		while (this.readTxnsUsed > 0) this.doneReadTxn(); // release the read snapshot when we abort, we assume we don't need it
 		this.open = TRANSACTION_STATE.CLOSED;
+		this.drainCompletions();
 		// any blobs that were pre-saved as part of these writes will never be referenced; schedule deletion
 		// (retaining any fileId the current on-disk record still references — an aborted write may carry an
 		// already-saved blob shared with the surviving record; see harper-pro#406).

resources/Table.ts

@@ -4511,7 +4511,7 @@ export function makeTable(options) {
 					logger.trace?.(`Publishing message to id: ${id}, timestamp: ${new Date(txnTime).toISOString()}`);
 					// always audit this, but don't change existing version
 					// TODO: Use direct writes in the future (copying binary data is hard because it invalidates the cache)
-					updateRecord(
+					return updateRecord(
 						id,
 						existingEntry?.value ?? null,
 						existingEntry,
@@ -4555,8 +4555,9 @@ export function makeTable(options) {
 				tableTxn.addWrite({
 					key: null,
 					store: primaryStore,
+					skipReplicationConfirmation: true,
 					commit: (txnTime: number, _existingEntry: any, _retry: any, transaction: any) => {
-						updateRecord(
+						return updateRecord(
 							null, // recordId: null — a whole-table signal, not a per-row change
 							undefined, // no record to store: this writes the audit entry only
 							undefined,

unitTests/apiTests/cache-test.mjs

@@ -3,7 +3,6 @@
 import { assert } from 'chai';
 import axios from 'axios';
 import { setupTestApp, baseUrl } from './setupTestApp.mjs';
-import { setTimeout as delay } from 'node:timers/promises';
 import { waitFor } from '../waitFor.js';
 
 describe('test REST calls with cache table', () => {
@@ -35,8 +34,13 @@ describe('test REST calls with cache table', () => {
 		delete data.ageInMonths; // harper#1484: computed scalars now also surface on default reads
 		response = await axios.put(`${baseUrl}/FourProp/3`, data);
 		assert.equal(response.status, 204);
-		await delay(20);
-		response = await axios(`${baseUrl}/SimpleCache/3`);
+		response = await waitFor(
+			async () => {
+				const cacheResponse = await axios(`${baseUrl}/SimpleCache/3`, { validateStatus: (status) => status < 500 });
+				return cacheResponse.status === 200 && cacheResponse.data?.name === 'name change' ? cacheResponse : false;
+			},
+			{ timeout: 5000, interval: 20, message: 'source update did not reach SimpleCache' }
+		);
 		assert.equal(response.status, 200);
 		assert.equal(response.data.id, 3);
 		assert.equal(response.data.name, 'name change');