WIP outline of the Blockchain sync procedure using the ETH wire protocol

eth_p2p/rlpx_protocols/eth.nim

@@ -12,8 +12,6 @@ import
   rlp/types, stint, rlpx, eth_common
 
 type
-  P = UInt256
-
   NewBlockHashesAnnounce* = object
     hash: KeccakHash
     number: uint
@@ -22,34 +20,76 @@ type
     header: BlockHeader
     body {.rlpInline.}: BlockBody
 
+  NetworkState = object
+    syncing: bool
+
+  PeerState = object
+    reportedTotalDifficulty: Difficulty
+    latestBlockHash: KeccakHash
+
+const
+  maxStateFetch = 384
+  maxBodiesFetch = 128
+  maxReceiptsFetch = 256
+  maxHeadersFetch = 192
+
 rlpxProtocol eth, 63:
   useRequestIds = false
 
-  proc status(p: Peer, protocolVersion, networkId, td: P,
+  proc status(peer: Peer,
+              protocolVersion, networkId: uint,
+              totalDifficulty: Difficulty,
               bestHash, genesisHash: KeccakHash) =
-    discard
+    # verify that the peer is on the same chain:
+    if peer.network.id != networkId or
+       peer.network.chain.genesisHash != genesisHash:
+      peer.disconnect()
+      return
 
-  proc newBlockHashes(p: Peer, hashes: openarray[NewBlockHashesAnnounce]) =
+    p.state.reportedTotalDifficulty = totalDifficulty
+
+  proc newBlockHashes(peer: Peer, hashes: openarray[NewBlockHashesAnnounce]) =
     discard
 
   proc transactions(p: Peer, transactions: openarray[Transaction]) =
     discard
 
   requestResponse:
-    proc getBlockHeaders(p: Peer, hash: BlocksRequest) =
+    proc getBlockHeaders(peer: Peer, request: BlocksRequest) =
-      discard
+      if request.maxResults > maxHeadersFetch:
+        peer.disconnect()
+        return
 
-    proc blockHeaders(p: Peer, hashes: openarray[BlockHeader]) =
+      var foundBlock = peer.network.chain.locateBlock(startBlock)
-      discard
+      if not foundBlock.isNil:
+        var headers = newSeqOfCap[BlockHeader](request.maxResults)
+
+        while headers.len < request.maxResults:
+          headers.add peer.network.chain.getBlockHeader(foundBlock)
+          foundBlock = foundBlock.nextBlock()
+          if foundBlock.isNil: break
+
+        discard await peer.blockHeaders(headers)
+
+    proc blockHeaders(p: Peer, headers: openarray[BlockHeader])
 
   requestResponse:
     proc getBlockBodies(p: Peer, hashes: openarray[KeccakHash]) =
-      discard
+      if hashes.len > maxBodiesFetch:
+        peer.disconnect()
+        return
 
-    proc blockBodies(p: Peer, blocks: openarray[BlockBody]) =
+      var blockBodies = newSeqOfCap[BlockBody](hashes.len)
-      discard
+      for hash in hashes:
+        let blockBody = peer.network.chain.getBlockBody(hash)
+        if not blockBody.isNil:
+          blockBodies.add deref(blockBody)
 
-  proc newBlock(p: Peer, bh: NewBlockAnnounce, totalDificulty: P) =
+      discard await peer.blockBodies(blockBodies)
+
+    proc blockBodies(p: Peer, blocks: openarray[BlockBody])
+
+  proc newBlock(p: Peer, bh: NewBlockAnnounce, totalDifficulty: Difficulty) =
     discard
 
   nextID 13
@@ -68,3 +108,40 @@ rlpxProtocol eth, 63:
     proc receipts(p: Peer, receipts: openarray[Receipt]) =
       discard
 
+proc fastBlockchainSync*(network: EthereumNode) {.async.} =
+  # 1. obtain last N block headers from all peers
+  var latestBlocksRequest: BlocksRequest
+  var requests = newSeqOfCap[Future[eth.blockHeaders]](32)
+  for peer in network.peerPool:
+    if peer.supports(eth):
+      requests.add peer.getBlockHeaders(latestBlocksRequest)
+
+  await all(requests)
+
+  # 2. find out what is the block with best total difficulty
+  var bestBlockDifficulty: Difficulty = 0
+  for req in requests:
+    if req.read.isNone: continue
+    for header in req.read.get.headers:
+      if header.difficulty > bestBlockDifficulty:
+        discard
+
+  # 3. establish the highest valid block for each peer
+  # keep in mind that some of the peers may report an alternative history, so
+  # we must find the last block where each peer agreed with the best peer
+
+  # 4. Start making requests in parallel for the block headers that we are
+  # missing (by requesting blocks from peers while honoring maxHeadersFetch).
+  # Make sure the blocks hashes add up. Don't count on everyone replying, ask
+  # a different peer in case of time-out. Handle invalid or incomplete replies
+  # properly. The peer may response with fewer headers than requested (or with
+  # different ones if the peer is not behaving properly).
+
+  # 5. Store the obtained headers in the blockchain DB
+
+  # 6. Once the sync is complete, repeat from 1. until to further progress is
+  # possible
+
+  # 7. Start downloading the blockchain state in parallel
+  # (maybe this could start earlier).
+

eth_p2p/rlpx_protocols/les.nim

@@ -9,7 +9,7 @@
 #
 
 import
-  rlp/types, rlpx, eth_common
+  rlp/types, rlpx, eth_common, times
 
 type
   ProofRequest* = object
@@ -40,8 +40,79 @@ type
     status*: TransactionStatus
     data*: Blob
 
+  PeerState = object
+    buffer: int
+    lastRequestTime: float
+    reportedTotalDifficulty: Difficulty
+
+  KeyValuePair = object
+    key: string
+    value: Rlp
+
+const
+  maxHeadersFetch = 192
+  maxBodiesFetch = 32
+  maxReceiptsFetch = 128
+  maxCodeFetch = 64
+  maxProofsFetch = 64
+  maxHeaderProofsFetch = 64
+
+# Handshake properties:
+# https://github.com/zsfelfoldi/go-ethereum/wiki/Light-Ethereum-Subprotocol-(LES)
+const
+  keyProtocolVersion = "protocolVersion"
+    ## P: is 1 for the LPV1 protocol version.
+
+  keyNetworkId = "networkId"
+    ## P: should be 0 for testnet, 1 for mainnet.
+
+  keyHeadTotalDifficulty = "headTd"
+    ## P: Total Difficulty of the best chain.
+    ## Integer, as found in block header.
+
+  keyHeadHash = "headHash"
+    ## B_32: the hash of the best (i.e. highest TD) known block.
+
+  keyHeadNumber = "headNum"
+    ## P: the number of the best (i.e. highest TD) known block.
+
+  keyGenesisHash = "genesisHash"
+    ## B_32: the hash of the Genesis block.
+
+  keyServeHeaders = "serveHeaders"
+    ## (optional, no value)
+    ## present if the peer can serve header chain downloads.
+
+  keyServeChainSince = "serveChainSince"
+    ## P (optional)
+    ## present if the peer can serve Body/Receipts ODR requests
+    ## starting from the given block number.
+
+  keyServeStateSince = "serveStateSince"
+    ## P (optional):
+    ## present if the peer can serve Proof/Code ODR requests
+    ## starting from the given block number.
+
+  keyRelaysTransactions = "txRelay"
+    ## (optional, no value)
+    ## present if the peer can relay transactions to the ETH network.
+
+  keyFlowControlBL = "flowControl/BL"
+  keyFlowControlMRC = "flowControl/MRC"
+  keyFlowControlMRR = "flowControl/MRR"
+    ## see Client Side Flow Control:
+    ## https://github.com/zsfelfoldi/go-ethereum/wiki/Client-Side-Flow-Control-model-for-the-LES-protocol
+
+const
+  rechargeRate = 0.3
+
+proc getPeerWithNewestChain(pool: PeerPool): Peer =
+  discard
+
 rlpxProtocol les, 2:
 
+  type State = PeerState
+
   ## Handshake
   ##
 
@@ -52,8 +123,11 @@ rlpxProtocol les, 2:
   ##
 
   proc announce(p: Peer, headHash: KeccakHash,
-                headNumber, headTd, reorgDepth: P,
+                headNumber: BlockNumber,
-                values: openarray[KeyValuePair], announceType: uint) =
+                headTotalDifficulty: Difficulty,
+                reorgDepth: BlockNumber,
+                values: openarray[KeyValuePair],
+                announceType: uint) =
     discard
 
   requestResponse: