Add missing percentage.

Fix edge case when SQ uses single supernode to get sidecars.
Adopt test.

Author: cheatfate

beacon_chain/sync/sync_overseer2.nim

@@ -101,8 +101,7 @@ template blobsCount(blck: ForkedSignedBeaconBlock): int =
 func slimLog(blck: ref ForkedSignedBeaconBlock): string =
   "(" & $blck.kind & ",slot:" & $blck[].slot() &
     ",root:" & shortLog(blck[].root()) &
-    ",parent_root:" & shortLog(blck[].parent_root()) &
-    ",blobs_count:" & $blck[].blobsCount() & ")"
+    ",parent_root:" & shortLog(blck[].parent_root()) & ")"
 
 func slimLog(blocks: openArray[ref ForkedSignedBeaconBlock]): string =
   "[" & blocks.mapIt(slimLog(it)).join(",") & "]"
@@ -206,22 +205,34 @@ func getColumnsFillRate(
 func getMissingColumnsLog(
     overseer: SyncOverseerRef2,
     blocks: openArray[ref ForkedSignedBeaconBlock]
-): string =
-  var res: seq[string]
+): (string, string) =
+  var
+    res: seq[string]
+    missingCount = 0.0
+    totalCount = 0.0
+
+  let blocksColumnsCount = float(len(overseer.columnQuarantine[].custodyMap))
+
   for blck in blocks:
     withBlck(blck[]):
       when consensusFork == ConsensusFork.Fulu:
-        res.add(
-          if len(forkyBlck.message.body.blob_kzg_commitments) == 0:
-            shortLog(forkyBlck.root) & ":[]"
-          else:
-            let map =
-              overseer.columnQuarantine[].getMissingColumnsMap(forkyBlck.root)
-            shortLog(forkyBlck.root) & ":" & $map
-        )
+        if len(forkyBlck.message.body.blob_kzg_commitments) > 0:
+          let map =
+            overseer.columnQuarantine[].getMissingColumnsMap(forkyBlck.root)
+          res.add(shortLog(forkyBlck.root) & ":" & $map)
+          missingCount += float(len(map))
+          totalCount += blocksColumnsCount
       else:
         raiseAssert "Unsupported fork"
-  "[" & res.join(",") & "]"
+
+  let missing =
+    if totalCount > 0.0:
+      ((missingCount * 100.0) / totalCount).formatBiggestFloat(ffDecimal, 2) &
+        "%"
+    else:
+      "0.00%"
+
+  (missing, "[" & res.join(",") & "]")
 
 func getLastSeenHeadLog(
     overseer: SyncOverseerRef2
@@ -1986,12 +1997,14 @@ proc doRangeSidecarsStep(
             # blocks.
             (false, false)
 
-        let missingLog = overseer.getMissingColumnsLog(blocks)
+        let (missingCount, missingLog) =
+          overseer.getMissingColumnsLog(blocks)
 
         debug "Peer columns compatibility",
            custody_map = shortLog(custodyMap),
            peer_map = shortLog(peerMap),
            intersect_map = shortLog(intersectMap),
+           missing_count = missingCount,
            missing_log = missingLog
 
         if (len(blocks) > 0) and (columnsNeeded and not(columnsHave)):
@@ -2000,6 +2013,7 @@ proc doRangeSidecarsStep(
             custody_map = shortLog(custodyMap),
             peer_map = shortLog(peerMap),
             intersect_map = shortLog(intersectMap),
+            missing_count = missingCount,
             missing_log = missingLog
           overseer.ssqueue(direction).push(request)
           return true
@@ -2022,6 +2036,7 @@ proc doRangeSidecarsStep(
             columns_map = getShortMap(request, intersectMap, data.toSeq()),
             peer_map = shortLog(peerMap),
             intersection_map = shortLog(intersectMap),
+            missing_count = missingCount,
             columns = slimLog(data.asSeq()),
             missing_log = missingLog
 
@@ -2064,6 +2079,7 @@ proc doRangeSidecarsStep(
             custody_map = shortLog(custodyMap),
             peer_map = shortLog(peerMap),
             intersect_map = shortLog(intersectMap),
+            missing_count = missingCount,
             missing_log = missingLog,
             columns_needed = columnsNeeded, columns_have = columnsHave
 

beacon_chain/sync/sync_queue.nim

@@ -7,7 +7,8 @@
 
 {.push raises: [].}
 
-import std/[deques, heapqueue, tables, strutils, sequtils, math, typetraits]
+import std/[
+  deques, heapqueue, sets, tables, strutils, sequtils, math, typetraits]
 import stew/base10, chronos, chronicles, results
 import
   ../spec/[helpers, forks, column_map],
@@ -53,6 +54,7 @@ type
 
   ColumnCompleteness* = object
     map: ColumnMap
+    keys: HashSet[string]
     done: bool
 
   SyncRequest*[T] = object
@@ -334,6 +336,7 @@ func isComplete[M, N](
     peer: M,
     criteria: var N
 ): bool =
+  mixin getKey
   if criteria.done:
     return true
   when N is BlockCompleteness:
@@ -343,6 +346,10 @@ func isComplete[M, N](
       # If criteria's map is empty, it means that we do not need columns for
       # the range.
       return true
+    if $(peer.getKey()) in criteria.keys:
+      # Columns was already downloaded from this peer for the range, there is no
+      # reason to request it one more time.
+      return true
     # If the peer has columns that we are still missing, we should return
     # `false`, so that peer will get request for that range, but if the peer
     # does not have the columns we need, we should return `true`.
@@ -377,13 +384,20 @@ func fillCompleteness[M](
     peer: M,
     missingMap: Opt[ColumnMap],
     done: bool,
+    storePeer: bool,
     criteria: var ColumnCompleteness
 ) =
+  mixin getKey
+
   if done:
     criteria.done = true
     criteria.map = ColumnMap()
     return
 
+  if storePeer:
+    let key = peer.getKey()
+    criteria.keys.incl($key)
+
   if missingMap.isSome():
     criteria.map = missingMap.get()
 
@@ -1028,7 +1042,7 @@ proc push*[M, N](sq: SyncQueue[M, N], requests: openArray[SyncRequest[M]]) =
     elif N is ColumnCompleteness:
       sq.fillCompleteness(
         sq.requests[pos.qindex].data, request.item, Opt.none(ColumnMap),
-        done = false, sq.requests[pos.qindex].completeness)
+        done = false, storePeer = false, sq.requests[pos.qindex].completeness)
     sq.del(pos)
 
 proc push*[M, N](sq: SyncQueue[M, N], sr: SyncRequest[M]) =
@@ -1141,7 +1155,7 @@ proc push*[M, N](
       return SyncPushResponse(
         code: SyncProcessError.NoRelevant, count: 0'i64)
 
-  template fillCompleteness(pdone, pblck: untyped) =
+  template fillCompleteness(pdone, pblck, pstore: untyped) =
     when N is BlockCompleteness:
       sq.fillCompleteness(
         sq.requests[position.qindex].data, sr.item, done = pdone,
@@ -1151,11 +1165,13 @@ proc push*[M, N](
         let map = sq.getMissingMap(data, pblck.get().root)
         sq.fillCompleteness(
           sq.requests[position.qindex].data, sr.item, Opt.some(map),
-          done = pdone, sq.requests[position.qindex].completeness)
+          done = pdone, storePeer = pstore,
+          sq.requests[position.qindex].completeness)
       else:
         sq.fillCompleteness(
           sq.requests[position.qindex].data, sr.item, Opt.none(ColumnMap),
-          done = pdone, sq.requests[position.qindex].completeness)
+          done = pdone, storePeer = pstore,
+          sq.requests[position.qindex].completeness)
 
   # This is backpressure handling algorithm, this algorithm is blocking
   # all pending `push` requests if `request` is not in range.
@@ -1234,18 +1250,18 @@ proc push*[M, N](
       # peers returns empty response for the same range.
       if sq.requests[position.qindex].voidsCount >= sq.requestsCount:
         when N is BlockCompleteness:
-          fillCompleteness(true, Opt.none(BlockId))
+          fillCompleteness(true, Opt.none(BlockId), false)
           sq.advanceQueue(res)
         elif N is ColumnCompleteness:
           let localMap = sq.cbGetLocalColumnMap()
           # If completeness map was changed it proves that specific range is
           # not actually empty and we should not move forward.
           if sq.requests[position.qindex].completeness.map != localMap:
-            fillCompleteness(false, Opt.none(BlockId))
+            fillCompleteness(false, Opt.none(BlockId), false)
           else:
-            fillCompleteness(true, Opt.none(BlockId))
+            fillCompleteness(true, Opt.none(BlockId), false)
       else:
-        fillCompleteness(false, Opt.none(BlockId))
+        fillCompleteness(false, Opt.none(BlockId), false)
 
     of SyncProcessError.Duplicate:
       # Duplicate responses does not affect failures count
@@ -1260,7 +1276,7 @@ proc push*[M, N](
             topics = "sync"
 
       sq.gapList.reset()
-      fillCompleteness(true, Opt.none(BlockId))
+      fillCompleteness(true, Opt.none(BlockId), false)
       sq.advanceQueue(res)
 
     of SyncProcessError.MissingSidecars:
@@ -1275,7 +1291,7 @@ proc push*[M, N](
             topics = "sync"
 
       inc(sq.requests[position.qindex].failuresCount)
-      fillCompleteness(false, pres.blck)
+      fillCompleteness(false, pres.blck, true)
       sq.del(position)
       res = 0'i64
 
@@ -1292,7 +1308,7 @@ proc push*[M, N](
             topics = "sync"
 
       inc(sq.requests[position.qindex].failuresCount)
-      fillCompleteness(false, pres.blck)
+      fillCompleteness(false, pres.blck, false)
       sq.del(position)
       res = 0'i64
 
@@ -1310,7 +1326,7 @@ proc push*[M, N](
 
       sr.item.updateScore(PeerScoreUnviableFork)
       inc(sq.requests[position.qindex].failuresCount)
-      fillCompleteness(false, pres.blck)
+      fillCompleteness(false, pres.blck, false)
       sq.del(position)
       res = 0'i64
 
@@ -1331,7 +1347,7 @@ proc push*[M, N](
       sq.rewardForGaps(PeerScoreMissingValues)
       sq.gapList.reset()
       inc(sq.requests[position.qindex].failuresCount)
-      fillCompleteness(false, pres.blck)
+      fillCompleteness(false, pres.blck, false)
       sq.del(position)
       res = 0'i64
 
@@ -1351,7 +1367,7 @@ proc push*[M, N](
             topics = "sync"
 
       sr.item.updateScore(PeerScoreMissingValues)
-      fillCompleteness(false, pres.blck)
+      fillCompleteness(false, pres.blck, false)
       sq.del(position)
       res = 0'i64
 
@@ -1364,7 +1380,7 @@ proc push*[M, N](
       if sr.hasEndGap(data):
         sq.gapList.add(GapItem.init(sr))
 
-      fillCompleteness(true, Opt.none(BlockId))
+      fillCompleteness(true, Opt.none(BlockId), false)
       sq.advanceQueue(res)
     of SyncProcessError.NoRelevant:
       raiseAssert "Processor should not return this error code"
@@ -1392,7 +1408,7 @@ proc push*[M, N](
   except CancelledError as exc:
     let pos = sq.find(sr)
     if pos.isSome():
-      fillCompleteness(false, Opt.none(BlockId))
+      fillCompleteness(false, Opt.none(BlockId), false)
       sq.del(pos.get())
     raise exc
   finally:

tests/test_sync_manager.nim

@@ -10,7 +10,7 @@
 
 import std/sequtils
 import unittest2
-import chronos, stew/base10, chronos/unittest2/asynctests
+import chronos, stew/base10, chronos/unittest2/asynctests, libp2p/peerid
 import ../beacon_chain/networking/peer_scores
 import ../beacon_chain/gossip_processing/block_processor,
        ../beacon_chain/sync/[sync_queue, response_utils],
@@ -19,6 +19,7 @@ import ../beacon_chain/gossip_processing/block_processor,
 type
   SomeTPeer = ref object
     id: string
+    peerId: PeerId
     score: int
     map: ColumnMap
 
@@ -27,8 +28,8 @@ type
 func init(t: typedesc[SomeTPeer], id: string, score = 1000): SomeTPeer =
   SomeTPeer(id: id, score: score)
 
-func init(t: typedesc[SomeTPeer], id: string, map: ColumnMap): SomeTPeer =
-  SomeTPeer(id: id, map: map)
+proc init(t: typedesc[SomeTPeer], id: string, map: ColumnMap): SomeTPeer =
+  SomeTPeer(id: id, map: map, peerId: PeerId.random().get())
 
 func init(t: typedesc[ColumnMap], columns: openArray[int]): ColumnMap =
   var res = columns.mapIt(ColumnIndex(it))
@@ -37,6 +38,9 @@ func init(t: typedesc[ColumnMap], columns: openArray[int]): ColumnMap =
 func `$`(peer: SomeTPeer): string =
   "peer#" & peer.id
 
+func getKey*(peer: SomeTPeer): PeerId =
+  peer.peerId
+
 template shortLog(peer: SomeTPeer): string =
   $peer
 
@@ -1613,6 +1617,8 @@ suite "SyncManager test suite":
         peer8 = SomeTPeer.init("8", getLocalMap())
         peer9 = SomeTPeer.init("9", getLocalMap())
         peer10 = SomeTPeer.init("10", getLocalMap())
+        peer11 = SomeTPeer.init("11", getLocalMap())
+        peer12 = SomeTPeer.init("12", getLocalMap())
 
       let
         r1 = sq.pop(Slot(127), peer1)
@@ -1676,7 +1682,7 @@ suite "SyncManager test suite":
       check p8.code == SyncProcessError.MissingSidecars
 
       let
-        r9 = sq.pop(Slot(127), peer8)
+        r9 = sq.pop(Slot(127), peer9)
         (d9, c9) = createFuluChain(r9, r9.item.map)
 
       var columns9 =
@@ -1690,17 +1696,18 @@ suite "SyncManager test suite":
 
       # Finish this range with full 32 blocks and columns
       let
-        r10 = sq.pop(Slot(127), peer8)
+        r10 = sq.pop(Slot(127), peer10)
         (d10, c10) = createFuluChain(r10, r10.item.map)
 
       let p10 = await sq.push(r10, d10, c10)
       check p10.code == SyncProcessError.NoError
 
       let
-        r11 = sq.pop(Slot(127), peer9)
-        r12 = sq.pop(Slot(127), peer10)
+        r11 = sq.pop(Slot(127), peer11)
+        r12 = sq.pop(Slot(127), peer12)
         r13 = sq.pop(Slot(127), peer8)
         r14 = sq.pop(Slot(127), peer9)
+        r15 = sq.pop(Slot(127), peer10)
         (d11, c11) = createFuluChain(r11, r11.item.map)
         (d12, c12) = createFuluChain(r12, r12.item.map)
 
@@ -1709,6 +1716,7 @@ suite "SyncManager test suite":
         r12.isEmpty() == false
         r13.isEmpty() == true
         r14.isEmpty() == true
+        r15.isEmpty() == true
 
       let p11 = await sq.push(r11, d11, c11)
       check p11.code == SyncProcessError.NoError