<commit_before>{-# LANGUAGE ScopedTypeVariables #-}

-- | Server which deals with blocks processing.

module Pos.Block.Network.Retrieval
       ( retrievalWorker
       , triggerRecovery
       , handleUnsolicitedHeaders
       , requestTip
       , requestHeaders
       , addToBlockRequestQueue
       , mkHeadersRequest
       , requestTipOuts
       ) where

import           Control.Concurrent.STM     (isEmptyTBQueue, isFullTBQueue, putTMVar,
                                             tryReadTBQueue, tryReadTMVar, tryTakeTMVar,
                                             writeTBQueue)
import           Control.Lens               (_Wrapped)
import           Control.Monad.Except       (ExceptT, runExceptT, throwError)
import           Data.List.NonEmpty         ((<|))
import qualified Data.List.NonEmpty         as NE
import           Formatting                 (build, int, sformat, shown, stext, (%))
import           Mockable                   (fork, handleAll, throw)
import           Paths_cardano_sl           (version)
import           Serokell.Util.Text         (listJson)
import           Serokell.Util.Verify       (isVerSuccess)
import           System.Wlog                (logDebug, logError, logInfo, logWarning)
import           Universum

import           Pos.Binary.Communication   ()
import           Pos.Block.Logic            (ClassifyHeaderRes (..),
                                             ClassifyHeadersRes (..), classifyHeaders,
                                             classifyNewHeader, getHeadersOlderExp,
                                             lcaWithMainChain, verifyAndApplyBlocks,
                                             withBlkSemaphore)
import qualified Pos.Block.Logic            as L
import           Pos.Block.Network.Announce (announceBlock, announceBlockOuts)
import           Pos.Block.Network.Types    (MsgBlock (..), MsgGetBlocks (..),
                                             MsgGetHeaders (..), MsgHeaders (..))
import           Pos.Block.Types            (Blund)
import           Pos.Communication.Limits   (LimitedLength, recvLimited, reifyMsgLimit)
import           Pos.Communication.Protocol (ConversationActions (..), NodeId, OutSpecs,
                                             SendActions (..), WorkerSpec, convH,
                                             toOutSpecs, worker)
import           Pos.Constants              (blkSecurityParam)
import           Pos.Context                (NodeContext (..), getNodeContext,
                                             isRecoveryMode)
import           Pos.Crypto                 (shortHashF)
import qualified Pos.DB.DB                  as DB
import           Pos.DHT.Model              (converseToNeighbors)
import           Pos.Reporting.Methods      (reportMisbehaviourMasked, reportingFatal)
import           Pos.Shutdown               (runIfNotShutdown)
import           Pos.Ssc.Class              (Ssc, SscWorkersClass)
import           Pos.Types                  (Block, BlockHeader, HasHeaderHash (..),
                                             HeaderHash, blockHeader, difficultyL,
                                             gbHeader, prevBlockL, verifyHeaders)
import           Pos.Util                   (NE, NewestFirst (..), OldestFirst (..),
                                             inAssertMode, _neHead, _neLast)
import           Pos.WorkMode               (WorkMode)

data VerifyBlocksException = VerifyBlocksException Text deriving Show
instance Exception VerifyBlocksException

retrievalWorker
    :: forall ssc m.
       (SscWorkersClass ssc, WorkMode ssc m)
    => (WorkerSpec m, OutSpecs)
retrievalWorker = worker outs retrievalWorkerImpl
  where
    outs = announceBlockOuts <>
           toOutSpecs [convH (Proxy :: Proxy MsgGetBlocks)
                             (Proxy :: Proxy (MsgBlock ssc))
                      ]

retrievalWorkerImpl
    :: forall ssc m.
       (SscWorkersClass ssc, WorkMode ssc m)
    => SendActions m -> m ()
retrievalWorkerImpl sendActions = handleAll handleTop $ do
    logDebug "Starting retrievalWorker loop"
    NodeContext{..} <- getNodeContext
    mainLoop ncBlockRetrievalQueue ncRecoveryHeader
  where
    mainLoop queue recHeaderVar = ifNotShutdown $ reportingFatal $ do
        logDebug "Waiting on the queue"
        loopCont <- atomically $ do
            qV <- tryReadTBQueue queue
            rV <- tryReadTMVar recHeaderVar
            case (qV, rV) of
                (Nothing, Nothing) -> retry
                (Just v, _)        -> pure $ Left v
                (Nothing, Just r)  -> pure $ Right r
        let onLeft ph =
                handleAll (handleBlockRetrieval recHeaderVar ph) $
                reportingFatal $ handle ph
        let tryFillQueue peerId (0 :: Int) _ =
                dropRecoveryHeaderAndRepeat recHeaderVar peerId
            tryFillQueue peerId tries action = do
                () <- action
                whenM (atomically $ isEmptyTBQueue queue) $
                    tryFillQueue peerId (tries - 1) action
        let onRight (peerId, rHeader) = tryFillQueue peerId 5 $ do
                logDebug "Queue is empty, we're in recovery mode -> querying more"
                whenJustM (mkHeadersRequest (Just $ headerHash rHeader)) $ \mghNext ->
                    handleAll (handleHeadersRecovery recHeaderVar peerId) $
                    reportingFatal $
                    reifyMsgLimit (Proxy @(MsgHeaders ssc)) $ \limPx ->
                    withConnectionTo sendActions peerId $ \_peerData ->
                        requestHeaders mghNext (Just rHeader) peerId limPx
        either onLeft onRight loopCont
        mainLoop queue recHeaderVar

    dropUpdateHeader = do
        progressHeaderVar <- ncProgressHeader <$> getNodeContext
        void $ atomically $ tryTakeTMVar progressHeaderVar<commit_msg>[CSL-779] Restore @pva701's fixes in Pos.Block.Network.Retrieval<commit_after>{-# LANGUAGE ScopedTypeVariables #-}

-- | Server which deals with blocks processing.

module Pos.Block.Network.Retrieval
       ( retrievalWorker
       , triggerRecovery
       , handleUnsolicitedHeaders
       , requestTip
       , requestHeaders
       , addToBlockRequestQueue
       , mkHeadersRequest
       , requestTipOuts
       ) where

import           Control.Concurrent.STM     (isEmptyTBQueue, isFullTBQueue, putTMVar,
                                             tryReadTBQueue, tryReadTMVar, tryTakeTMVar,
                                             writeTBQueue)
import           Control.Lens               (_Wrapped)
import           Control.Monad.Except       (ExceptT, runExceptT, throwError)
import           Data.List.NonEmpty         ((<|))
import qualified Data.List.NonEmpty         as NE
import           Formatting                 (build, int, sformat, shown, stext, (%))
import           Mockable                   (fork, handleAll, throw)
import           Paths_cardano_sl           (version)
import           Serokell.Util.Text         (listJson)
import           Serokell.Util.Verify       (isVerSuccess)
import           System.Wlog                (logDebug, logError, logInfo, logWarning)
import           Universum

import           Pos.Binary.Communication   ()
import           Pos.Block.Logic            (ClassifyHeaderRes (..),
                                             ClassifyHeadersRes (..), classifyHeaders,
                                             classifyNewHeader, getHeadersOlderExp,
                                             lcaWithMainChain, verifyAndApplyBlocks,
                                             withBlkSemaphore)
import qualified Pos.Block.Logic            as L
import           Pos.Block.Network.Announce (announceBlock, announceBlockOuts)
import           Pos.Block.Network.Types    (MsgBlock (..), MsgGetBlocks (..),
                                             MsgGetHeaders (..), MsgHeaders (..))
import           Pos.Block.Types            (Blund)
import           Pos.Communication.Limits   (LimitedLength, recvLimited, reifyMsgLimit)
import           Pos.Communication.Protocol (ConversationActions (..), NodeId, OutSpecs,
                                             SendActions (..), WorkerSpec, convH,
                                             toOutSpecs, worker)
import           Pos.Constants              (blkSecurityParam)
import           Pos.Context                (NodeContext (..), getNodeContext,
                                             isRecoveryMode)
import           Pos.Crypto                 (shortHashF)
import qualified Pos.DB.DB                  as DB
import           Pos.DHT.Model              (converseToNeighbors)
import           Pos.Reporting.Methods      (reportMisbehaviourMasked, reportingFatal)
import           Pos.Shutdown               (runIfNotShutdown)
import           Pos.Ssc.Class              (Ssc, SscWorkersClass)
import           Pos.Types                  (Block, BlockHeader, HasHeaderHash (..),
                                             HeaderHash, blockHeader, difficultyL,
                                             gbHeader, prevBlockL, verifyHeaders)
import           Pos.Util                   (NE, NewestFirst (..), OldestFirst (..),
                                             inAssertMode, _neHead, _neLast)
import           Pos.WorkMode               (WorkMode)

data VerifyBlocksException = VerifyBlocksException Text deriving Show
instance Exception VerifyBlocksException

retrievalWorker
    :: forall ssc m.
       (SscWorkersClass ssc, WorkMode ssc m)
    => (WorkerSpec m, OutSpecs)
retrievalWorker = worker outs retrievalWorkerImpl
  where
    outs = announceBlockOuts <>
           toOutSpecs [convH (Proxy :: Proxy MsgGetBlocks)
                             (Proxy :: Proxy (MsgBlock ssc))
                      ]

retrievalWorkerImpl
    :: forall ssc m.
       (SscWorkersClass ssc, WorkMode ssc m)
    => SendActions m -> m ()
retrievalWorkerImpl sendActions = handleAll handleTop $ do
    logDebug "Starting retrievalWorker loop"
    NodeContext{..} <- getNodeContext
    mainLoop ncBlockRetrievalQueue ncRecoveryHeader
  where
    mainLoop queue recHeaderVar = runIfNotShutdown $ reportingFatal version $ do
        logDebug "Waiting on the queue"
        loopCont <- atomically $ do
            qV <- tryReadTBQueue queue
            rV <- tryReadTMVar recHeaderVar
            case (qV, rV) of
                (Nothing, Nothing) -> retry
                (Just v, _)        -> pure $ Left v
                (Nothing, Just r)  -> pure $ Right r
        let onLeft ph =
                handleAll (handleBlockRetrieval recHeaderVar ph) $
                reportingFatal version $ handle ph
        let tryFillQueue peerId (0 :: Int) _ =
                dropRecoveryHeaderAndRepeat recHeaderVar peerId
            tryFillQueue peerId tries action = do
                () <- action
                whenM (atomically $ isEmptyTBQueue queue) $
                    tryFillQueue peerId (tries - 1) action
        let onRight (peerId, rHeader) = tryFillQueue peerId 5 $ do
                logDebug "Queue is empty, we're in recovery mode -> querying more"
                whenJustM (mkHeadersRequest (Just $ headerHash rHeader)) $ \mghNext ->
                    handleAll (handleHeadersRecovery recHeaderVar peerId) $
                    reportingFatal version $
                    reifyMsgLimit (Proxy @(MsgHeaders ssc)) $ \limPx ->
                    withConnectionTo sendActions peerId $ \_peerData ->
                        requestHeaders mghNext (Just rHeader) peerId limPx
        either onLeft onRight loopCont
        mainLoop queue recHeaderVar

    dropUpdateHeader = do
        progressHeaderVar <- ncProgressHeader <$> getNodeContext
        void $ atomically $ tryTakeTMVar progressHeaderVar
{'fraction_non_alphanumeric': 0.06625995593967124, 'fraction_numerical': 0.0010167768174885613, 'mean_word_length': 1.530117899249732, 'pattern_counts': {'":': 0, '<': 21, '<?xml version=': 0, '>': 27, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 52, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '650908', 'n_tokens_mistral': 2914, 'n_tokens_neox': 2788, 'n_words': 737}