feat(analytics): expose write/read transaction queue depth (count) metrics

resources/DatabaseTransaction.ts

@@ -26,6 +26,52 @@ const MAX_RETRIES = 40;
 // cap (see the commit rejection handler), don't grow the delay unbounded.
 const MAX_RETRY_DELAY_MS = 1000;
 let outstandingCommit, outstandingCommitStart;
+
+// Queue-depth gauges surfaced through the analytics pipeline (write-transaction-queue-depth /
+// read-transaction-queue-depth). Per-thread state; the analytics aggregator sums across threads.
+// `writeTxnQueueDepth` counts write commits handed to the storage engine but not yet resolved —
+// this is the backlog that, when it drains too slowly, produces the "Outstanding write transactions
+// have too long of queue" overload error. Read depth is derived from the live `trackedTxns` set
+// (every tracked transaction holds an open read snapshot). We also retain a high-water mark per
+// sampling window because the queue can fill and drain within a single (~1s) analytics period, so an
+// instantaneous sample taken at emit time would routinely miss the spike operators need to see.
+// RocksDB-write-path only: LMDB routes through the separate LMDBTransaction.commit()/getReadTxn()
+// overrides (resources/LMDBTransaction.ts), which maintain their own unrelated `trackedTxns` set and
+// do not call into this accounting.
+let writeTxnQueueDepth = 0;
+let writeTxnQueueDepthHighWater = 0;
+let readTxnQueueDepthHighWater = 0;
+
+function enterWriteQueue() {
+	if (++writeTxnQueueDepth > writeTxnQueueDepthHighWater) writeTxnQueueDepthHighWater = writeTxnQueueDepth;
+}
+function leaveWriteQueue() {
+	// Floor at zero: accounting is balanced by construction (every enterWriteQueue has exactly one
+	// matching settlement), but the guard is cheap insurance against a future call-site imbalance
+	// producing a negative depth that would corrupt every subsequent sample.
+	if (writeTxnQueueDepth > 0) writeTxnQueueDepth--;
+}
+
+/**
+ * Returns the current write/read transaction queue depths for this thread along with the high-water
+ * mark observed since the previous call, then resets the high-water marks to the current depth so the
+ * next sampling window starts fresh. Consumed by the analytics writer (see analytics/write.ts).
+ */
+export function getTransactionQueueDepths() {
+	// `readTxnQueueDepthHighWater` is maintained at the single trackedTxns growth site, so it already
+	// dominates the current size here — no need to reconcile against `readDepth` before reporting.
+	const readDepth = trackedTxns.size;
+	const depths = {
+		writeDepth: writeTxnQueueDepth,
+		writeMaxDepth: writeTxnQueueDepthHighWater,
+		readDepth,
+		readMaxDepth: readTxnQueueDepthHighWater,
+	};
+	writeTxnQueueDepthHighWater = writeTxnQueueDepth;
+	readTxnQueueDepthHighWater = readDepth;
+	return depths;
+}
+
 let confirmReplication;
 export function replicationConfirmation(callback) {
 	confirmReplication = callback;
@@ -135,6 +181,7 @@ export class DatabaseTransaction implements Transaction {
 		}
 		if ((this.transaction as any).openTimer) (this.transaction as any).openTimer = 0;
 		trackedTxns.add(this);
+		if (trackedTxns.size > readTxnQueueDepthHighWater) readTxnQueueDepthHighWater = trackedTxns.size;
 		return this.transaction;
 	}
 
@@ -288,6 +335,19 @@ export class DatabaseTransaction implements Transaction {
 							// coordinatedRetry, so it never resolves to the sentinel here. Cast
 							// away the sentinel to keep commitResolution void.
 							commitResolution = transaction.commit() as Promise<void>;
+							// Count this commit against the write queue depth until the storage engine
+							// resolves it. A transient-conflict retry rejects this promise and issues a
+							// fresh commit() (re-entering here), so the enter/leave stays balanced. leaveWriteQueue
+							// never throws, so the settled promise resolves and needs no rejection handling of its own.
+							// The thenable guard protects against a future caller passing a non-Promise
+							// `commitResolution` (today it is always rocksdb-js's async Transaction.commit()
+							// result, guaranteed to be a Promise).
+							enterWriteQueue();
+							if (commitResolution && typeof (commitResolution as any).then === 'function') {
+								commitResolution.then(leaveWriteQueue, leaveWriteQueue);
+							} else {
+								leaveWriteQueue();
+							}
 						} else {
 							try {
 								commitResolution = transaction.abort();

resources/analytics/metadata.ts

@@ -8,6 +8,8 @@ export const METRIC = {
 	NODE_STORAGE: 'node-storage',
 	ROCKSDB_STATS: 'rocksdb-stats',
 	ROCKSDB_TXNLOG_STATS: 'rocksdb-txnlog-stats',
+	WRITE_TRANSACTION_QUEUE_DEPTH: 'write-transaction-queue-depth',
+	READ_TRANSACTION_QUEUE_DEPTH: 'read-transaction-queue-depth',
 } as const;
 
 export type BuiltInMetricName = (typeof METRIC)[keyof typeof METRIC];

resources/analytics/write.ts

@@ -15,6 +15,7 @@ import * as fs from 'node:fs';
 import { getAnalyticsHostnameTable, nodeIds, stableNodeId } from './hostnames.ts';
 import { METRIC } from './metadata.ts';
 import { RocksDatabase, type TransactionLogStats } from '@harperfast/rocksdb-js';
+import { getTransactionQueueDepths } from '../DatabaseTransaction.ts';
 
 const log = forComponent('analytics').conditional;
 const isBun = typeof globalThis.Bun !== 'undefined';
@@ -213,6 +214,25 @@ function sendAnalytics() {
 			byThread: true,
 			...memoryUsage,
 		});
+		// Transaction queue depth gauges. `depth` is the instantaneous depth at emit time; `maxDepth` is
+		// the high-water mark over this sampling window (the queue can fill and drain within a single
+		// period, so the instantaneous sample alone would miss short spikes). Reported per-thread and
+		// summed across threads by the aggregator, mirroring the `memory` gauge above.
+		const queueDepths = getTransactionQueueDepths();
+		metrics.push({
+			metric: METRIC.WRITE_TRANSACTION_QUEUE_DEPTH,
+			threadId,
+			byThread: true,
+			depth: queueDepths.writeDepth,
+			maxDepth: queueDepths.writeMaxDepth,
+		});
+		metrics.push({
+			metric: METRIC.READ_TRANSACTION_QUEUE_DEPTH,
+			threadId,
+			byThread: true,
+			depth: queueDepths.readDepth,
+			maxDepth: queueDepths.readMaxDepth,
+		});
 		for (const listener of analyticsListeners) {
 			listener(metrics);
 		}

unitTests/resources/transactionQueueDepth.test.js

@@ -0,0 +1,63 @@
+require('../testUtils');
+const assert = require('assert');
+const { setupTestDBPath } = require('../testUtils');
+const { table } = require('#src/resources/databases');
+const { setMainIsWorker } = require('#js/server/threads/manageThreads');
+const { transaction } = require('#src/resources/transaction');
+const { getTransactionQueueDepths } = require('#src/resources/DatabaseTransaction');
+// The queue-depth accounting lives on the base DatabaseTransaction (RocksDB path). LMDB writes/reads
+// route through the separate LMDBTransaction overrides (resources/LMDBTransaction.ts), which maintain
+// their own unrelated trackedTxns set and do not feed this accounting — matching the existing
+// RocksDB-only carve-outs in transaction.test.js.
+const isLMDB = process.env.HARPER_STORAGE_ENGINE === 'lmdb';
+
+describe('Transaction queue depth metrics', () => {
+	let QueueTest;
+
+	before(function () {
+		setupTestDBPath();
+		setMainIsWorker(true);
+		QueueTest = table({
+			table: 'QueueDepthTest',
+			database: 'test',
+			attributes: [{ name: 'id', isPrimaryKey: true }, { name: 'name' }],
+		});
+	});
+
+	it('returns the expected shape and resets high-water marks on read', function () {
+		const depths = getTransactionQueueDepths();
+		for (const key of ['writeDepth', 'writeMaxDepth', 'readDepth', 'readMaxDepth']) {
+			assert.equal(typeof depths[key], 'number', `${key} should be a number`);
+			assert.ok(depths[key] >= 0, `${key} should be non-negative`);
+		}
+		// After reading, the high-water marks are reset to the current instantaneous depth, so an
+		// immediate second read (with no intervening activity) reports maxDepth === depth.
+		const after = getTransactionQueueDepths();
+		assert.equal(after.writeMaxDepth, after.writeDepth);
+		assert.equal(after.readMaxDepth, after.readDepth);
+	});
+
+	it('records a committed write on the write-queue high-water mark, then drains to zero', async function () {
+		if (isLMDB) return;
+		getTransactionQueueDepths(); // reset the sampling window
+		await QueueTest.put(1, { name: 'first' });
+		const depths = getTransactionQueueDepths();
+		// The commit resolved before put() returned, so the instantaneous depth is back to zero, but the
+		// high-water mark captured the in-flight commit.
+		assert.ok(depths.writeMaxDepth >= 1, `writeMaxDepth should have captured the commit, got ${depths.writeMaxDepth}`);
+		assert.equal(depths.writeDepth, 0, 'write depth should drain to zero after the commit settles');
+	});
+
+	it('reflects an open read transaction in the read-queue depth', async function () {
+		if (isLMDB) return;
+		getTransactionQueueDepths(); // reset the sampling window
+		await QueueTest.put(2, { name: 'second' });
+		let observedReadDepth = 0;
+		await transaction({}, async (context) => {
+			// A read inside the transaction opens a tracked read snapshot.
+			await QueueTest.get(2, context);
+			observedReadDepth = getTransactionQueueDepths().readDepth;
+		});
+		assert.ok(observedReadDepth >= 1, `an open read transaction should be counted, got ${observedReadDepth}`);
+	});
+});