Add lease transaction statistics

README.md

@@ -48,13 +48,21 @@ paths concurrently.
     clean up items written to a queue by a process outside your control (e.g., webhooks).
   * `healthyPingLatency: {number | string}` the maximum response latency to pings that is considered
     "healthy" for this queue.
+  * `captureLeaseTransactionMetrics: {function(string, number, number)}` a callback invoked after
+    each acquired, contended, or failed task lease transaction.  It receives the acquisition
+    outcome, NodeFire transaction tries, and transaction duration in milliseconds.  Missing optional
+    NodeFire metadata is reported as zero.  The callback must be synchronous.  Callback errors are
+    reported through
+    `settings.captureError` and do not affect task processing.
 
 * `@param {function(Object):RETRY | number | string | undefined}` worker The worker function that
   handles enqueued tasks.  It will be given a task object as argument, with a special $ref attribute
-  set to the Nodefire ref of that task.  The worker can perform arbitrary computation whose duration
-  should not exceed the queue's minLease value.  It can manipulate the task itself in Firebase as
-  well, e.g. to delete it (to get at-most-once queue semantics) or otherwise modify it.  The worker
-  can return any of the following:
+  set to the Nodefire ref of that task.  On a task's first acquisition, the worker-facing `_lease`
+  object also has a non-enumerable `firstAcquisition: true` property that is not saved to Firebase;
+  the property is absent on subsequent acquisitions.  The worker can perform arbitrary computation
+  whose duration should not exceed the queue's minLease value.  It can manipulate the task itself in
+  Firebase as well, e.g. to delete it (to get at-most-once queue semantics) or otherwise modify it.
+  The worker can return any of the following:
   * undefined or null to cause the task to be retired from the queue.
   * firelease.RETRY to cause the task to be retried after the current lease expires (and reset the
     lease backoff counter).
@@ -142,8 +150,16 @@ Mutable default option values for all subsequent attachWorker calls.  See that f
 ```stats: {Object}```
 
 The live stats object also passed to the ping callback.  Global fields include `healthy`,
-`sickQueues`, `sickSources`, `stuckTasks`, `maxLatency`, and `tasksAcquired`.  Each entry in `queues`
-includes its own health, latency, acquisition count, and all physical `sources`.  Queue-level
+`sickQueues`, `sickSources`, `stuckTasks`, `maxLatency`, `leaseTransactions`, and the legacy
+`tasksAcquired`.  `leaseTransactions` contains lifetime `acquired`, `contended`, `failed`, and
+`tries` counts for task lease transactions.  `tries` includes failed transactions and comes from
+NodeFire transaction metadata.  `duration` is an exponential moving average of the NodeFire
+transaction duration in milliseconds, using an alpha of 0.1.  These stats are available for every
+physical source.  Logical queue and global counts are additive, while their duration is the average
+of the underlying source or queue duration values that have recorded at least one attempt.  The
+legacy `tasksAcquired` field also remains
+lifetime-cumulative.  Each entry in `queues` includes its own health, latency, leasing totals, and
+all physical `sources`.  Queue-level
 `size` and `sizeDelta` sum their source values when all are known, `sizeTimestamp` is the oldest
 source timestamp, and `mode` is `full`, `safe`, or `mixed`.  Source stats include `connected`,
 current `mode` (`full` or `safe`), last known `size`, `sizeTimestamp` when the size came from a

package.json

@@ -1,6 +1,6 @@
 {
   "name": "firelease",
-  "version": "4.1.0",
+  "version": "4.2.0",
   "packageManager": "yarn@4.13.0",
   "description": "Firebase queue consumer for Node with at-least-once semantics",
   "main": "built/index.js",

src/firelease.ts

@@ -1,6 +1,6 @@
 import _ from 'lodash';
 import ms from 'ms';
-import NodeFire from 'nodefire';
+import NodeFire, {type TransactionMetadata} from 'nodefire';
 import * as timers from 'safe-timers';
 import {
   FireleaseStats, QueueSourceStats, QueueStats, type QueueSourceMode
@@ -14,6 +14,7 @@ const QUEUE_CHECK_TIMEOUT = ms('15s');
 const QUEUE_SIZE_HYSTERESIS = 0.15;
 const QUEUE_SIZE_MISMATCH_THRESHOLD = 100;
 const DEMOTION_JITTER = ms('30s');
+const LEASE_TRANSACTION_DURATION_ALPHA = 0.1;
 
 declare const RETRY_DIRECTIVE: unique symbol;
 
@@ -29,6 +30,10 @@ export interface Lease {
   extendLeasePromise?: Promise<void>;
 }
 
+export type AcquiredLease = Lease & {
+  expiry: number, time: number, attempts: number, initial: number, readonly firstAcquisition?: true
+};
+
 export interface LeaseItem {
   _lease?: Lease;
   [key: string]: any;
@@ -39,7 +44,7 @@ export interface RetryDirective {
 }
 
 export interface WorkerItem extends LeaseItem {
-  _lease: Lease & {expiry: number};
+  _lease: AcquiredLease;
   readonly $ref: NodeFire;
   readonly $leaseTimeRemaining: number;
 }
@@ -69,13 +74,19 @@ export interface FireleaseError extends Error {
   level?: FireleaseErrorLevel;
 }
 
+export type LeaseTransactionOutcome = 'acquired' | 'contended' | 'failed';
+export type CaptureLeaseTransactionMetrics = (
+  outcome: LeaseTransactionOutcome, tries: number, duration: number
+) => undefined;
+
 export interface QueueOptions {
   maxConcurrent?: number;
   bufferSize?: number;
   minLease?: Duration;
   maxLease?: Duration;
   healthyPingLatency?: Duration;
   preprocess?: (item: LeaseItem) => LeaseItem;
+  captureLeaseTransactionMetrics?: CaptureLeaseTransactionMetrics;
 }
 
 export type PingReport = FireleaseStats;
@@ -121,6 +132,7 @@ interface NormalizedQueueOptions {
   maxLease: number;
   healthyPingLatency: number;
   preprocess?: (item: LeaseItem) => LeaseItem;
+  captureLeaseTransactionMetrics?: CaptureLeaseTransactionMetrics;
 }
 
 const queues: Queue[] = [];
@@ -261,12 +273,16 @@ class Task {
   async process() {
     let startTimestamp = 0;
     let acquired = false;
+    let contended = false;
     let reschedule = true;
+    let firstAcquisition = false;
     this.working = true;
     this.phase = 'lease';
     const transactionPromise = this.ref.transaction(itemValue => {
       const item = itemValue as LeaseItem | null;
       acquired = false;
+      contended = false;
+      firstAcquisition = false;
       if (tasks[this.key] !== this || this.removed) return;
       if (!item || this.ref.key === PING_KEY) {
         acquired = true;
@@ -276,9 +292,11 @@ class Task {
       // console.log('txn  ', this.ref.key, 'lease', item._lease, 'now', startTimestamp);
       // Check if another process beat us to it.
       if (item._lease?.expiry && item._lease.expiry > startTimestamp) {
+        contended = true;
         return item;
       }
       acquired = true;
+      firstAcquisition = _.isNil(item._lease?.initial);
       item._lease ??= {};
       item._lease.time = this.queue.constrainLeaseDuration((item._lease.time ?? 0) * 2);
       item._lease.expiry = startTimestamp + item._lease.time;
@@ -287,14 +305,22 @@ class Task {
       item._lease.busy = true;
       return this.queue.callPreprocess(item);
     }, {detectStuck: 5, prefetchValue: false, timeout: ms('15s')});
+    let transactionCompleted = false;
     try {
       const item = await transactionPromise;
+      transactionCompleted = true;
       if (acquired && item !== null && this.ref.key !== PING_KEY) {
-        if (!_.isObject(item)) throw new Error(`item not an object: ${item}`);
+        this.recordLeaseTransaction('acquired', transactionPromise.transaction);
+        if (firstAcquisition) Object.defineProperty(item._lease, 'firstAcquisition', {value: true});
         this.queue.stats.tasksAcquired++;
         await this.run(item as WorkerItem, startTimestamp);
+      } else if (contended) {
+        this.recordLeaseTransaction('contended', transactionPromise.transaction);
       }
     } catch (error) {
+      if (!transactionCompleted && this.ref.key !== PING_KEY) {
+        this.recordLeaseTransaction('failed', transactionPromise.transaction);
+      }
       reschedule = false;
       // Hardcoded retry -- hard to do anything smarter, since we failed to update the task in
       // Firebase.
@@ -315,6 +341,37 @@ class Task {
     }
   }
 
+  recordLeaseTransaction(
+    outcome: LeaseTransactionOutcome, transaction: TransactionMetadata | undefined
+  ) {
+    try {
+      const tries = transaction?.tries ?? 0;
+      const transactionDuration = transaction?.duration ?? 0;
+      const leaseStats = this.source.stats.leaseTransactions;
+      const priorAttempts = leaseStats.acquired + leaseStats.contended + leaseStats.failed;
+      leaseStats[outcome] += 1;
+      leaseStats.tries += tries;
+      leaseStats.duration = priorAttempts === 0 ?
+        transactionDuration :
+        leaseStats.duration * (1 - LEASE_TRANSACTION_DURATION_ALPHA) +
+          transactionDuration * LEASE_TRANSACTION_DURATION_ALPHA;
+      this.queue.options.captureLeaseTransactionMetrics?.(outcome, tries, transactionDuration);
+    } catch (error) {
+      try {
+        const metricError: FireleaseError = _.isError(error) ? error : new Error(String(error));
+        metricError.firelease = _.assign(
+          metricError.firelease ?? {}, {itemKey: this.key, phase: 'lease-metric'});
+        settings.captureError(metricError);
+      } catch (captureError) {
+        try {
+          console.error('Error capturing lease transaction metric error:', captureError);
+        } catch {
+          // Metric recording must never interrupt task processing.
+        }
+      }
+    }
+  }
+
   async run(item: WorkerItem, startTimestamp: number) {
     Object.defineProperty(item, '$ref', {value: this.ref});
     Object.defineProperty(item, '$leaseTimeRemaining', {get: () => {
@@ -395,7 +452,7 @@ class Task {
         if (currentItem._lease) delete currentItem._lease.busy;
         return currentItem;
       }, {prefetchValue: false}) as LeaseItem | null | undefined;
-      if (item2) item._lease = item2._lease as Lease & {expiry: number};
+      if (item2) item._lease = item2._lease as AcquiredLease;
     } catch (postProcessingError) {
       this.handlePostProcessingError(postProcessingError);
     }
@@ -873,7 +930,7 @@ class QueueSource {
   }
 
   recordPingResult(startedAt: number, succeeded: boolean) {
-    const latency = performance.now() - startedAt;
+    const latency = Math.round(performance.now() - startedAt);
     this.stats.latency = latency;
     this.stats.healthy = succeeded && latency < this.queue.options.healthyPingLatency;
     this.stats.pingTimestamp = Date.now();
@@ -1069,9 +1126,15 @@ class Queue {
  *          control (e.g., webhooks).
  *        healthyPingLatency: {number | string} the maximum response latency to pings that is
  *          considered "healthy" for this queue.
+ *        captureLeaseTransactionMetrics: {function(string, number, number)} a callback invoked
+ *          after each acquired, contended, or failed task lease transaction with its outcome,
+ *          NodeFire transaction tries, and duration in milliseconds.  The callback must be
+ *          synchronous.
  * @param {function(Object):RETRY | number | string | undefined} worker The worker function that
  *        handles enqueued tasks.  It will be given a task object as argument, with a special $ref
- *        attribute set to the Nodefire ref of that task.  The worker can perform arbitrary
+ *        attribute set to the Nodefire ref of that task.  On a task's first acquisition its _lease
+ *        also has a non-enumerable firstAcquisition property set to true; it is not saved to
+ *        Firebase and is absent on subsequent acquisitions.  The worker can perform arbitrary
  *        computation whose duration should not exceed the queue's minLease value.  It can
  *        manipulate the task itself in Firebase as well, e.g. to delete it (to get at-most-once
  *        queue semantics) or otherwise modify it.  The worker can return any of the following:

src/stats.ts

@@ -3,6 +3,33 @@ import _ from 'lodash';
 export type QueueSourceMode = 'full' | 'safe';
 export type QueueMode = QueueSourceMode | 'mixed';
 
+export interface LeaseTransactionStats {
+  acquired: number;
+  contended: number;
+  failed: number;
+  tries: number;
+  duration: number;
+}
+
+function createLeaseTransactionStats(): LeaseTransactionStats {
+  return {acquired: 0, contended: 0, failed: 0, tries: 0, duration: 0};
+}
+
+function rollUpLeaseTransactions(items: LeaseTransactionStats[]) {
+  const result = createLeaseTransactionStats();
+  result.acquired = _.sumBy(items, 'acquired');
+  result.contended = _.sumBy(items, 'contended');
+  result.failed = _.sumBy(items, 'failed');
+  result.tries = _.sumBy(items, 'tries');
+  const attemptedItems = _.filter(items, countLeaseAttempts);
+  if (attemptedItems.length) result.duration = _.meanBy(attemptedItems, 'duration');
+  return result;
+}
+
+function countLeaseAttempts(stats: LeaseTransactionStats) {
+  return stats.acquired + stats.contended + stats.failed;
+}
+
 function exposeGetters(instance: object, properties: string[]) {
   const prototype = Object.getPrototypeOf(instance);
   for (const property of properties) {
@@ -20,6 +47,7 @@ export class QueueSourceStats {
   healthy = true;
   latency: number | null = null;
   declare pingTimestamp?: number;
+  readonly leaseTransactions = createLeaseTransactionStats();
 
   constructor(readonly ref: string) {}
 }
@@ -32,7 +60,10 @@ export class QueueStats {
     readonly key: string | null,
     readonly sources: QueueSourceStats[]
   ) {
-    exposeGetters(this, ['mode', 'size', 'sizeDelta', 'sizeTimestamp', 'healthy', 'maxLatency']);
+    exposeGetters(
+      this,
+      ['mode', 'size', 'sizeDelta', 'sizeTimestamp', 'healthy', 'maxLatency', 'leaseTransactions'],
+    );
   }
 
   get mode(): QueueMode {
@@ -61,6 +92,10 @@ export class QueueStats {
   get maxLatency() {
     return _(this.sources).map('latency').max() || 0;
   }
+
+  get leaseTransactions() {
+    return rollUpLeaseTransactions(_.map(this.sources, 'leaseTransactions'));
+  }
 }
 
 export class FireleaseStats {
@@ -70,7 +105,11 @@ export class FireleaseStats {
   constructor(getStuckTasks: () => number) {
     this.#getStuckTasks = getStuckTasks;
     exposeGetters(
-      this, ['healthy', 'sickQueues', 'sickSources', 'stuckTasks', 'maxLatency', 'tasksAcquired'],
+      this,
+      [
+        'healthy', 'sickQueues', 'sickSources', 'stuckTasks', 'maxLatency', 'leaseTransactions',
+        'tasksAcquired'
+      ],
     );
   }
 
@@ -99,6 +138,10 @@ export class FireleaseStats {
     return _(this.queues).map('maxLatency').max() || 0;
   }
 
+  get leaseTransactions() {
+    return rollUpLeaseTransactions(_.map(this.queues, 'leaseTransactions'));
+  }
+
   get tasksAcquired() {
     return _.sumBy(this.queues, queue => queue.tasksAcquired);
   }

tests/fake_firebase.ts

@@ -1,5 +1,6 @@
 import assert from 'node:assert';
 import type NodeFire from 'nodefire';
+import type {TransactionMetadata} from 'nodefire';
 
 interface FakeLease {
   [key: string]: unknown;
@@ -92,12 +93,24 @@ export class FakeTaskRef {
   transaction(
     update: (value: FakeTaskValue | null) => FakeTaskValue | null | undefined
   ) {
-    if (this.queueRef.transactionError) return Promise.reject(this.queueRef.transactionError);
-    const previous = clone(this.value);
-    const updated = update(clone(this.value));
-    if (updated !== undefined) this.value = clone(updated);
-    this.queueRef.notifyTaskChange(this, previous);
-    return Promise.resolve(clone(this.value));
+    this.queueRef.beforeTransaction?.(this);
+    const metadata: TransactionMetadata = {
+      outcome: this.queueRef.transactionError ? 'error' : 'commit',
+      tries: this.queueRef.transactionTries,
+      duration: this.queueRef.transactionDuration
+    };
+    let transactionPromise: Promise<FakeTaskValue | null>;
+    if (this.queueRef.transactionError) {
+      transactionPromise = Promise.reject(this.queueRef.transactionError);
+    } else {
+      const previous = clone(this.value);
+      const updated = update(clone(this.value));
+      if (updated !== undefined) this.value = clone(updated);
+      this.queueRef.notifyTaskChange(this, previous);
+      transactionPromise = Promise.resolve(clone(this.value));
+    }
+    return this.queueRef.omitTransactionMetadata ?
+      transactionPromise : Object.assign(transactionPromise, {transaction: metadata});
   }
 
   get() {
@@ -214,6 +227,10 @@ export class FakeQueueRef {
   childrenKeysCalls = 0;
   listenerError?: Error;
   transactionError?: Error;
+  omitTransactionMetadata = false;
+  transactionTries = 1;
+  transactionDuration = 0;
+  beforeTransaction?: (ref: FakeTaskRef) => void;
   fixedNow?: number;
 
   constructor(databaseName: string, readonly path: string) {