Syncing. (#909)

beacon_chain/beacon_node.nim

@@ -16,7 +16,7 @@ import
   attestation_pool, block_pool, eth2_network, eth2_discovery,
   beacon_node_types, mainchain_monitor, version, ssz, ssz/dynamic_navigator,
   sync_protocol, request_manager, validator_keygen, interop, statusbar,
-  attestation_aggregation
+  attestation_aggregation, sync_manager
 
 const
   genesisFile = "genesis.ssz"
@@ -67,6 +67,7 @@ type
     forkDigest: ForkDigest
     topicBeaconBlocks: string
     topicAggregateAndProofs: string
+    syncLoop: Future[void]
 
 proc onBeaconBlock*(node: BeaconNode, signedBlock: SignedBeaconBlock) {.gcsafe.}
 proc updateHead(node: BeaconNode): BlockRef
@@ -486,9 +487,7 @@ proc onAttestation(node: BeaconNode, attestation: Attestation) =
 
   node.attestationPool.add(attestation)
 
-proc onBeaconBlock(node: BeaconNode, signedBlock: SignedBeaconBlock) =
-  # We received a block but don't know much about it yet - in particular, we
-  # don't know if it's part of the chain we're currently building.
+proc storeBlock(node: BeaconNode, signedBlock: SignedBeaconBlock): bool =
   let blockRoot = hash_tree_root(signedBlock.message)
   debug "Block received",
     signedBlock = shortLog(signedBlock.message),
@@ -497,9 +496,8 @@ proc onBeaconBlock(node: BeaconNode, signedBlock: SignedBeaconBlock) =
     pcs = "receive_block"
 
   beacon_blocks_received.inc()
-
   if node.blockPool.add(blockRoot, signedBlock).isNil:
-    return
+    return false
 
   # The block we received contains attestations, and we might not yet know about
   # all of them. Let's add them to the attestation pool - in case they block
@@ -511,6 +509,12 @@ proc onBeaconBlock(node: BeaconNode, signedBlock: SignedBeaconBlock) =
     signedBlock.message.slot.epoch + 1 >= currentSlot.slot.epoch:
     for attestation in signedBlock.message.body.attestations:
       node.onAttestation(attestation)
+  return true
+
+proc onBeaconBlock(node: BeaconNode, signedBlock: SignedBeaconBlock) =
+  # We received a block but don't know much about it yet - in particular, we
+  # don't know if it's part of the chain we're currently building.
+  discard node.storeBlock(signedBlock)
 
 proc handleAttestations(node: BeaconNode, head: BlockRef, slot: Slot) =
   ## Perform all attestations that the validators attached to this node should
@@ -900,10 +904,10 @@ proc handleMissingBlocks(node: BeaconNode) =
       #      good because the onSecond fetching also kicks in regardless but
       #      whatever - this is just a quick fix for making the testnet easier
       #      work with while the sync problem is dealt with more systematically
-      dec left
-      if left == 0:
-        discard setTimer(Moment.now()) do (p: pointer):
-          handleMissingBlocks(node)
+      # dec left
+      # if left == 0:
+      #   discard setTimer(Moment.now()) do (p: pointer):
+      #     handleMissingBlocks(node)
 
 proc onSecond(node: BeaconNode, moment: Moment) {.async.} =
   node.handleMissingBlocks()
@@ -912,6 +916,62 @@ proc onSecond(node: BeaconNode, moment: Moment) {.async.} =
   discard setTimer(nextSecond) do (p: pointer):
     asyncCheck node.onSecond(nextSecond)
 
+proc getHeadSlot*(peer: Peer): Slot {.inline.} =
+  ## Returns head slot for specific peer ``peer``.
+  result = peer.state(BeaconSync).statusMsg.headSlot
+
+proc updateStatus*(peer: Peer): Future[bool] {.async.} =
+  ## Request `status` of remote peer ``peer``.
+  let
+    nstate = peer.networkState(BeaconSync)
+    finalizedHead = nstate.blockPool.finalizedHead
+    headBlock = nstate.blockPool.head.blck
+
+    ourStatus = StatusMsg(
+      forkDigest: nstate.forkDigest,
+      finalizedRoot: finalizedHead.blck.root,
+      finalizedEpoch: finalizedHead.slot.compute_epoch_at_slot(),
+      headRoot: headBlock.root,
+      headSlot: headBlock.slot
+    )
+
+  let theirStatus = await peer.status(ourStatus,
+                                      timeout = chronos.seconds(60))
+  if theirStatus.isSome():
+    peer.state(BeaconSync).statusMsg = theirStatus.get()
+    result = true
+
+proc updateScore*(peer: Peer, score: int) =
+  ## Update peer's ``peer`` score with value ``score``.
+  peer.score = peer.score + score
+
+proc runSyncLoop(node: BeaconNode) {.async.} =
+
+  proc getLocalHeadSlot(): Slot =
+    result = node.blockPool.head.blck.slot
+
+  proc getLocalWallSlot(): Slot {.gcsafe.} =
+    let epoch = node.beaconClock.now().toSlot().slot.compute_epoch_at_slot() + 1'u64
+    result = epoch.compute_start_slot_at_epoch()
+
+  proc updateLocalBlocks(list: openarray[SignedBeaconBlock]): bool =
+    debug "Forward sync imported blocks", count = len(list),
+          local_head_slot = $getLocalHeadSlot()
+    for blk in list:
+      if not(node.storeBlock(blk)):
+        return false
+    discard node.updateHead()
+    info "Forward sync blocks got imported sucessfully", count = $len(list),
+         local_head_slot = $getLocalHeadSlot()
+    result = true
+
+  var syncman = newSyncManager[Peer, PeerID](
+    node.network.peerPool, getLocalHeadSlot, getLocalWallSlot,
+    updateLocalBlocks
+  )
+
+  await syncman.sync()
+
 # TODO: Should we move these to other modules?
 # This would require moving around other type definitions
 proc installValidatorApiHandlers(rpcServer: RpcServer, node: BeaconNode) =
@@ -1060,6 +1120,8 @@ proc run*(node: BeaconNode) =
   discard setTimer(second) do (p: pointer):
     asyncCheck node.onSecond(second)
 
+  node.syncLoop = runSyncLoop(node)
+
   runForever()
 
 var gPidFile: string

beacon_chain/eth2_network.nim

@@ -27,7 +27,7 @@ import
 
 export
   version, multiaddress, peer_pool, peerinfo, p2pProtocol,
-  libp2p_json_serialization, ssz
+  libp2p_json_serialization, ssz, peer
 
 logScope:
   topics = "networking"
@@ -195,6 +195,9 @@ proc getFuture*(peer: Peer): Future[void] {.inline.} =
 proc `<`*(a, b: Peer): bool =
   result = `<`(a.score, b.score)
 
+proc getScore*(a: Peer): int =
+  result = a.score
+
 proc disconnect*(peer: Peer, reason: DisconnectionReason,
                  notifyOtherPeer = false) {.async.} =
   # TODO: How should we notify the other peer?

beacon_chain/sync_protocol.nim

@@ -8,6 +8,13 @@ logScope:
   topics = "sync"
 
 type
+  StatusMsg* = object
+    forkDigest*: ForkDigest
+    finalizedRoot*: Eth2Digest
+    finalizedEpoch*: Epoch
+    headRoot*: Eth2Digest
+    headSlot*: Slot
+
   ValidatorSetDeltaFlags {.pure.} = enum
     Activation = 0
     Exit = 1
@@ -27,7 +34,8 @@ type
     onBeaconBlock*: BeaconBlockCallback
 
   BeaconSyncPeerState* = ref object
-    initialStatusReceived: bool
+    initialStatusReceived*: bool
+    statusMsg*: StatusMsg
 
   BlockRootSlot* = object
     blockRoot: Eth2Digest
@@ -42,14 +50,6 @@ proc importBlocks(state: BeaconSyncNetworkState,
     state.onBeaconBlock(blk)
   info "Forward sync imported blocks", len = blocks.len
 
-type
-  StatusMsg = object
-    forkDigest*: ForkDigest
-    finalizedRoot*: Eth2Digest
-    finalizedEpoch*: Epoch
-    headRoot*: Eth2Digest
-    headSlot*: Slot
-
 proc getCurrentStatus(state: BeaconSyncNetworkState): StatusMsg {.gcsafe.} =
   let
     blockPool = state.blockPool
@@ -69,7 +69,7 @@ proc getCurrentStatus(state: BeaconSyncNetworkState): StatusMsg {.gcsafe.} =
 proc handleInitialStatus(peer: Peer,
                          state: BeaconSyncNetworkState,
                          ourStatus: StatusMsg,
-                         theirStatus: StatusMsg) {.async, gcsafe.}
+                         theirStatus: StatusMsg): Future[bool] {.async, gcsafe.}
 
 p2pProtocol BeaconSync(version = 1,
                        rlpxName = "bcs",
@@ -85,7 +85,11 @@ p2pProtocol BeaconSync(version = 1,
         theirStatus = await peer.status(ourStatus, timeout = 60.seconds)
 
       if theirStatus.isSome:
-        await peer.handleInitialStatus(peer.networkState, ourStatus, theirStatus.get)
+        let tstatus = theirStatus.get()
+        let res = await peer.handleInitialStatus(peer.networkState,
+                                                 ourStatus, tstatus)
+        if res:
+          peer.state(BeaconSync).statusMsg = tstatus
       else:
         warn "Status response not received in time"
 
@@ -99,7 +103,10 @@ p2pProtocol BeaconSync(version = 1,
 
       if not peer.state.initialStatusReceived:
         peer.state.initialStatusReceived = true
-        await peer.handleInitialStatus(peer.networkState, ourStatus, theirStatus)
+        let res = await peer.handleInitialStatus(peer.networkState,
+                                                 ourStatus, theirStatus)
+        if res:
+          peer.state(BeaconSync).statusMsg = theirStatus
 
     proc statusResp(peer: Peer, msg: StatusMsg)
 
@@ -120,12 +127,11 @@ p2pProtocol BeaconSync(version = 1,
   requestResponse:
     proc beaconBlocksByRange(
             peer: Peer,
-            headBlockRoot: Eth2Digest,
             startSlot: Slot,
             count: uint64,
             step: uint64) {.
             libp2pProtocol("beacon_blocks_by_range", 1).} =
-      trace "got range request", peer, count, startSlot, headBlockRoot, step
+      trace "got range request", peer, count, startSlot, step
 
       if count > 0'u64:
         let count = if step != 0: min(count, MAX_REQUESTED_BLOCKS.uint64) else: 1
@@ -133,7 +139,7 @@ p2pProtocol BeaconSync(version = 1,
         var results: array[MAX_REQUESTED_BLOCKS, BlockRef]
         let
           lastPos = min(count.int, results.len) - 1
-          firstPos = pool.getBlockRange(headBlockRoot, startSlot, step,
+          firstPos = pool.getBlockRange(pool.head.blck.root, startSlot, step,
                                         results.toOpenArray(0, lastPos))
         for i in firstPos.int .. lastPos.int:
           trace "wrote response block", slot = results[i].slot
@@ -158,77 +164,17 @@ p2pProtocol BeaconSync(version = 1,
 proc handleInitialStatus(peer: Peer,
                          state: BeaconSyncNetworkState,
                          ourStatus: StatusMsg,
-                         theirStatus: StatusMsg) {.async, gcsafe.} =
+                       theirStatus: StatusMsg): Future[bool] {.async, gcsafe.} =
   if theirStatus.forkDigest != state.forkDigest:
-    notice "Irrelevant peer", peer,
-           theirFork = theirStatus.forkDigest, ourFork = state.forkDigest
+    notice "Irrelevant peer",
+      peer, theirFork = theirStatus.forkDigest, ourFork = state.forkDigest
     await peer.disconnect(IrrelevantNetwork)
-    return
-
-  # TODO: onPeerConnected runs unconditionally for every connected peer, but we
-  # don't need to sync with everybody. The beacon node should detect a situation
-  # 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,
-          localHeadSlot = ourStatus.headSlot,
-          remoteHeadSlot = theirStatus.headSlot,
-          remoteHeadRoot = theirStatus.headRoot
-
-    let bestDiff = cmp((ourStatus.finalizedEpoch, ourStatus.headSlot),
-                       (theirStatus.finalizedEpoch, theirStatus.headSlot))
-    if bestDiff >= 0:
-      # Nothing to do?
-      debug "Nothing to sync", peer
-    else:
-      # TODO: Check for WEAK_SUBJECTIVITY_PERIOD difference and terminate the
-      # connection if it's too big.
-      var s = ourStatus.headSlot + 1
-      var theirStatus = theirStatus
-      while s <= theirStatus.headSlot:
-        let numBlocksToRequest = min(uint64(theirStatus.headSlot - s) + 1,
-                                     MAX_REQUESTED_BLOCKS)
-
-        debug "Requesting blocks", peer, remoteHeadSlot = theirStatus.headSlot,
-                                         ourHeadSlot = s,
-                                         numBlocksToRequest
-
-        # TODO: The timeout here is so high only because we fail to
-        # respond in time due to high CPU load in our single thread.
-        let blocks = await peer.beaconBlocksByRange(theirStatus.headRoot, s,
-                                                    numBlocksToRequest, 1'u64,
-                                                    timeout = 60.seconds)
-        if blocks.isSome:
-          info "got blocks", total = blocks.get.len
-          if blocks.get.len == 0:
-            info "Got 0 blocks while syncing", peer
-            break
-
-          state.importBlocks(blocks.get)
-          let lastSlot = blocks.get[^1].message.slot
-          if lastSlot <= s:
-            info "Slot did not advance during sync", peer
-            break
-
-          s = lastSlot + 1
-
-          # TODO: Maybe this shouldn't happen so often.
-          # The alternative could be watching up a timer here.
-
-          let statusResp = await peer.status(state.getCurrentStatus())
-          if statusResp.isSome:
-            theirStatus = statusResp.get
-          else:
-            # We'll ignore this error and we'll try to request
-            # another range optimistically. If that fails, the
-            # syncing will be interrupted.
-            discard
-        else:
-          error "Did not get any blocks from peer. Aborting sync."
-          break
-
-  except CatchableError as e:
-    warn "Failed to sync with peer", peer, err = e.msg
+    return false
+  debug "Peer connected", peer,
+                          localHeadSlot = ourStatus.headSlot,
+                          remoteHeadSlot = theirStatus.headSlot,
+                          remoteHeadRoot = theirStatus.headRoot
+  return true
 
 proc initBeaconSync*(network: Eth2Node,
                      blockPool: BlockPool,
@@ -238,4 +184,3 @@ proc initBeaconSync*(network: Eth2Node,
   networkState.blockPool = blockPool
   networkState.forkDigest = forkDigest
   networkState.onBeaconBlock = onBeaconBlock
-