diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,12 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/) and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.html). +## 0.9.6 +### Change +- Randomize known peers instead of keeping scores. +- Simplify peer management code to avoid freezes. +- Merge logic for chain and manager. + ## 0.9.5 ### Change - Do not record new peers in database when peer discovery is disabled. diff --git a/haskoin-node.cabal b/haskoin-node.cabal --- a/haskoin-node.cabal +++ b/haskoin-node.cabal @@ -1,11 +1,13 @@ --- This file has been generated from package.yaml by hpack version 0.28.2. +cabal-version: 1.12 + +-- This file has been generated from package.yaml by hpack version 0.31.1. -- -- see: https://github.com/sol/hpack -- --- hash: de83a76e2e9d853065c60347d1ed2c080fe552cf04ca51797f0e873974581ea0 +-- hash: 153005bf793d856dc57a198e58d683234d72138315022bdd1f65e5ef649d5109 name: haskoin-node -version: 0.9.5 +version: 0.9.6 synopsis: Haskoin Node P2P library for Bitcoin and Bitcoin Cash description: Bitcoin and Bitcoin Cash peer-to-peer protocol library featuring headers-first synchronisation. category: Bitcoin, Finance, Network @@ -16,10 +18,9 @@ license: PublicDomain license-file: UNLICENSE build-type: Simple -cabal-version: >= 1.10 extra-source-files: - CHANGELOG.md README.md + CHANGELOG.md source-repository head type: git @@ -29,10 +30,8 @@ exposed-modules: Haskoin.Node Network.Haskoin.Node.Chain - Network.Haskoin.Node.Chain.Logic Network.Haskoin.Node.Common Network.Haskoin.Node.Manager - Network.Haskoin.Node.Manager.Logic Network.Haskoin.Node.Peer other-modules: Paths_haskoin_node @@ -44,6 +43,7 @@ , cereal , conduit , conduit-extra + , containers , data-default , hashable , haskoin-core @@ -76,6 +76,11 @@ , base >=4.9 && <5 , bytestring , cereal + , conduit + , conduit-extra + , containers + , data-default + , hashable , haskoin-core , haskoin-node , hspec @@ -84,8 +89,15 @@ , network , nqe , random + , resourcet , rocksdb-haskell + , rocksdb-query , safe + , string-conversions + , text + , time + , transformers , unliftio + , unordered-containers default-language: Haskell2010 build-tool-depends: hspec-discover:hspec-discover diff --git a/src/Network/Haskoin/Node/Chain.hs b/src/Network/Haskoin/Node/Chain.hs --- a/src/Network/Haskoin/Node/Chain.hs +++ b/src/Network/Haskoin/Node/Chain.hs @@ -7,6 +7,7 @@ {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE TemplateHaskell #-} {-# LANGUAGE UndecidableInstances #-} +{-# OPTIONS_GHC -fno-warn-orphans #-} {-| Module : Network.Haskoin.Node.Chain Copyright : No rights reserved @@ -21,20 +22,28 @@ ( chain ) where -import Control.Monad +import Control.Monad.Except import Control.Monad.Logger import Control.Monad.Reader import Control.Monad.Trans.Maybe +import qualified Data.ByteString as B +import Data.Default +import Data.List +import Data.Maybe +import Data.Serialize as S import Data.String.Conversions import Data.Time.Clock.POSIX -import Network.Haskoin.Block -import Network.Haskoin.Network -import Network.Haskoin.Node.Chain.Logic +import Data.Word +import Database.RocksDB (DB) +import qualified Database.RocksDB as R +import Database.RocksDB.Query as R +import Haskoin import Network.Haskoin.Node.Common import NQE import System.Random import UnliftIO import UnliftIO.Concurrent +import UnliftIO.Resource type MonadChain m = (MonadLoggerIO m, MonadChainLogic ChainConfig Peer m) @@ -176,3 +185,256 @@ threadDelay =<< liftIO (randomRIO (250 * 1000, 1000 * 1000)) ChainPing `send` ch + +-- | Version of the database. +dataVersion :: Word32 +dataVersion = 1 + +-- | Database key for version. +data ChainDataVersionKey = ChainDataVersionKey + deriving (Eq, Ord, Show) + +instance Key ChainDataVersionKey +instance KeyValue ChainDataVersionKey Word32 + +instance Serialize ChainDataVersionKey where + get = do + guard . (== 0x92) =<< S.getWord8 + return ChainDataVersionKey + put ChainDataVersionKey = S.putWord8 0x92 + +data ChainSync p = ChainSync + { chainSyncPeer :: !p + , chainTimestamp :: !Timestamp + , chainHighest :: !(Maybe BlockNode) + } + +-- | Mutable state for the header chain process. +data ChainState p = ChainState + { chainSyncing :: !(Maybe (ChainSync p)) + -- ^ peer to sync against and time of last received message + , newPeers :: ![p] + -- ^ queue of peers to sync against + , mySynced :: !Bool + -- ^ has the header chain ever been considered synced? + } + +-- | Key for block header in database. +newtype BlockHeaderKey = BlockHeaderKey BlockHash deriving (Eq, Show) + +instance Serialize BlockHeaderKey where + get = do + guard . (== 0x90) =<< getWord8 + BlockHeaderKey <$> get + put (BlockHeaderKey bh) = do + putWord8 0x90 + put bh + +-- | Key for best block in database. +data BestBlockKey = BestBlockKey deriving (Eq, Show) + +instance KeyValue BlockHeaderKey BlockNode +instance KeyValue BestBlockKey BlockNode + +instance Serialize BestBlockKey where + get = do + guard . (== 0x91) =<< getWord8 + return BestBlockKey + put BestBlockKey = putWord8 0x91 + +-- | Type alias for monad commonly used in this module. +type MonadChainLogic a p m + = (BlockHeaders m, MonadReader (ChainReader a p) m) + +-- | Reader for header synchronization code. +data ChainReader a p = ChainReader + { myReader :: !a + -- ^ placeholder for upstream data + , myChainDB :: !DB + -- ^ database handle + , chainState :: !(TVar (ChainState p)) + -- ^ mutable state for header synchronization + } + +instance (Monad m, MonadIO m, MonadReader (ChainReader a p) m) => + BlockHeaders m where + addBlockHeader bn = do + db <- asks myChainDB + R.insert db (BlockHeaderKey (headerHash (nodeHeader bn))) bn + getBlockHeader bh = do + db <- asks myChainDB + retrieve db def (BlockHeaderKey bh) + getBestBlockHeader = do + db <- asks myChainDB + retrieve db def BestBlockKey >>= \case + Nothing -> error "Could not get best block from database" + Just b -> return b + setBestBlockHeader bn = do + db <- asks myChainDB + R.insert db BestBlockKey bn + addBlockHeaders bns = do + db <- asks myChainDB + writeBatch db (map f bns) + where + f bn = insertOp (BlockHeaderKey (headerHash (nodeHeader bn))) bn + +-- | Initialize header database. If version is different from current, the +-- database is purged of conflicting elements first. +initChainDB :: (MonadChainLogic a p m, MonadUnliftIO m) => Network -> m () +initChainDB net = do + db <- asks myChainDB + ver <- retrieve db def ChainDataVersionKey + when (ver /= Just dataVersion) purgeChainDB + R.insert db ChainDataVersionKey dataVersion + retrieve db def BestBlockKey >>= \b -> + when (isNothing (b :: Maybe BlockNode)) $ do + addBlockHeader (genesisNode net) + setBestBlockHeader (genesisNode net) + +-- | Purge database of elements having keys that may conflict with those used in +-- this module. +purgeChainDB :: (MonadChainLogic a p m, MonadUnliftIO m) => m () +purgeChainDB = do + db <- asks myChainDB + runResourceT . R.withIterator db def $ \it -> do + R.iterSeek it $ B.singleton 0x90 + recurse_delete it db + where + recurse_delete it db = + R.iterKey it >>= \case + Nothing -> return () + Just k + | B.head k == 0x90 || B.head k == 0x91 -> do + R.delete db def k + R.iterNext it + recurse_delete it db + | otherwise -> return () + +-- | Import a bunch of continuous headers. Returns 'True' if the number of +-- headers is 2000, which means that there are possibly more headers to sync +-- from whatever peer delivered these. +importHeaders :: + (MonadIO m, BlockHeaders m, MonadChainLogic a p m) + => Network + -> Timestamp + -> [BlockHeader] + -> m (Either PeerException Bool) +importHeaders net now hs = + runExceptT $ + lift (connectBlocks net now hs) >>= \case + Right _ -> do + case hs of + [] -> return () + _ -> do + bb <- getBlockHeader (headerHash (last hs)) + box <- asks chainState + atomically . modifyTVar box $ \s -> + s + { chainSyncing = + (\x -> x {chainHighest = bb}) <$> + chainSyncing s + } + case length hs of + 2000 -> return False + _ -> return True + Left _ -> throwError PeerSentBadHeaders + +-- | Check if best block header is in sync with the rest of the block chain by +-- comparing the best block with the current time, verifying that there are no +-- peers in the queue to be synced, and no peer is being synced at the moment. +-- This function will only return 'True' once. It should be used to decide +-- whether to notify other processes that the header chain has been synced. The +-- state of the chain will be flipped to synced when this function returns +-- 'True'. +notifySynced :: (MonadIO m, MonadChainLogic a p m) => Timestamp -> m Bool +notifySynced now = + fmap isJust $ + runMaybeT $ do + bb <- getBestBlockHeader + guard (now - blockTimestamp (nodeHeader bb) < 2 * 60 * 60) + st <- asks chainState + MaybeT . atomically . runMaybeT $ do + s <- lift $ readTVar st + guard . isNothing $ chainSyncing s + guard . null $ newPeers s + guard . not $ mySynced s + lift $ writeTVar st s {mySynced = True} + return () + +-- | Get next peer to sync against from the queue. +nextPeer :: (MonadIO m, MonadChainLogic a p m) => m (Maybe p) +nextPeer = listToMaybe . newPeers <$> (asks chainState >>= readTVarIO) + +-- | Set a syncing peer and generate a 'GetHeaders' data structure with a block +-- locator to send to that peer for syncing. +syncHeaders :: + (Eq p, MonadChainLogic a p m, MonadIO m) + => Timestamp + -> BlockNode + -> p + -> m GetHeaders +syncHeaders now bb p = do + st <- asks chainState + atomically . modifyTVar st $ \s -> + s + { chainSyncing = + Just + ChainSync + { chainSyncPeer = p + , chainTimestamp = now + , chainHighest = Nothing + } + , newPeers = delete p (newPeers s) + } + loc <- blockLocator bb + return + GetHeaders + { getHeadersVersion = myVersion + , getHeadersBL = loc + , getHeadersHashStop = + "0000000000000000000000000000000000000000000000000000000000000000" + } + +-- | Set the time of last received data to now if a syncing peer is active. +setLastReceived :: (MonadChainLogic a p m, MonadIO m) => Timestamp -> m () +setLastReceived now = do + st <- asks chainState + atomically . modifyTVar st $ \s -> + s {chainSyncing = (\p -> p {chainTimestamp = now}) <$> chainSyncing s} + +-- | Add a new peer to the queue of peers to sync against. +addPeer :: (Eq p, MonadIO m, MonadChainLogic a p m) => p -> m () +addPeer p = do + st <- asks chainState + atomically . modifyTVar st $ \s -> s {newPeers = nub (p : newPeers s)} + +-- | Get syncing peer if there is one. +getSyncingPeer :: (MonadChainLogic a p m, MonadIO m) => m (Maybe p) +getSyncingPeer = fmap chainSyncPeer . chainSyncing <$> (readTVarIO =<< asks chainState) + +-- | Return 'True' if the chain has ever been considered synced. it will always +-- return 'True', even if the chain gets out of sync for any reason. +isSynced :: (MonadChainLogic a p m, MonadIO m) => m Bool +isSynced = mySynced <$> (asks chainState >>= readTVarIO) + +-- | Remove a peer from the queue of peers to sync and unset the syncing peer if +-- it is set to the provided value. +finishPeer :: (Eq p, MonadIO m, MonadChainLogic a p m) => p -> m () +finishPeer p = + asks chainState >>= \st -> + atomically . modifyTVar st $ \s -> + s + { newPeers = delete p (newPeers s) + , chainSyncing = + case chainSyncing s of + Just ChainSync { chainSyncPeer = p' + , chainTimestamp = _ + , chainHighest = _ + } + | p == p' -> Nothing + _ -> chainSyncing s + } + +-- | Return syncing peer data. +chainSyncingPeer :: (MonadChainLogic a p m, MonadIO m) => m (Maybe (ChainSync p)) +chainSyncingPeer = chainSyncing <$> (readTVarIO =<< asks chainState) diff --git a/src/Network/Haskoin/Node/Chain/Logic.hs b/src/Network/Haskoin/Node/Chain/Logic.hs deleted file mode 100644 --- a/src/Network/Haskoin/Node/Chain/Logic.hs +++ /dev/null @@ -1,312 +0,0 @@ -{-# LANGUAGE ConstraintKinds #-} -{-# LANGUAGE ExistentialQuantification #-} -{-# LANGUAGE FlexibleContexts #-} -{-# LANGUAGE FlexibleInstances #-} -{-# LANGUAGE LambdaCase #-} -{-# LANGUAGE MultiParamTypeClasses #-} -{-# LANGUAGE OverloadedStrings #-} -{-# LANGUAGE UndecidableInstances #-} -{-# OPTIONS_GHC -fno-warn-orphans #-} -{-| -Module : Network.Haskoin.Node.Chain.Logic -Copyright : No rights reserved -License : UNLICENSE -Maintainer : xenog@protonmail.com -Stability : experimental -Portability : POSIX - -State and code for block header synchronization. --} -module Network.Haskoin.Node.Chain.Logic where - -import Control.Monad -import Control.Monad.Except -import Control.Monad.Reader -import Control.Monad.Trans.Maybe -import qualified Data.ByteString as B -import Data.Default -import Data.List -import Data.Maybe -import Data.Serialize as S -import Data.Word -import Database.RocksDB (DB) -import qualified Database.RocksDB as R -import Database.RocksDB.Query as R -import Haskoin -import Network.Haskoin.Node.Common -import UnliftIO -import UnliftIO.Resource - --- | Version of the database. -dataVersion :: Word32 -dataVersion = 1 - --- | Database key for version. -data ChainDataVersionKey = ChainDataVersionKey - deriving (Eq, Ord, Show) - -instance Key ChainDataVersionKey -instance KeyValue ChainDataVersionKey Word32 - -instance Serialize ChainDataVersionKey where - get = do - guard . (== 0x92) =<< S.getWord8 - return ChainDataVersionKey - put ChainDataVersionKey = S.putWord8 0x92 - -data ChainSync p = ChainSync - { chainSyncPeer :: !p - , chainTimestamp :: !Timestamp - , chainHighest :: !(Maybe BlockNode) - } - --- | Mutable state for the header chain process. -data ChainState p = ChainState - { chainSyncing :: !(Maybe (ChainSync p)) - -- ^ peer to sync against and time of last received message - , newPeers :: ![p] - -- ^ queue of peers to sync against - , mySynced :: !Bool - -- ^ has the header chain ever been considered synced? - } - --- | Key for block header in database. -newtype BlockHeaderKey = BlockHeaderKey BlockHash deriving (Eq, Show) - -instance Serialize BlockHeaderKey where - get = do - guard . (== 0x90) =<< getWord8 - BlockHeaderKey <$> get - put (BlockHeaderKey bh) = do - putWord8 0x90 - put bh - --- | Key for best block in database. -data BestBlockKey = BestBlockKey deriving (Eq, Show) - -instance KeyValue BlockHeaderKey BlockNode -instance KeyValue BestBlockKey BlockNode - -instance Serialize BestBlockKey where - get = do - guard . (== 0x91) =<< getWord8 - return BestBlockKey - put BestBlockKey = putWord8 0x91 - --- | Type alias for monad commonly used in this module. -type MonadChainLogic a p m - = (BlockHeaders m, MonadReader (ChainReader a p) m) - --- | Reader for header synchronization code. -data ChainReader a p = ChainReader - { myReader :: !a - -- ^ placeholder for upstream data - , myChainDB :: !DB - -- ^ database handle - , chainState :: !(TVar (ChainState p)) - -- ^ mutable state for header synchronization - } - -instance (Monad m, MonadIO m, MonadReader (ChainReader a p) m) => - BlockHeaders m where - addBlockHeader bn = do - db <- asks myChainDB - R.insert db (BlockHeaderKey (headerHash (nodeHeader bn))) bn - getBlockHeader bh = do - db <- asks myChainDB - retrieve db def (BlockHeaderKey bh) - getBestBlockHeader = do - db <- asks myChainDB - retrieve db def BestBlockKey >>= \case - Nothing -> error "Could not get best block from database" - Just b -> return b - setBestBlockHeader bn = do - db <- asks myChainDB - R.insert db BestBlockKey bn - addBlockHeaders bns = do - db <- asks myChainDB - writeBatch db (map f bns) - where - f bn = insertOp (BlockHeaderKey (headerHash (nodeHeader bn))) bn - --- | Initialize header database. If version is different from current, the --- database is purged of conflicting elements first. -initChainDB :: (MonadChainLogic a p m, MonadUnliftIO m) => Network -> m () -initChainDB net = do - db <- asks myChainDB - ver <- retrieve db def ChainDataVersionKey - when (ver /= Just dataVersion) purgeChainDB - R.insert db ChainDataVersionKey dataVersion - retrieve db def BestBlockKey >>= \b -> - when (isNothing (b :: Maybe BlockNode)) $ do - addBlockHeader (genesisNode net) - setBestBlockHeader (genesisNode net) - --- | Purge database of elements having keys that may conflict with those used in --- this module. -purgeChainDB :: (MonadChainLogic a p m, MonadUnliftIO m) => m () -purgeChainDB = do - db <- asks myChainDB - runResourceT . R.withIterator db def $ \it -> do - R.iterSeek it $ B.singleton 0x90 - recurse_delete it db - where - recurse_delete it db = - R.iterKey it >>= \case - Nothing -> return () - Just k - | B.head k == 0x90 || B.head k == 0x91 -> do - R.delete db def k - R.iterNext it - recurse_delete it db - | otherwise -> return () - --- | Import a bunch of continuous headers. Returns 'True' if the number of --- headers is 2000, which means that there are possibly more headers to sync --- from whatever peer delivered these. -importHeaders :: - (MonadIO m, BlockHeaders m, MonadChainLogic a p m) - => Network - -> Timestamp - -> [BlockHeader] - -> m (Either PeerException Bool) -importHeaders net now hs = - runExceptT $ - lift (connectBlocks net now hs) >>= \case - Right _ -> do - case hs of - [] -> return () - _ -> do - bb <- getBlockHeader (headerHash (last hs)) - box <- asks chainState - atomically . modifyTVar box $ \s -> - s - { chainSyncing = - (\x -> x {chainHighest = bb}) <$> - chainSyncing s - } - case length hs of - 2000 -> return False - _ -> return True - Left _ -> throwError PeerSentBadHeaders - --- | Check if best block header is in sync with the rest of the block chain by --- comparing the best block with the current time, verifying that there are no --- peers in the queue to be synced, and no peer is being synced at the moment. --- This function will only return 'True' once. It should be used to decide --- whether to notify other processes that the header chain has been synced. The --- state of the chain will be flipped to synced when this function returns --- 'True'. -notifySynced :: (MonadIO m, MonadChainLogic a p m) => Timestamp -> m Bool -notifySynced now = - fmap isJust $ - runMaybeT $ do - bb <- getBestBlockHeader - guard (now - blockTimestamp (nodeHeader bb) < 2 * 60 * 60) - st <- asks chainState - MaybeT . atomically . runMaybeT $ do - s <- lift $ readTVar st - guard . isNothing $ chainSyncing s - guard . null $ newPeers s - guard . not $ mySynced s - lift $ writeTVar st s {mySynced = True} - return () - --- | Get next peer to sync against from the queue. -nextPeer :: (MonadIO m, MonadChainLogic a p m) => m (Maybe p) -nextPeer = listToMaybe . newPeers <$> (asks chainState >>= readTVarIO) - --- | Set a syncing peer and generate a 'GetHeaders' data structure with a block --- locator to send to that peer for syncing. -syncHeaders :: - (Eq p, MonadChainLogic a p m, MonadIO m) - => Timestamp - -> BlockNode - -> p - -> m GetHeaders -syncHeaders now bb p = do - st <- asks chainState - atomically . modifyTVar st $ \s -> - s - { chainSyncing = - Just - ChainSync - { chainSyncPeer = p - , chainTimestamp = now - , chainHighest = Nothing - } - , newPeers = delete p (newPeers s) - } - loc <- blockLocator bb - return - GetHeaders - { getHeadersVersion = myVersion - , getHeadersBL = loc - , getHeadersHashStop = - "0000000000000000000000000000000000000000000000000000000000000000" - } - --- | Set the time of last received data to now if a syncing peer is active. -setLastReceived :: (MonadChainLogic a p m, MonadIO m) => Timestamp -> m () -setLastReceived now = do - st <- asks chainState - atomically . modifyTVar st $ \s -> - s {chainSyncing = (\p -> p {chainTimestamp = now}) <$> chainSyncing s} - --- | Add a new peer to the queue of peers to sync against. -addPeer :: (Eq p, MonadIO m, MonadChainLogic a p m) => p -> m () -addPeer p = do - st <- asks chainState - atomically . modifyTVar st $ \s -> s {newPeers = nub (p : newPeers s)} - --- | Get syncing peer if there is one. -getSyncingPeer :: (MonadChainLogic a p m, MonadIO m) => m (Maybe p) -getSyncingPeer = fmap chainSyncPeer . chainSyncing <$> (readTVarIO =<< asks chainState) - --- | Set syncing peer to the pone provided. -setSyncingPeer :: (MonadChainLogic a p m, MonadIO m) => Timestamp -> p -> m () -setSyncingPeer now p = - asks chainState >>= \v -> - atomically . modifyTVar v $ \s -> - s - { chainSyncing = - Just - ChainSync - { chainSyncPeer = p - , chainTimestamp = now - , chainHighest = Nothing - } - } - --- | Return 'True' if the chain has ever been considered synced. it will always --- return 'True', even if the chain gets out of sync for any reason. -isSynced :: (MonadChainLogic a p m, MonadIO m) => m Bool -isSynced = mySynced <$> (asks chainState >>= readTVarIO) - --- | Set chain as synced. -setSynced :: (MonadChainLogic a p m, MonadIO m) => m () -setSynced = - asks chainState >>= \v -> - atomically . modifyTVar v $ \s -> s {mySynced = True} - --- | Remove a peer from the queue of peers to sync and unset the syncing peer if --- it is set to the provided value. -finishPeer :: (Eq p, MonadIO m, MonadChainLogic a p m) => p -> m () -finishPeer p = - asks chainState >>= \st -> - atomically . modifyTVar st $ \s -> - s - { newPeers = delete p (newPeers s) - , chainSyncing = - case chainSyncing s of - Just ChainSync { chainSyncPeer = p' - , chainTimestamp = _ - , chainHighest = _ - } - | p == p' -> Nothing - _ -> chainSyncing s - } - --- | Return syncing peer data. -chainSyncingPeer :: (MonadChainLogic a p m, MonadIO m) => m (Maybe (ChainSync p)) -chainSyncingPeer = chainSyncing <$> (readTVarIO =<< asks chainState) diff --git a/src/Network/Haskoin/Node/Manager.hs b/src/Network/Haskoin/Node/Manager.hs --- a/src/Network/Haskoin/Node/Manager.hs +++ b/src/Network/Haskoin/Node/Manager.hs @@ -27,21 +27,32 @@ import Control.Monad.Logger import Control.Monad.Reader import Control.Monad.Trans.Maybe +import Data.Bits +import qualified Data.ByteString as B +import Data.Default +import Data.List +import Data.Maybe +import Data.Serialize (Get, Put, Serialize) +import Data.Serialize as S +import Data.Set (Set) +import qualified Data.Set as Set import Data.String.Conversions +import Data.Text (Text) import Data.Time.Clock import Data.Time.Clock.POSIX -import Database.RocksDB (DB) -import Network.Haskoin.Block -import Network.Haskoin.Constants -import Network.Haskoin.Network +import Data.Word +import Database.RocksDB (DB) +import qualified Database.RocksDB as R +import Database.RocksDB.Query as R +import Haskoin import Network.Haskoin.Node.Common -import Network.Haskoin.Node.Manager.Logic import Network.Haskoin.Node.Peer -import Network.Socket (SockAddr (..)) +import Network.Socket (SockAddr (..)) import NQE import System.Random import UnliftIO import UnliftIO.Concurrent +import UnliftIO.Resource as U -- | Monad used by most functions in this module. type MonadManager m = (MonadLoggerIO m, MonadReader ManagerReader m) @@ -52,6 +63,7 @@ , mySupervisor :: !Supervisor , myMailbox :: !Manager , myBestBlock :: !(TVar BlockHeight) + , knownPeers :: !(TVar (Set SockAddr)) , onlinePeers :: !(TVar [OnlinePeer]) } @@ -65,6 +77,7 @@ manager cfg inbox = withSupervisor (Notify f) $ \sup -> do bb <- newTVarIO 0 + kp <- newTVarIO Set.empty ob <- newTVarIO [] let rd = ManagerReader @@ -72,9 +85,10 @@ , mySupervisor = sup , myMailbox = mgr , myBestBlock = bb + , knownPeers = kp , onlinePeers = ob } - go `runReaderT` rd + (go `finally` storePeersInDB) `runReaderT` rd where mgr = inboxToMailbox inbox discover = mgrConfDiscover cfg @@ -86,8 +100,7 @@ ManagerBestBlock b -> Just b _ -> Nothing $(logDebugS) "Manager" "Initialization complete" - withConnectLoop mgr $ \a -> do - link a + withConnectLoop mgr $ forever $ receive inbox >>= managerMessage f (a, mex) = ManagerPeerDied a mex `sendSTM` mgr @@ -100,19 +113,42 @@ getNetwork :: MonadManager m => m Network getNetwork = mgrConfNetwork <$> asks myConfig -populatePeerDB :: (MonadUnliftIO m, MonadManager m) => m () -populatePeerDB = do - ManagerConfig { mgrConfPeers = statics - , mgrConfDB = db - , mgrConfDiscover = discover - } <- asks myConfig - cfg_peers <- concat <$> mapM toSockAddr statics - forM_ cfg_peers $ newPeerDB db staticPeerScore - when discover $ do - net <- getNetwork - net_seeds <- concat <$> mapM toSockAddr (networkSeeds net) - forM_ net_seeds $ newPeerDB db netSeedScore +loadPeers :: (MonadUnliftIO m, MonadManager m) => m () +loadPeers = do + d <- mgrConfDiscover <$> asks myConfig + loadStaticPeers + when d $ loadPeersFromDB >> loadNetSeeds +loadStaticPeers :: (MonadUnliftIO m, MonadManager m) => m () +loadStaticPeers = do + xs <- mgrConfPeers <$> asks myConfig + mapM_ newPeer =<< concat <$> mapM toSockAddr xs + +loadPeersFromDB :: (MonadUnliftIO m, MonadManager m) => m () +loadPeersFromDB = do + db <- getManagerDB + xs <- matchingAsList db R.defaultReadOptions PeerAddressBase + let f (PeerAddress x, ()) = x + f _ = undefined + ys = map f xs + mapM_ newPeer ys + +storePeersInDB :: (MonadUnliftIO m, MonadManager m) => m () +storePeersInDB = do + db <- getManagerDB + os <- readTVarIO =<< asks onlinePeers + unless (null os) $ do + ks <- readTVarIO =<< asks knownPeers + let ps = map onlinePeerAddress os <> Set.toList ks + xs = map ((`insertOp` ()) . PeerAddress) ps + purgePeerDB db + writeBatch db xs + +loadNetSeeds :: (MonadUnliftIO m, MonadManager m) => m () +loadNetSeeds = do + net <- getNetwork + mapM_ newPeer =<< concat <$> mapM toSockAddr (networkSeeds net) + logConnectedPeers :: MonadManager m => m () logConnectedPeers = do m <- mgrConfMaxPeers <$> asks myConfig @@ -137,7 +173,7 @@ purgePeerDB db forwardMessage :: MonadManager m => Peer -> Message -> m () -forwardMessage p msg = managerEvent $ PeerMessage p msg +forwardMessage p = managerEvent . PeerMessage p managerEvent :: MonadManager m => PeerEvent -> m () managerEvent e = mgrConfEvents <$> asks myConfig >>= \l -> atomically $ l e @@ -176,15 +212,14 @@ managerMessage (ManagerPeerMessage p (MAddr (Addr nas))) = mgrConfDiscover <$> asks myConfig >>= \d -> when d $ do - db <- getManagerDB b <- asks onlinePeers s <- atomically $ peerString b p $(logInfoS) "Manager" $ "Received " <> cs (show (length nas)) <> " peers from " <> s let sas = map (naAddress . snd) nas - forM_ sas $ newPeerDB db netPeerScore + forM_ sas newPeer -managerMessage (ManagerPeerMessage p msg@(MPong (Pong n))) = do +managerMessage (ManagerPeerMessage p m@(MPong (Pong n))) = do now <- liftIO getCurrentTime b <- asks onlinePeers s <- atomically $ peerString b p @@ -192,7 +227,7 @@ Nothing -> do $(logDebugS) "Manager" $ "Forwarding pong " <> cs (show n) <> " from " <> s - forwardMessage p msg + forwardMessage p m Just d -> do let ms = fromRational . toRational $ d * 1000 :: Double $(logDebugS) "Manager" $ @@ -205,13 +240,13 @@ "Responding to ping " <> cs (show n) <> " from " <> s MPong (Pong n) `sendMessage` p -managerMessage (ManagerPeerMessage p msg) = do +managerMessage (ManagerPeerMessage p m) = do b <- asks onlinePeers s <- atomically $ peerString b p - let cmd = commandToString $ msgType msg + let cmd = commandToString $ msgType m $(logDebugS) "Manager" $ "Forwarding message " <> cs cmd <> " from peer " <> s - forwardMessage p msg + forwardMessage p m managerMessage (ManagerBestBlock h) = putBestBlock h @@ -287,8 +322,6 @@ managerEvent $ PeerDisconnected p d else log_not_connect s e atomically $ removePeer b p - db <- getManagerDB - demotePeerDB db (onlinePeerAddress o) logConnectedPeers where log_unknown Nothing = $(logErrorS) "Manager" "Disconnected unknown peer" @@ -315,21 +348,25 @@ managerEvent $ PeerConnected p a logConnectedPeers managerCheck p mgr - db <- getManagerDB - promotePeerDB db a Just OnlinePeer {onlinePeerConnected = False} -> $(logErrorS) "Manager" $ "Not announcing because not handshaken: " <> s Nothing -> $(logErrorS) "Manager" "Will not announce unknown peer" getNewPeer :: (MonadUnliftIO m, MonadManager m) => m (Maybe SockAddr) -getNewPeer = do - ManagerConfig {mgrConfDB = db} <- asks myConfig - exclude <- map onlinePeerAddress <$> getOnlinePeers - runMaybeT $ - MaybeT (getNewPeerDB db exclude) <|> - MaybeT (populatePeerDB >> getNewPeerDB db exclude) +getNewPeer = runMaybeT $ go <|> (lift loadPeers >> go) + where + go = do + b <- asks knownPeers + ks <- readTVarIO b + guard . not $ Set.null ks + let xs = Set.toList ks + a <- liftIO $ randomRIO (0, length xs - 1) + let p = xs !! a + atomically . modifyTVar b $ Set.delete p + return p + connectPeer :: (MonadUnliftIO m, MonadManager m) => SockAddr -> m () connectPeer sa = do $(logInfoS) "Manager" $ "Connecting to peer: " <> cs (show sa) @@ -372,9 +409,249 @@ threadDelay =<< liftIO (randomRIO (30 * 1000 * 1000, 90 * 1000 * 1000)) ManagerCheckPeer p `send` mgr -withConnectLoop :: MonadUnliftIO m => Manager -> (Async a -> m a) -> m a -withConnectLoop mgr = withAsync go +withConnectLoop :: MonadUnliftIO m => Manager -> m a -> m a +withConnectLoop mgr act = withAsync go (\a -> link a >> act) where go = forever $ do ManagerConnect `send` mgr threadDelay =<< liftIO (randomRIO (250 * 1000, 1000 * 1000)) + +-- | Database version. +versionPeerDB :: Word32 +versionPeerDB = 5 + +-- | Peer address key in database. +data PeerAddress + = PeerAddress { + getPeerAddress :: !SockAddr + -- ^ peer network address and port + } + | PeerAddressBase + deriving (Eq, Ord, Show) + +-- | Database version key. +data PeerDataVersionKey = PeerDataVersionKey + deriving (Eq, Ord, Show) + +instance Serialize PeerDataVersionKey where + get = do + guard . (== 0x82) =<< S.getWord8 + return PeerDataVersionKey + put PeerDataVersionKey = S.putWord8 0x82 + +instance Key PeerDataVersionKey +instance KeyValue PeerDataVersionKey Word32 + +instance Serialize PeerAddress where + get = do + guard . (== 0x81) =<< S.getWord8 + PeerAddress <$> decodeSockAddr + put PeerAddress {getPeerAddress = a} = do + S.putWord8 0x81 + encodeSockAddr a + put PeerAddressBase = S.putWord8 0x81 + +instance Key PeerAddress +instance KeyValue PeerAddress () + +-- | Add a peer. +newPeer :: (MonadIO m, MonadManager m) => SockAddr -> m () +newPeer sa = do + b <- asks knownPeers + o <- asks onlinePeers + i <- atomically $ findPeerAddress o sa + when (isNothing i) $ atomically . modifyTVar b $ Set.insert sa + +-- | Initialize peer database, purging it of conflicting records if version +-- doesn't match current one, or if peer discovery is disabled. +initPeerDB :: MonadUnliftIO m => DB -> Bool -> m () +initPeerDB db discover = do + ver <- retrieve db def PeerDataVersionKey + when (ver /= Just versionPeerDB || not discover) $ purgePeerDB db + R.insert db PeerDataVersionKey versionPeerDB + +-- | Purge peer records from database. +purgePeerDB :: MonadUnliftIO m => DB -> m () +purgePeerDB db = do + ops <- (++) <$> purge_byte 0x81 <*> purge_byte 0x83 + writeBatch db ops + where + purge_byte byte = + U.runResourceT . R.withIterator db def $ \it -> do + R.iterSeek it $ B.singleton byte + recurse_delete it byte + recurse_delete it byte = + R.iterKey it >>= \case + Just k + | B.head k == byte -> do + R.iterNext it + (R.Del k :) <$> recurse_delete it byte + _ -> return [] + +-- | Get static network seeds. +networkSeeds :: Network -> [HostPort] +networkSeeds net = map (, getDefaultPort net) (getSeeds net) + +-- | Serialize a network address/port. +encodeSockAddr :: SockAddr -> Put +encodeSockAddr (SockAddrInet6 p _ (a, b, c, d) _) = do + S.putWord32be a + S.putWord32be b + S.putWord32be c + S.putWord32be d + S.putWord16be (fromIntegral p) +encodeSockAddr (SockAddrInet p a) = do + S.putWord32be 0x00000000 + S.putWord32be 0x00000000 + S.putWord32be 0x0000ffff + S.putWord32host a + S.putWord16be (fromIntegral p) +encodeSockAddr x = error $ "Could not encode address: " <> show x + +-- | Deserialize a network address/port. +decodeSockAddr :: Get SockAddr +decodeSockAddr = do + a <- S.getWord32be + b <- S.getWord32be + c <- S.getWord32be + if a == 0x00000000 && b == 0x00000000 && c == 0x0000ffff + then do + d <- S.getWord32host + p <- S.getWord16be + return $ SockAddrInet (fromIntegral p) d + else do + d <- S.getWord32be + p <- S.getWord16be + return $ SockAddrInet6 (fromIntegral p) 0 (a, b, c, d) 0 + +-- | Report receiving a pong from a connected peer. Will store ping roundtrip +-- time in a window of latest eleven. Peers are returned by the manager in order +-- of median roundtrip time. +gotPong :: + TVar [OnlinePeer] + -> Word64 + -> UTCTime + -> Peer + -> STM (Maybe NominalDiffTime) +gotPong b nonce now p = + runMaybeT $ do + o <- MaybeT $ findPeer b p + (time, old_nonce) <- MaybeT . return $ onlinePeerPing o + guard $ nonce == old_nonce + let diff = now `diffUTCTime` time + lift $ + insertPeer + b + o + { onlinePeerPing = Nothing + , onlinePeerPings = take 11 $ diff : onlinePeerPings o + } + return diff + +-- | Return time of last ping sent to peer, if any. +lastPing :: TVar [OnlinePeer] -> Peer -> STM (Maybe UTCTime) +lastPing b p = + findPeer b p >>= \case + Just OnlinePeer {onlinePeerPing = Just (time, _)} -> return (Just time) + _ -> return Nothing + +-- | Set nonce and time of last ping sent to peer. +setPeerPing :: TVar [OnlinePeer] -> Word64 -> UTCTime -> Peer -> STM () +setPeerPing b nonce now p = + modifyPeer b p $ \o -> o {onlinePeerPing = Just (now, nonce)} + +-- | Set version for online peer. Will set the peer connected status to 'True' +-- if a verack message has already been registered for that peer. +setPeerVersion :: + TVar [OnlinePeer] + -> Peer + -> Version + -> STM (Either PeerException OnlinePeer) +setPeerVersion b p v = + runExceptT $ do + when (services v .&. nodeNetwork == 0) $ throwError NotNetworkPeer + ops <- lift $ readTVar b + when (any ((verNonce v ==) . onlinePeerNonce) ops) $ + throwError PeerIsMyself + lift (findPeer b p) >>= \case + Nothing -> throwError UnknownPeer + Just o -> do + let n = + o + { onlinePeerVersion = Just v + , onlinePeerConnected = onlinePeerVerAck o + } + lift $ insertPeer b n + return n + +-- | Register that a verack message was received from a peer. +setPeerVerAck :: TVar [OnlinePeer] -> Peer -> STM (Maybe OnlinePeer) +setPeerVerAck b p = + runMaybeT $ do + o <- MaybeT $ findPeer b p + let n = + o + { onlinePeerVerAck = True + , onlinePeerConnected = isJust (onlinePeerVersion o) + } + lift $ insertPeer b n + return n + +-- | Create 'OnlinePeer' data structure. +newOnlinePeer :: + TVar [OnlinePeer] + -> SockAddr + -- ^ peer address + -> Word64 + -- ^ nonce sent to peer + -> Peer + -- ^ peer mailbox + -> Async () + -- ^ peer asynchronous handle + -> STM OnlinePeer +newOnlinePeer b sa n p a = do + let op = + OnlinePeer + { onlinePeerAddress = sa + , onlinePeerVerAck = False + , onlinePeerConnected = False + , onlinePeerVersion = Nothing + , onlinePeerAsync = a + , onlinePeerMailbox = p + , onlinePeerNonce = n + , onlinePeerPings = [] + , onlinePeerPing = Nothing + } + insertPeer b op + return op + +-- | Get a human-readable string for the peer address. +peerString :: TVar [OnlinePeer] -> Peer -> STM Text +peerString b p = + maybe "[unknown]" (cs . show . onlinePeerAddress) <$> findPeer b p + +-- | Find a connected peer. +findPeer :: TVar [OnlinePeer] -> Peer -> STM (Maybe OnlinePeer) +findPeer b p = find ((== p) . onlinePeerMailbox) <$> readTVar b + +-- | Insert or replace a connected peer. +insertPeer :: TVar [OnlinePeer] -> OnlinePeer -> STM () +insertPeer b o = modifyTVar b $ \x -> sort . nub $ o : x + +-- | Modify an online peer. +modifyPeer :: TVar [OnlinePeer] -> Peer -> (OnlinePeer -> OnlinePeer) -> STM () +modifyPeer b p f = + findPeer b p >>= \case + Nothing -> return () + Just o -> insertPeer b $ f o + +-- | Remove an online peer. +removePeer :: TVar [OnlinePeer] -> Peer -> STM () +removePeer b p = modifyTVar b $ \x -> filter ((/= p) . onlinePeerMailbox) x + +-- | Find online peer by asynchronous handle. +findPeerAsync :: TVar [OnlinePeer] -> Async () -> STM (Maybe OnlinePeer) +findPeerAsync b a = find ((== a) . onlinePeerAsync) <$> readTVar b + +findPeerAddress :: TVar [OnlinePeer] -> SockAddr -> STM (Maybe OnlinePeer) +findPeerAddress b a = find ((== a) . onlinePeerAddress) <$> readTVar b diff --git a/src/Network/Haskoin/Node/Manager/Logic.hs b/src/Network/Haskoin/Node/Manager/Logic.hs deleted file mode 100644 --- a/src/Network/Haskoin/Node/Manager/Logic.hs +++ /dev/null @@ -1,406 +0,0 @@ -{-# LANGUAGE ConstraintKinds #-} -{-# LANGUAGE FlexibleContexts #-} -{-# LANGUAGE FlexibleInstances #-} -{-# LANGUAGE LambdaCase #-} -{-# LANGUAGE MultiParamTypeClasses #-} -{-# LANGUAGE OverloadedStrings #-} -{-# LANGUAGE TupleSections #-} -{-| -Module : Network.Haskoin.Manager.Logic -Copyright : No rights reserved -License : UNLICENSE -Maintainer : xenog@protonmail.com -Stability : experimental -Portability : POSIX - -Logic and data for peer manager. --} -module Network.Haskoin.Node.Manager.Logic where - -import Conduit -import Control.Monad -import Control.Monad.Except -import Control.Monad.Trans.Maybe -import Data.Bits -import qualified Data.ByteString as B -import Data.Default -import Data.List -import Data.Maybe -import Data.Serialize (Get, Put, Serialize) -import Data.Serialize as S -import Data.String.Conversions -import Data.Text (Text) -import Data.Time.Clock -import Data.Word -import Database.RocksDB (DB) -import qualified Database.RocksDB as R -import Database.RocksDB.Query as R -import Haskoin -import Network.Haskoin.Node.Common -import Network.Socket (SockAddr (..)) -import UnliftIO -import UnliftIO.Resource as U - --- | Database version. -versionPeerDB :: Word32 -versionPeerDB = 4 - --- | Peer score. Lower is better. -type Score = Word8 - --- | Peer address key in database. -data PeerAddress - = PeerAddress { - getPeerAddress :: !SockAddr - -- ^ peer network address and port - } - | PeerAddressBase - deriving (Eq, Ord, Show) - --- | Peer score key in database. -data PeerScore - = PeerScore !Score !SockAddr - | PeerScoreBase - deriving (Eq, Ord, Show) - -instance Serialize PeerScore where - put (PeerScore s sa) = do - putWord8 0x83 - S.put s - encodeSockAddr sa - put PeerScoreBase = S.putWord8 0x83 - get = do - guard . (== 0x83) =<< S.getWord8 - PeerScore <$> S.get <*> decodeSockAddr - -instance Key PeerScore -instance KeyValue PeerScore () - --- | Database version key. -data PeerDataVersionKey = PeerDataVersionKey - deriving (Eq, Ord, Show) - -instance Serialize PeerDataVersionKey where - get = do - guard . (== 0x82) =<< S.getWord8 - return PeerDataVersionKey - put PeerDataVersionKey = S.putWord8 0x82 - -instance Key PeerDataVersionKey -instance KeyValue PeerDataVersionKey Word32 - -instance Serialize PeerAddress where - get = do - guard . (== 0x81) =<< S.getWord8 - PeerAddress <$> decodeSockAddr - put PeerAddress {getPeerAddress = a} = do - S.putWord8 0x81 - encodeSockAddr a - put PeerAddressBase = S.putWord8 0x81 - --- | Peer data. -data PeerData = PeerData !Score !Bool - deriving (Eq, Show, Ord) - -instance Serialize PeerData where - put (PeerData s p) = S.put s >> S.put p - get = PeerData <$> S.get <*> S.get - -instance Key PeerAddress -instance KeyValue PeerAddress PeerData - --- | Update peer score and pass in database. -updatePeerDB :: MonadIO m => DB -> SockAddr -> Score -> Bool -> m () -updatePeerDB db sa score pass = - retrieve db def (PeerAddress sa) >>= \case - Nothing -> - writeBatch - db - [ insertOp (PeerAddress sa) (PeerData score pass) - , insertOp (PeerScore score sa) () - ] - Just (PeerData score' _) -> - writeBatch - db - [ deleteOp (PeerScore score' sa) - , insertOp (PeerAddress sa) (PeerData score pass) - , insertOp (PeerScore score sa) () - ] - --- | Add a peer to database if it is not already there. -newPeerDB :: MonadIO m => DB -> Score -> SockAddr -> m () -newPeerDB db score sa = - retrieve db def (PeerAddress sa) >>= \case - Just PeerData {} -> return () - Nothing -> - writeBatch - db - [ insertOp (PeerAddress sa) (PeerData score False) - , insertOp (PeerScore score sa) () - ] - --- | Initialize peer database, purging it of conflicting records if version --- doesn't match current one. -initPeerDB :: MonadUnliftIO m => DB -> Bool -> m () -initPeerDB db discover = do - ver <- retrieve db def PeerDataVersionKey - when (ver /= Just versionPeerDB || not discover) $ purgePeerDB db - R.insert db PeerDataVersionKey versionPeerDB - --- | Purge conflicting records from database. -purgePeerDB :: MonadUnliftIO m => DB -> m () -purgePeerDB db = purge_byte 0x81 >> purge_byte 0x83 - where - purge_byte byte = - U.runResourceT . R.withIterator db def $ \it -> do - R.iterSeek it $ B.singleton byte - recurse_delete it byte - recurse_delete it byte = - R.iterKey it >>= \case - Nothing -> return () - Just k - | B.head k == byte -> do - R.delete db def k - R.iterNext it - recurse_delete it byte - | otherwise -> return () - --- | Get static network seeds. -networkSeeds :: Network -> [HostPort] -networkSeeds net = map (, getDefaultPort net) (getSeeds net) - --- | Get default score for statically-defined peers. -staticPeerScore :: Score -staticPeerScore = maxBound `div` 4 - --- | Get default score for network seeds. -netSeedScore :: Score -netSeedScore = maxBound `div` 2 - --- | Get default score for discovered peers. -netPeerScore :: Score -netPeerScore = maxBound `div` 3 * 4 - --- | Serialize a network address/port. -encodeSockAddr :: SockAddr -> Put -encodeSockAddr (SockAddrInet6 p _ (a, b, c, d) _) = do - S.putWord32be a - S.putWord32be b - S.putWord32be c - S.putWord32be d - S.putWord16be (fromIntegral p) -encodeSockAddr (SockAddrInet p a) = do - S.putWord32be 0x00000000 - S.putWord32be 0x00000000 - S.putWord32be 0x0000ffff - S.putWord32host a - S.putWord16be (fromIntegral p) -encodeSockAddr x = error $ "Could not encode address: " <> show x - --- | Deserialize a network address/port. -decodeSockAddr :: Get SockAddr -decodeSockAddr = do - a <- S.getWord32be - b <- S.getWord32be - c <- S.getWord32be - if a == 0x00000000 && b == 0x00000000 && c == 0x0000ffff - then do - d <- S.getWord32host - p <- S.getWord16be - return $ SockAddrInet (fromIntegral p) d - else do - d <- S.getWord32be - p <- S.getWord16be - return $ SockAddrInet6 (fromIntegral p) 0 (a, b, c, d) 0 - --- | Get database entry for provided peer address. -getPeerDB :: MonadIO m => DB -> SockAddr -> m (Maybe PeerData) -getPeerDB db sa = retrieve db def (PeerAddress sa) - --- | Promote a peer by improving its score (decreasing by one). -promotePeerDB :: MonadIO m => DB -> SockAddr -> m () -promotePeerDB db sa = - getPeerDB db sa >>= \case - Nothing -> return () - Just (PeerData score pass) -> - updatePeerDB - db - sa - (if score == 0 - then score - else score - 1) - pass - --- | Demote a peer increasing its score by one. -demotePeerDB :: MonadIO m => DB -> SockAddr -> m () -demotePeerDB db sa = - getPeerDB db sa >>= \case - Nothing -> return () - Just (PeerData score pass) -> - updatePeerDB - db - sa - (if score == maxBound - then score - else score + 1) - pass - --- | Get a peer from database. Pass a list of addresses to exclude. -getNewPeerDB :: MonadUnliftIO m => DB -> [SockAddr] -> m (Maybe SockAddr) -getNewPeerDB db exclude = go >>= maybe (reset_pass >> go) (return . Just) - where - go = - U.runResourceT . runConduit $ - matching db def PeerScoreBase .| filterMC filter_pass .| - mapC get_address .| - filterC (`notElem` exclude) .| - headC - reset_pass = - U.runResourceT . runConduit $ - matching db def PeerAddressBase .| mapM_C reset_peer_pass - reset_peer_pass (PeerAddress addr, PeerData score _) = - updatePeerDB db addr score False - reset_peer_pass _ = return () - filter_pass (PeerScore _ addr, ()) = - retrieve db def (PeerAddress addr) >>= \case - Just (PeerData score pass) - | not pass -> do - updatePeerDB db addr score True - return True - _ -> return False - filter_pass _ = return False - get_address (PeerScore _ addr, ()) = addr - get_address _ = error "Something is wrong with peer database" - --- | Report receiving a pong from a connected peer. Will store ping roundtrip --- time in a window of latest eleven. Peers are returned by the manager in order --- of median roundtrip time. -gotPong :: - TVar [OnlinePeer] - -> Word64 - -> UTCTime - -> Peer - -> STM (Maybe NominalDiffTime) -gotPong b nonce now p = - runMaybeT $ do - o <- MaybeT $ findPeer b p - (time, old_nonce) <- MaybeT . return $ onlinePeerPing o - guard $ nonce == old_nonce - let diff = now `diffUTCTime` time - lift $ - insertPeer - b - o - { onlinePeerPing = Nothing - , onlinePeerPings = take 11 $ diff : onlinePeerPings o - } - return diff - --- | Return time of last ping sent to peer, if any. -lastPing :: TVar [OnlinePeer] -> Peer -> STM (Maybe UTCTime) -lastPing b p = - findPeer b p >>= \case - Just OnlinePeer {onlinePeerPing = Just (time, _)} -> return (Just time) - _ -> return Nothing - --- | Set nonce and time of last ping sent to peer. -setPeerPing :: TVar [OnlinePeer] -> Word64 -> UTCTime -> Peer -> STM () -setPeerPing b nonce now p = - modifyPeer b p $ \o -> o {onlinePeerPing = Just (now, nonce)} - --- | Set version for online peer. Will set the peer connected status to 'True' --- if a verack message has already been registered for that peer. -setPeerVersion :: - TVar [OnlinePeer] - -> Peer - -> Version - -> STM (Either PeerException OnlinePeer) -setPeerVersion b p v = - runExceptT $ do - when (services v .&. nodeNetwork == 0) $ throwError NotNetworkPeer - ops <- lift $ readTVar b - when (any ((verNonce v ==) . onlinePeerNonce) ops) $ - throwError PeerIsMyself - lift (findPeer b p) >>= \case - Nothing -> throwError UnknownPeer - Just o -> do - let n = - o - { onlinePeerVersion = Just v - , onlinePeerConnected = onlinePeerVerAck o - } - lift $ insertPeer b n - return n - --- | Register that a verack message was received from a peer. -setPeerVerAck :: TVar [OnlinePeer] -> Peer -> STM (Maybe OnlinePeer) -setPeerVerAck b p = - runMaybeT $ do - o <- MaybeT $ findPeer b p - let n = - o - { onlinePeerVerAck = True - , onlinePeerConnected = isJust (onlinePeerVersion o) - } - lift $ insertPeer b n - return n - --- | Create 'OnlinePeer' data structure. -newOnlinePeer :: - TVar [OnlinePeer] - -> SockAddr - -- ^ peer address - -> Word64 - -- ^ nonce sent to peer - -> Peer - -- ^ peer mailbox - -> Async () - -- ^ peer asynchronous handle - -> STM OnlinePeer -newOnlinePeer b sa nonce peer peer_async = do - let op = - OnlinePeer - { onlinePeerAddress = sa - , onlinePeerVerAck = False - , onlinePeerConnected = False - , onlinePeerVersion = Nothing - , onlinePeerAsync = peer_async - , onlinePeerMailbox = peer - , onlinePeerNonce = nonce - , onlinePeerPings = [] - , onlinePeerPing = Nothing - } - insertPeer b op - return op - --- | Get a human-readable string for the peer address. -peerString :: TVar [OnlinePeer] -> Peer -> STM Text -peerString b p = - maybe "[unknown]" (cs . show . onlinePeerAddress) <$> findPeer b p - --- | Find a connected peer. -findPeer :: TVar [OnlinePeer] -> Peer -> STM (Maybe OnlinePeer) -findPeer b p = find ((== p) . onlinePeerMailbox) <$> readTVar b - --- | Insert or replace a connected peer. -insertPeer :: TVar [OnlinePeer] -> OnlinePeer -> STM () -insertPeer b o = modifyTVar b $ \x -> sort . nub $ o : x - --- | Modify an online peer. -modifyPeer :: TVar [OnlinePeer] -> Peer -> (OnlinePeer -> OnlinePeer) -> STM () -modifyPeer b p f = - findPeer b p >>= \case - Nothing -> return () - Just o -> insertPeer b $ f o - --- | Remove an online peer. -removePeer :: TVar [OnlinePeer] -> Peer -> STM () -removePeer b p = modifyTVar b $ \x -> filter ((/= p) . onlinePeerMailbox) x - --- | Find online peer by asynchronous handle. -findPeerAsync :: TVar [OnlinePeer] -> Async () -> STM (Maybe OnlinePeer) -findPeerAsync b a = find ((== a) . onlinePeerAsync) <$> readTVar b - --- | Remove online peer by asynchronous handle. -removePeerAsync :: TVar [OnlinePeer] -> Async () -> STM () -removePeerAsync b a = modifyTVar b $ \x -> filter ((/= a) . onlinePeerAsync) x