[WIP] ETH2.0 Wire protocol

package.json

@@ -62,6 +62,7 @@
     "pouchdb-adapter-memory": "^7.0.0",
     "pouchdb-core": "^7.0.0",
     "promisify-es6": "^1.0.3",
+    "pull-stream": "^3.6.10",
     "winston": "^3.2.1",
     "ws": "^6.2.1"
   },

src/p2p/index.ts

@@ -4,9 +4,11 @@ import {LodestarNode} from "./node";
 import logger, {AbstractLogger} from "../logger";
 import {PeerInfo} from "peer-info";
 import LibP2p from "libp2p";
+import {pull} from "pull-stream";
 import {PeerBook} from "peer-book";
 import {PeerId} from "peer-id";
 import {promisify} from "promisify-es6";
+import {Hello, Goodbye} from "../rpc/api/wire/messages";
 
 export interface P2pOptions {
   maxPeers: number;
@@ -16,6 +18,10 @@ export interface P2pOptions {
   bootnodes: string[];
 }
 
+export interface ChainOptions {
+
+}
+
 /**
  * The P2PNetwork service manages p2p connection/subscription objects
  */
@@ -48,7 +54,7 @@ export class P2PNetwork extends EventEmitter implements Service {
     this.refreshInterval = this.options.refreshInterval;
     this.peerBook = this.options.peerBook;
     this.privateKey = this.options.privateKey;
-    this.bootnodes = this.options.bootnodes;
+    this.bootnodes = this.options.bootnodes || [];
 
 
     this.started = false;
@@ -89,18 +95,45 @@ export class P2PNetwork extends EventEmitter implements Service {
 
       });
 
+      // 2-way handshake
+      const protocol: string = "/eth/serenity/beacon/rpc/1";
+      const helloMsg: Hello = {
+
+      };
+      this.node.handle(protocol, (proto, conn) => {
+        pull(
+          pull.values([Buffer.from(JSON.stringify(helloMsg))]),
+          conn,
+          pull.collect((values) => {
+	    // Peers' responses
+
+	  })
+	);	    
+      });
       this.node.on('peer:connect', (peerInfo) => {
         try {
           this.log.info(`Peer connected: ${peerInfo}`);
           this.peerBook.put(peerInfo);
-          this.discoveredPeers.add(peerInfo);
+	  this.discoveredPeers.add(peerInfo);
+	  this.node.dialProtocol(peerInfo, protocol, (err, conn) => {
+	    pull(
+              pull.values([Buffer.from(JSON.stringify(helloMsg))]),
+              conn,
+              pull.collect((values) => {
+	        // Peers responses
+
+	      })
+	    );
+	  })
+
         } catch (err) {
           this.log.error(err);
         }
       });
 
       this.node.on('peer:disconnect', (peerInfo) => {
-        try {
+	try {
+
           this.peerBook.remove(peerInfo);
           this.discoveredPeers.delete(peerInfo);
         } catch (err) {

src/p2p/wire/messages.ts

@@ -0,0 +1,79 @@
+
+export interface Request {
+  id: number;
+  method_id: number;
+  body: string;
+}
+
+export interface Response {
+  id: number;
+  response_code: number;
+  result: Buffer;
+}
+
+// Method ID: 0
+
+export interface Hello {
+  network_id: number;
+  chain_id: number;
+  latest_finalized_root: Buffer;
+  latest_finalized_epoch: number;
+  best_root: Buffer;
+  best_slot: number;
+}
+
+// Method ID: 1
+
+export interface Goodbye {
+  reason: number;
+}
+
+// Method ID: 2
+
+export interface GetStatus {
+  sha: Buffer;
+  user_agent: Buffer;
+  timestamp: number;
+} 
+
+// Method ID: 10
+
+export interface RequestBeaconBlockRoots {
+  start_slot: number;
+  count: number;
+}
+
+export interface BeaconBlockRoots {
+  block_root: Buffer;
+  slot: number;
+  // Doesn't currently exist as a standalone type
+  roots: []BlockRootSlot;
+}
+
+// Method ID: 11
+export interface RequestBeaconBlockHeader {
+  // Doesn't currently exist as a standalone type	
+  start_root: HashTreeRoot; 
+  start_slot: number;
+  max_headers: number;
+  skip_slots: number;
+}
+
+export interface BeaconBlockHeaders {
+  // Doesn't currently exist as a standalone type
+  headers: []BeaconBlockHeader
+}
+
+// Method ID: 12
+export interface RequestBeaconBlockBodies {
+  block_roots: []HashTreeRoot;
+}
+
+export interface BeaconBlockBodies {
+  block_bodies: []BeaconBlockBody;
+}
+
+// Method ID: 13
+export interface RequestBeaconChainState {
+  hashes: []HashTreeRoot;
+}

src/p2p/wire/types.ts

@@ -0,0 +1,10 @@
+import {Slot} from "../../types/primitive"
+
+export interface BlockRootSlot {
+  block_root: Buffer;
+  slot: Slot;
+}
+
+export interface HashTreeRoot {
+  hash: Buffer;
+}

src/rpc/api/wire/index.ts

@@ -0,0 +1,7 @@
+import {WireProtocolApi} from "./wire";
+import {IWireProtocolApi} from "./interface";
+
+export {
+  WireProtocolApi,
+  IWireProtocolApi
+};

src/rpc/api/wire/interface.ts

@@ -0,0 +1,46 @@
+import {IApi} from "../interface";
+
+import {
+  Request,
+  Response,
+  Hello,
+  Goodbye,
+  GetStatus,
+  BeaconBlockRootsRequest,
+  BeaconBlockRootsResponse,
+  BeaconBlockHeaderRequest,
+  BeaconBlockHeaderResponse,
+  BeaconBlockBodiesRequest,
+  BeaconBlockBodiesResponse,
+  BeaconChainStateRequest,
+  BeaconChainStateResponse
+} from "./messages";
+
+export interface IWireProtocolApi extends IApi {
+
+  /**
+   * Returns metadata about the remote node.
+   */
+  GetStatus(): Promise<GetStatus>;
+
+  /**
+   * Returns list of block roots and slots from the peer
+   */
+  RequestBeaconBlockRoots(request: BeaconBlockRootsRequest): Promise<BeaconBlockRootsResponse>;
+
+  /**
+   * Returns beacon block headers from peer
+   */
+  RequestBeaconBlockHeaders(request: BeaconBlockHeadersRequest): Promise<BeaconBlockHeaderResponse>;
+
+  /**
+   * Returns block bodies associated with block roots from a peer
+   */
+  RequestBeaconBlockBodies(request: BeaconBlockBodiesRequest): Promise<BeaconBlockBodiesResponse>;
+
+  /**
+   * Returns the hashes of merkle tree nodes from merkelizing the block's state root.
+   */
+  RequestBeaconChainState(request: BeaconChainStateRequest): Promise<BeaconChainStateResponse>;
+
+}

src/rpc/api/wire/messages.ts

@@ -0,0 +1,89 @@
+import {Slot} from "../../types/primitive";
+import {BlockRootSlot, HashTreeRoot} from "./types";
+
+type RPCMethod = Hello | Goodbye | GetStatus;
+
+export interface Request {
+  id: number;
+  method_id: number;
+  body: string;
+}
+
+export interface Response {
+  id: number;
+  response_code: number;
+  result: Buffer;
+}
+
+// Method ID: 0
+
+export interface Hello {
+  network_id: number;
+  chain_id: number;
+  latest_finalized_root: Buffer;
+  latest_finalized_epoch: number;
+  best_root: Buffer;
+  best_slot: Slot;
+}
+
+// Method ID: 1
+
+export interface Goodbye {
+  reason: number;
+}
+
+// Method ID: 2
+
+export interface GetStatusRequest {
+  sha: Buffer;
+  user_agent: Buffer;
+  timestamp: number;
+}
+
+// Method ID: 10
+
+export interface BeaconBlockRootsRequest {
+  start_slot: Slot;
+  count: number;
+}
+
+export interface BeaconBlockRootsResponse {
+  block_root: Buffer;
+  slot: Slot;
+  // Doesn't currently exist as a standalone type
+  roots: []BlockRootSlot;
+}
+
+// Method ID: 11
+export interface BeaconBlockHeadersRequest {
+  // Doesn't currently exist as a standalone type	
+  start_root: HashTreeRoot; 
+  start_slot: Slot;
+  max_headers: number;
+  skip_slots: number;
+}
+
+export interface BeaconBlockHeadersResponse {
+  // Doesn't currently exist as a standalone type
+  headers: []BeaconBlockHeader
+}
+
+// Method ID: 12
+export interface BeaconBlockBodiesRequest {
+  block_roots: []HashTreeRoot;
+}
+
+export interface BeaconBlockBodiesResponse {
+  block_bodies: []BeaconBlockBody;
+}
+
+// Method ID: 13
+export interface BeaconChainStateRequest {
+  hashes: []HashTreeRoot;
+}
+
+// Method ID: 14
+// Not yet defined in the ETH2.0 Wire spec.
+export interface BeaconChainStateResponse {
+
+}

src/rpc/api/wire/types.ts

@@ -0,0 +1,10 @@
+import {Slot} from "../../types/primitive"
+
+export interface BlockRootSlot {
+  block_root: Buffer;
+  slot: Slot;
+}
+
+export interface HashTreeRoot {
+  hash: Buffer;
+}

src/rpc/api/wire/wire.ts

@@ -0,0 +1,58 @@
+import {IWireProtocolApi} from "./interface";
+import {BeaconChain} from "../../../chain";
+import {DB} from "../../../db";
+
+import {
+  Request,
+  Response,
+  Hello,
+  Goodbye,
+  BeaconBlockRootsRequest,
+  BeaconBlockRootsResponse,
+  BeaconBlockHeaderRequest,
+  BeaconBlockHeaderResponse,
+  BeaconBlockBodiesRequest,
+  BeaconBlockBodiesResponse,
+  BeaconChainStateRequest,
+  BeaconChainStateResponse
+} from "./messages";
+
+import {
+  BlockRootSlot,
+  HashTreeRoot
+} from "./types";
+
+export class WireProtocolApi implements IWireProtocolApi {
+
+   public namespace: string;
+
+   private chain: BeaconChain;
+   private db: DB;
+
+   public constructor(opts, {chain, db}) {
+     this.namespace = 'eth2-wire';
+     this.db;
+     this.chain;
+   }
+
+   public async GetStatus(): Promise<GetStatus> {
+
+   }
+
+   public async RequestBeaconBlockRoots(request: BeaconBlockRootsRequest): Promise<BeaconBlockRootsResponse> {
+
+   }
+
+   public async RequestBeaconBlockHeaders(request: BeaconBlockHeadersRequest): Promise<BeaconBlockHeadersResponse> {
+
+   }
+
+   public async RequestBeaconBlockBodies(request: BeaconBlockBodiesRequest): Promise<BeaconBlockBodiesResponse> {
+
+   }
+
+   public async RequestBeaconChainState(request: BeaconChainStateRequest): Promise<BeaconChainStateResponse> {
+
+   }
+
+}