haskoin-node 0.13.0 → 0.14.0
raw patch · 11 files changed
+1522/−1491 lines, 11 filesdep ~haskoin-coredep ~nqe
Dependency ranges changed: haskoin-core, nqe
Files
- CHANGELOG.md +10/−0
- LICENSE +19/−0
- README.md +2/−1
- UNLICENSE +0/−24
- haskoin-node.cabal +11/−12
- src/Haskoin/Node.hs +166/−144
- src/Haskoin/Node/Chain.hs +430/−273
- src/Haskoin/Node/Common.hs +0/−573
- src/Haskoin/Node/Manager.hs +528/−337
- src/Haskoin/Node/Peer.hs +285/−56
- test/Haskoin/NodeSpec.hs +71/−71
CHANGELOG.md view
@@ -4,6 +4,16 @@ 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.14.0+### Changed+- Massively refactor everything in a non-backwards-compatible way.+- Use MIT license.+- Bump haskoin-core.+- Bump secp256k1-haskell.++### Fixed+- Fix getting stuck on a single peer.+ ## 0.13.0 ### Changed - Depend on Haskoin Store 0.13.3.
+ LICENSE view
@@ -0,0 +1,19 @@+Copyright 2020 Haskoin Developers++Permission is hereby granted, free of charge, to any person obtaining a copy of+this software and associated documentation files (the "Software"), to deal in+the Software without restriction, including without limitation the rights to+use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of+the Software, and to permit persons to whom the Software is furnished to do so,+subject to the following conditions:++The above copyright notice and this permission notice shall be included in all+copies or substantial portions of the Software.++THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS+FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR+COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER+IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN+CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.+
README.md view
@@ -1,4 +1,5 @@ # Haskoin Node -Haskoin Node is a peer-to-peer library for Bitcoin and Bitcoin Cash. It uses a RocksDB database to store blockchain headers and peers.+Haskoin Node is a peer-to-peer library for Bitcoin and Bitcoin Cash. It uses a+RocksDB database to store blockchain headers and peers.
− UNLICENSE
@@ -1,24 +0,0 @@-This is free and unencumbered software released into the public domain.--Anyone is free to copy, modify, publish, use, compile, sell, or-distribute this software, either in source code form or as a compiled-binary, for any purpose, commercial or non-commercial, and by any-means.--In jurisdictions that recognize copyright laws, the author or authors-of this software dedicate any and all copyright interest in the-software to the public domain. We make this dedication for the benefit-of the public at large and to the detriment of our heirs and-successors. We intend this dedication to be an overt act of-relinquishment in perpetuity of all present and future rights to this-software under copyright law.--THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,-EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF-MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.-IN NO EVENT SHALL THE AUTHORS BE LIABLE FOR ANY CLAIM, DAMAGES OR-OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE,-ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR-OTHER DEALINGS IN THE SOFTWARE.--For more information, please refer to <http://unlicense.org/>
haskoin-node.cabal view
@@ -1,22 +1,22 @@ cabal-version: 1.12 --- This file has been generated from package.yaml by hpack version 0.31.2.+-- This file has been generated from package.yaml by hpack version 0.33.0. -- -- see: https://github.com/sol/hpack ----- hash: dd35e7fe4a17f0db9c152f68c7e00a0dd8d92536e04aced986b481b462f30705+-- hash: 92f53efa89a440e6a96b4108646c17c6fc9315a7bf8ba2ecadbac8fb2d295f73 name: haskoin-node-version: 0.13.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.+version: 0.14.0+synopsis: P2P library for Bitcoin and Bitcoin Cash+description: Please see the README on GitHub at <https://github.com/haskoin/haskoin-node#readme> category: Bitcoin, Finance, Network homepage: http://github.com/haskoin/haskoin#readme bug-reports: http://github.com/haskoin/haskoin/issues author: Jean-Pierre Rupp maintainer: jprupp@protonmail.ch-license: PublicDomain-license-file: UNLICENSE+license: MIT+license-file: LICENSE build-type: Simple extra-source-files: README.md@@ -31,7 +31,6 @@ Haskoin.Node other-modules: Haskoin.Node.Chain- Haskoin.Node.Common Haskoin.Node.Manager Haskoin.Node.Peer Paths_haskoin_node@@ -46,11 +45,11 @@ , containers , data-default , hashable- , haskoin-core >=0.13.3+ , haskoin-core >=0.14.1 , monad-logger , mtl , network- , nqe+ , nqe >=0.6.3 , random , resourcet , rocksdb-haskell@@ -82,13 +81,13 @@ , containers , data-default , hashable- , haskoin-core >=0.13.3+ , haskoin-core >=0.14.1 , haskoin-node , hspec , monad-logger , mtl , network- , nqe+ , nqe >=0.6.3 , random , resourcet , rocksdb-haskell
src/Haskoin/Node.hs view
@@ -1,163 +1,185 @@ {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE GADTs #-}+{-# LANGUAGE LambdaCase #-} {-# LANGUAGE MultiParamTypeClasses #-}-{-|-Module : Haskoin.Node-Copyright : No rights reserved-License : UNLICENSE-Maintainer : jprupp@protonmail.ch-Stability : experimental-Portability : POSIX+module Haskoin.Node+ ( module Haskoin.Node.Peer+ , module Haskoin.Node.Manager+ , module Haskoin.Node.Chain+ , NodeConfig (..)+ , NodeEvent (..)+ , Node (..)+ , withNode+ , withConnection+ ) where -Integrates peers, manager, and block header synchronization processes.+import Control.Monad (forever)+import Control.Monad.Logger (MonadLoggerIO)+import Data.Conduit.Network (appSink, appSource, clientSettings,+ runTCPClient)+import Data.String.Conversions (cs)+import Data.Time.Clock (NominalDiffTime)+import Database.RocksDB (DB)+import Haskoin (Addr (..), BlockNode (..),+ Headers (..), Message (..), Network,+ NetworkAddress, Ping (..), Pong (..))+import Haskoin.Node.Chain+import Haskoin.Node.Manager+import Haskoin.Node.Peer+import Network.Socket (NameInfoFlag (..), SockAddr,+ getNameInfo)+import NQE (Inbox, Publisher, publish, receive,+ withPublisher, withSubscription)+import Text.Read (readMaybe)+import UnliftIO (MonadUnliftIO, SomeException, catch,+ liftIO, link, throwIO, withAsync) -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- ( -- * Types- Host- , Port- , HostPort+-- | General node configuration.+data NodeConfig = NodeConfig+ { nodeConfMaxPeers :: !Int+ -- ^ maximum number of connected peers allowed+ , nodeConfDB :: !DB+ -- ^ database handler+ , nodeConfPeers :: ![HostPort]+ -- ^ static list of peers to connect to+ , nodeConfDiscover :: !Bool+ -- ^ activate peer discovery+ , nodeConfNetAddr :: !NetworkAddress+ -- ^ network address for the local host+ , nodeConfNet :: !Network+ -- ^ network constants+ , nodeConfEvents :: !(Publisher NodeEvent)+ -- ^ node events are sent to this publisher+ , nodeConfTimeout :: !NominalDiffTime+ -- ^ timeout in seconds+ , nodeConfPeerMaxLife :: !NominalDiffTime+ -- ^ peer disconnect after seconds+ , nodeConfConnect :: !(SockAddr -> WithConnection)+ } - -- * Node- , NodeConfig(..)- , NodeEvent(..)- , withNode- , node+data Node = Node { nodeManager :: !PeerManager+ , nodeChain :: !Chain+ } - -- * Header Chain- , Chain- , ChainMessage- , ChainConfig(..)- , ChainEvent(..)- , chainGetBlock- , chainGetBest- , chainGetAncestor- , chainGetParents- , chainGetSplitBlock- , chainBlockMain- , chainIsSynced+data NodeEvent+ = ChainEvent !ChainEvent+ | PeerEvent !PeerEvent+ | PeerMessage !Peer !Message+ deriving Eq - -- * Peer Manager- , PeerManager- , PeerManagerMessage- , PeerManagerConfig(..)- , PeerEvent(..)- , OnlinePeer(..)- , managerGetPeers- , managerGetPeer- , managerPeerText+withConnection :: SockAddr -> WithConnection+withConnection na f =+ fromSockAddr na >>= \case+ Nothing -> throwIO PeerAddressInvalid+ Just (host, port) -> do+ let cset = clientSettings port (cs host)+ runTCPClient cset $ \ad ->+ f (Conduits (appSource ad) (appSink ad)) - -- * Peer- , Peer- , WithConnection- , PeerMessage- , PeerConfig(..)- , PeerException(..)- , Conduits(..)- , withConnection- , killPeer- , sendMessage- , peerGetPublisher- , peerGetBlocks- , peerGetTxs+fromSockAddr ::+ (MonadUnliftIO m) => SockAddr -> m (Maybe HostPort)+fromSockAddr sa = go `catch` e+ where+ go = do+ (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 - -- * Misc- , myVersion- , buildVersion- ) where+chainForwarder :: MonadLoggerIO m+ => PeerManager+ -> Publisher NodeEvent+ -> Inbox ChainEvent+ -> m ()+chainForwarder mgr pub inbox =+ forever $ receive inbox >>= \event -> do+ case event of+ ChainBestBlock bb ->+ managerBest (nodeHeight bb) mgr+ _ -> return ()+ publish (ChainEvent event) pub -import Control.Monad.Logger (MonadLoggerIO)-import Haskoin (BlockNode (..), Headers (..),- Message (..))-import Haskoin.Node.Chain (chain)-import Haskoin.Node.Common (Chain, ChainConfig (..), ChainEvent (..),- ChainMessage (..), Conduits (..), Host,- HostPort, NodeConfig (..),- NodeEvent (..), OnlinePeer (..), Peer,- PeerConfig (..), PeerEvent (..),- PeerException (..), PeerManager,- PeerManagerConfig (..),- PeerManagerMessage (..),- PeerMessage (..), Port, WithConnection,- buildVersion, chainBlockMain,- chainGetAncestor, chainGetBest,- chainGetBlock, chainGetParents,- chainGetSplitBlock, chainIsSynced,- killPeer, managerGetPeer,- managerGetPeers, managerPeerText,- myVersion, peerGetBlocks,- peerGetPublisher, peerGetTxs,- sendMessage, withConnection)-import Haskoin.Node.Manager (peerManager)-import NQE (Inbox, inboxToMailbox, newInbox, sendSTM)-import UnliftIO (MonadUnliftIO, link, withAsync)+managerForwarder :: MonadLoggerIO m+ => Chain+ -> Publisher NodeEvent+ -> Inbox PeerEvent+ -> m ()+managerForwarder ch pub inbox =+ forever $ receive inbox >>= \event -> do+ case event of+ PeerConnected p ->+ chainPeerConnected p ch+ PeerDisconnected p ->+ chainPeerDisconnected p ch+ publish (PeerEvent event) pub --- | 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)- => NodeConfig- -> ((PeerManager, Chain) -> m a)- -> m a-withNode cfg f = do- mgr_inbox <- newInbox- ch_inbox <- newInbox- withAsync (node cfg mgr_inbox ch_inbox) $ \a -> do- link a- f (inboxToMailbox mgr_inbox, inboxToMailbox ch_inbox)+peerForwarder :: MonadLoggerIO m+ => Chain+ -> PeerManager+ -> Publisher NodeEvent+ -> Inbox (Peer, Message)+ -> m ()+peerForwarder ch mgr pub inbox =+ forever $ receive inbox >>= \(p, msg) -> do+ case msg of+ MVersion v ->+ managerVersion p v mgr+ MVerAck ->+ managerVerAck p mgr+ MPing (Ping n) ->+ managerPing p n mgr+ MPong (Pong n) ->+ managerPong p n mgr+ MAddr (Addr ns) ->+ managerAddrs p (map snd ns) mgr+ MHeaders (Headers hs) ->+ chainHeaders p (map fst hs) ch+ _ -> return ()+ managerTickle p mgr+ publish (PeerMessage p msg) pub -- | Launch node process in the foreground.-node ::+withNode :: ( MonadLoggerIO m , MonadUnliftIO m ) => NodeConfig- -> Inbox PeerManagerMessage- -> Inbox ChainMessage- -> m ()-node cfg mgr_inbox ch_inbox = do- let mgr_config =- PeerManagerConfig- { peerManagerMaxPeers = nodeConfMaxPeers cfg- , peerManagerPeers = nodeConfPeers cfg- , peerManagerDiscover = nodeConfDiscover cfg- , peerManagerNetAddr = nodeConfNetAddr cfg- , peerManagerNetwork = nodeConfNet cfg- , peerManagerEvents = mgr_events- , peerManagerTimeout = nodeConfTimeout cfg- , peerManagerTooOld = nodeConfPeerOld cfg- , peerManagerConnect = nodeConfConnect cfg- }- withAsync (peerManager mgr_config mgr_inbox) $ \mgr_async -> do- link mgr_async- let chain_config =- ChainConfig- { chainConfDB = nodeConfDB cfg- , chainConfNetwork = nodeConfNet cfg- , chainConfEvents = chain_events- , chainConfTimeout = nodeConfTimeout cfg- , chainConfManager = mgr- }- chain chain_config ch_inbox+ -> (Node -> m a)+ -> m a+withNode cfg action =+ withPublisher $ \peer_pub ->+ withPublisher $ \mgr_pub ->+ withPublisher $ \ch_pub ->+ withSubscription peer_pub $ \peer_inbox ->+ withSubscription mgr_pub $ \mgr_inbox ->+ withSubscription ch_pub $ \ch_inbox ->+ withPeerManager (mgr_config mgr_pub peer_pub) $ \mgr ->+ withChain (chain_config ch_pub) $ \ch ->+ withAsync (peerForwarder ch mgr pub peer_inbox) $ \a ->+ withAsync (managerForwarder ch pub mgr_inbox) $ \b ->+ withAsync (chainForwarder mgr pub ch_inbox) $ \c ->+ link a >> link b >> link c >>+ action Node { nodeManager = mgr, nodeChain = ch } where- 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- nodeConfEvents cfg $ PeerEvent event- PeerDisconnected p a -> do- ChainPeerDisconnected p a `sendSTM` ch- nodeConfEvents cfg $ PeerEvent event- _ -> nodeConfEvents cfg $ PeerEvent event- chain_events event = do- nodeConfEvents cfg $ ChainEvent event- case event of- ChainBestBlock b ->- PeerManagerBestBlock (nodeHeight b) `sendSTM` mgr- _ -> return ()+ pub = nodeConfEvents cfg+ chain_config ch_pub =+ ChainConfig+ { chainConfDB = nodeConfDB cfg+ , chainConfNetwork = nodeConfNet cfg+ , chainConfEvents = ch_pub+ , chainConfTimeout = nodeConfTimeout cfg+ }+ mgr_config mgr_pub peer_pub =+ PeerManagerConfig+ { peerManagerMaxPeers = nodeConfMaxPeers cfg+ , peerManagerPeers = nodeConfPeers cfg+ , peerManagerDiscover = nodeConfDiscover cfg+ , peerManagerNetAddr = nodeConfNetAddr cfg+ , peerManagerNetwork = nodeConfNet cfg+ , peerManagerEvents = mgr_pub+ , peerManagerMaxLife = nodeConfPeerMaxLife cfg+ , peerManagerTimeout = nodeConfTimeout cfg+ , peerManagerConnect = nodeConfConnect cfg+ , peerManagerPub = peer_pub+ }
src/Haskoin/Node/Chain.hs view
@@ -8,55 +8,58 @@ {-# LANGUAGE TemplateHaskell #-} {-# LANGUAGE UndecidableInstances #-} {-# OPTIONS_GHC -fno-warn-orphans #-}-{-|-Module : Network.Haskoin.Node.Chain-Copyright : No rights reserved-License : UNLICENSE-Maintainer : jprupp@protonmail.ch-Stability : experimental-Portability : POSIX--Block chain headers synchronizing process.--} module Haskoin.Node.Chain- ( chain+ ( ChainConfig (..)+ , ChainEvent (..)+ , Chain+ , withChain+ , chainGetBlock+ , chainGetBest+ , chainGetAncestor+ , chainGetParents+ , chainGetSplitBlock+ , chainPeerConnected+ , chainPeerDisconnected+ , chainIsSynced+ , chainBlockMain+ , chainHeaders ) where -import Control.Monad (forM_, forever, guard, unless, void,- when)+import Control.Monad (forM_, forever, guard, void, when) import Control.Monad.Except (runExceptT, throwError) import Control.Monad.Logger (MonadLoggerIO, logDebugS, logErrorS, logInfoS)-import Control.Monad.Reader (MonadReader, asks, runReaderT)+import Control.Monad.Reader (MonadReader, ReaderT, ask, asks,+ runReaderT) import Control.Monad.Trans (lift) import Control.Monad.Trans.Maybe (MaybeT (..), runMaybeT) import qualified Data.ByteString as B import Data.Default (def)+import Data.Function (on) import Data.List (delete, nub)-import Data.Maybe (isJust, isNothing, listToMaybe)+import Data.Maybe (isJust, isNothing) import Data.Serialize (Serialize, get, getWord8, put, putWord8) import Data.String.Conversions (cs)-import Data.Time.Clock.POSIX (getPOSIXTime)+import Data.Time.Clock (NominalDiffTime, UTCTime,+ diffUTCTime, getCurrentTime)+import Data.Time.Clock.POSIX (posixSecondsToUTCTime,+ utcTimeToPOSIXSeconds) import Data.Word (Word32) import Database.RocksDB (DB) import qualified Database.RocksDB as R import Database.RocksDB.Query (Key, KeyValue, insert, insertOp, retrieve, writeBatch) import Haskoin (BlockHash, BlockHeader (..),- BlockHeaders (..), BlockNode (..),- GetHeaders (..), Message (..),- Network, Timestamp, blockHashToHex,- blockLocator, connectBlocks,- genesisNode, getAncestor,- headerHash, splitPoint)-import Haskoin.Node.Common (Chain, ChainConfig (..),- ChainEvent (..), ChainMessage (..),- Peer, PeerException (..), killPeer,- managerPeerText, myVersion,- sendMessage)-import NQE (Inbox, inboxToMailbox, receive,- send)+ BlockHeaders (..), BlockHeight,+ BlockNode (..), GetHeaders (..),+ Message (..), Network, blockLocator,+ connectBlocks, genesisNode,+ getAncestor, headerHash, splitPoint)+import Haskoin.Node.Manager (myVersion)+import Haskoin.Node.Peer+import NQE (Mailbox, Publisher, newMailbox,+ publish, receive, send) import System.Random (randomRIO) import UnliftIO (MonadIO, MonadUnliftIO, TVar, atomically, liftIO, link,@@ -65,159 +68,53 @@ import UnliftIO.Concurrent (threadDelay) import UnliftIO.Resource (runResourceT) -type MonadChain m- = (MonadLoggerIO m, MonadChainLogic ChainConfig Peer m)---- | Launch process to synchronize block headers in current thread.-chain ::- (MonadUnliftIO m, MonadLoggerIO m)- => ChainConfig- -> Inbox ChainMessage- -> m ()-chain cfg inbox = do- st <-- newTVarIO- ChainState {chainSyncing = 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- initChainDB net- getBestBlockHeader >>= chainEvent . ChainBestBlock- forever (receive inbox >>= chainMessage)--chainEvent :: MonadChain m => ChainEvent -> m ()-chainEvent e = do- l <- asks (chainConfEvents . myReader)- case e of- ChainBestBlock b ->- $(logInfoS) "Chain" $- "Best block header at height: " <> cs (show (nodeHeight b))- ChainSynced b ->- $(logInfoS) "Chain" $- "Headers in sync at height: " <> cs (show (nodeHeight b))- atomically $ l e--processHeaders ::- MonadChain m => Peer -> [BlockHeader] -> m ()-processHeaders p hs =- void . runMaybeT $ do- net <- asks (chainConfNetwork . myReader)- mgr <- asks (chainConfManager . myReader)- p' <- managerPeerText p mgr- unless (null hs) $- forM_ (zip [(1 :: Int) ..] hs) $ \(i, h) ->- $(logDebugS) "Chain" $- "Importing " <> cs (show i) <> "/" <> cs (show (length hs)) <>- " header " <>- blockHashToHex (headerHash h) <>- " from peer " <>- p'- now <- round <$> liftIO getPOSIXTime- pbest <- getBestBlockHeader- importHeaders net now hs >>= \case- Left e -> do- $(logErrorS) "Chain" $- "Could not connect received headers from peer " <> p'- e `killPeer` p- Right done -> do- setLastReceived now- best <- getBestBlockHeader- when (nodeHeader pbest /= nodeHeader best) . chainEvent $- ChainBestBlock best- if done- then do- MSendHeaders `sendMessage` p- finishPeer p- syncNewPeer- syncNotif- else do- syncPeer p+-- | Mailbox for chain header syncing process.+data Chain = Chain { chainMailbox :: !(Mailbox ChainMessage)+ , chainReader :: !ChainReader+ } -syncNewPeer :: MonadChain m => m ()-syncNewPeer =- getSyncingPeer >>= \case- Nothing -> do- nextPeer >>= \case- Nothing ->- $(logDebugS) "Chain" "No more peers to sync headers against"- Just p -> syncPeer p- Just _ -> return ()+instance Eq Chain where+ (==) = (==) `on` chainMailbox -syncNotif :: MonadChain m => m ()-syncNotif =- round <$> liftIO getPOSIXTime >>= notifySynced >>= \x ->- when x $ getBestBlockHeader >>= chainEvent . ChainSynced+-- | Configuration for chain syncing process.+data ChainConfig =+ ChainConfig+ { chainConfDB :: !DB+ -- ^ database handle+ , chainConfNetwork :: !Network+ -- ^ network constants+ , chainConfEvents :: !(Publisher ChainEvent)+ -- ^ send header chain events here+ , chainConfTimeout :: !NominalDiffTime+ -- ^ timeout in seconds+ } -syncPeer :: MonadChain m => Peer -> m ()-syncPeer p = do- mgr <- asks (chainConfManager . myReader)- bb <-- chainSyncingPeer >>= \case- Just ChainSync {chainSyncPeer = p2, chainHighest = Just g}- | p == p2 -> return g- _ -> getBestBlockHeader- now <- round <$> liftIO getPOSIXTime- gh <- syncHeaders now bb p- p' <- managerPeerText p mgr- $(logDebugS) "Chain" $ "Syncing headers against peer: " <> p'- MGetHeaders gh `sendMessage` p+data ChainMessage+ = ChainHeaders !Peer ![BlockHeader]+ | ChainPeerConnected !Peer+ | ChainPeerDisconnected !Peer+ | ChainPing -chainMessage :: MonadChain m => ChainMessage -> m ()-chainMessage (ChainGetBest reply) = getBestBlockHeader >>= atomically . reply-chainMessage (ChainHeaders p hs) = processHeaders p hs-chainMessage (ChainPeerConnected p _) = do- mgr <- asks (chainConfManager . myReader)- p' <- managerPeerText p mgr- $(logDebugS) "Chain" $ "New peer connected: " <> p'- addPeer p- syncNewPeer-chainMessage (ChainPeerDisconnected p _) = do- finishPeer p- syncNewPeer-chainMessage (ChainGetAncestor h n reply) = do- a <- getAncestor h n- atomically $ reply a-chainMessage (ChainGetSplit l r reply) = do- s <- splitPoint l r- atomically $ reply s-chainMessage (ChainGetBlock h reply) = do- m <- getBlockHeader h- atomically $ reply m-chainMessage (ChainIsSynced reply) = do- s <- isSynced- atomically $ reply s-chainMessage ChainPing = do- ChainConfig {chainConfTimeout = to} <- asks myReader- now <- round <$> liftIO getPOSIXTime- chainSyncingPeer >>= \case- Nothing -> return ()- Just ChainSync {chainSyncPeer = p, chainTimestamp = t}- | now - t > fromIntegral to -> do- mgr <- asks (chainConfManager . myReader)- p' <- managerPeerText p mgr- $(logErrorS) "Chain" $- "Syncing peer timed out after " <> cs (show (now - t)) <>- " seconds: " <>- p'- PeerTimeout `killPeer` p- | otherwise -> return ()+-- | Events originating from chain syncing process.+data ChainEvent+ = ChainBestBlock !BlockNode+ -- ^ chain has new best block+ | ChainSynced !BlockNode+ -- ^ chain is in sync with the network+ deriving (Eq, Show) -withSyncLoop :: (MonadUnliftIO m, MonadLoggerIO m) => Chain -> m a -> m a-withSyncLoop ch f = withAsync go $ \a -> link a >> f- where- go =- forever $ do- threadDelay =<<- liftIO (randomRIO (2 * 1000 * 1000, 20 * 1000 * 1000))- ChainPing `send` ch+type MonadChain m =+ ( MonadLoggerIO m+ , MonadUnliftIO m+ , MonadReader ChainReader m ) --- | Version of the database.-dataVersion :: Word32-dataVersion = 1+-- | Reader for header synchronization code.+data ChainReader = ChainReader+ { myConfig :: !ChainConfig+ -- ^ placeholder for upstream data+ , chainState :: !(TVar ChainState)+ -- ^ mutable state for header synchronization+ } -- | Database key for version. data ChainDataVersionKey = ChainDataVersionKey@@ -232,17 +129,17 @@ return ChainDataVersionKey put ChainDataVersionKey = putWord8 0x92 -data ChainSync p = ChainSync- { chainSyncPeer :: !p- , chainTimestamp :: !Timestamp+data ChainSync = ChainSync+ { chainSyncPeer :: !Peer+ , chainTimestamp :: !UTCTime , chainHighest :: !(Maybe BlockNode) } -- | Mutable state for the header chain process.-data ChainState p = ChainState- { chainSyncing :: !(Maybe (ChainSync p))+data ChainState = ChainState+ { chainSyncing :: !(Maybe ChainSync) -- ^ peer to sync against and time of last received message- , newPeers :: ![p]+ , newPeers :: ![Peer] -- ^ queue of peers to sync against , mySynced :: !Bool -- ^ has the header chain ever been considered synced?@@ -271,60 +168,220 @@ 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) =>+instance (Monad m, MonadIO m, MonadReader DB m) => BlockHeaders m where addBlockHeader bn = do- db <- asks myChainDB- insert db (BlockHeaderKey (headerHash (nodeHeader bn))) bn+ db <- ask+ insert db (BlockHeaderKey h) bn+ where+ h = headerHash (nodeHeader bn) getBlockHeader bh = do- db <- asks myChainDB+ db <- ask retrieve db def (BlockHeaderKey bh) getBestBlockHeader = do- db <- asks myChainDB+ db <- ask retrieve db def BestBlockKey >>= \case Nothing -> error "Could not get best block from database" Just b -> return b setBestBlockHeader bn = do- db <- asks myChainDB+ db <- ask insert db BestBlockKey bn addBlockHeaders bns = do- db <- asks myChainDB+ db <- ask writeBatch db (map f bns) where- f bn = insertOp (BlockHeaderKey (headerHash (nodeHeader bn))) bn+ h bn = headerHash (nodeHeader bn)+ f bn = insertOp (BlockHeaderKey (h bn)) bn +withBlockHeaders :: MonadChain m => ReaderT DB m a -> m a+withBlockHeaders f = do+ db <- asks (chainConfDB . myConfig)+ runReaderT f db++-- | Launch process to synchronize block headers in current thread.+withChain ::+ (MonadUnliftIO m, MonadLoggerIO m)+ => ChainConfig+ -> (Chain -> m a)+ -> m a+withChain cfg action = do+ (inbox, mailbox) <- newMailbox+ $(logDebugS) "Chain" "Starting chain actor"+ st <- newTVarIO ChainState { chainSyncing = Nothing+ , mySynced = False+ , newPeers = []+ }+ let rd = ChainReader { myConfig = cfg+ , chainState = st+ }+ ch = Chain { chainReader = rd+ , chainMailbox = mailbox+ }+ withAsync (main_loop ch rd inbox) $ \a ->+ link a >> action ch+ where+ main_loop ch rd inbox = withSyncLoop ch $+ run inbox `runReaderT` rd+ run inbox = do+ initChainDB+ withBlockHeaders getBestBlockHeader >>=+ chainEvent . ChainBestBlock+ forever $ do+ msg <- receive inbox+ chainMessage msg++chainEvent :: MonadChain m => ChainEvent -> m ()+chainEvent e = do+ pub <- asks (chainConfEvents . myConfig)+ case e of+ ChainBestBlock b ->+ $(logInfoS) "Chain" $+ "Best block header at height: "+ <> cs (show (nodeHeight b))+ ChainSynced b ->+ $(logInfoS) "Chain" $+ "Headers in sync at height: "+ <> cs (show (nodeHeight b))+ publish e pub++processHeaders :: MonadChain m => Peer -> [BlockHeader] -> m ()+processHeaders p hs = void . runMaybeT $ do+ $(logDebugS) "Chain" $+ "Processing " <> cs (show (length hs))+ <> " headers from peer: " <> peerText p+ net <- asks (chainConfNetwork . myConfig)+ now <- liftIO getCurrentTime+ pbest <- lift $ withBlockHeaders getBestBlockHeader+ lift (importHeaders net now hs) >>= \case+ Left e -> do+ $(logErrorS) "Chain" $+ "Could not connect headers from peer: "+ <> peerText p+ e `killPeer` p+ Right done -> lift $ do+ setLastReceived+ best <- withBlockHeaders getBestBlockHeader+ when (nodeHeader pbest /= nodeHeader best) $+ chainEvent (ChainBestBlock best)+ if done+ then do+ MSendHeaders `sendMessage` p+ finishPeer p+ syncNewPeer+ syncNotif+ else syncPeer p++syncNewPeer :: MonadChain m => m ()+syncNewPeer =+ getSyncingPeer >>= \s ->+ when (isNothing s) $+ nextPeer >>= \case+ Nothing ->+ $(logDebugS) "Chain"+ "No more peers to sync against"+ Just p -> do+ $(logDebugS) "Chain" $+ "Syncing against peer: " <> peerText p+ syncPeer p++syncNotif :: MonadChain m => m ()+syncNotif =+ notifySynced >>= \x ->+ when x $ withBlockHeaders getBestBlockHeader >>=+ chainEvent . ChainSynced++syncPeer :: MonadChain m => Peer -> m ()+syncPeer p = do+ t <- liftIO getCurrentTime+ m <- chainSyncingPeer >>= \case+ Just ChainSync { chainSyncPeer = s+ , chainHighest = m+ }+ | p == s -> syncing_me t m+ | otherwise -> return Nothing+ Nothing -> syncing_new t+ forM_ m $ \g -> do+ $(logDebugS) "Chain" $+ "Requesting headers from peer: " <> peerText p+ MGetHeaders g `sendMessage` p+ where+ syncing_new t =+ setBusy p >>= \case+ False -> return Nothing+ True -> do+ $(logDebugS) "Chain" $+ "Locked peer: " <> peerText p+ h <- withBlockHeaders getBestBlockHeader+ Just <$> syncHeaders t h p+ syncing_me t m = do+ h <- case m of+ Nothing -> withBlockHeaders getBestBlockHeader+ Just h -> return h+ Just <$> syncHeaders t h p++chainMessage :: MonadChain m => ChainMessage -> m ()++chainMessage (ChainHeaders p hs) =+ processHeaders p hs++chainMessage (ChainPeerConnected p) = do+ $(logDebugS) "Chain" $ "Peer connected: " <> peerText p+ addPeer p+ syncNewPeer++chainMessage (ChainPeerDisconnected p) = do+ $(logDebugS) "Chain" $ "Peer disconnected: " <> peerText p+ finishPeer p+ syncNewPeer++chainMessage ChainPing = do+ to <- asks (chainConfTimeout . myConfig)+ now <- liftIO getCurrentTime+ chainSyncingPeer >>= \case+ Just ChainSync {chainSyncPeer = p, chainTimestamp = t}+ | now `diffUTCTime` t > to -> do+ $(logErrorS) "Chain" $+ "Syncing peer timed out: " <> peerText p+ PeerTimeout `killPeer` p+ _ -> return ()++withSyncLoop :: (MonadUnliftIO m, MonadLoggerIO m)+ => Chain -> m a -> m a+withSyncLoop ch f =+ withAsync go $ \a ->+ link a >> f+ where+ go = forever $ do+ delay <- liftIO $+ randomRIO ( 2 * 1000 * 1000+ , 20 * 1000 * 1000 )+ threadDelay delay+ ChainPing `send` chainMailbox ch++-- | Version of the database.+dataVersion :: Word32+dataVersion = 1+ -- | 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+initChainDB :: MonadChain m => m ()+initChainDB = do+ db <- asks (chainConfDB . myConfig)+ net <- asks (chainConfNetwork . myConfig) ver <- retrieve db def ChainDataVersionKey when (ver /= Just dataVersion) $ purgeChainDB >>= writeBatch db insert db ChainDataVersionKey dataVersion retrieve db def BestBlockKey >>= \b ->- when (isNothing (b :: Maybe BlockNode)) $ do+ when (isNothing (b :: Maybe BlockNode)) $+ withBlockHeaders $ 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 [R.BatchOp]+purgeChainDB :: MonadChain m => m [R.BatchOp] purgeChainDB = do- db <- asks myChainDB+ db <- asks (chainConfDB . myConfig) runResourceT . R.withIterator db def $ \it -> do R.iterSeek it $ B.singleton 0x90 recurse_delete it db@@ -341,31 +398,33 @@ -- | Import a bunch of continuous headers. Returns 'True' if the number of -- headers is 2000, which means that there are possibly more headers to sync -- from whatever peer delivered these.-importHeaders ::- (MonadIO m, BlockHeaders m, MonadChainLogic a p m)- => Network- -> Timestamp- -> [BlockHeader]- -> m (Either PeerException Bool)+importHeaders :: MonadChain m+ => Network+ -> UTCTime+ -> [BlockHeader]+ -> m (Either PeerException Bool) importHeaders net now hs = runExceptT $- lift (connectBlocks net now hs) >>= \case+ lift connect >>= \case Right _ -> do case hs of [] -> return () _ -> do- bb <- getBlockHeader (headerHash (last hs))+ bb <- lift get_last box <- asks chainState atomically . modifyTVar box $ \s ->- s- { chainSyncing =- (\x -> x {chainHighest = bb}) <$>- chainSyncing s- }+ s { chainSyncing =+ (\x -> x {chainHighest = bb})+ <$> chainSyncing s+ } case length hs of 2000 -> return False _ -> return True Left _ -> throwError PeerSentBadHeaders+ where+ timestamp = floor (utcTimeToPOSIXSeconds now)+ connect = withBlockHeaders $ connectBlocks net timestamp hs+ get_last = withBlockHeaders . getBlockHeader . headerHash $ last hs -- | 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@@ -374,12 +433,13 @@ -- whether to notify other processes that the header chain has been synced. The -- state of the chain will be flipped to synced when this function returns -- 'True'.-notifySynced :: (MonadIO m, MonadChainLogic a p m) => Timestamp -> m Bool-notifySynced now =+notifySynced :: MonadChain m => m Bool+notifySynced = fmap isJust $ runMaybeT $ do- bb <- getBestBlockHeader- guard (now - blockTimestamp (nodeHeader bb) < 2 * 60 * 60)+ bb <- lift $ withBlockHeaders getBestBlockHeader+ now <- liftIO getCurrentTime+ guard $ now `diffUTCTime` block_time bb > 7200 st <- asks chainState MaybeT . atomically . runMaybeT $ do s <- lift $ readTVar st@@ -388,81 +448,178 @@ guard . not $ mySynced s lift $ writeTVar st s {mySynced = True} return ()+ where+ block_time =+ posixSecondsToUTCTime . fromIntegral . blockTimestamp . nodeHeader -- | 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)+nextPeer :: MonadChain m => m (Maybe Peer)+nextPeer =+ go =<< newPeers <$> (readTVarIO =<< asks chainState)+ where+ go [] = return Nothing+ go (p:ps) =+ getBusy p >>= \case+ True -> go ps+ False -> return (Just p) -- | Set a syncing peer and generate a 'GetHeaders' data structure with a block -- locator to send to that peer for syncing. syncHeaders ::- (Eq p, MonadChainLogic a p m, MonadIO m)- => Timestamp+ MonadChain m+ => UTCTime -> BlockNode- -> p+ -> Peer -> m GetHeaders syncHeaders now bb p = do st <- asks chainState- atomically . modifyTVar st $ \s ->- s- { chainSyncing =- Just- ChainSync- { chainSyncPeer = p- , chainTimestamp = now- , chainHighest = Nothing- }- , newPeers = delete p (newPeers s)- }- loc <- blockLocator bb+ atomically $+ modifyTVar st $ \s ->+ s { chainSyncing =+ Just+ ChainSync+ { chainSyncPeer = p+ , chainTimestamp = now+ , chainHighest = Nothing+ }+ , newPeers = delete p (newPeers s)+ }+ loc <- withBlockHeaders $ blockLocator bb return GetHeaders { getHeadersVersion = myVersion , getHeadersBL = loc- , getHeadersHashStop =- "0000000000000000000000000000000000000000000000000000000000000000"+ , getHeadersHashStop = z }+ where+ z = "0000000000000000000000000000000000000000000000000000000000000000" -- | Set the time of last received data to now if a syncing peer is active.-setLastReceived :: (MonadChainLogic a p m, MonadIO m) => Timestamp -> m ()-setLastReceived now = do+setLastReceived :: MonadChain m => m ()+setLastReceived = do+ now <- liftIO getCurrentTime st <- asks chainState+ let f p = p { chainTimestamp = now } atomically . modifyTVar st $ \s ->- s {chainSyncing = (\p -> p {chainTimestamp = now}) <$> chainSyncing s}+ s { chainSyncing = f <$> chainSyncing s } -- | Add a new peer to the queue of peers to sync against.-addPeer :: (Eq p, MonadIO m, MonadChainLogic a p m) => p -> m ()+addPeer :: MonadChain m => Peer -> m () addPeer p = do st <- asks chainState atomically . modifyTVar st $ \s -> s {newPeers = nub (p : newPeers s)} -- | Get syncing peer if there is one.-getSyncingPeer :: (MonadChainLogic a p m, MonadIO m) => m (Maybe p)-getSyncingPeer = fmap chainSyncPeer . chainSyncing <$> (readTVarIO =<< asks chainState)---- | Return 'True' if the chain has ever been considered synced. it will always--- return 'True', even if the chain gets out of sync for any reason.-isSynced :: (MonadChainLogic a p m, MonadIO m) => m Bool-isSynced = mySynced <$> (asks chainState >>= readTVarIO)+getSyncingPeer :: MonadChain m => m (Maybe Peer)+getSyncingPeer =+ fmap chainSyncPeer . chainSyncing <$> (readTVarIO =<< asks chainState) -- | 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 :: MonadChain m => Peer -> m () finishPeer p =- asks chainState >>= \st ->- atomically . modifyTVar st $ \s ->- s- { newPeers = delete p (newPeers s)- , chainSyncing =- case chainSyncing s of- Just ChainSync { chainSyncPeer = p'- , chainTimestamp = _- , chainHighest = _- }- | p == p' -> Nothing- _ -> chainSyncing s- }+ asks chainState >>=+ remove_peer >>= \x -> when x $ do+ $(logDebugS) "Chain" $ "Free peer: " <> peerText p+ setFree p+ where+ remove_peer st =+ atomically $ readTVar st >>= \s -> do+ let is_me = maybe False ((== p) . chainSyncPeer) (chainSyncing s)+ new_sync = if is_me then Nothing else chainSyncing s+ new_peers = delete p (newPeers s)+ writeTVar st $+ s { newPeers = new_peers+ , chainSyncing = new_sync }+ return is_me -- | Return syncing peer data.-chainSyncingPeer :: (MonadChainLogic a p m, MonadIO m) => m (Maybe (ChainSync p))-chainSyncingPeer = chainSyncing <$> (readTVarIO =<< asks chainState)+chainSyncingPeer :: MonadChain m => m (Maybe ChainSync)+chainSyncingPeer =+ chainSyncing <$> (readTVarIO =<< asks chainState)++-- | Get a block header from 'Chain' process.+chainGetBlock :: MonadIO m+ => BlockHash -> Chain -> m (Maybe BlockNode)+chainGetBlock bh ch =+ runReaderT (getBlockHeader bh) (chainConfDB (myConfig (chainReader ch)))++-- | Get best block header from chain process.+chainGetBest :: MonadIO m => Chain -> m BlockNode+chainGetBest ch =+ runReaderT getBestBlockHeader (chainConfDB (myConfig (chainReader ch)))++-- | Get ancestor of 'BlockNode' at 'BlockHeight' from chain process.+chainGetAncestor :: MonadIO m+ => BlockHeight+ -> BlockNode+ -> Chain+ -> m (Maybe BlockNode)+chainGetAncestor h bn ch =+ runReaderT (getAncestor h bn) (chainConfDB (myConfig (chainReader ch)))++-- | Get parents of 'BlockNode' starting at 'BlockHeight' from chain process.+chainGetParents :: MonadIO m+ => BlockHeight+ -> BlockNode+ -> Chain+ -> m [BlockNode]+chainGetParents height top ch =+ go [] top+ where+ go acc b+ | height >= nodeHeight b = return acc+ | otherwise = do+ m <- chainGetBlock (prevBlock $ nodeHeader b) ch+ case m of+ Nothing -> return acc+ Just p -> go (p : acc) p++-- | Get last common block from chain process.+chainGetSplitBlock :: MonadIO m+ => BlockNode+ -> BlockNode+ -> Chain+ -> m BlockNode+chainGetSplitBlock l r ch =+ runReaderT (splitPoint l r) (chainConfDB (myConfig (chainReader ch)))++-- | Notify chain that a new peer is connected.+chainPeerConnected :: MonadIO m+ => Peer+ -> Chain+ -> m ()+chainPeerConnected p ch =+ ChainPeerConnected p `send` chainMailbox ch++-- | Notify chain that a peer has disconnected.+chainPeerDisconnected :: MonadIO m+ => Peer+ -> Chain+ -> m ()+chainPeerDisconnected p ch =+ ChainPeerDisconnected p `send` chainMailbox ch++-- | Is given 'BlockHash' in the main chain?+chainBlockMain :: MonadIO m+ => BlockHash+ -> Chain+ -> m Bool+chainBlockMain bh ch =+ chainGetBest ch >>= \bb ->+ chainGetBlock bh ch >>= \case+ Nothing ->+ return False+ bm@(Just bn) ->+ (== bm) <$> chainGetAncestor (nodeHeight bn) bb ch++-- | Is chain in sync with network?+chainIsSynced :: MonadIO m => Chain -> m Bool+chainIsSynced ch =+ mySynced <$> readTVarIO (chainState (chainReader 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` chainMailbox ch
− src/Haskoin/Node/Common.hs
@@ -1,573 +0,0 @@-{-# LANGUAGE FlexibleContexts #-}-{-# LANGUAGE LambdaCase #-}-{-# LANGUAGE MultiParamTypeClasses #-}-{-# LANGUAGE OverloadedStrings #-}-{-# LANGUAGE RankNTypes #-}-{-|-Module : Network.Haskoin.Node.Common-Copyright : No rights reserved-License : UNLICENSE-Maintainer : jprupp@protonmail.ch-Stability : experimental-Portability : POSIX--Common functions used by Haskoin Node.--}-module Haskoin.Node.Common where--import Conduit (ConduitT)-import Control.Monad (join)-import Control.Monad.Trans.Maybe (MaybeT (..), runMaybeT)-import Data.ByteString (ByteString)-import Data.Conduit.Network (appSink, appSource, clientSettings,- runTCPClient)-import Data.Function (on)-import Data.List (union)-import Data.Maybe (fromMaybe, isJust)-import Data.String.Conversions (cs)-import Data.Text (Text)-import Data.Time.Clock (NominalDiffTime, UTCTime)-import Data.Void (Void)-import Data.Word (Word32, Word64)-import Database.RocksDB (DB)-import Haskoin (Block (..), BlockHash,- BlockHeader (..), BlockHeight,- BlockNode (..), GetData (..),- InvType (..), InvVector (..),- Message (..), Network (..),- NetworkAddress (..), NotFound (..),- Ping (..), Pong (..), Tx, TxHash,- VarString (..), Version (..),- getBlockHash, getTxHash, headerHash,- txHash)-import Network.Socket (AddrInfo (..), AddrInfoFlag (..),- Family (..), NameInfoFlag (..),- SockAddr, SocketType (..),- defaultHints, getAddrInfo,- getNameInfo)-import NQE (Child, Listen, Mailbox, Publisher,- query, queryS, receive,- receiveMatchS, send,- withSubscription)-import System.Random (randomIO)-import Text.Read (readMaybe)-import UnliftIO (Async (..), Exception, MonadIO,- MonadUnliftIO, SomeException, catch,- liftIO, throwIO, timeout)---- | 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- -- ^ network address- , onlinePeerVerAck :: !Bool- -- ^ got version acknowledgement from peer- , onlinePeerConnected :: !Bool- -- ^ peer is connected and ready- , onlinePeerVersion :: !(Maybe Version)- -- ^ protocol version- , onlinePeerAsync :: !(Async ())- -- ^ peer asynchronous process- , onlinePeerMailbox :: !Peer- -- ^ peer mailbox- , onlinePeerNonce :: !Word64- -- ^ random nonce sent during handshake- , onlinePeerPing :: !(Maybe (UTCTime, Word64))- -- ^ last sent ping time and nonce- , onlinePeerLastMessage :: !Word64- -- ^ last message received by peer- , onlinePeerPings :: ![NominalDiffTime]- -- ^ last few ping rountrip duration- , onlinePeerConnectTime :: !Word64- -- ^ when connection was opened- }--instance Eq OnlinePeer where- (==) = (==) `on` f- where- f OnlinePeer {onlinePeerMailbox = p} = p--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 PeerManager = Mailbox PeerManagerMessage--data Conduits =- Conduits- { inboundConduit :: ConduitT () ByteString IO ()- , outboundConduit :: ConduitT ByteString Void IO ()- }--type WithConnection = SockAddr -> (Conduits -> IO ()) -> IO ()---- | General node configuration.-data NodeConfig = NodeConfig- { nodeConfMaxPeers :: !Int- -- ^ maximum number of connected peers allowed- , nodeConfDB :: !DB- -- ^ database handler- , nodeConfPeers :: ![HostPort]- -- ^ static list of peers to connect to- , nodeConfDiscover :: !Bool- -- ^ activate peer discovery- , nodeConfNetAddr :: !NetworkAddress- -- ^ network address for the local host- , nodeConfNet :: !Network- -- ^ network constants- , nodeConfEvents :: !(Listen NodeEvent)- -- ^ node events are sent to this publisher- , nodeConfTimeout :: !Int- -- ^ timeout in seconds- , nodeConfPeerOld :: !Int- -- ^ peer disconnect after seconds- , nodeConfConnect :: !WithConnection- }---- | Peer manager configuration.-data PeerManagerConfig =- PeerManagerConfig- { peerManagerMaxPeers :: !Int- -- ^ maximum number of peers to connect to- , peerManagerPeers :: ![HostPort]- -- ^ static list of peers to connect to- , peerManagerDiscover :: !Bool- -- ^ activate peer discovery- , peerManagerNetAddr :: !NetworkAddress- -- ^ network address for the local host- , peerManagerNetwork :: !Network- -- ^ network constants- , peerManagerEvents :: !(Listen PeerEvent)- -- ^ send manager and peer messages to this mailbox- , peerManagerTimeout :: !Int- -- ^ timeout in seconds- , peerManagerTooOld :: !Int- -- ^ disconnect peers after connected for so long- , peerManagerConnect :: !WithConnection- }---- | Messages that can be sent to the peer manager.-data PeerManagerMessage- = PeerManagerConnect- -- ^ try to connect to peers- | PeerManagerGetPeers !(Listen [OnlinePeer])- -- ^ get all connected peers- | PeerManagerGetOnlinePeer !Peer !(Listen (Maybe OnlinePeer))- -- ^ get a peer information- | PeerManagerCheckPeer !Peer- -- ^ check this peer- | PeerManagerPeerMessage !Peer !Message- -- ^ peer got a message that is forwarded to manager- | PeerManagerPeerDied !Child !(Maybe SomeException)- -- ^ child died- | PeerManagerBestBlock !BlockHeight- -- ^ set this as our best block---- | Configuration for chain syncing process.-data ChainConfig = ChainConfig- { chainConfDB :: !DB- -- ^ database handle- , chainConfManager :: !PeerManager- -- ^ peer manager- , 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- = ChainGetBest !(Listen BlockNode)- -- ^ get best block known- | ChainHeaders !Peer- ![BlockHeader]- | ChainGetAncestor !BlockHeight- !BlockNode- !(Listen (Maybe BlockNode))- -- ^ get ancestor for 'BlockNode' at 'BlockHeight'- | ChainGetSplit !BlockNode- !BlockNode- !(Listen BlockNode)- -- ^ get highest common node- | ChainGetBlock !BlockHash- !(Listen (Maybe BlockNode))- -- ^ get a block header- | 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 syncing process.-data ChainEvent- = 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- { peerConfListen :: !(Publisher Message)- -- ^ Send peer messages to publisher- , peerConfNetwork :: !Network- -- ^ network constants- , peerConfAddress :: !SockAddr- -- ^ peer address- , peerConfConnect :: !WithConnection- }---- | Reasons why a peer may stop working.-data PeerException- = PeerMisbehaving !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 not valid- | PeerSentBadHeaders- -- ^ peer sent wrong headers- | NotNetworkPeer- -- ^ peer cannot serve block chain data- | PeerNoSegWit- -- ^ peer has no segwit support- | PeerTimeout- -- ^ request to peer timed out- | UnknownPeer- -- ^ peer is unknown- | PeerTooOld- -- ^ peer has been connected too long- deriving (Eq, Show)--instance Exception PeerException---- | Events originating from peers and the peer manager.-data PeerEvent- = PeerConnected !Peer- !SockAddr- -- ^ new peer connected- | PeerDisconnected !Peer- !SockAddr- -- ^ peer disconnected- | PeerMessage !Peer- !Message- -- ^ peer sent a message- deriving Eq---- | Incoming messages that a peer accepts.-data PeerMessage- = GetPublisher !(Listen (Publisher Message))- | KillPeer !PeerException- | SendMessage !Message---- | Resolve a host and port to a list of 'SockAddr'. May do DNS lookups.-toSockAddr :: MonadUnliftIO m => HostPort -> m [SockAddr]-toSockAddr (host, port) = go `catch` e- where- go =- fmap (map addrAddress) . liftIO $- getAddrInfo- (Just- defaultHints- { addrFlags = [AI_ADDRCONFIG]- , addrSocketType = Stream- , addrFamily = AF_INET- })- (Just host)- (Just (show port))- e :: Monad m => SomeException -> m [SockAddr]- e _ = return []---- | 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- (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---- | Our protocol version.-myVersion :: Word32-myVersion = 70012---- | Internal function used by peer to send a message to the peer manager.-managerPeerMessage :: MonadIO m => Peer -> Message -> PeerManager -> m ()-managerPeerMessage p msg mgr = PeerManagerPeerMessage p msg `send` mgr---- | Get list of connected peers from manager.-managerGetPeers ::- MonadIO m => PeerManager -> m [OnlinePeer]-managerGetPeers mgr = PeerManagerGetPeers `query` mgr---- | Get information for an online peer from manager.-managerGetPeer :: MonadIO m => Peer -> PeerManager -> m (Maybe OnlinePeer)-managerGetPeer p mgr = PeerManagerGetOnlinePeer p `query` mgr---- | Kill a peer with the provided exception.-killPeer :: MonadIO m => PeerException -> Peer -> m ()-killPeer e p = KillPeer e `send` p---- | Internal function used by manager to check peers periodically.-managerCheck :: MonadIO m => Peer -> PeerManager -> m ()-managerCheck p mgr = PeerManagerCheckPeer p `send` mgr---- | Internal function used to ask manager to connect to a new peer.-managerConnect :: MonadIO m => PeerManager -> m ()-managerConnect mgr = PeerManagerConnect `send` mgr---- | Set the best block that the manager knows about.-managerSetBest :: MonadIO m => BlockHeight -> PeerManager -> m ()-managerSetBest bh mgr = PeerManagerBestBlock bh `send` mgr---- | Send a network message to peer.-sendMessage :: MonadIO m => Message -> Peer -> m ()-sendMessage msg p = SendMessage msg `send` p---- | Get a publisher associated to a peer. Must provide timeout as peer may--- disconnect and become unresponsive.-peerGetPublisher ::- MonadUnliftIO m => Int -> Peer -> m (Maybe (Publisher Message))-peerGetPublisher time = queryS time GetPublisher---- | 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 ::- MonadUnliftIO m- => Network- -> Int- -> Peer- -> [BlockHash]- -> m (Maybe [Block])-peerGetBlocks net time p bhs =- runMaybeT $ mapM f =<< MaybeT (peerGetData time p (GetData ivs))- where- f (Right b) = return b- f (Left _) = MaybeT $ return Nothing- c- | getSegWit net = InvWitnessBlock- | otherwise = InvBlock- ivs = map (InvVector c . getBlockHash) bhs---- | 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- f (Right _) = MaybeT $ return Nothing- f (Left t) = return t- c- | getSegWit net = InvWitnessTx- | otherwise = InvTx- ivs = map (InvVector c . getTxHash) ths---- | 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 ::- Network- -> Word64- -> BlockHeight- -> NetworkAddress- -> NetworkAddress- -> 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.-chainGetBlock :: MonadIO m => BlockHash -> Chain -> m (Maybe BlockNode)-chainGetBlock bh ch = ChainGetBlock bh `query` ch---- | Get best block header from chain process.-chainGetBest :: MonadIO m => Chain -> m BlockNode-chainGetBest ch = ChainGetBest `query` ch---- | Get ancestor of 'BlockNode' at 'BlockHeight' from chain process.-chainGetAncestor ::- MonadIO m => BlockHeight -> BlockNode -> Chain -> m (Maybe BlockNode)-chainGetAncestor h n c = ChainGetAncestor h n `query` c---- | Get parents of 'BlockNode' starting at 'BlockHeight' from chain process.-chainGetParents ::- MonadIO m => BlockHeight -> BlockNode -> Chain -> m [BlockNode]-chainGetParents height top ch = go [] top- where- go acc b- | height >= nodeHeight b = return acc- | otherwise = do- m <- chainGetBlock (prevBlock $ nodeHeader b) ch- case m of- Nothing -> return acc- Just p -> go (p : acc) p---- | Get last common block from chain process.-chainGetSplitBlock ::- 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 =- chainGetBest ch >>= \bb ->- chainGetBlock bh ch >>= \case- Nothing -> return False- bm@(Just bn) -> (== bm) <$> chainGetAncestor (nodeHeight bn) bb ch---- | 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 :: WithConnection-withConnection na f =- fromSockAddr na >>= \case- Nothing -> throwIO PeerAddressInvalid- Just (host, port) -> do- let cset = clientSettings port (cs host)- runTCPClient cset $ \ad -> f (Conduits (appSource ad) (appSink ad))---- | 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---- | Peer string for logging-peerString :: SockAddr -> Text-peerString a = "Peer<" <> cs (show a) <> ">"--managerPeerText :: MonadIO m => Peer -> PeerManager -> m Text-managerPeerText p mgr =- managerGetPeer p mgr >>= \case- Nothing -> return "???"- Just op -> return $ cs (show (onlinePeerAddress op))
src/Haskoin/Node/Manager.hs view
@@ -6,131 +6,203 @@ {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE TemplateHaskell #-} {-# LANGUAGE TupleSections #-}-{-|-Module : Network.Haskoin.Node.Manager-Copyright : No rights reserved-License : UNLICENSE-Maintainer : jprupp@protonmail.ch-Stability : experimental-Portability : POSIX--Peer manager process.--} module Haskoin.Node.Manager- ( peerManager+ ( PeerManagerConfig (..)+ , PeerEvent (..)+ , OnlinePeer (..)+ , HostPort+ , Host+ , Port+ , PeerManager+ , withPeerManager+ , managerBest+ , managerVersion+ , managerPing+ , managerPong+ , managerAddrs+ , managerVerAck+ , managerTickle+ , getPeers+ , getOnlinePeer+ , buildVersion+ , myVersion ) where -import Control.Monad (forM_, forever, guard, when, (<=<))+import Control.Monad (forM_, forever, guard, void, when,+ (<=<)) import Control.Monad.Except (ExceptT (..), runExceptT, throwError) import Control.Monad.Logger (MonadLogger, MonadLoggerIO, logDebugS, logErrorS, logInfoS, logWarnS)-import Control.Monad.Reader (MonadReader, asks, runReaderT)+import Control.Monad.Reader (MonadReader, ReaderT (ReaderT), ask,+ asks, runReaderT) import Control.Monad.Trans (lift) import Control.Monad.Trans.Maybe (MaybeT (..), runMaybeT) import Data.Bits ((.&.))+import Data.Function (on) import Data.List (find, nub, sort)-import Data.Maybe (isJust, isNothing)+import Data.Maybe (fromMaybe, isJust) import Data.Set (Set) import qualified Data.Set as Set import Data.String.Conversions (cs)-import Data.Text (Text) import Data.Time.Clock (NominalDiffTime, UTCTime,- diffUTCTime)-import Data.Time.Clock.POSIX (getCurrentTime, getPOSIXTime)-import Data.Word (Word64)-import Haskoin (Addr (..), BlockHeight,- Message (..), Network (..),- NetworkAddress (..), Ping (..),- Pong (..), Version (..),+ diffUTCTime, getCurrentTime)+import Data.Time.Clock.POSIX (utcTimeToPOSIXSeconds)+import Data.Word (Word32, Word64)+import Haskoin (BlockHeight, Message (..),+ Network (..), NetworkAddress (..),+ Ping (..), Pong (..),+ VarString (..), Version (..), hostToSockAddr, nodeNetwork, sockToHostAddress)-import Haskoin.Node.Common (HostPort, OnlinePeer (..), Peer,- PeerConfig (..), PeerEvent (..),- PeerException (..), PeerManager,- PeerManagerConfig (..),- PeerManagerMessage (..),- buildVersion, killPeer,- managerCheck, sendMessage,- toSockAddr)-import Haskoin.Node.Peer (peer)-import Network.Socket (SockAddr (..))-import NQE (Child, Inbox, Strategy (..),- Supervisor, addChild,- inboxToMailbox, newMailbox, receive,+import Haskoin.Node.Peer+import Network.Socket (AddrInfo (..), AddrInfoFlag (..),+ Family (..), SockAddr (..),+ SocketType (..), defaultHints,+ getAddrInfo)+import NQE (Child, Inbox, Mailbox, Publisher,+ Strategy (..), Supervisor, addChild,+ inboxToMailbox, newInbox,+ newMailbox, publish, receive, receiveMatch, send, sendSTM,- subscribe, unsubscribe,- withPublisher, withSupervisor)+ withSupervisor) import System.Random (randomIO, randomRIO) import UnliftIO (Async, MonadIO, MonadUnliftIO, STM, SomeException, TVar, atomically,- bracket, liftIO, link, modifyTVar,+ catch, liftIO, link, modifyTVar, newTVarIO, readTVar, readTVarIO, withAsync, withRunInIO, writeTVar) import UnliftIO.Concurrent (threadDelay) --- | Monad used by most functions in this module.-type MonadManager m = (MonadLoggerIO m, MonadReader ManagerReader m)+type HostPort = (Host, Port)+type Host = String+type Port = Int --- | Reader for peer configuration and state.-data ManagerReader = ManagerReader- { myConfig :: !PeerManagerConfig- , mySupervisor :: !Supervisor- , myMailbox :: !PeerManager- , myBestBlock :: !(TVar BlockHeight)- , knownPeers :: !(TVar (Set SockAddr))- , onlinePeers :: !(TVar [OnlinePeer])- }+type MonadManager m = (MonadIO m, MonadReader PeerManager m) --- | Peer Manager process. In order to fully start it needs to receive a--- 'ManageBestBlock' event.-peerManager ::- (MonadUnliftIO m, MonadLoggerIO m)- => PeerManagerConfig- -> Inbox PeerManagerMessage- -> m ()-peerManager cfg inbox =- withSupervisor (Notify f) $ \sup -> do+data PeerEvent+ = PeerConnected !Peer+ | PeerDisconnected !Peer+ deriving Eq++data PeerManagerConfig =+ PeerManagerConfig+ { peerManagerMaxPeers :: !Int+ , peerManagerPeers :: ![HostPort]+ , peerManagerDiscover :: !Bool+ , peerManagerNetAddr :: !NetworkAddress+ , peerManagerNetwork :: !Network+ , peerManagerEvents :: !(Publisher PeerEvent)+ , peerManagerTimeout :: !NominalDiffTime+ , peerManagerMaxLife :: !NominalDiffTime+ , peerManagerConnect :: !(SockAddr -> WithConnection)+ , peerManagerPub :: !(Publisher (Peer, Message))+ }++data PeerManager =+ PeerManager+ { myConfig :: !PeerManagerConfig+ , mySupervisor :: !Supervisor+ , myMailbox :: !(Mailbox ManagerMessage)+ , myBestBlock :: !(TVar BlockHeight)+ , knownPeers :: !(TVar (Set SockAddr))+ , onlinePeers :: !(TVar [OnlinePeer])+ }++data ManagerMessage+ = Connect !SockAddr+ | CheckPeer !Peer+ | PeerDied !Child !(Maybe SomeException)+ | ManagerBest !BlockHeight+ | PeerVerAck !Peer+ | PeerVersion !Peer !Version+ | PeerPing !Peer !Word64+ | PeerPong !Peer !Word64+ | PeerAddrs !Peer ![NetworkAddress]+ | PeerTickle !Peer++-- | Data structure representing an online peer.+data OnlinePeer =+ OnlinePeer+ { onlinePeerAddress :: !SockAddr+ , onlinePeerVerAck :: !Bool+ , onlinePeerConnected :: !Bool+ , onlinePeerVersion :: !(Maybe Version)+ , onlinePeerAsync :: !(Async ())+ , onlinePeerMailbox :: !Peer+ , onlinePeerNonce :: !Word64+ , onlinePeerPing :: !(Maybe (UTCTime, Word64))+ , onlinePeerPings :: ![NominalDiffTime]+ , onlinePeerConnectTime :: !UTCTime+ , onlinePeerTickled :: !UTCTime+ }++instance Eq OnlinePeer where+ (==) = (==) `on` f+ where+ f OnlinePeer {onlinePeerMailbox = p} = p++instance Ord OnlinePeer where+ compare = compare `on` f+ where+ f OnlinePeer {onlinePeerPings = pings} = fromMaybe 60 (median pings)++withPeerManager :: (MonadUnliftIO m, MonadLoggerIO m)+ => PeerManagerConfig+ -> (PeerManager -> m a)+ -> m a+withPeerManager cfg action = do+ inbox <- newInbox+ let mgr = inboxToMailbox inbox+ withSupervisor (Notify (death mgr)) $ \sup -> do bb <- newTVarIO 0 kp <- newTVarIO Set.empty ob <- newTVarIO []- let rd =- ManagerReader- { myConfig = cfg- , mySupervisor = sup- , myMailbox = mgr- , myBestBlock = bb- , knownPeers = kp- , onlinePeers = ob- }- go `runReaderT` rd+ let rd = PeerManager { myConfig = cfg+ , mySupervisor = sup+ , myMailbox = mgr+ , myBestBlock = bb+ , knownPeers = kp+ , onlinePeers = ob+ }+ go inbox `runReaderT` rd where- mgr = inboxToMailbox inbox- go = do- putBestBlock <=< receiveMatch inbox $ \case- PeerManagerBestBlock b -> Just b- _ -> Nothing- withConnectLoop mgr $- forever $ do- m <- receive inbox- case m of- PeerManagerPeerMessage p _ -> do- now <- round <$> liftIO getPOSIXTime- o <- asks onlinePeers- atomically $ setLastMessage o now p- _ -> return ()- managerMessage m- f (a, mex) = PeerManagerPeerDied a mex `sendSTM` mgr+ death mgr (a, ex) = PeerDied a ex `sendSTM` mgr+ go inbox =+ withAsync (peerManager inbox) $ \a ->+ withConnectLoop $+ link a >> ReaderT action +peerManager :: ( MonadUnliftIO m+ , MonadManager m+ , MonadLoggerIO m )+ => Inbox ManagerMessage+ -> m ()+peerManager inb = do+ $(logDebugS) "PeerManager" "Awaiting best block"+ putBestBlock <=< receiveMatch inb $ \case+ ManagerBest b -> Just b+ _ -> Nothing+ $(logDebugS) "PeerManager" "Starting peer manager actor"+ forever loop+ where+ loop = do+ m <- receive inb+ managerMessage m+ putBestBlock :: MonadManager m => BlockHeight -> m ()-putBestBlock bb = asks myBestBlock >>= \b -> atomically $ writeTVar b bb+putBestBlock bb = do+ b <- asks myBestBlock+ atomically $ writeTVar b bb getBestBlock :: MonadManager m => m BlockHeight-getBestBlock = asks myBestBlock >>= readTVarIO+getBestBlock =+ asks myBestBlock >>= readTVarIO getNetwork :: MonadManager m => m Network-getNetwork = asks (peerManagerNetwork . myConfig)+getNetwork =+ asks (peerManagerNetwork . myConfig) loadPeers :: (MonadUnliftIO m, MonadManager m) => m () loadPeers = do@@ -150,7 +222,7 @@ ss <- concat <$> mapM toSockAddr (networkSeeds net) mapM_ newPeer ss -logConnectedPeers :: MonadManager m => m ()+logConnectedPeers :: (MonadManager m, MonadLoggerIO m) => m () logConnectedPeers = do m <- asks (peerManagerMaxPeers . myConfig) l <- length <$> getConnectedPeers@@ -158,173 +230,206 @@ "Peers connected: " <> cs (show l) <> "/" <> cs (show m) getOnlinePeers :: MonadManager m => m [OnlinePeer]-getOnlinePeers = asks onlinePeers >>= readTVarIO+getOnlinePeers =+ asks onlinePeers >>= readTVarIO getConnectedPeers :: MonadManager m => m [OnlinePeer]-getConnectedPeers = filter onlinePeerConnected <$> getOnlinePeers--forwardMessage :: MonadManager m => Peer -> Message -> m ()-forwardMessage p = managerEvent . PeerMessage p+getConnectedPeers =+ filter onlinePeerConnected <$> getOnlinePeers managerEvent :: MonadManager m => PeerEvent -> m ()-managerEvent e = asks (peerManagerEvents . myConfig) >>= \l -> atomically $ l e+managerEvent e =+ publish e =<< asks (peerManagerEvents . myConfig) -managerMessage :: (MonadUnliftIO m, MonadManager m) => PeerManagerMessage -> m ()-managerMessage (PeerManagerPeerMessage p (MVersion v)) = do+managerMessage :: ( MonadUnliftIO m+ , MonadManager m+ , MonadLoggerIO m )+ => ManagerMessage+ -> m ()++managerMessage (PeerVersion p v) = do b <- asks onlinePeers- s <- atomically $ peerString b p- e <-- runExceptT $ do- o <- ExceptT . atomically $ setPeerVersion b p v- when (onlinePeerConnected o) $ announcePeer p+ e <- runExceptT $ do+ o <- ExceptT . atomically $ setPeerVersion b p v+ when (onlinePeerConnected o) $ announcePeer p case e of Right () -> do+ $(logDebugS) "PeerManager" $+ "Sending version ack to peer: " <> peerText p MVerAck `sendMessage` p Left x -> do $(logErrorS) "PeerManager" $- "Version rejected for peer " <> s <> ": " <> cs (show x)+ "Version rejected for peer "+ <> peerText p <> ": " <> cs (show x) killPeer x p -managerMessage (PeerManagerPeerMessage p MVerAck) = do+managerMessage (PeerVerAck p) = do b <- asks onlinePeers- s <- atomically $ peerString b p atomically (setPeerVerAck b p) >>= \case- Just o -> when (onlinePeerConnected o) $ announcePeer p+ Just o -> do+ $(logDebugS) "PeerManager" $+ "Received version ack from peer: "+ <> peerText p+ when (onlinePeerConnected o) $+ announcePeer p Nothing -> do- $(logErrorS) "PeerManager" $ "Received verack from unknown peer: " <> s+ $(logErrorS) "PeerManager" $+ "Received verack from unknown peer: "+ <> peerText p killPeer UnknownPeer p -managerMessage (PeerManagerPeerMessage p (MAddr (Addr nas))) =- asks (peerManagerDiscover . myConfig) >>= \discover ->- when discover $ do- let sas = map (hostToSockAddr . naAddress . snd) nas- b <- asks onlinePeers- s <- atomically $ peerString b p- forM_ (zip [(1 :: Int) ..] sas) $ \(i, a) -> do- $(logDebugS) "PeerManager" $- "Received peer address " <> cs (show i) <> "/" <>- cs (show (length sas)) <>- ": " <>- cs (show a) <>- " from peer " <>- s- newPeer a+managerMessage (PeerAddrs p nas) = do+ discover <- asks (peerManagerDiscover . myConfig)+ when discover $ do+ let sas = map (hostToSockAddr . naAddress) nas+ forM_ (zip [(1 :: Int) ..] sas) $ \(i, a) -> do+ $(logDebugS) "PeerManager" $+ "Got peer address "+ <> cs (show i) <> "/" <> cs (show (length sas))+ <> ": " <> cs (show a)+ <> " from peer " <> peerText p+ newPeer a -managerMessage (PeerManagerPeerMessage p m@(MPong (Pong n))) = do- now <- liftIO getCurrentTime+managerMessage (PeerPong p n) = do b <- asks onlinePeers- atomically (gotPong b n now p) >>= \case- Nothing -> forwardMessage p m- Just _ -> return ()+ $(logDebugS) "PeerManager" $+ "Received pong "+ <> cs (show n)+ <> " from: " <> peerText p+ now <- liftIO getCurrentTime+ atomically (gotPong b n now p) -managerMessage (PeerManagerPeerMessage p (MPing (Ping n))) =+managerMessage (PeerPing p n) = do+ $(logDebugS) "PeerManager" $+ "Responding to ping "+ <> cs (show n)+ <> " from: " <> peerText p MPong (Pong n) `sendMessage` p -managerMessage (PeerManagerPeerMessage p m) = forwardMessage p m--managerMessage (PeerManagerBestBlock h) = putBestBlock h+managerMessage (ManagerBest h) =+ putBestBlock h -managerMessage PeerManagerConnect = do- l <- length <$> getConnectedPeers- x <- asks (peerManagerMaxPeers . myConfig)- when (l < x) $- getNewPeer >>= \case- Nothing -> return ()- Just sa -> connectPeer sa+managerMessage (Connect sa) =+ connectPeer sa -managerMessage (PeerManagerPeerDied a e) = processPeerOffline a e+managerMessage (PeerDied a e) =+ processPeerOffline a e -managerMessage (PeerManagerGetPeers reply) = do- ps <- getConnectedPeers- atomically $ reply ps+managerMessage (CheckPeer p) =+ checkPeer p -managerMessage (PeerManagerGetOnlinePeer p reply) = do+managerMessage (PeerTickle p) = do b <- asks onlinePeers- atomically $ findPeer b p >>= reply--managerMessage (PeerManagerCheckPeer p) = checkPeer p+ now <- liftIO getCurrentTime+ atomically $+ modifyPeer b p $ \o ->+ o { onlinePeerTickled = now } -checkPeer :: MonadManager m => Peer -> m ()-checkPeer p = do- PeerManagerConfig {peerManagerTimeout = to, peerManagerTooOld = old} <-- asks myConfig- b <- asks onlinePeers- s <- atomically $ peerString b p- atomically (findPeer b p) >>= \case- Nothing -> return ()- Just o -> do- now <- round <$> liftIO getPOSIXTime- when- (onlinePeerConnectTime o < now - fromIntegral old)- (killPeer PeerTooOld p)- atomically (lastMessage b p) >>= \case- Nothing -> pingPeer p- Just t -> do- now <- round <$> liftIO getPOSIXTime- when (now - t > fromIntegral to) $ do- $(logErrorS) "PeerManager" $- "Peer " <> s <> " did not respond ping on time"+checkPeer :: (MonadManager m, MonadLoggerIO m) => Peer -> m ()+checkPeer p =+ getBusy p >>= \case+ True -> return ()+ False -> do+ PeerManagerConfig+ { peerManagerMaxLife = max_age+ , peerManagerTimeout = to+ } <- asks myConfig+ b <- asks onlinePeers+ now <- liftIO getCurrentTime+ atomically (findPeer b p) >>= \case+ Nothing -> return ()+ Just o -> do+ check_conn max_age now o+ when (check_tickle now to o) (check_ping o)+ where+ check_tickle now to o =+ now `diffUTCTime` onlinePeerTickled o > to+ check_conn max_age now o = do+ let t = onlinePeerConnectTime o+ when (now `diffUTCTime` t > max_age) $+ killPeer PeerTooOld p+ check_ping o =+ case onlinePeerPing o of+ Nothing ->+ sendPing p+ Just _ -> do+ $(logWarnS) "PeerManager" $+ "Peer ping timeout: " <> peerText p killPeer PeerTimeout p -pingPeer :: MonadManager m => Peer -> m ()-pingPeer p = do+sendPing :: (MonadManager m, MonadLoggerIO m) => Peer -> m ()+sendPing p = do b <- asks onlinePeers atomically (findPeer b p) >>= \case- Nothing -> $(logDebugS) "PeerManager" $ "Will not ping unknown peer"+ Nothing ->+ $(logWarnS) "PeerManager" $+ "Will not ping unknown peer: " <> peerText p Just o | onlinePeerConnected o -> do n <- liftIO randomIO now <- liftIO getCurrentTime atomically (setPeerPing b n now p)+ $(logDebugS)" PeerManager" $+ "Sending ping " <> cs (show n)+ <> " to: " <> peerText p MPing (Ping n) `sendMessage` p | otherwise -> return () -processPeerOffline :: MonadManager m => Child -> Maybe SomeException -> m ()+processPeerOffline :: (MonadManager m, MonadLoggerIO 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+ log_disconnected p e+ managerEvent $ PeerDisconnected p+ else log_not_connect p e atomically $ removePeer b p logConnectedPeers where- log_unknown Nothing = $(logErrorS) "PeerManager" "Disconnected unknown peer"+ log_unknown Nothing =+ $(logErrorS) "PeerManager"+ "Disconnected unknown peer" log_unknown (Just x) =- $(logErrorS) "PeerManager" $ "Unknown peer died: " <> cs (show x)- log_disconnected s Nothing =- $(logWarnS) "PeerManager" $ "Disconnected peer: " <> s- log_disconnected s (Just x) =- $(logErrorS) "PeerManager" $ "Peer " <> s <> " died: " <> cs (show x)- log_not_connect s Nothing =- $(logWarnS) "PeerManager" $ "Could not connect to peer " <> s- log_not_connect s (Just x) = $(logErrorS) "PeerManager" $- "Could not connect to peer " <> s <> ": " <> cs (show x)+ "Unknown peer died: " <> cs (show x)+ log_disconnected p Nothing =+ $(logWarnS) "PeerManager" $+ "Disconnected peer: " <> peerText p+ log_disconnected p (Just x) =+ $(logErrorS) "PeerManager" $+ "Peer " <> peerText p <> " died: " <> cs (show x)+ log_not_connect p Nothing =+ $(logWarnS) "PeerManager" $+ "Could not connect to peer " <> peerText p+ log_not_connect p (Just x) =+ $(logErrorS) "PeerManager" $+ "Could not connect to peer "+ <> peerText p <> ": " <> cs (show x) -announcePeer :: MonadManager m => Peer -> m ()+announcePeer :: (MonadManager m, MonadLoggerIO 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) "PeerManager" $ "Connected to peer " <> s- managerEvent $ PeerConnected p a+ Just OnlinePeer {onlinePeerConnected = True} -> do+ $(logInfoS) "PeerManager" $+ "Connected to peer " <> peerText p+ managerEvent $ PeerConnected p logConnectedPeers- managerCheck p mgr- Just OnlinePeer {onlinePeerConnected = False} -> return ()- Nothing -> $(logErrorS) "PeerManager" "Not announcing disconnected peer"+ Just OnlinePeer {onlinePeerConnected = False} ->+ return ()+ Nothing ->+ $(logErrorS) "PeerManager" $+ "Not announcing disconnected peer: "+ <> peerText p getNewPeer :: (MonadUnliftIO m, MonadManager m) => m (Maybe SockAddr)-getNewPeer = runMaybeT $ lift loadPeers >> go+getNewPeer =+ runMaybeT $ lift loadPeers >> go where go = do b <- asks knownPeers@@ -333,14 +438,19 @@ let xs = Set.toList ks a <- liftIO $ randomRIO (0, length xs - 1) let p = xs !! a- atomically . modifyTVar b $ Set.delete p o <- asks onlinePeers- atomically (findPeerAddress o p) >>= \case- Nothing -> return p- Just _ -> go+ m <- atomically $ do+ modifyTVar b $ Set.delete p+ findPeerAddress o p+ maybe (return p) (const go) m -connectPeer :: (MonadUnliftIO m, MonadManager m) => SockAddr -> m ()+connectPeer :: ( MonadUnliftIO m+ , MonadManager m+ , MonadLoggerIO m+ )+ => SockAddr+ -> m () connectPeer sa = do os <- asks onlinePeers atomically (findPeerAddress os sa) >>= \case@@ -352,38 +462,38 @@ PeerManagerConfig { peerManagerNetAddr = ad , peerManagerNetwork = net } <- asks myConfig- mgr <- asks myMailbox sup <- asks mySupervisor conn <- asks (peerManagerConnect . myConfig)+ pub <- asks (peerManagerPub . myConfig) nonce <- liftIO randomIO bb <- getBestBlock- now <- round <$> liftIO getPOSIXTime+ now <- liftIO getCurrentTime let rmt = NetworkAddress (srv net) (sockToHostAddress sa)- ver = buildVersion net nonce bb ad rmt now- (inbox, p) <- newMailbox- let pc pub =- PeerConfig- { peerConfListen = pub- , peerConfNetwork = net- , peerConfAddress = sa- , peerConfConnect = conn- }- a <- withRunInIO $ \io -> sup `addChild` io (launch mgr pc inbox p)+ unix = floor (utcTimeToPOSIXSeconds now)+ ver = buildVersion net nonce bb ad rmt unix+ text = cs (show sa)+ (inbox, mailbox) <- newMailbox+ let pc = PeerConfig { peerConfPub = pub+ , peerConfNetwork = net+ , peerConfText = text+ , peerConfConnect = conn sa+ }+ busy <- newTVarIO False+ p <- wrapPeer pc busy mailbox+ a <- withRunInIO $ \io ->+ sup `addChild` io (launch pc busy inbox p) MVersion ver `sendMessage` p b <- asks onlinePeers _ <- atomically $ newOnlinePeer b sa nonce p a now return () where- l mgr p m = PeerManagerPeerMessage p m `sendSTM` mgr srv net | getSegWit net = 8 | otherwise = 0- launch mgr pc inbox p =- withPublisher $ \pub ->- bracket (subscribe pub (l mgr p)) (unsubscribe pub) $ \_ ->- withPeerLoop sa p mgr $ \a -> do- link a- peer (pc pub) inbox+ launch pc busy inbox p =+ ask >>= \mgr ->+ withPeerLoop sa p mgr $ \a ->+ link a >> peer pc busy inbox withPeerLoop :: (MonadUnliftIO m, MonadLogger m)@@ -392,167 +502,248 @@ -> PeerManager -> (Async a -> m a) -> m a-withPeerLoop _ p mgr = withAsync go- where- go =- forever $ do- threadDelay =<<- liftIO (randomRIO (30 * 1000 * 1000, 90 * 1000 * 1000))- PeerManagerCheckPeer p `send` mgr+withPeerLoop _ p mgr =+ withAsync . forever $ do+ let time_out = floor $ peerManagerTimeout $ myConfig mgr+ threadDelay (time_out * 1000000)+ managerCheck p mgr -withConnectLoop :: (MonadLogger m, MonadUnliftIO m) => PeerManager -> m a -> m a-withConnectLoop mgr act = withAsync go (\a -> link a >> act)+withConnectLoop :: (MonadUnliftIO m, MonadManager m)+ => m a+ -> m a+withConnectLoop act =+ withAsync go $ \a ->+ link a >> act where- go =- forever $ do- PeerManagerConnect `send` mgr- threadDelay =<<- liftIO (randomRIO (100 * 1000, 10 * 500 * 1000))+ go = forever $ do+ l <- length <$> getConnectedPeers+ x <- asks (peerManagerMaxPeers . myConfig)+ when (l < x) $+ getNewPeer >>= mapM_ (\sa -> ask >>= managerConnect sa)+ delay <- liftIO $+ randomRIO ( 100 * 1000+ , 10 * 500 * 1000 )+ threadDelay delay --- | Add a peer. newPeer :: (MonadIO m, MonadManager m) => SockAddr -> m () newPeer sa = do b <- asks knownPeers o <- asks onlinePeers- i <- atomically $ findPeerAddress o sa- when (isNothing i) $ atomically . modifyTVar b $ Set.insert sa+ atomically $+ findPeerAddress o sa >>= \case+ Just _ -> return ()+ Nothing -> modifyTVar b $ Set.insert sa --- | Get static network seeds. networkSeeds :: Network -> [HostPort] networkSeeds net = map (, getDefaultPort net) (getSeeds net) --- | 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--setLastMessage :: TVar [OnlinePeer] -> Word64 -> Peer -> STM ()-setLastMessage b t p = modifyPeer b p $ \o -> o {onlinePeerLastMessage = t}---- | Return time of last ping sent to peer, if any.-lastMessage :: TVar [OnlinePeer] -> Peer -> STM (Maybe Word64)-lastMessage b p = fmap onlinePeerLastMessage <$> findPeer b p+gotPong :: TVar [OnlinePeer] -> Word64 -> UTCTime -> Peer -> STM ()+gotPong b nonce now p = void . 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 = sort $ take 11 $ diff : onlinePeerPings o+ } --- | 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+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+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- -> Word64- -- ^ when connection was established+ -> UTCTime -> STM OnlinePeer-newOnlinePeer b sa n p a t = do- let op =- OnlinePeer- { onlinePeerAddress = sa- , onlinePeerVerAck = False- , onlinePeerConnected = False- , onlinePeerVersion = Nothing- , onlinePeerAsync = a- , onlinePeerMailbox = p- , onlinePeerNonce = n- , onlinePeerPings = []- , onlinePeerPing = Nothing- , onlinePeerConnectTime = t- , onlinePeerLastMessage = t- }- insertPeer b op+newOnlinePeer box addr nonce p peer_async connect_time = do+ let op = OnlinePeer+ { onlinePeerAddress = addr+ , onlinePeerVerAck = False+ , onlinePeerConnected = False+ , onlinePeerVersion = Nothing+ , onlinePeerAsync = peer_async+ , onlinePeerMailbox = p+ , onlinePeerNonce = nonce+ , onlinePeerPings = []+ , onlinePeerPing = Nothing+ , onlinePeerConnectTime = connect_time+ , onlinePeerTickled = connect_time+ }+ insertPeer box 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+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+insertPeer b o =+ modifyTVar b $ \x -> sort . nub $ o : x --- | Modify an online peer.-modifyPeer :: TVar [OnlinePeer] -> Peer -> (OnlinePeer -> OnlinePeer) -> STM ()+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+removePeer b p =+ modifyTVar b $+ filter ((/= p) . onlinePeerMailbox) --- | Find online peer by asynchronous handle.-findPeerAsync :: TVar [OnlinePeer] -> Async () -> STM (Maybe OnlinePeer)-findPeerAsync b a = find ((== a) . onlinePeerAsync) <$> readTVar b+findPeerAsync :: TVar [OnlinePeer]+ -> Async ()+ -> STM (Maybe OnlinePeer)+findPeerAsync b a =+ find ((== a) . onlinePeerAsync)+ <$> readTVar b -findPeerAddress :: TVar [OnlinePeer] -> SockAddr -> STM (Maybe OnlinePeer)-findPeerAddress b a = find ((== a) . onlinePeerAddress) <$> readTVar b+findPeerAddress :: TVar [OnlinePeer]+ -> SockAddr+ -> STM (Maybe OnlinePeer)+findPeerAddress b a =+ find ((== a) . onlinePeerAddress)+ <$> readTVar b++getPeers :: MonadIO m => PeerManager -> m [OnlinePeer]+getPeers = runReaderT getConnectedPeers++getOnlinePeer :: MonadIO m+ => Peer+ -> PeerManager+ -> m (Maybe OnlinePeer)+getOnlinePeer p =+ runReaderT $ asks onlinePeers >>= atomically . (`findPeer` p)++managerCheck :: MonadIO m => Peer -> PeerManager -> m ()+managerCheck p mgr =+ CheckPeer p `send` myMailbox mgr++managerConnect :: MonadIO m => SockAddr -> PeerManager -> m ()+managerConnect sa mgr =+ Connect sa `send` myMailbox mgr++managerBest :: MonadIO m => BlockHeight -> PeerManager -> m ()+managerBest bh mgr =+ ManagerBest bh `send` myMailbox mgr++managerVerAck :: MonadIO m => Peer -> PeerManager -> m ()+managerVerAck p mgr =+ PeerVerAck p `send` myMailbox mgr++managerVersion :: MonadIO m+ => Peer -> Version -> PeerManager -> m ()+managerVersion p ver mgr =+ PeerVersion p ver `send` myMailbox mgr++managerPing :: MonadIO m+ => Peer -> Word64 -> PeerManager -> m ()+managerPing p nonce mgr =+ PeerPing p nonce `send` myMailbox mgr++managerPong :: MonadIO m+ => Peer -> Word64 -> PeerManager -> m ()+managerPong p nonce mgr =+ PeerPong p nonce `send` myMailbox mgr++managerAddrs :: MonadIO m+ => Peer -> [NetworkAddress] -> PeerManager -> m ()+managerAddrs p addrs mgr =+ PeerAddrs p addrs `send` myMailbox mgr++managerTickle :: MonadIO m+ => Peer -> PeerManager -> m ()+managerTickle p mgr =+ PeerTickle p `send` myMailbox mgr++toSockAddr :: MonadUnliftIO m => HostPort -> m [SockAddr]+toSockAddr (host, port) = go `catch` e+ where+ go =+ fmap (map addrAddress) . liftIO $+ getAddrInfo+ (Just+ defaultHints+ { addrFlags = [AI_ADDRCONFIG]+ , addrSocketType = Stream+ , addrFamily = AF_INET+ })+ (Just host)+ (Just (show port))+ e :: Monad m => SomeException -> m [SockAddr]+ e _ = return []++median :: (Ord a, 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'+ where+ ls' = sort ls++buildVersion+ :: Network+ -> Word64+ -> BlockHeight+ -> NetworkAddress+ -> NetworkAddress+ -> 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+ }++myVersion :: Word32+myVersion = 70012
src/Haskoin/Node/Peer.hs view
@@ -1,97 +1,200 @@ {-# 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 : jprupp@protonmail.ch-Stability : experimental-Portability : POSIX--Network peer process. Represents a network peer connection locally.--} module Haskoin.Node.Peer- ( peer+ ( PeerConfig(..)+ , Conduits(..)+ , PeerException(..)+ , WithConnection+ , Peer+ , peer+ , wrapPeer+ , peerPublisher+ , peerText+ , sendMessage+ , killPeer+ , getBlocks+ , getTxs+ , getData+ , pingPeer+ , getBusy+ , setBusy+ , setFree ) where -import Conduit (ConduitT, awaitForever, foldC, mapM_C,- runConduit, takeCE, transPipe, yield,- (.|))-import Control.Monad (forever, when)-import Control.Monad.Logger (MonadLoggerIO, logErrorS)-import Data.ByteString (ByteString)-import qualified Data.ByteString as B-import Data.Serialize (decode, runGet, runPut)-import Data.String.Conversions (cs)-import Haskoin (Message, MessageHeader (..), Network,- getMessage, putMessage)-import Haskoin.Node.Common (Conduits (..), PeerConfig (..),- PeerException (..), PeerMessage (..),- peerString)-import Network.Socket (SockAddr)-import NQE (Inbox, PublisherMessage (..), receive,- send)-import UnliftIO (MonadUnliftIO, atomically, liftIO,- link, throwIO, withAsync, withRunInIO)+import Conduit (ConduitT, Void, awaitForever, foldC,+ mapM_C, runConduit, takeCE,+ transPipe, yield, (.|))+import Control.Monad (forever, join, when)+import Control.Monad.Logger (MonadLoggerIO, logErrorS)+import Control.Monad.Trans.Maybe (MaybeT (MaybeT), runMaybeT)+import Data.ByteString (ByteString)+import qualified Data.ByteString as B+import Data.Function (on)+import Data.List (union)+import Data.Maybe (isJust)+import Data.Serialize (decode, runGet, runPut)+import Data.String.Conversions (cs)+import Data.Text (Text)+import Data.Word (Word32)+import Haskoin (Block (..), BlockHash (..),+ GetData (..), InvType (..),+ InvVector (..), Message (..),+ MessageHeader (..), Network (..),+ NotFound (..), Ping (..), Pong (..),+ Tx, TxHash (..), getMessage,+ headerHash, putMessage, txHash)+import NQE (Inbox, Mailbox, Publisher,+ inboxToMailbox, publish, receive,+ receiveMatchS, send,+ withSubscription)+import System.Random (randomIO)+import UnliftIO (Exception, MonadIO, MonadUnliftIO,+ TVar, atomically, liftIO, link,+ readTVar, readTVarIO, throwIO,+ timeout, withAsync, withRunInIO,+ writeTVar) +data Conduits =+ Conduits+ { inboundConduit :: ConduitT () ByteString IO ()+ , outboundConduit :: ConduitT ByteString Void IO ()+ }++type WithConnection = (Conduits -> IO ()) -> IO ()++data PeerConfig = PeerConfig+ { peerConfPub :: !(Publisher (Peer, Message))+ , peerConfNetwork :: !Network+ , peerConfText :: !Text+ , peerConfConnect :: !WithConnection+ }++data PeerException+ = PeerMisbehaving !String+ | DuplicateVersion+ | DecodeHeaderError+ | CannotDecodePayload+ | PeerIsMyself+ | PayloadTooLarge !Word32+ | PeerAddressInvalid+ | PeerSentBadHeaders+ | NotNetworkPeer+ | PeerNoSegWit+ | PeerTimeout+ | UnknownPeer+ | PeerTooOld+ deriving Eq++instance Show PeerException where+ show (PeerMisbehaving s) = "Peer misbehaving: " <> s+ show DuplicateVersion = "Duplicate version"+ show DecodeHeaderError = "Error decoding header"+ show CannotDecodePayload = "Cannot decode payload"+ show PeerIsMyself = "Peer is myself"+ show (PayloadTooLarge s) = "Payload too large: " <> show s+ show PeerAddressInvalid = "Peer address invalid"+ show PeerSentBadHeaders = "Peer sent bad headers"+ show NotNetworkPeer = "Not network peer"+ show PeerNoSegWit = "Segwit not supported by peer"+ show PeerTimeout = "Peer timed out"+ show UnknownPeer = "Unknown peer"+ show PeerTooOld = "Peer too old"++instance Exception PeerException++-- | Mailbox for a peer.+data Peer = Peer { peerMailbox :: !(Mailbox PeerMessage)+ , peerPublisher :: !(Publisher (Peer, Message))+ , peerText :: !Text+ , peerBusy :: !(TVar Bool)+ }++instance Eq Peer where+ (==) = (==) `on` peerMailbox++instance Show Peer where+ show = cs . peerText++-- | Incoming messages that a peer accepts.+data PeerMessage+ = KillPeer !PeerException+ | SendMessage !Message++wrapPeer :: MonadIO m+ => PeerConfig+ -> TVar Bool+ -> Mailbox PeerMessage+ -> m Peer+wrapPeer cfg busy mbox =+ return Peer { peerMailbox = mbox+ , peerPublisher = peerConfPub cfg+ , peerText = peerConfText cfg+ , peerBusy = busy+ }+ -- | Run peer process in current thread.-peer ::- (MonadUnliftIO m, MonadLoggerIO m)- => PeerConfig- -> Inbox PeerMessage- -> m ()-peer pc inbox = withRunInIO $ \r -> connect a (r . peer_session)+peer :: (MonadUnliftIO m, MonadLoggerIO m)+ => PeerConfig+ -> TVar Bool+ -> Inbox PeerMessage+ -> m ()+peer cfg busy inbox = do+ p <- wrapPeer cfg busy (inboxToMailbox inbox)+ withRunInIO $ connect . (. peer_session p) where- connect = peerConfConnect pc- a = peerConfAddress pc- go = forever $ receive inbox >>= dispatchMessage pc- net = peerConfNetwork pc- peer_session ad =+ connect = peerConfConnect cfg+ go = forever $ receive inbox >>= dispatchMessage cfg+ net = peerConfNetwork cfg+ peer_session p ad = let ins = transPipe liftIO (inboundConduit ad) ons = transPipe liftIO (outboundConduit ad)- src = runConduit $ ins .| inPeerConduit net a .| mapM_C send_msg+ src = runConduit $+ ins+ .| inPeerConduit net (peerConfText cfg)+ .| mapM_C (send_msg p) snk = outPeerConduit net .| ons in withAsync src $ \as -> do link as runConduit (go .| snk)- send_msg = (`send` peerConfListen pc) . Event+ send_msg p msg = publish (p, msg) (peerConfPub cfg) -- | Internal function to dispatch peer messages.-dispatchMessage ::- MonadLoggerIO m => PeerConfig -> PeerMessage -> ConduitT i Message m ()+dispatchMessage :: MonadLoggerIO m+ => PeerConfig+ -> PeerMessage+ -> ConduitT i Message m () dispatchMessage _ (SendMessage msg) = yield msg-dispatchMessage cfg (GetPublisher reply) =- atomically (reply (peerConfListen cfg))-dispatchMessage _ (KillPeer e) = throwIO e+dispatchMessage _ (KillPeer e) = throwIO e -- | Internal conduit to parse messages coming from peer.-inPeerConduit ::- MonadLoggerIO m- => Network- -> SockAddr- -> ConduitT ByteString Message m ()+inPeerConduit :: MonadLoggerIO m+ => Network+ -> Text+ -> ConduitT ByteString Message m () inPeerConduit net a = forever $ do x <- takeCE 24 .| foldC case decode x of Left _ -> do $(logErrorS)- (peerString a)+ (peerLog 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"+ $(logErrorS) (peerLog 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) $+ $(logErrorS) (peerLog a) $ "Cannot decode payload: " <> cs (show e) throwIO CannotDecodePayload Right msg -> yield msg@@ -99,3 +202,129 @@ -- | Outgoing peer conduit to serialize and send messages. outPeerConduit :: Monad m => Network -> ConduitT Message ByteString m () outPeerConduit net = awaitForever $ yield . runPut . putMessage net++-- | Kill a peer with the provided exception.+killPeer :: MonadIO m => PeerException -> Peer -> m ()+killPeer e p = KillPeer e `send` peerMailbox p++-- | Send a network message to peer.+sendMessage :: MonadIO m => Message -> Peer -> m ()+sendMessage msg p = SendMessage msg `send` peerMailbox p++getBusy :: MonadIO m => Peer -> m Bool+getBusy p = readTVarIO (peerBusy p)++setBusy :: MonadIO m => Peer -> m Bool+setBusy p =+ atomically $+ readTVar (peerBusy p) >>= \case+ True -> return False+ False -> writeTVar (peerBusy p) True >>+ return True++setFree :: MonadIO m => Peer -> m ()+setFree p = atomically $ writeTVar (peerBusy p) False++-- | 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.+getBlocks :: MonadUnliftIO m+ => Network+ -> Int+ -> Peer+ -> [BlockHash]+ -> m (Maybe [Block])+getBlocks net time p bhs =+ runMaybeT $ mapM f =<< MaybeT (getData time p (GetData ivs))+ where+ f (Right b) = return b+ f (Left _) = MaybeT $ return Nothing+ c+ | getSegWit net = InvWitnessBlock+ | otherwise = InvBlock+ ivs = map (InvVector c . getBlockHash) bhs++-- | 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.+getTxs :: MonadUnliftIO m+ => Network+ -> Int+ -> Peer+ -> [TxHash]+ -> m (Maybe [Tx])+getTxs net time p ths =+ runMaybeT $ mapM f =<< MaybeT (getData time p (GetData ivs))+ where+ f (Right _) = MaybeT $ return Nothing+ f (Left t) = return t+ c+ | getSegWit net = InvWitnessTx+ | otherwise = InvTx+ ivs = map (InvVector c . getTxHash) ths++-- | Request transactions and/or blocks from peer. Return 'Nothing' if any+-- single inventory fails to be retrieved, if they come out of order, or if+-- timeout is reached.+getData :: MonadUnliftIO m+ => Int+ -> Peer+ -> GetData+ -> m (Maybe [Either Tx Block])+getData seconds p gd@(GetData ivs) =+ withSubscription (peerPublisher p) $ \inb -> do+ MGetData gd `sendMessage` p+ r <- liftIO randomIO+ MPing (Ping r) `sendMessage` p+ fmap join . timeout (seconds * 1000 * 1000) .+ runMaybeT $ get_thing inb r [] ivs+ where+ get_thing _inb _r acc [] =+ return $ reverse acc+ get_thing inb r acc hss@(InvVector t h : hs) =+ filterReceive p inb >>= \case+ MTx tx+ | is_tx t && getTxHash (txHash tx) == h ->+ get_thing inb r (Left tx : acc) hs+ MBlock b@(Block bh _)+ | is_block t && getBlockHash (headerHash bh) == h ->+ get_thing inb 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 inb 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.+pingPeer :: MonadUnliftIO m => Int -> Peer -> m Bool+pingPeer time p =+ fmap isJust . withSubscription (peerPublisher p) $ \sub -> do+ r <- liftIO randomIO+ MPing (Ping r) `sendMessage` p+ receiveMatchS time sub $ \case+ (p', MPong (Pong r'))+ | p == p' && r == r' -> Just ()+ _ -> Nothing++-- | Peer string for logging+peerLog :: Text -> Text+peerLog = mappend "Peer|"++filterReceive :: MonadIO m => Peer -> Inbox (Peer, Message) -> m Message+filterReceive p inb =+ receive inb >>= \case+ (p', msg) | p == p' -> return msg+ _ -> filterReceive p inb
test/Haskoin/NodeSpec.hs view
@@ -3,57 +3,46 @@ {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE RecordWildCards #-}-{-# LANGUAGE TemplateHaskell #-} module Haskoin.NodeSpec ( spec ) where -import Conduit (awaitForever, concatMapC, foldC, mapC,- mapMC, runConduit, takeCE, yield,- (.|))-import Control.Monad (forM_, forever, replicateM,- replicateM_)-import Control.Monad.Logger (logDebug, runNoLoggingT)-import Control.Monad.Trans (lift)-import Data.ByteString (ByteString)-import qualified Data.ByteString as B-import Data.ByteString.Base64 (decodeBase64Lenient)-import Data.Either (fromRight)-import Data.Maybe (isJust, listToMaybe, mapMaybe)-import Data.Serialize (decode, get, runGet, runPut)-import Data.String.Conversions (cs)-import Data.Time.Clock.POSIX (getCurrentTime, getPOSIXTime)-import qualified Database.RocksDB as R-import Haskoin (Block (..), BlockHash (..),- BlockHeader (..), BlockNode (..),- GetData (..), GetHeaders (..),- Headers (..), InvType (..),- InvVector (..), Message (..),- MessageHeader (..), Network (..),- NetworkAddress (..), Ping (..),- Pong (..), VarInt (..), Version (..),- bchRegTest, btcTest, buildMerkleRoot,- getGenesisHeader, getMessage,- headerHash, nodeNetwork, putMessage,- sockToHostAddress, txHash)-import Haskoin.Node (Chain, ChainEvent (..), Conduits (..),- NodeConfig (..), NodeEvent (..),- OnlinePeer (..), Peer, PeerEvent (..),- PeerManager, WithConnection,- buildVersion, chainGetAncestor,- chainGetBest, chainGetParents,- managerGetPeer, peerGetBlocks,- peerGetTxs, withConnection, withNode)-import Network.Socket (SockAddr (..))-import NQE (Inbox, Mailbox, inboxToMailbox,- newInbox, receive, receiveMatch, send,- sendSTM)-import System.Random (randomIO, randomRIO)-import Test.Hspec (Spec, describe, it, shouldBe,- shouldReturn, shouldSatisfy)-import UnliftIO (MonadIO, MonadUnliftIO, liftIO, link,- throwString, withAsync,- withSystemTempDirectory)+import Conduit (awaitForever, concatMapC, foldC, mapMC,+ runConduit, takeCE, yield, (.|))+import Control.Monad (forM_, forever, replicateM)+import Control.Monad.Logger (runNoLoggingT)+import Control.Monad.Trans (lift)+import Data.ByteString (ByteString)+import qualified Data.ByteString as B+import Data.ByteString.Base64 (decodeBase64Lenient)+import Data.Either (fromRight)+import Data.List (find)+import Data.Maybe (isJust, mapMaybe)+import Data.Serialize (decode, get, runGet, runPut)+import Data.Time.Clock.POSIX (getPOSIXTime)+import qualified Database.RocksDB as R+import Haskoin (Block (..), BlockHash (..),+ BlockHeader (..), BlockNode (..),+ GetData (..), GetHeaders (..),+ Headers (..), InvType (..),+ InvVector (..), Message (..),+ MessageHeader (..), Network (..),+ NetworkAddress (..), Ping (..),+ Pong (..), VarInt (..), Version (..),+ bchRegTest, buildMerkleRoot,+ getMessage, headerHash, nodeNetwork,+ putMessage, sockToHostAddress, txHash)+import Haskoin.Node+import Network.Socket (SockAddr (..))+import NQE (Inbox, Mailbox, inboxToMailbox,+ newInbox, receive, receiveMatch, send,+ withPublisher, withSubscription)+import System.Random (randomIO)+import Test.Hspec (Spec, describe, it, shouldBe,+ shouldSatisfy)+import UnliftIO (MonadIO, MonadUnliftIO, liftIO,+ throwString, withAsync,+ withSystemTempDirectory) data TestNode = TestNode { testMgr :: PeerManager@@ -61,7 +50,10 @@ , nodeEvents :: Inbox NodeEvent } -dummyPeerConnect :: Network -> NetworkAddress -> WithConnection+dummyPeerConnect :: Network+ -> NetworkAddress+ -> SockAddr+ -> WithConnection dummyPeerConnect net ad sa f = do r <- newInbox s <- newInbox@@ -105,7 +97,7 @@ hs' = map f allBlocks mockPeerReact (MGetData (GetData ivs)) = mapMaybe f ivs where- f (InvVector InvBlock h) = MBlock <$> listToMaybe (filter (l h) allBlocks)+ f (InvVector InvBlock h) = MBlock <$> find (l h) allBlocks f _ = Nothing l h b = headerHash (blockHeader b) == BlockHash h mockPeerReact _ = []@@ -119,7 +111,7 @@ withTestNode net "connect-one-peer" $ \TestNode {..} -> do p <- waitForPeer nodeEvents Just OnlinePeer {onlinePeerVersion = Just Version {version = ver}} <-- managerGetPeer p testMgr+ getOnlinePeer p testMgr ver `shouldSatisfy` (>= 70002) it "downloads some blocks" $ withTestNode net "get-blocks" $ \TestNode {..} -> do@@ -128,7 +120,7 @@ h2 = "0c89955fc5c9f98ecc71954f167b938138c90c6a094c4737f2e901669d26763f" p <- waitForPeer nodeEvents- pbs <- peerGetBlocks net 10 p [h1, h2]+ pbs <- getBlocks net 10 p [h1, h2] pbs `shouldSatisfy` isJust let Just [b1, b2] = pbs headerHash (blockHeader b1) `shouldBe` h1@@ -178,7 +170,7 @@ waitForPeer :: MonadIO m => Inbox NodeEvent -> m Peer waitForPeer inbox = receiveMatch inbox $ \case- PeerEvent (PeerConnected p _) -> Just p+ PeerEvent (PeerConnected p) -> Just p _ -> Nothing withTestNode ::@@ -189,8 +181,8 @@ -> m a withTestNode net str f = runNoLoggingT $- withSystemTempDirectory ("haskoin-node-test-" <> str <> "-") $ \w -> do- node_inbox <- newInbox+ withSystemTempDirectory ("haskoin-node-test-" <> str <> "-") $ \w ->+ withPublisher $ \pub -> do db <- R.open w@@ -199,30 +191,38 @@ , R.errorIfExists = True , R.compression = R.SnappyCompression }- let ad = NetworkAddress nodeNetwork (sockToHostAddress (SockAddrInet 0 0))- cfg =- NodeConfig- { nodeConfMaxPeers = 20- , nodeConfDB = db- , nodeConfPeers = [("127.0.0.1", 17486)]- , nodeConfDiscover = False- , nodeConfNetAddr = NetworkAddress 0 (sockToHostAddress (SockAddrInet 0 0))- , nodeConfNet = net- , nodeConfEvents = (`sendSTM` node_inbox)- , nodeConfTimeout = 120- , nodeConfPeerOld = 48 * 3600- , nodeConfConnect = dummyPeerConnect net ad- }- withNode cfg $ \(mgr, ch) ->+ let ad = NetworkAddress+ nodeNetwork+ (sockToHostAddress (SockAddrInet 0 0))+ na = NetworkAddress+ 0+ (sockToHostAddress (SockAddrInet 0 0))+ cfg = NodeConfig+ { nodeConfMaxPeers = 20+ , nodeConfDB = db+ , nodeConfPeers = [("127.0.0.1", 17486)]+ , nodeConfDiscover = False+ , nodeConfNetAddr = na+ , nodeConfNet = net+ , nodeConfEvents = pub+ , nodeConfTimeout = 120+ , nodeConfPeerMaxLife = 48 * 3600+ , nodeConfConnect = dummyPeerConnect net ad+ }+ withNode cfg $ \(Node mgr ch) ->+ withSubscription pub $ \sub -> lift $- f TestNode {testMgr = mgr, testChain = ch, nodeEvents = node_inbox}+ f TestNode { testMgr = mgr+ , testChain = ch+ , nodeEvents = sub+ } allBlocks :: [Block] allBlocks = fromRight (error "Could not decode blocks") $ runGet f (decodeBase64Lenient allBlocksBase64) where- f = mapM (const get) [1..15]+ f = mapM (const get) [(1 :: Int) .. 15] allBlocksBase64 :: ByteString allBlocksBase64 =