Skip to content

Commit c5a7a8c

Browse files
bestmike007bestmike007
authored
feat: remove reconcile mempool & debounce stats (#1815)
* feat: debounce reconcile mempool Signed-off-by: bestmike007 <[email protected]> * feat: wait for next mempool reconcile Signed-off-by: bestmike007 <[email protected]> * chore: add comments Signed-off-by: bestmike007 <[email protected]> * chore: reconcile inserted and debounce stat Signed-off-by: bestmike007 <[email protected]> * chore: use CTE to check and update pruned mempool tx Signed-off-by: bestmike007 <[email protected]> * chore: make mempool stat debounce interval configurable Signed-off-by: bestmike007 <[email protected]> * chore: add entry to env file Signed-off-by: bestmike007 <[email protected]> * feat: mempool stat debounce max interval Signed-off-by: bestmike007 <[email protected]> * chore: fix logging Signed-off-by: bestmike007 <[email protected]> * chore: fix getUintEnvOrDefault Signed-off-by: bestmike007 <[email protected]> * chore: revert use CTE to check and update pruned mempool tx Signed-off-by: bestmike007 <[email protected]> * chore: get rid of reconcileMempoolStatus Signed-off-by: bestmike007 <[email protected]> * chore: add tests for getUintEnvOrDefault Signed-off-by: bestmike007 <[email protected]> --------- Signed-off-by: bestmike007 <[email protected]>
1 parent ee0da75 commit c5a7a8c

File tree

5 files changed

+121
-52
lines changed

5 files changed

+121
-52
lines changed

‎.env

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -61,6 +61,11 @@ PG_APPLICATION_NAME=stacks-blockchain-api
6161
# the same.
6262
# STACKS_MEMPOOL_TX_GARBAGE_COLLECTION_THRESHOLD=256
6363

64+
# To avoid running unnecessary mempool stats during transaction influx, we use a debounce mechanism for the process.
65+
# This variable controls the duration it waits until there are no further mempool updates
66+
# MEMPOOL_STATS_DEBOUNCE_INTERVAL=1000
67+
# MEMPOOL_STATS_DEBOUNCE_MAX_INTERVAL=10000
68+
6469
# If specified, an http server providing profiling capability endpoints will be opened on the given port.
6570
# This port should not be publicly exposed.
6671
# STACKS_PROFILER_PORT=9119

‎src/datastore/pg-write-store.ts

Lines changed: 79 additions & 50 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
import { getOrAdd, I32_MAX, getIbdBlockHeight } from '../helpers';
1+
import { getOrAdd, I32_MAX, getIbdBlockHeight, getUintEnvOrDefault } from '../helpers';
22
import {
33
DbBlock,
44
DbTx,
@@ -97,6 +97,14 @@ import { PgServer, getConnectionArgs, getConnectionConfig } from './connection';
9797

9898
const MIGRATIONS_TABLE = 'pgmigrations';
9999
const INSERT_BATCH_SIZE = 500;
100+
const MEMPOOL_STATS_DEBOUNCE_INTERVAL = getUintEnvOrDefault(
101+
'MEMPOOL_STATS_DEBOUNCE_INTERVAL',
102+
1000
103+
);
104+
const MEMPOOL_STATS_DEBOUNCE_MAX_INTERVAL = getUintEnvOrDefault(
105+
'MEMPOOL_STATS_DEBOUNCE_MAX_INTERVAL',
106+
10000
107+
);
100108

101109
class MicroblockGapError extends Error {
102110
constructor(message: string) {
@@ -271,10 +279,7 @@ export class PgWriteStore extends PgStore {
271279
}
272280

273281
if (!this.isEventReplay) {
274-
await this.reconcileMempoolStatus(sql);
275-
276-
const mempoolStats = await this.getMempoolStatsInternal({ sql });
277-
this.eventEmitter.emit('mempoolStatsUpdate', mempoolStats);
282+
this.debounceMempoolStat();
278283
}
279284
if (isCanonical)
280285
await sql`
@@ -664,10 +669,7 @@ export class PgWriteStore extends PgStore {
664669
}
665670

666671
if (!this.isEventReplay) {
667-
await this.reconcileMempoolStatus(sql);
668-
669-
const mempoolStats = await this.getMempoolStatsInternal({ sql });
670-
this.eventEmitter.emit('mempoolStatsUpdate', mempoolStats);
672+
this.debounceMempoolStat();
671673
}
672674
if (currentMicroblockTip.microblock_canonical)
673675
await sql`
@@ -705,42 +707,6 @@ export class PgWriteStore extends PgStore {
705707
}
706708
}
707709

708-
// Find any transactions that are erroneously still marked as both `pending` in the mempool table
709-
// and also confirmed in the mined txs table. Mark these as pruned in the mempool and log warning.
710-
// This must be called _after_ any writes to txs/mempool tables during block and microblock ingestion,
711-
// but _before_ any reads or view refreshes that depend on the mempool table.
712-
// NOTE: this is essentially a work-around for whatever bug is causing the underlying problem.
713-
async reconcileMempoolStatus(sql: PgSqlClient): Promise<void> {
714-
const txsResult = await sql<{ tx_id: string }[]>`
715-
WITH pruned AS (
716-
UPDATE mempool_txs
717-
SET pruned = true
718-
FROM txs
719-
WHERE
720-
mempool_txs.tx_id = txs.tx_id AND
721-
mempool_txs.pruned = false AND
722-
txs.canonical = true AND
723-
txs.microblock_canonical = true AND
724-
txs.status IN ${sql([
725-
DbTxStatus.Success,
726-
DbTxStatus.AbortByResponse,
727-
DbTxStatus.AbortByPostCondition,
728-
])}
729-
RETURNING mempool_txs.tx_id
730-
),
731-
count_update AS (
732-
UPDATE chain_tip SET
733-
mempool_tx_count = mempool_tx_count - (SELECT COUNT(*) FROM pruned),
734-
mempool_updated_at = NOW()
735-
)
736-
SELECT tx_id FROM pruned
737-
`;
738-
if (txsResult.length > 0) {
739-
const txs = txsResult.map(tx => tx.tx_id);
740-
logger.warn(`Reconciled mempool txs as pruned for ${txsResult.length} txs`, { txs });
741-
}
742-
}
743-
744710
async fixBlockZeroData(sql: PgSqlClient, blockOne: DbBlock): Promise<void> {
745711
const tablesUpdates: Record<string, number> = {};
746712
const txsResult = await sql<TxQueryResult[]>`
@@ -1696,21 +1662,84 @@ export class PgWriteStore extends PgStore {
16961662
SELECT tx_id FROM inserted
16971663
`;
16981664
txIds.push(...result.map(r => r.tx_id));
1665+
// The incoming mempool transactions might have already been settled
1666+
// We need to mark them as pruned to avoid inconsistent tx state
1667+
const pruned_tx = await sql<{ tx_id: string }[]>`
1668+
SELECT tx_id
1669+
FROM txs
1670+
WHERE
1671+
tx_id IN ${sql(batch.map(b => b.tx_id))} AND
1672+
canonical = true AND
1673+
microblock_canonical = true`;
1674+
if (pruned_tx.length > 0) {
1675+
await sql<{ tx_id: string }[]>`
1676+
WITH pruned AS (
1677+
UPDATE mempool_txs
1678+
SET pruned = true
1679+
WHERE
1680+
tx_id IN ${sql(pruned_tx.map(t => t.tx_id))} AND
1681+
pruned = false
1682+
RETURNING tx_id
1683+
),
1684+
count_update AS (
1685+
UPDATE chain_tip SET
1686+
mempool_tx_count = mempool_tx_count - (SELECT COUNT(*) FROM pruned),
1687+
mempool_updated_at = NOW()
1688+
)
1689+
SELECT tx_id FROM pruned`;
1690+
}
16991691
}
17001692
return txIds;
17011693
}
17021694

1695+
private _debounceMempoolStat: {
1696+
triggeredAt?: number | null;
1697+
debounce?: NodeJS.Timeout | null;
1698+
running: boolean;
1699+
} = { running: false };
1700+
/**
1701+
* Debounce the mempool stat process in case new transactions pour in.
1702+
*/
1703+
private debounceMempoolStat() {
1704+
if (this._debounceMempoolStat.triggeredAt == null) {
1705+
this._debounceMempoolStat.triggeredAt = Date.now();
1706+
}
1707+
if (this._debounceMempoolStat.running) return;
1708+
const waited = Date.now() - this._debounceMempoolStat.triggeredAt;
1709+
const delay = Math.max(
1710+
0,
1711+
Math.min(MEMPOOL_STATS_DEBOUNCE_MAX_INTERVAL - waited, MEMPOOL_STATS_DEBOUNCE_INTERVAL)
1712+
);
1713+
if (this._debounceMempoolStat.debounce != null) {
1714+
clearTimeout(this._debounceMempoolStat.debounce);
1715+
}
1716+
this._debounceMempoolStat.debounce = setTimeout(async () => {
1717+
this._debounceMempoolStat.running = true;
1718+
this._debounceMempoolStat.triggeredAt = null;
1719+
try {
1720+
const mempoolStats = await this.getMempoolStatsInternal({ sql: this.sql });
1721+
this.eventEmitter.emit('mempoolStatsUpdate', mempoolStats);
1722+
} catch (e) {
1723+
logger.error(e, `failed to run mempool stats update`);
1724+
} finally {
1725+
this._debounceMempoolStat.running = false;
1726+
this._debounceMempoolStat.debounce = null;
1727+
if (this._debounceMempoolStat.triggeredAt != null) {
1728+
this.debounceMempoolStat();
1729+
}
1730+
}
1731+
}, delay);
1732+
}
1733+
17031734
async updateMempoolTxs({ mempoolTxs: txs }: { mempoolTxs: DbMempoolTxRaw[] }): Promise<void> {
17041735
const updatedTxIds: string[] = [];
17051736
await this.sqlWriteTransaction(async sql => {
17061737
const chainTip = await this.getChainTip();
17071738
updatedTxIds.push(...(await this.insertDbMempoolTxs(txs, chainTip, sql)));
1708-
if (!this.isEventReplay) {
1709-
await this.reconcileMempoolStatus(sql);
1710-
const mempoolStats = await this.getMempoolStatsInternal({ sql });
1711-
this.eventEmitter.emit('mempoolStatsUpdate', mempoolStats);
1712-
}
17131739
});
1740+
if (!this.isEventReplay) {
1741+
this.debounceMempoolStat();
1742+
}
17141743
for (const txId of updatedTxIds) {
17151744
await this.notifier?.sendTx({ txId });
17161745
}

‎src/helpers.ts

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -764,3 +764,13 @@ export enum BootContractAddress {
764764
mainnet = 'SP000000000000000000002Q6VF78',
765765
testnet = 'ST000000000000000000002AMW42H',
766766
}
767+
768+
export function getUintEnvOrDefault(envName: string, defaultValue = 0) {
769+
const v = BigInt(process.env[envName] ?? defaultValue);
770+
if (v < 0n) {
771+
throw new Error(
772+
`Expecting ENV ${envName} to be non-negative number but it is configured as ${process.env[envName]}`
773+
);
774+
}
775+
return Number(v);
776+
}

‎src/tests/helpers-tests.ts

Lines changed: 11 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,7 @@ import * as c32check from 'c32check';
33
import { bitcoinToStacksAddress, stacksToBitcoinAddress } from 'stacks-encoding-native-js';
44
import * as c32AddrCache from '../c32-addr-cache';
55
import { ADDR_CACHE_ENV_VAR } from '../c32-addr-cache';
6-
import { isValidBitcoinAddress } from '../helpers';
6+
import { isValidBitcoinAddress, getUintEnvOrDefault } from '../helpers';
77
import { ECPair, getBitcoinAddressFromKey } from '../ec-helpers';
88
import { decodeBtcAddress, poxAddressToBtcAddress } from '@stacks/stacking';
99
import { has0xPrefix } from '@hirosystems/api-toolkit';
@@ -536,3 +536,13 @@ describe('Bitcoin address encoding formats', () => {
536536
expect(randP2TRTestnet).toMatch(/^tb1p/);
537537
});
538538
});
539+
540+
test('getUintEnvOrDefault tests', () => {
541+
const key = 'SOME_UINT_ENV';
542+
process.env[key] = '123';
543+
expect(getUintEnvOrDefault(key)).toBe(123);
544+
process.env[key] = '-123';
545+
expect(() => getUintEnvOrDefault(key)).toThrowError();
546+
process.env[key] = 'ABC';
547+
expect(() => getUintEnvOrDefault(key)).toThrowError();
548+
});

‎src/tests/mempool-tests.ts

Lines changed: 16 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1691,7 +1691,22 @@ describe('mempool tests', () => {
16911691
block: dbBlock2,
16921692
microblocks: [],
16931693
minerRewards: [],
1694-
txs: [],
1694+
txs: [
1695+
{
1696+
tx: dbTx1,
1697+
stxEvents: [],
1698+
stxLockEvents: [],
1699+
ftEvents: [],
1700+
nftEvents: [],
1701+
contractLogEvents: [],
1702+
smartContracts: [],
1703+
names: [],
1704+
namespaces: [],
1705+
pox2Events: [],
1706+
pox3Events: [],
1707+
pox4Events: [],
1708+
},
1709+
],
16951710
});
16961711

16971712
// Verify tx pruned from mempool

0 commit comments

Comments
 (0)