Chunked sync second try

This reverts commit cdf9baf5e6.

beacon_chain/sync_protocol.nim

@@ -23,6 +23,10 @@ type
     node*: BeaconNode
     db*: BeaconChainDB
 
+const
+  MaxRootsToRequest = 512
+  MaxHeadersToRequest = MaxRootsToRequest
+
 func toHeader(b: BeaconBlock): BeaconBlockHeaderRLP =
   BeaconBlockHeaderRLP(
     slot: b.slot.uint64,
@@ -77,12 +81,13 @@ p2pProtocol BeaconSync(version = 1,
       networkId = peer.networkState.networkId
       blockPool = node.blockPool
       latestState = blockPool.latestState()
+      headBlock = blockPool.head
 
     var
       latestFinalizedRoot: Eth2Digest # TODO
       latestFinalizedEpoch = latestState.finalized_epoch
       bestRoot: Eth2Digest # TODO
-      bestSlot = latestState.slot
+      bestSlot = headBlock.slot
 
     let m = await handshake(peer, timeout = 10.seconds,
                             status(networkId, latestFinalizedRoot,
@@ -97,42 +102,49 @@ p2pProtocol BeaconSync(version = 1,
     # where it needs to sync and it should execute the sync algorithm with a certain
     # number of randomly selected peers. The algorithm itself must be extracted in a proc.
     try:
-      debug "Peer connected. Initiating sync", peer
+      debug "Peer connected. Initiating sync", peer, bestSlot, remoteBestSlot = m.bestSlot
 
       let bestDiff = cmp((latestFinalizedEpoch, bestSlot), (m.latestFinalizedEpoch, m.bestSlot))
-      if bestDiff == 0:
+      if bestDiff >= 0:
         # Nothing to do?
         trace "Nothing to sync", peer = peer.remote
       else:
         # TODO: Check for WEAK_SUBJECTIVITY_PERIOD difference and terminate the
         # connection if it's too big.
 
-        if bestDiff > 0:
+        var s = bestSlot + 1
-          # Send roots
+        while s <= m.bestSlot:
-          # TODO: Currently we send all block roots in one "packet". Maybe
+          debug "Waiting for block roots", fromPeer = peer, remoteBestSlot = m.bestSlot, peer
-          # they should be split to multiple packets.
+          let r = await peer.getBeaconBlockRoots(s, MaxRootsToRequest)
-          type Root = (Eth2Digest, Slot)
+          if not r.isSome:
-          var roots = newSeqOfCap[Root](128)
+            debug "Block roots not received", peer
-          for i in int(m.bestSlot) + 1 .. int(bestSlot):
+            break
-            for r in blockPool.blockRootsForSlot(i.Slot):
+          let roots = r.get.roots
-              roots.add((r, i.Slot))
+          debug "Received block roots", len = roots.len, peer
+          if roots.len != 0:
+            let headers = await peer.getBeaconBlockHeaders(bestRoot, s, roots.len, 0)
+            var bodiesRequest = newSeqOfCap[Eth2Digest](roots.len)
+            for r in roots:
+              bodiesRequest.add(r[0])
 
-          debug "Sending block roots", peer, coveredSlots = roots.len
+            debug "Block headers received. Requesting block bodies", peer
-          await peer.beaconBlockRoots(roots)
+            let bodies = await peer.getBeaconBlockBodies(bodiesRequest)
-        else:
+            node.importBlocks(roots, headers.get.blockHeaders, bodies.get.blockBodies)
-          # Receive roots
-          debug "Waiting for block roots", fromPeer = peer
-          let roots = await peer.nextMsg(BeaconSync.beaconBlockRoots)
 
-          debug "Block roots received. Requesting block headers", bestRoot, bestSlot
+            let lastSlot = roots[^1][1]
-          let headers = await peer.getBeaconBlockHeaders(bestRoot, bestSlot, roots.roots.len, 0)
+            if roots.len >= MaxRootsToRequest:
-          var bodiesRequest = newSeqOfCap[Eth2Digest](roots.roots.len)
+              # Next batch of roots starts with the last slot of the current one
-          for r in roots.roots:
+              # to make sure we did not miss any roots with this slot that did
-            bodiesRequest.add(r[0])
+              # not fit into the response.
 
-          debug "Block headers received. Requesting block bodies", blocks = bodiesRequest
+              if s == lastSlot:
-          let bodies = await peer.getBeaconBlockBodies(bodiesRequest)
+                info "Too many roots for a single slot while syncing"
-          node.importBlocks(roots.roots, headers.get.blockHeaders, bodies.get.blockBodies)
+                break
+              s = lastSlot
+            else:
+              s = lastSlot + 1
+          else:
+            break
 
     except CatchableError:
       warn "Failed to sync with peer", peer, err = getCurrentExceptionMsg()
@@ -145,9 +157,21 @@ p2pProtocol BeaconSync(version = 1,
             bestRoot: Eth2Digest,
             bestSlot: Slot) {.libp2pProtocol("hello", "1.0.0").}
 
-  proc beaconBlockRoots(
+  requestResponse:
-            peer: Peer,
+    proc getBeaconBlockRoots(peer: Peer, fromSlot: Slot, maxRoots: int) =
-            roots: openarray[(Eth2Digest, Slot)]) {.libp2pProtocol("rpc/beacon_block_roots", "1.0.0").}
+      doAssert(maxRoots <= MaxRootsToRequest) # TODO: Validate maxRoots properly
+      var s = fromSlot
+      var roots = newSeqOfCap[(Eth2Digest, Slot)](maxRoots)
+      let blockPool = peer.networkState.node.blockPool
+      let maxSlot = blockPool.head.slot
+      while s <= maxSlot:
+        for r in blockPool.blockRootsForSlot(s):
+          roots.add((r, s))
+          if roots.len == maxRoots: break
+        s += 1
+      await response.send(roots)
+
+    proc beaconBlockRoots(peer: Peer, roots: openarray[(Eth2Digest, Slot)])
 
   requestResponse:
     proc getBeaconBlockHeaders(
@@ -157,15 +181,16 @@ p2pProtocol BeaconSync(version = 1,
             maxHeaders: int,
             skipSlots: int) {.libp2pProtocol("rpc/beacon_block_headers", "1.0.0").} =
       # TODO: validate maxHeaders and implement slipSlots
-      var s = slot.int
+      var s = slot
       var headers = newSeqOfCap[BeaconBlockHeaderRLP](maxHeaders)
       let db = peer.networkState.db
       let blockPool = peer.networkState.node.blockPool
-      while headers.len < maxHeaders:
+      let maxSlot = blockPool.head.slot
-        for r in blockPool.blockRootsForSlot(s.Slot):
+      while s <= maxSlot:
+        for r in blockPool.blockRootsForSlot(s):
           headers.add(db.getBlock(r).get().toHeader)
           if headers.len == maxHeaders: break
-        inc s
+        s += 1
       await response.send(headers)
 
     proc beaconBlockHeaders(peer: Peer, blockHeaders: openarray[BeaconBlockHeaderRLP])