diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,15 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/) and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.html). +## 0.9.7 +### Added +- More debugging. + +### Change +- Be defensive against duplicate peers. +- Increase interval between housekeeping pings. +- Replace peers in database atomically. + ## 0.9.6 ### Change - Randomize known peers instead of keeping scores. 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: 153005bf793d856dc57a198e58d683234d72138315022bdd1f65e5ef649d5109 +-- hash: 2a766eed5c0c8979b073e662fc635448c7381959bd7e02195104296ba88bda01 name: haskoin-node -version: 0.9.6 +version: 0.9.7 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/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 @@ -73,7 +73,9 @@ initChainDB net getBestBlockHeader >>= chainEvent . ChainBestBlock $(logInfoS) "Chain" "Initialization complete" - forever $ receive inbox >>= chainMessage + forever $ do + $(logDebugS) "Chain" "Awaiting message..." + receive inbox >>= chainMessage chainEvent :: MonadChain m => ChainEvent -> m () chainEvent e = do @@ -109,16 +111,19 @@ then do $(logDebugS) "Chain" - "Finished importing headers from peer" + "Finished importing headers from syncing peer" MSendHeaders `sendMessage` p finishPeer p syncNewPeer syncNotif - else syncPeer p + else do + $(logDebugS) + "Chain" + "Continuing importing headers from syncing peer" + syncPeer p syncNewPeer :: MonadChain m => m () -syncNewPeer = do - $(logDebugS) "Chain" "Attempting to sync against a new peer" +syncNewPeer = getSyncingPeer >>= \case Nothing -> do $(logDebugS) "Chain" "Getting next peer to sync from" @@ -135,13 +140,17 @@ syncPeer :: MonadChain m => Peer -> m () syncPeer p = do - $(logInfoS) "Chain" "Syncing against selected peer" - bb <- chainSyncingPeer >>= \case - Just ChainSync {chainSyncPeer = p', chainHighest = Just g} - | p == p' -> return g - _ -> getBestBlockHeader + $(logInfoS) "Chain" "Syncing hedears against selected peer" + bb <- + chainSyncingPeer >>= \case + Just ChainSync {chainSyncPeer = p', chainHighest = Just g} + | p == p' -> return g + _ -> getBestBlockHeader now <- round <$> liftIO getPOSIXTime gh <- syncHeaders now bb p + $(logDebugS) "Chain" $ + "Requesting headers from syncing peer for chain head at height: " <> + cs (show (nodeHeight bb)) MGetHeaders gh `sendMessage` p chainMessage :: MonadChain m => ChainMessage -> m () @@ -167,6 +176,7 @@ chainMessage (ChainIsSynced reply) = isSynced >>= atomically . reply chainMessage ChainPing = do + $(logDebugS) "Chain" "Housekeeping..." ChainConfig {chainConfTimeout = to} <- asks myReader now <- round <$> liftIO getPOSIXTime chainSyncingPeer >>= \case @@ -183,7 +193,10 @@ go = forever $ do threadDelay =<< - liftIO (randomRIO (250 * 1000, 1000 * 1000)) + liftIO (randomRIO (2 * 1000 * 1000, 20 * 1000 * 1000)) + $(logDebugS) + "ChainSyncLoop" + "Ping chain for housekeeping" ChainPing `send` ch -- | Version of the database. @@ -284,7 +297,7 @@ initChainDB net = do db <- asks myChainDB ver <- retrieve db def ChainDataVersionKey - when (ver /= Just dataVersion) purgeChainDB + when (ver /= Just dataVersion) $ purgeChainDB >>= writeBatch db R.insert db ChainDataVersionKey dataVersion retrieve db def BestBlockKey >>= \b -> when (isNothing (b :: Maybe BlockNode)) $ do @@ -293,7 +306,7 @@ -- | Purge database of elements having keys that may conflict with those used in -- this module. -purgeChainDB :: (MonadChainLogic a p m, MonadUnliftIO m) => m () +purgeChainDB :: (MonadChainLogic a p m, MonadUnliftIO m) => m [R.BatchOp] purgeChainDB = do db <- asks myChainDB runResourceT . R.withIterator db def $ \it -> do @@ -302,13 +315,12 @@ where recurse_delete it db = R.iterKey it >>= \case - Nothing -> return () Just k | B.head k == 0x90 || B.head k == 0x91 -> do R.delete db def k R.iterNext it - recurse_delete it db - | otherwise -> return () + (R.Del k :) <$> recurse_delete it db + _ -> return [] -- | Import a bunch of continuous headers. Returns 'True' if the number of -- headers is 2000, which means that there are possibly more headers to sync 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 @@ -21,7 +21,6 @@ ) where import Conduit -import Control.Applicative import Control.Monad import Control.Monad.Except import Control.Monad.Logger @@ -101,11 +100,17 @@ _ -> Nothing $(logDebugS) "Manager" "Initialization complete" withConnectLoop mgr $ - forever $ receive inbox >>= managerMessage + forever $ do + $(logDebugS) "Manager" "Awaiting message..." + m <- receive inbox + $(logDebugS) "Manager" "Processing message.." + managerMessage m f (a, mex) = ManagerPeerDied a mex `sendSTM` mgr putBestBlock :: MonadManager m => BlockHeight -> m () -putBestBlock bb = asks myBestBlock >>= \b -> atomically $ writeTVar b bb +putBestBlock bb = do + $(logDebugS) "Manager" $ "Best block at height " <> cs (show bb) + asks myBestBlock >>= \b -> atomically $ writeTVar b bb getBestBlock :: MonadManager m => m BlockHeight getBestBlock = asks myBestBlock >>= readTVarIO @@ -115,39 +120,52 @@ loadPeers :: (MonadUnliftIO m, MonadManager m) => m () loadPeers = do - d <- mgrConfDiscover <$> asks myConfig - loadStaticPeers - when d $ loadPeersFromDB >> loadNetSeeds + os <- readTVarIO =<< asks onlinePeers + ks <- readTVarIO =<< asks knownPeers + if null os && Set.null ks + then do + loadStaticPeers + d <- mgrConfDiscover <$> asks myConfig + when d $ loadPeersFromDB >> loadNetSeeds + else $(logDebugS) "Manager" "Peers already available, not initialising" loadStaticPeers :: (MonadUnliftIO m, MonadManager m) => m () loadStaticPeers = do + $(logDebugS) "Manager" "Loading static peers" xs <- mgrConfPeers <$> asks myConfig mapM_ newPeer =<< concat <$> mapM toSockAddr xs loadPeersFromDB :: (MonadUnliftIO m, MonadManager m) => m () loadPeersFromDB = do + $(logDebugS) "Manager" "Loading peers from database" db <- getManagerDB xs <- matchingAsList db R.defaultReadOptions PeerAddressBase let f (PeerAddress x, ()) = x f _ = undefined ys = map f xs + $(logInfoS) "Manager" $ + "Loaded " <> cs (show (length ys)) <> " peers from database" mapM_ newPeer ys storePeersInDB :: (MonadUnliftIO m, MonadManager m) => m () storePeersInDB = do db <- getManagerDB os <- readTVarIO =<< asks onlinePeers - unless (null os) $ do - ks <- readTVarIO =<< asks knownPeers - let ps = map onlinePeerAddress os <> Set.toList ks - xs = map ((`insertOp` ()) . PeerAddress) ps - purgePeerDB db - writeBatch db xs + ks <- readTVarIO =<< asks knownPeers + let ps = map onlinePeerAddress os <> Set.toList ks + xs = map ((`insertOp` ()) . PeerAddress) ps + unless (null xs) $ do + $(logInfoS) "Manager" $ + "Storing " <> cs (show (length xs)) <> " peers in database" + writeBatch db =<< (++ xs) <$> purgePeerDB db loadNetSeeds :: (MonadUnliftIO m, MonadManager m) => m () loadNetSeeds = do net <- getNetwork - mapM_ newPeer =<< concat <$> mapM toSockAddr (networkSeeds net) + $(logDebugS) "Manager" "Loading network seeds" + ss <- concat <$> mapM toSockAddr (networkSeeds net) + $(logDebugS) "Manager" $ "Adding " <> cs (show (length ss)) <> " seed peers" + mapM_ newPeer ss logConnectedPeers :: MonadManager m => m () logConnectedPeers = do @@ -170,7 +188,7 @@ db <- getManagerDB ops <- getOnlinePeers forM_ ops $ \OnlinePeer {onlinePeerMailbox = p} -> killPeer PurgingPeer p - purgePeerDB db + writeBatch db =<< purgePeerDB db forwardMessage :: MonadManager m => Peer -> Message -> m () forwardMessage p = managerEvent . PeerMessage p @@ -209,15 +227,20 @@ $(logErrorS) "Manager" $ "Received verack from unknown peer: " <> s killPeer UnknownPeer p -managerMessage (ManagerPeerMessage p (MAddr (Addr nas))) = - mgrConfDiscover <$> asks myConfig >>= \d -> - when d $ do - b <- asks onlinePeers - s <- atomically $ peerString b p - $(logInfoS) "Manager" $ - "Received " <> cs (show (length nas)) <> " peers from " <> s +managerMessage (ManagerPeerMessage p (MAddr (Addr nas))) = do + b <- asks onlinePeers + s <- atomically $ peerString b p + let n = length nas + $(logDebugS) "Manager" $ + "Received " <> cs (show n) <> " addresses from peer " <> s + mgrConfDiscover <$> asks myConfig >>= \case + True -> do let sas = map (naAddress . snd) nas forM_ sas newPeer + False -> + $(logDebugS) + "Manager" + "Ignoring received peers (discovery disabled)" managerMessage (ManagerPeerMessage p m@(MPong (Pong n))) = do now <- liftIO getCurrentTime @@ -253,10 +276,12 @@ managerMessage ManagerConnect = do l <- length <$> getConnectedPeers x <- mgrConfMaxPeers <$> asks myConfig - when (l < x) $ - getNewPeer >>= \case - Nothing -> return () - Just sa -> connectPeer sa + if l < x + then getNewPeer >>= \case + Nothing -> + $(logDebugS) "Manager" "No peers available to connect" + Just sa -> connectPeer sa + else $(logDebugS) "Manager" "Enough peers connected." managerMessage (ManagerPeerDied a e) = processPeerOffline a e @@ -278,6 +303,7 @@ ManagerConfig {mgrConfTimeout = to} <- asks myConfig b <- asks onlinePeers s <- atomically $ peerString b p + $(logDebugS) "Manager" $ "Checking on peer " <> s atomically (lastPing b p) >>= \case Nothing -> pingPeer p Just t -> do @@ -354,7 +380,7 @@ Nothing -> $(logErrorS) "Manager" "Will not announce unknown peer" getNewPeer :: (MonadUnliftIO m, MonadManager m) => m (Maybe SockAddr) -getNewPeer = runMaybeT $ go <|> (lift loadPeers >> go) +getNewPeer = runMaybeT $ lift loadPeers >> go where go = do b <- asks knownPeers @@ -364,32 +390,42 @@ a <- liftIO $ randomRIO (0, length xs - 1) let p = xs !! a atomically . modifyTVar b $ Set.delete p - return p + o <- asks onlinePeers + atomically (findPeerAddress o p) >>= \case + Nothing -> return p + Just _ -> go connectPeer :: (MonadUnliftIO m, MonadManager m) => SockAddr -> m () connectPeer sa = do - $(logInfoS) "Manager" $ "Connecting to peer: " <> cs (show sa) - ManagerConfig {mgrConfNetAddr = ad, mgrConfNetwork = net} <- asks myConfig - mgr <- asks myMailbox - sup <- asks mySupervisor - nonce <- liftIO randomIO - bb <- getBestBlock - now <- round <$> liftIO getPOSIXTime - let rmt = NetworkAddress (srv net) sa - ver = buildVersion net nonce bb ad rmt now - (inbox, p) <- newMailbox - let pc pub = - PeerConfig - { peerConfListen = pub - , peerConfNetwork = net - , peerConfAddress = sa - } - a <- withRunInIO $ \io -> sup `addChild` io (launch mgr pc inbox p) - MVersion ver `sendMessage` p - b <- asks onlinePeers - _ <- atomically $ newOnlinePeer b sa nonce p a - return () + os <- asks onlinePeers + atomically (findPeerAddress os sa) >>= \case + Just _ -> + $(logErrorS) "Manager" $ + "Attempted to connect to peer twice: " <> cs (show sa) + Nothing -> do + $(logInfoS) "Manager" $ "Connecting to " <> cs (show sa) + ManagerConfig {mgrConfNetAddr = ad, mgrConfNetwork = net} <- + asks myConfig + mgr <- asks myMailbox + sup <- asks mySupervisor + nonce <- liftIO randomIO + bb <- getBestBlock + now <- round <$> liftIO getPOSIXTime + let rmt = NetworkAddress (srv net) sa + ver = buildVersion net nonce bb ad rmt now + (inbox, p) <- newMailbox + let pc pub = + PeerConfig + { peerConfListen = pub + , peerConfNetwork = net + , peerConfAddress = sa + } + a <- withRunInIO $ \io -> sup `addChild` io (launch mgr pc inbox p) + MVersion ver `sendMessage` p + b <- asks onlinePeers + _ <- atomically $ newOnlinePeer b sa nonce p a + return () where l mgr p m = ManagerPeerMessage p m `sendSTM` mgr srv net @@ -398,23 +434,37 @@ launch mgr pc inbox p = withPublisher $ \pub -> bracket (subscribe pub (l mgr p)) (unsubscribe pub) $ \_ -> - withPeerLoop p mgr $ \a -> do + withPeerLoop sa p mgr $ \a -> do link a peer (pc pub) inbox -withPeerLoop :: MonadUnliftIO m => Peer -> Manager -> (Async a -> m a) -> m a -withPeerLoop p mgr = withAsync go +withPeerLoop :: + (MonadUnliftIO m, MonadLogger m) + => SockAddr + -> Peer + -> Manager + -> (Async a -> m a) + -> m a +withPeerLoop sa p mgr = withAsync go where - go = forever $ do - threadDelay =<< liftIO (randomRIO (30 * 1000 * 1000, 90 * 1000 * 1000)) - ManagerCheckPeer p `send` mgr + go = + forever $ do + threadDelay =<< + liftIO (randomRIO (30 * 1000 * 1000, 90 * 1000 * 1000)) + $(logDebugS) "ManagerPeerLoop" $ + "Ping manager about peer: " <> cs (show sa) + ManagerCheckPeer p `send` mgr -withConnectLoop :: MonadUnliftIO m => Manager -> m a -> m a +withConnectLoop :: (MonadLogger m, MonadUnliftIO m) => Manager -> m a -> m a withConnectLoop mgr act = withAsync go (\a -> link a >> act) where - go = forever $ do - ManagerConnect `send` mgr - threadDelay =<< liftIO (randomRIO (250 * 1000, 1000 * 1000)) + go = + forever $ do + $(logDebugS) + "ManagerConnectLoop" + "Ping manager for housekeeping" + ManagerConnect `send` mgr + threadDelay =<< liftIO (randomRIO (2 * 1000 * 1000, 10 * 1000 * 1000)) -- | Database version. versionPeerDB :: Word32 @@ -467,14 +517,13 @@ initPeerDB :: MonadUnliftIO m => DB -> Bool -> m () initPeerDB db discover = do ver <- retrieve db def PeerDataVersionKey - when (ver /= Just versionPeerDB || not discover) $ purgePeerDB db + when (ver /= Just versionPeerDB || not discover) $ + writeBatch db =<< purgePeerDB db R.insert db PeerDataVersionKey versionPeerDB -- | Purge peer records from database. -purgePeerDB :: MonadUnliftIO m => DB -> m () -purgePeerDB db = do - ops <- (++) <$> purge_byte 0x81 <*> purge_byte 0x83 - writeBatch db ops +purgePeerDB :: MonadUnliftIO m => DB -> m [R.BatchOp] +purgePeerDB db = (++) <$> purge_byte 0x81 <*> purge_byte 0x83 where purge_byte byte = U.runResourceT . R.withIterator db def $ \it -> do 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 @@ -45,8 +45,11 @@ peer pc inbox = withConnection a $ \ad -> runReaderT (peer_session ad) pc where a = peerConfAddress pc - go = forever $ receive inbox >>= dispatchMessage pc + go = forever $ do + $(logDebugS) s "Awaiting message..." + receive inbox >>= dispatchMessage pc net = peerConfNetwork pc + s = peerString (peerConfAddress pc) peer_session ad = let ins = appSource ad ons = appSink ad @@ -66,8 +69,11 @@ yield msg dispatchMessage cfg (GetPublisher reply) = atomically $ reply (peerConfListen cfg) -dispatchMessage _ (KillPeer e) = +dispatchMessage cfg (KillPeer e) = do + $(logErrorS) s $ "Killing peer via mailbox request: " <> cs (show e) throwIO e + where + s = peerString (peerConfAddress cfg) -- | Internal conduit to parse messages coming from peer. inPeerConduit :: @@ -104,4 +110,4 @@ -- | Peer string for logging peerString :: SockAddr -> Text -peerString a = "Peer{" <> cs (show a) <> "}" +peerString a = "Peer<" <> cs (show a) <> ">"