Merge pull request #6538 from mempool/mononaut/refactor-redis

refactor redis cache

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();
       }
 

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();
     }
 

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<string>();
+  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<void> {
     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<void> {
+  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<void> {
+    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<void> {
+    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<void> {
+    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<string>): 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<MempoolTransactionExtended>('mempool:tx:*');
+      const mempoolList = validTxids?.size
+        ? await this.loadKeys<MempoolTransactionExtended>('mempool:tx:*', Array.from(validTxids))
+        : await this.scanKeys<MempoolTransactionExtended>('mempool:tx:*');
       for (const tx of mempoolList) {
         mempool[tx.key] = tx.value;
       }
@@ -350,14 +413,14 @@ class RedisCache {
   }
 
   /** @asyncUnsafe */
-  async $loadCache(): Promise<void> {
+  async $loadCache(validTxids?: Set<string>): Promise<void> {
     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<T>(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<void> => {
+      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;
   }

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<string> | 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);
       }
     }