haskoin-node 0.6.1 → 0.7.0
raw patch · 11 files changed
+1803/−1605 lines, 11 filesdep +HUnitdep +data-defaultdep +hashable
Dependencies added: HUnit, data-default, hashable, safe, text, transformers, unordered-containers
Files
- CHANGELOG.md +22/−0
- haskoin-node.cabal +16/−6
- src/Haskoin/Node.hs +80/−38
- src/Network/Haskoin/Node/Chain.hs +140/−227
- src/Network/Haskoin/Node/Chain/Logic.hs +284/−0
- src/Network/Haskoin/Node/Common.hs +315/−274
- src/Network/Haskoin/Node/Manager.hs +317/−531
- src/Network/Haskoin/Node/Manager/Logic.hs +406/−0
- src/Network/Haskoin/Node/Peer.hs +75/−310
- test/Haskoin/NodeSpec.hs +147/−0
- test/Spec.hs +1/−219
CHANGELOG.md view
@@ -4,6 +4,28 @@ 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.7.0+### Added+- Versioning for chain and peer database.+- Automatic purging of chain and peer database when version changes.+- Add extra timers.+- Add publishers to every peer.++### Changed+- Full reimplementation of node API.+- Simplify peer selection and management.+- Merge manager and peer events.+- Rename configuration variables for node.+- Separate logic from actors for peer manager and chain.++### Removed+- Remove irrelevant fields from peer information.+- Remove unreliable peer block head tracking.+- Remove dependency on deprecated binary conduits.+- Remove Bloom filter support from manager.+- Remove unreliable peer request tracking code.+- Remove separate manager events.+ ## 0.6.1 ### Changed - Fix bug where peer height did not update in certain cases.
haskoin-node.cabal view
@@ -2,10 +2,10 @@ -- -- see: https://github.com/sol/hpack ----- hash: 442a53cc8f60132054fb6ad206821818652bc1a8f6f0f0661f2c0609db575380+-- hash: 0e33d7c5993dd152a270e0957b4aec8b09e67fb8f760b01945022ae62df8ac7b name: haskoin-node-version: 0.6.1+version: 0.7.0 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@@ -28,11 +28,13 @@ library exposed-modules: Haskoin.Node- other-modules: 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 hs-source-dirs: src@@ -42,6 +44,8 @@ , cereal , conduit , conduit-extra+ , data-default+ , hashable , haskoin-core , monad-logger , mtl@@ -52,20 +56,24 @@ , rocksdb-haskell , rocksdb-query , string-conversions+ , text , time+ , transformers , unliftio+ , unordered-containers default-language: Haskell2010 -test-suite haskoin-node-test+test-suite spec type: exitcode-stdio-1.0 main-is: Spec.hs other-modules:+ Haskoin.NodeSpec Paths_haskoin_node hs-source-dirs: test- ghc-options: -threaded -rtsopts -with-rtsopts=-N build-depends:- base >=4.7 && <5+ HUnit+ , base >=4.7 && <5 , bytestring , cereal , haskoin-core@@ -77,5 +85,7 @@ , nqe , random , rocksdb-haskell+ , safe , unliftio default-language: Haskell2010+ build-tool-depends: hspec-discover:hspec-discover
src/Haskoin/Node.hs view
@@ -1,25 +1,38 @@ {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE GADTs #-} {-# LANGUAGE MultiParamTypeClasses #-}+{-|+Module : Haskoin.Node+Copyright : No rights reserved+License : UNLICENSE+Maintainer : xenog@protonmail.com+Stability : experimental+Portability : POSIX++Integrates peers, manager, and block header synchronization processes.++Messages from peers that aren't consumed by the peer manager or chain are+forwarded to the listen action provided in the node configuration.+-} module Haskoin.Node- ( Host, Port, HostPort- , Peer, Chain, Manager+ ( Host+ , Port+ , HostPort+ , Peer+ , Chain+ , Manager , OnlinePeer(..) , NodeConfig(..) , NodeEvent(..)- , ManagerEvent(..) , ChainEvent(..) , PeerEvent(..) , PeerException(..) , withNode- , managerGetPeerVersion- , managerGetPeerBest+ , node , managerGetPeers , managerGetPeer , managerKill- , setManagerFilter , sendMessage- , getMerkleBlocks , peerGetBlocks , peerGetTxs , chainGetBlock@@ -33,46 +46,75 @@ ) where import Control.Monad.Logger+import Haskoin import Network.Haskoin.Node.Chain import Network.Haskoin.Node.Common import Network.Haskoin.Node.Manager import NQE import UnliftIO +-- | Launch a node in the background. Pass a 'Manager' and 'Chain' to a+-- function. Node will stop once the function ends. withNode ::- ( MonadLoggerIO m- , MonadUnliftIO m- )+ (MonadLoggerIO m, MonadUnliftIO m) => NodeConfig -> ((Manager, Chain) -> m a) -> m a withNode cfg f = do- c <- newInbox =<< newTQueueIO- m <- newInbox =<< newTQueueIO- withAsync (chain (chain_conf c m)) $ \ch ->- withAsync (manager (manager_conf c m)) $ \mgr -> do- link ch- link mgr- f (m, c)+ mgr_inbox <- newInbox+ ch_inbox <- newInbox+ withAsync (node cfg mgr_inbox ch_inbox) $ \a -> do+ link a+ f (inboxToMailbox mgr_inbox, inboxToMailbox ch_inbox)++-- | Launch node process in the foreground.+node ::+ ( MonadLoggerIO m+ , MonadUnliftIO m+ )+ => NodeConfig+ -> Inbox ManagerMessage+ -> Inbox ChainMessage+ -> m ()+node cfg mgr_inbox ch_inbox = do+ let mgr_config =+ ManagerConfig+ { mgrConfMaxPeers = nodeConfMaxPeers cfg+ , mgrConfDB = nodeConfDB cfg+ , mgrConfPeers = nodeConfPeers cfg+ , mgrConfDiscover = nodeConfDiscover cfg+ , mgrConfNetAddr = nodeConfNetAddr cfg+ , mgrConfNetwork = nodeConfNet cfg+ , mgrConfEvents = mgr_events+ , mgrConfTimeout = nodeConfTimeout cfg+ }+ withAsync (manager mgr_config mgr_inbox) $ \mgr_async -> do+ link mgr_async+ let chain_config =+ ChainConfig+ { chainConfDB = nodeConfDB cfg+ , chainConfManager = mgr+ , chainConfNetwork = nodeConfNet cfg+ , chainConfEvents = chain_events+ , chainConfTimeout = nodeConfTimeout cfg+ }+ chain chain_config ch_inbox where- chain_conf c m =- ChainConfig- { chainConfDB = database cfg- , chainConfListener = nodeEvents cfg . ChainEvent- , chainConfManager = m- , chainConfChain = c- , chainConfNetwork = nodeNet cfg- }- manager_conf c m =- ManagerConfig- { mgrConfMaxPeers = maxPeers cfg- , mgrConfDB = database cfg- , mgrConfDiscover = discover cfg- , mgrConfMgrListener = nodeEvents cfg . ManagerEvent- , mgrConfPeerListener = nodeEvents cfg . PeerEvent- , mgrConfNetAddr = netAddress cfg- , mgrConfPeers = initPeers cfg- , mgrConfManager = m- , mgrConfChain = c- , mgrConfNetwork = nodeNet cfg- }+ ch = inboxToMailbox ch_inbox+ mgr = inboxToMailbox mgr_inbox+ mgr_events event =+ case event of+ PeerMessage p (MHeaders (Headers hcs)) ->+ ChainHeaders p (map fst hcs) `sendSTM` ch+ PeerConnected p a -> do+ ChainPeerConnected p a `sendSTM` ch+ Event (PeerEvent event) `sendSTM` nodeConfEvents cfg+ PeerDisconnected p a -> do+ ChainPeerDisconnected p a `sendSTM` ch+ Event (PeerEvent event) `sendSTM` nodeConfEvents cfg+ _ -> Event (PeerEvent event) `sendSTM` nodeConfEvents cfg+ chain_events event = do+ Event (ChainEvent event) `sendSTM` nodeConfEvents cfg+ case event of+ ChainBestBlock b -> ManagerBestBlock (nodeHeight b) `sendSTM` mgr+ _ -> return ()
src/Network/Haskoin/Node/Chain.hs view
@@ -5,269 +5,182 @@ {-# LANGUAGE LambdaCase #-} {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE OverloadedStrings #-}-{-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE TemplateHaskell #-} {-# LANGUAGE UndecidableInstances #-}-{-# OPTIONS_GHC -fno-warn-orphans #-}+{-|+Module : Network.Haskoin.Node.Chain+Copyright : No rights reserved+License : UNLICENSE+Maintainer : xenog@protonmail.com+Stability : experimental+Portability : POSIX++Block chain headers synchronizing process.+-} module Network.Haskoin.Node.Chain-( chain-) where+ ( chain+ ) where import Control.Monad import Control.Monad.Logger import Control.Monad.Reader-import qualified Data.ByteString as BS-import Data.Either-import Data.List (delete, nub)-import Data.Maybe-import Data.Serialize-import Data.String import Data.String.Conversions-import Database.RocksDB (DB)-import Database.RocksDB.Query as R+import Data.Text (Text)+import Data.Time.Clock import Network.Haskoin.Block import Network.Haskoin.Network+import Network.Haskoin.Node.Chain.Logic import Network.Haskoin.Node.Common import NQE+import System.Random import UnliftIO+import UnliftIO.Concurrent type MonadChain m- = ( BlockHeaders m- , MonadLoggerIO m- , MonadReader ChainReader m)--data ChainState = ChainState- { syncingPeer :: !(Maybe Peer)- , newPeers :: ![Peer]- , mySynced :: !Bool- }--data ChainReader = ChainReader- { headerDB :: !DB- , myConfig :: !ChainConfig- , chainState :: !(TVar ChainState)- }--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--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--instance (Monad m, MonadLoggerIO m, MonadReader ChainReader m) =>- BlockHeaders m where- addBlockHeader bn = do- db <- asks headerDB- insert db (BlockHeaderKey (headerHash (nodeHeader bn))) bn- getBlockHeader bh = do- db <- asks headerDB- retrieve db Nothing (BlockHeaderKey bh)- getBestBlockHeader = do- db <- asks headerDB- retrieve db Nothing BestBlockKey >>= \case- Nothing -> error "Could not get best block from database"- Just b -> return b- setBestBlockHeader bn = do- db <- asks headerDB- insert db BestBlockKey bn- addBlockHeaders bns = do- db <- asks headerDB- writeBatch db (map f bns)- where- f bn = insertOp (BlockHeaderKey (headerHash (nodeHeader bn))) bn+ = (MonadLoggerIO m, MonadChainLogic ChainConfig Peer m) +-- | Launch process to synchronize block headers in current thread. chain ::- ( MonadUnliftIO m- , MonadLoggerIO m- )+ (MonadUnliftIO m, MonadLoggerIO m) => ChainConfig+ -> Inbox ChainMessage -> m ()-chain cfg = do+chain cfg inbox = do st <- newTVarIO- ChainState {syncingPeer = Nothing, mySynced = False, newPeers = []}- let rd =- ChainReader- {myConfig = cfg, headerDB = chainConfDB cfg, chainState = st}- run `runReaderT` rd+ ChainState+ { syncingPeer = Nothing+ , mySynced = False+ , newPeers = []+ }+ let rd = ChainReader {myReader = cfg, myChainDB = db, chainState = st}+ withSyncLoop ch $ run `runReaderT` rd where net = chainConfNetwork cfg+ db = chainConfDB cfg+ ch = inboxToMailbox inbox run = do- db <- asks headerDB- m :: Maybe BlockNode <- retrieve db Nothing BestBlockKey- when (isNothing m) $ do- addBlockHeader (genesisNode net)- insert db BestBlockKey (genesisNode net)- forever $ do- msg <- receive $ chainConfChain cfg- processChainMessage msg+ $(logDebugS) "Chain" "Initializing..."+ initChainDB net+ getBestBlockHeader >>= chainEvent . ChainBestBlock+ $(logInfoS) "Chain" "Initialization complete"+ forever $ receive inbox >>= chainMessage -processChainMessage :: MonadChain m => ChainMessage -> m ()-processChainMessage (ChainNewHeaders p hcs) = do- stb <- asks chainState- st <- readTVarIO stb- net <- chainConfNetwork <$> asks myConfig- let spM = syncingPeer st- t <- computeTime- bb <- getBestBlockHeader- bhsE <- connectBlocks net t (map fst hcs)- case bhsE of- Right bhs -> conn bb bhs spM- Left e -> do- $(logWarnS) "Chain" $ "Could not connect headers: " <> cs e- case spM of- Nothing -> do- bb' <- getBestBlockHeader- atomically . modifyTVar stb $ \s ->- s {newPeers = nub $ p : newPeers s}- syncHeaders bb' p- Just sp- | sp == p -> do- pstr <- peerString p- $(logErrorS) "Chain" $- "Syncing peer " <> pstr <> " sent bad headers"- mgr <- chainConfManager <$> asks myConfig- managerKill PeerSentBadHeaders p mgr- atomically . modifyTVar stb $ \s ->- s {syncingPeer = Nothing}- processSyncQueue- | otherwise ->- atomically . modifyTVar stb $ \s ->- s {newPeers = nub $ p : newPeers s}- where- synced = do- st <- asks chainState- atomically . modifyTVar st $ \s -> s {syncingPeer = Nothing}- MSendHeaders `sendMessage` p- processSyncQueue- upeer bb = do- mgr <- chainConfManager <$> asks myConfig- managerSetPeerBest p bb mgr- conn bb bhs spM = do- bb' <- getBestBlockHeader- when (bb /= bb') $ do+chainEvent :: MonadChain m => ChainEvent -> m ()+chainEvent e = do+ l <- chainConfEvents <$> asks myReader+ case e of+ ChainBestBlock b -> $(logInfoS) "Chain" $- "Best header at height " <> cs (show (nodeHeight bb'))- mgr <- chainConfManager <$> asks myConfig- managerSetBest bb' mgr- l <- chainConfListener <$> asks myConfig- atomically . l $ ChainNewBest bb'- case length hcs of- 0 -> do- upeer bb- synced- 2000 ->- case spM of- Just sp- | sp == p -> do- upeer $ head bhs- syncHeaders (head bhs) p- _ -> do- st <- asks chainState- atomically . modifyTVar st $ \s ->- s {newPeers = nub $ p : newPeers s}- _ -> do- upeer $ head bhs- synced+ "Best block header at height " <> cs (show (nodeHeight b))+ ChainSynced b ->+ $(logInfoS) "Chain" $+ "Headers now synced at height " <> cs (show (nodeHeight b))+ atomically $ l e -processChainMessage (ChainNewPeer p) = do- st <- asks chainState- sp <-- atomically $ do- modifyTVar st $ \s -> s {newPeers = nub $ p : newPeers s}- syncingPeer <$> readTVar st- case sp of- Nothing -> processSyncQueue- Just _ -> return ()+processHeaders ::+ MonadChain m => Peer -> [BlockHeader] -> m ()+processHeaders p hs = do+ s <- peerString p+ net <- chainConfNetwork <$> asks myReader+ mgr <- chainConfManager <$> asks myReader+ $(logDebugS) "Chain" $+ "Importing " <> cs (show (length hs)) <> " headers from peer " <> s+ importHeaders net hs >>= \case+ Left e -> do+ $(logErrorS) "Chain" $+ "Could not connect headers sent by peer " <> s <> ": " <>+ cs (show e)+ managerKill e p mgr+ Right done -> do+ setLastReceived+ best <- getBestBlockHeader+ chainEvent $ ChainBestBlock best+ if done+ then do+ $(logDebugS) "Chain" $+ "Finished importing headers from peer: " <> s+ MSendHeaders `sendMessage` p+ finishPeer p+ syncNewPeer+ syncNotif+ else syncPeer p --- Getting a new block should trigger an action equivalent to getting a new peer-processChainMessage (ChainNewBlocks p _) = processChainMessage (ChainNewPeer p)+syncNewPeer :: MonadChain m => m ()+syncNewPeer = do+ $(logDebugS) "Chain" "Attempting to sync against a new peer"+ getSyncingPeer >>= \case+ Nothing -> do+ $(logDebugS) "Chain" "Getting next peer to sync from"+ nextPeer >>= \case+ Nothing ->+ $(logInfoS) "Chain" "Finished syncing against all peers"+ Just p -> syncPeer p+ Just p -> do+ s <- peerString p+ $(logDebugS) "Chain" $ "Already syncing against peer " <> s -processChainMessage (ChainRemovePeer p) = do- st <- asks chainState- sp <-- atomically $ do- modifyTVar st $ \s -> s {newPeers = delete p (newPeers s)}- syncingPeer <$> readTVar st- case sp of- Just p' ->- when (p == p') $ do- atomically . modifyTVar st $ \s ->- s {syncingPeer = Nothing}- processSyncQueue- Nothing -> return ()+syncNotif :: MonadChain m => m ()+syncNotif =+ notifySynced >>= \x ->+ when x $ getBestBlockHeader >>= chainEvent . ChainSynced -processChainMessage (ChainGetBest reply) =- getBestBlockHeader >>= atomically . reply+syncPeer :: MonadChain m => Peer -> m ()+syncPeer p = do+ s <- peerString p+ $(logInfoS) "Chain" $ "Syncing against peer " <> s+ bb <- getBestBlockHeader+ gh <- syncHeaders bb p+ MGetHeaders gh `sendMessage` p -processChainMessage (ChainGetAncestor h n reply) =+chainMessage :: MonadChain m => ChainMessage -> m ()+chainMessage (ChainGetBest reply) =+ getBestBlockHeader >>= atomically . reply+chainMessage (ChainHeaders p hs) = do+ s <- peerString p+ $(logDebugS) "Chain" $+ "Processing " <> cs (show (length hs)) <> " headers from peer " <> s+ processHeaders p hs+chainMessage (ChainPeerConnected p a) = do+ $(logDebugS) "Chain" $ "Adding new peer to sync queue: " <> cs (show a)+ addPeer p+ syncNewPeer+chainMessage (ChainPeerDisconnected p a) = do+ $(logWarnS) "Chain" $ "Removing a peer from sync queue: " <> cs (show a)+ finishPeer p+ syncNewPeer+chainMessage (ChainGetAncestor h n reply) = getAncestor h n >>= atomically . reply--processChainMessage (ChainGetSplit r l reply) =+chainMessage (ChainGetSplit r l reply) = splitPoint r l >>= atomically . reply--processChainMessage (ChainGetBlock h reply) =+chainMessage (ChainGetBlock h reply) = getBlockHeader h >>= atomically . reply--processChainMessage (ChainSendHeaders _) = return ()--processChainMessage (ChainIsSynced reply) = do- st <- asks chainState- s <- mySynced <$> readTVarIO st- atomically (reply s)+chainMessage (ChainIsSynced reply) =+ isSynced >>= atomically . reply+chainMessage ChainPing = do+ ChainConfig {chainConfManager = mgr, chainConfTimeout = to} <- asks myReader+ now <- liftIO getCurrentTime+ lastMessage >>= \case+ Nothing -> return ()+ Just (p, t)+ | diffUTCTime now t > fromIntegral to -> do+ s <- peerString p+ $(logErrorS) "Chain" $ "Syncing peer timed out: " <> s+ managerKill PeerTimeout p mgr+ | otherwise -> return () -processSyncQueue :: MonadChain m => m ()-processSyncQueue = do- s <- asks chainState >>= readTVarIO- when (isNothing (syncingPeer s)) $ getBestBlockHeader >>= go s+withSyncLoop :: (MonadUnliftIO m, MonadLoggerIO m) => Chain -> m a -> m a+withSyncLoop ch f = withAsync go $ \a -> link a >> f where- go s bb =- case newPeers s of- [] ->- unless (mySynced s) $ do- l <- chainConfListener <$> asks myConfig- st <- asks chainState- atomically $ do- l (ChainSynced bb)- writeTVar st s {mySynced = True}- p:_ -> syncHeaders bb p--syncHeaders :: MonadChain m => BlockNode -> Peer -> m ()-syncHeaders bb p = do- st <- asks chainState- s <- readTVarIO st- atomically . writeTVar st $- s {syncingPeer = Just p, newPeers = delete p (newPeers s)}- loc <- blockLocator bb- let m =- MGetHeaders- GetHeaders- { getHeadersVersion = myVersion- , getHeadersBL = loc- , getHeadersHashStop =- fromRight (error "Could not decode zero hash") . decode $- BS.replicate 32 0- }- PeerOutgoing m `send` p+ go =+ forever $ do+ threadDelay =<<+ liftIO (randomRIO (250 * 1000, 1000 * 1000))+ ChainPing `send` ch -peerString :: (MonadChain m, IsString a) => Peer -> m a+peerString :: MonadChain m => Peer -> m Text peerString p = do- mgr <- chainConfManager <$> asks myConfig- managerGetPeer mgr p >>= \case+ ChainConfig {chainConfManager = mgr} <- asks myReader+ managerGetPeer p mgr >>= \case Nothing -> return "[unknown]"- Just o -> return $ fromString $ show (onlinePeerAddress o)+ Just o -> return . cs . show $ onlinePeerAddress o
+ src/Network/Haskoin/Node/Chain/Logic.hs view
@@ -0,0 +1,284 @@+{-# 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.Arrow+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.Time.Clock+import Data.Time.Clock.POSIX+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++-- | Mutable state for the header chain process.+data ChainState p = ChainState+ { syncingPeer :: !(Maybe (p, UTCTime))+ -- ^ 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)+ => Network+ -> [BlockHeader]+ -> m (Either PeerException Bool)+importHeaders net hs =+ runExceptT $ do+ now <- fromIntegral <$> computeTime+ lift (connectBlocks net now hs) >>= \case+ Right _ ->+ 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) => m Bool+notifySynced =+ fmap isJust $+ runMaybeT $ do+ bb <- getBestBlockHeader+ now <- liftIO getCurrentTime+ let bt =+ posixSecondsToUTCTime . realToFrac . blockTimestamp $+ nodeHeader bb+ guard (diffUTCTime now bt < 2 * 60 * 60)+ st <- asks chainState+ MaybeT . atomically . runMaybeT $ do+ s <- lift $ readTVar st+ guard . isNothing $ syncingPeer 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)+ => BlockNode+ -> p+ -> m GetHeaders+syncHeaders bb p = do+ st <- asks chainState+ now <- liftIO getCurrentTime+ atomically . modifyTVar st $ \s ->+ s {syncingPeer = Just (p, now), 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) => m ()+setLastReceived = do+ st <- asks chainState+ now <- liftIO getCurrentTime+ atomically . modifyTVar st $ \s ->+ s {syncingPeer = second (const now) <$> syncingPeer 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 fst . syncingPeer <$> (readTVarIO =<< asks chainState)++-- | Set syncing peer to the pone provided.+setSyncingPeer :: (MonadChainLogic a p m, MonadIO m) => p -> m ()+setSyncingPeer p = do+ now <- liftIO getCurrentTime+ asks chainState >>= \v ->+ atomically . modifyTVar v $ \s -> s {syncingPeer = Just (p, now)}++-- | 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)+ , syncingPeer =+ case syncingPeer s of+ Just (p', _)+ | p == p' -> Nothing+ _ -> syncingPeer s+ }++-- | Return the syncing peer and time of last communication received, if any.+lastMessage :: (MonadChainLogic a p m, MonadIO m) => m (Maybe (p, UTCTime))+lastMessage = syncingPeer <$> (readTVarIO =<< asks chainState)
src/Network/Haskoin/Node/Common.hs view
@@ -2,9 +2,26 @@ {-# LANGUAGE LambdaCase #-} {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE RankNTypes #-}+{-|+Module : Network.Haskoin.Node.Common+Copyright : No rights reserved+License : UNLICENSE+Maintainer : xenog@protonmail.com+Stability : experimental+Portability : POSIX++Common functions used by Haskoin Node.+-} module Network.Haskoin.Node.Common where -import Data.ByteString (ByteString)+import Conduit+import Control.Monad+import Control.Monad.Trans.Maybe+import Data.Conduit.Network+import Data.Function+import Data.List+import Data.Maybe+import Data.String.Conversions import Data.Time.Clock import Data.Time.Clock.POSIX import Data.Word@@ -13,287 +30,244 @@ import Network.Haskoin.Constants import Network.Haskoin.Network import Network.Haskoin.Transaction-import Network.Socket (AddrInfo (..), AddrInfoFlag (..),- Family (..), NameInfoFlag (..),- SockAddr (..), SocketType (..),- addrAddress, defaultHints,- getAddrInfo, getNameInfo)+import Network.Socket hiding (send) import NQE+import System.Random import Text.Read import UnliftIO +-- | Type alias for a combination of hostname and port. type HostPort = (Host, Port)++-- | Type alias for a hostname. type Host = String++-- | Type alias for a port number. type Port = Int -- | Data structure representing an online peer. data OnlinePeer = OnlinePeer- { onlinePeerAddress :: !SockAddr+ { onlinePeerAddress :: !SockAddr -- ^ network address- , onlinePeerConnected :: !Bool- -- ^ has it finished handshake- , onlinePeerVersion :: !Word32+ , onlinePeerVerAck :: !Bool+ -- ^ got version acknowledgement from peer+ , onlinePeerConnected :: !Bool+ -- ^ peer is connected and ready+ , onlinePeerVersion :: !(Maybe Version) -- ^ protocol version- , onlinePeerServices :: !Word64- -- ^ services field- , onlinePeerRemoteNonce :: !Word64- -- ^ random nonce sent by peer- , onlinePeerUserAgent :: !ByteString- -- ^ user agent string- , onlinePeerRelay :: !Bool- -- ^ peer will relay transactions (BIP-37)- , onlinePeerBestBlock :: !BlockNode- -- ^ estimated best block that peer has- , onlinePeerAsync :: !(Async ())+ , onlinePeerAsync :: !(Async ()) -- ^ peer asynchronous process- , onlinePeerMailbox :: !Peer+ , onlinePeerMailbox :: !Peer -- ^ peer mailbox- , onlinePeerNonce :: !Word64+ , onlinePeerNonce :: !Word64 -- ^ random nonce sent during handshake- , onlinePeerPings :: ![NominalDiffTime]+ , onlinePeerPing :: !(Maybe (UTCTime, Word64))+ -- ^ last sent ping time and nonce+ , onlinePeerPings :: ![NominalDiffTime] -- ^ last few ping rountrip duration } --- | Mailbox for a peer process.-type Peer = Inbox PeerMessage+instance Eq OnlinePeer where+ (==) = (==) `on` f+ where+ f OnlinePeer {onlinePeerMailbox = p} = p --- | Mailbox for chain headers process.-type Chain = Inbox ChainMessage+instance Ord OnlinePeer where+ compare = compare `on` f+ where+ f OnlinePeer {onlinePeerPings = pings} = fromMaybe 60 (median pings) +-- | Mailbox for a peer.+type Peer = Mailbox PeerMessage++-- | Mailbox for chain header syncing process.+type Chain = Mailbox ChainMessage+ -- | Mailbox for peer manager process.-type Manager = Inbox ManagerMessage+type Manager = Mailbox ManagerMessage --- | Node configuration. Mailboxes for manager and chain processes must be--- created before launching the node. The node will start those processes and--- receive any messages sent to those mailboxes.+-- | General node configuration. data NodeConfig = NodeConfig- { maxPeers :: !Int+ { nodeConfMaxPeers :: !Int -- ^ maximum number of connected peers allowed- , database :: !DB- -- ^ RocksDB database handler- , initPeers :: ![HostPort]+ , nodeConfDB :: !DB+ -- ^ database handler+ , nodeConfPeers :: ![HostPort] -- ^ static list of peers to connect to- , discover :: !Bool+ , nodeConfDiscover :: !Bool -- ^ activate peer discovery- , nodeEvents :: !(Listen NodeEvent)- -- ^ listener for events originated by the node- , netAddress :: !NetworkAddress+ , nodeConfNetAddr :: !NetworkAddress -- ^ network address for the local host- , nodeNet :: !Network+ , nodeConfNet :: !Network -- ^ network constants+ , nodeConfEvents :: !(Publisher NodeEvent)+ -- ^ node events are sent to this publisher+ , nodeConfTimeout :: !Int+ -- ^ timeout in seconds } --- | Peer manager configuration. Mailbox must be created before starting the--- process.+-- | Peer manager configuration. data ManagerConfig = ManagerConfig- { mgrConfMaxPeers :: !Int+ { mgrConfMaxPeers :: !Int -- ^ maximum number of peers to connect to- , mgrConfDB :: !DB- -- ^ RocksDB database handler to store peer information- , mgrConfPeers :: ![HostPort]+ , mgrConfDB :: !DB+ -- ^ database handler to store peer information+ , mgrConfPeers :: ![HostPort] -- ^ static list of peers to connect to- , mgrConfDiscover :: !Bool+ , mgrConfDiscover :: !Bool -- ^ activate peer discovery- , mgrConfMgrListener :: !(Listen ManagerEvent)- -- ^ listener for events originating from peer manager- , mgrConfPeerListener :: !(Listen (Peer, PeerEvent))- -- ^ listener for events originating from individual peers- , mgrConfNetAddr :: !NetworkAddress+ , mgrConfNetAddr :: !NetworkAddress -- ^ network address for the local host- , mgrConfManager :: !Manager- -- ^ peer manager mailbox- , mgrConfChain :: !Chain- -- ^ chain process mailbox- , mgrConfNetwork :: !Network+ , mgrConfNetwork :: !Network -- ^ network constants+ , mgrConfEvents :: !(Listen PeerEvent)+ -- ^ send manager and peer messages to this mailbox+ , mgrConfTimeout :: !Int+ -- ^ timeout in seconds } --- | Event originating from the node. Aggregates events from the peer manager,--- chain, and any connected peers.-data NodeEvent- = ManagerEvent !ManagerEvent- -- ^ event originating from peer manager- | ChainEvent !ChainEvent- -- ^ event originating from chain process- | PeerEvent !(Peer, PeerEvent)- -- ^ event originating from a peer---- | Peer manager event.-data ManagerEvent- = ManagerConnect !Peer- -- ^ a new peer connected and its handshake completed- | ManagerDisconnect !Peer- -- ^ a peer disconnected- -- | Messages that can be sent to the peer manager. data ManagerMessage- = ManagerSetFilter !BloomFilter- -- ^ set a bloom filter in all peers- | ManagerSetBest !BlockNode- -- ^ set our best block- | ManagerPing- -- ^ internal timer signal that triggers housekeeping tasks- | ManagerGetAddr !Peer- -- ^ peer requests all peers we know about- | ManagerNewPeers !Peer- ![NetworkAddressTime]- -- ^ peer sent list of peers it knows about+ = ManagerConnect+ -- ^ try to connect to peers | ManagerKill !PeerException !Peer- -- ^ please kill this peer with supplied exception- | ManagerSetPeerBest !Peer- !BlockNode- -- ^ set best block for this peer- | ManagerGetPeerBest !Peer- !(Reply (Maybe BlockNode))- -- ^ get best block that manager thinks peer has- | ManagerSetPeerVersion !Peer- !Version- -- ^ set version for this peer- | ManagerGetPeerVersion !Peer- !(Reply (Maybe Word32))- -- ^ get protocol version for this peer- | ManagerGetPeers !(Reply [OnlinePeer])+ -- ^ kill this peer with supplied exception+ | ManagerGetPeers !(Listen [OnlinePeer]) -- ^ get all connected peers- | ManagerGetOnlinePeer !Peer !(Reply (Maybe OnlinePeer))+ | ManagerGetOnlinePeer !Peer !(Listen (Maybe OnlinePeer)) -- ^ get a peer information- | ManagerPeerPing !Peer- !NominalDiffTime- -- ^ add a peer roundtrip time for this peer- | PeerStopped !(Async (), Either SomeException ())- -- ^ peer corresponding to 'Async' has stopped+ | ManagerPurgePeers+ -- ^ delete all known peers+ | ManagerCheckPeer !Peer+ -- ^ check this peer+ | ManagerPeerMessage !Peer !Message+ -- ^ peer got a message that is forwarded to manager+ | ManagerPeerDied !Child !(Maybe SomeException)+ -- ^ child died+ | ManagerBestBlock !BlockHeight+ -- ^ set this as our best block --- | Configuration for the chain process.+-- | Configuration for chain syncing process. data ChainConfig = ChainConfig- { chainConfDB :: !DB+ { chainConfDB :: !DB -- ^ RocksDB database handle- , chainConfListener :: !(Listen ChainEvent)- -- ^ listener for events originating from the chain process- , chainConfManager :: !Manager+ , chainConfManager :: !Manager -- ^ peer manager mailbox- , chainConfChain :: !Chain- -- ^ chain process mailbox- , chainConfNetwork :: !Network+ , chainConfNetwork :: !Network -- ^ network constants+ , chainConfEvents :: !(Listen ChainEvent)+ -- ^ send header chain events here+ , chainConfTimeout :: !Int+ -- ^ timeout in seconds } -- | Messages that can be sent to the chain process. data ChainMessage- = ChainNewHeaders !Peer- ![BlockHeaderCount]- -- ^ peer sent some block headers- | ChainNewPeer !Peer- -- ^ a new peer connected- | ChainRemovePeer !Peer- -- ^ a peer disconnected- | ChainGetBest !(Reply BlockNode)+ = ChainGetBest !(Listen BlockNode) -- ^ get best block known+ | ChainHeaders !Peer+ ![BlockHeader] | ChainGetAncestor !BlockHeight !BlockNode- !(Reply (Maybe BlockNode))+ !(Listen (Maybe BlockNode)) -- ^ get ancestor for 'BlockNode' at 'BlockHeight' | ChainGetSplit !BlockNode !BlockNode- !(Reply BlockNode)+ !(Listen BlockNode) -- ^ get highest common node | ChainGetBlock !BlockHash- !(Reply (Maybe BlockNode))+ !(Listen (Maybe BlockNode)) -- ^ get a block header- | ChainNewBlocks !Peer ![BlockHash]- -- ^ peer sent block inventory- | ChainSendHeaders !Peer- -- ^ peer asks for our block headers in the future- | ChainIsSynced !(Reply Bool)+ | ChainIsSynced !(Listen Bool) -- ^ is chain in sync with network?+ | ChainPing+ -- ^ internal message for process housekeeping+ | ChainPeerConnected !Peer+ !SockAddr+ -- ^ internal message to notify that a peer has connected+ | ChainPeerDisconnected !Peer+ !SockAddr+ -- ^ internal message to notify that a peer has disconnected --- | Events originating from chain process.+-- | Events originating from chain syncing process. data ChainEvent- = ChainNewBest !BlockNode+ = ChainBestBlock !BlockNode -- ^ chain has new best block | ChainSynced !BlockNode -- ^ chain is in sync with the network deriving (Eq, Show) +-- | Chain and peer events generated by the node.+data NodeEvent+ = ChainEvent !ChainEvent+ -- ^ events from the chain syncing process+ | PeerEvent !PeerEvent+ -- ^ events from peers and peer manager+ deriving Eq+ -- | Configuration for a particular peer. data PeerConfig = PeerConfig- { peerConfConnect :: !NetworkAddress- -- ^ address of remote peer- , peerConfLocal :: !NetworkAddress- -- ^ our address to send to peer- , peerConfManager :: !Manager- -- ^ peer manager mailbox- , peerConfChain :: !Chain- -- ^ chain process mailbox- , peerConfListener :: !(Listen (Peer, PeerEvent))- -- ^ listener for peer events- , peerConfNonce :: !Word64- -- ^ our random nonce to send to peer- , peerConfNetwork :: !Network+ { peerConfListen :: !(Publisher Message)+ -- ^ Send peer messages to publisher+ , peerConfNetwork :: !Network -- ^ network constants+ , peerConfAddress :: !SockAddr+ -- ^ peer address } -- | Reasons why a peer may stop working. data PeerException = PeerMisbehaving !String- -- ^ peer was a naughty boy- | DecodeMessageError !String- -- ^ incoming message could not be decoded- | CannotDecodePayload !String+ -- ^ peer is being a naughty boy+ | DuplicateVersion+ -- ^ peer sent an extra version message+ | DecodeHeaderError+ -- ^ incoming message headers could not be decoded+ | CannotDecodePayload -- ^ incoming message payload could not be decoded | PeerIsMyself -- ^ nonce for peer matches ours | PayloadTooLarge !Word32 -- ^ message payload too large | PeerAddressInvalid- -- ^ peer address did not parse with 'fromSockAddr'- | BloomFiltersNotSupported- -- ^ peer does not support bloom filters+ -- ^ peer address not valid | PeerSentBadHeaders -- ^ peer sent wrong headers | NotNetworkPeer- -- ^ peer is SPV and cannot serve blockchain data+ -- ^ peer cannot serve block chain data | PeerNoSegWit -- ^ peer has no segwit support | PeerTimeout -- ^ request to peer timed out+ | PurgingPeer+ -- ^ peers are being purged+ | UnknownPeer+ -- ^ peer is unknown deriving (Eq, Show) instance Exception PeerException --- | Events originating from a peer.+-- | Events originating from peers and the peer manager. data PeerEvent- = TxAvail ![TxHash]- -- ^ peer sent transaction inventory- | GotBlock !Block- -- ^ peer sent a 'Block'- | GotMerkleBlock !MerkleBlock- -- ^ peer sent a 'MerkleBlock'- | GotTx !Tx- -- ^ peer sent a 'Tx'- | GotPong !Word64- -- ^ peer responded to a 'Ping'- | SendBlocks !GetBlocks- -- ^ peer is requesting some blocks- | SendHeaders !GetHeaders- -- ^ peer is requesting some headers- | SendData ![InvVector]- -- ^ per is requesting an inventory- | TxNotFound !TxHash- -- ^ peer could not find transaction- | BlockNotFound !BlockHash- -- ^ peer could not find block- | WantMempool- -- ^ peer wants our mempool- | Rejected !Reject- -- ^ peer rejected something we sent+ = PeerConnected !Peer+ !SockAddr+ -- ^ new peer connected+ | PeerDisconnected !Peer+ !SockAddr+ -- ^ peer disconnected+ | PeerMessage !Peer+ !Message+ -- ^ peer sent a message+ deriving Eq --- | Internal type for peer messages.+-- | Incoming messages that a peer accepts. data PeerMessage- = PeerOutgoing !Message- | PeerIncoming !Message+ = GetPublisher !(Listen (Publisher Message))+ | SendMessage !Message --- | Convert a host and port into a list of matching 'SockAddr'.-toSockAddr :: (MonadUnliftIO m) => HostPort -> m [SockAddr]+-- | Resolve a host and port to a list of 'SockAddr'. May make use DNS resolver.+toSockAddr :: MonadUnliftIO m => HostPort -> m [SockAddr] toSockAddr (host, port) = go `catch` e where go =@@ -310,141 +284,178 @@ e :: Monad m => SomeException -> m [SockAddr] e _ = return [] --- | Convert a 'SockAddr' into a host and port.+-- | Convert a 'SockAddr' to a a numeric host and port. fromSockAddr :: (MonadUnliftIO m) => SockAddr -> m (Maybe HostPort) fromSockAddr sa = go `catch` e where go = do- (hostM, portM) <- liftIO (getNameInfo flags True True sa)- return $ (,) <$> hostM <*> (readMaybe =<< portM)+ (maybe_host, maybe_port) <- liftIO (getNameInfo flags True True sa)+ return $ (,) <$> maybe_host <*> (readMaybe =<< maybe_port) flags = [NI_NUMERICHOST, NI_NUMERICSERV] e :: Monad m => SomeException -> m (Maybe a) e _ = return Nothing -- | Integer current time in seconds from 1970-01-01T00:00Z.-computeTime :: MonadIO m => m Word32+computeTime :: MonadIO m => m Word64 computeTime = round <$> liftIO getPOSIXTime -- | Our protocol version. myVersion :: Word32 myVersion = 70012 --- | Set best block in the manager.-managerSetBest :: MonadIO m => BlockNode -> Manager -> m ()-managerSetBest bn mgr = ManagerSetBest bn `send` mgr---- | Set version of peer in manager.-managerSetPeerVersion :: MonadIO m => Peer -> Version -> Manager -> m ()-managerSetPeerVersion p v mgr = ManagerSetPeerVersion p v `send` mgr---- | Get version of peer from manager.-managerGetPeerVersion :: MonadIO m => Peer -> Manager -> m (Maybe Word32)-managerGetPeerVersion p mgr = ManagerGetPeerVersion p `query` mgr---- | Get best block for peer from manager.-managerGetPeerBest :: MonadIO m => Peer -> Manager -> m (Maybe BlockNode)-managerGetPeerBest p mgr = ManagerGetPeerBest p `query` mgr---- | Set best block for peer in manager.-managerSetPeerBest :: MonadIO m => Peer -> BlockNode -> Manager -> m ()-managerSetPeerBest p bn mgr = ManagerSetPeerBest p bn `send` mgr+-- | Internal function used by peer to send a message to the peer manager.+managerPeerMessage :: MonadIO m => Peer -> Message -> Manager -> m ()+managerPeerMessage p msg mgr = ManagerPeerMessage p msg `send` mgr --- | Get list of peers from manager.-managerGetPeers :: MonadIO m => Manager -> m [OnlinePeer]+-- | Get list of connected peers from manager.+managerGetPeers ::+ MonadIO m => Manager -> m [OnlinePeer] managerGetPeers mgr = ManagerGetPeers `query` mgr --- | Get peer information for a peer from manager.-managerGetPeer :: MonadIO m => Manager -> Peer -> m (Maybe OnlinePeer)-managerGetPeer mgr p = ManagerGetOnlinePeer p `query` mgr---- | Ask manager to send all known peers to a peer.-managerGetAddr :: MonadIO m => Peer -> Manager -> m ()-managerGetAddr p mgr = ManagerGetAddr p `send` mgr+-- | Get information for an online peer from manager.+managerGetPeer :: MonadIO m => Peer -> Manager -> m (Maybe OnlinePeer)+managerGetPeer p mgr = ManagerGetOnlinePeer p `query` mgr -- | Ask manager to kill a peer with the provided exception. managerKill :: MonadIO m => PeerException -> Peer -> Manager -> m () managerKill e p mgr = ManagerKill e p `send` mgr --- | Peer sends manager list of known peers.-managerNewPeers ::- MonadIO m => Peer -> [NetworkAddressTime] -> Manager -> m ()-managerNewPeers p as mgr = ManagerNewPeers p as `send` mgr+-- | Internal function used by manager to check peers periodically.+managerCheck :: MonadIO m => Peer -> Manager -> m ()+managerCheck p mgr = ManagerCheckPeer p `send` mgr --- | Set bloom filters in peer manager.-setManagerFilter :: MonadIO m => BloomFilter -> Manager -> m ()-setManagerFilter bf mgr = ManagerSetFilter bf `send` mgr+-- | Internal function used to ask manager to connect to a new peer.+managerConnect :: MonadIO m => Manager -> m ()+managerConnect mgr = ManagerConnect `send` mgr +-- | Set the best block that the manager knows about.+managerSetBest :: MonadIO m => BlockHeight -> Manager -> m ()+managerSetBest bh mgr = ManagerBestBlock bh `send` mgr+ -- | Send a network message to peer. sendMessage :: MonadIO m => Message -> Peer -> m ()-sendMessage msg p = PeerOutgoing msg `send` p---- | Upload bloom filter to remote peer.-peerSetFilter :: MonadIO m => BloomFilter -> Peer -> m ()-peerSetFilter f p = MFilterLoad (FilterLoad f) `sendMessage` p---- | Request Merkle blocks from peer.-getMerkleBlocks ::- (MonadIO m)- => Peer- -> [BlockHash]- -> m ()-getMerkleBlocks p bhs = PeerOutgoing (MGetData (GetData ivs)) `send` p- where- ivs = map (InvVector InvMerkleBlock . getBlockHash) bhs+sendMessage msg p = SendMessage msg `send` p --- | Request full blocks from peer.+-- | Request full blocks from peer. Will return 'Nothing' if the list of blocks+-- returned by the peer is incomplete, comes out of order, or a timeout is+-- reached. peerGetBlocks ::- MonadIO m => Network -> Peer -> [BlockHash] -> m ()-peerGetBlocks net p bhs = PeerOutgoing (MGetData (GetData ivs)) `send` p+ MonadUnliftIO m+ => Network+ -> Int+ -> Peer+ -> [BlockHash]+ -> m (Maybe [Block])+peerGetBlocks net time p bhs =+ runMaybeT $ mapM f =<< MaybeT (peerGetData time p (GetData ivs)) where- con+ f (Right b) = return b+ f (Left _) = MaybeT $ return Nothing+ c | getSegWit net = InvWitnessBlock | otherwise = InvBlock- ivs = map (InvVector con . getBlockHash) bhs+ ivs = map (InvVector c . getBlockHash) bhs --- | Request transactions from peer.-peerGetTxs :: MonadIO m => Network -> Peer -> [TxHash] -> m ()-peerGetTxs net p ths = PeerOutgoing (MGetData (GetData ivs)) `send` p+-- | Request transactions from peer. Will return 'Nothing' if the list of+-- transactions returned by the peer is incomplete, comes out of order, or a+-- timeout is reached.+peerGetTxs ::+ MonadUnliftIO m+ => Network+ -> Int+ -> Peer+ -> [TxHash]+ -> m (Maybe [Tx])+peerGetTxs net time p ths =+ runMaybeT $ mapM f =<< MaybeT (peerGetData time p (GetData ivs)) where- con+ f (Right _) = MaybeT $ return Nothing+ f (Left t) = return t+ c | getSegWit net = InvWitnessTx | otherwise = InvTx- ivs = map (InvVector con . getTxHash) ths+ ivs = map (InvVector c . getTxHash) ths --- | Build my version structure.+-- | Request transactions and/or blocks from peer. Return maybe if any single+-- inventory fails to be retrieved, if they come out of order, or if timeout is+-- reached.+peerGetData ::+ MonadUnliftIO m => Int -> Peer -> GetData -> m (Maybe [Either Tx Block])+peerGetData time p gd@(GetData ivs) =+ runMaybeT $ do+ pub <- MaybeT $ queryS time GetPublisher p+ MaybeT $+ withSubscription pub $ \sub -> do+ MGetData gd `sendMessage` p+ r <- liftIO randomIO+ MPing (Ping r) `sendMessage` p+ join <$>+ timeout+ (time * 1000 * 1000)+ (runMaybeT (get_thing sub r [] ivs))+ where+ get_thing _ _ acc [] = return $ reverse acc+ get_thing sub r acc hss@(InvVector t h:hs) =+ receive sub >>= \case+ MTx tx+ | is_tx t && getTxHash (txHash tx) == h ->+ get_thing sub r (Left tx : acc) hs+ MBlock b@(Block bh _)+ | is_block t && getBlockHash (headerHash bh) == h ->+ get_thing sub r (Right b : acc) hs+ MNotFound (NotFound nvs)+ | not (null (nvs `union` hs)) -> MaybeT $ return Nothing+ MPong (Pong r')+ | r == r' -> MaybeT $ return Nothing+ _+ | null acc -> get_thing sub r acc hss+ | otherwise -> MaybeT $ return Nothing+ is_tx InvWitnessTx = True+ is_tx InvTx = True+ is_tx _ = False+ is_block InvWitnessBlock = True+ is_block InvBlock = True+ is_block _ = False++-- | Ping a peer and await response. Return 'False' if response not received+-- before timeout.+peerPing :: MonadUnliftIO m => Int -> Peer -> m Bool+peerPing time p =+ fmap isJust . runMaybeT $ do+ pub <- MaybeT $ queryS time GetPublisher p+ MaybeT $+ withSubscription pub $ \sub -> do+ r <- liftIO randomIO+ MPing (Ping r) `sendMessage` p+ receiveMatchS time sub $ \case+ MPong (Pong r')+ | r == r' -> Just ()+ _ -> Nothing++-- | Create version data structure. buildVersion ::- MonadIO m- => Network+ Network -> Word64 -> BlockHeight -> NetworkAddress -> NetworkAddress- -> m Version-buildVersion net nonce height loc rmt = do- time <- fromIntegral <$> computeTime- return- Version- { version = myVersion- , services = naServices loc- , timestamp = time- , addrRecv = rmt- , addrSend = loc- , verNonce = nonce- , userAgent = VarString (getHaskoinUserAgent net)- , startHeight = height- , relay = True- }---- | Notify chain of a new peer that connected.-chainNewPeer :: MonadIO m => Peer -> Chain -> m ()-chainNewPeer p ch = ChainNewPeer p `send` ch---- | Notify chain that a peer has disconnected.-chainRemovePeer :: MonadIO m => Peer -> Chain -> m ()-chainRemovePeer p ch = ChainRemovePeer p `send` ch+ -> Word64+ -> Version+buildVersion net nonce height loc rmt time =+ Version+ { version = myVersion+ , services = naServices loc+ , timestamp = time+ , addrRecv = rmt+ , addrSend = loc+ , verNonce = nonce+ , userAgent = VarString (getHaskoinUserAgent net)+ , startHeight = height+ , relay = True+ } --- | Get a block header from chain process.+-- | Get a block header from 'Chain' process. chainGetBlock :: MonadIO m => BlockHash -> Chain -> m (Maybe BlockNode) chainGetBlock bh ch = ChainGetBlock bh `query` ch @@ -475,6 +486,14 @@ MonadIO m => BlockNode -> BlockNode -> Chain -> m BlockNode chainGetSplitBlock l r c = ChainGetSplit l r `query` c +-- | Notify chain that a new peer is connected.+chainPeerConnected :: MonadIO m => Peer -> SockAddr -> Chain -> m ()+chainPeerConnected p a ch = ChainPeerConnected p a `send` ch++-- | Notify chain that a peer has disconnected.+chainPeerDisconnected :: MonadIO m => Peer -> SockAddr -> Chain -> m ()+chainPeerDisconnected p a ch = ChainPeerDisconnected p a `send` ch+ -- | Is given 'BlockHash' in the main chain? chainBlockMain :: MonadIO m => BlockHash -> Chain -> m Bool chainBlockMain bh ch =@@ -486,3 +505,25 @@ -- | Is chain in sync with network? chainIsSynced :: MonadIO m => Chain -> m Bool chainIsSynced ch = ChainIsSynced `query` ch++-- | Peer sends a bunch of headers to the chain process.+chainHeaders :: MonadIO m => Peer -> [BlockHeader] -> Chain -> m ()+chainHeaders p hs ch = ChainHeaders p hs `send` ch++-- | Connect to a socket via TCP.+withConnection ::+ MonadUnliftIO m => SockAddr -> (AppData -> m a) -> m a+withConnection na f =+ fromSockAddr na >>= \case+ Nothing -> throwIO PeerAddressInvalid+ Just (host, port) ->+ let cset = clientSettings port (cs host)+ in runGeneralTCPClient cset f++-- | Calculate the median value from a list. The list must not be empty.+median :: Fractional a => [a] -> Maybe a+median ls+ | null ls = Nothing+ | length ls `mod` 2 == 0 =+ Just . (/ 2) . sum . take 2 $ drop (length ls `div` 2 - 1) ls+ | otherwise = Just . head $ drop (length ls `div` 2) ls
src/Network/Haskoin/Node/Manager.hs view
@@ -1,594 +1,380 @@ {-# LANGUAGE ConstraintKinds #-} {-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE LambdaCase #-} {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE OverloadedStrings #-}-{-# LANGUAGE RecordWildCards #-} {-# LANGUAGE TemplateHaskell #-} {-# LANGUAGE TupleSections #-}+{-|+Module : Network.Haskoin.Node.Manager+Copyright : No rights reserved+License : UNLICENSE+Maintainer : xenog@protonmail.com+Stability : experimental+Portability : POSIX++Peer manager process.+-} module Network.Haskoin.Node.Manager ( manager ) where +import Conduit import Control.Applicative import Control.Monad import Control.Monad.Except import Control.Monad.Logger import Control.Monad.Reader-import Data.Bits-import qualified Data.ByteString as BS-import Data.Conduit-import qualified Data.Conduit.Combinators as CC-import Data.Function-import Data.List-import Data.Maybe-import Data.Serialize (Get, Put, Serialize, get, put)-import qualified Data.Serialize as S-import Data.String+import Control.Monad.Trans.Maybe import Data.String.Conversions-import Data.Word-import Database.RocksDB (DB)-import Database.RocksDB.Query+import Data.Time.Clock+import Database.RocksDB (DB) import Network.Haskoin.Block import Network.Haskoin.Constants import Network.Haskoin.Network 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 -type MonadManager n m- = ( MonadUnliftIO n- , MonadLoggerIO n- , MonadUnliftIO m- , MonadLoggerIO m- , MonadReader (ManagerReader n) m)+-- | Monad used by most functions in this module.+type MonadManager m = (MonadLoggerIO m, MonadReader ManagerReader m) -data ManagerReader m = ManagerReader- { mySelf :: !Manager- , myChain :: !Chain- , myConfig :: !ManagerConfig- , myPeerDB :: !DB- , myPeerSupervisor :: !(Inbox (SupervisorMessage m))- , onlinePeers :: !(TVar [OnlinePeer])- , myBloomFilter :: !(TVar (Maybe BloomFilter))- , myBestBlock :: !(TVar BlockNode)+-- | Reader for peer configuration and state.+data ManagerReader = ManagerReader+ { myConfig :: !ManagerConfig+ , mySupervisor :: !Supervisor+ , myMailbox :: !Manager+ , myBestBlock :: !(TVar BlockHeight)+ , onlinePeers :: !(TVar [OnlinePeer]) } -data Priority- = PriorityNetwork- | PrioritySeed- | PriorityManual- deriving (Eq, Show, Ord)--data PeerAddress- = PeerAddress { getPeerAddress :: SockAddr }- | PeerAddressBase- deriving (Eq, Show)--instance Serialize Priority where- get =- S.getWord8 >>= \case- 0x00 -> return PriorityManual- 0x01 -> return PrioritySeed- 0x02 -> return PriorityNetwork- _ -> mzero- put PriorityManual = S.putWord8 0x00- put PrioritySeed = S.putWord8 0x01- put PriorityNetwork = S.putWord8 0x02--instance Serialize PeerAddress where- get = do- guard . (== 0x81) =<< S.getWord8- record <|> return PeerAddressBase- where- record = do- getPeerAddress <- decodeSockAddr- return PeerAddress {..}- put PeerAddress {..} = do- S.putWord8 0x81- encodeSockAddr getPeerAddress- put PeerAddressBase = S.putWord8 0x81+-- | Peer Manager process. In order to fully start it needs to receive a+-- 'ManageBestBlock' event.+manager ::+ (MonadUnliftIO m, MonadLoggerIO m)+ => ManagerConfig+ -> Inbox ManagerMessage+ -> m ()+manager cfg inbox =+ withSupervisor (Notify f) $ \sup -> do+ bb <- newTVarIO 0+ ob <- newTVarIO []+ let rd =+ ManagerReader+ { myConfig = cfg+ , mySupervisor = sup+ , myMailbox = mgr+ , myBestBlock = bb+ , onlinePeers = ob+ }+ go `runReaderT` rd+ where+ mgr = inboxToMailbox inbox+ discover = mgrConfDiscover cfg+ go = do+ db <- getManagerDB+ $(logDebugS) "Manager" "Initializing..."+ initPeerDB db discover+ putBestBlock <=< receiveMatch inbox $ \case+ ManagerBestBlock b -> Just b+ _ -> Nothing+ $(logDebugS) "Manager" "Initialization complete"+ withConnectLoop mgr $ \a -> do+ link a+ forever $ receive inbox >>= managerMessage+ f (a, mex) = ManagerPeerDied a mex `sendSTM` mgr -data PeerTimeAddress- = PeerTimeAddress { getPeerPrio :: !Priority- , getPeerBanned :: !Word32- , getPeerLastConnect :: !Word32- , getPeerNextConnect :: !Word32- , getPeerTimeAddress :: !PeerAddress }- | PeerTimeAddressBase- deriving (Eq, Show)+putBestBlock :: MonadManager m => BlockHeight -> m ()+putBestBlock bb = asks myBestBlock >>= \b -> atomically $ writeTVar b bb -instance Key PeerTimeAddress-instance KeyValue PeerAddress PeerTimeAddress-instance KeyValue PeerTimeAddress PeerAddress+getBestBlock :: MonadManager m => m BlockHeight+getBestBlock = asks myBestBlock >>= readTVarIO -instance Serialize PeerTimeAddress where- get = do- guard . (== 0x80) =<< S.getWord8- record <|> return PeerTimeAddressBase- where- record = do- getPeerPrio <- S.get- getPeerBanned <- S.get- getPeerLastConnect <- (maxBound -) <$> S.get- getPeerNextConnect <- S.get- getPeerTimeAddress <- S.get- return PeerTimeAddress {..}- put PeerTimeAddress {..} = do- S.putWord8 0x80- S.put getPeerPrio- S.put getPeerBanned- S.put (maxBound - getPeerLastConnect)- S.put getPeerNextConnect- S.put getPeerTimeAddress- put PeerTimeAddressBase = S.putWord8 0x80+getNetwork :: MonadManager m => m Network+getNetwork = mgrConfNetwork <$> asks myConfig -manager :: (MonadUnliftIO m, MonadLoggerIO m) => ManagerConfig -> m ()-manager cfg = do- psup <- newInbox =<< newTQueueIO- withAsync (supervisor (Notify dead) psup []) $ \sup -> do- link sup- bb <- chainGetBest $ mgrConfChain cfg- opb <- newTVarIO []- bfb <- newTVarIO Nothing- bbb <- newTVarIO bb- withConnectLoop (mgrConfManager cfg) $ do- let rd =- ManagerReader- { mySelf = mgrConfManager cfg- , myChain = mgrConfChain cfg- , myConfig = cfg- , myPeerDB = mgrConfDB cfg- , myPeerSupervisor = psup- , onlinePeers = opb- , myBloomFilter = bfb- , myBestBlock = bbb- }- run `runReaderT` rd- where- dead ex = PeerStopped ex `sendSTM` mgrConfManager cfg- run = do- connectNewPeers- managerLoop+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 -resolvePeers :: (MonadUnliftIO m, MonadManager n m) => m [(SockAddr, Priority)]-resolvePeers = do- cfg <- asks myConfig- let net = mgrConfNetwork cfg- confPeers <-- fmap- (map (, PriorityManual) . concat)- (mapM toSockAddr (mgrConfPeers cfg))- if mgrConfDiscover cfg- then do- seedPeers <-- fmap- (map (, PrioritySeed) . concat)- (mapM (toSockAddr . (, getDefaultPort net)) (getSeeds net))- return (confPeers ++ seedPeers)- else return confPeers+logConnectedPeers :: MonadManager m => m ()+logConnectedPeers = do+ m <- mgrConfMaxPeers <$> asks myConfig+ l <- length <$> getConnectedPeers+ $(logInfoS) "Manager" $+ "Peers connected: " <> cs (show l) <> "/" <> cs (show m) -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)+getManagerDB :: MonadManager m => m DB+getManagerDB = mgrConfDB <$> asks myConfig -encodeSockAddr (SockAddrInet p a) = do- S.putWord32be 0x00000000- S.putWord32be 0x00000000- S.putWord32be 0x0000ffff- S.putWord32host a- S.putWord16be (fromIntegral p)+getOnlinePeers :: MonadManager m => m [OnlinePeer]+getOnlinePeers = asks onlinePeers >>= readTVarIO -encodeSockAddr x = error $ "Colud not encode address: " <> show x+getConnectedPeers :: MonadManager m => m [OnlinePeer]+getConnectedPeers = filter onlinePeerConnected <$> getOnlinePeers -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+purgePeers :: (MonadUnliftIO m, MonadManager m) => m ()+purgePeers = do+ db <- getManagerDB+ ops <- getOnlinePeers+ forM_ ops $ \OnlinePeer {onlinePeerMailbox = p} -> killPeer PurgingPeer p+ purgePeerDB db -connectPeer :: MonadManager n m => SockAddr -> m ()-connectPeer sa = do- db <- asks myPeerDB- let k = PeerAddress sa- retrieve db Nothing k >>= \case- Nothing -> error "Could not find peer to mark connected"- Just v -> do- now <- computeTime- let v' = v {getPeerLastConnect = now}- writeBatch db [deleteOp v, insertOp v' k, insertOp k v']- logPeersConnected+forwardMessage :: MonadManager m => Peer -> Message -> m ()+forwardMessage p msg = managerEvent $ PeerMessage p msg +managerEvent :: MonadManager m => PeerEvent -> m ()+managerEvent e = mgrConfEvents <$> asks myConfig >>= \l -> atomically $ l e -logPeersConnected :: MonadManager n m => m ()-logPeersConnected = do- mo <- mgrConfMaxPeers <$> asks myConfig- ps <- getOnlinePeers+managerMessage :: (MonadUnliftIO m, MonadManager m) => ManagerMessage -> m ()+managerMessage (ManagerPeerMessage p (MVersion v)) = do+ b <- asks onlinePeers+ s <- atomically $ peerString b p+ e <-+ runExceptT $ do+ let ua = getVarString $ userAgent v+ $(logDebugS) "Manager" $+ "Got version from peer " <> s <> ": " <> cs ua+ o <- ExceptT . atomically $ setPeerVersion b p v+ when (onlinePeerConnected o) $ announcePeer p+ case e of+ Right () -> do+ $(logDebugS) "Manager" $ "Version accepted for peer " <> s+ MVerAck `sendMessage` p+ Left x -> do+ $(logErrorS) "Manager" $+ "Version rejected for peer " <> s <> ": " <> cs (show x)+ killPeer x p+managerMessage (ManagerPeerMessage p MVerAck) = do+ b <- asks onlinePeers+ s <- atomically $ peerString b p+ atomically (setPeerVerAck b p) >>= \case+ Just o -> do+ $(logDebugS) "Manager" $ "Received verack from peer: " <> s+ when (onlinePeerConnected o) $ announcePeer p+ Nothing -> do+ $(logErrorS) "Manager" $ "Received verack from unknown peer: " <> s+ killPeer UnknownPeer p+managerMessage (ManagerPeerMessage p (MAddr (Addr nas))) = do+ db <- getManagerDB+ b <- asks onlinePeers+ s <- atomically $ peerString b p $(logInfoS) "Manager" $- "Peers connected: " <> cs (show (length ps)) <> "/" <> cs (show mo)--storePeer :: MonadManager n m => SockAddr -> Priority -> m ()-storePeer sa prio = do- db <- asks myPeerDB- let k = PeerAddress sa- retrieve db Nothing k >>= \case+ "Received " <> cs (show (length nas)) <> " peers from " <> s+ let sas = map (naAddress . snd) nas+ forM_ sas $ newPeerDB db netPeerScore+managerMessage (ManagerPeerMessage p msg@(MPong (Pong n))) = do+ now <- liftIO getCurrentTime+ b <- asks onlinePeers+ s <- atomically $ peerString b p+ atomically (gotPong b n now p) >>= \case Nothing -> do- let v =- PeerTimeAddress- { getPeerPrio = prio- , getPeerBanned = 0- , getPeerLastConnect = 0- , getPeerNextConnect = 0- , getPeerTimeAddress = k- }- writeBatch- db- [insertOp v k, insertOp k v]- Just v@PeerTimeAddress {..} ->- when (getPeerPrio < prio) $ do- let v' = v {getPeerPrio = prio}- writeBatch- db- [deleteOp v, insertOp v' k, insertOp k v']- Just PeerTimeAddressBase ->- error "Key for peer is corrupted"+ $(logDebugS) "Manager" $+ "Forwarding pong " <> cs (show n) <> " from " <> s+ forwardMessage p msg+ Just d -> do+ let ms = d * 1000+ $(logDebugS) "Manager" $+ "Ping " <> cs (show n) <> " to " <> s <> " took " <>+ cs (show ms) <>+ " milliseconds"+managerMessage (ManagerPeerMessage p (MPing (Ping n))) = do+ b <- asks onlinePeers+ s <- atomically $ peerString b p+ $(logDebugS) "Manager" $+ "Responding to ping " <> cs (show n) <> " from " <> s+ MPong (Pong n) `sendMessage` p+managerMessage (ManagerPeerMessage p msg) = do+ b <- asks onlinePeers+ s <- atomically $ peerString b p+ let cmd = commandToString $ msgType msg+ $(logDebugS) "Manager" $+ "Forwarding message " <> cs cmd <> " from peer " <> s+ forwardMessage p msg+managerMessage (ManagerBestBlock h) = putBestBlock h+managerMessage ManagerConnect = do+ l <- length <$> getConnectedPeers+ x <- mgrConfMaxPeers <$> asks myConfig+ when (l < x) $+ getNewPeer >>= \case+ Nothing -> return ()+ Just sa -> connectPeer sa+managerMessage (ManagerKill e p) = killPeer e p+managerMessage (ManagerPeerDied a e) = processPeerOffline a e+managerMessage ManagerPurgePeers = do+ $(logWarnS) "Manager" "Purging connected peers and peer database"+ purgePeers+managerMessage (ManagerGetPeers reply) =+ getConnectedPeers >>= atomically . reply+managerMessage (ManagerGetOnlinePeer p reply) = do+ b <- asks onlinePeers+ atomically $ findPeer b p >>= reply+managerMessage (ManagerCheckPeer p) = checkPeer p -banPeer :: MonadManager n m => SockAddr -> m ()-banPeer sa = do- db <- asks myPeerDB- let k = PeerAddress sa- retrieve db Nothing k >>= \case- Nothing -> error "Cannot find peer to be banned"- Just v -> do- now <- computeTime- let v' =- v- { getPeerBanned = now- , getPeerNextConnect = now + 6 * 60 * 60- }- when (getPeerPrio v == PriorityNetwork) $ do- $(logWarnS) "Manager" $ "Banning peer " <> cs (show sa)- writeBatch db [deleteOp v, insertOp k v', insertOp v' k]+checkPeer :: MonadManager m => Peer -> m ()+checkPeer p = do+ ManagerConfig {mgrConfTimeout = to} <- asks myConfig+ b <- asks onlinePeers+ s <- atomically $ peerString b p+ atomically (lastPing b p) >>= \case+ Nothing -> pingPeer p+ Just t -> do+ now <- liftIO getCurrentTime+ if diffUTCTime now t > fromIntegral to+ then do+ $(logErrorS) "Manager" $+ "Peer " <> s <> " did not respond ping on time"+ killPeer PeerTimeout p+ else $(logDebugS) "Manager" $ "peer " <> s <> " awaiting pong" -backoffPeer :: MonadManager n m => SockAddr -> m ()-backoffPeer sa = do- db <- asks myPeerDB- onlinePeers <- map onlinePeerAddress <$> getOnlinePeers- let k = PeerAddress sa- retrieve db Nothing k >>= \case- Nothing -> error "Cannot find peer to backoff in database"- Just v -> do- now <- computeTime- r <-- liftIO . randomRIO $- if null onlinePeers- then (90, 300) -- Don't backoff so much if possibly offline- else (900, 1800)- let t = max (now + r) (getPeerNextConnect v)- v' = v {getPeerNextConnect = t}- when (getPeerPrio v == PriorityNetwork) $ do+pingPeer :: MonadManager m => Peer -> m ()+pingPeer p = do+ b <- asks onlinePeers+ s <- atomically $ peerString b p+ atomically (findPeer b p) >>= \case+ Nothing -> $(logErrorS) "Manager" $ "Will not ping unknown peer " <> s+ Just o+ | onlinePeerConnected o -> do+ n <- liftIO randomIO+ now <- liftIO getCurrentTime+ atomically (setPeerPing b n now p)+ $(logDebugS) "Manager" $+ "Sending ping " <> cs (show n) <> " to peer " <> s+ MPing (Ping n) `sendMessage` p+ | otherwise -> $(logWarnS) "Manager" $- "Backing off peer " <> cs (show sa) <> " for " <>- cs (show r) <>- " seconds"- writeBatch db [deleteOp v, insertOp k v', insertOp v' k]--getNewPeer :: (MonadUnliftIO m, MonadManager n m) => m (Maybe SockAddr)-getNewPeer = do- ManagerConfig {..} <- asks myConfig- online_peers <- map onlinePeerAddress <$> getOnlinePeers- config_peers <- concat <$> mapM toSockAddr mgrConfPeers- if mgrConfDiscover- then do- db <- asks myPeerDB- now <- computeTime- runResourceT . runConduit $- matching db Nothing PeerTimeAddressBase .|- CC.filter ((<= now) . getPeerNextConnect . fst) .|- CC.map (getPeerAddress . snd) .|- CC.find (not . (`elem` online_peers))- else return $ find (not . (`elem` online_peers)) config_peers+ "Will not ping peer " <> s <> " until handshake complete" -getConnectedPeers :: MonadManager n m => m [OnlinePeer]-getConnectedPeers = filter onlinePeerConnected <$> getOnlinePeers+killPeer :: MonadManager m => PeerException -> Peer -> m ()+killPeer e p = void . runMaybeT $ do+ b <- asks onlinePeers+ o <- MaybeT . atomically $ findPeer b p+ s <- atomically $ peerString b p+ $(logErrorS) "Manager" $ "Killing peer " <> s <> ": " <> cs (show e)+ onlinePeerAsync o `cancelWith` e -withConnectLoop :: (MonadUnliftIO m, MonadLoggerIO m) => Manager -> m a -> m a-withConnectLoop mgr f = withAsync go $ const f+processPeerOffline :: MonadManager m => Child -> Maybe SomeException -> m ()+processPeerOffline a e = do+ b <- asks onlinePeers+ atomically (findPeerAsync b a) >>= \case+ Nothing -> log_unknown e+ Just o -> do+ let p = onlinePeerMailbox o+ d = onlinePeerAddress o+ s <- atomically $ peerString b p+ if onlinePeerConnected o+ then do+ log_disconnected s e+ managerEvent $ PeerDisconnected p d+ else log_not_connect s e+ atomically $ removePeer b p+ db <- getManagerDB+ demotePeerDB db (onlinePeerAddress o)+ logConnectedPeers where- go =- forever $ do- ManagerPing `send` mgr- i <- liftIO (randomRIO (30, 90))- threadDelay (i * 1000 * 1000)--managerLoop :: (MonadUnliftIO m, MonadManager n m) => m ()-managerLoop =- forever $ do- mgr <- asks mySelf- msg <- receive mgr- processManagerMessage msg--processManagerMessage ::- (MonadUnliftIO m, MonadManager n m) => ManagerMessage -> m ()--processManagerMessage (ManagerSetFilter bf) = setFilter bf--processManagerMessage (ManagerSetBest bb) =- asks myBestBlock >>= atomically . (`writeTVar` bb)--processManagerMessage ManagerPing = connectNewPeers--processManagerMessage (ManagerGetAddr p) = do- pn <- peerString p- -- TODO: send list of peers we know about- $(logWarnS) "Manager" $ "Ignoring address request from peer " <> fromString pn--processManagerMessage (ManagerNewPeers p as) =- asks myConfig >>= \case- ManagerConfig {..}- | not mgrConfDiscover -> return ()- | otherwise -> do- pn <- peerString p- $(logInfoS) "Manager" $- "Received " <> cs (show (length as)) <>- " peers from " <>- fromString pn- forM_ as $ \(_, na) ->- let sa = naAddress na- in storePeer sa PriorityNetwork+ log_unknown Nothing = $(logErrorS) "Manager" "Disconnected unknown peer"+ log_unknown (Just x) =+ $(logErrorS) "Manager" $ "Unknown peer died: " <> cs (show x)+ log_disconnected s Nothing =+ $(logWarnS) "Manager" $ "Disconnected peer: " <> s+ log_disconnected s (Just x) =+ $(logErrorS) "Manager" $ "Peer " <> s <> " died: " <> cs (show x)+ log_not_connect s Nothing =+ $(logWarnS) "Manager" $ "Could not connect to peer " <> s+ log_not_connect s (Just x) =+ $(logErrorS) "Manager" $+ "Could not connect to peer " <> s <> ": " <> cs (show x) -processManagerMessage (ManagerKill e p) =- findPeer p >>= \case- Nothing -> return ()- Just op -> do+announcePeer :: MonadManager m => Peer -> m ()+announcePeer p = do+ b <- asks onlinePeers+ s <- atomically $ peerString b p+ mgr <- asks myMailbox+ atomically (findPeer b p) >>= \case+ Just OnlinePeer {onlinePeerAddress = a, onlinePeerConnected = True} -> do+ $(logInfoS) "Manager" $ "Handshake completed for peer " <> s+ managerEvent $ PeerConnected p a+ logConnectedPeers+ managerCheck p mgr+ db <- getManagerDB+ promotePeerDB db a+ Just OnlinePeer {onlinePeerConnected = False} -> $(logErrorS) "Manager" $- "Killing peer " <> cs (show (onlinePeerAddress op))- banPeer $ onlinePeerAddress op- onlinePeerAsync op `cancelWith` e--processManagerMessage (ManagerSetPeerBest p bn) = modifyPeer f p- where- f op = op {onlinePeerBestBlock = bn}--processManagerMessage (ManagerGetPeerBest p reply) =- findPeer p >>= \op -> atomically (reply (fmap onlinePeerBestBlock op))--processManagerMessage (ManagerSetPeerVersion p v) =- modifyPeer f p >> findPeer p >>= \case- Nothing -> return ()- Just op ->- runExceptT testVersion >>= \case- Left ex -> do- banPeer (onlinePeerAddress op)- onlinePeerAsync op `cancelWith` ex- Right () -> do- loadFilter- askForPeers- connectPeer (onlinePeerAddress op)- announcePeer- where- f op =- op- { onlinePeerVersion = version v- , onlinePeerServices = services v- , onlinePeerRemoteNonce = verNonce v- , onlinePeerUserAgent = getVarString (userAgent v)- , onlinePeerRelay = relay v- }- testVersion = do- when (services v .&. nodeNetwork == 0) $ throwError NotNetworkPeer- bfb <- asks myBloomFilter- bf <- readTVarIO bfb- when (isJust bf && services v .&. nodeBloom == 0) $- throwError BloomFiltersNotSupported- myself <-- any ((verNonce v ==) . onlinePeerNonce) <$> lift getOnlinePeers- when myself $ throwError PeerIsMyself- loadFilter = do- bfb <- asks myBloomFilter- bf <- readTVarIO bfb- case bf of- Nothing -> return ()- Just b -> b `peerSetFilter` p- askForPeers =- mgrConfDiscover <$> asks myConfig >>= \discover ->- when discover (MGetAddr `sendMessage` p)- announcePeer =- findPeer p >>= \case- Nothing -> return ()- Just op- | onlinePeerConnected op -> return ()- | otherwise -> do- $(logInfoS) "Manager" $- "Connected to " <> cs (show (onlinePeerAddress op))- l <- mgrConfMgrListener <$> asks myConfig- atomically (l (ManagerConnect p))- ch <- asks myChain- chainNewPeer p ch- setPeerAnnounced p--processManagerMessage (ManagerGetPeerVersion p reply) =- fmap onlinePeerVersion <$> findPeer p >>= atomically . reply--processManagerMessage (ManagerGetPeers reply) =- getPeers >>= atomically . reply--processManagerMessage (ManagerGetOnlinePeer p reply) =- getOnlinePeer p >>= atomically . reply--processManagerMessage (ManagerPeerPing p i) =- modifyPeer (\x -> x {onlinePeerPings = take 11 $ i : onlinePeerPings x}) p+ "Not announcing because not handshaken: " <> s+ Nothing -> $(logErrorS) "Manager" "Will not announce unknown peer" -processManagerMessage (PeerStopped (p, _ex)) = do- opb <- asks onlinePeers- m <- atomically $ do- m <- findPeerAsync p opb- when (isJust m) $ removePeer p opb- return m+getNewPeer :: (MonadUnliftIO m, MonadManager m) => m (Maybe SockAddr)+getNewPeer = do+ ManagerConfig {mgrConfDB = db} <- asks myConfig+ exclude <- map onlinePeerAddress <$> getOnlinePeers+ m <-+ runMaybeT $+ MaybeT (getNewPeerDB db exclude) <|>+ MaybeT (populatePeerDB >> getNewPeerDB db exclude) case m of- Just op -> do- backoffPeer (onlinePeerAddress op)- processPeerOffline op- Nothing -> return ()--processPeerOffline :: MonadManager n m => OnlinePeer -> m ()-processPeerOffline op- | onlinePeerConnected op = do- let p = onlinePeerMailbox op- $(logWarnS) "Manager" $- "Disconnected peer " <> cs (show (onlinePeerAddress op))- asks myChain >>= chainRemovePeer p- l <- mgrConfMgrListener <$> asks myConfig- atomically (l (ManagerDisconnect p))- logPeersConnected- | otherwise =- $(logWarnS) "Manager" $- "Could not connect to peer " <> cs (show (onlinePeerAddress op))--getPeers :: MonadManager n m => m [OnlinePeer]-getPeers = sortBy (compare `on` median . onlinePeerPings) <$> getConnectedPeers--getOnlinePeer :: MonadManager n m => Peer -> m (Maybe OnlinePeer)-getOnlinePeer p = find ((== p) . onlinePeerMailbox) <$> getConnectedPeers+ Just a -> return $ Just a+ Nothing -> return Nothing -connectNewPeers :: MonadManager n m => m ()-connectNewPeers = do- mo <- mgrConfMaxPeers <$> asks myConfig- ps <- getOnlinePeers- let n = mo - length ps- when (null ps) $ do- ps' <- resolvePeers- mapM_ (uncurry storePeer) ps'- go n+connectPeer :: (MonadUnliftIO m, MonadManager m) => SockAddr -> m ()+connectPeer sa = do+ $(logInfoS) "Manager" $ "Connecting to peer: " <> cs (show sa)+ ManagerConfig {mgrConfNetAddr = ad, mgrConfNetwork = net} <- asks myConfig+ mgr <- asks myMailbox+ sup <- asks mySupervisor+ nonce <- liftIO randomIO+ bb <- getBestBlock+ now <- computeTime+ let rmt = NetworkAddress (srv net) sa+ ver = buildVersion net nonce bb ad rmt now+ (inbox, p) <- newMailbox+ let pc pub =+ PeerConfig+ { peerConfListen = pub+ , peerConfNetwork = net+ , peerConfAddress = sa+ }+ a <- withRunInIO $ \io -> sup `addChild` io (launch mgr pc inbox p)+ MVersion ver `sendMessage` p+ b <- asks onlinePeers+ _ <- atomically $ newOnlinePeer b sa nonce p a+ return () where- go 0 = return ()- go n =- getNewPeer >>= \case- Nothing -> return ()- Just sa -> conn sa >> go (n - 1)- conn sa = do- ad <- mgrConfNetAddr <$> asks myConfig- mgr <- asks mySelf- ch <- asks myChain- pl <- mgrConfPeerListener <$> asks myConfig- net <- mgrConfNetwork <$> asks myConfig- $(logInfoS) "Manager" $ "Connecting to peer " <> cs (show sa)- nonce <- liftIO randomIO- let pc =- PeerConfig- { peerConfConnect = NetworkAddress (srv net) sa- , peerConfLocal = ad- , peerConfManager = mgr- , peerConfChain = ch- , peerConfListener = pl- , peerConfNonce = nonce- , peerConfNetwork = net- }- psup <- asks myPeerSupervisor- pmbox <- newTBQueueIO 100- p <- newInbox pmbox- a <- psup `addChild` peer pc p- newPeerConnection net sa nonce p a+ l mgr p m = ManagerPeerMessage p m `sendSTM` mgr srv net | getSegWit net = 8 | otherwise = 0--newPeerConnection ::- MonadManager n m- => Network- -> SockAddr- -> Word64- -> Peer- -> Async ()- -> m ()-newPeerConnection net sa nonce p a =- addPeer- OnlinePeer- { onlinePeerAddress = sa- , onlinePeerConnected = False- , onlinePeerVersion = 0- , onlinePeerServices = 0- , onlinePeerRemoteNonce = 0- , onlinePeerUserAgent = BS.empty- , onlinePeerRelay = False- , onlinePeerBestBlock = genesisNode net- , onlinePeerAsync = a- , onlinePeerMailbox = p- , onlinePeerNonce = nonce- , onlinePeerPings = []- }--peerString :: MonadManager n m => Peer -> m String-peerString p = maybe "[unknown]" (show . onlinePeerAddress) <$> findPeer p--setPeerAnnounced :: MonadManager n m => Peer -> m ()-setPeerAnnounced = modifyPeer (\x -> x {onlinePeerConnected = True})--setFilter :: MonadManager n m => BloomFilter -> m ()-setFilter bl = do- bfb <- asks myBloomFilter- atomically . writeTVar bfb $ Just bl- ops <- getOnlinePeers- forM_ ops $ \op ->- when (onlinePeerConnected op) $- if acceptsFilters $ onlinePeerServices op- then bl `peerSetFilter` onlinePeerMailbox op- else do- $(logErrorS) "Manager" $- "Peer " <> cs (show (onlinePeerAddress op)) <>- "does not support bloom filters"- banPeer (onlinePeerAddress op)- onlinePeerAsync op `cancelWith` BloomFiltersNotSupported--findPeer :: MonadManager n m => Peer -> m (Maybe OnlinePeer)-findPeer p = find ((== p) . onlinePeerMailbox) <$> getOnlinePeers--findPeerAsync :: Async () -> TVar [OnlinePeer] -> STM (Maybe OnlinePeer)-findPeerAsync a t = find ((== a) . onlinePeerAsync) <$> readTVar t+ launch mgr pc inbox p =+ withPublisher $ \pub ->+ bracket (subscribe pub (l mgr p)) (unsubscribe pub) $ \_ ->+ withPeerLoop p mgr $ \a -> do+ link a+ peer (pc pub) inbox -modifyPeer :: MonadManager n m => (OnlinePeer -> OnlinePeer) -> Peer -> m ()-modifyPeer f p = modifyOnlinePeers $ map upd+withPeerLoop :: MonadUnliftIO m => Peer -> Manager -> (Async a -> m a) -> m a+withPeerLoop p mgr = withAsync go where- upd op =- if onlinePeerMailbox op == p- then f op- else op+ go = forever $ do+ threadDelay =<< liftIO (randomRIO (30 * 1000 * 1000, 90 * 1000 * 1000))+ ManagerCheckPeer p `send` mgr -addPeer :: MonadManager n m => OnlinePeer -> m ()-addPeer op = modifyOnlinePeers $ nubBy f . (op :)+withConnectLoop :: MonadUnliftIO m => Manager -> (Async a -> m a) -> m a+withConnectLoop mgr = withAsync go where- f = (==) `on` onlinePeerMailbox--removePeer :: Async () -> TVar [OnlinePeer] -> STM ()-removePeer a t = modifyTVar t $ filter ((/= a) . onlinePeerAsync)--getOnlinePeers :: MonadManager n m => m [OnlinePeer]-getOnlinePeers = asks onlinePeers >>= readTVarIO--modifyOnlinePeers :: MonadManager n m => ([OnlinePeer] -> [OnlinePeer]) -> m ()-modifyOnlinePeers f = asks onlinePeers >>= atomically . (`modifyTVar` f)--median :: Fractional a => [a] -> Maybe a-median ls- | null ls = Nothing- | length ls `mod` 2 == 0 =- Just . (/ 2) . sum . take 2 $ drop (length ls `div` 2 - 1) ls- | otherwise = Just . head $ drop (length ls `div` 2) ls+ go = forever $ do+ ManagerConnect `send` mgr+ threadDelay =<< liftIO (randomRIO (250 * 1000, 1000 * 1000))
+ src/Network/Haskoin/Node/Manager/Logic.hs view
@@ -0,0 +1,406 @@+{-# 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
src/Network/Haskoin/Node/Peer.hs view
@@ -1,341 +1,106 @@ {-# LANGUAGE ConstraintKinds #-} {-# LANGUAGE FlexibleContexts #-}-{-# LANGUAGE LambdaCase #-} {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-} {-# LANGUAGE TemplateHaskell #-}-{-# LANGUAGE TupleSections #-} {-# LANGUAGE TypeFamilies #-}+{-|+Module : Network.Haskoin.Node.Peer+Copyright : No rights reserved+License : UNLICENSE+Maintainer : xenog@protonmail.com+Stability : experimental+Portability : POSIX++Network peer process. Represents a network peer connection locally.+-} module Network.Haskoin.Node.Peer ( peer ) where +import Conduit import Control.Monad import Control.Monad.Logger import Control.Monad.Reader-import Data.Bits import Data.ByteString (ByteString)-import qualified Data.ByteString as BS-import qualified Data.ByteString.Char8 as C8-import qualified Data.ByteString.Lazy as BL-import Data.Conduit-import qualified Data.Conduit.Binary as CB+import qualified Data.ByteString as B import Data.Conduit.Network-import Data.Maybe import Data.Serialize-import Data.String import Data.String.Conversions-import Data.Time.Clock-import Data.Word-import Network.Haskoin.Block+import Data.Text (Text) import Network.Haskoin.Constants import Network.Haskoin.Network import Network.Haskoin.Node.Common-import Network.Haskoin.Transaction import Network.Socket (SockAddr) import NQE-import System.Random import UnliftIO -type MonadPeer m = (MonadUnliftIO m, MonadLoggerIO m, MonadReader PeerReader m)--data Pending- = PendingTx !TxHash- | PendingBlock !BlockHash- | PendingMerkle !BlockHash- | PendingHeaders- deriving (Show, Eq)--data PeerReader = PeerReader- { mySelf :: !Peer- , myConfig :: !PeerConfig- , mySockAddr :: !SockAddr- , myHostPort :: !(Host, Port)- , myPending :: !(TVar [(Pending, Word32)])- }--time :: Int-time = 15 * 1000 * 1000--logMsg :: IsString a => Message -> a-logMsg = fromString . cs . commandToString . msgType--logPeer ::- (ConvertibleStrings String a, Semigroup a, IsString a) => SockAddr -> a-logPeer sa = "Peer<" <> cs (show sa) <> ">"--peer :: (MonadUnliftIO m, MonadLoggerIO m) => PeerConfig -> Peer -> m ()-peer pc p =- fromSockAddr na >>= \case- Nothing -> do- $(logErrorS) (logPeer na) "Invalid network address"- throwIO PeerAddressInvalid- Just (host, port) -> do- let cset = clientSettings port (C8.pack host)- runGeneralTCPClient cset (peerSession (host, port))+-- | Run peer process in current thread.+peer ::+ (MonadUnliftIO m, MonadLoggerIO m)+ => PeerConfig+ -> Inbox PeerMessage+ -> m ()+peer pc inbox = withConnection a $ \ad -> runReaderT (peer_session ad) pc where- na = naAddress (peerConfConnect pc)- go = handshake >> exchangePing >> peerLoop+ a = peerConfAddress pc+ go = forever $ receive inbox >>= dispatchMessage pc net = peerConfNetwork pc- peerSession hp ad = do- let src =- runConduit $- appSource ad .| inPeerConduit net .| conduitMailbox p- snk = outPeerConduit net .| appSink ad- withAsync src $ \as -> do- link as- pbox <- newTVarIO []- let rd =- PeerReader- { myConfig = pc- , mySelf = p- , myHostPort = hp- , mySockAddr = na- , myPending = pbox- }- runReaderT (runConduit (go .| snk)) rd--handshake :: MonadPeer m => ConduitT () Message m ()-handshake = do- p <- asks mySelf- ch <- peerConfChain <$> asks myConfig- rmt <- peerConfConnect <$> asks myConfig- loc <- peerConfLocal <$> asks myConfig- net <- peerConfNetwork <$> asks myConfig- nonce <- peerConfNonce <$> asks myConfig- bb <- chainGetBest ch- ver <- buildVersion net nonce (nodeHeight bb) loc rmt- yield $ MVersion ver- lift (remoteVer p) >>= \case- v- | testSegWit net v -> do- yield MVerAck- lift (remoteVerAck p)- mgr <- peerConfManager <$> asks myConfig- managerSetPeerVersion p v mgr- | otherwise -> do- yield . MReject $- reject MCVersion RejectObsolete "No SegWit support"- throwIO PeerNoSegWit- where- testSegWit net v- | getSegWit net = services v `testBit` 3- | otherwise = True- remoteVer p = do- m <-- timeout time . receiveMatch p $ \case- PeerIncoming (MVersion v) -> Just v- _ -> Nothing- case m of- Just v -> return v- Nothing -> throwIO PeerTimeout- remoteVerAck p = do- m <-- timeout time . receiveMatch p $ \case- PeerIncoming MVerAck -> Just ()- _ -> Nothing- when (isNothing m) $ throwIO PeerTimeout--peerLoop :: MonadPeer m => ConduitT () Message m ()-peerLoop =- forever $ do- me <- asks mySelf- m <- lift $ timeout (2 * 60 * 1000 * 1000) (receive me)- case m of- Nothing -> exchangePing- Just msg -> processMessage msg--exchangePing :: MonadPeer m => ConduitT () Message m ()-exchangePing = do- lp <- logMe- i <- liftIO randomIO- yield $ MPing (Ping i)- me <- asks mySelf- mgr <- peerConfManager <$> asks myConfig- t1 <- liftIO getCurrentTime- m <-- lift . timeout time . receiveMatch me $ \case- PeerIncoming (MPong (Pong j))- | i == j -> Just ()- _ -> Nothing- case m of- Nothing -> do- $(logErrorS) lp "Timeout while waiting for pong"- throwIO PeerTimeout- Just () -> do- t2 <- liftIO getCurrentTime- let d = t2 `diffUTCTime` t1- $(logDebugS) lp $- "Roundtrip: " <> cs (show (d * 1000)) <> " ms"- ManagerPeerPing me d `send` mgr--checkStale :: MonadPeer m => ConduitM () Message m ()-checkStale = do- pbox <- asks myPending- ps <- readTVarIO pbox- case ps of- [] -> return ()- (_, ts):_ -> do- cur <- computeTime- when (cur > ts + 30) $ throwIO PeerTimeout--registerOutgoing :: MonadPeer m => Message -> m ()-registerOutgoing (MGetData (GetData ivs)) = do- pbox <- asks myPending- cur <- computeTime- ms <-- fmap catMaybes . forM ivs $ \iv ->- case toPending iv of- Nothing -> return Nothing- Just p -> return $ Just (p, cur)- atomically (modifyTVar pbox (++ ms))- where- toPending InvVector {invType = InvTx, invHash = hash} =- Just (PendingTx (TxHash hash))- toPending InvVector {invType = InvWitnessTx, invHash = hash} =- Just (PendingTx (TxHash hash))- toPending InvVector {invType = InvBlock, invHash = hash} =- Just (PendingBlock (BlockHash hash))- toPending InvVector {invType = InvWitnessBlock, invHash = hash} =- Just (PendingBlock (BlockHash hash))- toPending InvVector {invType = InvMerkleBlock, invHash = hash} =- Just (PendingMerkle (BlockHash hash))- toPending InvVector {invType = InvWitnessMerkleBlock, invHash = hash} =- Just (PendingMerkle (BlockHash hash))- toPending _ = Nothing-registerOutgoing MGetHeaders {} = do- pbox <- asks myPending- cur <- computeTime- atomically (modifyTVar pbox (reverse . ((PendingHeaders, cur) :) . reverse))-registerOutgoing _ = return ()--registerIncoming :: MonadPeer m => Message -> m ()-registerIncoming (MNotFound (NotFound ivs)) =- asks myPending >>= \pbox ->- atomically (modifyTVar pbox (filter (matchNotFound . fst)))- where- matchNotFound (PendingTx (TxHash hash)) =- InvVector InvTx hash `notElem` ivs &&- InvVector InvWitnessTx hash `notElem` ivs- matchNotFound (PendingBlock (BlockHash hash)) =- InvVector InvBlock hash `notElem` ivs &&- InvVector InvWitnessBlock hash `notElem` ivs- matchNotFound (PendingMerkle (BlockHash hash)) =- InvVector InvBlock hash `notElem` ivs &&- InvVector InvMerkleBlock hash `notElem` ivs &&- InvVector InvWitnessMerkleBlock hash `notElem` ivs- matchNotFound _ = False-registerIncoming (MTx t) =- asks myPending >>= \pbox ->- atomically (modifyTVar pbox (filter ((/= PendingTx (txHash t)) . fst)))-registerIncoming (MBlock b) =- asks myPending >>= \pbox ->- atomically $- modifyTVar- pbox- (filter ((/= PendingBlock (headerHash (blockHeader b))) . fst))-registerIncoming (MMerkleBlock b) =- asks myPending >>= \pbox ->- atomically $- modifyTVar- pbox- (filter ((/= PendingMerkle (headerHash (merkleHeader b))) . fst))-registerIncoming MHeaders {} =- asks myPending >>= \pbox ->- atomically $ modifyTVar pbox (filter ((/= PendingHeaders) . fst))-registerIncoming _ = return ()--processMessage :: MonadPeer m => PeerMessage -> ConduitM () Message m ()-processMessage m = do- checkStale- case m of- PeerOutgoing msg -> do- lift (registerOutgoing msg)- yield msg- PeerIncoming msg -> do- lift (registerIncoming msg)- incoming msg--logMe ::- ( ConvertibleStrings String a- , Semigroup a- , IsString a- , MonadReader PeerReader m- )- => m a-logMe = logPeer <$> asks mySockAddr+ peer_session ad =+ let ins = appSource ad+ ons = appSink ad+ src = runConduit $ ins .| inPeerConduit net a .| mapM_C send_msg+ snk = outPeerConduit net .| ons+ in withAsync src $ \as -> do+ link as+ runConduit (go .| snk)+ send_msg = (`send` peerConfListen pc) . Event -incoming :: MonadPeer m => Message -> ConduitT () Message m ()-incoming m = do- lp <- lift logMe- p <- asks mySelf- l <- peerConfListener <$> asks myConfig- mgr <- peerConfManager <$> asks myConfig- ch <- peerConfChain <$> asks myConfig- case m of- MVersion _ -> do- $(logErrorS) lp $ "Received duplicate " <> logMsg m- yield $- MReject- Reject- { rejectMessage = MCVersion- , rejectCode = RejectDuplicate- , rejectReason = VarString BS.empty- , rejectData = BS.empty- }- MPing (Ping n) -> yield $ MPong (Pong n)- MPong (Pong n) -> atomically (l (p, GotPong n))- MSendHeaders {} -> ChainSendHeaders p `send` ch- MAlert {} -> $(logWarnS) lp $ "Deprecated " <> logMsg m- MAddr (Addr as) -> managerNewPeers p as mgr- MInv (Inv is) -> do- let ts = [TxHash (invHash i) | i <- is, invType i == InvTx]- bs =- [ BlockHash (invHash i)- | i <- is- , invType i == InvBlock || invType i == InvMerkleBlock- ]- unless (null ts) $ atomically $ l (p, TxAvail ts)- unless (null bs) $ ChainNewBlocks p bs `send` ch- MTx tx -> atomically (l (p, GotTx tx))- MBlock b -> atomically (l (p, GotBlock b))- MMerkleBlock b -> atomically (l (p, GotMerkleBlock b))- MHeaders (Headers hcs) -> ChainNewHeaders p hcs `send` ch- MGetData (GetData d) -> atomically (l (p, SendData d))- MNotFound (NotFound ns) -> do- let f (InvVector InvTx hash) = Just (TxNotFound (TxHash hash))- f (InvVector InvWitnessTx hash) =- Just (TxNotFound (TxHash hash))- f (InvVector InvBlock hash) =- Just (BlockNotFound (BlockHash hash))- f (InvVector InvWitnessBlock hash) =- Just (BlockNotFound (BlockHash hash))- f (InvVector InvMerkleBlock hash) =- Just (BlockNotFound (BlockHash hash))- f (InvVector InvWitnessMerkleBlock hash) =- Just (BlockNotFound (BlockHash hash))- f _ = Nothing- events = mapMaybe f ns- atomically (mapM_ (l . (p, )) events)- MGetBlocks g -> atomically (l (p, SendBlocks g))- MGetHeaders h -> atomically (l (p, SendHeaders h))- MReject r -> atomically (l (p, Rejected r))- MMempool -> atomically (l (p, WantMempool))- MGetAddr -> managerGetAddr p mgr- _ -> $(logWarnS) lp $ "Ignoring message: " <> logMsg m+-- | Internal function to dispatch peer messages.+dispatchMessage ::+ MonadLoggerIO m => PeerConfig -> PeerMessage -> ConduitT i Message m ()+dispatchMessage cfg (SendMessage msg) = do+ $(logDebugS) (peerString (peerConfAddress cfg)) $+ "Outgoing: " <> cs (commandToString (msgType msg))+ yield msg+dispatchMessage cfg (GetPublisher reply) =+ atomically $ reply (peerConfListen cfg) -inPeerConduit :: MonadIO m => Network -> ConduitT ByteString PeerMessage m ()-inPeerConduit net = do- headerBytes <- CB.take 24- case decodeLazy headerBytes of- Left e -> throwIO $ DecodeMessageError e- Right (MessageHeader _ _cmd len _) -> do- when (len > 32 * 2 ^ (20 :: Int)) . throwIO $ PayloadTooLarge len- payloadBytes <- CB.take (fromIntegral len)- case runGetLazy (getMessage net) $ headerBytes `BL.append` payloadBytes of- Left e -> throwIO $ CannotDecodePayload e- Right msg -> yield $ PeerIncoming msg- inPeerConduit net+-- | Internal conduit to parse messages coming from peer.+inPeerConduit ::+ MonadLoggerIO m+ => Network+ -> SockAddr+ -> ConduitT ByteString Message m ()+inPeerConduit net a = do+ x <- takeCE 24 .| foldC+ case decode x of+ Left _ -> do+ $(logErrorS)+ (peerString a)+ "Could not decode incoming message header"+ throwIO DecodeHeaderError+ Right (MessageHeader _ _ len _) -> do+ when (len > 32 * 2 ^ (20 :: Int)) $ do+ $(logErrorS) (peerString a) "Payload too large"+ throwIO $ PayloadTooLarge len+ y <- takeCE (fromIntegral len) .| foldC+ case runGet (getMessage net) $ x `B.append` y of+ Left e -> do+ $(logErrorS) (peerString a) $+ "Cannot decode payload: " <> cs (show e)+ throwIO CannotDecodePayload+ Right msg -> do+ $(logDebugS) (peerString a) $+ "Incoming: " <> cs (commandToString (msgType msg))+ yield msg+ inPeerConduit net a +-- | Outgoing peer conduit to serialize and send messages. outPeerConduit :: Monad m => Network -> ConduitT Message ByteString m () outPeerConduit net = awaitForever $ yield . runPut . putMessage net++-- | Peer string for logging+peerString :: SockAddr -> Text+peerString a = "Peer{" <> cs (show a) <> "}"
+ test/Haskoin/NodeSpec.hs view
@@ -0,0 +1,147 @@+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-}+module Haskoin.NodeSpec+ ( spec+ ) where++import Control.Monad+import Control.Monad.Logger+import Control.Monad.Trans+import qualified Database.RocksDB as R+import Haskoin+import Haskoin.Node+import Network.Socket (SockAddr (..))+import NQE+import Test.Hspec+import UnliftIO++data TestNode = TestNode+ { testMgr :: Manager+ , testChain :: Chain+ , nodeEvents :: Inbox NodeEvent+ }++spec :: Spec+spec = do+ let net = btcTest+ describe "peer manager on test network" $ do+ it "connects to a peer" $+ withTestNode net "connect-one-peer" $ \TestNode {..} -> do+ p <- waitForPeer nodeEvents+ Just OnlinePeer {onlinePeerVersion = Just Version {version = ver}} <-+ managerGetPeer p testMgr+ ver `shouldSatisfy` (>= 70002)+ it "downloads some blocks" $+ withTestNode net "get-blocks" $ \TestNode {..} -> do+ let h1 =+ "000000000babf10e26f6cba54d9c282983f1d1ce7061f7e875b58f8ca47db932"+ h2 =+ "00000000851f278a8b2c466717184aae859af5b83c6f850666afbc349cf61577"+ p <- waitForPeer nodeEvents+ Just [b1, b2] <- peerGetBlocks net 30 p [h1, h2]+ headerHash (blockHeader b1) `shouldBe` h1+ headerHash (blockHeader b2) `shouldBe` h2+ let testMerkle b =+ merkleRoot (blockHeader b) `shouldBe`+ buildMerkleRoot (map txHash (blockTxns b))+ testMerkle b1+ testMerkle b2+ it "connects to two peers" $+ withTestNode net "connect-peers" $ \TestNode {..} ->+ replicateM_ 2 (waitForPeer nodeEvents) `shouldReturn` ()+ it "attempts to get inexistent things" $+ withTestNode net "download-fail" $ \TestNode {..} -> do+ let hs =+ [ "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"+ , "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"+ ]+ p <- waitForPeer nodeEvents+ peerGetTxs net 30 p hs `shouldReturn` Nothing+ describe "chain on test network" $ do+ it "syncs some headers" $+ withTestNode net "connect-sync" $ \TestNode {..} -> do+ let h =+ "000000009ec921df4bb16aedd11567e27ede3c0b63835b257475d64a059f102b"+ hs =+ [ headerHash (getGenesisHeader net)+ , "0000000005bdbddb59a3cd33b69db94fa67669c41d9d32751512b5d7b68c71cf"+ , "00000000185b36fa6e406626a722793bea80531515e0b2a99ff05b73738901f1"+ , "000000001ab69b12b73ccdf46c9fbb4489e144b54f1565e42e481c8405077bdd"+ ]+ bns <-+ replicateM 4 . receiveMatch nodeEvents $ \case+ ChainEvent (ChainBestBlock bn) -> Just bn+ _ -> Nothing+ bb <- chainGetBest testChain+ nodeHeight bb `shouldSatisfy` (>= 6000)+ an <-+ maybe (throwString "No ancestor found") return =<<+ chainGetAncestor 2357 (last bns) testChain+ map (headerHash . nodeHeader) bns `shouldBe` hs+ headerHash (nodeHeader an) `shouldBe` h+ it "downloads some block parents" $+ withTestNode net "parents" $ \TestNode {..} -> do+ let hs =+ [ "00000000c74a24e1b1f2c04923c514ed88fc785cf68f52ed0ccffd3c6fe3fbd9"+ , "000000007e5c5f40e495186ac4122f2e4ee25788cc36984a5760c55ecb376cb1"+ , "00000000a6299059b2bff3479bc569019792e75f3c0f39b10a0bc85eac1b1615"+ ]+ [_, bn] <-+ replicateM 2 $+ receiveMatch nodeEvents $ \case+ ChainEvent (ChainBestBlock bn) -> Just bn+ _ -> Nothing+ nodeHeight bn `shouldBe` 2000+ ps <- chainGetParents 1997 bn testChain+ length ps `shouldBe` 3+ forM_ (zip ps hs) $ \(p, h) ->+ headerHash (nodeHeader p) `shouldBe` h++waitForPeer :: MonadIO m => Inbox NodeEvent -> m Peer+waitForPeer inbox =+ receiveMatch inbox $ \case+ PeerEvent (PeerConnected p _) -> Just p+ _ -> Nothing+++withTestNode ::+ MonadUnliftIO m+ => Network+ -> String+ -> (TestNode -> m a)+ -> m a+withTestNode net str f =+ runNoLoggingT $+ withSystemTempDirectory ("haskoin-node-test-" <> str <> "-") $ \w ->+ withPublisher $ \pub ->+ withSubscription pub $ \node_inbox -> do+ db <-+ R.open+ w+ R.defaultOptions+ { R.createIfMissing = True+ , R.compression = R.SnappyCompression+ }+ let cfg =+ NodeConfig+ { nodeConfMaxPeers = 20+ , nodeConfDB = db+ , nodeConfPeers = []+ , nodeConfDiscover = True+ , nodeConfNetAddr =+ NetworkAddress 0 (SockAddrInet 0 0)+ , nodeConfNet = net+ , nodeConfEvents = pub+ , nodeConfTimeout = 10+ }+ withNode cfg $ \(mgr, ch) ->+ lift $+ f+ TestNode+ { testMgr = mgr+ , testChain = ch+ , nodeEvents = node_inbox+ }
test/Spec.hs view
@@ -1,219 +1,1 @@-{-# LANGUAGE FlexibleContexts #-}-{-# LANGUAGE LambdaCase #-}-{-# LANGUAGE MultiParamTypeClasses #-}-{-# LANGUAGE OverloadedStrings #-}-{-# LANGUAGE RecordWildCards #-}-import Control.Monad-import Control.Monad.Logger-import Control.Monad.Trans-import qualified Data.ByteString as BS-import Data.Either-import Data.Maybe-import Data.Serialize-import qualified Database.RocksDB as RocksDB-import Haskoin-import Haskoin.Node-import Network.Socket (SockAddr (..))-import NQE-import System.Random-import Test.Hspec-import UnliftIO--data TestNode = TestNode- { testMgr :: Manager- , testChain :: Chain- , testEvents :: Inbox NodeEvent- }--main :: IO ()-main = do- let net = btcTest- hspec . describe "peer-to-peer client" $ do- it "connects to a peer" $- withTestNode net "connect-one-peer" $ \TestNode {..} -> do- p <-- receiveMatch testEvents $ \case- ManagerEvent (ManagerConnect p) -> Just p- _ -> Nothing- v <-- fromMaybe (error "No version") <$>- managerGetPeerVersion p testMgr- v `shouldSatisfy` (>= 70002)- bb <-- fromMaybe (error "No best block") <$>- managerGetPeerBest p testMgr- bb `shouldBe` genesisNode net- it "downloads some blocks" $- withTestNode net "get-blocks" $ \TestNode {..} -> do- let hs = [h1, h2]- h1 =- "000000000babf10e26f6cba54d9c282983f1d1ce7061f7e875b58f8ca47db932"- h2 =- "00000000851f278a8b2c466717184aae859af5b83c6f850666afbc349cf61577"- p <-- receiveMatch testEvents $ \case- ManagerEvent (ManagerConnect p) -> Just p- _ -> Nothing- peerGetBlocks net p hs- b1 <-- receiveMatch testEvents $ \case- PeerEvent (p', GotBlock b)- | p == p' -> Just b- _ -> Nothing- b2 <-- receiveMatch testEvents $ \case- PeerEvent (p', GotBlock b)- | p == p' -> Just b- _ -> Nothing- headerHash (blockHeader b1) `shouldSatisfy` (`elem` hs)- headerHash (blockHeader b2) `shouldSatisfy` (`elem` hs)- let testMerkle b =- merkleRoot (blockHeader b) `shouldBe`- buildMerkleRoot (map txHash (blockTxns b))- testMerkle b1- testMerkle b2- it "downloads some merkle blocks" $- withTestNode net "get-merkle-blocks" $ \TestNode {..} -> do- let a =- fromJust $- stringToAddr net "mgpS4Zis8iwNhriKMro1QSGDAbY6pqzRtA"- k :: PubKey- k =- "02c3cface1777c70251cb206f7c80cabeae195dfbeeff0767cbd2a58d22be383da"- h1 =- "000000006cf9d53d65522002a01d8c7091c78d644106832bc3da0b7644f94d36"- h2 =- "000000000babf10e26f6cba54d9c282983f1d1ce7061f7e875b58f8ca47db932"- bhs = [h1, h2]- n <- randomIO- let f0 = bloomCreate 2 0.001 n BloomUpdateAll- f1 = bloomInsert f0 $ exportPubKey True k- f2 = bloomInsert f1 $ encode $ getAddrHash160 a- f2 `setManagerFilter` testMgr- p <-- receiveMatch testEvents $ \case- ManagerEvent (ManagerConnect p) -> Just p- _ -> Nothing- getMerkleBlocks p bhs- b1 <-- receiveMatch testEvents $ \case- PeerEvent (p', GotMerkleBlock b)- | p == p' -> Just b- _ -> Nothing- b2 <-- receiveMatch testEvents $ \case- PeerEvent (p', GotMerkleBlock b)- | p == p' -> Just b- _ -> Nothing- liftIO $ do- a `shouldBe` pubKeyAddr net (wrapPubKey True k)- b1 `shouldSatisfy` testMerkleRoot net- b2 `shouldSatisfy` testMerkleRoot net- it "connects to multiple peers" $- withTestNode net "connect-peers" $ \TestNode {..} -> do- replicateM_ 3 $ do- pc <- receive testEvents- case pc of- ManagerEvent (ManagerDisconnect _) ->- expectationFailure "Received peer disconnection"- _ -> return ()- ps <- managerGetPeers testMgr- length ps `shouldSatisfy` (>= 2)- it "connects and syncs some headers" $- withTestNode net "connect-sync" $ \TestNode {..} -> do- let h =- "000000009ec921df4bb16aedd11567e27ede3c0b63835b257475d64a059f102b"- hs =- [ "0000000005bdbddb59a3cd33b69db94fa67669c41d9d32751512b5d7b68c71cf"- , "00000000185b36fa6e406626a722793bea80531515e0b2a99ff05b73738901f1"- , "000000001ab69b12b73ccdf46c9fbb4489e144b54f1565e42e481c8405077bdd"- ]- bns <-- replicateM 3 . receiveMatch testEvents $ \case- ChainEvent (ChainNewBest bn) -> Just bn- _ -> Nothing- bb <- chainGetBest testChain- an <-- fromMaybe (error "No ancestor found") <$>- chainGetAncestor 2357 (last bns) testChain- map (headerHash . nodeHeader) bns `shouldBe` hs- nodeHeight bb `shouldSatisfy` (>= 6000)- headerHash (nodeHeader an) `shouldBe` h- it "downloads a single block" $- withTestNode net "download-block" $ \TestNode {..} -> do- let h =- "000000009ec921df4bb16aedd11567e27ede3c0b63835b257475d64a059f102b"- p <-- receiveMatch testEvents $ \case- ManagerEvent (ManagerConnect p) -> Just p- _ -> Nothing- peerGetBlocks net p [h]- b <-- receiveMatch testEvents $ \case- PeerEvent (p', GotBlock b)- | p == p' -> Just b- _ -> Nothing- headerHash (blockHeader b) `shouldBe` h- it "attempts to get inexistent things" $- withTestNode net "download-fail" $ \TestNode {..} -> do- let h =- TxHash .- fromRight (error "We will, we will rock you!") . decode $- BS.replicate 32 0xaa- p <-- receiveMatch testEvents $ \case- ManagerEvent (ManagerConnect p) -> Just p- _ -> Nothing- peerGetTxs net p [h]- n <-- receiveMatch testEvents $ \case- PeerEvent (p', TxNotFound n)- | p == p' -> Just n- _ -> Nothing- n `shouldBe` h- it "downloads some block parents" $- withTestNode net "parents" $ \TestNode {..} -> do- let hs =- [ "00000000c74a24e1b1f2c04923c514ed88fc785cf68f52ed0ccffd3c6fe3fbd9"- , "000000007e5c5f40e495186ac4122f2e4ee25788cc36984a5760c55ecb376cb1"- , "00000000a6299059b2bff3479bc569019792e75f3c0f39b10a0bc85eac1b1615"- ]- bn <-- receiveMatch testEvents $ \case- ChainEvent (ChainNewBest bn) -> Just bn- _ -> Nothing- nodeHeight bn `shouldBe` 2000- ps <- chainGetParents 1997 bn testChain- length ps `shouldBe` 3- forM_ (zip ps hs) $ \(p, h) ->- headerHash (nodeHeader p) `shouldBe` h--withTestNode ::- (MonadUnliftIO m)- => Network- -> String- -> (TestNode -> m ())- -> m ()-withTestNode net t f =- runNoLoggingT . withSystemTempDirectory ("haskoin-node-test-" <> t <> "-") $ \w -> do- events <- newInbox =<< newTQueueIO- db <-- RocksDB.open- w- RocksDB.defaultOptions- { RocksDB.createIfMissing = True- , RocksDB.compression = RocksDB.SnappyCompression- }- let cfg =- NodeConfig- { maxPeers = 20- , database = db- , initPeers = []- , discover = True- , nodeEvents = (`sendSTM` events)- , netAddress = NetworkAddress 0 (SockAddrInet 0 0)- , nodeNet = net- }- withNode cfg $ \(mgr, ch) ->- lift $- f TestNode {testMgr = mgr, testChain = ch, testEvents = events}+{-# OPTIONS_GHC -F -pgmF hspec-discover #-}