content: {-# LANGUAGE RankNTypes      #-}
  {-# LANGUAGE RecordWildCards #-}
  
  -- | Server which deals with blocks processing.
  
  module Pos.Network.Block.Retrieval
         ( retrievalWorker
         ) where
  
  import           Universum
  
  import           Control.Concurrent.STM (TVar, newTVar, putTMVar, swapTMVar,
                       swapTVar, tryReadTBQueue, tryReadTMVar, tryTakeTMVar)
  import           Control.Exception.Safe (IOException, handleAny)
  import           Control.Lens (to)
  import           Control.Monad.STM (retry)
  import qualified Data.List.NonEmpty as NE
  import           Data.Time.Units (Second)
  import           Formatting (build, int, sformat, shown, (%))
  
  import           Pos.Chain.Block (Block, BlockHeader, HasHeaderHash (..),
                       HeaderHash, headerHashF)
  import           Pos.Chain.Genesis as Genesis (Config)
  import           Pos.Chain.Txp (TxpConfiguration)
  import           Pos.Core (difficultyL, isMoreDifficult)
  import           Pos.Core.Chrono (NE, OldestFirst (..), _OldestFirst)
  import           Pos.Core.Conc (delay)
  import           Pos.Core.Reporting (HasMisbehaviorMetrics)
  import           Pos.Crypto (shortHashF)
  import           Pos.DB.Block (ClassifyHeaderRes (..), classifyNewHeader,
                       getHeadersOlderExp)
  import qualified Pos.DB.BlockIndex as DB
  import           Pos.Infra.Communication.Protocol (NodeId)
  import           Pos.Infra.Diffusion.Types (Diffusion, StreamBlocks (..))
  import qualified Pos.Infra.Diffusion.Types as Diffusion
  import           Pos.Infra.Recovery.Types (RecoveryHeaderTag)
  import           Pos.Infra.Reporting (reportOrLogE, reportOrLogW)
  import           Pos.Network.Block.Logic (BlockNetLogicException (..),
                       handleBlocks, triggerRecovery)
  import           Pos.Network.Block.RetrievalQueue (BlockRetrievalQueueTag,
                       BlockRetrievalTask (..))
  import           Pos.Network.Block.WorkMode (BlockWorkMode)
  import           Pos.Util.Util (HasLens (..))
  import           Pos.Util.Wlog (logDebug, logError, logInfo, logWarning)
  
  -- I really don't like join
  {-# ANN retrievalWorker ("HLint: ignore Use join" :: Text) #-}
  
  -- | Worker that queries blocks. It has two jobs:
  --
  -- * If there are headers in 'BlockRetrievalQueue', this worker retrieves
  --   blocks according to that queue.
  --
  -- * If recovery is in progress, this worker keeps recovery going by asking
  --   headers (and then switching to block retrieval on next loop iteration).
  --
  -- If both happen at the same time, 'BlockRetrievalQueue' takes precedence.
  --
  retrievalWorker
      :: forall ctx m.
         ( BlockWorkMode ctx m
         , HasMisbehaviorMetrics ctx
         )
      => Genesis.Config
      -> TxpConfiguration
      -> Diffusion m -> m ()
  retrievalWorker genesisConfig txpConfig diffusion = do
      logInfo "Starting retrievalWorker loop"
      mainLoop
    where
      mainLoop = do
          queue        <- view (lensOf @BlockRetrievalQueueTag)
          recHeaderVar <- view (lensOf @RecoveryHeaderTag)
          logDebug "Waiting on the block queue or recovery header var"
          -- Reading the queue is a priority, because it sets the recovery
          -- variable in case the header is classified as alternative. So if the
          -- queue contains lots of headers after a long delay, we'll first
          -- iterate over them and set recovery variable to the latest one, and
          -- only then we'll do recovery.
          thingToDoNext <- atomically $ do
              mbQueuedHeadersChunk <- tryReadTBQueue queue
              mbRecHeader <- tryReadTMVar recHeaderVar
              case (mbQueuedHeadersChunk, mbRecHeader) of
                  (Nothing, Nothing) -> retry
                  -- Dispatch the task
                  (Just (nodeId, task), _) ->
                      pure (handleBlockRetrieval nodeId task)
                  -- No tasks & the recovery header is set => do the recovery
                  (_, Just (nodeId, rHeader))  ->
                      pure (handleRecoveryWithHandler nodeId rHeader)
  
          -- Exception handlers are installed locally, on the 'thingToDoNext',
          -- to ensure that network troubles, for instance, do not kill the
          -- worker.
          () <- thingToDoNext
          mainLoop
  
      -----------------
  
      -- That's the first queue branch (task dispatching).
      handleBlockRetrieval nodeId BlockRetrievalTask{..} =
          handleAny (handleRetrievalE nodeId brtHeader) $ do
              logDebug $ sformat
                  ("Block retrieval queue task received, nodeId="%build%
                   ", header="%build%", continues="%build)
                  nodeId
                  (headerHash brtHeader)
                  brtContinues
              (if brtContinues then handleContinues else handleAlternative)
                  nodeId
                  brtHeader
  
      -- When we have a continuation of the chain, just try to get and apply it.
      handleContinues nodeId header = do
          let hHash = headerHash header
          logDebug $ "handleContinues: " <> pretty hHash
          classifyNewHeader genesisConfig header >>= \case
              CHContinues ->
                  void $ getProcessBlocks genesisConfig txpConfig diffusion nodeId hHash [hHash]
              res -> logDebug $
                  "processContHeader: expected header to " <>
                  "be continuation, but it's " <> show res
  
      -- When we have an alternative header, we should check whether it's actually
      -- recovery mode (server side should send us headers as a proof) and then
      -- enter recovery mode.
      handleAlternative nodeId header = do
          logDebug $ "handleAlternative: " <> pretty (headerHash header)
          classifyNewHeader genesisConfig header >>= \case
              CHInvalid _ ->
                  logError "handleAlternative: invalid header got into retrievalWorker queue"
              CHUseless _ ->
                  logDebug $
                  sformat ("handleAlternative: header "%build%" became useless, ignoring it")
                          header
              _ -> do
                  logDebug "handleAlternative: considering header for recovery mode"
                  -- CSL-1514
                  updateRecoveryHeader nodeId header
  
      -- Squelch the exception and continue. Used with 'handleAny' from
      -- safe-exceptions so it will let async exceptions pass.
      handleRetrievalE nodeId cHeader e = do
          reportOrLogW (sformat
              ("handleRetrievalE: error handling nodeId="%build%", header="%build%": ")
              nodeId (headerHash cHeader)) e
  
      -----------------
  
      handleRecoveryWithHandler nodeId rHeader =
          handleAny (handleRecoveryE nodeId rHeader) $
          handleRecovery nodeId rHeader
  
      -- We immediately drop recovery mode/header and request tips
      -- again.
      handleRecoveryE nodeId rHeader e = do
          -- REPORT:ERROR 'reportOrLogW' in block retrieval worker/recovery.
          reportOrLogW (sformat errfmt nodeId (headerHash rHeader)) e
              `catch` handleIOException
          dropRecoveryHeaderAndRepeat genesisConfig diffusion nodeId
          where
              errfmt = "handleRecoveryE: error handling nodeId="%build%", header="%headerHashF%": "
  
              handleIOException :: IOException -> m ()
              handleIOException _ = logError $ sformat (errfmt%shown) nodeId (headerHash rHeader) e
  
      -- Recovery handling. We assume that header in the recovery variable is
      -- appropriate and just query headers/blocks.
      handleRecovery :: NodeId -> BlockHeader -> m ()
      handleRecovery nodeId rHeader = do
          logDebug "Block retrieval queue is empty and we're in recovery mode,\
                   \ so we will fetch more blocks"
          whenM (fmap isJust $ DB.getHeader $ headerHash rHeader) $
              -- How did we even got into recovery then?
              throwM $ DialogUnexpected $ "handleRecovery: recovery header is " <>
                                          "already present in db"
          logDebug "handleRecovery: fetching blocks"
          checkpoints <- toList <$> getHeadersOlderExp genesisConfig Nothing
          void $ streamProcessBlocks genesisConfig
                                     txpConfig
                                     diffusion
                                     nodeId
                                     (headerHash rHeader)
                                     checkpoints
  
  ----------------------------------------------------------------------------
  -- Entering and exiting recovery mode
  ----------------------------------------------------------------------------
  
  -- | Result of attempt to update recovery header.
  data UpdateRecoveryResult ssc
      = RecoveryStarted NodeId BlockHeader
        -- ^ Recovery header was absent, so we've set it.
      | RecoveryShifted NodeId BlockHeader NodeId BlockHeader
        -- ^ Header was present, but we've replaced it with another
        -- (more difficult) one.
      | RecoveryContinued NodeId BlockHeader
        -- ^ Header is good, but is irrelevant, so recovery variable is
        -- unchanged.
  
  -- | Be careful to run this in the same thread that ends recovery mode
  -- (or synchronise those threads with an MVar), otherwise a race
  -- condition can occur where we are caught in the recovery mode
  -- indefinitely.
  updateRecoveryHeader
      :: BlockWorkMode ctx m
      => NodeId
      -> BlockHeader
      -> m ()
  updateRecoveryHeader nodeId hdr = do
      recHeaderVar <- view (lensOf @RecoveryHeaderTag)
      logDebug "Updating recovery header..."
      updated <- atomically $ do
          mbRecHeader <- tryReadTMVar recHeaderVar
          case mbRecHeader of
              Nothing -> do
                  putTMVar recHeaderVar (nodeId, hdr)
                  return $ RecoveryStarted nodeId hdr
              Just (oldNodeId, oldHdr) -> do
                  let needUpdate = hdr `isMoreDifficult` oldHdr
                  if needUpdate
                      then swapTMVar recHeaderVar (nodeId, hdr) $>
                           RecoveryShifted oldNodeId oldHdr nodeId hdr
                      else return $ RecoveryContinued oldNodeId oldHdr
      logDebug $ case updated of
          RecoveryStarted rNodeId rHeader -> sformat
              ("Recovery started with nodeId="%build%" and tip="%build)
              rNodeId
              (headerHash rHeader)
          RecoveryShifted rNodeId' rHeader' rNodeId rHeader -> sformat
              ("Recovery shifted from nodeId="%build%" and tip="%build%
               " to nodeId="%build%" and tip="%build)
              rNodeId' (headerHash rHeader')
              rNodeId  (headerHash rHeader)
          RecoveryContinued rNodeId rHeader -> sformat
              ("Recovery continued with nodeId="%build%" and tip="%build)
              rNodeId
              (headerHash rHeader)
  
  -- | The returned 'Bool' signifies whether given peer was kicked and recovery
  -- was stopped.
  --
  -- NB. The reason @nodeId@ is passed is that we want to avoid a race
  -- condition. If you work with peer P and header H, after failure you want to
  -- drop communication with P; however, if at the same time a new block
  -- arrives and another thread replaces peer and header to (P2, H2), you want
  -- to continue working with P2 and ignore the exception that happened with P.
  -- So, @nodeId@ is used to check that the peer wasn't replaced mid-execution.
  dropRecoveryHeader
      :: BlockWorkMode ctx m
      => NodeId
      -> m Bool
  dropRecoveryHeader nodeId = do
      recHeaderVar <- view (lensOf @RecoveryHeaderTag)
      (kicked,realPeer) <- atomically $ do
          let processKick (peer,_) = do
                  let p = peer == nodeId
                  when p $ void $ tryTakeTMVar recHeaderVar
                  pure (p, Just peer)
          maybe (pure (True,Nothing)) processKick =<< tryReadTMVar recHeaderVar
      when kicked $ logWarning $
          sformat ("Recovery mode communication dropped with peer "%build) nodeId
      unless kicked $
          logDebug $ "Recovery mode wasn't disabled: " <>
                     maybe "noth" show realPeer <> " vs " <> show nodeId
      pure kicked
  
  -- | Drops the recovery header and, if it was successful, queries the tips.
  dropRecoveryHeaderAndRepeat
      :: BlockWorkMode ctx m
      => Genesis.Config
      -> Diffusion m
      -> NodeId
      -> m ()
  dropRecoveryHeaderAndRepeat genesisConfig diffusion nodeId = do
      kicked <- dropRecoveryHeader nodeId
      when kicked $ attemptRestartRecovery
    where
      attemptRestartRecovery = do
          logDebug "Attempting to restart recovery"
          -- FIXME why delay? Why 2 seconds?
          delay (2 :: Second)
          handleAny handleRecoveryTriggerE $ triggerRecovery genesisConfig diffusion
          logDebug "Attempting to restart recovery over"
      handleRecoveryTriggerE =
          -- REPORT:ERROR 'reportOrLogE' somewhere in block retrieval.
          reportOrLogE $ "Exception happened while trying to trigger " <>
                         "recovery inside dropRecoveryHeaderAndRepeat: "
  
  -- Returns only if blocks were successfully downloaded and
  -- processed. Throws exception if something goes wrong.
  getProcessBlocks
      :: forall ctx m
       . (BlockWorkMode ctx m, HasMisbehaviorMetrics ctx)
      => Genesis.Config
      -> TxpConfiguration
      -> Diffusion m
      -> NodeId
      -> HeaderHash
      -> [HeaderHash]
      -> m ()
  getProcessBlocks genesisConfig txpConfig diffusion nodeId desired checkpoints = do
      result <- Diffusion.getBlocks diffusion nodeId desired checkpoints
      case OldestFirst <$> nonEmpty (getOldestFirst result) of
        Nothing -> do
            let msg = sformat ("getProcessBlocks: diffusion returned []"%
                               " on request to fetch "%shortHashF%" from peer "%build)
                              desired nodeId
            throwM $ DialogUnexpected msg
        Just (blocks :: OldestFirst NE Block) -> do
            recHeaderVar <- view (lensOf @RecoveryHeaderTag)
            logDebug $ sformat
                ("Retrieved "%int%" blocks")
                (blocks ^. _OldestFirst . to NE.length)
            handleBlocks genesisConfig txpConfig blocks diffusion
            -- If we've downloaded any block with bigger
            -- difficulty than ncRecoveryHeader, we're
            -- gracefully exiting recovery mode.
            let isMoreDifficultThan b x = b ^. difficultyL >= x ^. difficultyL
            exitedRecovery <- atomically $ tryReadTMVar recHeaderVar >>= \case
                -- We're not in recovery mode? That must be ok.
                Nothing -> pure False
                -- If we're in recovery mode we should exit it if
                -- any block is more difficult than one in
                -- recHeader.
                Just (_, rHeader) ->
                    if any (`isMoreDifficultThan` rHeader) blocks
                    then isJust <$> tryTakeTMVar recHeaderVar
                    else pure False
            when exitedRecovery $
                logInfo "Recovery mode exited gracefully on receiving block we needed"
  
  -- Attempts to catch up by streaming blocks from peer.
  -- Will fall back to getProcessBlocks if streaming is disabled
  -- or not supported by peer.
  streamProcessBlocks
      :: forall ctx m
       . (BlockWorkMode ctx m, HasMisbehaviorMetrics ctx)
      => Genesis.Config
      -> TxpConfiguration
      -> Diffusion m
      -> NodeId
      -> HeaderHash
      -> [HeaderHash]
      -> m ()
  streamProcessBlocks genesisConfig txpConfig diffusion nodeId desired checkpoints = do
      logInfo "streaming start"
      mostDifficultBlock <- atomically $ newTVar Nothing
      r <- Diffusion.streamBlocks diffusion nodeId desired checkpoints (writeCallback mostDifficultBlock)
      case r of
           Nothing -> do
               logInfo "streaming not supported, reverting to batch mode"
               getProcessBlocks genesisConfig txpConfig diffusion nodeId desired checkpoints
           Just _  -> do
               logInfo "streaming done"
  
               recHeaderVar <- view (lensOf @RecoveryHeaderTag)
               exitedRecovery <- atomically $ do
                   mbMostDifficult <- readTVar mostDifficultBlock
                   mbRecHeader <- tryReadTMVar recHeaderVar
                   case (mbMostDifficult, mbRecHeader) of
                        (Nothing, _)                 -> pure False -- We have not gotten a single block
                        (Just _, Nothing)            -> pure False -- We where not in recovery?
                        (Just mostDifficult, Just (_, recHeader)) ->
                            if (mostDifficult ^. difficultyL) >= (recHeader ^. difficultyL)
                               then isJust <$> tryTakeTMVar recHeaderVar
                               else pure False
  
               if exitedRecovery
                  then do
                      logInfo "Recovery mode exited gracefully on receiving block we needed"
                      return ()
                  else do -- Streaming stopped but we didn't make any progress
                      _ <- dropRecoveryHeaderAndRepeat genesisConfig diffusion nodeId
                      return ()
    where
      writeCallback :: TVar (Maybe Block) -> StreamBlocks Block m ()
      writeCallback mostDifficultBlock = StreamBlocks
        { streamBlocksMore = \blks -> do
            _ <- atomically $ swapTVar mostDifficultBlock (Just (NE.head blks))
            _ <- handleBlocks genesisConfig txpConfig (OldestFirst (NE.reverse $ blks)) diffusion
            pure (writeCallback mostDifficultBlock)
        , streamBlocksDone = pure ()
        }
  