feat: add mempool_tx_confirmation_seconds metric by tx type

src/datastore/common.ts

@@ -288,6 +288,14 @@ export interface DbTxGlobalStatus {
   microblock_hash?: string;
 }
 
+/** A mempool tx that was pruned because a canonical block confirmed it. */
+export interface MinedMempoolTx {
+  tx_id: string;
+  type_id: DbTxTypeId;
+  /** Unix time (seconds) at which the tx was first received into the mempool. */
+  receipt_time: number;
+}
+
 export interface DbMempoolStats {
   tx_type_counts: Record<string, number>;
   tx_simple_fee_averages: Record<

src/datastore/helpers.ts

@@ -11,6 +11,7 @@
   DbFaucetRequestCurrency,
   DbFtEvent,
   DbMempoolStats,
+  MinedMempoolTx,
   DbMempoolTx,
   DbMempoolTxRaw,
   DbMicroblock,
@@ -61,7 +62,7 @@
 import { getTxSenderAddress } from '../event-stream/reader.js';
 import postgres from 'postgres';
 import * as prom from 'prom-client';
-import { getAssetEventTypeString } from '../api/controllers/db-controller.js';
+import { getAssetEventTypeString, getTxTypeString } from '../api/controllers/db-controller.js';
 import { PgStoreEventEmitter } from './pg-store-event-emitter.js';
 import { logger, PgSqlClient } from '@stacks/api-toolkit';
 import PQueue from 'p-queue';
@@ -222,7 +223,7 @@
   'names',
   'namespaces',
   'subdomains',
   // TODO: add pox_set table here
 ] as const;
 
 export const POX_SYNTHETIC_EVENT_COLUMNS = [
@@ -1343,6 +1344,24 @@
   return dbTx;
 }
 
+/**
+ * Computes how long each mined tx waited in the mempool, in seconds, from its mempool receipt time
+ * to its block's timestamp. Returns nothing when the block is older than `maxLagSeconds`, which
+ * means the API is catching up to the chain tip. Negative durations caused by clock differences
+ * between the miner and the mempool observer are clamped to zero.
+ */
+export function getMempoolTxConfirmationTimes(
+  info: { blockTime: number; txs: MinedMempoolTx[] },
+  nowSeconds: number,
+  maxLagSeconds: number
+): { type: string; seconds: number }[] {
+  if (nowSeconds - info.blockTime > maxLagSeconds) return [];
+  return info.txs.map(tx => ({
+    type: getTxTypeString(tx.type_id),
+    seconds: Math.max(0, info.blockTime - tx.receipt_time),
+  }));
+}
+
 export function registerMempoolPromStats(pgEvents: PgStoreEventEmitter) {
   const mempoolTxCountGauge = new prom.Gauge({
     name: `mempool_tx_count`,
@@ -1364,6 +1383,12 @@
     help: 'Average byte size of txs in the mempool, by tx type',
     labelNames: ['type', 'percentile'] as const,
   });
+  const mempoolTxConfirmationHistogram = new prom.Histogram({
+    name: 'mempool_tx_confirmation_seconds',
+    help: 'Time between a tx first being received into the mempool and its block timestamp, by tx type',
+    labelNames: ['type'] as const,
+    buckets: [1, 2, 3, 4, 5, 7, 10, 15, 20, 30, 45, 60, 120, 300, 600, 1800, 3600, 7200, 21600],
+  });
   const updatePromMempoolStats = (mempoolStats: DbMempoolStats) => {
     for (const txType in mempoolStats.tx_type_counts) {
       const entry = mempoolStats.tx_type_counts[txType];
@@ -1397,6 +1422,20 @@
       }
     });
   });
+  pgEvents.addListener('mempoolTxsMined', info => {
+    try {
+      const times = getMempoolTxConfirmationTimes(
+        info,
+        Math.round(Date.now() / 1000),
+        ENV.MEMPOOL_CONFIRMATION_METRIC_MAX_LAG
+      );
+      for (const { type, seconds } of times) {
+        mempoolTxConfirmationHistogram.observe({ type }, seconds);
+      }
+    } catch (error) {
+      logger.error(error, 'Error updating prometheus mempool confirmation stats');
+    }
+  });
 }
 
 export function convertTxQueryResultToDbMempoolTx(txs: TxQueryResult[]): DbMempoolTxRaw[] {

src/datastore/pg-store-event-emitter.ts

@@ -1,5 +1,5 @@
 import { EventEmitter } from 'events';
-import { DbConfigState, DbMempoolStats } from './common.js';
+import { DbConfigState, DbMempoolStats, MinedMempoolTx } from './common.js';
 
 interface DataStoreEvents {
   txUpdate: (txId: string) => void;
@@ -12,6 +12,8 @@ interface DataStoreEvents {
   tokensUpdate: (contractID: string) => void;
   tokenMetadataUpdateQueued: (queueId: number) => void;
   mempoolStatsUpdate: (mempoolStats: DbMempoolStats) => void;
+  /** Emitted after a canonical block that confirmed mempool txs has been committed. */
+  mempoolTxsMined: (info: { blockTime: number; txs: MinedMempoolTx[] }) => void;
   configStateUpdate: (configState: DbConfigState) => void;
 }
 

src/datastore/pg-write-store.ts

@@ -54,6 +54,7 @@ import {
   DbTxRaw,
   DbMempoolTxRaw,
   DbChainTip,
+  MinedMempoolTx,
   NftCustodyInsertValues,
   DataStoreBnsBlockTxData,
   DbPox4SyntheticEvent,
@@ -302,6 +303,7 @@ export class PgWriteStore extends PgStore {
 
   async update(data: DataStoreBlockUpdateData): Promise<void> {
     let garbageCollectedMempoolTxs: string[] = [];
+    let minedMempoolTxs: MinedMempoolTx[] = [];
     let reorg: ReOrgUpdatedEntities = newReOrgUpdatedEntities();
     let isCanonical = true;
     let skippedDuplicateBlock = false;
@@ -341,7 +343,11 @@ export class PgWriteStore extends PgStore {
           nonce: d.tx.nonce,
           sponsor_nonce: d.tx.sponsor_nonce,
         }));
-        await this.pruneMempoolTxs(sql, prunableTxs);
+        const pruneResult = await this.pruneMempoolTxs(sql, prunableTxs);
+        // Only the txs included in this block were mined; the rest of the pruned txs were
+        // displaced from their nonce slots.
+        const blockTxIds = new Set(prunableTxs.map(tx => tx.txId));
+        minedMempoolTxs = pruneResult.prunedTxs.filter(tx => blockTxIds.has(tx.tx_id));
       }
 
       if (isCanonical) {
@@ -438,6 +444,12 @@ export class PgWriteStore extends PgStore {
       }
     });
     if (skippedDuplicateBlock) return;
+    if (minedMempoolTxs.length > 0 && !this.isEventReplay) {
+      this.eventEmitter.emit('mempoolTxsMined', {
+        blockTime: data.block.block_time,
+        txs: minedMempoolTxs,
+      });
+    }
     if (isCanonical) {
       await this.redisNotifier?.notify(
         {
@@ -4381,8 +4393,8 @@ export class PgWriteStore extends PgStore {
   async pruneMempoolTxs(
     sql: PgSqlClient,
     transactions: TransactionHeader[]
-  ): Promise<{ removedTxs: string[] }> {
-    if (transactions.length === 0) return { removedTxs: [] };
+  ): Promise<{ removedTxs: string[]; prunedTxs: MinedMempoolTx[] }> {
+    if (transactions.length === 0) return { removedTxs: [], prunedTxs: [] };
     if (logger.isLevelEnabled('debug'))
       for (const tx of transactions)
         logger.debug(
@@ -4391,22 +4403,22 @@ export class PgWriteStore extends PgStore {
 
     // Prune confirmed txs from the mempool. Also prune transactions that share a nonce slot with
     // them. We'll recalculate replaced-by-fee data later when new block data is written to the DB.
-    const updateResults = await sql<{ tx_id: string }[]>`
+    const updateResults = await sql<MinedMempoolTx[]>`
       WITH ${this.sameNonceSlotMempoolTxIdsCte(sql, transactions)},
       pruned AS (
         UPDATE mempool_txs
         SET pruned = true, replaced_by_tx_id = NULL
         WHERE pruned = false AND tx_id IN (SELECT tx_id FROM affected_mempool_tx_ids)
-        RETURNING tx_id
+        RETURNING tx_id, type_id, receipt_time
       ),
       count_update AS (
         UPDATE chain_tip SET
           mempool_tx_count = mempool_tx_count - (SELECT COUNT(*) FROM pruned),
           mempool_updated_at = NOW()
       )
-      SELECT tx_id FROM pruned
+      SELECT tx_id, type_id, receipt_time FROM pruned
     `;
-    return { removedTxs: updateResults.map(r => r.tx_id) };
+    return { removedTxs: updateResults.map(r => r.tx_id), prunedTxs: updateResults };
   }
 
   /**

src/env.ts

@@ -200,6 +200,12 @@ const schema = Type.Object({
    * The maximum duration to wait for further mempool updates after the debounce interval.
    */
   MEMPOOL_STATS_DEBOUNCE_MAX_INTERVAL: Type.Integer({ default: 10000, minimum: 0 }),
+  /**
+   * Maximum age (in seconds) of a block, relative to the current time, for its mempool txs to be
+   * recorded in the `mempool_tx_confirmation_seconds` metric. Blocks ingested while catching up to
+   * the chain tip are skipped so they don't flood the metric with historical observations.
+   */
+  MEMPOOL_CONFIRMATION_METRIC_MAX_LAG: Type.Integer({ default: 300, minimum: 0 }),
 
   BTC_RPC_HOST: Type.String({ default: 'http://127.0.0.1' }),
   BTC_RPC_PORT: Type.Integer({ default: 18443, minimum: 0, maximum: 65535 }),

src/event-stream/event-server.ts

@@ -94,6 +94,27 @@ export function eventErrorResponse(error: unknown): { error: string } {
   return { error: normalizeErrorMessage(messages.join(': ')) };
 }
 
+/**
+ * Header the SNP stream handler sets on injected event requests, carrying the time SNP received the
+ * event from the Stacks node. SNP replays keep the original value, so it stays accurate while the
+ * API catches up to the chain tip.
+ */
+export const SNP_TIMESTAMP_HEADER = 'x-snp-timestamp';
+
+/**
+ * Resolves the unix time (in seconds) at which an event was first received. Uses the SNP timestamp
+ * header when present (epoch milliseconds, or any `Date`-parseable string), falling back to the
+ * current time for events posted directly by a Stacks node.
+ */
+export function getEventReceiptTime(header: string | string[] | undefined): number {
+  const value = Array.isArray(header) ? header[0] : header;
+  if (value) {
+    const ms = /^\d+$/.test(value) ? Number(value) : Date.parse(value);
+    if (Number.isFinite(ms)) return Math.round(ms / 1000);
+  }
+  return Math.round(Date.now() / 1000);
+}
+
 /** Collapse all runs of whitespace (incl. newlines) to single spaces and trim. */
 function normalizeErrorMessage(message: string): string {
   return message.replace(/\s+/g, ' ').trim();
@@ -183,9 +204,12 @@ async function handleBurnBlockMessage(
   });
 }
 
-async function handleMempoolTxsMessage(rawTxs: string[], db: PgWriteStore): Promise<void> {
+async function handleMempoolTxsMessage(
+  rawTxs: string[],
+  receiptDate: number,
+  db: PgWriteStore
+): Promise<void> {
   logger.debug(`Received ${rawTxs.length} mempool transactions`);
-  const receiptDate = Math.round(Date.now() / 1000);
   const decodedTxs = rawTxs.map(str => {
     const parsedTx = decodeTransaction(str);
     const txSender = getTxSenderAddress(parsedTx);
@@ -619,7 +643,7 @@ interface EventMessageHandler {
     msg: NewBlockMessage,
     db: PgWriteStore
   ): Promise<void> | void;
-  handleMempoolTxs(rawTxs: string[], db: PgWriteStore): Promise<void> | void;
+  handleMempoolTxs(rawTxs: string[], receiptTime: number, db: PgWriteStore): Promise<void> | void;
   handleBurnBlock(msg: NewBurnBlockMessage, db: PgWriteStore): Promise<void> | void;
   handleDroppedMempoolTxs(msg: DropMempoolTxMessage, db: PgWriteStore): Promise<void> | void;
   handleNewAttachment(msg: AttachmentsNewMessage[], db: PgWriteStore): Promise<void> | void;
@@ -708,9 +732,11 @@ function createMessageProcessorQueue(db: PgWriteStore): EventMessageHandler {
           throw e;
         });
     },
-    handleMempoolTxs: (rawTxs: string[], db: PgWriteStore) => {
+    handleMempoolTxs: (rawTxs: string[], receiptTime: number, db: PgWriteStore) => {
       return secondaryQueue
-        .add(() => observeEvent('mempool_txs', () => handleMempoolTxsMessage(rawTxs, db)))
+        .add(() =>
+          observeEvent('mempool_txs', () => handleMempoolTxsMessage(rawTxs, receiptTime, db))
+        )
         .catch(e => {
           logger.error(e, 'Error processing core node mempool message');
           throw e;
@@ -867,7 +893,8 @@ export async function startEventServer(opts: {
   app.post('/new_mempool_tx', async (req, res) => {
     try {
       const rawTxs = req.body as string[];
-      await messageHandler.handleMempoolTxs(rawTxs, db);
+      const receiptTime = getEventReceiptTime(req.headers[SNP_TIMESTAMP_HEADER]);
+      await messageHandler.handleMempoolTxs(rawTxs, receiptTime, db);
       await handleRawEventRequest(req);
       await res.status(200).send({ result: 'ok' });
     } catch (error) {

src/event-stream/snp-event-stream.ts

@@ -1,7 +1,7 @@
 import { SERVER_VERSION } from '@stacks/api-toolkit';
 import { logger as defaultLogger } from '@stacks/api-toolkit';
 import { EventEmitter } from 'node:events';
-import { EventStreamServer } from './event-server.js';
+import { EventStreamServer, SNP_TIMESTAMP_HEADER } from './event-server.js';
 import { PgWriteStore } from '../datastore/pg-write-store.js';
 import { StacksMessageStream, MessagePath } from '@stacks/node-publisher-client';
 import { ENV } from '../env.js';
@@ -70,7 +70,7 @@ export class SnpEventStreamHandler {
   }
 
   // eslint-disable-next-line @typescript-eslint/no-explicit-any
-  async handleMsg(messageId: string, _timestamp: string, path: string, body: any) {
+  async handleMsg(messageId: string, timestamp: string, path: string, body: any) {
     this.logger.debug(`Received SNP stream event ${path}, msgId: ${messageId}`);
     let response;
 
@@ -79,6 +79,7 @@ export class SnpEventStreamHandler {
         method: 'POST',
         url: path,
         payload: body,
+        headers: { [SNP_TIMESTAMP_HEADER]: timestamp },
       });
     } catch (error) {
       const errorMessage = `Failed to process SNP message ${messageId} at path ${path}: ${error}`;