nimbus-eth2/beacon_chain/sync/sync_manager.nim

import chronicles
import options, deques, heapqueue, tables, strutils, sequtils, math, algorithm
import stew/results, chronos, chronicles
import ../spec/[datatypes, digest, helpers, eth2_apis/callsigs_types],
       ../networking/[peer_pool, eth2_network]

import ../gossip_processing/gossip_to_consensus
import ../consensus_object_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.
  PeerScoreUseless* = -10
    ## Peer's latest head is lower then ours.
  PeerScoreGoodStatus* = 50
    ## Peer's `status` answer is fine.
  PeerScoreNoBlocks* = -100
    ## Peer did not respond in time on `blocksByRange` request.
  PeerScoreGoodBlocks* = 100
    ## Peer's `blocksByRange` answer is fine.
  PeerScoreBadBlocks* = -1000
    ## Peer's response contains incorrect blocks.
  PeerScoreBadResponse* = -1000
    ## Peer's response is not in requested range.
  PeerScoreMissingBlocks* = -200
    ## Peer response contains too many empty blocks.

  SyncWorkersCount* = 10
    ## Number of sync workers to spawn

  StatusUpdateInterval* = chronos.minutes(1)
    ## Minimum time between two subsequent calls to update peer's status

  StatusExpirationTime* = chronos.minutes(2)
    ## Time time it takes for the peer's status information to expire.

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

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

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

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

  SyncWaiter*[T] = object
    future: Future[bool]
    request: SyncRequest[T]

  RewindPoint = object
    failSlot: Slot
    epochCount: uint64

  SyncQueue*[T] = ref object
    inpSlot*: Slot
    outSlot*: Slot
    startSlot*: Slot
    lastSlot: Slot
    chunkSize*: uint64
    queueSize*: int
    counter*: uint64
    opcounter*: uint64
    pending*: Table[uint64, SyncRequest[T]]
    waiters: seq[SyncWaiter[T]]
    getFinalizedSlot*: GetSlotCallback
    debtsQueue: HeapQueue[SyncRequest[T]]
    debtsCount: uint64
    readyQueue: HeapQueue[SyncResult[T]]
    rewind: Option[RewindPoint]
    verifQueues: ref VerifQueueManager

  SyncWorkerStatus* {.pure.} = enum
    Sleeping, WaitingPeer, UpdatingStatus, Requesting, Downloading, Processing

  SyncWorker*[A, B] = object
    future: Future[void]
    status: SyncWorkerStatus

  SyncManager*[A, B] = ref object
    pool: PeerPool[A, B]
    responseTimeout: chronos.Duration
    sleepTime: chronos.Duration
    maxStatusAge: uint64
    maxHeadAge: uint64
    toleranceValue: uint64
    getLocalHeadSlot: GetSlotCallback
    getLocalWallSlot: GetSlotCallback
    getFinalizedSlot: GetSlotCallback
    workers: array[SyncWorkersCount, SyncWorker[A, B]]
    notInSyncEvent: AsyncEvent
    rangeAge: uint64
    inRangeEvent*: AsyncEvent
    notInRangeEvent*: AsyncEvent
    chunkSize: uint64
    queue: SyncQueue[A]
    syncFut: Future[void]
    verifQueues: ref VerifQueueManager
    inProgress*: bool
    insSyncSpeed*: float
    avgSyncSpeed*: float
    timeLeft*: Duration
    syncCount*: uint64
    syncStatus*: string

  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 validate*[T](sq: SyncQueue[T],
           blk: SignedBeaconBlock): Future[Result[void, BlockError]] {.async.} =
  let sblock = SyncBlock(
    blk: blk,
    resfut: newFuture[Result[void, BlockError]]("sync.manager.validate")
  )
  sq.verifQueues[].addBlock(sblock)
  return await sblock.resfut

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 contains*[T](req: SyncRequest[T], slot: Slot): bool {.inline.} =
  slot >= req.slot and slot < req.slot + req.count * req.step and
    ((slot - req.slot) mod req.step == 0)

proc cmp*[T](a, b: SyncRequest[T]): int =
  result = cmp(uint64(a.slot), uint64(b.slot))

proc checkResponse*[T](req: SyncRequest[T],
                       data: openArray[SignedBeaconBlock]): bool =
  if len(data) == 0:
    # Impossible to verify empty response.
    return true

  if uint64(len(data)) > req.count:
    # Number of blocks in response should be less or equal to number of
    # requested blocks.
    return false

  var slot = req.slot
  var rindex = 0'u64
  var dindex = 0

  while (rindex < req.count) and (dindex < len(data)):
    if slot < data[dindex].message.slot:
      discard
    elif slot == data[dindex].message.slot:
      inc(dindex)
    else:
      return false
    slot = slot + req.step
    rindex = rindex + 1'u64

  if dindex == len(data):
    return true
  else:
    return false

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] =
  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] =
  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] =
  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] =
  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] =
  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,
              getFinalizedSlotCb: GetSlotCallback,
              verifQueues: ref VerifQueueManager,
              syncQueueSize: 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.
  ##
  ## ``syncQueueSize`` maximum queue size for incoming data. If ``syncQueueSize > 0``
  ## queue will help to keep backpressure under control. If ``syncQueueSize <= 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 we going to perform rollback to the latest finalized
  # epoch's first slot.
  doAssert(chunkSize > 0'u64, "Chunk size should not be zero")
  result = SyncQueue[T](
    startSlot: start,
    lastSlot: last,
    chunkSize: chunkSize,
    queueSize: syncQueueSize,
    getFinalizedSlot: getFinalizedSlotCb,
    waiters: newSeq[SyncWaiter[T]](),
    counter: 1'u64,
    pending: initTable[uint64, SyncRequest[T]](),
    debtsQueue: initHeapQueue[SyncRequest[T]](),
    inpSlot: start,
    outSlot: start,
    verifQueues: verifQueues
  )

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

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) =
  ## Wakeup one or all blocked waiters.
  for item in sq.waiters:
    if not(item.future.finished()):
      item.future.complete(flag)

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

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(SyncRequest.empty(T))
  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`.
  #
  # Queue's `outSlot` is the lowest slot we added to `block_pool`, but
  # `toSlot` 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]) =
  sq.debtsQueue.push(sr)
  sq.debtsCount = sq.debtsCount + sr.count

proc getRewindPoint*[T](sq: SyncQueue[T], failSlot: Slot,
                        finalizedSlot: Slot): Slot =
  # Calculate exponential rewind point in number of epochs.
  let epochCount =
    if sq.rewind.isSome():
      let rewind = sq.rewind.get()
      if failSlot == rewind.failSlot:
        # `MissingParent` happened at same slot so we increase rewind point by
        # factor of 2.
        let epochs = rewind.epochCount * 2
        sq.rewind = some(RewindPoint(failSlot: failSlot, epochCount: epochs))
        epochs
      else:
        # `MissingParent` happened at different slot so we going to rewind for
        # 1 epoch only.
        sq.rewind = some(RewindPoint(failSlot: failSlot, epochCount: 1'u64))
        1'u64
    else:
      # `MissingParent` happened first time.
      sq.rewind = some(RewindPoint(failSlot: failSlot, epochCount: 1'u64))
      1'u64

  # Calculate the latest finalized epoch.
  let finalizedEpoch = compute_epoch_at_slot(finalizedSlot)

  # Calculate the rewind epoch, which should not be less than the latest
  # finalized epoch.
  let rewindEpoch =
    block:
      let failEpoch = compute_epoch_at_slot(failSlot)
      if failEpoch < finalizedEpoch + epochCount:
        finalizedEpoch
      else:
        failEpoch - epochCount

  compute_start_slot_at_epoch(rewindEpoch)

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(sr)
      if res:
        continue
      else:
        # SyncQueue reset happens. 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()

    # Validating received blocks one by one
    var res: Result[void, BlockError]
    var failSlot: Option[Slot]
    if len(item.data) > 0:
      for blk in item.data:
        trace "Pushing block", block_root = blk.root,
                               block_slot = blk.message.slot
        res = await sq.validate(blk)
        if not(res.isOk):
          failSlot = some(blk.message.slot)
          break
    else:
      res = Result[void, BlockError].ok()

    # Increase progress counter, so watch task will be able to know that we are
    # not stuck.
    inc(sq.opcounter)

    if res.isOk:
      sq.outSlot = sq.outSlot + item.request.count
      if len(item.data) > 0:
        # If there no error and response was not empty we should reward peer
        # with some bonus score.
        item.request.item.updateScore(PeerScoreGoodBlocks)
      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,
            topics = "syncman"

      var resetSlot: Option[Slot]

      if res.error == BlockError.MissingParent:
        # If we got `BlockError.MissingParent` it means that peer returns chain
        # of blocks with holes or `block_pool` is in incomplete state. We going
        # to rewind to the first slot at latest finalized epoch.
        let req = item.request
        let finalizedSlot = sq.getFinalizedSlot()
        if finalizedSlot < req.slot:
          let rewindSlot = sq.getRewindPoint(failSlot.get(), finalizedSlot)
          warn "Unexpected missing parent, rewind happens",
               peer = req.item, rewind_to_slot = rewindSlot,
               rewind_epoch_count = sq.rewind.get().epochCount,
               rewind_fail_slot = failSlot.get(),
               finalized_slot = finalized_slot,
               request_slot = req.slot, request_count = req.count,
               request_step = req.step, blocks_count = len(item.data),
               blocks_map = getShortMap(req, item.data), topics = "syncman"
          resetSlot = some(rewindSlot)
          req.item.updateScore(PeerScoreMissingBlocks)
        else:
          error "Unexpected missing parent at finalized epoch slot",
                peer = req.item, to_slot = finalizedSlot,
                request_slot = req.slot, request_count = req.count,
                request_step = req.step, blocks_count = len(item.data),
                blocks_map = getShortMap(req, item.data), topics = "syncman"
          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), topics = "syncman"
        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,
             topics = "syncman"
        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 "Rewind to slot was happened", reset_slot = reset_slot.get(),
              queue_input_slot = sq.inpSlot, queue_output_slot = sq.outSlot,
              rewind_epoch_count = sq.rewind.get().epochCount,
              rewind_fail_slot = sq.rewind.get().failSlot,
              reset_slot = resetSlot, topics = "syncman"
      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,
                           getFinalizedSlotCb: GetSlotCallback,
                           verifQueues: ref VerifQueueManager,
                           maxStatusAge = uint64(SLOTS_PER_EPOCH * 4),
                           maxHeadAge = uint64(SLOTS_PER_EPOCH * 1),
                           sleepTime = (int(SLOTS_PER_EPOCH) *
                                        int(SECONDS_PER_SLOT)).seconds,
                           chunkSize = uint64(SLOTS_PER_EPOCH),
                           toleranceValue = uint64(1),
                           rangeAge = uint64(SLOTS_PER_EPOCH * 4)
                           ): SyncManager[A, B] =

  let queue = SyncQueue.init(A, getLocalHeadSlotCb(), getLocalWallSlotCb(),
                             chunkSize, getFinalizedSlotCb, verifQueues, 1)

  result = SyncManager[A, B](
    pool: pool,
    maxStatusAge: maxStatusAge,
    getLocalHeadSlot: getLocalHeadSlotCb,
    getLocalWallSlot: getLocalWallSlotCb,
    getFinalizedSlot: getFinalizedSlotCb,
    maxHeadAge: maxHeadAge,
    sleepTime: sleepTime,
    chunkSize: chunkSize,
    queue: queue,
    verifQueues: verifQueues,
    notInSyncEvent: newAsyncEvent(),
    inRangeEvent: newAsyncEvent(),
    notInRangeEvent: newAsyncEvent(),
    rangeAge: rangeAge
  )

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(), peer_speed = peer.netKbps(),
        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, peer_speed = peer.netKbps(),
          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, peer_speed = peer.netKbps(),
            topics = "syncman", error = $res.error()
    result = res

template headAge(): uint64 =
  wallSlot - headSlot

template queueAge(): uint64 =
  wallSlot - man.queue.outSlot

template peerStatusAge(): Duration =
  Moment.now() - peer.state(BeaconSync).statusLastTime

func syncQueueLen*[A, B](man: SyncManager[A, B]): uint64 =
  man.queue.len

proc syncStep[A, B](man: SyncManager[A, B], index: int, peer: A) {.async.} =
  let wallSlot = man.getLocalWallSlot()
  let headSlot = man.getLocalHeadSlot()
  var peerSlot = peer.getHeadSlot()

  # We updating SyncQueue's last slot all the time
  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, index = index,
        peer_speed = peer.netKbps(), topics = "syncman"

  # Check if peer's head slot is bigger than our wall clock slot.
  if peerSlot > wallSlot + man.toleranceValue:
    warn "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, index = index,
          tolerance_value = man.toleranceValue, peer_speed = peer.netKbps(),
          peer_score = peer.getScore(), topics = "syncman"
    discard SyncFailure.init(SyncFailureKind.StatusInvalid, peer)
    return

  # Check if we need to update peer's status information
  if peerStatusAge >= StatusExpirationTime:
    # Peer's status information is very old, its time to update it
    man.workers[index].status = SyncWorkerStatus.UpdatingStatus
    trace "Updating peer's status information", wall_clock_slot = wallSlot,
          remote_head_slot = peerSlot, local_head_slot = headSlot,
          peer = peer, peer_score = peer.getScore(), index = index,
          peer_speed = peer.netKbps(), topics = "syncman"

    try:
      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,
              peer_speed = peer.netKbps(), index = index, topics = "syncman"
        discard SyncFailure.init(SyncFailureKind.StatusDownload, peer)
        return
    except CatchableError as exc:
      debug "Unexpected exception while updating peer's status",
            peer = peer, peer_score = peer.getScore(),
            peer_head_slot = peerSlot, peer_speed = peer.netKbps(),
            index = index, errMsg = exc.msg, topics = "syncman"
      return

    let newPeerSlot = peer.getHeadSlot()
    if peerSlot >= newPeerSlot:
      peer.updateScore(PeerScoreStaleStatus)
      debug "Peer's status information is stale",
            wall_clock_slot = wallSlot, remote_old_head_slot = peerSlot,
            local_head_slot = headSlot, remote_new_head_slot = newPeerSlot,
            peer = peer, peer_score = peer.getScore(), index = index,
            peer_speed = peer.netKbps(), topics = "syncman"
    else:
      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(), peer_speed = peer.netKbps(),
            index = index, topics = "syncman"
      peer.updateScore(PeerScoreGoodStatus)
      peerSlot = newPeerSlot

  if headAge <= man.maxHeadAge:
    info "We are in sync with network", wall_clock_slot = wallSlot,
          remote_head_slot = peerSlot, local_head_slot = headSlot,
          peer = peer, peer_score = peer.getScore(), index = index,
          peer_speed = peer.netKbps(), topics = "syncman"
    # We clear SyncManager's `notInSyncEvent` so all the workers will become
    # sleeping soon.
    man.notInSyncEvent.clear()
    return

  if headSlot >= peerSlot - man.maxHeadAge:
    debug "We are in sync with peer; refreshing peer's status information",
          wall_clock_slot = wallSlot, remote_head_slot = peerSlot,
          local_head_slot = headSlot, peer = peer, peer_score = peer.getScore(),
          index = index, peer_speed = peer.netKbps(), topics = "syncman"

    man.workers[index].status = SyncWorkerStatus.UpdatingStatus

    if peerStatusAge <= StatusUpdateInterval:
      await sleepAsync(StatusUpdateInterval - peerStatusAge)

    try:
      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,
              peer_speed = peer.netKbps(), index = index, topics = "syncman"
        discard SyncFailure.init(SyncFailureKind.StatusDownload, peer)
        return
    except CatchableError as exc:
      debug "Unexpected exception while updating peer's status",
            peer = peer, peer_score = peer.getScore(),
            peer_head_slot = peerSlot, peer_speed = peer.netKbps(),
            index = index, errMsg = exc.msg, topics = "syncman"
      return

    let newPeerSlot = peer.getHeadSlot()
    if peerSlot >= newPeerSlot:
      peer.updateScore(PeerScoreStaleStatus)
      debug "Peer's status information is stale",
            wall_clock_slot = wallSlot, remote_old_head_slot = peerSlot,
            local_head_slot = headSlot, remote_new_head_slot = newPeerSlot,
            peer = peer, peer_score = peer.getScore(), index = index,
            peer_speed = peer.netKbps(), topics = "syncman"
    else:
      # This is not very good solution because we should not discriminate and/or
      # penalize peers which are in sync process too, but their latest head is
      # lower then our latest head. We should keep connections with such peers
      # (so this peers are able to get in sync using our data), but we should
      # not use this peers for syncing because this peers are useless for us.
      # Right now we decreasing peer's score a bit, so it will not be
      # disconnected due to low peer's score, but new fresh peers could replace
      # peers with low latest head.
      if headSlot >= newPeerSlot - man.maxHeadAge:
        # Peer's head slot is still lower then ours.
        debug "Peer's head slot is lower then local head slot",
              wall_clock_slot = wallSlot, remote_old_head_slot = peerSlot,
              local_head_slot = headSlot, remote_new_head_slot = newPeerSlot,
              peer = peer, peer_score = peer.getScore(),
              peer_speed = peer.netKbps(), index = index, topics = "syncman"
        peer.updateScore(PeerScoreUseless)
      else:
        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(), peer_speed = peer.netKbps(),
              index = index, topics = "syncman"
        peer.updateScore(PeerScoreGoodStatus)
        peerSlot = newPeerSlot

    return

  man.workers[index].status = SyncWorkerStatus.Requesting
  let req = man.queue.pop(peerSlot, peer)
  if req.isEmpty():
    # SyncQueue could return empty request in 2 cases:
    # 1. There no more slots in SyncQueue to download (we are synced, but
    #    our ``notInSyncEvent`` is not yet cleared).
    # 2. Current peer's known head slot is too low to satisfy request.
    #
    # To avoid endless loop we going to wait for RESP_TIMEOUT time here.
    # This time is enough for all pending requests to finish and it is also
    # enough for main sync loop to clear ``notInSyncEvent``.
    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_speed = peer.netKbps(), peer_score = peer.getScore(),
          index = index, topics = "syncman"
    await sleepAsync(RESP_TIMEOUT)
    return

  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_speed = peer.netKbps(),
        peer_score = peer.getScore(), index = index, topics = "syncman"

  man.workers[index].status = SyncWorkerStatus.Downloading

  try:
    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(),
            peer_speed = peer.netKbps(), index = index, topics = "syncman"

      if not(checkResponse(req, data)):
        peer.updateScore(PeerScoreBadResponse)
        warn "Received blocks sequence is not in requested range",
             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(), peer_speed = peer.netKbps(),
             index = index, topics = "syncman"
        discard SyncFailure.init(SyncFailureKind.BadResponse, peer)
        return

      # Scoring will happen in `syncUpdate`.
      man.workers[index].status = SyncWorkerStatus.Processing
      await man.queue.push(req, data)
    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, index = index,
            peer_score = peer.getScore(), peer_speed = peer.netKbps(),
            topics = "syncman"
      discard SyncFailure.init(SyncFailureKind.BlockDownload, peer)
      return

  except CatchableError as exc:
    debug "Unexpected exception while receiving blocks",
            request_slot = req.slot, request_count = req.count,
            request_step = req.step, peer = peer, index = index,
            peer_score = peer.getScore(), peer_speed = peer.netKbps(),
            errMsg = exc.msg, topics = "syncman"
    return

proc syncWorker[A, B](man: SyncManager[A, B], index: int) {.async.} =
  mixin getKey, getScore, getHeadSlot

  debug "Starting syncing worker", index = index, topics = "syncman"

  while true:
    man.workers[index].status = SyncWorkerStatus.Sleeping
    # This event is going to be set until we are not in sync with network
    await man.notInSyncEvent.wait()
    man.workers[index].status = SyncWorkerStatus.WaitingPeer
    let peer = await man.pool.acquire()
    await man.syncStep(index, peer)
    man.pool.release(peer)

proc getWorkersStats[A, B](man: SyncManager[A, B]): tuple[map: string,
                                                          sleeping: int,
                                                          waiting: int,
                                                          pending: int] =
  var map = newString(len(man.workers))
  var sleeping, waiting, pending: int
  for i in 0 ..< len(man.workers):
    var ch: char
    case man.workers[i].status
      of SyncWorkerStatus.Sleeping:
        ch = 's'
        inc(sleeping)
      of SyncWorkerStatus.WaitingPeer:
        ch = 'w'
        inc(waiting)
      of SyncWorkerStatus.UpdatingStatus:
        ch = 'U'
        inc(pending)
      of SyncWorkerStatus.Requesting:
        ch = 'R'
        inc(pending)
      of SyncWorkerStatus.Downloading:
        ch = 'D'
        inc(pending)
      of SyncWorkerStatus.Processing:
        ch = 'P'
        inc(pending)
    map[i] = ch
  (map, sleeping, waiting, pending)

proc guardTask[A, B](man: SyncManager[A, B]) {.async.} =
  var pending: array[SyncWorkersCount, Future[void]]

  # Starting all the synchronization workers.
  for i in 0 ..< len(man.workers):
    let future = syncWorker[A, B](man, i)
    man.workers[i].future = future
    pending[i] = future

  # Wait for synchronization worker's failure and replace it with new one.
  while true:
    let failFuture = await one(pending)
    let index = pending.find(failFuture)
    if failFuture.failed():
      warn "Synchronization worker stopped working unexpectedly with an error",
            index = index, errMsg = failFuture.error.msg
    else:
      warn "Synchronization worker stopped working unexpectedly without error",
            index = index

    let future = syncWorker[A, B](man, index)
    man.workers[index].future = future
    pending[index] = future

proc toTimeLeftString(d: Duration): string =
  var v = d
  var res = ""
  let ndays = chronos.days(v)
  if ndays > 0:
    res = res & (if ndays < 10: "0" & $ndays else: $ndays) & "d"
    v = v - chronos.days(ndays)

  let nhours = chronos.hours(v)
  if nhours > 0:
    res = res & (if nhours < 10: "0" & $nhours else: $nhours) & "h"
    v = v - chronos.hours(nhours)
  else:
    res =  res & "00h"

  let nmins = chronos.minutes(v)
  if nmins > 0:
    res = res & (if nmins < 10: "0" & $nmins else: $nmins) & "m"
    v = v - chronos.minutes(nmins)
  else:
    res = res & "00m"
  res

proc syncLoop[A, B](man: SyncManager[A, B]) {.async.} =
  mixin getKey, getScore
  var pauseTime = 0

  asyncSpawn man.guardTask()

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

  proc averageSpeedTask() {.async.} =
    while true:
      let wallSlot = man.getLocalWallSlot()
      let headSlot = man.getLocalHeadSlot()
      let lsm1 = SyncMoment.now(man.getLocalHeadSlot())
      await sleepAsync(chronos.seconds(int(SECONDS_PER_SLOT)))
      let lsm2 = SyncMoment.now(man.getLocalHeadSlot())
      let bps =
        if lsm2.slot - lsm1.slot == 0'u64:
          0.0
        else:
          speed(lsm1, lsm2)
      inc(man.syncCount)
      man.insSyncSpeed = bps
      man.avgSyncSpeed = man.avgSyncSpeed +
                         (bps - man.avgSyncSpeed) / float(man.syncCount)
      let nsec = (float(wallSlot - headSlot) / man.avgSyncSpeed) *
                 1_000_000_000.0
      man.timeLeft = chronos.nanoseconds(int64(nsec))

  asyncSpawn averageSpeedTask()

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

    let (map, sleeping, waiting, pending) = man.getWorkersStats()

    debug "Current syncing state", workers_map = map,
          sleeping_workers_count = sleeping,
          waiting_workers_count = waiting,
          pending_workers_count = pending,
          wall_head_slot = wallSlot, local_head_slot = headSlot,
          pause_time = $chronos.seconds(pauseTime),
          avg_sync_speed = man.avgSyncSpeed, ins_sync_speed = man.insSyncSpeed,
          topics = "syncman"

    # Update status string
    man.syncStatus = map & ":" & $pending & ":" &
                     man.insSyncSpeed.formatBiggestFloat(ffDecimal, 4) & ":" &
                     man.avgSyncSpeed.formatBiggestFloat(ffDecimal, 4) & ":" &
                     man.timeLeft.toTimeLeftString() &
                     " (" & $man.queue.outSlot & ")"

    if headAge <= man.maxHeadAge:
      man.notInSyncEvent.clear()
      # We are marking SyncManager as not working only when we are in sync and
      # all sync workers are in `Sleeping` state.
      if pending > 0:
        debug "Synchronization loop waits for workers completion",
              wall_head_slot = wallSlot, local_head_slot = headSlot,
              difference = (wallSlot - headSlot), max_head_age = man.maxHeadAge,
              sleeping_workers_count = sleeping,
              waiting_workers_count = waiting, pending_workers_count = pending,
              topics = "syncman"
        man.inProgress = true
      else:
        debug "Synchronization loop sleeping", wall_head_slot = wallSlot,
              local_head_slot = headSlot, difference = (wallSlot - headSlot),
              max_head_age = man.maxHeadAge, topics = "syncman"
        man.inProgress = false
    else:
      if not(man.notInSyncEvent.isSet()):
        # We get here only if we lost sync for more then `maxHeadAge` period.
        if pending == 0:
          man.queue = SyncQueue.init(A, man.getLocalHeadSlot(),
                                     man.getLocalWallSlot(),
                                     man.chunkSize, man.getFinalizedSlot,
                                     man.verifQueues, 1)
          man.notInSyncEvent.fire()
          man.inProgress = true
      else:
        man.notInSyncEvent.fire()
        man.inProgress = true

    if queueAge <= man.rangeAge:
      # We are in requested range ``man.rangeAge``.
      man.inRangeEvent.fire()
      man.notInRangeEvent.clear()
    else:
      # We are not in requested range anymore ``man.rangeAge``.
      man.inRangeEvent.clear()
      man.notInRangeEvent.fire()

    await sleepAsync(chronos.seconds(2))

proc start*[A, B](man: SyncManager[A, B]) =
  ## Starts SyncManager's main loop.
  man.syncFut = man.syncLoop()

proc getInfo*[A, B](man: SyncManager[A, B]): SyncInfo =
  ## Returns current synchronization information for RPC call.
  let wallSlot = man.getLocalWallSlot()
  let headSlot = man.getLocalHeadSlot()

  # Even with 6 second slots (minimal presets), this overflows about 1.75
  # trillion years into the chain's existence, if headSlot is 0.
  let sync_distance = wallSlot - headSlot
  doAssert sync_distance <= high(int64).uint64

  SyncInfo(head_slot: headSlot, sync_distance: int64(sync_distance))