diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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 diff --git a/haskoin-node.cabal b/haskoin-node.cabal --- a/haskoin-node.cabal +++ b/haskoin-node.cabal @@ -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: diff --git a/src/Haskoin/Node.hs b/src/Haskoin/Node.hs --- a/src/Haskoin/Node.hs +++ b/src/Haskoin/Node.hs @@ -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 () diff --git a/src/Haskoin/Node/Chain.hs b/src/Haskoin/Node/Chain.hs new file mode 100644 --- /dev/null +++ b/src/Haskoin/Node/Chain.hs @@ -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) diff --git a/src/Haskoin/Node/Common.hs b/src/Haskoin/Node/Common.hs new file mode 100644 --- /dev/null +++ b/src/Haskoin/Node/Common.hs @@ -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)) diff --git a/src/Haskoin/Node/Manager.hs b/src/Haskoin/Node/Manager.hs new file mode 100644 --- /dev/null +++ b/src/Haskoin/Node/Manager.hs @@ -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 diff --git a/src/Haskoin/Node/Peer.hs b/src/Haskoin/Node/Peer.hs new file mode 100644 --- /dev/null +++ b/src/Haskoin/Node/Peer.hs @@ -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 diff --git a/src/Network/Haskoin/Node/Chain.hs b/src/Network/Haskoin/Node/Chain.hs deleted file mode 100644 --- a/src/Network/Haskoin/Node/Chain.hs +++ /dev/null @@ -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) diff --git a/src/Network/Haskoin/Node/Common.hs b/src/Network/Haskoin/Node/Common.hs deleted file mode 100644 --- a/src/Network/Haskoin/Node/Common.hs +++ /dev/null @@ -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 diff --git a/src/Network/Haskoin/Node/Manager.hs b/src/Network/Haskoin/Node/Manager.hs deleted file mode 100644 --- a/src/Network/Haskoin/Node/Manager.hs +++ /dev/null @@ -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 diff --git a/src/Network/Haskoin/Node/Peer.hs b/src/Network/Haskoin/Node/Peer.hs deleted file mode 100644 --- a/src/Network/Haskoin/Node/Peer.hs +++ /dev/null @@ -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) <> ">" diff --git a/test/Haskoin/NodeSpec.hs b/test/Haskoin/NodeSpec.hs --- a/test/Haskoin/NodeSpec.hs +++ b/test/Haskoin/NodeSpec.hs @@ -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 }