Merge branch 'master' into mononaut/acceleration-viz

backend/src/api/mempool.ts

@@ -10,6 +10,7 @@ import bitcoinClient from './bitcoin/bitcoin-client';
 import bitcoinSecondClient from './bitcoin/bitcoin-second-client';
 import rbfCache from './rbf-cache';
 import accelerationApi, { Acceleration } from './services/acceleration';
+import redisCache from './redis-cache';
 
 class Mempool {
   private inSync: boolean = false;
@@ -88,6 +89,10 @@ class Mempool {
   public async $setMempool(mempoolData: { [txId: string]: MempoolTransactionExtended }) {
     this.mempoolCache = mempoolData;
     let count = 0;
+    const redisTimer = Date.now();
+    if (config.MEMPOOL.CACHE_ENABLED && config.REDIS.ENABLED) {
+      logger.debug(`Migrating ${Object.keys(this.mempoolCache).length} transactions from disk cache to Redis cache`);
+    }
     for (const txid of Object.keys(this.mempoolCache)) {
       if (!this.mempoolCache[txid].sigops || this.mempoolCache[txid].effectiveFeePerVsize == null) {
         this.mempoolCache[txid] = transactionUtils.extendMempoolTransaction(this.mempoolCache[txid]);
@@ -96,6 +101,13 @@ class Mempool {
         this.mempoolCache[txid].order = transactionUtils.txidToOrdering(txid);
       }
       count++;
+      if (config.MEMPOOL.CACHE_ENABLED && config.REDIS.ENABLED) {
+        await redisCache.$addTransaction(this.mempoolCache[txid]);
+      }
+    }
+    if (config.MEMPOOL.CACHE_ENABLED && config.REDIS.ENABLED) {
+      await redisCache.$flushTransactions();
+      logger.debug(`Finished migrating cache transactions in ${((Date.now() - redisTimer) / 1000).toFixed(2)} seconds`);
     }
     if (this.mempoolChangedCallback) {
       this.mempoolChangedCallback(this.mempoolCache, [], [], []);
@@ -140,8 +152,8 @@ class Mempool {
         logger.err('failed to fetch bulk mempool transactions from esplora');
       }
     }
-    return newTransactions;
     logger.info(`Done inserting loaded mempool transactions into local cache`);
+    return newTransactions;
   }
 
   public async $updateMemPoolInfo() {
@@ -173,7 +185,7 @@ class Mempool {
     return txTimes;
   }
 
-  public async $updateMempool(transactions: string[]): Promise<void> {
+  public async $updateMempool(transactions: string[], pollRate: number): Promise<void> {
     logger.debug(`Updating mempool...`);
 
     // warn if this run stalls the main loop for more than 2 minutes
@@ -210,6 +222,11 @@ class Mempool {
       logger.info(`Missing ${transactions.length - currentMempoolSize} mempool transactions, attempting to reload in bulk from esplora`);
       try {
         newTransactions = await this.$reloadMempool(transactions.length);
+        if (config.REDIS.ENABLED) {
+          for (const tx of newTransactions) {
+            await redisCache.$addTransaction(tx);
+          }
+        }
         loaded = true;
       } catch (e) {
         logger.err('failed to load mempool in bulk from esplora, falling back to fetching individual transactions');
@@ -232,6 +249,10 @@ class Mempool {
             }
             hasChange = true;
             newTransactions.push(transaction);
+
+            if (config.REDIS.ENABLED) {
+              await redisCache.$addTransaction(transaction);
+            }
           } catch (e: any) {
             if (config.MEMPOOL.BACKEND === 'esplora' && e.response?.status === 404) {
               this.missingTxCount++;
@@ -240,7 +261,7 @@ class Mempool {
           }
         }
 
-        if (Date.now() - intervalTimer > 5_000) {
+        if (Date.now() - intervalTimer > Math.max(pollRate * 2, 5_000)) {
           if (this.inSync) {
             // Break and restart mempool loop if we spend too much time processing
             // new transactions that may lead to falling behind on block height
@@ -252,7 +273,7 @@ class Mempool {
             if (Math.floor(progress) < 100) {
               loadingIndicators.setProgress('mempool', progress);
             }
-            intervalTimer = Date.now()
+            intervalTimer = Date.now();
           }
         }
       }
@@ -325,6 +346,13 @@ class Mempool {
       loadingIndicators.setProgress('mempool', 100);
     }
 
+    // Update Redis cache
+    if (config.REDIS.ENABLED) {
+      await redisCache.$flushTransactions();
+      await redisCache.$removeTransactions(deletedTransactions.map(tx => tx.txid));
+      await rbfCache.updateCache();
+    }
+
     const end = new Date().getTime();
     const time = end - start;
     logger.debug(`Mempool updated in ${time / 1000} seconds. New size: ${Object.keys(this.mempoolCache).length} (${diff > 0 ? '+' + diff : diff})`);