haskoin-node 0.9.21 → 0.10.0
raw patch · 12 files changed
+1707/−1686 lines, 12 files
Files
- CHANGELOG.md +7/−1
- haskoin-node.cabal +6/−6
- src/Haskoin/Node.hs +40/−42
- src/Haskoin/Node/Chain.hs +458/−0
- src/Haskoin/Node/Common.hs +553/−0
- src/Haskoin/Node/Manager.hs +540/−0
- src/Haskoin/Node/Peer.hs +101/−0
- src/Network/Haskoin/Node/Chain.hs +0/−447
- src/Network/Haskoin/Node/Common.hs +0/−541
- src/Network/Haskoin/Node/Manager.hs +0/−539
- src/Network/Haskoin/Node/Peer.hs +0/−108
- test/Haskoin/NodeSpec.hs +2/−2
CHANGELOG.md view
@@ -4,9 +4,15 @@ 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.10.0+### Changed+- Move modules out of Network.Haskoin namespace.+- Add better and more logging.+- Change Manager module name and related values to PeerManager.+ ## 0.9.21 ### Removed-- Removed unnecessary logging.+- Remove unnecessary logging. ## 0.9.20 ### Changed
haskoin-node.cabal view
@@ -4,10 +4,10 @@ -- -- see: https://github.com/sol/hpack ----- hash: 2cbb34333781c523b209d86f7674b1a21365315ce08863477817a85165e96af3+-- hash: 86210eacdc6447f4f56d49f633d67f529041ec0f96a689bfbd99005150653c2e name: haskoin-node-version: 0.9.21+version: 0.10.0 synopsis: Haskoin Node P2P library for Bitcoin and Bitcoin Cash description: Bitcoin and Bitcoin Cash peer-to-peer protocol library featuring headers-first synchronisation. category: Bitcoin, Finance, Network@@ -29,10 +29,10 @@ library exposed-modules: Haskoin.Node- Network.Haskoin.Node.Chain- Network.Haskoin.Node.Common- Network.Haskoin.Node.Manager- Network.Haskoin.Node.Peer+ Haskoin.Node.Chain+ Haskoin.Node.Common+ Haskoin.Node.Manager+ Haskoin.Node.Peer other-modules: Paths_haskoin_node hs-source-dirs:
src/Haskoin/Node.hs view
@@ -20,9 +20,9 @@ , HostPort , Peer , Chain- , Manager+ , PeerManager , ChainMessage(..)- , ManagerMessage(..)+ , PeerManagerMessage(..) , PeerMessage(..) , OnlinePeer(..) , NodeConfig(..)@@ -34,6 +34,7 @@ , node , managerGetPeers , managerGetPeer+ , managerPeerText , killPeer , sendMessage , peerGetPublisher@@ -49,41 +50,36 @@ , myVersion ) where -import Control.Monad.Logger (MonadLoggerIO)-import Haskoin (BlockNode (..), Headers (..),- Message (..))-import Network.Haskoin.Node.Chain (chain)-import Network.Haskoin.Node.Common (Chain, ChainConfig (..),- ChainEvent (..),- ChainMessage (..), Host,- HostPort, Manager,- ManagerConfig (..),- ManagerMessage (..),- NodeConfig (..), NodeEvent (..),- OnlinePeer (..), Peer,- PeerEvent (..),- PeerException (..),- PeerMessage (..), Port,- chainBlockMain, chainGetAncestor,- chainGetBest, chainGetBlock,- chainGetParents,- chainGetSplitBlock,- chainIsSynced, killPeer,- managerGetPeer, managerGetPeers,- myVersion, peerGetBlocks,- peerGetPublisher, peerGetTxs,- sendMessage)-import Network.Haskoin.Node.Manager (manager)-import NQE (Inbox, inboxToMailbox, newInbox,- sendSTM)-import UnliftIO (MonadUnliftIO, link, withAsync)+import Control.Monad.Logger (MonadLoggerIO)+import Haskoin (BlockNode (..), Headers (..),+ Message (..))+import Haskoin.Node.Chain (chain)+import Haskoin.Node.Common (Chain, ChainConfig (..), ChainEvent (..),+ ChainMessage (..), Host, HostPort,+ NodeConfig (..), NodeEvent (..),+ OnlinePeer (..), Peer, PeerEvent (..),+ PeerException (..), PeerManager,+ PeerManagerConfig (..),+ PeerManagerMessage (..),+ PeerMessage (..), Port, chainBlockMain,+ chainGetAncestor, chainGetBest,+ chainGetBlock, chainGetParents,+ chainGetSplitBlock, chainIsSynced,+ killPeer, managerGetPeer,+ managerGetPeers, managerPeerText,+ myVersion, peerGetBlocks,+ peerGetPublisher, peerGetTxs,+ sendMessage)+import Haskoin.Node.Manager (peerManager)+import NQE (Inbox, inboxToMailbox, newInbox, sendSTM)+import UnliftIO (MonadUnliftIO, link, withAsync) -- | 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- -> ((Manager, Chain) -> m a)+ -> ((PeerManager, Chain) -> m a) -> m a withNode cfg f = do mgr_inbox <- newInbox@@ -98,21 +94,21 @@ , MonadUnliftIO m ) => NodeConfig- -> Inbox ManagerMessage+ -> Inbox PeerManagerMessage -> Inbox ChainMessage -> m () node cfg mgr_inbox ch_inbox = do let mgr_config = ManagerConfig- { mgrConfMaxPeers = nodeConfMaxPeers cfg- , mgrConfPeers = nodeConfPeers cfg- , mgrConfDiscover = nodeConfDiscover cfg- , mgrConfNetAddr = nodeConfNetAddr cfg- , mgrConfNetwork = nodeConfNet cfg- , mgrConfEvents = mgr_events- , mgrConfTimeout = nodeConfTimeout cfg+ { peerManagerMaxPeers = nodeConfMaxPeers cfg+ , peerManagerPeers = nodeConfPeers cfg+ , peerManagerDiscover = nodeConfDiscover cfg+ , peerManagerNetAddr = nodeConfNetAddr cfg+ , peerManagerNetwork = nodeConfNet cfg+ , peerManagerEvents = mgr_events+ , peerManagerTimeout = nodeConfTimeout cfg }- withAsync (manager mgr_config mgr_inbox) $ \mgr_async -> do+ withAsync (peerManager mgr_config mgr_inbox) $ \mgr_async -> do link mgr_async let chain_config = ChainConfig@@ -120,6 +116,7 @@ , chainConfNetwork = nodeConfNet cfg , chainConfEvents = chain_events , chainConfTimeout = nodeConfTimeout cfg+ , chainConfManager = mgr } chain chain_config ch_inbox where@@ -139,5 +136,6 @@ chain_events event = do nodeConfEvents cfg $ ChainEvent event case event of- ChainBestBlock b -> ManagerBestBlock (nodeHeight b) `sendSTM` mgr- _ -> return ()+ ChainBestBlock b ->+ PeerManagerBestBlock (nodeHeight b) `sendSTM` mgr+ _ -> return ()
+ src/Haskoin/Node/Chain.hs view
@@ -0,0 +1,458 @@+{-# LANGUAGE ConstraintKinds #-}+{-# LANGUAGE ExistentialQuantification #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE OverloadedStrings #-}+{-# 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+ ) where++import Control.Monad (forM_, forever, guard, unless, 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.Trans (lift)+import Control.Monad.Trans.Maybe (MaybeT (..), runMaybeT)+import qualified Data.ByteString as B+import Data.Default (def)+import Data.List (delete, nub)+import Data.Maybe (isJust, isNothing, listToMaybe)+import Data.Serialize (Serialize, get, getWord8, put,+ putWord8)+import Data.String.Conversions (cs)+import Data.Time.Clock.POSIX (getPOSIXTime)+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)+import System.Random (randomRIO)+import UnliftIO (MonadIO, MonadUnliftIO, TVar,+ atomically, liftIO, link,+ modifyTVar, newTVarIO, readTVar,+ readTVarIO, withAsync, writeTVar)+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++syncNewPeer :: MonadChain m => m ()+syncNewPeer =+ getSyncingPeer >>= \case+ Nothing -> do+ nextPeer >>= \case+ Nothing -> return ()+ Just p -> syncPeer p+ Just _ -> return ()++syncNotif :: MonadChain m => m ()+syncNotif =+ round <$> liftIO getPOSIXTime >>= notifySynced >>= \x ->+ when x $ getBestBlockHeader >>= chainEvent . ChainSynced++syncPeer :: MonadChain m => Peer -> m ()+syncPeer p = do+ bb <-+ chainSyncingPeer >>= \case+ Just ChainSync {chainSyncPeer = p', chainHighest = Just g}+ | p == p' -> return g+ _ -> getBestBlockHeader+ now <- round <$> liftIO getPOSIXTime+ gh <- syncHeaders now bb p+ MGetHeaders gh `sendMessage` p++chainMessage :: MonadChain m => ChainMessage -> m ()+chainMessage (ChainGetBest reply) = getBestBlockHeader >>= atomically . reply+chainMessage (ChainHeaders p hs) = processHeaders p hs+chainMessage (ChainPeerConnected p _) = do+ 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: " <> p'+ PeerTimeout `killPeer` p+ | otherwise -> return ()++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++-- | Version of the database.+dataVersion :: Word32+dataVersion = 1++-- | Database key for version.+data ChainDataVersionKey = ChainDataVersionKey+ deriving (Eq, Ord, Show)++instance Key ChainDataVersionKey+instance KeyValue ChainDataVersionKey Word32++instance Serialize ChainDataVersionKey where+ get = do+ guard . (== 0x92) =<< getWord8+ return ChainDataVersionKey+ put ChainDataVersionKey = putWord8 0x92++data ChainSync p = ChainSync+ { chainSyncPeer :: !p+ , chainTimestamp :: !Timestamp+ , chainHighest :: !(Maybe BlockNode)+ }++-- | Mutable state for the header chain process.+data ChainState p = ChainState+ { chainSyncing :: !(Maybe (ChainSync p))+ -- ^ peer to sync against and time of last received message+ , newPeers :: ![p]+ -- ^ queue of peers to sync against+ , mySynced :: !Bool+ -- ^ has the header chain ever been considered synced?+ }++-- | Key for block header in database.+newtype BlockHeaderKey = BlockHeaderKey BlockHash deriving (Eq, Show)++instance Serialize BlockHeaderKey where+ get = do+ guard . (== 0x90) =<< getWord8+ BlockHeaderKey <$> get+ put (BlockHeaderKey bh) = do+ putWord8 0x90+ put bh++-- | Key for best block in database.+data BestBlockKey = BestBlockKey deriving (Eq, Show)++instance KeyValue BlockHeaderKey BlockNode+instance KeyValue BestBlockKey BlockNode++instance Serialize BestBlockKey where+ get = do+ guard . (== 0x91) =<< getWord8+ return BestBlockKey+ put BestBlockKey = putWord8 0x91++-- | Type alias for monad commonly used in this module.+type MonadChainLogic a p m+ = (BlockHeaders m, MonadReader (ChainReader a p) m)++-- | Reader for header synchronization code.+data ChainReader a p = ChainReader+ { myReader :: !a+ -- ^ placeholder for upstream data+ , myChainDB :: !DB+ -- ^ database handle+ , chainState :: !(TVar (ChainState p))+ -- ^ mutable state for header synchronization+ }++instance (Monad m, MonadIO m, MonadReader (ChainReader a p) m) =>+ BlockHeaders m where+ addBlockHeader bn = do+ db <- asks myChainDB+ insert db (BlockHeaderKey (headerHash (nodeHeader bn))) bn+ getBlockHeader bh = do+ db <- asks myChainDB+ retrieve db def (BlockHeaderKey bh)+ getBestBlockHeader = do+ db <- asks myChainDB+ retrieve db def BestBlockKey >>= \case+ Nothing -> error "Could not get best block from database"+ Just b -> return b+ setBestBlockHeader bn = do+ db <- asks myChainDB+ insert db BestBlockKey bn+ addBlockHeaders bns = do+ db <- asks myChainDB+ writeBatch db (map f bns)+ where+ f bn = insertOp (BlockHeaderKey (headerHash (nodeHeader bn))) bn++-- | Initialize header database. If version is different from current, the+-- database is purged of conflicting elements first.+initChainDB :: (MonadChainLogic a p m, MonadUnliftIO m) => Network -> m ()+initChainDB net = do+ db <- asks myChainDB+ ver <- retrieve db def ChainDataVersionKey+ when (ver /= Just dataVersion) $ purgeChainDB >>= writeBatch db+ insert db ChainDataVersionKey dataVersion+ retrieve db def BestBlockKey >>= \b ->+ when (isNothing (b :: Maybe BlockNode)) $ do+ addBlockHeader (genesisNode net)+ setBestBlockHeader (genesisNode net)++-- | Purge database of elements having keys that may conflict with those used in+-- this module.+purgeChainDB :: (MonadChainLogic a p m, MonadUnliftIO m) => m [R.BatchOp]+purgeChainDB = do+ db <- asks myChainDB+ runResourceT . R.withIterator db def $ \it -> do+ R.iterSeek it $ B.singleton 0x90+ recurse_delete it db+ where+ recurse_delete it db =+ R.iterKey it >>= \case+ Just k+ | B.head k == 0x90 || B.head k == 0x91 -> do+ R.delete db def k+ R.iterNext it+ (R.Del k :) <$> recurse_delete it db+ _ -> return []++-- | Import a bunch of continuous headers. Returns 'True' if the number of+-- headers is 2000, which means that there are possibly more headers to sync+-- from whatever peer delivered these.+importHeaders ::+ (MonadIO m, BlockHeaders m, MonadChainLogic a p m)+ => Network+ -> Timestamp+ -> [BlockHeader]+ -> m (Either PeerException Bool)+importHeaders net now hs =+ runExceptT $+ lift (connectBlocks net now hs) >>= \case+ Right _ -> do+ case hs of+ [] -> return ()+ _ -> do+ bb <- getBlockHeader (headerHash (last hs))+ box <- asks chainState+ atomically . modifyTVar box $ \s ->+ s+ { chainSyncing =+ (\x -> x {chainHighest = bb}) <$>+ chainSyncing s+ }+ case length hs of+ 2000 -> return False+ _ -> return True+ Left _ -> throwError PeerSentBadHeaders++-- | Check if best block header is in sync with the rest of the block chain by+-- comparing the best block with the current time, verifying that there are no+-- peers in the queue to be synced, and no peer is being synced at the moment.+-- This function will only return 'True' once. It should be used to decide+-- whether to notify other processes that the header chain has been synced. The+-- state of the chain will be flipped to synced when this function returns+-- 'True'.+notifySynced :: (MonadIO m, MonadChainLogic a p m) => Timestamp -> m Bool+notifySynced now =+ fmap isJust $+ runMaybeT $ do+ bb <- getBestBlockHeader+ guard (now - blockTimestamp (nodeHeader bb) < 2 * 60 * 60)+ st <- asks chainState+ MaybeT . atomically . runMaybeT $ do+ s <- lift $ readTVar st+ guard . isNothing $ chainSyncing s+ guard . null $ newPeers s+ guard . not $ mySynced s+ lift $ writeTVar st s {mySynced = True}+ return ()++-- | Get next peer to sync against from the queue.+nextPeer :: (MonadIO m, MonadChainLogic a p m) => m (Maybe p)+nextPeer = listToMaybe . newPeers <$> (asks chainState >>= readTVarIO)++-- | Set a syncing peer and generate a 'GetHeaders' data structure with a block+-- locator to send to that peer for syncing.+syncHeaders ::+ (Eq p, MonadChainLogic a p m, MonadIO m)+ => Timestamp+ -> BlockNode+ -> p+ -> m GetHeaders+syncHeaders now bb p = do+ st <- asks chainState+ atomically . modifyTVar st $ \s ->+ s+ { chainSyncing =+ Just+ ChainSync+ { chainSyncPeer = p+ , chainTimestamp = now+ , chainHighest = Nothing+ }+ , newPeers = delete p (newPeers s)+ }+ loc <- blockLocator bb+ return+ GetHeaders+ { getHeadersVersion = myVersion+ , getHeadersBL = loc+ , getHeadersHashStop =+ "0000000000000000000000000000000000000000000000000000000000000000"+ }++-- | Set the time of last received data to now if a syncing peer is active.+setLastReceived :: (MonadChainLogic a p m, MonadIO m) => Timestamp -> m ()+setLastReceived now = do+ st <- asks chainState+ atomically . modifyTVar st $ \s ->+ s {chainSyncing = (\p -> p {chainTimestamp = now}) <$> chainSyncing s}++-- | Add a new peer to the queue of peers to sync against.+addPeer :: (Eq p, MonadIO m, MonadChainLogic a p m) => p -> m ()+addPeer p = do+ st <- asks chainState+ atomically . modifyTVar st $ \s -> s {newPeers = nub (p : newPeers s)}++-- | Get syncing peer if there is one.+getSyncingPeer :: (MonadChainLogic a p m, MonadIO m) => m (Maybe p)+getSyncingPeer = fmap chainSyncPeer . chainSyncing <$> (readTVarIO =<< asks chainState)++-- | Return 'True' if the chain has ever been considered synced. it will always+-- return 'True', even if the chain gets out of sync for any reason.+isSynced :: (MonadChainLogic a p m, MonadIO m) => m Bool+isSynced = mySynced <$> (asks chainState >>= readTVarIO)++-- | Remove a peer from the queue of peers to sync and unset the syncing peer if+-- it is set to the provided value.+finishPeer :: (Eq p, MonadIO m, MonadChainLogic a p m) => p -> m ()+finishPeer p =+ asks chainState >>= \st ->+ atomically . modifyTVar st $ \s ->+ s+ { newPeers = delete p (newPeers s)+ , chainSyncing =+ case chainSyncing s of+ Just ChainSync { chainSyncPeer = p'+ , chainTimestamp = _+ , chainHighest = _+ }+ | p == p' -> Nothing+ _ -> chainSyncing s+ }++-- | Return syncing peer data.+chainSyncingPeer :: (MonadChainLogic a p m, MonadIO m) => m (Maybe (ChainSync p))+chainSyncingPeer = chainSyncing <$> (readTVarIO =<< asks chainState)
+ src/Haskoin/Node/Common.hs view
@@ -0,0 +1,553 @@+{-# 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 Control.Monad (join)+import Control.Monad.Trans.Maybe (MaybeT (..), runMaybeT)+import Data.Conduit.Network (AppData, clientSettings,+ runGeneralTCPClient)+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.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+ , 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++-- | 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+ }++-- | Peer manager configuration.+data PeerManagerConfig = ManagerConfig+ { 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+ }++-- | 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+ }++-- | 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 ::+ MonadUnliftIO m => SockAddr -> (AppData -> m a) -> m a+withConnection na f =+ fromSockAddr na >>= \case+ Nothing -> throwIO PeerAddressInvalid+ Just (host, port) ->+ let cset = clientSettings port (cs host)+ in runGeneralTCPClient cset f++-- | Calculate the median value from a list. The list must not be empty.+median :: Fractional a => [a] -> Maybe a+median ls+ | null ls = Nothing+ | length ls `mod` 2 == 0 =+ Just . (/ 2) . sum . take 2 $ drop (length ls `div` 2 - 1) ls+ | otherwise = Just . head $ drop (length ls `div` 2) ls++-- | 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
@@ -0,0 +1,540 @@+{-# LANGUAGE ConstraintKinds #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# 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+ ) where++import Control.Monad (forM_, forever, guard, 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.Trans (lift)+import Control.Monad.Trans.Maybe (MaybeT (..), runMaybeT)+import Data.Bits ((.&.))+import Data.List (find, nub, sort)+import Data.Maybe (isJust, isNothing)+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 (..),+ 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,+ receiveMatch, send, sendSTM,+ subscribe, unsubscribe,+ withPublisher, withSupervisor)+import System.Random (randomIO, randomRIO)+import UnliftIO (Async, MonadIO, MonadUnliftIO, STM,+ SomeException, TVar, atomically,+ bracket, 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)++-- | 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])+ }++-- | 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+ 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+ where+ mgr = inboxToMailbox inbox+ go = do+ putBestBlock <=< receiveMatch inbox $ \case+ PeerManagerBestBlock b -> Just b+ _ -> Nothing+ withConnectLoop mgr $+ forever $ do+ m <- receive inbox+ managerMessage m+ f (a, mex) = PeerManagerPeerDied a mex `sendSTM` mgr++putBestBlock :: MonadManager m => BlockHeight -> m ()+putBestBlock bb = asks myBestBlock >>= \b -> atomically $ writeTVar b bb++getBestBlock :: MonadManager m => m BlockHeight+getBestBlock = asks myBestBlock >>= readTVarIO++getNetwork :: MonadManager m => m Network+getNetwork = asks (peerManagerNetwork . myConfig)++loadPeers :: (MonadUnliftIO m, MonadManager m) => m ()+loadPeers = do+ loadStaticPeers+ loadNetSeeds++loadStaticPeers :: (MonadUnliftIO m, MonadManager m) => m ()+loadStaticPeers = do+ xs <- asks (peerManagerPeers . myConfig)+ mapM_ newPeer =<< concat <$> mapM toSockAddr xs++loadNetSeeds :: (MonadUnliftIO m, MonadManager m) => m ()+loadNetSeeds =+ asks (peerManagerDiscover . myConfig) >>= \discover ->+ when discover $ do+ net <- getNetwork+ ss <- concat <$> mapM toSockAddr (networkSeeds net)+ mapM_ newPeer ss++logConnectedPeers :: MonadManager m => m ()+logConnectedPeers = do+ m <- asks (peerManagerMaxPeers . myConfig)+ l <- length <$> getConnectedPeers+ $(logInfoS) "Manager" $+ "Peers connected: " <> cs (show l) <> "/" <> cs (show m)++getOnlinePeers :: MonadManager m => m [OnlinePeer]+getOnlinePeers = asks onlinePeers >>= readTVarIO++getConnectedPeers :: MonadManager m => m [OnlinePeer]+getConnectedPeers = filter onlinePeerConnected <$> getOnlinePeers++forwardMessage :: MonadManager m => Peer -> Message -> m ()+forwardMessage p = managerEvent . PeerMessage p++managerEvent :: MonadManager m => PeerEvent -> m ()+managerEvent e = asks (peerManagerEvents . myConfig) >>= \l -> atomically $ l e++managerMessage :: (MonadUnliftIO m, MonadManager m) => PeerManagerMessage -> m ()+managerMessage (PeerManagerPeerMessage p (MVersion 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+ case e of+ Right () -> do+ MVerAck `sendMessage` p+ Left x -> do+ $(logErrorS) "Manager" $+ "Version rejected for peer " <> s <> ": " <> cs (show x)+ killPeer x p++managerMessage (PeerManagerPeerMessage p MVerAck) = do+ b <- asks onlinePeers+ s <- atomically $ peerString b p+ atomically (setPeerVerAck b p) >>= \case+ Just o -> when (onlinePeerConnected o) $ announcePeer p+ Nothing -> do+ $(logErrorS) "Manager" $ "Received verack from unknown peer: " <> s+ 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+ $(logInfoS) "Manager" $+ "Received " <> cs (show (length sas)) <> " new peers from peer " <>+ s+ forM_ sas newPeer++managerMessage (PeerManagerPeerMessage p m@(MPong (Pong n))) = do+ now <- liftIO getCurrentTime+ b <- asks onlinePeers+ atomically (gotPong b n now p) >>= \case+ Nothing -> forwardMessage p m+ Just _ -> return ()++managerMessage (PeerManagerPeerMessage p (MPing (Ping n))) =+ MPong (Pong n) `sendMessage` p++managerMessage (PeerManagerPeerMessage p m) = forwardMessage p m++managerMessage (PeerManagerBestBlock 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 (PeerManagerPeerDied a e) = processPeerOffline a e++managerMessage (PeerManagerGetPeers reply) = do+ ps <- getConnectedPeers+ atomically $ reply ps++managerMessage (PeerManagerGetOnlinePeer p reply) = do+ b <- asks onlinePeers+ atomically $ findPeer b p >>= reply++managerMessage (PeerManagerCheckPeer p) = checkPeer p++checkPeer :: MonadManager m => Peer -> m ()+checkPeer p = do+ ManagerConfig {peerManagerTimeout = to} <- 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 - 1800) (killPeer PeerTooOld p)+ atomically (lastPing b p) >>= \case+ Nothing -> pingPeer p+ Just t -> do+ now <- liftIO getCurrentTime+ when (diffUTCTime now t > fromIntegral to) $ do+ $(logErrorS) "Manager" $+ "Peer " <> s <> " did not respond ping on time"+ killPeer PeerTimeout p++pingPeer :: MonadManager m => Peer -> m ()+pingPeer p = do+ b <- asks onlinePeers+ atomically (findPeer b p) >>= \case+ Nothing -> $(logDebugS) "Manager" $ "Will not ping unknown peer"+ Just o+ | onlinePeerConnected o -> do+ n <- liftIO randomIO+ now <- liftIO getCurrentTime+ atomically (setPeerPing b n now p)+ MPing (Ping n) `sendMessage` p+ | otherwise -> return ()++processPeerOffline :: MonadManager m => Child -> Maybe SomeException -> m ()+processPeerOffline a e = do+ b <- asks onlinePeers+ atomically (findPeerAsync b a) >>= \case+ Nothing -> log_unknown e+ Just o -> do+ let p = onlinePeerMailbox o+ d = onlinePeerAddress o+ s <- atomically $ peerString b p+ if onlinePeerConnected o+ then do+ log_disconnected s e+ managerEvent $ PeerDisconnected p d+ else log_not_connect s e+ atomically $ removePeer b p+ logConnectedPeers+ where+ log_unknown Nothing = $(logErrorS) "Manager" "Disconnected unknown peer"+ log_unknown (Just x) =+ $(logErrorS) "Manager" $ "Unknown peer died: " <> cs (show x)+ log_disconnected s Nothing =+ $(logWarnS) "Manager" $ "Disconnected peer: " <> s+ log_disconnected s (Just x) =+ $(logErrorS) "Manager" $ "Peer " <> s <> " died: " <> cs (show x)+ log_not_connect s Nothing =+ $(logWarnS) "Manager" $ "Could not connect to peer " <> s+ log_not_connect s (Just x) =+ $(logErrorS) "Manager" $+ "Could not connect to peer " <> s <> ": " <> cs (show x)++announcePeer :: MonadManager m => Peer -> m ()+announcePeer p = do+ b <- asks onlinePeers+ s <- atomically $ peerString b p+ mgr <- asks myMailbox+ atomically (findPeer b p) >>= \case+ Just OnlinePeer {onlinePeerAddress = a, onlinePeerConnected = True} -> do+ $(logInfoS) "Manager" $ "Handshake completed for peer " <> s+ managerEvent $ PeerConnected p a+ logConnectedPeers+ managerCheck p mgr+ Just OnlinePeer {onlinePeerConnected = False} -> return ()+ Nothing -> $(logErrorS) "Manager" "Will not announce unknown peer"++getNewPeer :: (MonadUnliftIO m, MonadManager m) => m (Maybe SockAddr)+getNewPeer = runMaybeT $ lift loadPeers >> go+ where+ go = do+ b <- asks knownPeers+ ks <- readTVarIO b+ guard . not $ Set.null ks+ let xs = Set.toList ks+ a <- liftIO $ randomRIO (0, length xs - 1)+ let p = xs !! a+ atomically . modifyTVar b $ Set.delete p+ o <- asks onlinePeers+ atomically (findPeerAddress o p) >>= \case+ Nothing -> return p+ Just _ -> go+++connectPeer :: (MonadUnliftIO m, MonadManager m) => SockAddr -> m ()+connectPeer sa = do+ os <- asks onlinePeers+ atomically (findPeerAddress os sa) >>= \case+ Just _ ->+ $(logErrorS) "Manager" $+ "Attempted to connect to peer twice: " <> cs (show sa)+ Nothing -> do+ $(logInfoS) "Manager" $ "Connecting to " <> cs (show sa)+ ManagerConfig {peerManagerNetAddr = ad, peerManagerNetwork = net} <-+ asks myConfig+ mgr <- asks myMailbox+ sup <- asks mySupervisor+ nonce <- liftIO randomIO+ bb <- getBestBlock+ now <- round <$> liftIO getPOSIXTime+ 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+ }+ a <- withRunInIO $ \io -> sup `addChild` io (launch mgr pc inbox p)+ MVersion ver `sendMessage` p+ b <- asks onlinePeers+ _ <- atomically $ newOnlinePeer b sa nonce p a 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++withPeerLoop ::+ (MonadUnliftIO m, MonadLogger m)+ => SockAddr+ -> Peer+ -> 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++withConnectLoop :: (MonadLogger m, MonadUnliftIO m) => PeerManager -> m a -> m a+withConnectLoop mgr act = withAsync go (\a -> link a >> act)+ where+ go =+ forever $ do+ threadDelay =<<+ liftIO (randomRIO (2 * 1000 * 1000, 10 * 1000 * 1000))+ PeerManagerConnect `send` mgr++-- | 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++-- | 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++-- | Return time of last ping sent to peer, if any.+lastPing :: TVar [OnlinePeer] -> Peer -> STM (Maybe UTCTime)+lastPing b p =+ findPeer b p >>= \case+ Just OnlinePeer {onlinePeerPing = Just (time, _)} -> return (Just time)+ _ -> return Nothing++-- | Set nonce and time of last ping sent to peer.+setPeerPing :: TVar [OnlinePeer] -> Word64 -> UTCTime -> Peer -> STM ()+setPeerPing b nonce now p =+ modifyPeer b p $ \o -> o {onlinePeerPing = Just (now, nonce)}++-- | Set version for online peer. Will set the peer connected status to 'True'+-- if a verack message has already been registered for that peer.+setPeerVersion ::+ TVar [OnlinePeer]+ -> Peer+ -> Version+ -> STM (Either PeerException OnlinePeer)+setPeerVersion b p v =+ runExceptT $ do+ when (services v .&. nodeNetwork == 0) $ throwError NotNetworkPeer+ ops <- lift $ readTVar b+ when (any ((verNonce v ==) . onlinePeerNonce) ops) $+ throwError PeerIsMyself+ lift (findPeer b p) >>= \case+ Nothing -> throwError UnknownPeer+ Just o -> do+ let n =+ o+ { onlinePeerVersion = Just v+ , onlinePeerConnected = onlinePeerVerAck o+ }+ lift $ insertPeer b n+ return n++-- | Register that a verack message was received from a peer.+setPeerVerAck :: TVar [OnlinePeer] -> Peer -> STM (Maybe OnlinePeer)+setPeerVerAck b p =+ runMaybeT $ do+ o <- MaybeT $ findPeer b p+ let n =+ o+ { onlinePeerVerAck = True+ , onlinePeerConnected = isJust (onlinePeerVersion o)+ }+ lift $ insertPeer b n+ return n++-- | Create 'OnlinePeer' data structure.+newOnlinePeer ::+ TVar [OnlinePeer]+ -> SockAddr+ -- ^ peer address+ -> Word64+ -- ^ nonce sent to peer+ -> Peer+ -- ^ peer mailbox+ -> Async ()+ -- ^ peer asynchronous handle+ -> Word64+ -- ^ when connection was established+ -> 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+ }+ insertPeer b op+ return op++-- | Get a human-readable string for the peer address.+peerString :: TVar [OnlinePeer] -> Peer -> STM Text+peerString b p =+ maybe "[unknown]" (cs . show . onlinePeerAddress) <$> findPeer b p++-- | Find a connected peer.+findPeer :: TVar [OnlinePeer] -> Peer -> STM (Maybe OnlinePeer)+findPeer b p = find ((== p) . onlinePeerMailbox) <$> readTVar b++-- | Insert or replace a connected peer.+insertPeer :: TVar [OnlinePeer] -> OnlinePeer -> STM ()+insertPeer b o = modifyTVar b $ \x -> sort . nub $ o : x++-- | Modify an online peer.+modifyPeer :: TVar [OnlinePeer] -> Peer -> (OnlinePeer -> OnlinePeer) -> STM ()+modifyPeer b p f =+ findPeer b p >>= \case+ Nothing -> return ()+ Just o -> insertPeer b $ f o++-- | Remove an online peer.+removePeer :: TVar [OnlinePeer] -> Peer -> STM ()+removePeer b p = modifyTVar b $ \x -> filter ((/= p) . onlinePeerMailbox) x++-- | Find online peer by asynchronous handle.+findPeerAsync :: TVar [OnlinePeer] -> Async () -> STM (Maybe OnlinePeer)+findPeerAsync b a = find ((== a) . onlinePeerAsync) <$> readTVar b++findPeerAddress :: TVar [OnlinePeer] -> SockAddr -> STM (Maybe OnlinePeer)+findPeerAddress b a = find ((== a) . onlinePeerAddress) <$> readTVar b
+ src/Haskoin/Node/Peer.hs view
@@ -0,0 +1,101 @@+{-# LANGUAGE ConstraintKinds #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE TemplateHaskell #-}+{-# 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+ ) where++import Conduit (ConduitT, awaitForever, foldC, mapM_C,+ runConduit, takeCE, yield, (.|))+import Control.Monad (forever, when)+import Control.Monad.Logger (MonadLoggerIO, logErrorS)+import Control.Monad.Reader (runReaderT)+import Data.ByteString (ByteString)+import qualified Data.ByteString as B+import Data.Conduit.Network (appSink, appSource)+import Data.Serialize (decode, runGet, runPut)+import Data.String.Conversions (cs)+import Haskoin (Message, MessageHeader (..), Network,+ getMessage, putMessage)+import Haskoin.Node.Common (PeerConfig (..), PeerException (..),+ PeerMessage (..), peerString,+ withConnection)+import Network.Socket (SockAddr)+import NQE (Inbox, PublisherMessage (..), receive,+ send)+import UnliftIO (MonadUnliftIO, atomically, link,+ throwIO, withAsync)++-- | Run peer process in current thread.+peer ::+ (MonadUnliftIO m, MonadLoggerIO m)+ => PeerConfig+ -> Inbox PeerMessage+ -> m ()+peer pc inbox = withConnection a $ \ad -> runReaderT (peer_session ad) pc+ where+ a = peerConfAddress pc+ go = forever $ do+ receive inbox >>= dispatchMessage pc+ net = peerConfNetwork pc+ peer_session ad =+ let ins = appSource ad+ ons = appSink ad+ src = runConduit $ ins .| inPeerConduit net a .| mapM_C send_msg+ snk = outPeerConduit net .| ons+ in withAsync src $ \as -> do+ link as+ runConduit (go .| snk)+ send_msg = (`send` peerConfListen pc) . Event++-- | Internal function to dispatch peer messages.+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++-- | Internal conduit to parse messages coming from peer.+inPeerConduit ::+ MonadLoggerIO m+ => Network+ -> SockAddr+ -> ConduitT ByteString Message m ()+inPeerConduit net a = forever $ do+ x <- takeCE 24 .| foldC+ case decode x of+ Left _ -> do+ $(logErrorS)+ (peerString a)+ "Could not decode incoming message header"+ throwIO DecodeHeaderError+ Right (MessageHeader _ _ len _) -> do+ when (len > 32 * 2 ^ (20 :: Int)) $ do+ $(logErrorS) (peerString a) "Payload too large"+ throwIO $ PayloadTooLarge len+ y <- takeCE (fromIntegral len) .| foldC+ case runGet (getMessage net) $ x `B.append` y of+ Left e -> do+ $(logErrorS) (peerString a) $+ "Cannot decode payload: " <> cs (show e)+ throwIO CannotDecodePayload+ Right msg -> yield msg++-- | Outgoing peer conduit to serialize and send messages.+outPeerConduit :: Monad m => Network -> ConduitT Message ByteString m ()+outPeerConduit net = awaitForever $ yield . runPut . putMessage net
− src/Network/Haskoin/Node/Chain.hs
@@ -1,447 +0,0 @@-{-# LANGUAGE ConstraintKinds #-}-{-# LANGUAGE ExistentialQuantification #-}-{-# LANGUAGE FlexibleContexts #-}-{-# LANGUAGE FlexibleInstances #-}-{-# LANGUAGE LambdaCase #-}-{-# LANGUAGE MultiParamTypeClasses #-}-{-# LANGUAGE OverloadedStrings #-}-{-# 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 Network.Haskoin.Node.Chain- ( chain- ) where--import Control.Monad (forever, guard, unless, void,- when)-import Control.Monad.Except (runExceptT, throwError)-import Control.Monad.Logger (MonadLoggerIO, logErrorS,- logInfoS)-import Control.Monad.Reader (MonadReader, 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.List (delete, nub)-import Data.Maybe (isJust, isNothing, listToMaybe)-import Data.Serialize (Serialize, get, getWord8, put,- putWord8)-import Data.String.Conversions (cs)-import Data.Time.Clock.POSIX (getPOSIXTime)-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, blockLocator,- connectBlocks, genesisNode,- getAncestor, headerHash,- splitPoint)-import Network.Haskoin.Node.Common (Chain, ChainConfig (..),- ChainEvent (..),- ChainMessage (..), Peer,- PeerException (..), killPeer,- myVersion, sendMessage)-import NQE (Inbox, inboxToMailbox, receive,- send)-import System.Random (randomRIO)-import UnliftIO (MonadIO, MonadUnliftIO, TVar,- atomically, liftIO, link,- modifyTVar, newTVarIO, readTVar,- readTVarIO, withAsync, writeTVar)-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 <- chainConfEvents <$> asks myReader- case e of- ChainBestBlock b ->- $(logInfoS) "Chain" $- "Best block header at height: " <> cs (show (nodeHeight b))- ChainSynced b ->- $(logInfoS) "Chain" $- "Headers now in sync at height: " <> cs (show (nodeHeight b))- atomically $ l e--processHeaders ::- MonadChain m => Peer -> [BlockHeader] -> m ()-processHeaders p hs =- void . runMaybeT $ do- net <- chainConfNetwork <$> asks myReader- unless (null hs) . $(logInfoS) "Chain" $- "Importing " <> cs (show (length hs)) <> " headers"- now <- round <$> liftIO getPOSIXTime- pbest <- getBestBlockHeader- importHeaders net now hs >>= \case- Left e -> do- $(logErrorS) "Chain" "Could not connect received headers"- 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--syncNewPeer :: MonadChain m => m ()-syncNewPeer =- getSyncingPeer >>= \case- Nothing -> do- nextPeer >>= \case- Nothing -> return ()- Just p -> syncPeer p- Just _ -> return ()--syncNotif :: MonadChain m => m ()-syncNotif =- round <$> liftIO getPOSIXTime >>= notifySynced >>= \x ->- when x $ getBestBlockHeader >>= chainEvent . ChainSynced--syncPeer :: MonadChain m => Peer -> m ()-syncPeer p = do- bb <-- chainSyncingPeer >>= \case- Just ChainSync {chainSyncPeer = p', chainHighest = Just g}- | p == p' -> return g- _ -> getBestBlockHeader- now <- round <$> liftIO getPOSIXTime- gh <- syncHeaders now bb p- MGetHeaders gh `sendMessage` p--chainMessage :: MonadChain m => ChainMessage -> m ()-chainMessage (ChainGetBest reply) = getBestBlockHeader >>= atomically . reply-chainMessage (ChainHeaders p hs) = processHeaders p hs-chainMessage (ChainPeerConnected p _) = do- 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- $(logErrorS) "Chain" "Syncing peer timed out"- PeerTimeout `killPeer` p- | otherwise -> return ()--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---- | Version of the database.-dataVersion :: Word32-dataVersion = 1---- | Database key for version.-data ChainDataVersionKey = ChainDataVersionKey- deriving (Eq, Ord, Show)--instance Key ChainDataVersionKey-instance KeyValue ChainDataVersionKey Word32--instance Serialize ChainDataVersionKey where- get = do- guard . (== 0x92) =<< getWord8- return ChainDataVersionKey- put ChainDataVersionKey = putWord8 0x92--data ChainSync p = ChainSync- { chainSyncPeer :: !p- , chainTimestamp :: !Timestamp- , chainHighest :: !(Maybe BlockNode)- }---- | Mutable state for the header chain process.-data ChainState p = ChainState- { chainSyncing :: !(Maybe (ChainSync p))- -- ^ peer to sync against and time of last received message- , newPeers :: ![p]- -- ^ queue of peers to sync against- , mySynced :: !Bool- -- ^ has the header chain ever been considered synced?- }---- | Key for block header in database.-newtype BlockHeaderKey = BlockHeaderKey BlockHash deriving (Eq, Show)--instance Serialize BlockHeaderKey where- get = do- guard . (== 0x90) =<< getWord8- BlockHeaderKey <$> get- put (BlockHeaderKey bh) = do- putWord8 0x90- put bh---- | Key for best block in database.-data BestBlockKey = BestBlockKey deriving (Eq, Show)--instance KeyValue BlockHeaderKey BlockNode-instance KeyValue BestBlockKey BlockNode--instance Serialize BestBlockKey where- get = do- guard . (== 0x91) =<< getWord8- return BestBlockKey- put BestBlockKey = putWord8 0x91---- | Type alias for monad commonly used in this module.-type MonadChainLogic a p m- = (BlockHeaders m, MonadReader (ChainReader a p) m)---- | Reader for header synchronization code.-data ChainReader a p = ChainReader- { myReader :: !a- -- ^ placeholder for upstream data- , myChainDB :: !DB- -- ^ database handle- , chainState :: !(TVar (ChainState p))- -- ^ mutable state for header synchronization- }--instance (Monad m, MonadIO m, MonadReader (ChainReader a p) m) =>- BlockHeaders m where- addBlockHeader bn = do- db <- asks myChainDB- insert db (BlockHeaderKey (headerHash (nodeHeader bn))) bn- getBlockHeader bh = do- db <- asks myChainDB- retrieve db def (BlockHeaderKey bh)- getBestBlockHeader = do- db <- asks myChainDB- retrieve db def BestBlockKey >>= \case- Nothing -> error "Could not get best block from database"- Just b -> return b- setBestBlockHeader bn = do- db <- asks myChainDB- insert db BestBlockKey bn- addBlockHeaders bns = do- db <- asks myChainDB- writeBatch db (map f bns)- where- f bn = insertOp (BlockHeaderKey (headerHash (nodeHeader bn))) bn---- | Initialize header database. If version is different from current, the--- database is purged of conflicting elements first.-initChainDB :: (MonadChainLogic a p m, MonadUnliftIO m) => Network -> m ()-initChainDB net = do- db <- asks myChainDB- ver <- retrieve db def ChainDataVersionKey- when (ver /= Just dataVersion) $ purgeChainDB >>= writeBatch db- insert db ChainDataVersionKey dataVersion- retrieve db def BestBlockKey >>= \b ->- when (isNothing (b :: Maybe BlockNode)) $ do- addBlockHeader (genesisNode net)- setBestBlockHeader (genesisNode net)---- | Purge database of elements having keys that may conflict with those used in--- this module.-purgeChainDB :: (MonadChainLogic a p m, MonadUnliftIO m) => m [R.BatchOp]-purgeChainDB = do- db <- asks myChainDB- runResourceT . R.withIterator db def $ \it -> do- R.iterSeek it $ B.singleton 0x90- recurse_delete it db- where- recurse_delete it db =- R.iterKey it >>= \case- Just k- | B.head k == 0x90 || B.head k == 0x91 -> do- R.delete db def k- R.iterNext it- (R.Del k :) <$> recurse_delete it db- _ -> return []---- | Import a bunch of continuous headers. Returns 'True' if the number of--- headers is 2000, which means that there are possibly more headers to sync--- from whatever peer delivered these.-importHeaders ::- (MonadIO m, BlockHeaders m, MonadChainLogic a p m)- => Network- -> Timestamp- -> [BlockHeader]- -> m (Either PeerException Bool)-importHeaders net now hs =- runExceptT $- lift (connectBlocks net now hs) >>= \case- Right _ -> do- case hs of- [] -> return ()- _ -> do- bb <- getBlockHeader (headerHash (last hs))- box <- asks chainState- atomically . modifyTVar box $ \s ->- s- { chainSyncing =- (\x -> x {chainHighest = bb}) <$>- chainSyncing s- }- case length hs of- 2000 -> return False- _ -> return True- Left _ -> throwError PeerSentBadHeaders---- | Check if best block header is in sync with the rest of the block chain by--- comparing the best block with the current time, verifying that there are no--- peers in the queue to be synced, and no peer is being synced at the moment.--- This function will only return 'True' once. It should be used to decide--- whether to notify other processes that the header chain has been synced. The--- state of the chain will be flipped to synced when this function returns--- 'True'.-notifySynced :: (MonadIO m, MonadChainLogic a p m) => Timestamp -> m Bool-notifySynced now =- fmap isJust $- runMaybeT $ do- bb <- getBestBlockHeader- guard (now - blockTimestamp (nodeHeader bb) < 2 * 60 * 60)- st <- asks chainState- MaybeT . atomically . runMaybeT $ do- s <- lift $ readTVar st- guard . isNothing $ chainSyncing s- guard . null $ newPeers s- guard . not $ mySynced s- lift $ writeTVar st s {mySynced = True}- return ()---- | Get next peer to sync against from the queue.-nextPeer :: (MonadIO m, MonadChainLogic a p m) => m (Maybe p)-nextPeer = listToMaybe . newPeers <$> (asks chainState >>= readTVarIO)---- | Set a syncing peer and generate a 'GetHeaders' data structure with a block--- locator to send to that peer for syncing.-syncHeaders ::- (Eq p, MonadChainLogic a p m, MonadIO m)- => Timestamp- -> BlockNode- -> p- -> m GetHeaders-syncHeaders now bb p = do- st <- asks chainState- atomically . modifyTVar st $ \s ->- s- { chainSyncing =- Just- ChainSync- { chainSyncPeer = p- , chainTimestamp = now- , chainHighest = Nothing- }- , newPeers = delete p (newPeers s)- }- loc <- blockLocator bb- return- GetHeaders- { getHeadersVersion = myVersion- , getHeadersBL = loc- , getHeadersHashStop =- "0000000000000000000000000000000000000000000000000000000000000000"- }---- | Set the time of last received data to now if a syncing peer is active.-setLastReceived :: (MonadChainLogic a p m, MonadIO m) => Timestamp -> m ()-setLastReceived now = do- st <- asks chainState- atomically . modifyTVar st $ \s ->- s {chainSyncing = (\p -> p {chainTimestamp = now}) <$> chainSyncing s}---- | Add a new peer to the queue of peers to sync against.-addPeer :: (Eq p, MonadIO m, MonadChainLogic a p m) => p -> m ()-addPeer p = do- st <- asks chainState- atomically . modifyTVar st $ \s -> s {newPeers = nub (p : newPeers s)}---- | Get syncing peer if there is one.-getSyncingPeer :: (MonadChainLogic a p m, MonadIO m) => m (Maybe p)-getSyncingPeer = fmap chainSyncPeer . chainSyncing <$> (readTVarIO =<< asks chainState)---- | Return 'True' if the chain has ever been considered synced. it will always--- return 'True', even if the chain gets out of sync for any reason.-isSynced :: (MonadChainLogic a p m, MonadIO m) => m Bool-isSynced = mySynced <$> (asks chainState >>= readTVarIO)---- | Remove a peer from the queue of peers to sync and unset the syncing peer if--- it is set to the provided value.-finishPeer :: (Eq p, MonadIO m, MonadChainLogic a p m) => p -> m ()-finishPeer p =- asks chainState >>= \st ->- atomically . modifyTVar st $ \s ->- s- { newPeers = delete p (newPeers s)- , chainSyncing =- case chainSyncing s of- Just ChainSync { chainSyncPeer = p'- , chainTimestamp = _- , chainHighest = _- }- | p == p' -> Nothing- _ -> chainSyncing s- }---- | Return syncing peer data.-chainSyncingPeer :: (MonadChainLogic a p m, MonadIO m) => m (Maybe (ChainSync p))-chainSyncingPeer = chainSyncing <$> (readTVarIO =<< asks chainState)
− src/Network/Haskoin/Node/Common.hs
@@ -1,541 +0,0 @@-{-# LANGUAGE FlexibleContexts #-}-{-# LANGUAGE LambdaCase #-}-{-# LANGUAGE MultiParamTypeClasses #-}-{-# 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 Network.Haskoin.Node.Common where--import Control.Monad (join)-import Control.Monad.Trans.Maybe (MaybeT (..), runMaybeT)-import Data.Conduit.Network (AppData, clientSettings,- runGeneralTCPClient)-import Data.Function (on)-import Data.List (union)-import Data.Maybe (fromMaybe, isJust)-import Data.String.Conversions (cs)-import Data.Time.Clock (NominalDiffTime, UTCTime)-import Data.Word (Word32, Word64)-import Database.RocksDB (DB)-import Network.Haskoin.Block (Block (..), BlockHash,- BlockHeader (..), BlockHeight,- BlockNode (..), getBlockHash,- headerHash)-import Network.Haskoin.Constants (Network (..))-import Network.Haskoin.Network (GetData (..), InvType (..),- InvVector (..), Message (..),- NetworkAddress (..),- NotFound (..), Ping (..),- Pong (..), VarString (..),- Version (..))-import Network.Haskoin.Transaction (Tx, TxHash, getTxHash, 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- , 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 Manager = Mailbox ManagerMessage---- | 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- }---- | Peer manager configuration.-data ManagerConfig = ManagerConfig- { mgrConfMaxPeers :: !Int- -- ^ maximum number of peers to connect to- , mgrConfPeers :: ![HostPort]- -- ^ static list of peers to connect to- , mgrConfDiscover :: !Bool- -- ^ activate peer discovery- , mgrConfNetAddr :: !NetworkAddress- -- ^ network address for the local host- , mgrConfNetwork :: !Network- -- ^ network constants- , mgrConfEvents :: !(Listen PeerEvent)- -- ^ send manager and peer messages to this mailbox- , mgrConfTimeout :: !Int- -- ^ timeout in seconds- }---- | Messages that can be sent to the peer manager.-data ManagerMessage- = ManagerConnect- -- ^ try to connect to peers- | ManagerGetPeers !(Listen [OnlinePeer])- -- ^ get all connected peers- | ManagerGetOnlinePeer !Peer !(Listen (Maybe OnlinePeer))- -- ^ get a peer information- | ManagerCheckPeer !Peer- -- ^ check this peer- | ManagerPeerMessage !Peer !Message- -- ^ peer got a message that is forwarded to manager- | ManagerPeerDied !Child !(Maybe SomeException)- -- ^ child died- | ManagerBestBlock !BlockHeight- -- ^ set this as our best block---- | Configuration for chain syncing process.-data ChainConfig = ChainConfig- { chainConfDB :: !DB- -- ^ database handle- , 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- }---- | 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 -> Manager -> m ()-managerPeerMessage p msg mgr = ManagerPeerMessage p msg `send` mgr---- | Get list of connected peers from manager.-managerGetPeers ::- MonadIO m => Manager -> m [OnlinePeer]-managerGetPeers mgr = ManagerGetPeers `query` mgr---- | Get information for an online peer from manager.-managerGetPeer :: MonadIO m => Peer -> Manager -> m (Maybe OnlinePeer)-managerGetPeer p mgr = ManagerGetOnlinePeer 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 -> Manager -> m ()-managerCheck p mgr = ManagerCheckPeer p `send` mgr---- | Internal function used to ask manager to connect to a new peer.-managerConnect :: MonadIO m => Manager -> m ()-managerConnect mgr = ManagerConnect `send` mgr---- | Set the best block that the manager knows about.-managerSetBest :: MonadIO m => BlockHeight -> Manager -> m ()-managerSetBest bh mgr = ManagerBestBlock bh `send` mgr---- | Send a network message to peer.-sendMessage :: MonadIO m => Message -> Peer -> m ()-sendMessage msg p = 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 ::- MonadUnliftIO m => SockAddr -> (AppData -> m a) -> m a-withConnection na f =- fromSockAddr na >>= \case- Nothing -> throwIO PeerAddressInvalid- Just (host, port) ->- let cset = clientSettings port (cs host)- in runGeneralTCPClient cset f---- | Calculate the median value from a list. The list must not be empty.-median :: Fractional a => [a] -> Maybe a-median ls- | null ls = Nothing- | length ls `mod` 2 == 0 =- Just . (/ 2) . sum . take 2 $ drop (length ls `div` 2 - 1) ls- | otherwise = Just . head $ drop (length ls `div` 2) ls
− src/Network/Haskoin/Node/Manager.hs
@@ -1,539 +0,0 @@-{-# LANGUAGE ConstraintKinds #-}-{-# LANGUAGE FlexibleContexts #-}-{-# LANGUAGE FlexibleInstances #-}-{-# LANGUAGE LambdaCase #-}-{-# LANGUAGE MultiParamTypeClasses #-}-{-# 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 Network.Haskoin.Node.Manager- ( manager- ) where--import Control.Monad (forM_, forever, guard, when,- (<=<))-import Control.Monad.Except (ExceptT (..), runExceptT,- throwError)-import Control.Monad.Logger (MonadLogger, MonadLoggerIO,- logErrorS, logInfoS, logWarnS)-import Control.Monad.Reader (MonadReader, asks, runReaderT)-import Control.Monad.Trans (lift)-import Control.Monad.Trans.Maybe (MaybeT (..), runMaybeT)-import Data.Bits ((.&.))-import Data.List (find, nub, sort)-import Data.Maybe (isJust, isNothing)-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 (..),- hostToSockAddr, nodeNetwork,- sockToHostAddress)-import Network.Haskoin.Node.Common (HostPort, Manager,- ManagerConfig (..),- ManagerMessage (..),- OnlinePeer (..), Peer,- PeerConfig (..), PeerEvent (..),- PeerException (..), buildVersion,- killPeer, managerCheck,- sendMessage, toSockAddr)-import Network.Haskoin.Node.Peer (peer)-import Network.Socket (SockAddr (..))-import NQE (Child, Inbox, Strategy (..),- Supervisor, addChild,- inboxToMailbox, newMailbox,- receive, receiveMatch, send,- sendSTM, subscribe, unsubscribe,- withPublisher, withSupervisor)-import System.Random (randomIO, randomRIO)-import UnliftIO (Async, MonadIO, MonadUnliftIO,- STM, SomeException, TVar,- atomically, bracket, 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)---- | Reader for peer configuration and state.-data ManagerReader = ManagerReader- { myConfig :: !ManagerConfig- , mySupervisor :: !Supervisor- , myMailbox :: !Manager- , myBestBlock :: !(TVar BlockHeight)- , knownPeers :: !(TVar (Set SockAddr))- , onlinePeers :: !(TVar [OnlinePeer])- }---- | Peer Manager process. In order to fully start it needs to receive a--- 'ManageBestBlock' event.-manager ::- (MonadUnliftIO m, MonadLoggerIO m)- => ManagerConfig- -> Inbox ManagerMessage- -> m ()-manager cfg inbox =- withSupervisor (Notify f) $ \sup -> do- bb <- newTVarIO 0- kp <- newTVarIO Set.empty- ob <- newTVarIO []- let rd =- ManagerReader- { myConfig = cfg- , mySupervisor = sup- , myMailbox = mgr- , myBestBlock = bb- , knownPeers = kp- , onlinePeers = ob- }- go `runReaderT` rd- where- mgr = inboxToMailbox inbox- go = do- putBestBlock <=< receiveMatch inbox $ \case- ManagerBestBlock b -> Just b- _ -> Nothing- withConnectLoop mgr $- forever $ do- m <- receive inbox- managerMessage m- f (a, mex) = ManagerPeerDied a mex `sendSTM` mgr--putBestBlock :: MonadManager m => BlockHeight -> m ()-putBestBlock bb = asks myBestBlock >>= \b -> atomically $ writeTVar b bb--getBestBlock :: MonadManager m => m BlockHeight-getBestBlock = asks myBestBlock >>= readTVarIO--getNetwork :: MonadManager m => m Network-getNetwork = asks (mgrConfNetwork . myConfig)--loadPeers :: (MonadUnliftIO m, MonadManager m) => m ()-loadPeers = do- loadStaticPeers- loadNetSeeds--loadStaticPeers :: (MonadUnliftIO m, MonadManager m) => m ()-loadStaticPeers = do- xs <- asks (mgrConfPeers . myConfig)- mapM_ newPeer =<< concat <$> mapM toSockAddr xs--loadNetSeeds :: (MonadUnliftIO m, MonadManager m) => m ()-loadNetSeeds =- asks (mgrConfDiscover . myConfig) >>= \discover ->- when discover $ do- net <- getNetwork- ss <- concat <$> mapM toSockAddr (networkSeeds net)- mapM_ newPeer ss--logConnectedPeers :: MonadManager m => m ()-logConnectedPeers = do- m <- asks (mgrConfMaxPeers . myConfig)- l <- length <$> getConnectedPeers- $(logInfoS) "Manager" $- "Peers connected: " <> cs (show l) <> "/" <> cs (show m)--getOnlinePeers :: MonadManager m => m [OnlinePeer]-getOnlinePeers = asks onlinePeers >>= readTVarIO--getConnectedPeers :: MonadManager m => m [OnlinePeer]-getConnectedPeers = filter onlinePeerConnected <$> getOnlinePeers--forwardMessage :: MonadManager m => Peer -> Message -> m ()-forwardMessage p = managerEvent . PeerMessage p--managerEvent :: MonadManager m => PeerEvent -> m ()-managerEvent e = asks (mgrConfEvents . myConfig) >>= \l -> atomically $ l e--managerMessage :: (MonadUnliftIO m, MonadManager m) => ManagerMessage -> m ()-managerMessage (ManagerPeerMessage p (MVersion 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- case e of- Right () -> do- MVerAck `sendMessage` p- Left x -> do- $(logErrorS) "Manager" $- "Version rejected for peer " <> s <> ": " <> cs (show x)- killPeer x p--managerMessage (ManagerPeerMessage p MVerAck) = do- b <- asks onlinePeers- s <- atomically $ peerString b p- atomically (setPeerVerAck b p) >>= \case- Just o -> when (onlinePeerConnected o) $ announcePeer p- Nothing -> do- $(logErrorS) "Manager" $ "Received verack from unknown peer: " <> s- killPeer UnknownPeer p--managerMessage (ManagerPeerMessage _ (MAddr (Addr nas))) =- asks (mgrConfDiscover . myConfig) >>= \discover ->- when discover $ do- let sas = map (hostToSockAddr . naAddress . snd) nas- $(logInfoS) "Manager" $- "Adding " <> cs (show (length sas)) <> " new peers"- forM_ sas newPeer--managerMessage (ManagerPeerMessage p m@(MPong (Pong n))) = do- now <- liftIO getCurrentTime- b <- asks onlinePeers- atomically (gotPong b n now p) >>= \case- Nothing -> forwardMessage p m- Just _ -> return ()--managerMessage (ManagerPeerMessage p (MPing (Ping n))) =- MPong (Pong n) `sendMessage` p--managerMessage (ManagerPeerMessage p m) = forwardMessage p m--managerMessage (ManagerBestBlock h) = putBestBlock h--managerMessage ManagerConnect = do- l <- length <$> getConnectedPeers- x <- asks (mgrConfMaxPeers . myConfig)- when (l < x) $- getNewPeer >>= \case- Nothing -> return ()- Just sa -> connectPeer sa--managerMessage (ManagerPeerDied a e) = processPeerOffline a e--managerMessage (ManagerGetPeers reply) = do- ps <- getConnectedPeers- atomically $ reply ps--managerMessage (ManagerGetOnlinePeer p reply) = do- b <- asks onlinePeers- atomically $ findPeer b p >>= reply--managerMessage (ManagerCheckPeer p) = checkPeer p--checkPeer :: MonadManager m => Peer -> m ()-checkPeer p = do- ManagerConfig {mgrConfTimeout = to} <- 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 - 1800) (killPeer PeerTooOld p)- atomically (lastPing b p) >>= \case- Nothing -> pingPeer p- Just t -> do- now <- liftIO getCurrentTime- when (diffUTCTime now t > fromIntegral to) $ do- $(logErrorS) "Manager" $- "Peer " <> s <> " did not respond ping on time"- killPeer PeerTimeout p--pingPeer :: MonadManager m => Peer -> m ()-pingPeer p = do- b <- asks onlinePeers- s <- atomically $ peerString b p- atomically (findPeer b p) >>= \case- Nothing -> $(logErrorS) "Manager" $ "Will not ping unknown peer " <> s- Just o- | onlinePeerConnected o -> do- n <- liftIO randomIO- now <- liftIO getCurrentTime- atomically (setPeerPing b n now p)- MPing (Ping n) `sendMessage` p- | otherwise -> return ()--processPeerOffline :: MonadManager m => Child -> Maybe SomeException -> m ()-processPeerOffline a e = do- b <- asks onlinePeers- atomically (findPeerAsync b a) >>= \case- Nothing -> log_unknown e- Just o -> do- let p = onlinePeerMailbox o- d = onlinePeerAddress o- s <- atomically $ peerString b p- if onlinePeerConnected o- then do- log_disconnected s e- managerEvent $ PeerDisconnected p d- else log_not_connect s e- atomically $ removePeer b p- logConnectedPeers- where- log_unknown Nothing = $(logErrorS) "Manager" "Disconnected unknown peer"- log_unknown (Just x) =- $(logErrorS) "Manager" $ "Unknown peer died: " <> cs (show x)- log_disconnected s Nothing =- $(logWarnS) "Manager" $ "Disconnected peer: " <> s- log_disconnected s (Just x) =- $(logErrorS) "Manager" $ "Peer " <> s <> " died: " <> cs (show x)- log_not_connect s Nothing =- $(logWarnS) "Manager" $ "Could not connect to peer " <> s- log_not_connect s (Just x) =- $(logErrorS) "Manager" $- "Could not connect to peer " <> s <> ": " <> cs (show x)--announcePeer :: MonadManager m => Peer -> m ()-announcePeer p = do- b <- asks onlinePeers- s <- atomically $ peerString b p- mgr <- asks myMailbox- atomically (findPeer b p) >>= \case- Just OnlinePeer {onlinePeerAddress = a, onlinePeerConnected = True} -> do- $(logInfoS) "Manager" $ "Handshake completed for peer " <> s- managerEvent $ PeerConnected p a- logConnectedPeers- managerCheck p mgr- Just OnlinePeer {onlinePeerConnected = False} -> return ()- Nothing -> $(logErrorS) "Manager" "Will not announce unknown peer"--getNewPeer :: (MonadUnliftIO m, MonadManager m) => m (Maybe SockAddr)-getNewPeer = runMaybeT $ lift loadPeers >> go- where- go = do- b <- asks knownPeers- ks <- readTVarIO b- guard . not $ Set.null ks- let xs = Set.toList ks- a <- liftIO $ randomRIO (0, length xs - 1)- let p = xs !! a- atomically . modifyTVar b $ Set.delete p- o <- asks onlinePeers- atomically (findPeerAddress o p) >>= \case- Nothing -> return p- Just _ -> go---connectPeer :: (MonadUnliftIO m, MonadManager m) => SockAddr -> m ()-connectPeer sa = do- os <- asks onlinePeers- atomically (findPeerAddress os sa) >>= \case- Just _ ->- $(logErrorS) "Manager" $- "Attempted to connect to peer twice: " <> cs (show sa)- Nothing -> do- $(logInfoS) "Manager" $ "Connecting to " <> cs (show sa)- ManagerConfig {mgrConfNetAddr = ad, mgrConfNetwork = net} <-- asks myConfig- mgr <- asks myMailbox- sup <- asks mySupervisor- nonce <- liftIO randomIO- bb <- getBestBlock- now <- round <$> liftIO getPOSIXTime- 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- }- a <- withRunInIO $ \io -> sup `addChild` io (launch mgr pc inbox p)- MVersion ver `sendMessage` p- b <- asks onlinePeers- _ <- atomically $ newOnlinePeer b sa nonce p a now- return ()- where- l mgr p m = ManagerPeerMessage 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--withPeerLoop ::- (MonadUnliftIO m, MonadLogger m)- => SockAddr- -> Peer- -> Manager- -> (Async a -> m a)- -> m a-withPeerLoop _ p mgr = withAsync go- where- go =- forever $ do- threadDelay =<<- liftIO (randomRIO (30 * 1000 * 1000, 90 * 1000 * 1000))- ManagerCheckPeer p `send` mgr--withConnectLoop :: (MonadLogger m, MonadUnliftIO m) => Manager -> m a -> m a-withConnectLoop mgr act = withAsync go (\a -> link a >> act)- where- go =- forever $ do- threadDelay =<<- liftIO (randomRIO (2 * 1000 * 1000, 10 * 1000 * 1000))- ManagerConnect `send` mgr---- | 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---- | 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---- | Return time of last ping sent to peer, if any.-lastPing :: TVar [OnlinePeer] -> Peer -> STM (Maybe UTCTime)-lastPing b p =- findPeer b p >>= \case- Just OnlinePeer {onlinePeerPing = Just (time, _)} -> return (Just time)- _ -> return Nothing---- | Set nonce and time of last ping sent to peer.-setPeerPing :: TVar [OnlinePeer] -> Word64 -> UTCTime -> Peer -> STM ()-setPeerPing b nonce now p =- modifyPeer b p $ \o -> o {onlinePeerPing = Just (now, nonce)}---- | Set version for online peer. Will set the peer connected status to 'True'--- if a verack message has already been registered for that peer.-setPeerVersion ::- TVar [OnlinePeer]- -> Peer- -> Version- -> STM (Either PeerException OnlinePeer)-setPeerVersion b p v =- runExceptT $ do- when (services v .&. nodeNetwork == 0) $ throwError NotNetworkPeer- ops <- lift $ readTVar b- when (any ((verNonce v ==) . onlinePeerNonce) ops) $- throwError PeerIsMyself- lift (findPeer b p) >>= \case- Nothing -> throwError UnknownPeer- Just o -> do- let n =- o- { onlinePeerVersion = Just v- , onlinePeerConnected = onlinePeerVerAck o- }- lift $ insertPeer b n- return n---- | Register that a verack message was received from a peer.-setPeerVerAck :: TVar [OnlinePeer] -> Peer -> STM (Maybe OnlinePeer)-setPeerVerAck b p =- runMaybeT $ do- o <- MaybeT $ findPeer b p- let n =- o- { onlinePeerVerAck = True- , onlinePeerConnected = isJust (onlinePeerVersion o)- }- lift $ insertPeer b n- return n---- | Create 'OnlinePeer' data structure.-newOnlinePeer ::- TVar [OnlinePeer]- -> SockAddr- -- ^ peer address- -> Word64- -- ^ nonce sent to peer- -> Peer- -- ^ peer mailbox- -> Async ()- -- ^ peer asynchronous handle- -> Word64- -- ^ when connection was established- -> 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- }- insertPeer b op- return op---- | Get a human-readable string for the peer address.-peerString :: TVar [OnlinePeer] -> Peer -> STM Text-peerString b p =- maybe "[unknown]" (cs . show . onlinePeerAddress) <$> findPeer b p---- | Find a connected peer.-findPeer :: TVar [OnlinePeer] -> Peer -> STM (Maybe OnlinePeer)-findPeer b p = find ((== p) . onlinePeerMailbox) <$> readTVar b---- | Insert or replace a connected peer.-insertPeer :: TVar [OnlinePeer] -> OnlinePeer -> STM ()-insertPeer b o = modifyTVar b $ \x -> sort . nub $ o : x---- | Modify an online peer.-modifyPeer :: TVar [OnlinePeer] -> Peer -> (OnlinePeer -> OnlinePeer) -> STM ()-modifyPeer b p f =- findPeer b p >>= \case- Nothing -> return ()- Just o -> insertPeer b $ f o---- | Remove an online peer.-removePeer :: TVar [OnlinePeer] -> Peer -> STM ()-removePeer b p = modifyTVar b $ \x -> filter ((/= p) . onlinePeerMailbox) x---- | Find online peer by asynchronous handle.-findPeerAsync :: TVar [OnlinePeer] -> Async () -> STM (Maybe OnlinePeer)-findPeerAsync b a = find ((== a) . onlinePeerAsync) <$> readTVar b--findPeerAddress :: TVar [OnlinePeer] -> SockAddr -> STM (Maybe OnlinePeer)-findPeerAddress b a = find ((== a) . onlinePeerAddress) <$> readTVar b
− src/Network/Haskoin/Node/Peer.hs
@@ -1,108 +0,0 @@-{-# LANGUAGE ConstraintKinds #-}-{-# LANGUAGE FlexibleContexts #-}-{-# LANGUAGE MultiParamTypeClasses #-}-{-# LANGUAGE OverloadedStrings #-}-{-# LANGUAGE RecordWildCards #-}-{-# LANGUAGE TemplateHaskell #-}-{-# 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 Network.Haskoin.Node.Peer- ( peer- ) where--import Conduit (ConduitT, awaitForever, foldC,- mapM_C, runConduit, takeCE, yield,- (.|))-import Control.Monad (forever, when)-import Control.Monad.Logger (MonadLoggerIO, logErrorS)-import Control.Monad.Reader (runReaderT)-import Data.ByteString (ByteString)-import qualified Data.ByteString as B-import Data.Conduit.Network (appSink, appSource)-import Data.Serialize (decode, runGet, runPut)-import Data.String.Conversions (cs)-import Data.Text (Text)-import Network.Haskoin.Constants (Network)-import Network.Haskoin.Network (Message, MessageHeader (..),- getMessage, putMessage)-import Network.Haskoin.Node.Common (PeerConfig (..),- PeerException (..),- PeerMessage (..), withConnection)-import Network.Socket (SockAddr)-import NQE (Inbox, PublisherMessage (..),- receive, send)-import UnliftIO (MonadUnliftIO, atomically, link,- throwIO, withAsync)---- | Run peer process in current thread.-peer ::- (MonadUnliftIO m, MonadLoggerIO m)- => PeerConfig- -> Inbox PeerMessage- -> m ()-peer pc inbox = withConnection a $ \ad -> runReaderT (peer_session ad) pc- where- a = peerConfAddress pc- go = forever $ do- receive inbox >>= dispatchMessage pc- net = peerConfNetwork pc- peer_session ad =- let ins = appSource ad- ons = appSink ad- src = runConduit $ ins .| inPeerConduit net a .| mapM_C send_msg- snk = outPeerConduit net .| ons- in withAsync src $ \as -> do- link as- runConduit (go .| snk)- send_msg = (`send` peerConfListen pc) . Event---- | Internal function to dispatch peer messages.-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---- | Internal conduit to parse messages coming from peer.-inPeerConduit ::- MonadLoggerIO m- => Network- -> SockAddr- -> ConduitT ByteString Message m ()-inPeerConduit net a = forever $ do- x <- takeCE 24 .| foldC- case decode x of- Left _ -> do- $(logErrorS)- (peerString a)- "Could not decode incoming message header"- throwIO DecodeHeaderError- Right (MessageHeader _ _ len _) -> do- when (len > 32 * 2 ^ (20 :: Int)) $ do- $(logErrorS) (peerString a) "Payload too large"- throwIO $ PayloadTooLarge len- y <- takeCE (fromIntegral len) .| foldC- case runGet (getMessage net) $ x `B.append` y of- Left e -> do- $(logErrorS) (peerString a) $- "Cannot decode payload: " <> cs (show e)- throwIO CannotDecodePayload- Right msg -> yield msg---- | Outgoing peer conduit to serialize and send messages.-outPeerConduit :: Monad m => Network -> ConduitT Message ByteString m ()-outPeerConduit net = awaitForever $ yield . runPut . putMessage net---- | Peer string for logging-peerString :: SockAddr -> Text-peerString a = "Peer<" <> cs (show a) <> ">"
test/Haskoin/NodeSpec.hs view
@@ -18,7 +18,7 @@ btcTest, buildMerkleRoot, getGenesisHeader, headerHash, sockToHostAddress, txHash)-import Haskoin.Node (Chain, ChainEvent (..), Manager,+import Haskoin.Node (Chain, ChainEvent (..), PeerManager, NodeConfig (..), NodeEvent (..), OnlinePeer (..), Peer, PeerEvent (..), chainGetAncestor, chainGetBest,@@ -32,7 +32,7 @@ withSystemTempDirectory) data TestNode = TestNode- { testMgr :: Manager+ { testMgr :: PeerManager , testChain :: Chain , nodeEvents :: Inbox NodeEvent }