packages feed

haskoin-node 0.9.6 → 0.9.7

raw patch · 5 files changed

+160/−84 lines, 5 files

Files

CHANGELOG.md view
@@ -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.
haskoin-node.cabal view
@@ -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
src/Network/Haskoin/Node/Chain.hs view
@@ -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
src/Network/Haskoin/Node/Manager.hs view
@@ -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
src/Network/Haskoin/Node/Peer.hs view
@@ -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) <> ">"