Beacon sync maint update (#3789)

* Move peer ranking/assessment to the `beacon` scheduler/wrapper

why
  So the `worker` logic can receive that assessment via function
  argument. This reduces complexity for the tracer/replay logic.

* Reduce trace-log spamming when waiting for peers for manual target

details
  Currently, the waiting for peers feature can be activated
  only for the trace wrapper binary.

* Re-calibrate waiting/re-activation time for low ranking peers

why
  In some cases there is no expectation that a peer might be
  ready for immediate action.

* Not repeating some time-consuming failed fetch request again

details
  With long fetch timeout, the request is stashed and checked
  before sending. It will be rejected right away (without
  sending.)

* Code cosmetics, log & assert messages updates

Author: mjfh

execution_chain/core/chain/header_chain_cache.nim

@@ -127,8 +127,8 @@ const
     ## Insert `persist()` statements in bulk action every `MaxDeleteBatch`
     ## `del()` directives.
 
-  RaisePfx = "Header Cache: "
-    ## Message prefix used when bailing out raising an exception
+  MsgPfx = "Header Cache: "
+    ## Message prefix used when logging or raising an exception
 
 const
   HccDbInfoKey = 0.beaconHeaderKey
@@ -163,7 +163,7 @@ func decodePayload(data: seq[byte]; T: type): T =
   try:
     result = rlp.decode(data, T)
   except RlpError as e:
-    raiseAssert RaisePfx & "rlp.decode(" & $T & ") failed:" &
+    raiseAssert MsgPfx & "rlp.decode(" & $T & ") failed:" &
       " name=" & $e.name & " error=" & e.msg
 
 # ------------------------------------------------------------------------------
@@ -172,7 +172,7 @@ func decodePayload(data: seq[byte]; T: type): T =
 
 proc putInfo(db: KvtTxRef; state: HccDbInfo) =
   db.put(HccDbInfoKey.toOpenArray, encodePayload(state)).isOkOr:
-    raiseAssert RaisePfx & "put(info) failed: " & $error
+    raiseAssert MsgPfx & "put(info) failed: " & $error
 
 proc getInfo(db: KvtTxRef): Opt[HccDbInfo] =
   let data = db.get(HccDbInfoKey.toOpenArray).valueOr:
@@ -194,11 +194,11 @@ proc putHeader(db: KvtTxRef; h: Header) =
   ## of the parent header.
   let data = encodePayload(h)
   db.put(beaconHeaderKey(h.number).toOpenArray, data).isOkOr:
-    raiseAssert RaisePfx & "put(header) failed: " & $error
+    raiseAssert MsgPfx & "put(header) failed: " & $error
 
   let parNumData = (h.number-1).toBytesBE
   db.put(genericHashKey(h.parentHash).toOpenArray, parNumData).isOkOr:
-    raiseAssert RaisePfx & "put(number-1) failed: " & $error
+    raiseAssert MsgPfx & "put(number-1) failed: " & $error
 
 
 proc getNumber(db: KvtTxRef, hash: Hash32): Opt[BlockNumber] =
@@ -300,7 +300,7 @@ proc resolveFinHash(hc: HeaderChainRef, f: Hash32) =
     return
 
   if hc.chain.tryUpdatePendingFCU(f, number):
-    debug "PendingFCU resolved to block number",
+    debug MsgPfx & "pendingFCU resolved to block number",
       hash = f.short,
       number = number.bnStr
 
@@ -480,7 +480,7 @@ proc put*(
   if rev.len == 0:
     return ok()                                    # nothing to do
 
-  debug "HC updated",
+  debug MsgPfx & "updated",
     minNum=rev[^1].bnStr,
     maxNum=rev[0].bnStr,
     numHeaders=rev.len
@@ -598,8 +598,11 @@ proc commit*(hc: HeaderChainRef): Result[void,string] =
         return ok()
 
       # Impossible situation!
-      raiseAssert RaisePfx &
-        "Missing finalised " & fin.bnStr & " parent on FC module"
+      raiseAssert MsgPfx &
+        "Missing finalised " & fin.bnStr & " parent on FC module" &
+           ", base=" & hc.chain.baseNumber.bnStr &
+           ", head=" & hc.session.head.bnStr &
+           ", finalized=" & hc.chain.latestFinalizedBlockNumber.bnStr
 
   hc.session.mode = orphan
   err("Parent on FC module has been lost: obsolete branch segment")

execution_chain/sync/beacon.nim

@@ -17,6 +17,7 @@ import
   ../networking/p2p,
   ./beacon/worker/headers/headers_target,
   ./beacon/[beacon_desc, worker],
+  ./beacon/worker/classify,
   ./[sync_sched, wire_protocol]
 
 export
@@ -30,28 +31,29 @@ logScope:
 # ------------------------------------------------------------------------------
 
 proc runSetup(ctx: BeaconCtxRef): bool =
-  worker.setup(ctx, "RunSetup")
+  worker.setup(ctx, "Setup")
 
 proc runRelease(ctx: BeaconCtxRef) =
-  worker.release(ctx, "RunRelease")
+  worker.release(ctx, "Release")
 
 proc runDaemon(ctx: BeaconCtxRef): Future[Duration] {.async: (raises: []).} =
-  return worker.runDaemon(ctx, "RunDaemon")
+  return worker.runDaemon(ctx, "Daemon")
 
 proc runTicker(ctx: BeaconCtxRef) =
-  worker.runTicker(ctx, "RunTicker")
+  worker.runTicker(ctx, "Ticker")
 
 proc runStart(buddy: BeaconBuddyRef): bool =
-  worker.start(buddy, "RunStart")
+  worker.start(buddy, "Start")
 
 proc runStop(buddy: BeaconBuddyRef) =
-  worker.stop(buddy, "RunStop")
+  worker.stop(buddy, "Stop")
 
 proc runPool(buddy: BeaconBuddyRef; last: bool; laps: int): bool =
-  worker.runPool(buddy, last, laps, "RunPool")
+  worker.runPool(buddy, last, laps, "SyncMode")
 
 proc runPeer(buddy: BeaconBuddyRef): Future[Duration] {.async: (raises: []).} =
-  return worker.runPeer(buddy, "RunPeer")
+  let rank = buddy.classifyForFetching()
+  return worker.runPeer(buddy, rank, "Peer")
 
 # ------------------------------------------------------------------------------
 # Public functions

execution_chain/sync/beacon/worker.nim

@@ -55,7 +55,7 @@ proc start*(buddy: BeaconBuddyRef; info: static[string]): bool =
 proc stop*(buddy: BeaconBuddyRef; info: static[string]) =
   ## Clean up this peer
   if not buddy.ctx.hibernate: debug info & ": release peer", peer=buddy.peer,
-    throughput=buddy.only.thruPutStats.toMeanVar.psStr,
+    thPut=buddy.only.thPutStats.toMeanVar.psStr,
     nSyncPeers=(buddy.ctx.pool.nBuddies-1), state=($buddy.syncState)
   buddy.stopBuddy()
 
@@ -76,7 +76,7 @@ proc runTicker*(ctx: BeaconCtxRef; info: static[string]) =
     if ctx.pool.lastNoPeersLog + noPeersLogWaitInterval < now:
       ctx.pool.lastNoPeersLog = now
       debug info & ": no sync peers yet",
-        elapsed=(now - ctx.pool.lastPeerSeen).toStr,
+        ela=(now - ctx.pool.lastPeerSeen).toStr,
         nOtherPeers=ctx.node.peerPool.connectedNodes.len
 
 
@@ -129,7 +129,11 @@ proc runPool*(
   true # stop
 
 
-template runPeer*(buddy: BeaconBuddyRef; info: static[string]): Duration =
+template runPeer*(
+    buddy: BeaconBuddyRef;
+    rank: PeerRanking;
+    info: static[string];
+      ): Duration =
   ## Async/template
   ##
   ## This peer worker method is repeatedly invoked (exactly one per peer) while
@@ -141,16 +145,18 @@ template runPeer*(buddy: BeaconBuddyRef; info: static[string]): Duration =
   block body:
     if buddy.somethingToCollectOrUnstage():
 
-      # Classify sync peer (aka buddy) performance
-      let (fetchPerf {.inject.}, rank) = buddy.classifyForFetching()
-
       trace info & ": start processing", peer=buddy.peer,
-        throughput=buddy.only.thruPutStats.toMeanVar.psStr,
-        fetchPerf, rank=(if rank < 0: "n/a" else: $rank),
+        thPut=buddy.only.thPutStats.toMeanVar.psStr,
+        rankInfo=($rank.assessed),
+        rank=(if rank.ranking < 0: "n/a" else: $rank.ranking),
         nSyncPeers=buddy.ctx.pool.nBuddies, state=($buddy.syncState)
 
-      if fetchPerf == rankingTooLow:
-        bodyRc = workerIdleWaitInterval
+      if rank.assessed == rankingTooLow:
+        # Tell the scheduler to wait a bit longer before next invocation.
+        # The reasoning is that in case of a low rank labelling, all slots
+        # for peers downloading can be filled with higher ranking peers. And
+        # this situation would not change immediately.
+        bodyRc = workerIdleLongWaitInterval
         break body                                # done, exit
 
       # Download and process headers and blocks
@@ -191,9 +197,11 @@ template runPeer*(buddy: BeaconBuddyRef; info: static[string]): Duration =
 
       # End block: `actionLoop`
 
-    else:
+    elif buddy.ctx.pool.lastState == SyncState.idle:
       # Potentially a manual sync target set up
-      buddy.headersTargetActivate info
+      if not buddy.headersTargetActivate info:
+        bodyRc = workerIdleLongWaitInterval
+      break body
 
     # Idle sleep unless there is something to do
     if not buddy.somethingToCollectOrUnstage():

execution_chain/sync/beacon/worker/blocks.nim

@@ -125,7 +125,7 @@ template blocksCollect*(
               head=ctx.chain.latestNumber.bnStr,
               target=ctx.subState.head.bnStr,
               targetHash=ctx.subState.headHash.short,
-              throughput=buddy.blkThroughput,
+              thPut=buddy.blkThroughput,
               nSyncPeers=ctx.pool.nBuddies
             ctx.pool.lastSyncUpdLog = Moment.now()
             nImported = 0
@@ -177,7 +177,7 @@ template blocksCollect*(
           head=ctx.chain.latestNumber.bnStr,
           target=ctx.subState.head.bnStr,
           targetHash=ctx.subState.headHash.short,
-          throughput=buddy.blkThroughput,
+          thPut=buddy.blkThroughput,
           nSyncPeers=ctx.pool.nBuddies
         ctx.pool.lastSyncUpdLog = Moment.now()
 

execution_chain/sync/beacon/worker/blocks/blocks_fetch.nim

@@ -29,8 +29,12 @@ proc getBlockBodies(
       {.async: (raises: []).} =
   ## Wrapper around `getBlockHeaders()`
   let start = Moment.now()
-  var resp: BlockBodiesPacket
 
+  if buddy.only.failedReq.state == SyncState.blocks and
+     buddy.only.failedReq.blockHash == req.blockHashes[0]:
+    return err((EAlreadyTriedAndFailed,"","",Moment.now()-start))
+
+  var resp: BlockBodiesPacket
   try:
     resp = (await buddy.peer.getBlockBodies(
       req, fetchBodiesRlpxTimeout)).valueOr:
@@ -86,17 +90,21 @@ template fetchBodies*(
           buddy.ctrl.zombie = true
         of ECatchableError:
           buddy.bdyFetchRegisterError()
+          discard buddy.only.thPutStats.hdr.bpsSample(elapsed, 0)
+        of EAlreadyTriedAndFailed:
+          # Just return `failed` (no error count or throughput stats)
+          discard
 
         chronicles.info trEthRecvReceivedBlockBodies & " error", peer, nReq,
-          elapsed=rc.error.elapsed.toStr, state=($buddy.syncState),
+          ela=rc.error.elapsed.toStr, state=($buddy.syncState),
           error=rc.error.name, msg=rc.error.msg, nErrors=buddy.nErrors.fetch.bdy
         break body                                  # return err()
 
     # Evaluate result
     if rc.isErr or buddy.ctrl.stopped:
       buddy.bdyFetchRegisterError()
       trace trEthRecvReceivedBlockBodies, peer, nReq, nResp=0,
-        elapsed=elapsed.toStr, state=($buddy.syncState),
+        ela=elapsed.toStr, state=($buddy.syncState),
         nErrors=buddy.nErrors.fetch.bdy
       break body                                    # return err()
 
@@ -107,17 +115,29 @@ template fetchBodies*(
         # Bogus peer returning additional rubbish
         buddy.bdyFetchRegisterError(forceZombie=true)
       else:
-        # Data not avail but fast enough answer: degrade througput stats only
-        discard buddy.only.thruPutStats.blk.bpsSample(elapsed, 0)
+        # No data available. For a fast enough rejection response, the
+        # througput stats are degraded, only.
+        discard buddy.only.thPutStats.blk.bpsSample(elapsed, 0)
+
+        # Slow response, definitely not fast enough
         if fetchBodiesErrTimeout <= elapsed:
           buddy.bdyFetchRegisterError(slowPeer=true)
+
+          # Do not repeat the same time-consuming failed request
+          buddy.only.failedReq = BuddyFirstFetchReq(
+            state:     SyncState.blocks,
+            blockHash: request.blockHashes[0])
+
       trace trEthRecvReceivedBlockBodies, peer, nReq, nResp=b.len,
-        elapsed=elapsed.toStr, state=($buddy.syncState),
+        ela=elapsed.toStr, state=($buddy.syncState),
         nErrors=buddy.nErrors.fetch.bdy
       break body                                    # return err()
 
     # Update download statistics
-    let bps = buddy.only.thruPutStats.blk.bpsSample(elapsed, b.getEncodedLength)
+    let bps = buddy.only.thPutStats.blk.bpsSample(elapsed, b.getEncodedLength)
+
+    # Request did not fail
+    buddy.only.failedReq.reset
 
     # Ban an overly slow peer for a while when observed consecutively.
     if fetchBodiesErrTimeout < elapsed:
@@ -127,7 +147,7 @@ template fetchBodies*(
       buddy.ctx.pool.lastSlowPeer = Opt.none(Hash)  # not last one or not error
 
     trace trEthRecvReceivedBlockBodies, peer, nReq, nResp=b.len,
-      elapsed=elapsed.toStr, throughput=(bps.toIECb(1) & "ps"),
+      ela=elapsed.toStr, thPut=(bps.toIECb(1) & "ps"),
       state=($buddy.syncState), nErrors=buddy.nErrors.fetch.bdy
 
     bodyRc = Opt[seq[BlockBody]].ok(b)

execution_chain/sync/beacon/worker/blocks/blocks_helpers.nim

@@ -45,7 +45,7 @@ func blkSessionStopped*(ctx: BeaconCtxRef): bool =
 
 func blkThroughput*(buddy: BeaconBuddyRef): string =
   ## Print throuhput sratistics
-  buddy.only.thruPutStats.blk.toMeanVar.psStr
+  buddy.only.thPutStats.blk.toMeanVar.psStr
 
 # ------------------------------------------------------------------------------
 # End

execution_chain/sync/beacon/worker/classify.nim

@@ -28,9 +28,7 @@ func somethingToCollectOrUnstage*(buddy: BeaconBuddyRef): bool =
   false
 
 
-func classifyForFetching*(
-    buddy: BeaconBuddyRef;
-      ): tuple[info: DownloadPerformance, ranking: int] =
+func classifyForFetching*(buddy: BeaconBuddyRef): PeerRanking =
   ## Rank and classify peers by whether they should be used for fetching
   ## data.
   ##
@@ -49,7 +47,7 @@ func classifyForFetching*(
       return (qSlotsAvail, -1)
 
     template hdr(b: BeaconBuddyRef): StatsCollect =
-      b.only.thruPutStats.hdr
+      b.only.thPutStats.hdr
 
     # Are there throughput data available for this peer (aka buddy), at all?
     if buddy.hdr.samples == 0:
@@ -79,7 +77,7 @@ func classifyForFetching*(
       return (qSlotsAvail, -1)
 
     template blk(b: BeaconBuddyRef): StatsCollect =
-      b.only.thruPutStats.blk
+      b.only.thPutStats.blk
 
     if buddy.blk.samples == 0:
       return (notEnoughData, -1)

execution_chain/sync/beacon/worker/headers.nim

@@ -138,7 +138,7 @@ template headersCollect*(buddy: BeaconBuddyRef; info: static[string]) =
               base=ctx.chain.baseNumber.bnStr,
               head=ctx.chain.latestNumber.bnStr,
               target=ctx.hdrCache.head.bnStr,
-              throughput=buddy.hdrThroughput,
+              thPut=buddy.hdrThroughput,
               nSyncPeers=ctx.pool.nBuddies
             ctx.pool.lastSyncUpdLog = Moment.now()
             nStashed = 0
@@ -183,7 +183,7 @@ template headersCollect*(buddy: BeaconBuddyRef; info: static[string]) =
           base=ctx.chain.baseNumber.bnStr,
           head=ctx.chain.latestNumber.bnStr,
           target=ctx.hdrCache.head.bnStr,
-          throughput=buddy.hdrThroughput,
+          thPut=buddy.hdrThroughput,
           nSyncPeers=ctx.pool.nBuddies
         ctx.pool.lastSyncUpdLog = Moment.now()