BN, add timeout for sync workers which are waiting in queue.

AllTests-mainnet.md

@@ -829,6 +829,7 @@ OK: 7/7 Fail: 0/7 Skip: 0/7
 + [SyncQueue#Backward] Pass through established limits test                                  OK
 + [SyncQueue#Backward] Smoke test                                                            OK
 + [SyncQueue#Backward] Start and finish slots equal                                          OK
++ [SyncQueue#Backward] SyncQueue worker timeout test                                         OK
 + [SyncQueue#Backward] Two full requests success/fail                                        OK
 + [SyncQueue#Backward] getRewindPoint() test                                                 OK
 + [SyncQueue#Forward] Async unordered push test                                              OK
@@ -838,14 +839,15 @@ OK: 7/7 Fail: 0/7 Skip: 0/7
 + [SyncQueue#Forward] Pass through established limits test                                   OK
 + [SyncQueue#Forward] Smoke test                                                             OK
 + [SyncQueue#Forward] Start and finish slots equal                                           OK
++ [SyncQueue#Forward] SyncQueue worker timeout test                                          OK
 + [SyncQueue#Forward] Two full requests success/fail                                         OK
 + [SyncQueue#Forward] getRewindPoint() test                                                  OK
 + [SyncQueue] checkResponse() test                                                           OK
 + [SyncQueue] contains() test                                                                OK
 + [SyncQueue] getLastNonEmptySlot() test                                                     OK
 + [SyncQueue] hasEndGap() test                                                               OK
 ```
-OK: 24/24 Fail: 0/24 Skip: 0/24
+OK: 26/26 Fail: 0/26 Skip: 0/26
 ## Type helpers
 ```diff
 + BeaconBlock                                                                                OK
@@ -979,4 +981,4 @@ OK: 2/2 Fail: 0/2 Skip: 0/2
 OK: 9/9 Fail: 0/9 Skip: 0/9
 
 ---TOTAL---
-OK: 664/669 Fail: 0/669 Skip: 5/669
+OK: 666/671 Fail: 0/671 Skip: 5/671

beacon_chain/nimbus_beacon_node.nim

@@ -390,14 +390,15 @@ proc initFullNode(
       dag.cfg.DENEB_FORK_EPOCH, dag.cfg.MIN_EPOCHS_FOR_BLOB_SIDECARS_REQUESTS,
       SyncQueueKind.Forward, getLocalHeadSlot,
       getLocalWallSlot, getFirstSlotAtFinalizedEpoch, getBackfillSlot,
-      getFrontfillSlot, dag.tail.slot, blockVerifier)
+      getFrontfillSlot, dag.tail.slot, blockVerifier,
+      workerBlockWaitTimeout = chronos.seconds(1))
     backfiller = newSyncManager[Peer, PeerId](
       node.network.peerPool,
       dag.cfg.DENEB_FORK_EPOCH, dag.cfg.MIN_EPOCHS_FOR_BLOB_SIDECARS_REQUESTS,
       SyncQueueKind.Backward, getLocalHeadSlot,
       getLocalWallSlot, getFirstSlotAtFinalizedEpoch, getBackfillSlot,
       getFrontfillSlot, dag.backfill.slot, blockVerifier,
-      maxHeadAge = 0)
+      maxHeadAge = 0, workerBlockWaitTimeout = chronos.seconds(1))
     router = (ref MessageRouter)(
       processor: processor,
       network: node.network)

beacon_chain/sync/sync_manager.nim

@@ -71,6 +71,7 @@ type
     syncStatus*: string
     direction: SyncQueueKind
     ident*: string
+    pendingWorkerBlockWaitTime*: Duration
     flags: set[SyncManagerFlag]
 
   SyncMoment* = object
@@ -100,7 +101,7 @@ proc initQueue[A, B](man: SyncManager[A, B]) =
     man.queue = SyncQueue.init(A, man.direction, man.getFirstSlot(),
                                man.getLastSlot(), man.chunkSize,
                                man.getSafeSlot, man.blockVerifier,
-                               1, man.ident)
+                               1, man.ident, man.pendingWorkerBlockWaitTime)
   of SyncQueueKind.Backward:
     let
       firstSlot = man.getFirstSlot()
@@ -113,7 +114,8 @@ proc initQueue[A, B](man: SyncManager[A, B]) =
                     Slot(firstSlot - 1'u64)
     man.queue = SyncQueue.init(A, man.direction, startSlot, lastSlot,
                                man.chunkSize, man.getSafeSlot,
-                               man.blockVerifier, 1, man.ident)
+                               man.blockVerifier, 1, man.ident,
+                               man.pendingWorkerBlockWaitTime)
 
 proc newSyncManager*[A, B](pool: PeerPool[A, B],
                            denebEpoch: Epoch,
@@ -128,6 +130,7 @@ proc newSyncManager*[A, B](pool: PeerPool[A, B],
                            blockVerifier: BlockVerifier,
                            maxHeadAge = uint64(SLOTS_PER_EPOCH * 1),
                            chunkSize = uint64(SLOTS_PER_EPOCH),
+                           workerBlockWaitTimeout = InfiniteDuration,
                            flags: set[SyncManagerFlag] = {},
                            ident = "main"
                            ): SyncManager[A, B] =
@@ -153,6 +156,7 @@ proc newSyncManager*[A, B](pool: PeerPool[A, B],
     notInSyncEvent: newAsyncEvent(),
     direction: direction,
     ident: ident,
+    pendingWorkerBlockWaitTime: workerBlockWaitTimeout,
     flags: flags
   )
   res.initQueue()

beacon_chain/sync/sync_queue.nim

@@ -75,6 +75,7 @@ type
     readyQueue: HeapQueue[SyncResult[T]]
     rewind: Option[RewindPoint]
     blockVerifier: BlockVerifier
+    pendingWorkerBlockWaitTime: Duration
     ident*: string
 
 chronicles.formatIt SyncQueueKind: toLowerAscii($it)
@@ -120,8 +121,9 @@ proc getShortMap*[T](req: SyncRequest[T],
       res.add('|')
       continue
     if slot == data[cur].signed_block_header.message.slot:
-      for k in cur..<cur+MAX_BLOBS_PER_BLOCK:
-        if k >= lenu64(data) or slot != data[k].signed_block_header.message.slot:
+      for k in cur ..< cur + MAX_BLOBS_PER_BLOCK:
+        if k >= lenu64(data) or
+           slot != data[k].signed_block_header.message.slot:
           res.add('|')
           break
         else:
@@ -197,7 +199,8 @@ proc init*[T](t1: typedesc[SyncQueue], t2: typedesc[T],
               getSafeSlotCb: GetSlotCallback,
               blockVerifier: BlockVerifier,
               syncQueueSize: int = -1,
-              ident: string = "main"): SyncQueue[T] =
+              ident: string = "main",
+              pendingWorkerWaitTime = InfiniteDuration): SyncQueue[T] =
   ## Create new synchronization queue with parameters
   ##
   ## ``start`` and ``final`` are starting and final Slots.
@@ -267,6 +270,7 @@ proc init*[T](t1: typedesc[SyncQueue], t2: typedesc[T],
     inpSlot: start,
     outSlot: start,
     blockVerifier: blockVerifier,
+    pendingWorkerBlockWaitTime: pendingWorkerWaitTime,
     ident: ident
   )
 
@@ -311,18 +315,22 @@ proc wakeupWaiters[T](sq: SyncQueue[T], reset = false) =
     if not(item.future.finished()):
       item.future.complete()
 
-proc waitForChanges[T](sq: SyncQueue[T]): Future[bool] {.async: (raises: [CancelledError]).} =
+proc waitForChanges[T](sq: SyncQueue[T]): Future[bool] {.
+     async: (raises: [CancelledError]).} =
   ## Create new waiter and wait for completion from `wakeupWaiters()`.
-  let waitfut = Future[void].Raising([CancelledError]).init("SyncQueue.waitForChanges")
-  let waititem = SyncWaiter(future: waitfut)
+  let
+    waitfut =
+      Future[void].Raising([CancelledError]).init("SyncQueue.waitForChanges")
+    waititem = SyncWaiter(future: waitfut)
   sq.waiters.add(waititem)
   try:
     await waitfut
     return waititem.reset
   finally:
     sq.waiters.delete(sq.waiters.find(waititem))
 
-proc wakeupAndWaitWaiters[T](sq: SyncQueue[T]) {.async: (raises: [CancelledError]).} =
+proc wakeupAndWaitWaiters[T](sq: SyncQueue[T]) {.
+     async: (raises: [CancelledError]).} =
   ## This procedure will perform wakeupWaiters(true) and blocks until last
   ## waiter will be awakened.
   var waitChanges = sq.waitForChanges()
@@ -333,7 +341,8 @@ proc clearAndWakeup*[T](sq: SyncQueue[T]) =
   sq.pending.clear()
   sq.wakeupWaiters(true)
 
-proc resetWait*[T](sq: SyncQueue[T], toSlot: Option[Slot]) {.async: (raises: [CancelledError]).} =
+proc resetWait*[T](sq: SyncQueue[T], toSlot: Option[Slot]) {.
+     async: (raises: [CancelledError]).} =
   ## Perform reset of all the blocked waiters in SyncQueue.
   ##
   ## We adding one more waiter to the waiters sequence and
@@ -546,8 +555,9 @@ func getOpt(blobs: Opt[seq[BlobSidecars]], i: int): Opt[BlobSidecars] =
   else:
     Opt.none(BlobSidecars)
 
-iterator blocks[T](sq: SyncQueue[T],
-                   sr: SyncResult[T]): (ref ForkedSignedBeaconBlock, Opt[BlobSidecars]) =
+iterator blocks[T](
+    sq: SyncQueue[T],
+    sr: SyncResult[T]): (ref ForkedSignedBeaconBlock, Opt[BlobSidecars]) =
   case sq.kind
   of SyncQueueKind.Forward:
     for i in countup(0, len(sr.data) - 1):
@@ -577,6 +587,24 @@ proc notInRange[T](sq: SyncQueue[T], sr: SyncRequest[T]): bool =
   of SyncQueueKind.Backward:
     (sq.queueSize > 0) and (sr.lastSlot < sq.outSlot)
 
+func chunksCount*[T](sq: SyncQueue[T], sr: SyncRequest[T]): uint64 =
+  ## Returns number of chunks between current queue position and request's
+  ## position.
+  case sq.kind
+  of SyncQueueKind.Forward:
+    (sr.slot - sq.outSlot) div sq.chunkSize
+  of SyncQueueKind.Backward:
+    (sq.outSlot - sr.lastSlot) div sq.chunkSize
+
+func expectedTime[T](sq: SyncQueue[T], sr: SyncRequest[T]): Duration =
+  ## Returns expected wait time of specific pending request.
+  ## SR3 SR2 SR1 SR0 <CURRENT QUEUE POSITION>
+  ## SR3 expected pending time would be
+  ## <CHUNK SIZE> * <SR2 .. SR0 CHUNKS COUNT> * <BLOCK WAIT TIME>
+  nanoseconds(
+    int64(sq.chunkSize * sq.chunksCount(sr)) *
+      sq.pendingWorkerBlockWaitTime.nanoseconds)
+
 func numAlreadyKnownSlots[T](sq: SyncQueue[T], sr: SyncRequest[T]): uint64 =
   ## Compute the number of slots covered by a given `SyncRequest` that are
   ## already known and, hence, no longer relevant for sync progression.
@@ -606,11 +634,14 @@ func numAlreadyKnownSlots[T](sq: SyncQueue[T], sr: SyncRequest[T]): uint64 =
       # Entire request is still relevant.
       0
 
-proc push*[T](sq: SyncQueue[T], sr: SyncRequest[T],
-              data: seq[ref ForkedSignedBeaconBlock],
-              blobs: Opt[seq[BlobSidecars]],
-              maybeFinalized: bool = false,
-              processingCb: ProcessingCallback = nil) {.async: (raises: [CancelledError]).} =
+proc push*[T](
+    sq: SyncQueue[T],
+    sr: SyncRequest[T],
+    data: seq[ref ForkedSignedBeaconBlock],
+    blobs: Opt[seq[BlobSidecars]],
+    maybeFinalized: bool = false,
+    processingCb: ProcessingCallback = nil
+) {.async: (raises: [CancelledError]).} =
   logScope:
     sync_ident = sq.ident
     topics = "syncman"
@@ -631,8 +662,18 @@ proc push*[T](sq: SyncQueue[T], sr: SyncRequest[T],
   # all pending `push` requests if `request.slot` not in range.
   while true:
     if sq.notInRange(sr):
-      let reset = await sq.waitForChanges()
-      if reset:
+      let
+        resetWorker =
+          if sq.pendingWorkerBlockWaitTime == InfiniteDuration:
+            await sq.waitForChanges()
+          else:
+            try:
+              await sq.waitForChanges().wait(sq.expectedTime(sr))
+            except AsyncTimeoutError:
+              # Moving request to debts queue without checking pending list.
+              sq.debtsQueue.push(sr)
+              return
+      if resetWorker:
         # SyncQueue reset happens. We are exiting to wake up sync-worker.
         return
     else:
@@ -685,10 +726,8 @@ proc push*[T](sq: SyncQueue[T], sr: SyncRequest[T],
       # Nim versions, remove workaround and move `res` into for loop
       res: Result[void, VerifierError]
 
-    var i=0
     for blk, blb in sq.blocks(item):
       res = await sq.blockVerifier(blk[], blb, maybeFinalized)
-      inc(i)
 
       if res.isOk():
         goodBlock = some(blk[].slot)

tests/test_sync_manager.nim

@@ -9,7 +9,7 @@
 
 import std/[strutils, sequtils]
 import unittest2
-import chronos
+import chronos, chronos/unittest2/asynctests
 import ../beacon_chain/gossip_processing/block_processor,
        ../beacon_chain/sync/sync_manager,
        ../beacon_chain/spec/forks
@@ -1134,8 +1134,6 @@ suite "SyncManager test suite":
     check:
       groupedRes3.isErr()
 
-
-
   test "[SyncQueue#Forward] getRewindPoint() test":
     let aq = newAsyncQueue[BlockEntry]()
     block:
@@ -1207,3 +1205,69 @@ suite "SyncManager test suite":
       let safeSlot = getSafeSlot()
       for i in countdown(1023, 0):
         check queue.getRewindPoint(Slot(i), safeSlot) == safeSlot
+
+  template workerTimeoutTest(startSlot, finishSlot: Slot,
+                             kind: SyncQueueKind) =
+    let aq = newAsyncQueue[BlockEntry]()
+    var
+      chain = createChain(startSlot, finishSlot)
+      queue =
+        case kind
+        of SyncQueueKind.Forward:
+          SyncQueue.init(SomeTPeer, kind, startSlot, finishSlot,
+                         1'u64, getStaticSlotCb(startSlot),
+                         collector(aq), 1, "worker-timeout-test",
+                         100.milliseconds)
+        of SyncQueueKind.Backward:
+          SyncQueue.init(SomeTPeer, kind, finishSlot, startSlot,
+                         1'u64, getStaticSlotCb(finishSlot),
+                         collector(aq), 1, "worker-timeout-test",
+                         100.milliseconds)
+
+    let
+      p1 = SomeTPeer()
+      p2 = SomeTPeer()
+      p3 = SomeTPeer()
+      p4 = SomeTPeer()
+      r11 {.used.} = queue.pop(finishSlot, p1)
+      r12 = queue.pop(finishSlot, p2)
+      r13 = queue.pop(finishSlot, p3)
+      r14 = queue.pop(finishSlot, p4)
+      f12 = queue.push(
+        r12, chain.getSlice(Slot(0), r12), Opt.none(seq[BlobSidecars]))
+      f13 = queue.push(
+        r13, chain.getSlice(Slot(0), r13), Opt.none(seq[BlobSidecars]))
+      f14 = queue.push(
+        r14, chain.getSlice(Slot(0), r14), Opt.none(seq[BlobSidecars]))
+
+    check:
+      f12.finished() == false
+      f13.finished() == false
+      f14.finished() == false
+
+    # Current timeout value is 100.milliseconds per block.
+    await sleepAsync(400.milliseconds)
+
+    check:
+      f12.finished() == true
+      f13.finished() == true
+      f14.finished() == true
+
+    let
+      r22 = queue.pop(finishSlot, p2)
+      r23 = queue.pop(finishSlot, p3)
+      r24 = queue.pop(finishSlot, p4)
+
+    check:
+      r22.slot == r12.slot
+      r22.count == r12.count
+      r23.slot == r13.slot
+      r23.count == r13.count
+      r24.slot == r14.slot
+      r24.count == r14.count
+
+  asyncTest "[SyncQueue#Forward] SyncQueue worker timeout test":
+    workerTimeoutTest(Slot(0), Slot(200), SyncQueueKind.Forward)
+
+  asyncTest "[SyncQueue#Backward] SyncQueue worker timeout test":
+    workerTimeoutTest(Slot(0), Slot(200), SyncQueueKind.Backward)