nimbus-eth2/beacon_chain/sync_manager.nim

import chronicles
import options, deques, heapqueue, tables, strutils, sequtils, math
import stew/[bitseqs, results], chronos, chronicles
import spec/datatypes, spec/digest, peer_pool, eth2_network
import eth/async_utils

import block_pools/block_pools_types
export datatypes, digest, chronos, chronicles, results, block_pools_types

logScope:
  topics = "syncman"

const
  PeerScoreNoStatus* = -100
    ## Peer did not answer `status` request.
  PeerScoreStaleStatus* = -50
    ## Peer's `status` answer do not progress in time.
  PeerScoreGoodStatus* = 50
    ## Peer's `status` answer is fine.
  PeerScoreNoBlocks* = -100
    ## Peer did not respond in time on `blocksByRange` request.
  PeerScoreGoodBlocks* = 100
    ## Peer' `blocksByRange` answer is fine.
  PeerScoreBadBlocks* = -1000
    ## Peer response contains incorrect blocks
  PeerScoreJokeBlocks* = -200
    ## Peer response contains too many empty blocks

type
  SyncFailureKind* = enum
    StatusInvalid,
    StatusDownload,
    StatusStale,
    EmptyProblem,
    BlockDownload

  GetSlotCallback* = proc(): Slot {.gcsafe, raises: [Defect].}

  UpdateLocalBlocksCallback* =
    proc(list: openarray[SignedBeaconBlock]): Result[void, BlockError] {.
      gcsafe.}

  SyncUpdateCallback*[T] =
    proc(req: SyncRequest[T],
         list: openarray[SignedBeaconBlock]): Result[void, BlockError] {.
      gcsafe.}

  SyncRequest*[T] = object
    index*: uint64
    slot*: Slot
    count*: uint64
    step*: uint64
    item*: T

  SyncResult*[T] = object
    request*: SyncRequest[T]
    data*: seq[SignedBeaconBlock]

  SyncQueue*[T] = ref object
    inpSlot*: Slot
    outSlot*: Slot

    startSlot*: Slot
    lastSlot: Slot
    chunkSize*: uint64
    queueSize*: int

    counter*: uint64
    pending*: Table[uint64, Slot]

    waiters: seq[Future[bool]]
    syncUpdate*: SyncUpdateCallback[T]

    debtsQueue: HeapQueue[SyncRequest[T]]
    debtsCount: uint64
    readyQueue: HeapQueue[SyncResult[T]]

    zeroPoint: Option[Slot]
    suspects: seq[SyncResult[T]]

  SyncManager*[A, B] = ref object
    pool: PeerPool[A, B]
    responseTimeout: chronos.Duration
    sleepTime: chronos.Duration
    maxStatusAge: uint64
    maxHeadAge: uint64
    maxRecurringFailures: int
    toleranceValue: uint64
    getLocalHeadSlot: GetSlotCallback
    getLocalWallSlot: GetSlotCallback
    syncUpdate: SyncUpdateCallback[A]
    chunkSize: uint64
    queue: SyncQueue[A]
    failures: seq[SyncFailure[A]]

  SyncMoment* = object
    stamp*: chronos.Moment
    slot*: Slot

  SyncFailure*[T] = object
    kind*: SyncFailureKind
    peer*: T
    stamp*: chronos.Moment

  SyncManagerError* = object of CatchableError
  BeaconBlocksRes* = NetRes[seq[SignedBeaconBlock]]

proc getShortMap*[T](req: SyncRequest[T],
                     data: openarray[SignedBeaconBlock]): string =
  ## Returns all slot numbers in ``data`` as placement map.
  var res = newStringOfCap(req.count)
  var slider = req.slot
  var last = 0
  for i in 0 ..< req.count:
    if last < len(data):
      for k in last ..< len(data):
        if slider == data[k].message.slot:
          res.add('x')
          last = k + 1
          break
        elif slider < data[k].message.slot:
          res.add('.')
          break
    else:
      res.add('.')
    slider = slider + req.step
  result = res

proc getFullMap*[T](req: SyncRequest[T],
                    data: openarray[SignedBeaconBlock]): string =
  # Returns all slot numbers in ``data`` as comma-delimeted string.
  result = mapIt(data, $it.message.slot).join(", ")

proc init*[T](t1: typedesc[SyncRequest], t2: typedesc[T], slot: Slot,
              count: uint64): SyncRequest[T] {.inline.} =
  result = SyncRequest[T](slot: slot, count: count, step: 1'u64)

proc init*[T](t1: typedesc[SyncRequest], t2: typedesc[T], start: Slot,
              finish: Slot): SyncRequest[T] {.inline.} =
  let count = finish - start + 1'u64
  result = SyncRequest[T](slot: start, count: count, step: 1'u64)

proc init*[T](t1: typedesc[SyncRequest], t2: typedesc[T], slot: Slot,
              count: uint64, item: T): SyncRequest[T] {.inline.} =
  result = SyncRequest[T](slot: slot, count: count, item: item, step: 1'u64)

proc init*[T](t1: typedesc[SyncRequest], t2: typedesc[T], start: Slot,
              finish: Slot, item: T): SyncRequest[T] {.inline.} =
  let count = finish - start + 1'u64
  result = SyncRequest[T](slot: start, count: count, step: 1'u64, item: item)

proc init*[T](t1: typedesc[SyncFailure], kind: SyncFailureKind,
              peer: T): SyncFailure[T] {.inline.} =
  result = SyncFailure[T](kind: kind, peer: peer, stamp: now(chronos.Moment))

proc empty*[T](t: typedesc[SyncRequest],
               t2: typedesc[T]): SyncRequest[T] {.inline.} =
  result = SyncRequest[T](step: 0'u64, count: 0'u64)

proc setItem*[T](sr: var SyncRequest[T], item: T) =
  sr.item = item

proc isEmpty*[T](sr: SyncRequest[T]): bool {.inline.} =
  result = (sr.step == 0'u64) and (sr.count == 0'u64)

proc init*[T](t1: typedesc[SyncQueue], t2: typedesc[T],
              start, last: Slot, chunkSize: uint64,
              updateCb: SyncUpdateCallback[T],
              queueSize: int = -1): SyncQueue[T] =
  ## Create new synchronization queue with parameters
  ##
  ## ``start`` and ``last`` are starting and finishing Slots.
  ##
  ## ``chunkSize`` maximum number of slots in one request.
  ##
  ## ``queueSize`` maximum queue size for incoming data. If ``queueSize > 0``
  ## queue will help to keep backpressure under control. If ``queueSize <= 0``
  ## then queue size is unlimited (default).
  ##
  ## ``updateCb`` procedure which will be used to send downloaded blocks to
  ## consumer. Procedure should return ``false`` only when it receives
  ## incorrect blocks, and ``true`` if sequence of blocks is correct.

  # SyncQueue is the core of sync manager, this data structure distributes
  # requests to peers and manages responses from peers.
  #
  # Because SyncQueue is async data structure it manages backpressure and
  # order of incoming responses and it also resolves "joker's" problem.
  #
  # Joker's problem
  #
  # According to current Ethereum2 network specification
  # > Clients MUST respond with at least one block, if they have it and it
  # > exists in the range. Clients MAY limit the number of blocks in the
  # > response.
  #
  # Such rule can lead to very uncertain responses, for example let slots from
  # 10 to 12 will be not empty. Client which follows specification can answer
  # with any response from this list (X - block, `-` empty space):
  #
  # 1.   X X X
  # 2.   - - X
  # 3.   - X -
  # 4.   - X X
  # 5.   X - -
  # 6.   X - X
  # 7.   X X -
  #
  # If peer answers with `1` everything will be fine and `block_pool` will be
  # able to process all 3 blocks. In case of `2`, `3`, `4`, `6` - `block_pool`
  # will fail immediately with chunk and report "parent is missing" error.
  # But in case of `5` and `7` blocks will be processed by `block_pool` without
  # any problems, however it will start producing problems right from this
  # uncertain last slot. SyncQueue will start producing requests for next
  # blocks, but all the responses from this point will fail with "parent is
  # missing" error. Lets call such peers "jokers", because they are joking
  # with responses.
  #
  # To fix "joker" problem i'm going to introduce "zero-point" which will
  # represent first non-empty slot in gap at the end of requested chunk.
  # If SyncQueue receives chunk of blocks with gap at the end and this chunk
  # will be successfully processed by `block_pool` it will set `zero_point` to
  # the first uncertain (empty) slot. For example:
  #
  # Case 1
  #   X  X  X  X  X  -
  #   3  4  5  6  7  8
  #
  # Case2
  #   X  X  -  -  -  -
  #   3  4  5  6  7  8
  #
  # In Case 1 `zero-point` will be equal to 8, in Case 2 `zero-point` will be
  # set to 5.
  #
  # When `zero-point` is set and the next received chunk of blocks will be
  # empty, then peer produced this chunk of blocks will be added to suspect
  # list.
  #
  # If the next chunk of blocks has at least one non-empty block and this chunk
  # will be successfully processed by `block_pool`, then `zero-point` will be
  # reset and suspect list will be cleared.
  #
  # If the `block_pool` failed to process next chunk of blocks, SyncQueue will
  # perform rollback to `zero-point` and penalize all the peers in suspect list.

  doAssert(chunkSize > 0'u64, "Chunk size should not be zero")
  result = SyncQueue[T](
    startSlot: start,
    lastSlot: last,
    chunkSize: chunkSize,
    queueSize: queueSize,
    syncUpdate: updateCb,
    waiters: newSeq[Future[bool]](),
    counter: 1'u64,
    pending: initTable[uint64, Slot](),
    debtsQueue: initHeapQueue[SyncRequest[T]](),
    inpSlot: start,
    outSlot: start
  )

proc `<`*[T](a, b: SyncRequest[T]): bool {.inline.} =
  result = (a.slot < b.slot)

proc `<`*[T](a, b: SyncResult[T]): bool {.inline.} =
  result = (a.request.slot < b.request.slot)

proc `==`*[T](a, b: SyncRequest[T]): bool {.inline.} =
  result = ((a.slot == b.slot) and (a.count == b.count) and
            (a.step == b.step))

proc lastSlot*[T](req: SyncRequest[T]): Slot {.inline.} =
  ## Returns last slot for request ``req``.
  result = req.slot + req.count - 1'u64

proc makePending*[T](sq: SyncQueue[T], req: var SyncRequest[T]) =
  req.index = sq.counter
  sq.counter = sq.counter + 1'u64
  sq.pending[req.index] = req.slot

proc updateLastSlot*[T](sq: SyncQueue[T], last: Slot) {.inline.} =
  ## Update last slot stored in queue ``sq`` with value ``last``.
  doAssert(sq.lastSlot <= last,
           "Last slot could not be lower then stored one " &
           $sq.lastSlot & " <= " & $last)
  sq.lastSlot = last

proc wakeupWaiters[T](sq: SyncQueue[T], flag = true) {.inline.} =
  ## Wakeup one or all blocked waiters.
  for waiter in sq.waiters:
    if not(waiter.finished()):
      waiter.complete(flag)

proc waitForChanges[T](sq: SyncQueue[T]): Future[bool] {.async.} =
  ## Create new waiter and wait for completion from `wakeupWaiters()`.
  var waiter = newFuture[bool]("SyncQueue.waitForChanges")
  sq.waiters.add(waiter)
  try:
    result = await waiter
  finally:
    sq.waiters.delete(sq.waiters.find(waiter))

proc wakeupAndWaitWaiters[T](sq: SyncQueue[T]) {.async.} =
  ## This procedure will perform wakeupWaiters(false) and blocks until last
  ## waiter will be awakened.
  var waitChanges = sq.waitForChanges()
  sq.wakeupWaiters(false)
  discard await waitChanges

proc resetWait*[T](sq: SyncQueue[T], toSlot: Option[Slot]) {.async.} =
  ## Perform reset of all the blocked waiters in SyncQueue.
  ##
  ## We adding one more waiter to the waiters sequence and
  ## call wakeupWaiters(false). Because our waiter is last in sequence of
  ## waiters it will be resumed only after all waiters will be awakened and
  ## finished.

  # We are clearing pending list, so that all requests that are still running
  # around (still downloading, but not yet pushed to the SyncQueue) will be
  # expired. Its important to perform this call first (before await), otherwise
  # you can introduce race problem.
  sq.pending.clear()

  # We calculating minimal slot number to which we will be able to reset,
  # without missing any blocks. There 3 sources:
  # 1. Debts queue.
  # 2. Processing queue (`inpSlot`, `outSlot`).
  # 3. Requested slot `toSlot` (which can be `zero-point` slot).
  #
  # Queue's `outSlot` is the lowest slot we added to `block_pool`, but
  # `zero-point` slot can be less then `outSlot`. `debtsQueue` holds only not
  # added slot requests, so it can't be bigger then `outSlot` value.
  var minSlot = sq.outSlot
  if toSlot.isSome():
    minSlot = min(toSlot.get(), sq.outSlot)
  sq.debtsQueue.clear()
  sq.debtsCount = 0
  sq.readyQueue.clear()
  sq.inpSlot = minSlot
  sq.outSlot = minSlot

  # We are going to wakeup all the waiters and wait for last one.
  await sq.wakeupAndWaitWaiters()

proc isEmpty*[T](sr: SyncResult[T]): bool {.inline.} =
  ## Returns ``true`` if response chain of blocks is empty (has only empty
  ## slots).
  len(sr.data) == 0

proc hasEndGap*[T](sr: SyncResult[T]): bool {.inline.} =
  ## Returns ``true`` if response chain of blocks has gap at the end.
  let lastslot = sr.request.slot + sr.request.count - 1'u64
  if len(sr.data) == 0:
    return true
  if sr.data[^1].message.slot != lastslot:
    return true
  return false

proc getLastNonEmptySlot*[T](sr: SyncResult[T]): Slot {.inline.} =
  ## Returns last non-empty slot from result ``sr``. If response has only
  ## empty slots, original request slot will be returned.
  if len(sr.data) == 0:
    # If response has only empty slots we going to use original request slot
    sr.request.slot
  else:
    sr.data[^1].message.slot

proc toDebtsQueue[T](sq: SyncQueue[T], sr: SyncRequest[T]) {.inline.} =
  sq.debtsQueue.push(sr)
  sq.debtsCount = sq.debtsCount + sr.count

proc push*[T](sq: SyncQueue[T], sr: SyncRequest[T],
              data: seq[SignedBeaconBlock]) {.async, gcsafe.} =
  ## Push successfull result to queue ``sq``.
  mixin updateScore

  if sr.index notin sq.pending:
    # If request `sr` not in our pending list, it only means that
    # SyncQueue.resetWait() happens and all pending requests are expired, so
    # we swallow `old` requests, and in such way sync-workers are able to get
    # proper new requests from SyncQueue.
    return

  sq.pending.del(sr.index)

  # This is backpressure handling algorithm, this algorithm is blocking
  # all pending `push` requests if `request.slot` not in range:
  # [current_queue_slot, current_queue_slot + sq.queueSize * sq.chunkSize].
  var exitNow = false
  while true:
    if (sq.queueSize > 0) and
       (sr.slot >= sq.outSlot + uint64(sq.queueSize) * sq.chunkSize):
      let res = await sq.waitForChanges()
      if res:
        continue
      else:
        # SyncQueue reset happens (it can't be `zero-point` reset, or continous
        # failure reset). We are exiting to wake up sync-worker.
        exitNow = true
        break
    let syncres = SyncResult[T](request: sr, data: data)
    sq.readyQueue.push(syncres)
    exitNow = false
    break

  if exitNow:
    return

  while len(sq.readyQueue) > 0:
    let minSlot = sq.readyQueue[0].request.slot
    if sq.outSlot != minSlot:
      break
    let item = sq.readyQueue.pop()
    let res = sq.syncUpdate(item.request, item.data)
    if res.isOk:
      if sq.zeroPoint.isSome():
        if item.isEmpty():
          # If the `zeropoint` is set and response is empty, we will add this
          # request to suspect list.
          debug "Adding peer to suspect list", peer = item.request.item,
                request_slot = item.request.slot,
                request_count = item.request.count,
                request_step = item.request.step,
                response_count = len(item.data), topics = "syncman"
          sq.suspects.add(item)
        else:
          # If the `zeropoint` is set and response is not empty, we will clean
          # suspect list and reset `zeropoint`.
          sq.suspects.setLen(0)
          sq.zeroPoint = none[Slot]()
          # At this point `zeropoint` is unset, but received response can have
          # gap at the end.
          if item.hasEndGap():
            debug "Zero-point reset and new zero-point found",
                  peer = item.request.item, request_slot = item.request.slot,
                  request_count = item.request.count,
                  request_step = item.request.step,
                  response_count = len(item.data),
                  blocks_map = getShortMap(item.request, item.data),
                  topics = "syncman"
            sq.suspects.add(item)
            sq.zeroPoint = some(item.getLastNonEmptySlot())
          else:
            debug "Zero-point reset", peer = item.request.item,
                  request_slot = item.request.slot,
                  request_count = item.request.count,
                  request_step = item.request.step,
                  response_count = len(item.data),
                  blocks_map = getShortMap(item.request, item.data),
                  topics = "syncman"
      else:
        # If the `zeropoint` is not set and response has gap at the end, we
        # will add first suspect to the suspect list and set `zeropoint`.
        if item.hasEndGap():
          debug "New zero-point found", peer = item.request.item,
                request_slot = item.request.slot,
                request_count = item.request.count,
                request_step = item.request.step,
                response_count = len(item.data),
                blocks_map = getShortMap(item.request, item.data),
                topics = "syncman"
          sq.suspects.add(item)
          sq.zeroPoint = some(item.getLastNonEmptySlot())

      sq.outSlot = sq.outSlot + item.request.count
      sq.wakeupWaiters()
    else:
      debug "Block pool rejected peer's response", peer = item.request.item,
            request_slot = item.request.slot,
            request_count = item.request.count,
            request_step = item.request.step,
            blocks_map = getShortMap(item.request, item.data),
            blocks_count = len(item.data), errCode = res.error

      var resetSlot: Option[Slot]

      if res.error == BlockError.MissingParent:
        if sq.zeroPoint.isSome():
          # If the `zeropoint` is set and we are unable to store response in
          # `block_pool` we are going to revert suspicious responses list.

          # If `zeropoint` is set, suspicious list should not be empty.
          var req: SyncRequest[T]
          if isEmpty(sq.suspects[0]):
            # If initial suspicious response is an empty list, then previous
            # chunk of blocks did not have a gap at the end. So we are going to
            # request suspicious response one more time without any changes.
            req = sq.suspects[0].request
          else:
            # If initial suspicious response is not an empty list, we are going
            # to request only gap at the end of the suspicious response.
            let startSlot = sq.suspects[0].getLastNonEmptySlot() + 1'u64
            let lastSlot = sq.suspects[0].request.lastSlot()
            req = SyncRequest.init(T, startSlot, lastSlot)

          debug "Resolve joker's problem", request_slot = req.slot,
                                           request_count = req.count,
                                           request_step = req.step,
                                         suspects_count = (len(sq.suspects) - 1)

          sq.suspects[0].request.item.updateScore(PeerScoreJokeBlocks)

          sq.toDebtsQueue(req)
          # We move all left suspicious responses to the debts queue.
          if len(sq.suspects) > 1:
            for i in 1 ..< len(sq.suspects):
              sq.toDebtsQueue(sq.suspects[i].request)
              sq.suspects[i].request.item.updateScore(PeerScoreJokeBlocks)

          # Reset state to the `zeropoint`.
          sq.suspects.setLen(0)
          resetSlot = sq.zeroPoint
          sq.zeroPoint = none[Slot]()
        else:
          # If we got `BlockError.MissingParent` and `zero-point` is not set
          # it means that peer returns chain of blocks with holes.
          let req = item.request
          warn "Received sequence of blocks with holes", peer = req.item,
               request_slot = req.slot, request_count = req.count,
               request_step = req.step, blocks_count = len(item.data),
               blocks_map = getShortMap(req, item.data)
          req.item.updateScore(PeerScoreBadBlocks)
      elif res.error == BlockError.Invalid:
        let req = item.request
        warn "Received invalid sequence of blocks", peer = req.item,
              request_slot = req.slot, request_count = req.count,
              request_step = req.step, blocks_count = len(item.data),
              blocks_map = getShortMap(req, item.data)
        req.item.updateScore(PeerScoreBadBlocks)
      else:
        let req = item.request
        warn "Received unexpected response from block_pool", peer = req.item,
             request_slot = req.slot, request_count = req.count,
             request_step = req.step, blocks_count = len(item.data),
             blocks_map = getShortMap(req, item.data), errorCode = res.error
        req.item.updateScore(PeerScoreBadBlocks)

      # We need to move failed response to the debts queue.
      sq.toDebtsQueue(item.request)
      if resetSlot.isSome():
        await sq.resetWait(resetSlot)
        debug "Zero-point reset happens", queue_input_slot = sq.inpSlot,
                                          queue_output_slot = sq.outSlot
      break

proc push*[T](sq: SyncQueue[T], sr: SyncRequest[T]) =
  ## Push failed request back to queue.
  if sr.index notin sq.pending:
    # If request `sr` not in our pending list, it only means that
    # SyncQueue.resetWait() happens and all pending requests are expired, so
    # we swallow `old` requests, and in such way sync-workers are able to get
    # proper new requests from SyncQueue.
    return
  sq.pending.del(sr.index)
  sq.toDebtsQueue(sr)

proc pop*[T](sq: SyncQueue[T], maxslot: Slot, item: T): SyncRequest[T] =
  if len(sq.debtsQueue) > 0:
    if maxSlot < sq.debtsQueue[0].slot:
      return SyncRequest.empty(T)

    var sr = sq.debtsQueue.pop()
    if sr.lastSlot() <= maxSlot:
      sq.debtsCount = sq.debtsCount - sr.count
      sr.setItem(item)
      sq.makePending(sr)
      return sr

    var sr1 = SyncRequest.init(T, sr.slot, maxslot, item)
    let sr2 = SyncRequest.init(T, maxslot + 1'u64, sr.lastSlot())
    sq.debtsQueue.push(sr2)
    sq.debtsCount = sq.debtsCount - sr1.count
    sq.makePending(sr1)
    return sr1
  else:
    if maxSlot < sq.inpSlot:
      return SyncRequest.empty(T)

    if sq.inpSlot > sq.lastSlot:
      return SyncRequest.empty(T)

    let lastSlot = min(maxslot, sq.lastSlot)
    let count = min(sq.chunkSize, lastSlot + 1'u64 - sq.inpSlot)
    var sr = SyncRequest.init(T, sq.inpSlot, count, item)
    sq.inpSlot = sq.inpSlot + count
    sq.makePending(sr)
    return sr

proc len*[T](sq: SyncQueue[T]): uint64 {.inline.} =
  ## Returns number of slots left in queue ``sq``.
  if sq.inpSlot > sq.lastSlot:
    result = sq.debtsCount
  else:
    result = sq.lastSlot - sq.inpSlot + 1'u64 - sq.debtsCount

proc total*[T](sq: SyncQueue[T]): uint64 {.inline.} =
  ## Returns total number of slots in queue ``sq``.
  result = sq.lastSlot - sq.startSlot + 1'u64

proc progress*[T](sq: SyncQueue[T]): uint64 =
  ## Returns queue's ``sq`` progress string.
  let curSlot = sq.outSlot - sq.startSlot
  result = (curSlot * 100'u64) div sq.total()

proc now*(sm: typedesc[SyncMoment], slot: Slot): SyncMoment {.inline.} =
  result = SyncMoment(stamp: now(chronos.Moment), slot: slot)

proc speed*(start, finish: SyncMoment): float {.inline.} =
  ## Returns number of slots per second.
  let slots = finish.slot - start.slot
  let dur = finish.stamp - start.stamp
  let secs = float(chronos.seconds(1).nanoseconds)
  if isZero(dur):
    result = 0.0
  else:
    let v = float(slots) * (secs / float(dur.nanoseconds))
    # We doing round manually because stdlib.round is deprecated
    result = round(v * 10000) / 10000

proc newSyncManager*[A, B](pool: PeerPool[A, B],
                           getLocalHeadSlotCb: GetSlotCallback,
                           getLocalWallSlotCb: GetSlotCallback,
                           updateLocalBlocksCb: UpdateLocalBlocksCallback,
                           maxStatusAge = uint64(SLOTS_PER_EPOCH * 4),
                           maxHeadAge = uint64(SLOTS_PER_EPOCH * 4),
                           sleepTime = (int(SLOTS_PER_EPOCH) *
                                        int(SECONDS_PER_SLOT)).seconds,
                           chunkSize = uint64(SLOTS_PER_EPOCH),
                           toleranceValue = uint64(1),
                           maxRecurringFailures = 3
                           ): SyncManager[A, B] =

  proc syncUpdate(req: SyncRequest[A],
      list: openarray[SignedBeaconBlock]): Result[void, BlockError] {.gcsafe.} =
    let peer = req.item
    let res = updateLocalBlocksCb(list)
    if res.isOk:
      peer.updateScore(PeerScoreGoodBlocks)
    return res

  let queue = SyncQueue.init(A, getLocalHeadSlotCb(), getLocalWallSlotCb(),
                             chunkSize, syncUpdate, 2)

  result = SyncManager[A, B](
    pool: pool,
    maxStatusAge: maxStatusAge,
    getLocalHeadSlot: getLocalHeadSlotCb,
    syncUpdate: syncUpdate,
    getLocalWallSlot: getLocalWallSlotCb,
    maxHeadAge: maxHeadAge,
    maxRecurringFailures: maxRecurringFailures,
    sleepTime: sleepTime,
    chunkSize: chunkSize,
    queue: queue
  )

proc getBlocks*[A, B](man: SyncManager[A, B], peer: A,
                      req: SyncRequest): Future[BeaconBlocksRes] {.async.} =
  mixin beaconBlocksByRange, getScore, `==`
  doAssert(not(req.isEmpty()), "Request must not be empty!")
  debug "Requesting blocks from peer", peer = peer,
        slot = req.slot, slot_count = req.count, step = req.step,
        peer_score = peer.getScore(), topics = "syncman"
  var workFut = awaitne beaconBlocksByRange(peer, req.slot, req.count, req.step)
  if workFut.failed():
    debug "Error, while waiting getBlocks response", peer = peer,
          slot = req.slot, slot_count = req.count, step = req.step,
          errMsg = workFut.readError().msg, topics = "syncman"
  else:
    let res = workFut.read()
    if res.isErr:
      debug "Error, while reading getBlocks response",
            peer = peer, slot = req.slot, count = req.count,
            step = req.step, topics = "syncman"
    result = res

template headAge(): uint64 =
  wallSlot - headSlot

template peerAge(): uint64 =
  if peerSlot > wallSlot: 0'u64 else: wallSlot - peerSlot

proc syncWorker*[A, B](man: SyncManager[A, B],
                       peer: A): Future[A] {.async.} =
  # Sync worker is the lowest level loop which performs syncing with single
  # peer.
  #
  # Logic here is pretty simple:
  # 1. Obtain request from SyncQueue.
  # 2. Send this request to a peer and obtain response.
  # 3. Push response to the SyncQueue, (doesn't matter if it success or failure)
  # 4. Update main SyncQueue last slot with wall time slot number.
  # 5. From time to time we also requesting peer's status information.
  # 6. If our current head slot is near equal to peer's head slot we are
  #    exiting this loop and finishing that sync-worker task.
  # 7. Repeat

  mixin getKey, getScore, getHeadSlot

  debug "Starting syncing with peer", peer = peer,
                                      peer_score = peer.getScore(),
                                      topics = "syncman"
  try:
    while true:
      var wallSlot = man.getLocalWallSlot()
      var headSlot = man.getLocalHeadSlot()
      var peerSlot = peer.getHeadSlot()

      man.queue.updateLastSlot(wallSlot)

      debug "Peer's syncing status", wall_clock_slot = wallSlot,
            remote_head_slot = peerSlot, local_head_slot = headSlot,
            peer_score = peer.getScore(), peer = peer, topics = "syncman"

      if peerSlot > wallSlot + man.toleranceValue:
        # Our wall timer is broken, or peer's status information is invalid.
        debug "Local timer is broken or peer's status information is invalid",
              wall_clock_slot = wallSlot, remote_head_slot = peerSlot,
              local_head_slot = headSlot, peer = peer,
              tolerance_value = man.toleranceValue,
              peer_score = peer.getScore(), topics = "syncman"
        # let failure = SyncFailure.init(SyncFailureKind.StatusInvalid, peer)
        # man.failures.add(failure)
        break

      if peerAge >= man.maxStatusAge:
        # Peer's status information is very old, we going to update it.
        debug "Updating peer's status information", wall_clock_slot = wallSlot,
              remote_head_slot = peerSlot, local_head_slot = headSlot,
              peer = peer, peer_score = peer.getScore(), topics = "syncman"
        let res = await peer.updateStatus()
        if not(res):
          peer.updateScore(PeerScoreNoStatus)
          debug "Failed to get remote peer's status, exiting", peer = peer,
                peer_score = peer.getScore(), peer_head_slot = peerSlot,
                topics = "syncman"
          # let failure = SyncFailure.init(SyncFailureKind.StatusDownload, peer)
          # man.failures.add(failure)
          break

        let newPeerSlot = peer.getHeadSlot()
        if peerSlot >= newPeerSlot:
          peer.updateScore(PeerScoreStaleStatus)
          debug "Peer's status information is stale, exiting",
                wall_clock_slot = wallSlot, remote_old_head_slot = peerSlot,
                local_head_slot = headSlot,
                remote_new_head_slot = newPeerSlot,
                peer = peer, peer_score = peer.getScore(), topics = "syncman"
          # let failure = SyncFailure.init(SyncFailureKind.StatusStale, peer)
          # man.failures.add(failure)
          break

        debug "Peer's status information updated", wall_clock_slot = wallSlot,
              remote_old_head_slot = peerSlot, local_head_slot = headSlot,
              remote_new_head_slot = newPeerSlot, peer = peer,
              peer_score = peer.getScore(), topics = "syncman"
        peer.updateScore(PeerScoreGoodStatus)
        peerSlot = newPeerSlot

      if (peerAge <= man.maxHeadAge) and (headAge <= man.maxHeadAge):
        debug "We are in sync with peer, exiting", wall_clock_slot = wallSlot,
              remote_head_slot = peerSlot, local_head_slot = headSlot,
              peer = peer, peer_score = peer.getScore(), topics = "syncman"
        break

      let req = man.queue.pop(peerSlot, peer)
      if req.isEmpty():
        debug "Empty request received from queue, exiting", peer = peer,
              local_head_slot = headSlot, remote_head_slot = peerSlot,
              queue_input_slot = man.queue.inpSlot,
              queue_output_slot = man.queue.outSlot,
              queue_last_slot = man.queue.lastSlot,
              peer_score = peer.getScore(), topics = "syncman"
        # Sometimes when syncing is almost done but last requests are still
        # pending, this can fall into endless cycle, when low number of peers
        # are available in PeerPool. We going to wait for RESP_TIMEOUT time,
        # so all pending requests should be finished at this moment.
        await sleepAsync(RESP_TIMEOUT)
        # let failure = SyncFailure.init(SyncFailureKind.EmptyProblem, peer)
        # man.failures.add(failure)
        break

      debug "Creating new request for peer", wall_clock_slot = wallSlot,
            remote_head_slot = peerSlot, local_head_slot = headSlot,
            request_slot = req.slot, request_count = req.count,
            request_step = req.step, peer = peer,
            peer_score = peer.getScore(), topics = "syncman"

      let blocks = await man.getBlocks(peer, req)
      if blocks.isOk:
        let data = blocks.get()
        let smap = getShortMap(req, data)
        debug "Received blocks on request", blocks_count = len(data),
              blocks_map = smap, request_slot = req.slot,
              request_count = req.count, request_step = req.step,
              peer = peer, peer_score = peer.getScore(), topics = "syncman"
        # Scoring will happen in `syncUpdate`.
        await man.queue.push(req, data)
        # Cleaning up failures.
        man.failures.setLen(0)
      else:
        peer.updateScore(PeerScoreNoBlocks)
        man.queue.push(req)
        debug "Failed to receive blocks on request",
              request_slot = req.slot, request_count = req.count,
              request_step = req.step, peer = peer,
              peer_score = peer.getScore(), topics = "syncman"
        # let failure = SyncFailure.init(SyncFailureKind.BlockDownload, peer)
        # man.failures.add(failure)
        break

    result = peer
  finally:
    man.pool.release(peer)

proc sync*[A, B](man: SyncManager[A, B]) {.async.} =
  # This procedure manages main loop of SyncManager and in this loop it
  # performs
  # 1. It checks for current sync status, "are we synced?".
  # 2. If we are in active syncing, it tries to acquire peers from PeerPool and
  #    spawns new sync-workers.
  # 3. It stops spawning sync-workers when we are "in sync".
  # 4. It calculates syncing performance.
  mixin getKey, getScore
  var pending = newSeq[Future[A]]()
  var acquireFut: Future[A]
  var wallSlot, headSlot: Slot
  var syncSpeed: float = 0.0

  template workersCount(): int =
    if isNil(acquireFut): len(pending) else: (len(pending) - 1)

  proc speedometerTask() {.async.} =
    while true:
      let lsm1 = SyncMoment.now(man.getLocalHeadSlot())
      await sleepAsync(chronos.seconds(int(SECONDS_PER_SLOT)))
      let lsm2 = SyncMoment.now(man.getLocalHeadSlot())
      if workersCount() == 0:
        syncSpeed = 0.0
      else:
        syncSpeed = speed(lsm1, lsm2)

  debug "Synchronization loop started", topics = "syncman"

  traceAsyncErrors speedometerTask()

  while true:
    wallSlot = man.getLocalWallSlot()
    headSlot = man.getLocalHeadSlot()

    var progress: uint64
    if headSlot <= man.queue.lastSlot:
      progress = man.queue.progress()
    else:
      progress = 100'u64

    debug "Synchronization loop start tick", wall_head_slot = wallSlot,
          local_head_slot = headSlot, queue_status = progress,
          queue_start_slot = man.queue.startSlot,
          queue_last_slot = man.queue.lastSlot,
          workers_count = workersCount(), topics = "syncman"

    if headAge <= man.maxHeadAge:
      debug "Synchronization loop sleeping", wall_head_slot = wallSlot,
              local_head_slot = headSlot, workers_count = workersCount(),
              difference = (wallSlot - headSlot),
              max_head_age = man.maxHeadAge, topics = "syncman"
      if len(pending) == 0:
        await sleepAsync(man.sleepTime)
      else:
        var peerFut = one(pending)
        # We do not care about result here because we going to check peerFut
        # later.
        discard await withTimeout(peerFut, man.sleepTime)
    else:
      if isNil(acquireFut):
        acquireFut = man.pool.acquire()
        pending.add(acquireFut)

      debug "Synchronization loop waiting for new peer",
              wall_head_slot = wallSlot, local_head_slot = headSlot,
              workers_count = workersCount(), topics = "syncman"
      var peerFut = one(pending)
      # We do not care about result here, because we going to check peerFut
      # later.
      discard await withTimeout(peerFut, man.sleepTime)

    var temp = newSeqOfCap[Future[A]](len(pending))
    # Update slots to with more recent data
    wallSlot = man.getLocalWallSlot()
    headSlot = man.getLocalHeadSlot()
    for fut in pending:
      if fut.finished():
        if fut == acquireFut:
          # We acquired new peer from PeerPool.
          if acquireFut.failed():
            debug "Synchronization loop failed to get new peer",
                  wall_head_slot = wallSlot, local_head_slot = headSlot,
                  workers_count = workersCount(),
                  errMsg = acquireFut.readError().msg, topics = "syncman"
          else:
            var peer = acquireFut.read()
            if headAge <= man.maxHeadAge:
              # If we are already in sync, we going to release just acquired
              # peer and do not acquire peers
              debug "Synchronization loop reached sync barrier", peer = peer,
                    wall_head_slot = wallSlot, local_head_slot = headSlot,
                    peer_score = peer.getScore(), topics = "syncman"
              man.pool.release(peer)
            else:
              if headSlot > man.queue.lastSlot:
                man.queue = SyncQueue.init(A, headSlot, wallSlot,
                                           man.chunkSize, man.syncUpdate, 2)
              debug "Synchronization loop starting new worker", peer = peer,
                    wall_head_slot = wallSlot, local_head_slot = headSlot,
                    peer_score = peer.getScore(), topics = "syncman"
              temp.add(syncWorker(man, peer))

          acquireFut = nil
          if headAge > man.maxHeadAge:
            acquireFut = man.pool.acquire()
            temp.add(acquireFut)
        else:
          # Worker finished its work
          if fut.failed():
            debug "Synchronization loop got worker finished with an error",
                   wall_head_slot = wallSlot, local_head_slot = headSlot,
                   errMsg = fut.readError().msg, topics = "syncman"
          else:
            let peer = fut.read()
            debug "Synchronization loop got worker finished",
                   wall_head_slot = wallSlot, local_head_slot = headSlot,
                   peer = peer, peer_score = peer.getScore(),
                   topics = "syncman"
      else:
        if fut == acquireFut:
          if headAge <= man.maxHeadAge:
            debug "Synchronization loop reached sync barrier",
                   wall_head_slot = wallSlot, local_head_slot = headSlot,
                   topics = "syncman"
            acquireFut = nil
            fut.cancel()
          else:
            temp.add(fut)
        else:
          temp.add(fut)

    pending = temp

    if len(man.failures) > man.maxRecurringFailures and (workersCount() > 1):
      debug "Number of recurring failures exceeds limit, reseting queue",
            workers_count = $workers_count(), rec_failures = $len(man.failures)
      await man.queue.resetWait(none[Slot]())

    debug "Synchronization loop end tick", wall_head_slot = wallSlot,
          local_head_slot = headSlot, workers_count = workersCount(),
          waiting_for_new_peer = $not(isNil(acquireFut)),
          sync_speed = syncSpeed, topics = "syncman"