cherry-pick: make read txn timeout configurable and set default to 5min (conflicts → v5.0)

resources/RecordEncoder.ts

@@ -6,6 +6,8 @@
  */
 
 import { Encoder } from 'msgpackr';
+import { get as envGet } from '../utility/environment/environmentManager.js';
+import { CONFIG_PARAMS } from '../utility/hdbTerms.ts';
 import {
 	HAS_PREVIOUS_RESIDENCY_ID,
 	HAS_CURRENT_RESIDENCY_ID,
@@ -483,6 +485,7 @@ export function handleLocalTimeForGets(store, rootStore) {
 				use.call(this);
 			};
 			Txn.prototype.done = function () {
+				if (this.isDone) return;
 				done.call(this);
 				this.openTimer = 0; // reset so idle pool time doesn't accumulate toward the stale-open threshold
 				if (this.isDone) {
@@ -501,16 +504,18 @@ export function handleLocalTimeForGets(store, rootStore) {
 	return store;
 }
 const trackedTxns: WeakRef<any>[] = [];
-setInterval(() => {
+const configValue = envGet(CONFIG_PARAMS.STORAGE_MAX_READ_TRANSACTION_OPEN_TIME) ?? 300000;
+let READ_TXN_TIMEOUT_TICKS = Math.round(configValue / 15000);
+export function checkReadTxnTimeouts() {
 	for (let i = 0; i < trackedTxns.length; i++) {
 		const txn = trackedTxns[i].deref();
 		if (!txn || txn.isDone || txn.isCommitted) trackedTxns.splice(i--, 1);
 		else if (txn.notCurrent) {
 			if (txn.openTimer) {
 				if (txn.openTimer > 3) {
-					if (txn.openTimer > 60) {
+					if (txn.openTimer > READ_TXN_TIMEOUT_TICKS) {
 						harperLogger.error(
-							'Read transaction detected that has been open too long (over 15 minutes), ending transaction',
+							`Read transaction detected that has been open too long (over ${Math.round(READ_TXN_TIMEOUT_TICKS * 15)} seconds), ending transaction`,
 							txn
 						);
 						trackedTxns.splice(i--, 1);
@@ -531,7 +536,12 @@ setInterval(() => {
 			} else txn.openTimer = 1;
 		}
 	}
-}, 15000).unref();
+}
+setInterval(checkReadTxnTimeouts, 15000).unref();
+export function setReadTxnExpiration(ms: number) {
+	READ_TXN_TIMEOUT_TICKS = Math.round(ms / 15000);
+	return trackedTxns;
+}
 export function setNextEncoding(timestamp: number, metadata: number, expiresAt = -1, nodeId = -1, residencyId = 0) {
 	timestampNextEncoding = timestamp;
 	metadataInNextEncoding = metadata;

unitTests/resources/txn-tracking.test.js

@@ -3,6 +3,7 @@ const assert = require('assert');
 const { setupTestDBPath } = require('../testUtils');
 const { setTxnExpiration } = 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 { setTimeout: delay } = require('node:timers/promises');
@@ -56,3 +57,65 @@ describe('Txn Expiration', () => {
 		setTxnExpiration(30000);
 	});
 });
+
+describe('Read Txn Expiration', () => {
+	let SlowReadResource;
+	before(async function () {
+		setupTestDBPath();
+		setMainIsWorker(true);
+		let BasicTable = table({
+			table: 'ReadTxnTable',
+			database: 'test',
+			attributes: [{ name: 'id', isPrimaryKey: true }, { name: 'name' }],
+		});
+		SlowReadResource = class extends BasicTable {
+			async get(query) {
+				const result = super.get(query);
+				await delay(50);
+				return result;
+			}
+		};
+		if (SlowReadResource.primaryStore instanceof RocksDatabase) this.skip();
+	});
+
+	it('Read txn will be ended after timeout', async function () {
+		await SlowReadResource.put(1, { name: 'one' });
+
+		// set timeout to minimum, 15s = 1 tick, openTimer > 1 means txn is expired
+		const trackedTxns = setReadTxnExpiration(15000);
+
+		const readPromise = SlowReadResource.get(1);
+		await delay(20);
+
+		const before = trackedTxns.length;
+		checkReadTxnTimeouts();
+		checkReadTxnTimeouts();
+		checkReadTxnTimeouts();
+		checkReadTxnTimeouts();
+		checkReadTxnTimeouts();
+
+		assert.ok(
+			trackedTxns.length < before,
+			`expected a txn to be removed; trackedTxns went ${before} -> ${trackedTxns.length}`
+		);
+		await readPromise;
+	});
+
+	it('Read txn below threshold is not expired', async function () {
+		setReadTxnExpiration(60000);
+
+		await SlowReadResource.put(2, { name: 'two' });
+		const readPromise = SlowReadResource.get(2);
+		await delay(20);
+
+		// only 2 ticks
+		checkReadTxnTimeouts();
+
+		const result = await readPromise;
+		assert.equal(result.name, 'two');
+	});
+
+	after(function () {
+		setReadTxnExpiration(300000);
+	});
+});

utility/hdbTerms.ts

@@ -560,6 +560,7 @@ export const CONFIG_PARAMS = {
 	STORAGE_ENCRYPTION: 'storage_encryption',
 	STORAGE_MAXTRANSACTIONQUEUETIME: 'storage_maxTransactionQueueTime',
 	STORAGE_MAXTRANSACTIONOPENTIME: 'storage_maxTransactionOpenTime',
+	STORAGE_MAX_READ_TRANSACTION_OPEN_TIME: 'storage_maxReadTransactionOpenTime',
 	STORAGE_DEBUGLONGTRANSACTIONS: 'storage_debugLongTransactions',
 	STORAGE_PATH: 'storage_path',
 	STORAGE_BLOBPATHS: 'storage_blobPaths',