From 3181d7d3c2c963f65dccbdd77075a7cfb8305861 Mon Sep 17 00:00:00 2001 From: Rafa Cardenas <253999660+rafa-stacks@users.noreply.github.com> Date: Wed, 7 Oct 2026 15:40:15 -0600 Subject: [PATCH 1/2] add mempool confirmation metrics --- src/datastore/common.ts | 8 + src/datastore/helpers.ts | 41 +++- src/datastore/pg-store-event-emitter.ts | 4 +- src/datastore/pg-write-store.ts | 26 ++- src/env.ts | 6 + src/event-stream/event-server.ts | 39 +++- src/event-stream/snp-event-stream.ts | 5 +- .../mempool-confirmation-metric.test.ts | 178 ++++++++++++++++++ tests/snp/snp-ingestion.test.ts | 12 ++ 9 files changed, 302 insertions(+), 17 deletions(-) create mode 100644 tests/api/mempool/mempool-confirmation-metric.test.ts diff --git a/src/datastore/common.ts b/src/datastore/common.ts index 66934df971..7feca2bc4d 100644 --- a/src/datastore/common.ts +++ b/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; tx_simple_fee_averages: Record< diff --git a/src/datastore/helpers.ts b/src/datastore/helpers.ts index 0c2905df39..714f5c5626 100644 --- a/src/datastore/helpers.ts +++ b/src/datastore/helpers.ts @@ -11,6 +11,7 @@ import { DbFaucetRequestCurrency, DbFtEvent, DbMempoolStats, + MinedMempoolTx, DbMempoolTx, DbMempoolTxRaw, DbMicroblock, @@ -61,7 +62,7 @@ import type { 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'; @@ -1343,6 +1344,24 @@ export function createDbTxFromCoreMsg(msg: CoreNodeParsedTxMessage): DbTxRaw { 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 @@ export function registerMempoolPromStats(pgEvents: PgStoreEventEmitter) { 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 @@ export function registerMempoolPromStats(pgEvents: PgStoreEventEmitter) { } }); }); + 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[] { diff --git a/src/datastore/pg-store-event-emitter.ts b/src/datastore/pg-store-event-emitter.ts index 2d2b9f0e8c..fd5cb2af09 100644 --- a/src/datastore/pg-store-event-emitter.ts +++ b/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; } diff --git a/src/datastore/pg-write-store.ts b/src/datastore/pg-write-store.ts index 322ec201a4..6dab5cfe9c 100644 --- a/src/datastore/pg-write-store.ts +++ b/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 { 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` 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 }; } /** diff --git a/src/env.ts b/src/env.ts index db88aafe4f..c598f90a54 100644 --- a/src/env.ts +++ b/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 }), diff --git a/src/event-stream/event-server.ts b/src/event-stream/event-server.ts index e6f9444681..1957ef7052 100644 --- a/src/event-stream/event-server.ts +++ b/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 { +async function handleMempoolTxsMessage( + rawTxs: string[], + receiptDate: number, + db: PgWriteStore +): Promise { 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; - handleMempoolTxs(rawTxs: string[], db: PgWriteStore): Promise | void; + handleMempoolTxs(rawTxs: string[], receiptTime: number, db: PgWriteStore): Promise | void; handleBurnBlock(msg: NewBurnBlockMessage, db: PgWriteStore): Promise | void; handleDroppedMempoolTxs(msg: DropMempoolTxMessage, db: PgWriteStore): Promise | void; handleNewAttachment(msg: AttachmentsNewMessage[], db: PgWriteStore): Promise | 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) { diff --git a/src/event-stream/snp-event-stream.ts b/src/event-stream/snp-event-stream.ts index 59f33706b1..fc5f909880 100644 --- a/src/event-stream/snp-event-stream.ts +++ b/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}`; diff --git a/tests/api/mempool/mempool-confirmation-metric.test.ts b/tests/api/mempool/mempool-confirmation-metric.test.ts new file mode 100644 index 0000000000..47bece6bc5 --- /dev/null +++ b/tests/api/mempool/mempool-confirmation-metric.test.ts @@ -0,0 +1,178 @@ +import { PgWriteStore } from '../../../src/datastore/pg-write-store.ts'; +import { DbTxTypeId, MinedMempoolTx } from '../../../src/datastore/common.ts'; +import { getMempoolTxConfirmationTimes } from '../../../src/datastore/helpers.ts'; +import { + EventStreamServer, + getEventReceiptTime, + SNP_TIMESTAMP_HEADER, + startEventServer, +} from '../../../src/event-stream/event-server.ts'; +import { TestBlockBuilder, testMempoolTx } from '../test-builders.ts'; +import { PgSqlClient } from '@stacks/api-toolkit'; +import { migrate } from '../../test-helpers.ts'; +import { afterEach, beforeEach, describe, test } from 'node:test'; +import assert from 'node:assert/strict'; +import { STACKS_TESTNET } from '@stacks/network'; + +// A real token transfer taken from the `epoch-3-transition` SNP dump. +const RAW_MEMPOOL_TX = + '0x80800000000400ad08341feab8ea788ef8045c343d21dcedc4483e0000000000000000000000000000012c00018ebc106d9348b2638573b891aa0ebaad6303d8f18f801a3600e4175baf7d44d412d2af6ec95a541a9224a538a22b7c82055730f8293fcd91899eb3757d426f6003020000000000051a62b0e91cc557e583c3d1f9dfe468ace76d2f037400000000000003e800000000000000000000000000000000000000000000000000000000000000000000'; + +describe('mempool confirmation time helpers', () => { + test('getEventReceiptTime parses SNP timestamps', () => { + assert.equal(getEventReceiptTime('1710343460976'), 1710343461); + assert.equal(getEventReceiptTime('1710343460276'), 1710343460); + assert.equal(getEventReceiptTime('2024-03-13 15:24:20.97639+00'), 1710343461); + assert.equal(getEventReceiptTime(['1710343460000', '1']), 1710343460); + }); + + test('getEventReceiptTime falls back to the current time', () => { + for (const header of [undefined, '', 'not-a-date']) { + const before = Math.round(Date.now() / 1000); + const receiptTime = getEventReceiptTime(header); + const after = Math.round(Date.now() / 1000); + assert.ok(receiptTime >= before && receiptTime <= after); + } + }); + + test('getMempoolTxConfirmationTimes labels by tx type and clamps negatives', () => { + const txs: MinedMempoolTx[] = [ + { tx_id: '0x01', type_id: DbTxTypeId.TokenTransfer, receipt_time: 940 }, + { tx_id: '0x02', type_id: DbTxTypeId.ContractCall, receipt_time: 1000 }, + { tx_id: '0x03', type_id: DbTxTypeId.VersionedSmartContract, receipt_time: 1003 }, + ]; + assert.deepEqual(getMempoolTxConfirmationTimes({ blockTime: 1000, txs }, 1010, 300), [ + { type: 'token_transfer', seconds: 60 }, + { type: 'contract_call', seconds: 0 }, + { type: 'smart_contract', seconds: 0 }, + ]); + }); + + test('getMempoolTxConfirmationTimes skips blocks older than the max lag', () => { + const txs: MinedMempoolTx[] = [ + { tx_id: '0x01', type_id: DbTxTypeId.TokenTransfer, receipt_time: 900 }, + ]; + assert.equal(getMempoolTxConfirmationTimes({ blockTime: 1000, txs }, 1300, 300).length, 1); + assert.deepEqual(getMempoolTxConfirmationTimes({ blockTime: 1000, txs }, 1301, 300), []); + }); +}); + +describe('mempool confirmation time ingestion', () => { + let db: PgWriteStore; + let client: PgSqlClient; + let eventServer: EventStreamServer; + + beforeEach(async () => { + await migrate('up'); + db = await PgWriteStore.connect({ + usageName: 'tests', + withNotifier: false, + skipMigrations: true, + }); + client = db.sql; + eventServer = await startEventServer({ + datastore: db, + chainId: STACKS_TESTNET.chainId, + serverHost: '127.0.0.1', + serverPort: 0, + }); + }); + + afterEach(async () => { + await eventServer.closeAsync(); + await db?.close(); + await migrate('down'); + }); + + const getReceiptTimes = async () => + (await client<{ receipt_time: number }[]>`SELECT receipt_time FROM mempool_txs`).map( + r => r.receipt_time + ); + + test('mempool receipt time comes from the SNP timestamp header', async () => { + const response = await eventServer.fastifyInstance.inject({ + method: 'POST', + url: '/new_mempool_tx', + payload: [RAW_MEMPOOL_TX], + headers: { [SNP_TIMESTAMP_HEADER]: '1710343460976' }, + }); + assert.equal(response.statusCode, 200); + assert.deepEqual(await getReceiptTimes(), [1710343461]); + }); + + test('mempool receipt time is the current time without an SNP timestamp', async () => { + const before = Math.round(Date.now() / 1000); + const response = await eventServer.fastifyInstance.inject({ + method: 'POST', + url: '/new_mempool_tx', + payload: [RAW_MEMPOOL_TX], + }); + const after = Math.round(Date.now() / 1000); + assert.equal(response.statusCode, 200); + const [receipt_time] = await getReceiptTimes(); + assert.ok(receipt_time >= before && receipt_time <= after); + }); + + test('emits only the mined txs of a canonical block', async () => { + const sender = 'SP3SBQ9PZEMBNBAWTR7FRPE3XK0EFW9JWVX4G80S2'; + await db.update(new TestBlockBuilder({ block_height: 1, index_block_hash: '0x01' }).build()); + await db.updateMempoolTxs({ + mempoolTxs: [ + // Mined in the next block. + testMempoolTx({ + tx_id: '0xaa', + sender_address: sender, + nonce: 0, + receipt_time: 1000, + }), + // Shares the mined tx's nonce slot, so it's pruned but was not mined. + testMempoolTx({ + tx_id: '0xbb', + type_id: DbTxTypeId.ContractCall, + sender_address: sender, + nonce: 0, + fee_rate: 1n, + receipt_time: 1001, + }), + // Unrelated, stays in the mempool. + testMempoolTx({ tx_id: '0xcc', nonce: 0, receipt_time: 1002 }), + ], + }); + + const emitted: { blockTime: number; txs: MinedMempoolTx[] }[] = []; + db.eventEmitter.on('mempoolTxsMined', info => emitted.push(info)); + + await db.update( + new TestBlockBuilder({ + block_height: 2, + index_block_hash: '0x02', + parent_index_block_hash: '0x01', + block_time: 1030, + }) + .addTx({ + tx_id: '0xaa', + sender_address: sender, + nonce: 0, + }) + .build() + ); + assert.deepEqual(emitted, [ + { + blockTime: 1030, + txs: [{ tx_id: '0xaa', type_id: DbTxTypeId.TokenTransfer, receipt_time: 1000 }], + }, + ]); + + // A block whose txs never went through the mempool emits nothing. + await db.update( + new TestBlockBuilder({ + block_height: 3, + index_block_hash: '0x03', + parent_index_block_hash: '0x02', + }) + .addTx({ tx_id: '0xdd', nonce: 5 }) + .build() + ); + assert.equal(emitted.length, 1); + }); +}); diff --git a/tests/snp/snp-ingestion.test.ts b/tests/snp/snp-ingestion.test.ts index a7c9f8f790..76a6e734ee 100644 --- a/tests/snp/snp-ingestion.test.ts +++ b/tests/snp/snp-ingestion.test.ts @@ -118,6 +118,18 @@ describe('SNP integration tests', { concurrency: 1 }, () => { assert.equal(chainTip.block_height, sampleEventsLastBlockHeight); }); + test('mempool receipt times use SNP timestamps', async () => { + // The dump's events were received by SNP between these times, long before this test replayed + // them, so receipt times based on ingestion time would fall outside this window. + const dumpStart = Date.parse('2024-03-13T15:34:10Z') / 1000; + const dumpEnd = Date.parse('2024-03-13T15:37:22Z') / 1000; + const rows = await client<{ receipt_time: number }[]>`SELECT receipt_time FROM mempool_txs`; + assert.ok(rows.length > 0); + for (const { receipt_time } of rows) { + assert.ok(receipt_time >= dumpStart && receipt_time <= dumpEnd, `${receipt_time}`); + } + }); + test('test block API fetch', async () => { const response = await supertest(apiServer.server) .get(`/extended/v1/block/by_height/${sampleEventsLastBlockHeight}`) From 1f3ea75e5650a497324761c9314e8ec958ca968e Mon Sep 17 00:00:00 2001 From: Rafa Cardenas <253999660+rafa-stacks@users.noreply.github.com> Date: Wed, 7 Oct 2026 16:20:37 -0600 Subject: [PATCH 2/2] rbf confirmations --- src/datastore/pg-write-store.ts | 30 ++++++---- src/event-stream/event-server.ts | 8 ++- .../mempool-confirmation-metric.test.ts | 58 ++++++++++++++++++- 3 files changed, 79 insertions(+), 17 deletions(-) diff --git a/src/datastore/pg-write-store.ts b/src/datastore/pg-write-store.ts index 6dab5cfe9c..ce9484223b 100644 --- a/src/datastore/pg-write-store.ts +++ b/src/datastore/pg-write-store.ts @@ -343,11 +343,17 @@ export class PgWriteStore extends PgStore { nonce: d.tx.nonce, sponsor_nonce: d.tx.sponsor_nonce, })); - 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)); + // Look up the mempool receipt data of this block's txs before pruning. Txs that were + // already pruned (e.g. replaced by fee or dropped) are included too, since a miner can + // still confirm them. + if (prunableTxs.length > 0 && !this.isEventReplay) { + minedMempoolTxs = await sql` + SELECT tx_id, type_id, receipt_time + FROM mempool_txs + WHERE tx_id IN ${sql(prunableTxs.map(tx => tx.txId))} + `; + } + await this.pruneMempoolTxs(sql, prunableTxs); } if (isCanonical) { @@ -444,7 +450,7 @@ export class PgWriteStore extends PgStore { } }); if (skippedDuplicateBlock) return; - if (minedMempoolTxs.length > 0 && !this.isEventReplay) { + if (minedMempoolTxs.length > 0) { this.eventEmitter.emit('mempoolTxsMined', { blockTime: data.block.block_time, txs: minedMempoolTxs, @@ -4393,8 +4399,8 @@ export class PgWriteStore extends PgStore { async pruneMempoolTxs( sql: PgSqlClient, transactions: TransactionHeader[] - ): Promise<{ removedTxs: string[]; prunedTxs: MinedMempoolTx[] }> { - if (transactions.length === 0) return { removedTxs: [], prunedTxs: [] }; + ): Promise<{ removedTxs: string[] }> { + if (transactions.length === 0) return { removedTxs: [] }; if (logger.isLevelEnabled('debug')) for (const tx of transactions) logger.debug( @@ -4403,22 +4409,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` + const updateResults = await sql<{ tx_id: string }[]>` 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, type_id, receipt_time + RETURNING tx_id ), count_update AS ( UPDATE chain_tip SET mempool_tx_count = mempool_tx_count - (SELECT COUNT(*) FROM pruned), mempool_updated_at = NOW() ) - SELECT tx_id, type_id, receipt_time FROM pruned + SELECT tx_id FROM pruned `; - return { removedTxs: updateResults.map(r => r.tx_id), prunedTxs: updateResults }; + return { removedTxs: updateResults.map(r => r.tx_id) }; } /** diff --git a/src/event-stream/event-server.ts b/src/event-stream/event-server.ts index 1957ef7052..7c94c7eb38 100644 --- a/src/event-stream/event-server.ts +++ b/src/event-stream/event-server.ts @@ -3,7 +3,7 @@ import * as net from 'net'; import Fastify, { FastifyInstance, FastifyRequest, FastifyServerOptions } from 'fastify'; import PQueue from 'p-queue'; import * as prom from 'prom-client'; -import { BitVec, ChainID, assertNotNullish, getChainIDNetwork } from '../helpers.js'; +import { BitVec, ChainID, I32_MAX, assertNotNullish, getChainIDNetwork } from '../helpers.js'; import { DbEventBase, DbSmartContractEvent, @@ -104,13 +104,15 @@ 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. + * current time for events posted directly by a Stacks node or when the header can't be stored as a + * positive 32-bit integer of seconds. */ 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); + const seconds = Math.round(ms / 1000); + if (Number.isFinite(seconds) && seconds > 0 && seconds <= I32_MAX) return seconds; } return Math.round(Date.now() / 1000); } diff --git a/tests/api/mempool/mempool-confirmation-metric.test.ts b/tests/api/mempool/mempool-confirmation-metric.test.ts index 47bece6bc5..bca61abb22 100644 --- a/tests/api/mempool/mempool-confirmation-metric.test.ts +++ b/tests/api/mempool/mempool-confirmation-metric.test.ts @@ -27,7 +27,8 @@ describe('mempool confirmation time helpers', () => { }); test('getEventReceiptTime falls back to the current time', () => { - for (const header of [undefined, '', 'not-a-date']) { + // Includes values that parse but don't fit a positive 32-bit integer of seconds. + for (const header of [undefined, '', 'not-a-date', '0', '999999999999999999', '1969-12-31']) { const before = Math.round(Date.now() / 1000); const receiptTime = getEventReceiptTime(header); const after = Math.round(Date.now() / 1000); @@ -140,7 +141,7 @@ describe('mempool confirmation time ingestion', () => { }); const emitted: { blockTime: number; txs: MinedMempoolTx[] }[] = []; - db.eventEmitter.on('mempoolTxsMined', info => emitted.push(info)); + db.eventEmitter.on('mempoolTxsMined', info => emitted.push({ ...info, txs: [...info.txs] })); await db.update( new TestBlockBuilder({ @@ -175,4 +176,57 @@ describe('mempool confirmation time ingestion', () => { ); assert.equal(emitted.length, 1); }); + + test('emits a mined tx that was already pruned by replace-by-fee', async () => { + const sender = 'SP3SBQ9PZEMBNBAWTR7FRPE3XK0EFW9JWVX4G80S2'; + await db.update(new TestBlockBuilder({ block_height: 1, index_block_hash: '0x01' }).build()); + await db.updateMempoolTxs({ + mempoolTxs: [ + testMempoolTx({ + tx_id: '0xaa', + sender_address: sender, + nonce: 0, + fee_rate: 1n, + receipt_time: 1000, + }), + ], + }); + // A higher fee tx for the same nonce slot replaces it. + await db.updateMempoolTxs({ + mempoolTxs: [ + testMempoolTx({ + tx_id: '0xbb', + sender_address: sender, + nonce: 0, + fee_rate: 100n, + receipt_time: 1010, + }), + ], + }); + const [replaced] = await client<{ pruned: boolean }[]>` + SELECT pruned FROM mempool_txs WHERE tx_id = '\\xaa' + `; + assert.equal(replaced.pruned, true); + + const emitted: { blockTime: number; txs: MinedMempoolTx[] }[] = []; + db.eventEmitter.on('mempoolTxsMined', info => emitted.push({ ...info, txs: [...info.txs] })); + + // The miner confirms the replaced tx anyway. + await db.update( + new TestBlockBuilder({ + block_height: 2, + index_block_hash: '0x02', + parent_index_block_hash: '0x01', + block_time: 1030, + }) + .addTx({ tx_id: '0xaa', sender_address: sender, nonce: 0 }) + .build() + ); + assert.deepEqual(emitted, [ + { + blockTime: 1030, + txs: [{ tx_id: '0xaa', type_id: DbTxTypeId.TokenTransfer, receipt_time: 1000 }], + }, + ]); + }); });