diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,10 @@ 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.9.17 +### Added +- Peers are disconnected automatically after awhile. + ## 0.9.16 ### Added - Lower bound versions for some dependencies. diff --git a/haskoin-node.cabal b/haskoin-node.cabal --- a/haskoin-node.cabal +++ b/haskoin-node.cabal @@ -4,10 +4,10 @@ -- -- see: https://github.com/sol/hpack -- --- hash: 06fee6ff6ad8e216f353bb90fadf1bfd26beb1bc0b8fe56bbb1142e60f1cbcca +-- hash: 8669769b56f62758da5356e662ad42d29c33cc14cc5463ffcd0b81166f56777f name: haskoin-node -version: 0.9.16 +version: 0.9.17 synopsis: Haskoin Node P2P library for Bitcoin and Bitcoin Cash description: Bitcoin and Bitcoin Cash peer-to-peer protocol library featuring headers-first synchronisation. category: Bitcoin, Finance, Network diff --git a/src/Haskoin/Node.hs b/src/Haskoin/Node.hs --- a/src/Haskoin/Node.hs +++ b/src/Haskoin/Node.hs @@ -5,7 +5,7 @@ Module : Haskoin.Node Copyright : No rights reserved License : UNLICENSE -Maintainer : xenog@protonmail.com +Maintainer : jprupp@protonmail.ch Stability : experimental Portability : POSIX @@ -21,8 +21,9 @@ , Peer , Chain , Manager - , ChainMessage - , ManagerMessage + , ChainMessage(..) + , ManagerMessage(..) + , PeerMessage(..) , OnlinePeer(..) , NodeConfig(..) , NodeEvent(..) @@ -48,13 +49,34 @@ , myVersion ) where -import Control.Monad.Logger -import Haskoin -import Network.Haskoin.Node.Chain -import Network.Haskoin.Node.Common -import Network.Haskoin.Node.Manager -import NQE -import UnliftIO +import Control.Monad.Logger (MonadLoggerIO) +import Haskoin (BlockNode (..), Headers (..), + Message (..)) +import Network.Haskoin.Node.Chain (chain) +import Network.Haskoin.Node.Common (Chain, ChainConfig (..), + ChainEvent (..), + ChainMessage (..), Host, + HostPort, Manager, + ManagerConfig (..), + ManagerMessage (..), + NodeConfig (..), NodeEvent (..), + OnlinePeer (..), Peer, + PeerEvent (..), + PeerException (..), + PeerMessage (..), Port, + chainBlockMain, chainGetAncestor, + chainGetBest, chainGetBlock, + chainGetParents, + chainGetSplitBlock, + chainIsSynced, killPeer, + managerGetPeer, managerGetPeers, + myVersion, peerGetBlocks, + peerGetPublisher, peerGetTxs, + sendMessage) +import Network.Haskoin.Node.Manager (manager) +import NQE (Inbox, inboxToMailbox, newInbox, + sendSTM) +import UnliftIO (MonadUnliftIO, link, withAsync) -- | Launch a node in the background. Pass a 'Manager' and 'Chain' to a -- function. Node will stop once the function ends. @@ -118,4 +140,4 @@ nodeConfEvents cfg $ ChainEvent event case event of ChainBestBlock b -> ManagerBestBlock (nodeHeight b) `sendSTM` mgr - _ -> return () + _ -> return () diff --git a/src/Network/Haskoin/Node/Chain.hs b/src/Network/Haskoin/Node/Chain.hs --- a/src/Network/Haskoin/Node/Chain.hs +++ b/src/Network/Haskoin/Node/Chain.hs @@ -12,7 +12,7 @@ Module : Network.Haskoin.Node.Chain Copyright : No rights reserved License : UNLICENSE -Maintainer : xenog@protonmail.com +Maintainer : jprupp@protonmail.ch Stability : experimental Portability : POSIX @@ -22,28 +22,48 @@ ( chain ) where -import Control.Monad.Except -import Control.Monad.Logger -import Control.Monad.Reader -import Control.Monad.Trans.Maybe -import qualified Data.ByteString as B -import Data.Default -import Data.List -import Data.Maybe -import Data.Serialize as S -import Data.String.Conversions -import Data.Time.Clock.POSIX -import Data.Word -import Database.RocksDB (DB) -import qualified Database.RocksDB as R -import Database.RocksDB.Query as R -import Haskoin -import Network.Haskoin.Node.Common -import NQE -import System.Random -import UnliftIO -import UnliftIO.Concurrent -import UnliftIO.Resource +import Control.Monad (forever, guard, void, when) +import Control.Monad.Except (runExceptT, throwError) +import Control.Monad.Logger (MonadLoggerIO, logDebugS, + logErrorS, logInfoS, logWarnS) +import Control.Monad.Reader (MonadReader, asks, runReaderT) +import Control.Monad.Trans (lift) +import Control.Monad.Trans.Maybe (MaybeT (..), runMaybeT) +import qualified Data.ByteString as B +import Data.Default (def) +import Data.List (delete, nub) +import Data.Maybe (isJust, isNothing, listToMaybe) +import Data.Serialize (Serialize, get, getWord8, put, + putWord8) +import Data.String.Conversions (cs) +import Data.Time.Clock.POSIX (getPOSIXTime) +import Data.Word (Word32) +import Database.RocksDB (DB) +import qualified Database.RocksDB as R +import Database.RocksDB.Query (Key, KeyValue, insert, insertOp, + retrieve, writeBatch) +import Haskoin (BlockHash, BlockHeader (..), + BlockHeaders (..), BlockNode (..), + GetHeaders (..), Message (..), + Network, Timestamp, + blockHashToHex, blockLocator, + connectBlocks, genesisNode, + getAncestor, headerHash, + splitPoint) +import Network.Haskoin.Node.Common (Chain, ChainConfig (..), + ChainEvent (..), + ChainMessage (..), Peer, + PeerException (..), killPeer, + myVersion, sendMessage) +import NQE (Inbox, inboxToMailbox, receive, + send) +import System.Random (randomRIO) +import UnliftIO (MonadIO, MonadUnliftIO, TVar, + atomically, liftIO, link, + modifyTVar, newTVarIO, readTVar, + readTVarIO, withAsync, writeTVar) +import UnliftIO.Concurrent (threadDelay) +import UnliftIO.Resource (runResourceT) type MonadChain m = (MonadLoggerIO m, MonadChainLogic ChainConfig Peer m) @@ -244,14 +264,14 @@ data ChainDataVersionKey = ChainDataVersionKey deriving (Eq, Ord, Show) -instance R.Key ChainDataVersionKey +instance Key ChainDataVersionKey instance KeyValue ChainDataVersionKey Word32 instance Serialize ChainDataVersionKey where get = do - guard . (== 0x92) =<< S.getWord8 + guard . (== 0x92) =<< getWord8 return ChainDataVersionKey - put ChainDataVersionKey = S.putWord8 0x92 + put ChainDataVersionKey = putWord8 0x92 data ChainSync p = ChainSync { chainSyncPeer :: !p @@ -310,7 +330,7 @@ BlockHeaders m where addBlockHeader bn = do db <- asks myChainDB - R.insert db (BlockHeaderKey (headerHash (nodeHeader bn))) bn + insert db (BlockHeaderKey (headerHash (nodeHeader bn))) bn getBlockHeader bh = do db <- asks myChainDB retrieve db def (BlockHeaderKey bh) @@ -321,7 +341,7 @@ Just b -> return b setBestBlockHeader bn = do db <- asks myChainDB - R.insert db BestBlockKey bn + insert db BestBlockKey bn addBlockHeaders bns = do db <- asks myChainDB writeBatch db (map f bns) @@ -335,7 +355,7 @@ db <- asks myChainDB ver <- retrieve db def ChainDataVersionKey when (ver /= Just dataVersion) $ purgeChainDB >>= writeBatch db - R.insert db ChainDataVersionKey dataVersion + insert db ChainDataVersionKey dataVersion retrieve db def BestBlockKey >>= \b -> when (isNothing (b :: Maybe BlockNode)) $ do addBlockHeader (genesisNode net) diff --git a/src/Network/Haskoin/Node/Common.hs b/src/Network/Haskoin/Node/Common.hs --- a/src/Network/Haskoin/Node/Common.hs +++ b/src/Network/Haskoin/Node/Common.hs @@ -6,7 +6,7 @@ Module : Network.Haskoin.Node.Common Copyright : No rights reserved License : UNLICENSE -Maintainer : xenog@protonmail.com +Maintainer : jprupp@protonmail.ch Stability : experimental Portability : POSIX @@ -14,26 +14,43 @@ -} module Network.Haskoin.Node.Common where -import Conduit -import Control.Monad -import Control.Monad.Trans.Maybe -import Data.Conduit.Network -import Data.Function -import Data.List -import Data.Maybe -import Data.String.Conversions -import Data.Time.Clock -import Data.Word +import Control.Monad (join) +import Control.Monad.Trans.Maybe (MaybeT (..), runMaybeT) +import Data.Conduit.Network (AppData, clientSettings, + runGeneralTCPClient) +import Data.Function (on) +import Data.List (union) +import Data.Maybe (fromMaybe, isJust) +import Data.String.Conversions (cs) +import Data.Time.Clock (NominalDiffTime, UTCTime) +import Data.Word (Word32, Word64) import Database.RocksDB (DB) -import Network.Haskoin.Block -import Network.Haskoin.Constants -import Network.Haskoin.Network -import Network.Haskoin.Transaction -import Network.Socket hiding (send) -import NQE -import System.Random -import Text.Read -import UnliftIO +import Network.Haskoin.Block (Block (..), BlockHash, + BlockHeader (..), BlockHeight, + BlockNode (..), getBlockHash, + headerHash) +import Network.Haskoin.Constants (Network (..)) +import Network.Haskoin.Network (GetData (..), InvType (..), + InvVector (..), Message (..), + NetworkAddress (..), + NotFound (..), Ping (..), + Pong (..), VarString (..), + Version (..)) +import Network.Haskoin.Transaction (Tx, TxHash, getTxHash, txHash) +import Network.Socket (AddrInfo (..), AddrInfoFlag (..), + Family (..), NameInfoFlag (..), + SockAddr, SocketType (..), + defaultHints, getAddrInfo, + getNameInfo) +import NQE (Child, Listen, Mailbox, Publisher, + query, queryS, receive, + receiveMatchS, send, + withSubscription) +import System.Random (randomIO) +import Text.Read (readMaybe) +import UnliftIO (Async (..), Exception, MonadIO, + MonadUnliftIO, SomeException, + catch, liftIO, throwIO, timeout) -- | Type alias for a combination of hostname and port. type HostPort = (Host, Port) @@ -46,24 +63,26 @@ -- | Data structure representing an online peer. data OnlinePeer = OnlinePeer - { onlinePeerAddress :: !SockAddr + { onlinePeerAddress :: !SockAddr -- ^ network address - , onlinePeerVerAck :: !Bool + , onlinePeerVerAck :: !Bool -- ^ got version acknowledgement from peer - , onlinePeerConnected :: !Bool + , onlinePeerConnected :: !Bool -- ^ peer is connected and ready - , onlinePeerVersion :: !(Maybe Version) + , onlinePeerVersion :: !(Maybe Version) -- ^ protocol version - , onlinePeerAsync :: !(Async ()) + , onlinePeerAsync :: !(Async ()) -- ^ peer asynchronous process - , onlinePeerMailbox :: !Peer + , onlinePeerMailbox :: !Peer -- ^ peer mailbox - , onlinePeerNonce :: !Word64 + , onlinePeerNonce :: !Word64 -- ^ random nonce sent during handshake - , onlinePeerPing :: !(Maybe (UTCTime, Word64)) + , onlinePeerPing :: !(Maybe (UTCTime, Word64)) -- ^ last sent ping time and nonce - , onlinePeerPings :: ![NominalDiffTime] + , onlinePeerPings :: ![NominalDiffTime] -- ^ last few ping rountrip duration + , onlinePeerConnectTime :: !Word64 + -- ^ when connection was opened } instance Eq OnlinePeer where @@ -232,6 +251,8 @@ -- ^ request to peer timed out | UnknownPeer -- ^ peer is unknown + | PeerTooOld + -- ^ peer has been connected too long deriving (Eq, Show) instance Exception PeerException diff --git a/src/Network/Haskoin/Node/Manager.hs b/src/Network/Haskoin/Node/Manager.hs --- a/src/Network/Haskoin/Node/Manager.hs +++ b/src/Network/Haskoin/Node/Manager.hs @@ -10,7 +10,7 @@ Module : Network.Haskoin.Node.Manager Copyright : No rights reserved License : UNLICENSE -Maintainer : xenog@protonmail.com +Maintainer : jprupp@protonmail.ch Stability : experimental Portability : POSIX @@ -20,30 +20,58 @@ ( manager ) where -import Conduit -import Control.Monad -import Control.Monad.Except -import Control.Monad.Logger -import Control.Monad.Reader -import Control.Monad.Trans.Maybe -import Data.Bits -import Data.List -import Data.Maybe +import Control.Monad (forM_, forever, guard, when, + (<=<)) +import Control.Monad.Except (ExceptT (..), runExceptT, + throwError) +import Control.Monad.Logger (MonadLogger, MonadLoggerIO, + logDebugS, logErrorS, logInfoS, + logWarnS) +import Control.Monad.Reader (MonadReader, asks, runReaderT) +import Control.Monad.Trans (lift) +import Control.Monad.Trans.Maybe (MaybeT (..), runMaybeT) +import Data.Bits ((.&.)) +import Data.List (find, nub, sort) +import Data.Maybe (isJust, isNothing) import Data.Set (Set) import qualified Data.Set as Set -import Data.String.Conversions +import Data.String.Conversions (cs) import Data.Text (Text) -import Data.Time.Clock -import Data.Time.Clock.POSIX -import Data.Word -import Haskoin -import Network.Haskoin.Node.Common -import Network.Haskoin.Node.Peer +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 (..), VarString (..), + Version (..), commandToString, + hostToSockAddr, msgType, + nodeNetwork, sockToHostAddress) +import Network.Haskoin.Node.Common (HostPort, Manager, + ManagerConfig (..), + ManagerMessage (..), + OnlinePeer (..), Peer, + PeerConfig (..), PeerEvent (..), + PeerException (..), buildVersion, + killPeer, managerCheck, + sendMessage, toSockAddr) +import Network.Haskoin.Node.Peer (peer) import Network.Socket (SockAddr (..)) -import NQE -import System.Random -import UnliftIO -import UnliftIO.Concurrent +import NQE (Child, Inbox, Strategy (..), + Supervisor, addChild, + inboxToMailbox, newMailbox, + receive, receiveMatch, send, + sendSTM, subscribe, unsubscribe, + withPublisher, withSupervisor) +import System.Random (randomIO, randomRIO) +import UnliftIO (Async, MonadIO, MonadUnliftIO, + STM, SomeException, TVar, + atomically, bracket, liftIO, link, + modifyTVar, newTVarIO, readTVar, + readTVarIO, withAsync, + withRunInIO, writeTVar) +import UnliftIO.Concurrent (threadDelay) -- | Monad used by most functions in this module. type MonadManager m = (MonadLoggerIO m, MonadReader ManagerReader m) @@ -82,7 +110,6 @@ go `runReaderT` rd where mgr = inboxToMailbox inbox - discover = mgrConfDiscover cfg go = do $(logDebugS) "Manager" "Initializing..." putBestBlock <=< receiveMatch inbox $ \case @@ -267,6 +294,11 @@ b <- asks onlinePeers s <- atomically $ peerString b p $(logDebugS) "Manager" $ "Checking on peer " <> s + atomically (findPeer b p) >>= \case + Nothing -> return () + Just o -> do + now <- round <$> liftIO getPOSIXTime + when (onlinePeerConnectTime o < now - 1800) (killPeer PeerTooOld p) atomically (lastPing b p) >>= \case Nothing -> pingPeer p Just t -> do @@ -387,7 +419,7 @@ a <- withRunInIO $ \io -> sup `addChild` io (launch mgr pc inbox p) MVersion ver `sendMessage` p b <- asks onlinePeers - _ <- atomically $ newOnlinePeer b sa nonce p a + _ <- atomically $ newOnlinePeer b sa nonce p a now return () where l mgr p m = ManagerPeerMessage p m `sendSTM` mgr @@ -429,10 +461,6 @@ ManagerConnect `send` mgr threadDelay =<< liftIO (randomRIO (2 * 1000 * 1000, 10 * 1000 * 1000)) --- | Database version. -versionPeerDB :: Word32 -versionPeerDB = 5 - -- | Add a peer. newPeer :: (MonadIO m, MonadManager m) => SockAddr -> m () newPeer sa = do @@ -529,8 +557,10 @@ -- ^ peer mailbox -> Async () -- ^ peer asynchronous handle + -> Word64 + -- ^ when connection was established -> STM OnlinePeer -newOnlinePeer b sa n p a = do +newOnlinePeer b sa n p a t = do let op = OnlinePeer { onlinePeerAddress = sa @@ -542,6 +572,7 @@ , onlinePeerNonce = n , onlinePeerPings = [] , onlinePeerPing = Nothing + , onlinePeerConnectTime = t } insertPeer b op return op diff --git a/src/Network/Haskoin/Node/Peer.hs b/src/Network/Haskoin/Node/Peer.hs --- a/src/Network/Haskoin/Node/Peer.hs +++ b/src/Network/Haskoin/Node/Peer.hs @@ -9,7 +9,7 @@ Module : Network.Haskoin.Node.Peer Copyright : No rights reserved License : UNLICENSE -Maintainer : xenog@protonmail.com +Maintainer : jprupp@protonmail.ch Stability : experimental Portability : POSIX @@ -19,22 +19,31 @@ ( peer ) where -import Conduit -import Control.Monad -import Control.Monad.Logger -import Control.Monad.Reader +import Conduit (ConduitT, awaitForever, foldC, + mapM_C, runConduit, takeCE, yield, + (.|)) +import Control.Monad (forever, when) +import Control.Monad.Logger (MonadLoggerIO, logDebugS, + logErrorS) +import Control.Monad.Reader (runReaderT) import Data.ByteString (ByteString) import qualified Data.ByteString as B -import Data.Conduit.Network -import Data.Serialize -import Data.String.Conversions +import Data.Conduit.Network (appSink, appSource) +import Data.Serialize (decode, runGet, runPut) +import Data.String.Conversions (cs) import Data.Text (Text) -import Network.Haskoin.Constants -import Network.Haskoin.Network -import Network.Haskoin.Node.Common +import Network.Haskoin.Constants (Network) +import Network.Haskoin.Network (Message, MessageHeader (..), + commandToString, getMessage, + msgType, putMessage) +import Network.Haskoin.Node.Common (PeerConfig (..), + PeerException (..), + PeerMessage (..), withConnection) import Network.Socket (SockAddr) -import NQE -import UnliftIO +import NQE (Inbox, PublisherMessage (..), + receive, send) +import UnliftIO (MonadUnliftIO, atomically, link, + throwIO, withAsync) -- | Run peer process in current thread. peer :: diff --git a/test/Haskoin/NodeSpec.hs b/test/Haskoin/NodeSpec.hs --- a/test/Haskoin/NodeSpec.hs +++ b/test/Haskoin/NodeSpec.hs @@ -7,17 +7,29 @@ ( spec ) where -import Control.Monad -import Control.Monad.Logger -import Control.Monad.Trans -import Data.Maybe +import Control.Monad (forM_, replicateM, replicateM_) +import Control.Monad.Logger (runNoLoggingT) +import Control.Monad.Trans (lift) +import Data.Maybe (isJust) import qualified Database.RocksDB as R -import Haskoin -import Haskoin.Node +import Haskoin (Block (..), BlockHeader (..), + BlockNode (..), Network, + NetworkAddress (..), Version (..), + btcTest, buildMerkleRoot, + getGenesisHeader, headerHash, + sockToHostAddress, txHash) +import Haskoin.Node (Chain, ChainEvent (..), Manager, + NodeConfig (..), NodeEvent (..), + OnlinePeer (..), Peer, PeerEvent (..), + chainGetAncestor, chainGetBest, + chainGetParents, managerGetPeer, + peerGetBlocks, peerGetTxs, withNode) import Network.Socket (SockAddr (..)) -import NQE -import Test.Hspec -import UnliftIO +import NQE (Inbox, newInbox, receiveMatch, sendSTM) +import Test.Hspec (Spec, describe, it, shouldBe, + shouldReturn, shouldSatisfy) +import UnliftIO (MonadIO, MonadUnliftIO, throwString, + withSystemTempDirectory) data TestNode = TestNode { testMgr :: Manager