From fe0806a626ddc21017b432e140d73fa5a65fbec3 Mon Sep 17 00:00:00 2001 From: mononaut Date: Tue, 2 Jun 2026 01:23:13 +0000 Subject: [PATCH] refactor redis --- backend/src/api/blocks.ts | 3 +- backend/src/api/mempool.ts | 3 +- backend/src/api/redis-cache.ts | 124 ++++++++++++++++++++++++++++----- backend/src/index.ts | 8 ++- 4 files changed, 116 insertions(+), 22 deletions(-) diff --git a/backend/src/api/blocks.ts b/backend/src/api/blocks.ts index 613b77932..b8da3cf80 100644 --- a/backend/src/api/blocks.ts +++ b/backend/src/api/blocks.ts @@ -520,6 +520,7 @@ class Blocks { delete _memPool[txId]; rbfCache.mined(txId); } + redisCache.queueTransactionsForRemoval(txIds); let candidates; let transactionIds: string[]; @@ -1288,7 +1289,7 @@ class Blocks { if (config.REDIS.ENABLED) { await redisCache.$updateBlocks(this.blocks); await redisCache.$updateBlockSummaries(this.blockSummaries); - await redisCache.$removeTransactions(txIds); + await redisCache.$removeTransactions(); await rbfCache.updateCache(); } diff --git a/backend/src/api/mempool.ts b/backend/src/api/mempool.ts index 1bd919c6c..7eb18cc54 100644 --- a/backend/src/api/mempool.ts +++ b/backend/src/api/mempool.ts @@ -382,6 +382,7 @@ class Mempool { for (const tx of deletedTransactions) { delete this.mempoolCache[tx.txid]; } + redisCache.queueTransactionsForRemoval(deletedTransactions.map(tx => tx.txid)); } const candidates = await this.getNextCandidates(minFeeMempool, minFeeTip, deletedTransactions); @@ -428,7 +429,7 @@ class Mempool { // Update Redis cache if (config.REDIS.ENABLED) { await redisCache.$flushTransactions(); - await redisCache.$removeTransactions(deletedTransactions.map(tx => tx.txid)); + await redisCache.$removeTransactions(); await rbfCache.updateCache(); } diff --git a/backend/src/api/redis-cache.ts b/backend/src/api/redis-cache.ts index 80a947c12..5858603c7 100644 --- a/backend/src/api/redis-cache.ts +++ b/backend/src/api/redis-cache.ts @@ -23,11 +23,14 @@ class RedisCache { private pauseFlush: boolean = false; private cacheQueue: MempoolTransactionExtended[] = []; - private removeQueue: string[] = []; + private removeQueue = new Set(); + private removeQueueFlushInProgress: boolean = false; private rbfCacheQueue: { type: string, txid: string, value: any }[] = []; private rbfRemoveQueue: { type: string, txid: string }[] = []; private txFlushLimit: number = 10000; private ignoreBlocksCache = false; + private reconciliationInProgress: boolean = false; + private reconciliationCursor: string = '0'; constructor() { if (config.REDIS.ENABLED) { @@ -39,6 +42,7 @@ class RedisCache { }; void this.$ensureConnected(); setInterval(() => { void this.$ensureConnected(); }, 10000); + setInterval(() => { void this.$reconcileMempoolTransactions(); }, 30000); } } @@ -92,7 +96,7 @@ class RedisCache { private async $onConnected(): Promise { await this.$flushTransactions(); - await this.$removeTransactions([]); + await this.$removeTransactions(); await this.$flushRbfQueues(); } @@ -134,6 +138,7 @@ class RedisCache { if (!config.REDIS.ENABLED) { return; } + this.removeQueue.delete(tx.txid); this.cacheQueue.push(tx); if (this.cacheQueue.length >= this.txFlushLimit) { if (!this.pauseFlush) { @@ -182,32 +187,88 @@ class RedisCache { } } - /** @asyncSafe */ - async $removeTransactions(transactions: string[]): Promise { + queueTransactionsForRemoval(transactions: string[]): void { if (!config.REDIS.ENABLED) { return; } - const toRemove = this.removeQueue.concat(transactions); - this.removeQueue = []; - let failed: string[] = []; - let numRemoved = 0; - if (this.connected) { + + for (const txid of transactions) { + this.removeQueue.add(txid); + } + } + + /** @asyncSafe */ + async $removeTransactions(transactions: string[] = []): Promise { + if (!config.REDIS.ENABLED) { + return; + } + for (const txid of transactions) { + this.removeQueue.add(txid); + } + await this.$flushQueuedMempoolTxRemovals(); + } + + // incrementally reconcile the redis cache with the in-memory mempool + // by scanning for cached txs no longer in the mempool and marking for deletion + // each invocation scans 1000 keys, cursor loops back to the start after completing a full scan + /** @asyncSafe */ + private async $reconcileMempoolTransactions(): Promise { + if (!config.REDIS.ENABLED || !this.connected || this.reconciliationInProgress || !memPool.isInSync()) { + return; + } + + this.reconciliationInProgress = true; + try { + const result = await this.client.scan(this.reconciliationCursor, { + MATCH: 'mempool:tx:*', + COUNT: 1000 + }); + const mempool = memPool.getMempool(); + let staleCount = 0; + this.reconciliationCursor = result.cursor.toString(); + + for (const key of result.keys) { + const txid = key.slice('mempool:tx:'.length); + if (!mempool[txid]) { + this.removeQueue.add(txid); + staleCount++; + } + } + + if (staleCount) { + logger.debug(`Removing ${staleCount} stale transactions from the redis cache`); + void this.$removeTransactions(); + } + } catch (e) { + logger.warn(`Failed to reconcile Redis mempool cache: ${e instanceof Error ? e.message : e}`); + } finally { + this.reconciliationInProgress = false; + } + } + + /** @asyncSafe */ + private async $flushQueuedMempoolTxRemovals(): Promise { + if (!this.connected || !this.removeQueue.size || this.removeQueueFlushInProgress) { + return; + } + + this.removeQueueFlushInProgress = true; + const toRemove = Array.from(this.removeQueue); + this.removeQueue.clear(); + try { const sliceLength = config.REDIS.BATCH_QUERY_BASE_SIZE; for (let i = 0; i < Math.ceil(toRemove.length / sliceLength); i++) { const slice = toRemove.slice(i * sliceLength, (i + 1) * sliceLength); try { await this.client.unlink(slice.map(txid => `mempool:tx:${txid}`)); - numRemoved+= sliceLength; logger.debug(`Deleted ${slice.length} transactions from the Redis cache`); } catch (e) { logger.warn(`Failed to remove ${slice.length} transactions from Redis cache: ${e instanceof Error ? e.message : e}`); - failed = failed.concat(slice); + this.queueTransactionsForRemoval(slice); } } - // concat instead of replace, in case more txs have been added in the meantime - this.removeQueue = this.removeQueue.concat(failed); - } else { - this.removeQueue = this.removeQueue.concat(toRemove); + } finally { + this.removeQueueFlushInProgress = false; } } @@ -308,7 +369,7 @@ class RedisCache { } /** @asyncSafe */ - async $getMempool(): Promise<{ [txid: string]: MempoolTransactionExtended }> { + async $getMempool(validTxids?: Set): Promise<{ [txid: string]: MempoolTransactionExtended }> { if (!config.REDIS.ENABLED) { return {}; } @@ -319,7 +380,9 @@ class RedisCache { const start = Date.now(); const mempool = {}; try { - const mempoolList = await this.scanKeys('mempool:tx:*'); + const mempoolList = validTxids?.size + ? await this.loadKeys('mempool:tx:*', Array.from(validTxids)) + : await this.scanKeys('mempool:tx:*'); for (const tx of mempoolList) { mempool[tx.key] = tx.value; } @@ -350,14 +413,14 @@ class RedisCache { } /** @asyncUnsafe */ - async $loadCache(): Promise { + async $loadCache(validTxids?: Set): Promise { if (!config.REDIS.ENABLED) { return; } logger.info('Restoring mempool and blocks data from Redis cache'); // Load mempool - const loadedMempool = await this.$getMempool(); + const loadedMempool = await this.$getMempool(validTxids); this.inflateLoadedTxs(loadedMempool); // Load rbf data const rbfTxs = await this.$getRbfEntries('tx'); @@ -432,6 +495,29 @@ class RedisCache { return result; } + /** @asyncUnsafe */ + private async loadKeys(pattern, keys: string[]): Promise<{ key: string, value: T }[]> { + const prefix = pattern.slice(0, -1); + const result: { key: string, value: T }[] = []; + let count = 0; + /** @asyncUnsafe */ + const processValues = async (slice: string[]): Promise => { + const values = await this.client.MGET(slice.map(key => `${prefix}${key}`)); + for (let i = 0; i < values.length; i++) { + if (values[i]) { + result.push({ key: slice[i], value: JSON.parse(values[i]) }); + count++; + } + } + logger.info(`loaded ${count} entries from Redis cache`); + }; + for (let i = 0; i < Math.ceil(keys.length / 10000); i++) { + const slice = keys.slice(i * 10000, (i + 1) * 10000); + await processValues(slice); + } + return result; + } + public setIgnoreBlocksCache(): void { this.ignoreBlocksCache = true; } diff --git a/backend/src/index.ts b/backend/src/index.ts index 2e7e9a13d..3b2296091 100644 --- a/backend/src/index.ts +++ b/backend/src/index.ts @@ -177,8 +177,14 @@ class Server { if (config.MEMPOOL.CACHE_ENABLED) { await diskCache.$loadMempoolCache(); } else if (config.REDIS.ENABLED) { + let currentMempoolTxids: Set | undefined; + try { + currentMempoolTxids = new Set(await bitcoinApi.$getRawMempool()); + } catch (e) { + logger.warn(`Failed to fetch raw mempool before loading Redis cache. Reason: ${e instanceof Error ? e.message : e}`); + } /** @asyncUnsafe */ - await redisCache.$loadCache(); + await redisCache.$loadCache(currentMempoolTxids); } }