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