haskoin-node 0.9.5 → 0.9.6
raw patch · 6 files changed
+609/−770 lines, 6 filesdep +containers
Dependencies added: containers
Files
- CHANGELOG.md +6/−0
- haskoin-node.cabal +19/−7
- src/Network/Haskoin/Node/Chain.hs +266/−4
- src/Network/Haskoin/Node/Chain/Logic.hs +0/−312
- src/Network/Haskoin/Node/Manager.hs +318/−41
- src/Network/Haskoin/Node/Manager/Logic.hs +0/−406
CHANGELOG.md view
@@ -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.
haskoin-node.cabal view
@@ -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
src/Network/Haskoin/Node/Chain.hs view
@@ -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)
− src/Network/Haskoin/Node/Chain/Logic.hs
@@ -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)
src/Network/Haskoin/Node/Manager.hs view
@@ -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
− src/Network/Haskoin/Node/Manager/Logic.hs
@@ -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