backend: refactor blockchain network

ctisbtes · Dec 5, 2023 · 73e550c · 73e550c
1 parent 66103d5
commit 73e550c
Showing 1 changed file with 66 additions and 53 deletions.
diff --git a/backend/src/core/blockchain/modules/BlockchainNetwork.ts b/backend/src/core/blockchain/modules/BlockchainNetwork.ts
@@ -3,6 +3,20 @@ import { ControlledTimerService } from '../../network/ControlledTimerService';
 import { BlockchainBlock } from '../../../common/blockchain/block/BlockchainBlock';
 import { BlockchainNetworkSnapshot } from '../../../common/blockchain/snapshots/BlockchainNetworkSnapshot';
 import { BlockchainTx } from '../../../common/blockchain/tx/BlockchainTx';
+import { NodeConnection } from '../../network/NodeConnection';
+import { SimulationNode } from '../../SimulationNode';
+import { fatalAssert } from '../../../common/utils/fatalAssert';
+
+type RpcTarget =
+  | {
+      arity: 'broadcast';
+    }
+  | {
+      arity: 'unicast';
+      targetNodeId: string;
+    };
+
+type Rpc = (recipientNode: SimulationNode) => void;
 
 export class BlockchainNetwork {
   private readonly connectionMap: NodeConnectionMap;
@@ -29,75 +43,74 @@ export class BlockchainNetwork {
   };
 
   public readonly broadcastBlock = (block: BlockchainBlock): void => {
-    for (const connection of this.connections) {
-      const recipientNode = connection.getOtherNode(this.nodeUid);
-
-      this.timerService.createTimer({
-        waitTimeInMs: connection.latencyInMs,
-        onDone: () => {
-          recipientNode.blockchainApp.receiveBlock(block, this.nodeUid);
-        },
-      });
-    }
+    this.broadcast((recipientNode) => {
+      recipientNode.blockchainApp.receiveBlock(block, this.nodeUid);
+    });
   };
 
   public readonly broadcastTx = (tx: BlockchainTx): void => {
-    for (const connection of this.connections) {
-      const recipientNode = connection.getOtherNode(this.nodeUid);
-
-      this.timerService.createTimer({
-        waitTimeInMs: connection.latencyInMs,
-        onDone: () => {
-          recipientNode.blockchainApp.receiveTx(tx);
-        },
-      });
-    }
+    this.broadcast((recipientNode) => {
+      recipientNode.blockchainApp.receiveTx(tx);
+    });
   };
 
   public readonly requestBlock = (
     blockHash: string,
     fromNodeUid: string
   ): void => {
-    for (const connection of this.connections) {
-      const otherNode = connection.getOtherNode(this.nodeUid);
-
-      if (otherNode.nodeUid === fromNodeUid) {
-        this.timerService.createTimer({
-          waitTimeInMs: connection.latencyInMs,
-          onDone: () => {
-            otherNode.blockchainApp.sendBlockIfWeHaveIt(
-              blockHash,
-              this.nodeUid
-            );
-          },
-        });
-
-        return;
-      }
-    }
-
-    throw new Error(`No connection to node with uid ${fromNodeUid}`);
+    this.unicast(fromNodeUid, (recipientNode) => {
+      recipientNode.blockchainApp.sendBlockIfWeHaveIt(blockHash, this.nodeUid);
+    });
   };
 
   public readonly sendBlock = (
     block: BlockchainBlock,
     toNodeUid: string
   ): void => {
-    for (const connection of this.connections) {
-      const otherNode = connection.getOtherNode(this.nodeUid);
-
-      if (otherNode.nodeUid === toNodeUid) {
-        this.timerService.createTimer({
-          waitTimeInMs: connection.latencyInMs,
-          onDone: () => {
-            otherNode.blockchainApp.receiveBlock(block, this.nodeUid);
-          },
-        });
-
-        return;
-      }
+    this.unicast(toNodeUid, (recipientNode) => {
+      recipientNode.blockchainApp.receiveBlock(block, this.nodeUid);
+    });
+  };
+
+  private readonly broadcast = (rpc: Rpc): void => {
+    this.sendRpc({ arity: 'broadcast' }, rpc);
+  };
+
+  private readonly unicast = (targetNodeId: string, rpc: Rpc): void => {
+    this.sendRpc({ arity: 'unicast', targetNodeId }, rpc);
+  };
+
+  private readonly sendRpc = (target: RpcTarget, rpc: Rpc): void => {
+    const connections = this.getRpcTargetConections(target);
+
+    for (const connection of connections) {
+      this.timerService.createTimer({
+        waitTimeInMs: connection.latencyInMs,
+        onDone: () => rpc(connection.getOtherNode(this.nodeUid)),
+      });
+    }
+  };
+
+  private readonly getRpcTargetConections = (
+    target: RpcTarget
+  ): NodeConnection[] => {
+    if (target.arity === 'broadcast') {
+      return this.connections;
+    }
+
+    if (target.arity === 'unicast') {
+      const filtered = this.connections.filter(
+        (c) => c.getOtherNode(this.nodeUid).nodeUid === target.targetNodeId
+      );
+
+      fatalAssert(
+        filtered.length === 1,
+        `Expected exactly one connection for unicast, got ${filtered.length}`
+      );
+
+      return filtered;
     }
 
-    throw new Error(`No connection to node with uid ${toNodeUid}`);
+    throw new Error(`Invalid rpc target: ${target}`);
   };
 }