never request blocks before `safeSlot` in sync (#3512)

Follows up on https://github.com/status-im/nimbus-eth2/pull/3461 which
ensured that repeated `beaconBlocksByRange` requests get shrinked to
account for potential out-of-band advancements to `safeSlot`, with
similar logic for the initial request.

beacon_chain/sync/sync_queue.nim

@@ -758,24 +758,38 @@ proc handlePotentialSafeSlotAdvancement[T](sq: SyncQueue[T]) =
   # through an out-of-band mechanism, e.g., VC / REST.
   # If that happens, advance to the new `safeSlot` to avoid repeating requests
   # for data that is considered immutable and no longer relevant.
+  let safeSlot = sq.getSafeSlot()
+  func numSlotsBehindSafeSlot(slot: Slot): uint64 =
+    case sq.kind
+    of SyncQueueKind.Forward:
+      if safeSlot > slot:
+        safeSlot - slot
+      else:
+        0
+    of SyncQueueKind.Backward:
+      if slot > safeSlot:
+        slot - safeSlot
+      else:
+        0
+
   let
-    safeSlot = sq.getSafeSlot()
-    numSlotsAdvanced: uint64 =
+    numOutSlotsAdvanced = sq.outSlot.numSlotsBehindSafeSlot
+    numInpSlotsAdvanced =
       case sq.kind
       of SyncQueueKind.Forward:
-        if safeSlot > sq.outSlot:
-          safeSlot - sq.outSlot
-        else:
-          0
+        sq.inpSlot.numSlotsBehindSafeSlot
       of SyncQueueKind.Backward:
-        if sq.outSlot > safeSlot:
-          sq.outSlot - safeSlot
+        if sq.inpSlot == 0xFFFF_FFFF_FFFF_FFFF'u64:
+          0'u64
         else:
-          0
-  if numSlotsAdvanced != 0:
+          sq.inpSlot.numSlotsBehindSafeSlot
+  if numOutSlotsAdvanced != 0 or numInpSlotsAdvanced != 0:
     debug "Sync progress advanced out-of-band",
-      slot_before = sq.outSlot, slot_after = safeSlot
-    sq.advanceOutput(numSlotsAdvanced)
+      safeSlot, outSlot = sq.outSlot, inpSlot = sq.inpSlot
+    if numOutSlotsAdvanced != 0:
+      sq.advanceOutput(numOutSlotsAdvanced)
+    if numInpSlotsAdvanced != 0:
+      sq.advanceInput(numInpSlotsAdvanced)
     sq.wakeupWaiters()
 
 func updateRequestForNewSafeSlot[T](sq: SyncQueue[T], sr: var SyncRequest[T]) =

tests/test_sync_manager.nim

@@ -618,6 +618,20 @@ suite "SyncManager test suite":
         await queue.push(request4, response4)
         check debtLen(queue) == request4.count
 
+        # Advance `safeSlot` out of band.
+        advanceSafeSlot()
+
+        # Fetch a request. It should take into account the new `safeSlot`.
+        let request5 = queue.pop(finish, p1)
+        if request5.isEmpty():
+          break
+        case kkind
+        of SyncQueueKind.Forward:
+          check request5.slot >= getFowardSafeSlotCb()
+        else:
+          check request5.lastSlot <= getBackwardSafeSlotCb()
+        queue.push(request5)
+
       await validatorFut.cancelAndWait()
 
     waitFor runTest()
@@ -664,7 +678,7 @@ suite "SyncManager test suite":
 
     test prefix & "Handle out-of-band sync progress advancement":
       const OutOfBandAdvancementTests = [
-        (Slot(0), Slot(200), SLOTS_PER_EPOCH.uint64)
+        (Slot(0), Slot(500), SLOTS_PER_EPOCH.uint64)
       ]
       for item in OutOfBandAdvancementTests:
         outOfBandAdvancementTest(k, item[0], item[1], item[2])