Fix assertion on syncing. (#7315)

* Fix assertion on empty response processing.

* Update AllTests.

Author: cheatfate

AllTests-mainnet.md

@@ -976,13 +976,15 @@ AllTests-mainnet
 + [SyncManager] groupBlobs() test                                                            OK
 + [SyncQueue# & Backward] Combination of missing parent and good blocks [3 peers] test       OK
 + [SyncQueue# & Backward] Empty responses should not advance queue until other peers will no OK
++ [SyncQueue# & Backward] Empty responses should not be accounted [3 peers] test             OK
 + [SyncQueue# & Backward] Failure request push test                                          OK
 + [SyncQueue# & Backward] Invalid block [3 peers] test                                       OK
 + [SyncQueue# & Backward] Smoke [3 peers] test                                               OK
 + [SyncQueue# & Backward] Smoke [single peer] test                                           OK
 + [SyncQueue# & Backward] Unviable block [3 peers] test                                      OK
 + [SyncQueue# & Forward] Combination of missing parent and good blocks [3 peers] test        OK
 + [SyncQueue# & Forward] Empty responses should not advance queue until other peers will not OK
++ [SyncQueue# & Forward] Empty responses should not be accounted [3 peers] test              OK
 + [SyncQueue# & Forward] Failure request push test                                           OK
 + [SyncQueue# & Forward] Invalid block [3 peers] test                                        OK
 + [SyncQueue# & Forward] Smoke [3 peers] test                                                OK

beacon_chain/sync/sync_queue.nim

@@ -37,9 +37,13 @@ type
   SyncQueueKind* {.pure.} = enum
     Forward, Backward
 
+  SyncRequestFlag* {.pure.} = enum
+    Void
+
   SyncRequest*[T] = object
     kind*: SyncQueueKind
     data*: SyncRange
+    flags*: set[SyncRequestFlag]
     item*: T
 
   SyncQueueItem[T] = object
@@ -472,11 +476,11 @@ func init*[T](t1: typedesc[SyncQueue], t2: typedesc[T],
     ident: ident
   )
 
-func contains[T](requests: openArray[SyncRequest[T]], source: T): bool =
-  for req in requests:
-    if req.item == source:
-      return true
-  false
+func searchPeer[T](requests: openArray[SyncRequest[T]], source: T): int =
+  for index, request in requests.pairs():
+    if request.item == source:
+      return index
+  -1
 
 func find[T](sq: SyncQueue[T], req: SyncRequest[T]): Opt[SyncPosition] =
   if len(sq.requests) == 0:
@@ -539,7 +543,8 @@ proc pop*[T](sq: SyncQueue[T], peerMaxSlot: Slot, item: T): SyncRequest[T] =
   var count = 0
   for qitem in sq.requests.mitems():
     if len(qitem.requests) < sq.requestsCount:
-      if item notin qitem.requests:
+      let sindex = qitem.requests.searchPeer(item)
+      if sindex < 0:
         return
           if qitem.data.slot > peerMaxSlot:
             # Peer could not satisfy our request, returning empty one.
@@ -551,7 +556,9 @@ proc pop*[T](sq: SyncQueue[T], peerMaxSlot: Slot, item: T): SyncRequest[T] =
             qitem.requests.add(request)
             request
       else:
-        inc(count)
+        if SyncRequestFlag.Void notin qitem.requests[sindex].flags:
+          # We only count non-empty requests.
+          inc(count)
 
   doAssert(count < sq.requestsCount,
            "You should not pop so many requests for single peer")
@@ -835,6 +842,10 @@ proc push*[T](
 
       sr.item.updateStats(SyncResponseKind.Empty, 1'u64)
       inc(sq.requests[position.qindex].voidsCount)
+      # Mark empty request in queue, so this range will not be requested by
+      # the same peer.
+      sq.requests[position.qindex].requests[position.sindex].flags.incl(
+        SyncRequestFlag.Void)
       sq.gapList.add(GapItem.init(sr))
       # With empty response - advance only when `requestsCount` of different
       # peers returns empty response for the same range.

tests/test_sync_manager.nim

@@ -855,6 +855,88 @@ suite "SyncManager test suite":
         sq.inpSlot == finishSlot
         sq.outSlot == finishSlot
 
+    asyncTest "[SyncQueue# & " & $kind & "] Empty responses should not " &
+              "be accounted [3 peers] test":
+      var emptyResponse: seq[ref ForkedSignedBeaconBlock]
+      let
+        scenario =
+          case kind
+          of SyncQueueKind.Forward:
+            [
+              (Slot(0) .. Slot(31), Opt.none(VerifierError)),
+              (Slot(32) .. Slot(63), Opt.none(VerifierError)),
+              (Slot(64) .. Slot(95), Opt.none(VerifierError)),
+              (Slot(96) .. Slot(127), Opt.none(VerifierError)),
+              (Slot(128) .. Slot(159), Opt.none(VerifierError))
+            ]
+          of SyncQueueKind.Backward:
+            [
+              (Slot(128) .. Slot(159), Opt.none(VerifierError)),
+              (Slot(96) .. Slot(127), Opt.none(VerifierError)),
+              (Slot(64) .. Slot(95), Opt.none(VerifierError)),
+              (Slot(32) .. Slot(63), Opt.none(VerifierError)),
+              (Slot(0) .. Slot(31), Opt.none(VerifierError))
+            ]
+        verifier = setupVerifier(kind, scenario)
+        sq =
+          case kind
+          of SyncQueueKind.Forward:
+            SyncQueue.init(SomeTPeer, kind, Slot(0), Slot(159),
+                           32'u64, # 32 slots per request
+                           3, # 3 concurrent requests
+                           2, # 2 failures allowed
+                           getStaticSlotCb(Slot(0)),
+                           verifier.collector)
+          of SyncQueueKind.Backward:
+            SyncQueue.init(SomeTPeer, kind, Slot(159), Slot(0),
+                           32'u64, # 32 slots per request
+                           3, # 3 concurrent requests
+                           2, # 2 failures allowed
+                           getStaticSlotCb(Slot(159)),
+                           verifier.collector)
+        slots =
+          case kind
+          of SyncQueueKind.Forward:
+            @[Slot(0), Slot(32), Slot(64), Slot(96), Slot(128)]
+          of SyncQueueKind.Backward:
+            @[Slot(128), Slot(96), Slot(64), Slot(32), Slot(0)]
+        peer1 = SomeTPeer.init("1")
+        peer2 = SomeTPeer.init("2")
+        peer3 = SomeTPeer.init("3")
+
+      let
+        r11 = sq.pop(Slot(159), peer1)
+        r21 = sq.pop(Slot(159), peer2)
+      await sq.push(r11, emptyResponse, Opt.none(seq[BlobSidecars]))
+      let
+        r12 = sq.pop(Slot(159), peer1)
+        r13 = sq.pop(Slot(159), peer1)
+        # This should not raise an assertion, as the previously sent empty
+        # response should not be taken into account.
+        r14 = sq.pop(Slot(159), peer1)
+
+      expect AssertionError:
+        let r1e {.used.} = sq.pop(Slot(159), peer1)
+
+      check:
+        r11.data.slot == slots[0]
+        r12.data.slot == slots[1]
+        r13.data.slot == slots[2]
+        r14.data.slot == slots[3]
+
+      # Scenario requires some finish steps
+      await sq.push(r21, createChain(r21.data), Opt.none(seq[BlobSidecars]))
+      let r22 = sq.pop(Slot(159), peer2)
+      await sq.push(r22, createChain(r22.data), Opt.none(seq[BlobSidecars]))
+      let r23 = sq.pop(Slot(159), peer2)
+      await sq.push(r23, createChain(r23.data), Opt.none(seq[BlobSidecars]))
+      let r24 = sq.pop(Slot(159), peer2)
+      await sq.push(r24, createChain(r24.data), Opt.none(seq[BlobSidecars]))
+      let r35 = sq.pop(Slot(159), peer3)
+      await sq.push(r35, createChain(r35.data), Opt.none(seq[BlobSidecars]))
+
+      await noCancel wait(verifier.verifier, 2.seconds)
+
     asyncTest "[SyncQueue# & " & $kind & "] Combination of missing parent " &
               "and good blocks [3 peers] test":
       let