Merge pull request #4670 from mempool/mononaut/optimize-mempool-websocket

Reduce the network size of mempool block websocket updates

backend/src/api/mempool-blocks.ts

@@ -1,6 +1,6 @@
 import { GbtGenerator, GbtResult, ThreadTransaction as RustThreadTransaction, ThreadAcceleration as RustThreadAcceleration } from 'rust-gbt';
 import logger from '../logger';
-import { MempoolBlock, MempoolTransactionExtended, MempoolBlockWithTransactions, MempoolBlockDelta, Ancestor, CompactThreadTransaction, EffectiveFeeStats, PoolTag, TransactionClassified } from '../mempool.interfaces';
+import { MempoolBlock, MempoolTransactionExtended, MempoolBlockWithTransactions, MempoolBlockDelta, Ancestor, CompactThreadTransaction, EffectiveFeeStats, PoolTag, TransactionClassified, TransactionCompressed, MempoolDeltaChange } from '../mempool.interfaces';
 import { Common, OnlineFeeStatsCalculator } from './common';
 import config from '../config';
 import { Worker } from 'worker_threads';
@@ -171,7 +171,7 @@ class MempoolBlocks {
     for (let i = 0; i < Math.max(mempoolBlocks.length, prevBlocks.length); i++) {
       let added: TransactionClassified[] = [];
       let removed: string[] = [];
-      const changed: { txid: string, rate: number | undefined, acc: boolean | undefined }[] = [];
+      const changed: TransactionClassified[] = [];
       if (mempoolBlocks[i] && !prevBlocks[i]) {
         added = mempoolBlocks[i].transactions;
       } else if (!mempoolBlocks[i] && prevBlocks[i]) {
@@ -194,14 +194,14 @@ class MempoolBlocks {
           if (!prevIds[tx.txid]) {
             added.push(tx);
           } else if (tx.rate !== prevIds[tx.txid].rate || tx.acc !== prevIds[tx.txid].acc) {
-            changed.push({ txid: tx.txid, rate: tx.rate, acc: tx.acc });
+            changed.push(tx);
           }
         });
       }
       mempoolBlockDeltas.push({
-        added,
+        added: added.map(this.compressTx),
         removed,
-        changed,
+        changed: changed.map(this.compressDeltaChange),
       });
     }
     return mempoolBlockDeltas;
@@ -691,6 +691,38 @@ class MempoolBlocks {
     });
     return { blocks: convertedBlocks, blockWeights, rates: convertedRates, clusters: convertedClusters, overflow: convertedOverflow };
   }
+
+  public compressTx(tx: TransactionClassified): TransactionCompressed {
+    if (tx.acc) {
+      return [
+        tx.txid,
+        tx.fee,
+        tx.vsize,
+        tx.value,
+        Math.round((tx.rate || (tx.fee / tx.vsize)) * 100) / 100,
+        tx.flags,
+        1
+      ];
+    } else {
+      return [
+        tx.txid,
+        tx.fee,
+        tx.vsize,
+        tx.value,
+        Math.round((tx.rate || (tx.fee / tx.vsize)) * 100) / 100,
+        tx.flags,
+      ];
+    }
+  }
+
+  public compressDeltaChange(tx: TransactionClassified): MempoolDeltaChange {
+    return [
+      tx.txid,
+      Math.round((tx.rate || (tx.fee / tx.vsize)) * 100) / 100,
+      tx.flags,
+      tx.acc ? 1 : 0,
+    ];
+  }
 }
 
 export default new MempoolBlocks();

backend/src/api/websocket-handler.ts

@@ -259,7 +259,7 @@ class WebsocketHandler {
               const mBlocksWithTransactions = mempoolBlocks.getMempoolBlocksWithTransactions();
               response['projected-block-transactions'] = JSON.stringify({
                 index: index,
-                blockTransactions: mBlocksWithTransactions[index]?.transactions || [],
+                blockTransactions: (mBlocksWithTransactions[index]?.transactions || []).map(mempoolBlocks.compressTx),
               });
             } else {
               client['track-mempool-block'] = null;
@@ -999,7 +999,7 @@ class WebsocketHandler {
           if (mBlockDeltas[index].added.length > (mBlocksWithTransactions[index]?.transactions.length / 2)) {
             response['projected-block-transactions'] = getCachedResponse(`projected-block-transactions-full-${index}`, {
               index: index,
-              blockTransactions: mBlocksWithTransactions[index].transactions,
+              blockTransactions: mBlocksWithTransactions[index].transactions.map(mempoolBlocks.compressTx),
             });
           } else {
             response['projected-block-transactions'] = getCachedResponse(`projected-block-transactions-delta-${index}`, {

backend/src/mempool.interfaces.ts

@@ -65,9 +65,9 @@ export interface MempoolBlockWithTransactions extends MempoolBlock {
 }
 
 export interface MempoolBlockDelta {
-  added: TransactionClassified[];
+  added: TransactionCompressed[];
   removed: string[];
-  changed: { txid: string, rate: number | undefined, flags?: number }[];
+  changed: MempoolDeltaChange[];
 }
 
 interface VinStrippedToScriptsig {
@@ -196,6 +196,11 @@ export interface TransactionClassified extends TransactionStripped {
   flags: number;
 }
 
+// [txid, fee, vsize, value, rate, flags, acceleration?]
+export type TransactionCompressed = [string, number, number, number, number, number, 1?];
+// [txid, rate, flags, acceleration?]
+export type MempoolDeltaChange = [string, number, number, (1|0)];
+
 // binary flags for transaction classification
 export const TransactionFlags = {
   // features

frontend/src/app/components/mempool-block-overview/mempool-block-overview.component.ts

@@ -100,7 +100,7 @@ export class MempoolBlockOverviewComponent implements OnInit, OnDestroy, OnChang
           const inOldBlock = {};
           const inNewBlock = {};
           const added: TransactionStripped[] = [];
-          const changed: { txid: string, rate: number | undefined, acc: boolean | undefined }[] = [];
+          const changed: { txid: string, rate: number | undefined, flags: number, acc: boolean | undefined }[] = [];
           const removed: string[] = [];
           for (const tx of transactionsStripped) {
             inNewBlock[tx.txid] = true;
@@ -118,6 +118,7 @@ export class MempoolBlockOverviewComponent implements OnInit, OnDestroy, OnChang
               changed.push({
                 txid: tx.txid,
                 rate: tx.rate,
+                flags: tx.flags,
                 acc: tx.acc
               });
             }

frontend/src/app/interfaces/websocket.interface.ts

@@ -70,9 +70,15 @@ export interface MempoolBlockWithTransactions extends MempoolBlock {
 }
 
 export interface MempoolBlockDelta {
-  added: TransactionStripped[],
+  added: TransactionStripped[];
-  removed: string[],
+  removed: string[];
-  changed?: { txid: string, rate: number | undefined, acc: boolean | undefined }[];
+  changed: { txid: string, rate: number, flags: number, acc: boolean }[];
+}
+
+export interface MempoolBlockDeltaCompressed {
+  added: TransactionCompressed[];
+  removed: string[];
+  changed: MempoolDeltaChange[];
 }
 
 export interface MempoolInfo {
@@ -97,6 +103,11 @@ export interface TransactionStripped {
   context?: 'projected' | 'actual';
 }
 
+// [txid, fee, vsize, value, rate, flags, acceleration?]
+export type TransactionCompressed = [string, number, number, number, number, number, 1?];
+// [txid, rate, flags, acceleration?]
+export type MempoolDeltaChange = [string, number, number, (1|0)];
+
 export interface IBackendInfo {
   hostname?: string;
   gitCommit: string;

frontend/src/app/services/state.service.ts

@@ -1,7 +1,7 @@
 import { Inject, Injectable, PLATFORM_ID, LOCALE_ID } from '@angular/core';
 import { ReplaySubject, BehaviorSubject, Subject, fromEvent, Observable, merge } from 'rxjs';
 import { Transaction } from '../interfaces/electrs.interface';
-import { IBackendInfo, MempoolBlock, MempoolBlockDelta, MempoolInfo, Recommendedfees, ReplacedTransaction, ReplacementInfo, TransactionStripped } from '../interfaces/websocket.interface';
+import { IBackendInfo, MempoolBlock, MempoolBlockDelta, MempoolInfo, Recommendedfees, ReplacedTransaction, ReplacementInfo, TransactionCompressed, TransactionStripped } from '../interfaces/websocket.interface';
 import { BlockExtended, CpfpInfo, DifficultyAdjustment, MempoolPosition, OptimizedMempoolStats, RbfTree } from '../interfaces/node-api.interface';
 import { Router, NavigationStart } from '@angular/router';
 import { isPlatformBrowser } from '@angular/common';

frontend/src/app/services/websocket.service.ts

@@ -8,6 +8,7 @@ import { ApiService } from './api.service';
 import { take } from 'rxjs/operators';
 import { TransferState, makeStateKey } from '@angular/platform-browser';
 import { CacheService } from './cache.service';
+import { uncompressDeltaChange, uncompressTx } from '../shared/common.utils';
 
 const OFFLINE_RETRY_AFTER_MS = 2000;
 const OFFLINE_PING_CHECK_AFTER_MS = 30000;
@@ -382,9 +383,9 @@ export class WebsocketService {
     if (response['projected-block-transactions']) {
       if (response['projected-block-transactions'].index == this.trackingMempoolBlock) {
         if (response['projected-block-transactions'].blockTransactions) {
-          this.stateService.mempoolBlockTransactions$.next(response['projected-block-transactions'].blockTransactions);
+          this.stateService.mempoolBlockTransactions$.next(response['projected-block-transactions'].blockTransactions.map(uncompressTx));
         } else if (response['projected-block-transactions'].delta) {
-          this.stateService.mempoolBlockDelta$.next(response['projected-block-transactions'].delta);
+          this.stateService.mempoolBlockDelta$.next(uncompressDeltaChange(response['projected-block-transactions'].delta));
         }
       }
     }

frontend/src/app/shared/common.utils.ts

@@ -1,3 +1,5 @@
+import { MempoolBlockDelta, MempoolBlockDeltaCompressed, MempoolDeltaChange, TransactionCompressed, TransactionStripped } from "../interfaces/websocket.interface";
+
 export function isMobile(): boolean {
   return (window.innerWidth <= 767.98);
 }
@@ -153,3 +155,28 @@ export function seoDescriptionNetwork(network: string): string {
   }
   return '';
 }
+
+export function uncompressTx(tx: TransactionCompressed): TransactionStripped {
+  return {
+    txid: tx[0],
+    fee: tx[1],
+    vsize: tx[2],
+    value: tx[3],
+    rate: tx[4],
+    flags: tx[5],
+    acc: !!tx[6],
+  };
+}
+
+export function uncompressDeltaChange(delta: MempoolBlockDeltaCompressed): MempoolBlockDelta {
+  return {
+    added: delta.added.map(uncompressTx),
+    removed: delta.removed,
+    changed: delta.changed.map(tx => ({
+      txid: tx[0],
+      rate: tx[1],
+      flags: tx[2],
+      acc: !!tx[3],
+    }))
+  };
+}