Merge branch 'master' into simon/load-more-mempool-txs

backend/src/__fixtures__/mempool-config.template.json

@@ -121,5 +121,11 @@
   },
   "CLIGHTNING": {
     "SOCKET": "__CLIGHTNING_SOCKET__"
+  },
+  "REPLICATION": {
+    "ENABLED": false,
+    "AUDIT": false,
+    "AUDIT_START_HEIGHT": 774000,
+    "SERVERS": []
   }
 }

backend/src/__tests__/config.test.ts

@@ -120,6 +120,13 @@ describe('Mempool Backend Config', () => {
         GEOLITE2_ASN: '/usr/local/share/GeoIP/GeoLite2-ASN.mmdb',
         GEOIP2_ISP: '/usr/local/share/GeoIP/GeoIP2-ISP.mmdb'
       });
+
+      expect(config.REPLICATION).toStrictEqual({
+        ENABLED: false,
+        AUDIT: false,
+        AUDIT_START_HEIGHT: 774000,
+        SERVERS: []
+      });
     });
   });
 

backend/src/api/audit.ts

@@ -1,12 +1,12 @@
 import config from '../config';
 import logger from '../logger';
-import { TransactionExtended, MempoolBlockWithTransactions } from '../mempool.interfaces';
+import { MempoolTransactionExtended, MempoolBlockWithTransactions } from '../mempool.interfaces';
 import rbfCache from './rbf-cache';
 
 const PROPAGATION_MARGIN = 180; // in seconds, time since a transaction is first seen after which it is assumed to have propagated to all miners
 
 class Audit {
-  auditBlock(transactions: TransactionExtended[], projectedBlocks: MempoolBlockWithTransactions[], mempool: { [txId: string]: TransactionExtended })
+  auditBlock(transactions: MempoolTransactionExtended[], projectedBlocks: MempoolBlockWithTransactions[], mempool: { [txId: string]: MempoolTransactionExtended })
    : { censored: string[], added: string[], fresh: string[], sigop: string[], fullrbf: string[], score: number, similarity: number } {
     if (!projectedBlocks?.[0]?.transactionIds || !mempool) {
       return { censored: [], added: [], fresh: [], sigop: [], fullrbf: [], score: 0, similarity: 1 };
@@ -14,7 +14,7 @@ class Audit {
 
     const matches: string[] = []; // present in both mined block and template
     const added: string[] = []; // present in mined block, not in template
-    const fresh: string[] = []; // missing, but firstSeen within PROPAGATION_MARGIN
+    const fresh: string[] = []; // missing, but firstSeen or lastBoosted within PROPAGATION_MARGIN
     const fullrbf: string[] = []; // either missing or present, and part of a fullrbf replacement
     const isCensored = {}; // missing, without excuse
     const isDisplaced = {};
@@ -36,10 +36,13 @@ class Audit {
     // look for transactions that were expected in the template, but missing from the mined block
     for (const txid of projectedBlocks[0].transactionIds) {
       if (!inBlock[txid]) {
-        // tx is recent, may have reached the miner too late for inclusion
         if (rbfCache.isFullRbf(txid)) {
           fullrbf.push(txid);
         } else if (mempool[txid]?.firstSeen != null && (now - (mempool[txid]?.firstSeen || 0)) <= PROPAGATION_MARGIN) {
+          // tx is recent, may have reached the miner too late for inclusion
+          fresh.push(txid);
+        } else if (mempool[txid]?.lastBoosted != null && (now - (mempool[txid]?.lastBoosted || 0)) <= PROPAGATION_MARGIN) {
+          // tx was recently cpfp'd, miner may not have the latest effective rate
           fresh.push(txid);
         } else {
           isCensored[txid] = true;

backend/src/api/mempool-blocks.ts

@@ -457,6 +457,7 @@ class MempoolBlocks {
               };
               if (matched) {
                 descendants.push(relative);
+                mempoolTx.lastBoosted = Math.max(mempoolTx.lastBoosted || 0, mempool[txid].firstSeen || 0);
               } else {
                 ancestors.push(relative);
               }

backend/src/api/websocket-handler.ts

@@ -236,7 +236,7 @@ class WebsocketHandler {
           }
 
           if (parsedMessage.action === 'init') {
-            if (!this.socketData['blocks']?.length || !this.socketData['da']) {
+            if (!this.socketData['blocks']?.length || !this.socketData['da'] || !this.socketData['backendInfo'] || !this.socketData['conversions']) {
               this.updateSocketData();
             }
             if (!this.socketData['blocks']?.length) {
@@ -419,7 +419,7 @@ class WebsocketHandler {
     memPool.addToSpendMap(newTransactions);
     const recommendedFees = feeApi.getRecommendedFee();
 
-    const latestTransactions = newTransactions.slice(0, 6).map((tx) => Common.stripTransaction(tx));
+    const latestTransactions = memPool.getLatestTransactions();
 
     // update init data
     const socketDataFields = {

backend/src/config.ts

@@ -132,6 +132,12 @@ interface IConfig {
     GEOLITE2_ASN: string;
     GEOIP2_ISP: string;
   },
+  REPLICATION: {
+    ENABLED: boolean;
+    AUDIT: boolean;
+    AUDIT_START_HEIGHT: number;
+    SERVERS: string[];
+  }
 }
 
 const defaults: IConfig = {
@@ -264,6 +270,12 @@ const defaults: IConfig = {
     'GEOLITE2_ASN': '/usr/local/share/GeoIP/GeoLite2-ASN.mmdb',
     'GEOIP2_ISP': '/usr/local/share/GeoIP/GeoIP2-ISP.mmdb'
   },
+  'REPLICATION': {
+    'ENABLED': false,
+    'AUDIT': false,
+    'AUDIT_START_HEIGHT': 774000,
+    'SERVERS': [],
+  }
 };
 
 class Config implements IConfig {
@@ -283,6 +295,7 @@ class Config implements IConfig {
   PRICE_DATA_SERVER: IConfig['PRICE_DATA_SERVER'];
   EXTERNAL_DATA_SERVER: IConfig['EXTERNAL_DATA_SERVER'];
   MAXMIND: IConfig['MAXMIND'];
+  REPLICATION: IConfig['REPLICATION'];
 
   constructor() {
     const configs = this.merge(configFromFile, defaults);
@@ -302,6 +315,7 @@ class Config implements IConfig {
     this.PRICE_DATA_SERVER = configs.PRICE_DATA_SERVER;
     this.EXTERNAL_DATA_SERVER = configs.EXTERNAL_DATA_SERVER;
     this.MAXMIND = configs.MAXMIND;
+    this.REPLICATION = configs.REPLICATION;
   }
 
   merge = (...objects: object[]): IConfig => {

backend/src/indexer.ts

@@ -7,6 +7,7 @@ import bitcoinClient from './api/bitcoin/bitcoin-client';
 import priceUpdater from './tasks/price-updater';
 import PricesRepository from './repositories/PricesRepository';
 import config from './config';
+import auditReplicator from './replication/AuditReplication';
 
 export interface CoreIndex {
   name: string;
@@ -136,6 +137,7 @@ class Indexer {
       await blocks.$generateBlocksSummariesDatabase();
       await blocks.$generateCPFPDatabase();
       await blocks.$generateAuditStats();
+      await auditReplicator.$sync();
     } catch (e) {
       this.indexerRunning = false;
       logger.err(`Indexer failed, trying again in 10 seconds. Reason: ` + (e instanceof Error ? e.message : e));

backend/src/mempool.interfaces.ts

@@ -100,6 +100,7 @@ export interface MempoolTransactionExtended extends TransactionExtended {
   adjustedVsize: number;
   adjustedFeePerVsize: number;
   inputs?: number[];
+  lastBoosted?: number;
 }
 
 export interface AuditTransaction {
@@ -236,6 +237,15 @@ export interface BlockSummary {
   transactions: TransactionStripped[];
 }
 
+export interface AuditSummary extends BlockAudit {
+  timestamp?: number,
+  size?: number,
+  weight?: number,
+  tx_count?: number,
+  transactions: TransactionStripped[];
+  template?: TransactionStripped[];
+}
+
 export interface BlockPrice {
   height: number;
   priceId: number;

backend/src/replication/AuditReplication.ts

@@ -0,0 +1,134 @@
+import DB from '../database';
+import logger from '../logger';
+import { AuditSummary } from '../mempool.interfaces';
+import blocksAuditsRepository from '../repositories/BlocksAuditsRepository';
+import blocksSummariesRepository from '../repositories/BlocksSummariesRepository';
+import { $sync } from './replicator';
+import config from '../config';
+import { Common } from '../api/common';
+import blocks from '../api/blocks';
+
+const BATCH_SIZE = 16;
+
+/**
+ * Syncs missing block template and audit data from trusted servers
+ */
+class AuditReplication {
+  inProgress: boolean = false;
+  skip: Set<string> = new Set();
+
+  public async $sync(): Promise<void> {
+    if (!config.REPLICATION.ENABLED || !config.REPLICATION.AUDIT) {
+      // replication not enabled
+      return;
+    }
+    if (this.inProgress) {
+      logger.info(`AuditReplication sync already in progress`, 'Replication');
+      return;
+    }
+    this.inProgress = true;
+
+    const missingAudits = await this.$getMissingAuditBlocks();
+
+    logger.debug(`Fetching missing audit data for ${missingAudits.length} blocks from trusted servers`, 'Replication');
+    
+    let totalSynced = 0;
+    let totalMissed = 0;
+    let loggerTimer = Date.now();
+    // process missing audits in batches of 
+    for (let i = 0; i < missingAudits.length; i += BATCH_SIZE) {
+      const slice = missingAudits.slice(i, i + BATCH_SIZE);
+      const results = await Promise.all(slice.map(hash => this.$syncAudit(hash)));
+      const synced = results.reduce((total, status) => status ? total + 1 : total, 0);
+      totalSynced += synced;
+      totalMissed += (slice.length - synced);
+      if (Date.now() - loggerTimer > 10000) {
+        loggerTimer = Date.now();
+        logger.info(`Found ${totalSynced} / ${totalSynced + totalMissed} of ${missingAudits.length} missing audits`, 'Replication');
+      }
+      await Common.sleep$(1000);
+    }
+
+    logger.debug(`Fetched ${totalSynced} audits, ${totalMissed} still missing`, 'Replication');
+
+    this.inProgress = false;
+  }
+
+  private async $syncAudit(hash: string): Promise<boolean> {
+    if (this.skip.has(hash)) {
+      // we already know none of our trusted servers have this audit
+      return false;
+    }
+
+    let success = false;
+    // start with a random server so load is uniformly spread
+    const syncResult = await $sync(`/api/v1/block/${hash}/audit-summary`);
+    if (syncResult) {
+      if (syncResult.data?.template?.length) {
+        await this.$saveAuditData(hash, syncResult.data);
+        logger.info(`Imported audit data from ${syncResult.server} for block ${syncResult.data.height} (${hash})`);
+        success = true;
+      }
+      if (!syncResult.data && !syncResult.exists) {
+        this.skip.add(hash);
+      }
+    }
+
+    return success;
+  }
+
+  private async $getMissingAuditBlocks(): Promise<string[]> {
+    try {
+      const startHeight = config.REPLICATION.AUDIT_START_HEIGHT || 0;
+      const [rows]: any[] = await DB.query(`
+        SELECT auditable.hash, auditable.height
+        FROM (
+          SELECT hash, height
+          FROM blocks
+          WHERE height >= ?
+        ) AS auditable
+        LEFT JOIN blocks_audits ON auditable.hash = blocks_audits.hash
+        WHERE blocks_audits.hash IS NULL
+        ORDER BY auditable.height DESC
+      `, [startHeight]);
+      return rows.map(row => row.hash);
+    } catch (e: any) {
+      logger.err(`Cannot fetch missing audit blocks from db. Reason: ` + (e instanceof Error ? e.message : e));
+      throw e;
+    }
+  }
+
+  private async $saveAuditData(blockHash: string, auditSummary: AuditSummary): Promise<void> {
+    // save audit & template to DB
+    await blocksSummariesRepository.$saveTemplate({
+      height: auditSummary.height,
+      template: {
+        id: blockHash,
+        transactions: auditSummary.template || []
+      }
+    });
+    await blocksAuditsRepository.$saveAudit({
+      hash: blockHash,
+      height: auditSummary.height,
+      time: auditSummary.timestamp || auditSummary.time,
+      missingTxs: auditSummary.missingTxs || [],
+      addedTxs: auditSummary.addedTxs || [],
+      freshTxs: auditSummary.freshTxs || [],
+      sigopTxs: auditSummary.sigopTxs || [],
+      fullrbfTxs: auditSummary.fullrbfTxs || [],
+      matchRate: auditSummary.matchRate,
+      expectedFees: auditSummary.expectedFees,
+      expectedWeight: auditSummary.expectedWeight,
+    });
+    // add missing data to cached blocks
+    const cachedBlock = blocks.getBlocks().find(block => block.id === blockHash);
+    if (cachedBlock) {
+      cachedBlock.extras.matchRate = auditSummary.matchRate;
+      cachedBlock.extras.expectedFees = auditSummary.expectedFees || null;
+      cachedBlock.extras.expectedWeight = auditSummary.expectedWeight || null;
+    }
+  }
+}
+
+export default new AuditReplication();
+

backend/src/replication/replicator.ts

@@ -0,0 +1,70 @@
+import config from '../config';
+import backendInfo from '../api/backend-info';
+import axios, { AxiosResponse } from 'axios';
+import { SocksProxyAgent } from 'socks-proxy-agent';
+import * as https from 'https';
+
+export async function $sync(path): Promise<{ data?: any, exists: boolean, server?: string }> {
+  // start with a random server so load is uniformly spread
+  let allMissing = true;
+  const offset = Math.floor(Math.random() * config.REPLICATION.SERVERS.length);
+  for (let i = 0; i < config.REPLICATION.SERVERS.length; i++) {
+    const server = config.REPLICATION.SERVERS[(i + offset) % config.REPLICATION.SERVERS.length];
+    // don't query ourself
+    if (server === backendInfo.getBackendInfo().hostname) {
+      continue;
+    }
+    
+    try {
+      const result = await query(`https://${server}${path}`);
+      if (result) {
+        return { data: result, exists: true, server };
+      }
+    } catch (e: any) {
+      if (e?.response?.status === 404) {
+        // this server is also missing this data
+      } else {
+        // something else went wrong
+        allMissing = false;
+      }
+    }
+  }
+
+  return { exists: !allMissing };
+}
+
+export async function query(path): Promise<object> {
+  type axiosOptions = {
+    headers: {
+      'User-Agent': string
+    };
+    timeout: number;
+    httpsAgent?: https.Agent;
+  };
+  const axiosOptions: axiosOptions = {
+    headers: {
+      'User-Agent': (config.MEMPOOL.USER_AGENT === 'mempool') ? `mempool/v${backendInfo.getBackendInfo().version}` : `${config.MEMPOOL.USER_AGENT}`
+    },
+    timeout: config.SOCKS5PROXY.ENABLED ? 30000 : 10000
+  };
+
+  if (config.SOCKS5PROXY.ENABLED) {
+    const socksOptions = {
+      agentOptions: {
+        keepAlive: true,
+      },
+      hostname: config.SOCKS5PROXY.HOST,
+      port: config.SOCKS5PROXY.PORT,
+      username: config.SOCKS5PROXY.USERNAME || 'circuit0',
+      password: config.SOCKS5PROXY.PASSWORD,
+    };
+
+    axiosOptions.httpsAgent = new SocksProxyAgent(socksOptions);
+  }
+
+  const data: AxiosResponse = await axios.get(path, axiosOptions);
+  if (data.statusText === 'error' || !data.data) {
+    throw new Error(`${data.status}`);
+  }
+  return data.data;
+}

backend/src/tasks/price-updater.ts

@@ -153,6 +153,7 @@ class PriceUpdater {
       try {
         const p = 60 * 60 * 1000; // milliseconds in an hour
         const nowRounded = new Date(Math.round(new Date().getTime() / p) * p); // https://stackoverflow.com/a/28037042
+        this.latestPrices.time = nowRounded.getTime() / 1000;
         await PricesRepository.$savePrices(nowRounded.getTime() / 1000, this.latestPrices);
       } catch (e) {
         this.lastRun = previousRun + 5 * 60;