Merge pull request #154 from ChainSafe/mikerah/libp2p_integration

[WIP] Integrating libp2p into Lodestar

package.json

@@ -33,6 +33,7 @@
   ],
   "dependencies": {
     "@chainsafe/ssz": "^0.2.0",
+    "@nodeutils/defaults-deep": "^1.1.0",
     "@polkadot/util-crypto": "^0.33.27",
     "@types/bn.js": "^4.11.4",
     "@types/levelup": "^3.1.0",
@@ -45,13 +46,23 @@
     "commander": "^2.19.0",
     "deepmerge": "^3.2.0",
     "ethers": "^4.0.27",
+    "ganache-core": "^2.5.3",
     "level": "^4.0.0",
+    "libp2p": "^0.24.4",
+    "libp2p-bootstrap": "^0.9.7",
+    "libp2p-floodsub": "^0.15.8",
+    "libp2p-kad-dht": "^0.14.12",
+    "libp2p-mplex": "^0.8.5",
+    "libp2p-tcp": "^0.13.0",
+    "libp2p-webrtc-star": "^0.15.8",
     "noice-json-rpc": "^1.2.0",
+    "peer-book": "^0.9.1",
+    "peer-id": "^0.12.2",
+    "peer-info": "^0.15.1",
     "pouchdb-adapter-memory": "^7.0.0",
     "pouchdb-core": "^7.0.0",
     "promisify-es6": "^1.0.3",
     "winston": "^3.2.1",
-    "ganache-core": "^2.5.3",
     "ws": "^6.2.1"
   },
   "bin": {
@@ -74,6 +85,7 @@
     "@types/sinon": "^7.0.11",
     "@typescript-eslint/eslint-plugin": "^1.3.0",
     "@typescript-eslint/parser": "^1.3.0",
+    "@polkadot/ts": "^0.1.56",
     "chai": "^4.2.0",
     "chai-as-promised": "^7.1.1",
     "codecov": "^3.1.0",

src/node/defaults.ts

@@ -1,4 +1,5 @@
 import eth1Defaults from "../eth1/defaults";
+import p2pDefaults from "../p2p/defaults";
 
 export default {
   chain: {
@@ -11,4 +12,5 @@ export default {
     port: 9545
   },
   eth1: eth1Defaults,
+  p2p: p2pDefaults
 };

src/node/index.ts

@@ -13,17 +13,19 @@ import {JSONRPC} from "../rpc/protocol";
 import {WSServer} from "../rpc/transport";
 import {BeaconAPI} from "../rpc/api";
 
-interface Service {
+export interface Service {
   start(): Promise<void>;
 
   stop(): Promise<void>;
 }
 
+// Temporarily have properties be optional until others portions of lodestar are ready
 interface BeaconNodeCtx {
   chain?: object;
   db?: object;
+  // Temporarily set to any. Will be changed to object later.
   eth1?: any;
-  network?: object;
+  network?: any;
   rpc?: object;
   sync?: object;
   opPool?: object;

src/p2p/defaults.ts

@@ -0,0 +1,8 @@
+import {PeerBook} from "peer-book";
+
+export default {
+  maxPeers: 25,
+  refreshInterval: 15000,
+  peerBook: new PeerBook(),
+  bootnodes: []
+};

src/p2p/index.ts

@@ -1,13 +1,149 @@
 import {EventEmitter} from "events";
+import {Service} from "../node";
+import {LodestarNode} from "./node";
+import logger, {AbstractLogger} from "../logger";
+import {PeerInfo} from "peer-info";
+import LibP2p from "libp2p";
+import {PeerBook} from "peer-book";
+import {PeerId} from "peer-id";
+import {promisify} from "promisify-es6";
+
+export interface P2pOptions {
+  maxPeers: number;
+  refreshInterval: number;
+  peerBook: PeerBook;
+  privateKey: Buffer;
+  bootnodes: string[];
+}
 
 /**
  * The P2PNetwork service manages p2p connection/subscription objects
  */
-export class P2PNetwork extends EventEmitter {
-  public constructor(opts) {
+export class P2PNetwork extends EventEmitter implements Service {
+
+  private options: P2pOptions;
+
+  private maxPeers: number;
+
+  private refreshInterval: number;
+
+  private peerBook: PeerBook;
+
+  private privateKey: Buffer;
+
+  private bootnodes: string[];
+
+  private started: boolean;
+
+  private node: LibP2p;
+
+  private discoveredPeers: Set<PeerInfo>;
+
+  private log: AbstractLogger;
+
+  public constructor(opts: P2pOptions) {
     super();
-    // build gossipsub object
+    this.options = opts;
+    this.maxPeers = this.options.maxPeers;
+    this.refreshInterval = this.options.refreshInterval;
+    this.peerBook = this.options.peerBook;
+    this.privateKey = this.options.privateKey;
+    this.bootnodes = this.options.bootnodes;
+
+
+    this.started = false;
+    this.node = null;
+    this.discoveredPeers = new Set();
+    this.log = logger;
+  }
+
+  public get isRunning(): boolean {
+    return this.started;
+  }
+
+  public async start(): Promise<void> {
+    if (this.started) {
+      throw new Error("P2P Network already started");
+    }
+
+    if (!this.node) {
+      this.node = LodestarNode.createNode({
+        peerInfo: await this.createPeerInfo(),
+        bootnodes: this.options.bootnodes
+      });
+
+      this.node.on('peer:discovery', (peerInfo) => {
+        try {
+          const peerId = peerInfo.id.toB58String();
+          // Check if peer has already been discovered
+          if (this.options.peerBook.has(peerId) || this.discoveredPeers.has(peerId)) {
+            return;
+          }
+          this.peerBook.put(peerInfo);
+	  this.node.dial(peerInfo, () => {});
+          this.log.info(`Peer discovered: ${peerInfo}`);
+          this.emit('connected', peerInfo);
+        } catch (err) {
+          this.log.error(err);
+        }
+
+      });
+
+      this.node.on('peer:connect', (peerInfo) => {
+        try {
+          this.log.info(`Peer connected: ${peerInfo}`);
+          this.peerBook.put(peerInfo);
+          this.discoveredPeers.add(peerInfo);
+        } catch (err) {
+          this.log.error(err);
+        }
+      });
+
+      this.node.on('peer:disconnect', (peerInfo) => {
+        try {
+          this.peerBook.remove(peerInfo);
+          this.discoveredPeers.delete(peerInfo);
+        } catch (err) {
+          this.log.error(err);
+        }
+      });
+    }
+    await promisify(this.node.start.bind(this.node))();
+
+    this.started = true;
+  }
+
+  public async stop(): Promise<void> {
+    if (!this.started) {
+      return;
+    }
+    this.node.removeAllListeners();
+    await promisify(this.node.stop.bind(this.node))();
+  }
+
+  private async createPeerInfo(): PeerInfo {
+    return new Promise((resolve, reject) => {
+      const handler = (err, peerInfo) => {
+        if (err) {
+	  return reject(err);
+        }
+        this.peerBook.getAll().forEach((peer) => {
+	  peer.multiaddrs.forEach((multiaddr) => {
+	    peerInfo.multiaddrs.add(multiaddr);
+	    resolve(peerInfo);
+	  });
+        });
+      };
+      if (this.privateKey) {
+        PeerId.createFromPrivKey(this.privateKey, (err, id) => {
+	  if (err) {
+	    return reject(err);
+	  }
+	  PeerInfo.create(id, handler);	  
+        });
+      } else {
+        PeerInfo.create(handler);
+      } 
+    });
   }
-  public async start() {}
-  public async stop() {}
 }

src/p2p/node.ts

@@ -0,0 +1,60 @@
+import LibP2p from "libp2p";
+import {TCP} from "libp2p-tcp";
+import {Mplex} from "libp2p-mplex";
+import {Bootstrap} from "libp2p-bootstrap";
+import {promisify} from "es6-promisify";
+import {PeerInfo} from "peer-info";
+import {PeerId} from "peer-id";
+import {defaultsDeep} from "@nodeutils/defaults-deep";
+import * as FloodSub from "libp2p-floodsub";
+
+export interface LodestarNodeOpts {
+  bootstrap?: string[];
+  peerInfo: PeerInfo;
+  bootnodes?: string[];
+}
+
+export class LodestarNode extends LibP2p {
+
+  private pubsub: FloodSub;
+
+  private constructor(_options: LodestarNodeOpts) {
+    const defaults = {
+      modules: {
+        transport: [TCP],
+        streamMuxer: [Mplex],
+        peerDiscovery: [Bootstrap]
+      },
+      config: {
+        peerDiscovery: {
+          bootstrap: {
+            interval: 2000,
+            enabled: true,
+            list: _options.bootstrap || []
+          }
+        }
+      }
+    };
+
+    super(defaultsDeep(_options, defaults));
+  }
+
+  public static createNode(callback): LodestarNode {
+    const id = promisify(PeerId.create)({bits: 1024});
+    const peerInfo = promisify(PeerInfo.create)(id);
+    peerInfo.multiaddrs.add('ip4/0.0.0.0/tcp/9000');
+    const node = new LodestarNode({
+      peerInfo
+    });
+
+    node.pubsub = new FloodSub(node);
+
+    return node;
+  }
+
+  public async start(): Promise<void> {
+    await promisify(super.start.bind(this))();
+    await promisify(this.pubsub.start.bind(this.pubsub))();
+  }
+}
+