Stop reads from extending the open-transaction limit while uncommitted writes are held

resources/DatabaseTransaction.ts

@@ -248,6 +248,12 @@ export class DatabaseTransaction implements Transaction {
 	readTxnRefCount: number;
 	readTxnsUsed: number;
 	timeout: number;
+	// Write recency, tracked separately from `timeout`: set only by addWrite, never by a read. `timeout`
+	// is re-armed by reads too (on a link with no pending writes of its own, see the fast path in
+	// getReadTxn), so chainStillActive can't use it to mean "this link was written recently" — a `.next`
+	// link with no writes that is being read in a loop would otherwise masquerade as write activity and
+	// keep a write-holding head immortal.
+	declare writeTimeout: number;
 	validated = 0;
 	timestamp = 0;
 	retries = 0;
@@ -281,7 +287,18 @@ export class DatabaseTransaction implements Transaction {
 
 	getReadTxn(disableSnapshot?: boolean): ReadTransaction {
 		this.readTxnRefCount = (this.readTxnRefCount || 0) + 1;
-		this.timeout = txnExpiration; // reset the timeout
+		// The limit is an IDLE limit. Writes always re-arm it (see addWrite), but reads only do so
+		// while no uncommitted writes are held: staged writes hold write intents that other writers'
+		// coordinated-retry commits park on, so a handler that wrote once and then only reads — an
+		// orphaned long-poll whose client had already gone, in harper#2001 — must not keep those
+		// intents alive by reading. A transaction that keeps writing stays alive; so does a purely
+		// read-only one. A committed transaction re-arms too: its intents went with the commit, and
+		// the monitor bounds its retained read snapshot separately.
+		// `writes`/`next` are checked inline first: this is a hot path and the dominant case is a
+		// single-store transaction that has never written.
+		if ((this.writes.length === 0 && !this.next) || this.open !== TRANSACTION_STATE.OPEN || !this.hasPendingWrites()) {
+			this.timeout = txnExpiration;
+		}
 		if (this.transaction) {
 			if ((this.transaction as any).openTimer) (this.transaction as any).openTimer = 0;
 			return this.transaction;
@@ -410,6 +427,13 @@ export class DatabaseTransaction implements Transaction {
 
 	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
+		// performs no longer do (see getReadTxn), so a transaction that keeps writing stays alive
+		// and only an idle one holding write intents is reaped.
+		this.timeout = txnExpiration;
+		// Independent write-recency signal for chainStillActive (see the field comment) — reads never
+		// touch this, only writes do.
+		this.writeTimeout = txnExpiration;
 		this.linkWrite(operation);
 		this.writes.push(operation);
 		if (!operation.deferSave) {
@@ -973,9 +997,31 @@ export class ImmediateTransaction extends DatabaseTransaction {
 
 let timer;
 
+/**
+ * True when a link other than `txn` in the same multi-store chain was written recently enough to
+ * still be active — i.e. its `writeTimeout` (set only by addWrite, see the field comment) hasn't
+ * decayed to zero. Writes re-arm only the link that receives them, so a chain writing database B
+ * while its head only reads A would otherwise be aborted by the head's own decay.
+ */
+function chainStillActive(txn: DatabaseTransaction): boolean {
+	for (let link: DatabaseTransaction = txn.next; link; link = link.next) {
+		// A write-only link (e.g. a blind write to a second database, never itself read) never calls
+		// getReadTxn, so it's never added to trackedTxns and the main loop below never decays it.
+		// Decay it here instead, so an idle write-only link eventually expires rather than keeping the
+		// whole chain immortal (harper#2001's blind-write shape).
+		if (!trackedTxns.has(link) && link.writeTimeout > 0) link.writeTimeout -= txnExpiration;
+		if (link.writeTimeout > 0) return true;
+	}
+	return false;
+}
+
 function startMonitoringTxns() {
 	timer = setInterval(function () {
 		for (const txn of trackedTxns) {
+			// Decay write recency once per tick for every tracked link, independent of the `timeout`
+			// branches below — a tracked link that keeps its own idle limit alive by reading must not
+			// thereby keep chainStillActive believing it was written recently too.
+			if (txn.writeTimeout > 0) txn.writeTimeout -= txnExpiration;
 			if (txn.timeout <= 0) {
 				const url = (txn.getContext() as any)?.url;
 				if (txn.open === TRANSACTION_STATE.CLOSED) {
@@ -991,6 +1037,12 @@ function startMonitoringTxns() {
 						}`
 					);
 					txn.releaseReadTxn();
+				} else if (txn.hasPendingWrites() && chainStillActive(txn)) {
+					// A later link in the chain was written recently (writes re-arm only the link that
+					// receives them, and a multi-store transaction can be writing database B while this
+					// head only reads A). The logical transaction is still active, so re-arm this link
+					// rather than aborting the whole chain out from under it.
+					txn.timeout = txnExpiration;
 				} else if (txn.hasPendingWrites() && !txn.sourceApply && !txn.isReplay) {
 					// Abort and surface an error rather than force-committing a partial write set: silently
 					// committing on the application's behalf breaks atomicity and can leave orphaned

unitTests/resources/txn-tracking.test.js

@@ -1,14 +1,15 @@
 require('../testUtils');
 const assert = require('assert');
 const { setupTestDBPath } = require('../testUtils');
-const { setTxnExpiration } = require('#src/resources/DatabaseTransaction');
+const { setTxnExpiration, DatabaseTransaction, TRANSACTION_STATE } = require('#src/resources/DatabaseTransaction');
 const { setTxnExpiration: setLMDBTxnExpiration } = require('#src/resources/LMDBTransaction');
 const { setReadTxnExpiration, checkReadTxnTimeouts } = require('#src/resources/RecordEncoder');
 const { setMainIsWorker } = require('#js/server/threads/manageThreads');
 const { table } = require('#src/resources/databases');
 const { transaction } = require('#src/resources/transaction');
 const { setTimeout: delay } = require('node:timers/promises');
 const { RocksDatabase } = require('@harperfast/rocksdb-js');
+const isLMDB = process.env.HARPER_STORAGE_ENGINE === 'lmdb';
 describe('Txn Expiration', () => {
 	let SlowResource,
 		performedDBInteractions = false;
@@ -84,8 +85,11 @@ describe('Write txn timeout', () => {
 				{ name: 't', indexed: true },
 			],
 		});
-		// a second table so a single transaction can span two databases (writes to the second live on the
-		// transaction's `next` chain), exercising the multi-store classification path
+		// A second table for the multi-store classification path. Note it shares `test` with
+		// IndexedResource, and the `next` chain is per-DATABASE, so both tables resolve to the same
+		// transaction link — a `next` link does not actually form here (a second `database:` in this
+		// suite resolves to the same store), which is why the chain-walk test below builds its links
+		// directly instead of going through the resource API.
 		OtherResource = table({
 			table: 'OtherTxnTable',
 			database: 'test',
@@ -125,6 +129,172 @@ describe('Write txn timeout', () => {
 		}
 	});
 
+	// A handler that keeps reading must not extend the limit once it is holding uncommitted writes:
+	// those hold write intents other writers' commits park on (harper#2001). The read-only arm below
+	// pins the other half — reads alone still re-arm. RocksDB-only: LMDBTransaction.getReadTxn()
+	// re-arms unconditionally, and that engine has no verification-table park to wedge.
+	it('does not let reads extend the limit for a txn holding uncommitted writes', async function () {
+		if (isLMDB) this.skip();
+		setExpiration(20);
+		try {
+			const context = {};
+			await assert.rejects(
+				transaction(context, async () => {
+					await IndexedResource.put(401, { t: 4001 }, context);
+					// Read repeatedly, well past the limit: pre-fix each read reset the clock and the
+					// monitor never fired.
+					for (let i = 0; i < 15; i++) {
+						await IndexedResource.get(401, context);
+						await delay(15);
+					}
+				}),
+				/open-transaction time/
+			);
+			assert.ok((await IndexedResource.get(401)) == null, 'the aborted write must not be committed');
+		} finally {
+			setExpiration(30000);
+		}
+	});
+
+	// The limit is an IDLE limit, so work in progress must not be killed: a transaction that keeps
+	// writing stays alive indefinitely, and only goes over when it stops. This is the counterpart to
+	// the arm above — reads don't extend a write-holding transaction, but writes do.
+	it('lets continued writes extend the limit well past it', async function () {
+		if (isLMDB) this.skip();
+		setExpiration(20);
+		try {
+			const context = {};
+			// Same total duration and cadence as the read-loop arm above, writing instead of reading.
+			await transaction(context, async () => {
+				for (let i = 0; i < 15; i++) {
+					await IndexedResource.put(500 + i, { t: 5000 + i }, context);
+					await delay(15);
+				}
+			});
+			assert.ok(await IndexedResource.get(514), 'a continuously-writing transaction must commit normally');
+		} finally {
+			setExpiration(30000);
+		}
+	});
+
+	// Direct-construction unit check of the chain walk: a head that holds NO writes of its own but
+	// whose `next` link does must not re-arm on a read. Driving this through the resource API is not
+	// reliable here — a second `database:` in this suite resolves to the same store, so no `next`
+	// link forms — so the links are built directly, as sourceApplyConflictRetry.test.js does.
+	it('does not re-arm the head on a read when the next chain holds the writes', function () {
+		if (isLMDB) this.skip();
+		const head = new DatabaseTransaction();
+		const next = new DatabaseTransaction();
+		head.next = next;
+		head.open = TRANSACTION_STATE.OPEN;
+		next.open = TRANSACTION_STATE.OPEN;
+		head.writes = [];
+		next.writes = [{ key: 'pending' }];
+		head.transaction = {}; // stand-in native read txn so getReadTxn returns before allocating one
+		assert.ok(head.hasPendingWrites(), "test setup: the head must see the next chain's write");
+
+		head.timeout = 5;
+		head.getReadTxn();
+		assert.equal(head.timeout, 5, 'a read must not re-arm a head whose next chain holds writes');
+
+		// Control: with the chain drained the same read re-arms normally.
+		next.writes = [];
+		head.timeout = 5;
+		head.getReadTxn();
+		assert.ok(head.timeout > 5, 'a read must still re-arm once no link holds writes');
+	});
+
+	// chainStillActive must tell "written recently" apart from "read recently": a next link with no
+	// writes of its own re-arms its own `timeout` on every read (the fast path above), so using that
+	// same field to decide the chain is write-active would let unrelated reads on the next link keep
+	// a write-holding head immortal — the harper#2001 shape, shifted onto a second store.
+	it('does not treat repeated reads on a write-free next link as write activity (chainStillActive)', async function () {
+		if (isLMDB) this.skip();
+		const trackedTxns = setExpiration(20);
+		try {
+			const head = new DatabaseTransaction();
+			head.open = TRANSACTION_STATE.OPEN;
+			head.writes = [{ key: 'pending' }]; // head itself holds the write
+			head.transaction = { abort() {} }; // stand-in native handle, as the chain-walk test above uses
+			head.readTxnsUsed = 1; // as a real getReadTxn() would leave behind
+			trackedTxns.add(head); // ...and as a real getReadTxn() would track it
+			head.timeout = 20;
+
+			const next = new DatabaseTransaction();
+			head.next = next;
+			next.open = TRANSACTION_STATE.OPEN;
+			next.transaction = { abort() {} };
+
+			// Read next repeatedly, well past the limit: each read re-arms next.timeout via the fast
+			// path, but must never touch writeTimeout — the signal chainStillActive actually consults.
+			for (let i = 0; i < 8; i++) {
+				next.getReadTxn();
+				assert.ok(next.timeout > 0, "test setup: next's own idle timeout does re-arm on reads");
+				await delay(15);
+			}
+			assert.ok(!next.writeTimeout, 'reads on a write-free link must never set writeTimeout');
+			assert.ok(
+				!trackedTxns.has(head),
+				'the head must not be kept immortal by unrelated reads on a write-free next link'
+			);
+		} finally {
+			setExpiration(30000);
+		}
+	});
+
+	// The other half of the same fix: a next link that receives a write but is never itself read (a
+	// blind write to a second database) is never added to trackedTxns, so nothing else decays it.
+	// Pre-fix, chainStillActive treated its permanently-armed timeout as ongoing write activity and
+	// kept the head immortal forever — a regression inside this PR's own target scenario.
+	it('reaps an idle chain whose next link received a write but was never itself read (chainStillActive decay)', async function () {
+		if (isLMDB) this.skip();
+		const trackedTxns = setExpiration(20);
+		try {
+			const head = new DatabaseTransaction();
+			head.open = TRANSACTION_STATE.OPEN;
+			head.transaction = { abort() {} };
+			head.readTxnsUsed = 1;
+			trackedTxns.add(head);
+			head.timeout = 20;
+
+			const next = new DatabaseTransaction();
+			head.next = next;
+			next.open = TRANSACTION_STATE.OPEN;
+			next.writes = [{ key: 'pending' }]; // as addWrite would leave staged
+			next.writeTimeout = 20; // as addWrite would set — but next.getReadTxn() is never called
+
+			assert.ok(!trackedTxns.has(next), 'test setup: next must not be tracked — it is never itself read');
+			assert.ok(head.hasPendingWrites(), "test setup: head must see the next chain's write");
+
+			await delay(150); // several monitor cycles with nothing touching either link
+			assert.ok(
+				!trackedTxns.has(head),
+				'an idle chain whose only write lives on an untracked next link must eventually be reaped'
+			);
+		} finally {
+			setExpiration(30000);
+		}
+	});
+
+	it('still lets reads extend the limit for a read-only txn', async function () {
+		if (isLMDB) this.skip();
+		await IndexedResource.put(402, { t: 4002 });
+		setExpiration(20);
+		try {
+			const context = {};
+			// Same duration and read cadence as the arm above, without a write: the transaction holds
+			// no write intents, so continued reads legitimately keep it alive.
+			await transaction(context, async () => {
+				for (let i = 0; i < 15; i++) {
+					assert.ok(await IndexedResource.get(402, context));
+					await delay(15);
+				}
+			});
+		} finally {
+			setExpiration(30000);
+		}
+	});
+
 	// Multi-store path: a transaction that reads one database and writes another holds the write on its
 	// `next` chain while the head (which only read) has no writes of its own. The head must still be treated
 	// as write-bearing and aborted, or the monitor would force-commit the second database's write (#1407).