diff --git a/resources/DatabaseTransaction.ts b/resources/DatabaseTransaction.ts index 30a08fd75e..fa8a09c0ae 100644 --- a/resources/DatabaseTransaction.ts +++ b/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; + // 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(); diff --git a/resources/analytics/metadata.ts b/resources/analytics/metadata.ts index 08e5bf94d3..d59809723f 100644 --- a/resources/analytics/metadata.ts +++ b/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]; diff --git a/resources/analytics/write.ts b/resources/analytics/write.ts index e325008fb3..51c5d332c5 100644 --- a/resources/analytics/write.ts +++ b/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); } diff --git a/unitTests/resources/transactionQueueDepth.test.js b/unitTests/resources/transactionQueueDepth.test.js new file mode 100644 index 0000000000..3945b2148b --- /dev/null +++ b/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}`); + }); +});