haskoin-node (empty) → 0.2.0
raw patch · 12 files changed
+2600/−0 lines, 12 filesdep +HUnitdep +QuickCheckdep +aesonsetup-changed
Dependencies added: HUnit, QuickCheck, aeson, async, base, binary, bytestring, concurrent-extra, conduit, conduit-extra, containers, data-default, deepseq, either, exceptions, haskoin-core, haskoin-node, leveldb-haskell, lifted-async, lifted-base, monad-control, monad-logger, mtl, network, random, stm, stm-chans, stm-conduit, string-conversions, test-framework, test-framework-hunit, test-framework-quickcheck2, text, time
Files
- Network/Haskoin/Node/BlockChain.hs +695/−0
- Network/Haskoin/Node/Checkpoints.hs +28/−0
- Network/Haskoin/Node/HeaderTree.hs +585/−0
- Network/Haskoin/Node/Peer.hs +703/−0
- Network/Haskoin/Node/STM.hs +408/−0
- Setup.hs +2/−0
- UNLICENSE +24/−0
- haskoin-node.cabal +99/−0
- stack.yaml +14/−0
- tests/Main.hs +13/−0
- tests/Network/Haskoin/Node/Tests.hs +12/−0
- tests/Network/Haskoin/Node/Units.hs +17/−0
+ Network/Haskoin/Node/BlockChain.hs view
@@ -0,0 +1,695 @@+{-# LANGUAGE TemplateHaskell #-}+module Network.Haskoin.Node.BlockChain where++import System.Random (randomIO)++import Control.Monad (void, liftM, when, unless, forever, forM, forM_)+import Control.Monad.Trans (MonadIO, lift, liftIO)+import Control.Monad.Trans.Control (MonadBaseControl, liftBaseOp_)+import Control.Monad.Logger (MonadLogger, logInfo, logWarn, logDebug, logError)+import Control.Monad.Reader (ask, asks)+import Control.Concurrent (threadDelay)+import Control.Concurrent.STM.Lock (locked)+import qualified Control.Concurrent.STM.Lock as Lock (with)+import Control.Exception.Lifted (throw)+import Control.Concurrent.STM.TBMChan (readTBMChan, isEmptyTBMChan)+import Control.Concurrent.Async.Lifted+ ( withAsync, link, waitAny, mapConcurrently )+import Control.Concurrent.STM+ ( STM, atomically, readTVar, putTMVar, isEmptyTMVar+ , takeTMVar, tryTakeTMVar, retry, tryReadTMVar+ )++import Data.List (nub)+import Data.Text (pack)+import Data.Maybe (listToMaybe)+import Data.Time.Clock.POSIX (getPOSIXTime)+import Data.Unique (hashUnique)+import Data.Conduit (Source, yield)+import qualified Data.Sequence as S (length)+import qualified Data.ByteString.Char8 as C (unpack)+import Data.String.Conversions (cs)+import qualified Data.Map as M (keys, lookup, null, delete)+import Data.Word (Word32)++import Network.Haskoin.Node+import Network.Haskoin.Node.STM+import Network.Haskoin.Node.HeaderTree+import Network.Haskoin.Node.Peer+import Network.Haskoin.Transaction+import Network.Haskoin.Block++startSPVNode :: (MonadLogger m, MonadIO m, MonadBaseControl IO m)+ => [PeerHost]+ -> BloomFilter+ -> Int+ -> NodeT m ()+startSPVNode hosts bloom elems = do+ $(logDebug) "Setting our bloom filter in the node"+ atomicallyNodeT $ sendBloomFilter bloom elems+ $(logDebug) $ pack $ unwords+ [ "Starting SPV node with", show $ length hosts, "hosts" ]+ withAsync (void $ mapConcurrently startReconnectPeer hosts) $ \a1 -> do+ link a1+ $(logInfo) "Starting the initial header sync"+ headerSync+ $(logInfo) "Initial header sync complete"+ $(logDebug) "Starting the tickle processing thread"+ withAsync processTickles $ \a2 -> link a2 >> do+ _ <- liftIO $ waitAny [a1, a2]+ return ()++-- Source of all transaction broadcasts+txSource :: (MonadLogger m, MonadIO m, MonadBaseControl IO m)+ => Source (NodeT m) Tx+txSource = do+ chan <- lift $ asks sharedTxChan+ $(logDebug) "Waiting to receive a transaction..."+ resM <- liftIO $ atomically $ readTBMChan chan+ case resM of+ Just (pid, ph, tx) -> do+ $(logInfo) $ formatPid pid ph $ unwords+ [ "Received transaction broadcast", cs $ txHashToHex $ txHash tx ]+ yield tx >> txSource+ _ -> $(logError) "Tx channel closed unexpectedly"++handleGetData :: (MonadLogger m, MonadIO m, MonadBaseControl IO m)+ => (TxHash -> m (Maybe Tx))+ -> NodeT m ()+handleGetData handler = forever $ do+ $(logDebug) "Waiting for GetData transaction requests..."+ -- Wait for tx GetData requests to be available+ txids <- atomicallyNodeT $ do+ datMap <- readTVarS sharedTxGetData+ if M.null datMap then lift retry else return $ M.keys datMap+ forM (nub txids) $ \tid -> lift (handler tid) >>= \txM -> do+ $(logDebug) $ pack $ unwords+ [ "Processing GetData txid request", cs $ txHashToHex tid ]+ pidsM <- atomicallyNodeT $ do+ datMap <- readTVarS sharedTxGetData+ writeTVarS sharedTxGetData $ M.delete tid datMap+ return $ M.lookup tid datMap+ case (txM, pidsM) of+ -- Send the transaction to the required peers+ (Just tx, Just pids) -> forM_ pids $ \(pid, ph) -> do+ $(logDebug) $ formatPid pid ph $ unwords+ [ "Sending tx", cs $ txHashToHex tid, "to peer" ]+ atomicallyNodeT $ trySendMessage pid $ MTx tx+ _ -> return ()++broadcastTxs :: (MonadLogger m, MonadIO m, MonadBaseControl IO m)+ => [TxHash]+ -> NodeT m ()+broadcastTxs txids = do+ forM_ txids $ \tid -> $(logInfo) $ pack $ unwords+ [ "Transaction INV broadcast:", cs $ txHashToHex tid ]+ -- Broadcast an INV message for new transactions+ let msg = MInv $ Inv $ map (InvVector InvTx . getTxHash) txids+ atomicallyNodeT $ sendMessageAll msg++rescanTs :: Timestamp -> NodeT STM ()+rescanTs ts = do+ rescanTMVar <- asks sharedRescan+ lift $ do+ -- Make sure the TMVar is empty+ _ <- tryTakeTMVar rescanTMVar+ putTMVar rescanTMVar $ Left ts++rescanHeight :: BlockHeight -> NodeT STM ()+rescanHeight h = do+ rescanTMVar <- asks sharedRescan+ lift $ do+ -- Make sure the TMVar is empty+ _ <- tryTakeTMVar rescanTMVar+ putTMVar rescanTMVar $ Right h++-- Wait for the next merkle batch to be available. This function will check for+-- rescans.+merkleDownload+ :: (MonadLogger m, MonadIO m, MonadBaseControl IO m)+ => BlockHash+ -> Int+ -> NodeT m ( BlockChainAction+ , Source (NodeT m) (Either (MerkleBlock, MerkleTxs) Tx)+ )+merkleDownload bh batchSize = do+ -- Store the best block received from the wallet for information only.+ -- This will be displayed in `hw status`+ atomicallyNodeT $ writeTVarS sharedBestBlock bh+ merkleCheckSync+ rescanTMVar <- asks sharedRescan+ -- Wait either for a new block to arrive or a rescan to be triggered+ $(logDebug) "Waiting for a new block or a rescan..."+ resE <- atomicallyNodeT $ orElseNodeT+ (liftM Left $ lift $ takeTMVar rescanTMVar)+ (liftM (const $ Right ()) $ waitNewBlock bh)+ resM <- case resE of+ -- A rescan was triggered+ Left valE -> do+ $(logInfo) $ pack $ unwords+ [ "Got rescan request", show valE ]+ -- Wait until rescan conditions are met+ newValE <- waitRescan rescanTMVar valE+ $(logDebug) $ pack $ unwords+ [ "Rescan condition reached:", show newValE ]+ case newValE of+ Left ts -> tryMerkleDwnTimestamp ts batchSize+ Right h -> tryMerkleDwnHeight h batchSize+ -- Continue download from a hash+ Right _ -> tryMerkleDwnBlock bh batchSize+ case resM of+ Just res -> return res+ _ -> do+ $(logWarn) "Invalid merkleDownload result. Retrying ..."+ -- Sleep 10 seconds and retry+ liftIO $ threadDelay $ 10*1000000+ merkleDownload bh batchSize+ where+ waitRescan rescanTMVar valE = do+ resE <- atomicallyNodeT $ orElseNodeT+ (liftM Left (lift $ takeTMVar rescanTMVar))+ (waitVal valE >> return (Right valE))+ case resE of+ Left newValE -> waitRescan rescanTMVar newValE+ Right res -> return res+ waitVal valE = case valE of+ Left ts -> waitFastCatchup ts+ Right h -> waitHeight h++merkleCheckSync :: (MonadLogger m, MonadIO m, MonadBaseControl IO m)+ => NodeT m ()+merkleCheckSync = do+ -- Check if we are synced+ (synced, mempool, ourHeight) <- atomicallyNodeT $ do+ synced <- areBlocksSynced+ when synced $ writeTVarS sharedMerklePeer Nothing+ ourHeight <- liftM nodeHeaderHeight $ readTVarS sharedBestHeader+ mempool <- readTVarS sharedMempool+ return (synced, mempool, ourHeight)+ when synced $ do+ $(logInfo) $ pack $ unwords+ [ "Merkle blocks are in sync with the"+ , "network at height", show ourHeight+ ]+ -- Do a mempool sync on the first merkle sync+ unless mempool $ do+ atomicallyNodeT $ do+ sendMessageAll MMempool+ writeTVarS sharedMempool True+ $(logInfo) "Requesting a mempool sync"++-- Wait for headers to catch up to the given height+waitHeight :: BlockHeight -> NodeT STM ()+waitHeight height = do+ node <- readTVarS sharedBestHeader+ -- Check if we passed the timestamp condition+ unless (height < nodeHeaderHeight node) $ lift retry++-- Wait for headers to catch up to the given timestamp+waitFastCatchup :: Timestamp -> NodeT STM ()+waitFastCatchup ts = do+ node <- readTVarS sharedBestHeader+ -- Check if we passed the timestamp condition+ unless (ts < blockTimestamp (nodeHeader node)) $ lift retry++-- Wait for a new block to be available for download+waitNewBlock :: BlockHash -> NodeT STM ()+waitNewBlock bh = do+ node <- readTVarS sharedBestHeader+ -- We have more merkle blocks to download+ unless (bh /= nodeBlockHash node) $ lift retry++tryMerkleDwnHeight+ :: (MonadLogger m, MonadIO m, MonadBaseControl IO m)+ => BlockHeight+ -> Int+ -> NodeT m ( Maybe ( BlockChainAction+ , Source (NodeT m) (Either (MerkleBlock, MerkleTxs) Tx)+ )+ )+tryMerkleDwnHeight height batchSize = do+ $(logInfo) $ pack $ unwords+ [ "Requesting merkle blocks at height", show height+ , "with batch size", show batchSize+ ]+ -- Request height - 1 as we want to start downloading at height+ nodeM <- runHeaderTree $ getBlockHeaderByHeight $ height - 1+ case nodeM of+ Just BlockHeaderNode{..} -> tryMerkleDwnBlock nodeBlockHash batchSize+ _ -> do+ $(logDebug) $ pack $ unwords+ [ "Can't download merkle blocks."+ , "Waiting for headers to sync ..."+ ]+ return Nothing++tryMerkleDwnTimestamp+ :: (MonadLogger m, MonadIO m, MonadBaseControl IO m)+ => Timestamp+ -> Int+ -> NodeT m ( Maybe ( BlockChainAction+ , Source (NodeT m) (Either (MerkleBlock, MerkleTxs) Tx)+ )+ )+tryMerkleDwnTimestamp ts batchSize = do+ $(logInfo) $ pack $ unwords+ [ "Requesting merkle blocks after timestamp", show ts+ , "with batch size", show batchSize+ ]+ nodeM <- runHeaderTree $ getNodeAtTimestamp ts+ case nodeM of+ Just BlockHeaderNode{..} -> tryMerkleDwnBlock nodeBlockHash batchSize+ _ -> do+ $(logDebug) $ pack $ unwords+ [ "Can't download merkle blocks."+ , "Waiting for headers to sync ..."+ ]+ return Nothing++tryMerkleDwnBlock+ :: (MonadLogger m, MonadIO m, MonadBaseControl IO m)+ => BlockHash+ -> Int+ -> NodeT m ( Maybe ( BlockChainAction+ , Source (NodeT m) (Either (MerkleBlock, MerkleTxs) Tx)+ )+ )+tryMerkleDwnBlock bh batchSize = do+ $(logDebug) $ pack $ unwords+ [ "Requesting merkle download from block"+ , cs $ blockHashToHex bh, "and batch size", show batchSize+ ]+ --Get the list of merkle blocks to download from our headers+ actionM <- runHeaderTree $ getNodeWindow bh batchSize+ case actionM of+ -- Nothing to download+ Nothing -> do+ $(logDebug) $ pack $ unwords+ [ "No more merkle blocks available from block"+ , cs $ blockHashToHex bh+ ]+ return Nothing+ -- A batch of merkle blocks is available for download+ Just action -> case actionNodes action of+ [] -> do+ $(logError) "BlockChainAction was empty"+ return Nothing+ ns -> do+ -- Wait for a peer available for merkle download+ (pid, PeerSession{..}) <- waitMerklePeer $+ nodeHeaderHeight $ last ns++ $(logDebug) $ formatPid pid peerSessionHost $ unwords+ [ "Found merkle downloading peer with score"+ , show peerSessionScore+ ]++ let source = peerMerkleDownload pid peerSessionHost action+ return $ Just (action, source)+ where+ waitMerklePeer height = atomicallyNodeT $ do+ pidM <- readTVarS sharedHeaderPeer+ allPeers <- getPeersAtHeight (>= height)+ let f (pid,_) = Just pid /= pidM+ -- Filter out the peer syncing headers (if there is one)+ peers = filter f allPeers+ case listToMaybe peers of+ Just res@(pid,_) -> do+ writeTVarS sharedMerklePeer $ Just pid+ return res+ _ -> lift retry++peerMerkleDownload+ :: (MonadLogger m, MonadIO m, MonadBaseControl IO m)+ => PeerId+ -> PeerHost+ -> BlockChainAction+ -> Source (NodeT m) (Either (MerkleBlock, MerkleTxs) Tx)+peerMerkleDownload pid ph action = do+ let bids = map nodeBlockHash $ actionNodes action+ vs = map (InvVector InvMerkleBlock . getBlockHash) bids+ $(logInfo) $ formatPid pid ph $ unwords+ [ "Requesting", show $ length bids, "merkle block(s)" ]+ nonce <- liftIO randomIO+ -- Request a merkle batch download+ sessM <- lift . atomicallyNodeT $ do+ _ <- trySendMessage pid $ MGetData $ GetData vs+ -- Send a ping to have a recognizable end message for+ -- the last merkle block download.+ _ <- trySendMessage pid $ MPing $ Ping nonce+ tryGetPeerSession pid+ case sessM of+ Just PeerSession{..} -> checkOrder peerSessionMerkleChan bids+ _ -> lift . atomicallyNodeT $+ writeTVarS sharedMerklePeer Nothing+ where+ -- Build a source that that will check the order of the received merkle+ -- blocks against the initial request. If merkle blocks are sent out of+ -- order, the source will close and the peer will be flagged as+ -- misbehaving. The source will also close once all the requested merkle+ -- blocks have been received from the peer.+ checkOrder _ [] = lift . atomicallyNodeT $+ writeTVarS sharedMerklePeer Nothing+ checkOrder chan (bid:bids) = do+ -- Read the channel or disconnect the peer after waiting for 2 minutes+ resM <- lift $ raceTimeout 120+ (disconnectPeer pid ph)+ (liftIO . atomically $ readTBMChan chan)+ case resM of+ -- Forward transactions+ Right (Just res@(Right _)) ->+ yield res >> checkOrder chan (bid:bids)+ Right (Just res@(Left (MerkleBlock mHead _ _ _, _))) -> do+ let mBid = headerHash mHead+ $(logDebug) $ formatPid pid ph $ unwords+ [ "Processing merkle block", cs $ blockHashToHex mBid ]+ -- Check if we were expecting this merkle block+ if mBid == bid+ then yield res >> checkOrder chan bids+ else lift $ do+ atomicallyNodeT $ writeTVarS sharedMerklePeer Nothing+ -- If we were not expecting this merkle block, do not+ -- yield the merkle block and close the source+ misbehaving pid ph moderateDoS $ unwords+ [ "Peer sent us merkle block hash"+ , cs $ blockHashToHex $ headerHash mHead+ , "but we expected merkle block hash"+ , cs $ blockHashToHex bid+ ]+ -- Not sure how to recover from this situation.+ -- Disconnect the peer. TODO: Is there a way to recover+ -- without buffering the whole batch in memory and+ -- re-order it?+ disconnectPeer pid ph+ -- The channel closed. Stop here.+ _ -> do+ $(logWarn) $ formatPid pid ph+ "Merkle channel closed unexpectedly"+ lift $ atomicallyNodeT $ writeTVarS sharedMerklePeer Nothing++processTickles :: (MonadLogger m, MonadIO m, MonadBaseControl IO m)+ => NodeT m ()+processTickles = forever $ do+ $(logDebug) $ pack "Waiting for a block tickle ..."+ (pid, ph, tickle) <- atomicallyNodeT waitTickle+ $(logInfo) $ formatPid pid ph $ unwords+ [ "Received block tickle", cs $ blockHashToHex tickle ]+ heightM <- runHeaderTree $ getBlockHeaderHeight tickle+ case heightM of+ Just height -> do+ $(logInfo) $ formatPid pid ph $ unwords+ [ "The block tickle", cs $ blockHashToHex tickle+ , "is already connected"+ ]+ updatePeerHeight pid ph height+ _ -> do+ $(logDebug) $ formatPid pid ph $ unwords+ [ "The tickle", cs $ blockHashToHex tickle+ , "is unknown. Requesting a peer header sync."+ ]+ peerHeaderSyncFull pid ph `catchAny` const (disconnectPeer pid ph)+ newHeightM <- runHeaderTree $ getBlockHeaderHeight tickle+ case newHeightM of+ Just height -> do+ $(logInfo) $ formatPid pid ph $ unwords+ [ "The block tickle", cs $ blockHashToHex tickle+ , "was connected successfully"+ ]+ updatePeerHeight pid ph height+ _ -> $(logWarn) $ formatPid pid ph $ unwords+ [ "Could not find the height of block tickle"+ , cs $ blockHashToHex tickle+ ]+ where+ updatePeerHeight pid ph height = do+ $(logInfo) $ formatPid pid ph $ unwords+ [ "Updating peer height to", show height ]+ atomicallyNodeT $ do+ modifyPeerSession pid $ \s ->+ s{ peerSessionHeight = height }+ updateNetworkHeight++waitTickle :: NodeT STM (PeerId, PeerHost, BlockHash)+waitTickle = do+ tickleChan <- asks sharedTickleChan+ resM <- lift $ readTBMChan tickleChan+ case resM of+ Just res -> return res+ _ -> throw $ NodeException "tickle channel closed unexpectedly"++syncedHeight :: MonadIO m => NodeT m (Bool, Word32)+syncedHeight = atomicallyNodeT $ do+ synced <- areHeadersSynced+ ourHeight <- liftM nodeHeaderHeight $ readTVarS sharedBestHeader+ return (synced, ourHeight)++headerSync :: (MonadLogger m, MonadIO m, MonadBaseControl IO m)+ => NodeT m ()+headerSync = do+ -- Start the header sync+ $(logDebug) "Syncing more headers. Finding the best peer..."+ (pid, PeerSession{..}) <- atomicallyNodeT $ do+ peers <- getPeersAtNetHeight+ case listToMaybe peers of+ Just res@(pid,_) -> do+ -- Save the header syncing peer+ writeTVarS sharedHeaderPeer $ Just pid+ return res+ _ -> lift retry++ $(logDebug) $ formatPid pid peerSessionHost $ unwords+ [ "Found best header syncing peer with score"+ , show peerSessionScore+ ]++ -- Run a maximum of 10 header downloads with this peer.+ -- Then we re-evaluate the best peer+ continue <- catchAny (peerHeaderSyncLimit pid peerSessionHost 10) $+ \e -> do+ $(logError) $ pack $ show e+ disconnectPeer pid peerSessionHost >> return True++ -- Reset the header syncing peer+ atomicallyNodeT $ writeTVarS sharedHeaderPeer Nothing++ -- Check if we should continue the header sync+ if continue then headerSync else do+ (synced, ourHeight) <- syncedHeight+ if synced+ then do+ -- Check if merkles are synced+ merkleCheckSync+ $(logInfo) $ formatPid pid peerSessionHost $ unwords+ [ "Block headers are in sync with the"+ , "network at height", show ourHeight+ ]+ -- Continue the download if we are not yet synced+ else headerSync++peerHeaderSyncLimit :: (MonadLogger m, MonadIO m, MonadBaseControl IO m)+ => PeerId+ -> PeerHost+ -> Int+ -> NodeT m Bool+peerHeaderSyncLimit pid ph initLimit+ | initLimit < 1 = error "Limit must be at least 1"+ | otherwise = go initLimit Nothing+ where+ go limit prevM = peerHeaderSync pid ph prevM >>= \actionM -> case actionM of+ Just action ->+ -- If we received a side chain or a known chain, we want to+ -- continue szncing from this peer even if the limit has been+ -- reached.+ if limit > 1 || isSideChain action || isKnownChain action+ then go (limit - 1) actionM+ -- We got a Just, so we can continue the download from+ -- this peer+ else return True+ _ -> return False++-- Sync all the headers from a given peer+peerHeaderSyncFull :: (MonadLogger m, MonadIO m, MonadBaseControl IO m)+ => PeerId+ -> PeerHost+ -> NodeT m ()+peerHeaderSyncFull pid ph =+ go Nothing+ where+ go prevM = peerHeaderSync pid ph prevM >>= \actionM -> case actionM of+ Just _ -> go actionM+ Nothing -> do+ (synced, ourHeight) <- syncedHeight+ when synced $ $(logInfo) $ formatPid pid ph $ unwords+ [ "Block headers are in sync with the"+ , "network at height", show ourHeight+ ]++areBlocksSynced :: NodeT STM Bool+areBlocksSynced = do+ headersSynced <- areHeadersSynced+ bestBlock <- readTVarS sharedBestBlock+ bestHeader <- readTVarS sharedBestHeader+ return $ headersSynced && bestBlock == nodeBlockHash bestHeader++-- Check if the block headers are synced with the network height+areHeadersSynced :: NodeT STM Bool+areHeadersSynced = do+ ourHeight <- liftM nodeHeaderHeight $ readTVarS sharedBestHeader+ netHeight <- readTVarS sharedNetworkHeight+ -- If netHeight == 0 then we did not connect to any peers yet+ return $ ourHeight >= netHeight && netHeight > 0++-- | Sync one batch of headers from the given peer. Accept the result of a+-- previous peerHeaderSync to correctly compute block locators in the+-- presence of side chains.+peerHeaderSync :: (MonadLogger m, MonadIO m, MonadBaseControl IO m)+ => PeerId+ -> PeerHost+ -> Maybe BlockChainAction+ -> NodeT m (Maybe BlockChainAction)+peerHeaderSync pid ph prevM = do+ $(logDebug) $ formatPid pid ph "Waiting for the HeaderSync lock"+ -- Aquire the header syncing lock+ lock <- asks sharedSyncLock+ liftBaseOp_ (Lock.with lock) $ do++ -- Retrieve the block locator+ loc <- case prevM of+ Just (KnownChain ns) -> do+ $(logDebug) $ formatPid pid ph "Building a KnownChain locator"+ runHeaderTree $ blockLocatorHeight $ nodeHeaderHeight $ last ns+ Just action@(SideChain _) -> do+ $(logDebug) $ formatPid pid ph "Building a SideChain locator"+ runHeaderTree $ blockLocatorSide action+ _ -> do+ $(logDebug) $ formatPid pid ph "Building a regular locator"+ runHeaderTree blockLocator++ $(logDebug) $ formatPid pid ph $ unwords+ [ "Requesting headers with block locator of size"+ , show $ length loc+ , "Start block:", cs $ blockHashToHex $ head loc+ , "End block:", cs $ blockHashToHex $ last loc+ ]++ -- Send a GetHeaders message to the peer+ atomicallyNodeT $ sendMessage pid $ MGetHeaders $ GetHeaders 0x01 loc z++ $(logDebug) $ formatPid pid ph "Waiting 2 minutes for headers..."++ -- Wait 120 seconds for a response or time out+ continueE <- raceTimeout 120+ (disconnectPeer pid ph)+ waitHeaders++ -- Return True if we can continue syncing from this peer+ return $ either (const Nothing) id continueE+ where+ z = "0000000000000000000000000000000000000000000000000000000000000000"+ -- Wait for the headers to be available+ waitHeaders = do+ (rPid, headers) <- atomicallyNodeT $ takeTMVarS sharedHeaders+ if rPid == pid+ then processHeaders headers+ else waitHeaders+ processHeaders (Headers []) = do+ $(logDebug) $ formatPid pid ph+ "Received empty headers. Finished downloading headers."+ -- Do not continue the header download+ return Nothing+ processHeaders (Headers hs) = do+ $(logDebug) $ formatPid pid ph $ unwords+ [ "Received", show $ length hs, "headers."+ , "Start blocks:", cs $ blockHashToHex $ headerHash $ fst $ head hs+ , "End blocks:", cs $ blockHashToHex $ headerHash $ fst $ last hs+ ]+ now <- liftM round $ liftIO getPOSIXTime+ actionE <- runHeaderTree $ connectHeaders (map fst hs) now True+ case actionE of+ Left err -> do+ misbehaving pid ph severeDoS err+ return Nothing+ Right action -> case actionNodes action of+ [] -> do+ $(logWarn) $ formatPid pid ph $ unwords+ [ "Received an empty blockchain action:", show action ]+ return Nothing+ nodes -> do+ $(logDebug) $ formatPid pid ph $ unwords+ [ "Received", show $ length nodes+ , "nodes in the action"+ ]+ let height = nodeHeaderHeight $ last nodes+ case action of+ KnownChain _ ->+ $(logInfo) $ formatPid pid ph $ unwords+ [ "KnownChain headers received"+ , "up to height", show height+ ]+ SideChain _ ->+ $(logInfo) $ formatPid pid ph $ unwords+ [ "SideChain headers connected successfully"+ , "up to height", show height+ ]+ -- Headers extend our current best head+ _ -> do+ $(logInfo) $ formatPid pid ph $ unwords+ [ "Best headers connected successfully"+ , "up to height", show height+ ]+ atomicallyNodeT $+ writeTVarS sharedBestHeader $ last nodes+ -- If we received less than 2000 headers, we are done+ -- syncing from this peer and we return Nothing.+ return $ if length hs < 2000+ then Nothing+ else Just action++nodeStatus :: NodeT STM NodeStatus+nodeStatus = do+ nodeStatusPeers <- mapM peerStatus =<< getPeers+ SharedNodeState{..} <- ask+ lift $ do+ nodeStatusNetworkHeight <- readTVar sharedNetworkHeight+ nodeStatusBestHeader <- liftM nodeBlockHash $+ readTVar sharedBestHeader+ nodeStatusBestHeaderHeight <- liftM nodeHeaderHeight $+ readTVar sharedBestHeader+ nodeStatusBestBlock <- readTVar sharedBestBlock+ nodeStatusBloomSize <- liftM (maybe 0 (S.length . bloomData . fst)) $+ readTVar sharedBloomFilter+ nodeStatusHeaderPeer <- liftM (fmap hashUnique) $+ readTVar sharedHeaderPeer+ nodeStatusMerklePeer <- liftM (fmap hashUnique) $+ readTVar sharedMerklePeer+ nodeStatusHaveHeaders <- liftM not $ isEmptyTMVar sharedHeaders+ nodeStatusHaveTickles <- liftM not $ isEmptyTBMChan sharedTickleChan+ nodeStatusHaveTxs <- liftM not $ isEmptyTBMChan sharedTxChan+ nodeStatusGetData <- liftM M.keys $ readTVar sharedTxGetData+ nodeStatusRescan <- tryReadTMVar sharedRescan+ nodeStatusMempool <- readTVar sharedMempool+ nodeStatusSyncLock <- locked sharedSyncLock+ nodeStatusLevelDBLock <- locked sharedLevelDBLock+ return NodeStatus{..}++peerStatus :: (PeerId, PeerSession) -> NodeT STM PeerStatus+peerStatus (pid, PeerSession{..}) = do+ hostM <- getHostSession peerSessionHost+ let peerStatusPeerId = hashUnique pid+ peerStatusHost = peerSessionHost+ peerStatusConnected = peerSessionConnected+ peerStatusHeight = peerSessionHeight+ peerStatusProtocol = version <$> peerSessionVersion+ peerStatusUserAgent =+ C.unpack . getVarString . userAgent <$> peerSessionVersion+ peerStatusPing = show <$> peerSessionScore+ peerStatusThreadId = show peerSessionThreadId+ peerStatusDoSScore = peerHostSessionScore <$> hostM+ peerStatusLog = peerHostSessionLog <$> hostM+ peerStatusReconnectTimer = peerHostSessionReconnect <$> hostM+ lift $ do+ peerStatusHaveMerkles <- liftM not $ isEmptyTBMChan peerSessionMerkleChan+ peerStatusHaveMessage <- liftM not $ isEmptyTBMChan peerSessionChan+ peerStatusPingNonces <- readTVar peerSessionPings+ return PeerStatus{..}+
+ Network/Haskoin/Node/Checkpoints.hs view
@@ -0,0 +1,28 @@+module Network.Haskoin.Node.Checkpoints+( checkpointMap+, checkpointList+, verifyCheckpoint+) where++import qualified Data.IntMap.Strict as M (IntMap, fromList, lookup)++import Network.Haskoin.Block+import Network.Haskoin.Constants++-- | Checkpoints from bitcoind reference implementation /src/checkpoints.cpp+-- presented as an IntMap.+checkpointMap :: M.IntMap BlockHash+checkpointMap = M.fromList checkpointList++-- | Checkpoints from bitcoind reference implementation /src/checkpoints.cpp+-- presented as a list.+checkpointList :: [(Int, BlockHash)]+checkpointList = checkpoints++-- | Verify that a block hash at a given height either matches an existing+-- checkpoint or is not a checkpoint.+verifyCheckpoint :: Int -> BlockHash -> Bool+verifyCheckpoint height hash = case M.lookup height checkpointMap of+ Just value -> hash == value+ Nothing -> True+
+ Network/Haskoin/Node/HeaderTree.hs view
@@ -0,0 +1,585 @@+module Network.Haskoin.Node.HeaderTree+( HeaderTree(..)+, BlockHeaderNode(..)+, BlockChainAction(..)+, BlockHeight+, Timestamp+, initHeaderTree+, connectHeader+, connectHeaders+, commitAction+, isBestChain+, isChainReorg+, isSideChain+, isKnownChain+, blockLocator+, blockLocatorHeight+, blockLocatorSide+, blockLocatorPartial+, getNodeWindow+, bestBlockHeaderHeight+, getBlockHeaderHeight+, getNodeAtTimestamp+, genesisNode+, getParentNode+) where++import Control.Monad (foldM, when, unless, liftM, (<=<), forM, forM_)+import Control.Monad.Trans (lift, liftIO, MonadIO)+import Control.Monad.Trans.Either (EitherT, left, runEitherT)+import Control.Monad.Reader (MonadReader(..), ReaderT, ask)+import Control.DeepSeq (NFData(..))++import Data.Word (Word32)+import Data.Bits (shiftL)+import Data.Maybe (fromMaybe, isNothing, isJust, catMaybes)+import Data.List (sort)+import Data.Binary.Get (getWord32le)+import Data.Binary.Put (putWord32le)+import Data.Default (def)+import qualified Data.Binary as B (Binary, get, put)+import qualified Data.ByteString as BS (ByteString, reverse, append)++import qualified Database.LevelDB.Base as L (DB, get, put)++import Network.Haskoin.Block+import Network.Haskoin.Constants+import Network.Haskoin.Util+import Network.Haskoin.Node.Checkpoints++type BlockHeight = Word32+type Timestamp = Word32++class Monad m => HeaderTree m where+ getBlockHeaderNode :: BlockHash -> m (Maybe BlockHeaderNode)+ putBlockHeaderNode :: BlockHeaderNode -> m ()+ -- The height is only updated when the node is part of the main chain.+ -- Side chains are not indexed by their height.+ putBlockHeaderHeight :: BlockHeaderNode -> m ()+ getBlockHeaderByHeight :: BlockHeight -> m (Maybe BlockHeaderNode)+ getBestBlockHeader :: m BlockHeaderNode+ setBestBlockHeader :: BlockHeaderNode -> m ()++-- | Data type representing a BlockHeader node in the header chain. It+-- contains additional data such as the chain work and chain height for this+-- node.+data BlockHeaderNode = BlockHeaderNode+ { nodeBlockHash :: !BlockHash+ , nodeHeader :: !BlockHeader+ , nodeHeaderHeight :: !BlockHeight+ , nodeChainWork :: !Integer+ , nodeChild :: !(Maybe BlockHash)+ , nodeMedianTimes :: ![Timestamp]+ , nodeMinWork :: !Word32 -- Only used for testnet+ } deriving (Show, Read, Eq)++instance NFData BlockHeaderNode where+ rnf BlockHeaderNode{..} =+ rnf nodeBlockHash `seq`+ rnf nodeHeader `seq`+ rnf nodeHeaderHeight `seq`+ rnf nodeChainWork `seq`+ rnf nodeChild `seq`+ rnf nodeMedianTimes `seq`+ rnf nodeMinWork++instance B.Binary BlockHeaderNode where++ get = do+ nodeBlockHash <- B.get+ nodeHeader <- B.get+ nodeHeaderHeight <- getWord32le+ nodeChainWork <- B.get+ nodeChild <- B.get+ nodeMedianTimes <- B.get+ nodeMinWork <- B.get+ return BlockHeaderNode{..}++ put BlockHeaderNode{..} = do+ B.put nodeBlockHash+ B.put nodeHeader+ putWord32le nodeHeaderHeight+ B.put nodeChainWork+ B.put nodeChild+ B.put nodeMedianTimes+ B.put nodeMinWork++data BlockChainAction+ = BestChain { actionNodes :: ![BlockHeaderNode] }+ | ChainReorg { actionSplitNode :: !BlockHeaderNode+ , actionOldNodes :: ![BlockHeaderNode]+ , actionNodes :: ![BlockHeaderNode]+ }+ | SideChain { actionNodes :: ![BlockHeaderNode] }+ | KnownChain { actionNodes :: ![BlockHeaderNode] }+ deriving (Read, Show, Eq)++instance NFData BlockChainAction where+ rnf bca = case bca of+ BestChain ns -> rnf ns+ ChainReorg s os ns -> rnf s `seq` rnf os `seq` rnf ns+ KnownChain ns -> rnf ns+ SideChain ns -> rnf ns++-- | Returns True if the action is a best chain+isBestChain :: BlockChainAction -> Bool+isBestChain (BestChain _) = True+isBestChain _ = False++-- | Returns True if the action is a chain reorg+isChainReorg :: BlockChainAction -> Bool+isChainReorg ChainReorg{} = True+isChainReorg _ = False++-- | Returns True if the action is a side chain+isSideChain :: BlockChainAction -> Bool+isSideChain (SideChain _) = True+isSideChain _ = False++-- | Returns True if the action is a known chain+isKnownChain :: BlockChainAction -> Bool+isKnownChain (KnownChain _) = True+isKnownChain _ = False++-- | Number of blocks on average between difficulty cycles (2016 blocks)+diffInterval :: Word32+diffInterval = targetTimespan `div` targetSpacing++-- | Genesis BlockHeaderNode+genesisNode :: BlockHeaderNode+genesisNode = BlockHeaderNode+ { nodeBlockHash = headerHash genesisHeader+ , nodeHeader = genesisHeader+ , nodeHeaderHeight = 0+ , nodeChainWork = headerWork genesisHeader+ , nodeChild = Nothing+ , nodeMedianTimes = [blockTimestamp genesisHeader]+ , nodeMinWork = blockBits genesisHeader+ }++-- | Initialize the block header chain by inserting the genesis block if+-- it doesn't already exist.+initHeaderTree :: HeaderTree m => m ()+initHeaderTree = do+ genM <- getBlockHeaderNode $ nodeBlockHash genesisNode+ when (isNothing genM) $ do+ putBlockHeaderNode genesisNode+ putBlockHeaderHeight genesisNode+ setBestBlockHeader genesisNode++-- A more efficient way of connecting a list of BlockHeaders than connecting+-- them individually. The work check will only be done once for the whole+-- chain. The list of BlockHeaders have to form a valid chain, linked by their+-- parents.+connectHeaders :: HeaderTree m+ => [BlockHeader]+ -> Timestamp+ -> Bool+ -> m (Either String BlockChainAction)+connectHeaders bhs adjustedTime commit+ | null bhs = return $ Left "Invalid empty BlockHeaders in connectHeaders"+ | validChain bhs = runEitherT $ do+ newNodes <- forM bhs $ \bh -> do+ parNode <- verifyBlockHeader bh adjustedTime+ lift $ storeBlockHeader bh parNode+ -- Best header will only be updated if we have no errors+ lift $ evalNewChain commit newNodes+ | otherwise = return $ Left "BlockHeaders do not form a valid chain."+ where+ validChain (a:b:xs) = prevBlock b == headerHash a && validChain (b:xs)+ validChain [_] = True+ validChain _ = False++-- | Connect a block header to this block header chain. Corresponds to bitcoind+-- function ProcessBlockHeader and AcceptBlockHeader in main.cpp.+connectHeader :: HeaderTree m+ => BlockHeader+ -> Timestamp+ -> Bool+ -> m (Either String BlockChainAction)+connectHeader bh adjustedTime commit = runEitherT $ do+ parNode <- verifyBlockHeader bh adjustedTime+ lift $ evalNewChain commit . (: []) =<< storeBlockHeader bh parNode++evalNewChain :: HeaderTree m+ => Bool+ -> [BlockHeaderNode]+ -> m BlockChainAction+evalNewChain commit newNodes = do+ currentHead <- getBestBlockHeader+ action <- go =<< findSplitNode currentHead (last newNodes)+ when commit $ commitAction action+ return action+ where+ go (split, old, new)+ | null old && not (null new) = return $ BestChain new+ | not (null old) && not (null new) =+ return $ if nodeChainWork (last new) > nodeChainWork (last old)+ then ChainReorg split old new+ else SideChain $ split : new+ | otherwise = return $ KnownChain newNodes++-- | Update the best block header of the action in the header tree+commitAction :: HeaderTree m => BlockChainAction -> m ()+commitAction action = do+ currentHead <- getBestBlockHeader+ case action of+ BestChain nodes -> unless (null nodes) $ do+ updateChildren $ currentHead:nodes+ forM_ nodes putBlockHeaderHeight+ setBestBlockHeader $ last nodes+ ChainReorg s _ ns -> unless (null ns) $ do+ updateChildren $ s:ns+ forM_ ns putBlockHeaderHeight+ setBestBlockHeader $ last ns+ KnownChain _ -> return ()+ SideChain _ -> return ()+ where+ updateChildren (a:b:xs) = do+ putBlockHeaderNode a{ nodeChild = Just $ nodeBlockHash b }+ updateChildren (b:xs)+ updateChildren _ = return ()++-- TODO: Add DOS return values+verifyBlockHeader :: HeaderTree m+ => BlockHeader+ -> Timestamp+ -> EitherT String m BlockHeaderNode+verifyBlockHeader bh adjustedTime = do+ unless (isValidPOW bh) $ left "Invalid proof of work"++ unless (blockTimestamp bh <= adjustedTime + 2 * 60 * 60) $+ left "Invalid header timestamp"++ parNodeM <- lift $ getBlockHeaderNode $ prevBlock bh+ parNode <- maybe (left "Parent block not found") return parNodeM++ nextWork <- lift $ nextWorkRequired parNode bh+ unless (blockBits bh == nextWork) $ left "Incorrect work transition (bits)"++ let sortedMedians = sort $ nodeMedianTimes parNode+ medianTime = sortedMedians !! (length sortedMedians `div` 2)+ when (blockTimestamp bh <= medianTime) $ left "Block timestamp is too early"++ chkPointM <- lift lastSeenCheckpoint+ let newHeight = nodeHeaderHeight parNode + 1+ unless (maybe True ((fromIntegral newHeight >) . fst) chkPointM) $+ left "Rewriting pre-checkpoint chain"++ unless (verifyCheckpoint (fromIntegral newHeight) bid) $+ left "Rejected by checkpoint lock-in"++ -- All block of height 227836 or more use version 2 in prodnet+ -- TODO: Find out the value here for testnet+ when ( networkName == "prodnet"+ && blockVersion bh == 1+ && nodeHeaderHeight parNode + 1 >= 227836) $+ left "Rejected version=1 block"++ return parNode+ where+ bid = headerHash bh++-- Build a new block header and store it+storeBlockHeader :: HeaderTree m+ => BlockHeader+ -> BlockHeaderNode+ -> m BlockHeaderNode+storeBlockHeader bh parNode = do+ let nodeBlockHash = bid+ nodeHeader = bh+ nodeHeaderHeight = newHeight+ nodeChainWork = newWork+ nodeChild = Nothing+ nodeMedianTimes = newMedian+ nodeMinWork = minWork+ newNode = BlockHeaderNode{..}+ prevM <- getBlockHeaderNode bid+ case prevM of+ Just prev -> return prev+ Nothing -> putBlockHeaderNode newNode >> return newNode+ where+ bid = headerHash bh+ newHeight = nodeHeaderHeight parNode + 1+ newWork = nodeChainWork parNode + headerWork bh+ newMedian = blockTimestamp bh : take 10 (nodeMedianTimes parNode)+ isDiffChange = newHeight `mod` diffInterval == 0+ isNotLimit = blockBits bh /= encodeCompact powLimit+ minWork | not allowMinDifficultyBlocks = 0+ | isDiffChange || isNotLimit = blockBits bh+ | otherwise = nodeMinWork parNode++-- | Return the window of nodes starting from the child of the given node.+-- Child links are followed to build the window. If the window ends in an+-- orphaned chain, we backtrack and return the window in the main chain.+-- The result is returned in a BlockChainAction to know if we had to+-- backtrack into the main chain or not.+getNodeWindow :: HeaderTree m+ => BlockHash -> Int -> m (Maybe BlockChainAction)+getNodeWindow bh cnt = getBlockHeaderNode bh >>= \nodeM -> case nodeM of+ Just node -> go [] cnt node+ Nothing -> return Nothing+ where+ go [] 0 _ = return Nothing+ go acc 0 _ = return $ Just $ BestChain $ reverse acc+ go acc i node = getChildNode node >>= \childM -> case childM of+ Just child -> go (child:acc) (i-1) child+ Nothing -> do+ -- We are at the end of our chain. Check if there is a better chain.+ currentHead <- getBestBlockHeader+ if nodeChainWork currentHead > nodeChainWork node+ -- We got stuck in an orphan chain. We need to backtrack.+ then findMainChain currentHead+ -- We are at the end of the main chain+ else return $ if null acc+ then Nothing+ else Just $ BestChain $ reverse acc+ findMainChain currentHead = do+ -- Compute the split point from the original input node so that the old+ -- chain doesn't contain blocks beyond the original node.+ node <- fromMaybe e <$> getBlockHeaderNode bh+ (split, old, new) <- findSplitNode node currentHead+ return $ Just $ ChainReorg split old $ take cnt new+ e = error "getNodeWindow: Could not get block header node"++-- Find the first node right after the given timestamp+getNodeAtTimestamp :: HeaderTree m+ => Timestamp+ -> m (Maybe BlockHeaderNode)+getNodeAtTimestamp ts = do+ best <- getBestBlockHeader+ if ts < blockTimestamp (nodeHeader best)+ -- there is a solution. Perform a binary search.+ then go 0 $ nodeHeaderHeight best+ -- There is no solution.+ else return Nothing+ where+ go a b+ | a == b = getBlockHeaderByHeight a+ | otherwise = do+ -- compute the middle point+ let m = a + ((b - a) `div` 2)+ nodeM <- getBlockHeaderByHeight m+ case nodeM of+ Just node -> if ts < blockTimestamp (nodeHeader node)+ -- The block is after the timestamp. The block could be the+ -- solution, or the solution is smaller.+ then go a m+ -- The block is before the timestamp. The solution is+ -- striclty higher than m.+ else go (m+1) b+ _ -> error $ unwords+ [ "getNodeAtTimestamp: Possibly corrupted chain"+ , "at height", show m+ ]++-- | Find the split point between two nodes. It also returns the two partial+-- chains leading from the split point to the respective nodes.+findSplitNode :: HeaderTree m+ => BlockHeaderNode+ -> BlockHeaderNode+ -> m (BlockHeaderNode, [BlockHeaderNode], [BlockHeaderNode])+findSplitNode =+ go [] []+ where+ go xs ys x y+ | nodeBlockHash x == nodeBlockHash y = return (x, xs, ys)+ | nodeHeaderHeight x > nodeHeaderHeight y = do+ par <- fromMaybe e <$> getParentNode x+ go (x:xs) ys par y+ | otherwise = do+ par <- fromMaybe e <$> getParentNode y+ go xs (y:ys) x par+ e = error "findSplitNode: Could not get parent node"++-- | Finds the parent of a BlockHeaderNode+getParentNode :: HeaderTree m => BlockHeaderNode -> m (Maybe BlockHeaderNode)+getParentNode node+ | p == z = return Nothing+ | otherwise = getBlockHeaderNode p+ where+ p = prevBlock $ nodeHeader node+ z = "0000000000000000000000000000000000000000000000000000000000000000"++-- | Finds the child of a BlockHeaderNode if it exists. If a node has+-- multiple children, this function will always return the child on the+-- main branch.+getChildNode :: HeaderTree m => BlockHeaderNode -> m (Maybe BlockHeaderNode)+getChildNode node = case nodeChild node of+ Just child -> getBlockHeaderNode child+ Nothing -> return Nothing++-- | Get the last checkpoint that we have seen+lastSeenCheckpoint :: HeaderTree m => m (Maybe (Int, BlockHash))+lastSeenCheckpoint =+ go $ reverse checkpointList+ where+ go ((i, chk):xs) = do+ existsChk <- liftM isJust $ getBlockHeaderNode chk+ if existsChk then return $ Just (i, chk) else go xs+ go [] = return Nothing++-- | Returns the work required for a BlockHeader given the previous+-- BlockHeaderNode. This function coresponds to bitcoind function+-- GetNextWorkRequired in main.cpp.+nextWorkRequired :: HeaderTree m => BlockHeaderNode -> BlockHeader -> m Word32+nextWorkRequired lastNode bh+ -- Genesis block+ | prevBlock (nodeHeader lastNode) == z = return $ encodeCompact powLimit+ -- Only change the difficulty once per interval+ | (nodeHeaderHeight lastNode + 1) `mod` diffInterval /= 0 = return $+ if allowMinDifficultyBlocks+ then minPOW+ else blockBits $ nodeHeader lastNode+ | otherwise = do+ -- TODO: Can this break if there are not enough blocks in the chain?+ firstNode <- foldM (flip ($)) lastNode fs+ let lastTs = blockTimestamp $ nodeHeader firstNode+ return $ workFromInterval lastTs (nodeHeader lastNode)+ where+ len = fromIntegral diffInterval - 1+ fs = replicate len (liftM (fromMaybe e) . getParentNode)+ e = error "nextWorkRequired: Could not get parent node"+ delta = targetSpacing * 2+ minPOW+ | blockTimestamp bh > blockTimestamp (nodeHeader lastNode) + delta =+ encodeCompact powLimit+ | otherwise = nodeMinWork lastNode+ z = "0000000000000000000000000000000000000000000000000000000000000000"++-- | Computes the work required for the next block given a timestamp and the+-- current block. The timestamp should come from the block that matched the+-- last jump in difficulty (spaced out by 2016 blocks in prodnet).+workFromInterval :: Timestamp -> BlockHeader -> Word32+workFromInterval ts lastB+ | newDiff > powLimit = encodeCompact powLimit+ | otherwise = encodeCompact newDiff+ where+ t = fromIntegral $ blockTimestamp lastB - ts+ actualTime+ | t < targetTimespan `div` 4 = targetTimespan `div` 4+ | t > targetTimespan * 4 = targetTimespan * 4+ | otherwise = t+ lastDiff = decodeCompact $ blockBits lastB+ newDiff = lastDiff * toInteger actualTime `div` toInteger targetTimespan++-- | Returns a BlockLocator object (newest block first, genesis at the end)+blockLocator :: HeaderTree m => m BlockLocator+blockLocator = bestBlockHeaderHeight >>= blockLocatorHeight++-- | Returns a BlockLocator object for a given block height on the main chain.+blockLocatorHeight :: HeaderTree m => BlockHeight -> m BlockLocator+blockLocatorHeight height = do+ -- Take only indices > 0 to avoid the genesis block+ let is = takeWhile (> (0 :: Int)) $+ [h, (h - 1) .. (h - 9)] ++ [(h - 10) - 2^x | x <- [(0 :: Int)..]]+ ns <- liftM catMaybes $ forM (map fromIntegral is) getBlockHeaderByHeight+ return $ map nodeBlockHash ns ++ [headerHash genesisHeader]+ where+ h = fromIntegral height++-- | Build a block locator starting from a side chain. We do not have access+-- to the height index for the portion of the side chain so we build the+-- part in the side chain manually, then use the block height index for the+-- rest of the locator once we're back on the main chain.+blockLocatorSide :: HeaderTree m => BlockChainAction -> m BlockLocator+blockLocatorSide action = case action of+ SideChain (split:nodes) -> do+ mainLoc <- blockLocatorHeight $ nodeHeaderHeight split+ return $ take 10 (map nodeBlockHash $ reverse nodes) ++ mainLoc+ _ -> error "blockLocatorSide: Invalid blockchain action provided"++-- | Returns a partial BlockLocator object.+blockLocatorPartial :: HeaderTree m => Int -> m BlockLocator+blockLocatorPartial i+ | i < 1 = error "Locator length must be greater than 0"+ | otherwise = do+ h <- getBestBlockHeader+ liftM (map nodeBlockHash . reverse) $ go [] i h+ where+ go acc 1 node = return $ node:acc+ go acc step node = getParentNode node >>= \parM -> case parM of+ Just par -> go (node:acc) (step - 1) par+ Nothing -> return $ node:acc++bestBlockHeaderHeight :: HeaderTree m => m BlockHeight+bestBlockHeaderHeight = liftM nodeHeaderHeight getBestBlockHeader++getBlockHeaderHeight :: HeaderTree m => BlockHash -> m (Maybe BlockHeight)+getBlockHeaderHeight = return . fmap nodeHeaderHeight <=< getBlockHeaderNode++-- | Returns True if the difficulty target (bits) of the header is valid+-- and the proof of work of the header matches the advertised difficulty target.+-- This function corresponds to the function CheckProofOfWork from bitcoind+-- in main.cpp+isValidPOW :: BlockHeader -> Bool+isValidPOW bh+ | target <= 0 || target > powLimit = False+ | otherwise = headerPOW bh <= fromIntegral target+ where+ target = decodeCompact $ blockBits bh++-- | Returns the proof of work of a block header as an Integer number.+headerPOW :: BlockHeader -> Integer+headerPOW = bsToInteger . BS.reverse . encode' . headerHash++-- | Returns the work represented by this block. Work is defined as the number+-- of tries needed to solve a block in the average case with respect to the+-- target.+headerWork :: BlockHeader -> Integer+headerWork bh =+ largestHash `div` (target + 1)+ where+ target = decodeCompact (blockBits bh)+ largestHash = 1 `shiftL` 256++{- Default LevelDB implementation -}++blockHashKey :: BlockHash -> BS.ByteString+blockHashKey bid = "b_" `BS.append` encode' bid++bestBlockKey :: BS.ByteString+bestBlockKey = "bestblockheader_"++heightKey :: BlockHeight -> BS.ByteString+heightKey h = "h_" `BS.append` encode' h++-- Get a node which is directly referenced by the key+getLevelDBNode :: MonadIO m+ => BS.ByteString -> ReaderT L.DB m (Maybe BlockHeaderNode)+getLevelDBNode key = do+ db <- ask+ resM <- liftIO $ L.get db def key+ return $ decodeToMaybe =<< resM++-- Get a node that has 1 level of indirection+getLevelDBNode' :: MonadIO m+ => BS.ByteString -> ReaderT L.DB m (Maybe BlockHeaderNode)+getLevelDBNode' key = do+ db <- ask+ resM <- liftIO $ L.get db def key+ maybe (return Nothing) getLevelDBNode resM++instance MonadIO m => HeaderTree (ReaderT L.DB m) where+ getBlockHeaderNode = getLevelDBNode . blockHashKey+ putBlockHeaderNode node = do+ db <- ask+ liftIO $ L.put db def (blockHashKey $ nodeBlockHash node) $ encode' node+ getBlockHeaderByHeight = getLevelDBNode' . heightKey+ putBlockHeaderHeight node = do+ db <- ask+ let val = blockHashKey $ nodeBlockHash node+ liftIO $ L.put db def (heightKey $ nodeHeaderHeight node) val+ getBestBlockHeader = do+ db <- ask+ keyM <- liftIO $ L.get db def bestBlockKey+ case keyM of+ Just key -> fromMaybe e <$> getLevelDBNode key+ Nothing -> error+ "GetBestBlockHeader: Best block header does not exist"+ where+ e = error "getBestBlockHeader: Could not get LevelDB node"+ setBestBlockHeader node = do+ db <- ask+ liftIO $ L.put db def bestBlockKey $ blockHashKey $ nodeBlockHash node+
+ Network/Haskoin/Node/Peer.hs view
@@ -0,0 +1,703 @@+{-# LANGUAGE TemplateHaskell #-}+module Network.Haskoin.Node.Peer where++import System.Random (randomIO)++import Control.Monad (liftM, when, unless, join, forM_, forever)+import Control.Monad.Trans (MonadIO, lift, liftIO)+import Control.Monad.Trans.Control (MonadBaseControl)+import Control.Monad.Logger (MonadLogger, logInfo, logWarn, logDebug, logError)+import Control.Monad.Reader (ReaderT, asks, runReaderT)+import Control.Monad.State (StateT, evalStateT, get, put)+import Control.Concurrent (threadDelay, killThread, myThreadId)+import qualified Control.Concurrent.STM.Lock as Lock (with)+import Control.Exception.Lifted (fromException, finally, throwIO, throw)+import Control.Concurrent.STM.TBMChan+ ( TBMChan, writeTBMChan, closeTBMChan, newTBMChan )+import Control.Concurrent.Async.Lifted+ ( race, withAsync, waitAnyCancel, link, waitCatch )+import Control.Concurrent.STM+ ( STM, atomically, readTVar, modifyTVar', swapTVar, retry, newTVarIO )++import Data.List (nub, sort, sortBy)+import Data.Text (Text, pack)+import Data.Word (Word32)+import Data.Maybe (fromMaybe, isJust, listToMaybe)+import Data.Time.Clock.POSIX (getPOSIXTime)+import Data.Time.Clock (getCurrentTime, diffUTCTime)+import Data.Unique (hashUnique, newUnique)+import Data.Conduit (Conduit, Sink, awaitForever, yield, ($$), ($=))+import Data.Conduit.TMChan (sourceTBMChan)+import qualified Data.Conduit.Binary as CB (take)+import qualified Data.ByteString.Char8 as C (pack)+import qualified Data.ByteString as BS (ByteString, null, append)+import qualified Data.ByteString.Lazy as BL (toStrict)+import Data.String.Conversions (cs)+import qualified Data.Map as M+ ( keys , lookup, assocs, elems, fromList, unionWith )+import Data.Conduit.Network+ ( runGeneralTCPClient, appSink, appSource, clientSettings )++import qualified Database.LevelDB.Base as L (DB, withDB)++import Network.Socket (SockAddr (SockAddrInet))++import Network.Haskoin.Node+import Network.Haskoin.Node.STM+import Network.Haskoin.Node.HeaderTree+import Network.Haskoin.Transaction+import Network.Haskoin.Block+import Network.Haskoin.Constants+import Network.Haskoin.Util++-- TODO: Move constants elsewhere ?+minProtocolVersion :: Word32+minProtocolVersion = 70001++-- Start a reconnecting peer that will idle once the connection is established+-- and the handshake is performed.+startPeer :: (MonadIO m, MonadLogger m, MonadBaseControl IO m)+ => PeerHost+ -> NodeT m ()+startPeer ph@PeerHost{..} = do+ -- Create a new unique ID for this peer+ pid <- liftIO newUnique+ -- Start the peer with the given PID+ startPeerPid pid ph++-- Start a peer that will try to reconnect when the connection is closed. The+-- reconnections are performed using an expoential backoff time. This function+-- blocks until the peer cannot reconnect (either the peer is banned or we+-- already have a peer connected to the given peer host).+startReconnectPeer :: (MonadIO m, MonadLogger m, MonadBaseControl IO m)+ => PeerHost+ -> NodeT m ()+startReconnectPeer ph@PeerHost{..} = do+ -- Create a new unique ID for this peer+ pid <- liftIO newUnique+ -- Wait if there is a reconnection timeout+ maybeWaitReconnect pid+ -- Launch the peer+ withAsync (startPeerPid pid ph) $ \a -> do+ resE <- liftIO $ waitCatch a+ reconnect <- case resE of+ Left se -> do+ $(logError) $ formatPid pid ph $ unwords+ [ "Peer thread stopped with exception:", show se ]+ return $ case fromException se of+ Just NodeExceptionBanned -> False+ Just NodeExceptionConnected -> False+ _ -> True+ Right _ -> do+ $(logDebug) $ formatPid pid ph "Peer thread stopped"+ return True+ -- Try to reconnect+ when reconnect $ startReconnectPeer ph+ where+ maybeWaitReconnect pid = do+ reconnect <- atomicallyNodeT $ do+ sessM <- getHostSession ph+ case sessM of+ Just PeerHostSession{..} -> do+ -- Compute the new reconnection time (max 15 minutes)+ let reconnect = min 900 $ 2 * peerHostSessionReconnect+ -- Save the reconnection time+ modifyHostSession ph $ \s ->+ s{ peerHostSessionReconnect = reconnect }+ return reconnect+ _ -> return 0++ when (reconnect > 0) $ do+ $(logInfo) $ formatPid pid ph $ unwords+ [ "Reconnecting peer in", show reconnect, "seconds" ]+ -- Wait for some time before calling a reconnection+ liftIO $ threadDelay $ reconnect * 1000000++-- Start a peer with with the given peer host/peer id and initiate the+-- network protocol handshake. This function will block until the peer+-- connection is closed or an exception is raised.+startPeerPid :: (MonadIO m, MonadLogger m, MonadBaseControl IO m)+ => PeerId+ -> PeerHost+ -> NodeT m ()+startPeerPid pid ph@PeerHost{..} = do+ -- Check if the peer host is banned+ banned <- atomicallyNodeT $ isPeerHostBanned ph+ when banned $ do+ $(logWarn) $ formatPid pid ph "Failed to start banned host"+ liftIO $ throwIO NodeExceptionBanned++ -- Check if the peer host is already connected+ connected <- atomicallyNodeT $ isPeerHostConnected ph+ when connected $ do+ $(logWarn) $ formatPid pid ph "This host is already connected"+ liftIO $ throwIO NodeExceptionConnected++ tid <- liftIO myThreadId+ chan <- liftIO . atomically $ newTBMChan 1024+ mChan <- liftIO . atomically $ newTBMChan 1024+ pings <- liftIO $ newTVarIO []+ atomicallyNodeT $ do+ newPeerSession pid PeerSession+ { peerSessionConnected = False+ , peerSessionVersion = Nothing+ , peerSessionHeight = 0+ , peerSessionChan = chan+ , peerSessionHost = ph+ , peerSessionThreadId = tid+ , peerSessionMerkleChan = mChan+ , peerSessionPings = pings+ , peerSessionScore = Nothing+ }+ newHostSession ph PeerHostSession+ { peerHostSessionScore = 0+ , peerHostSessionReconnect = 1+ , peerHostSessionLog = []+ }++ $(logDebug) $ formatPid pid ph "Starting a new client TCP connection"++ -- Start the client TCP connection+ let c = clientSettings peerPort $ C.pack peerHost+ runGeneralTCPClient c (peerTCPClient chan) `finally` cleanupPeer+ return ()+ where+ peerTCPClient chan ad = do+ -- Conduit for receiving messages from the remote host+ let recvMsg = appSource ad $$ decodeMessage pid ph+ -- Conduit for sending messages to the remote host+ sendMsg = sourceTBMChan chan $= encodeMessage $$ appSink ad++ withAsync (evalStateT recvMsg Nothing) $ \a1 -> link a1 >> do+ $(logDebug) $ formatPid pid ph+ "Receiving message thread started..."+ withAsync sendMsg $ \a2 -> link a2 >> do+ $(logDebug) $ formatPid pid ph+ "Sending message thread started..."+ -- Perform the peer handshake before we continue+ -- Timeout after 2 minutes+ resE <- raceTimeout 120 (disconnectPeer pid ph)+ (peerHandshake pid ph chan)+ case resE of+ Left _ -> $(logError) $ formatPid pid ph+ "Peer timed out during the connection handshake"+ _ -> do+ -- Send the bloom filter if we have one+ $(logDebug) $ formatPid pid ph+ "Sending the bloom filter if we have one"+ atomicallyNodeT $ do+ bloomM <- readTVarS sharedBloomFilter+ case bloomM of+ Just (bloom, _) ->+ sendMessage pid $+ MFilterLoad $ FilterLoad bloom+ _ -> return ()+ withAsync (peerPing pid ph) $ \a3 -> link a3 >> do+ $(logDebug) $ formatPid pid ph "Ping thread started"+ _ <- liftIO $ waitAnyCancel [a1, a2, a3]+ return ()++ cleanupPeer = do+ $(logWarn) $ formatPid pid ph "Peer is closing. Running cleanup..."+ atomicallyNodeT $ do+ -- Remove the header syncing peer if necessary+ hPidM <- readTVarS sharedHeaderPeer+ when (hPidM == Just pid) $ writeTVarS sharedHeaderPeer Nothing+ -- Remove the merkle syncing peer if necessary+ mPidM <- readTVarS sharedMerklePeer+ when (mPidM == Just pid) $ writeTVarS sharedMerklePeer Nothing+ -- Remove the session and close the channels+ sessM <- removePeerSession pid+ case sessM of+ Just PeerSession{..} -> lift $ do+ closeTBMChan peerSessionChan+ closeTBMChan peerSessionMerkleChan+ _ -> return ()+ -- Update the network height+ updateNetworkHeight++-- Return True if the PeerHost is banned+isPeerHostBanned :: PeerHost -> NodeT STM Bool+isPeerHostBanned ph = do+ hostMap <- readTVarS sharedHostMap+ case M.lookup ph hostMap of+ Just sessTVar -> do+ sess <- lift $ readTVar sessTVar+ return $ isHostScoreBanned $ peerHostSessionScore sess+ _ -> return False++-- Returns True if we have a peer connected to that PeerHost already+isPeerHostConnected :: PeerHost -> NodeT STM Bool+isPeerHostConnected ph = do+ peerMap <- readTVarS sharedPeerMap+ sess <- lift $ mapM readTVar $ M.elems peerMap+ return $ ph `elem` map peerSessionHost sess++-- | Decode messages sent from the remote host and send them to the peers main+-- message queue for processing. If we receive invalid messages, this function+-- will also notify the PeerManager about a misbehaving remote host.+decodeMessage+ :: (MonadLogger m, MonadIO m, MonadBaseControl IO m)+ => PeerId+ -> PeerHost+ -> Sink BS.ByteString (StateT (Maybe (MerkleBlock, MerkleTxs)) (NodeT m)) ()+decodeMessage pid ph = do+ -- Message header is always 24 bytes+ headerBytes <- BL.toStrict <$> CB.take 24+ -- If headerBytes is empty, the conduit has disconnected and we need to+ -- exit (not recurse). Otherwise, we go into an infinite loop here.+ unless (BS.null headerBytes) $ do+ -- Introspection required to know the length of the payload+ case decodeToEither headerBytes of+ Left err -> lift . lift $ misbehaving pid ph moderateDoS $ unwords+ [ "Could not decode message header:", err+ , "Bytes:", cs (encodeHex headerBytes)+ ]+ Right (MessageHeader _ cmd len _) -> do+ $(logDebug) $ formatPid pid ph $ unwords+ [ "Received message header of type", show cmd ]+ payloadBytes <- BL.toStrict <$> CB.take (fromIntegral len)+ case decodeToEither $ headerBytes `BS.append` payloadBytes of+ Left err -> lift . lift $ misbehaving pid ph moderateDoS $+ unwords [ "Could not decode message payload:", err ]+ Right msg -> lift $ processMessage pid ph msg+ decodeMessage pid ph++-- Handle a message from a peer+processMessage :: (MonadLogger m, MonadIO m, MonadBaseControl IO m)+ => PeerId+ -> PeerHost+ -> Message+ -> StateT (Maybe (MerkleBlock, MerkleTxs)) (NodeT m) ()+processMessage pid ph msg = checkMerkleEnd >> case msg of+ MVersion v -> lift $ do+ $(logDebug) $ formatPid pid ph "Processing MVersion message"+ join . atomicallyNodeT $ do+ oldVerM <- liftM peerSessionVersion $ getPeerSession pid+ case oldVerM of+ Just _ -> do+ _ <- trySendMessage pid $ MReject $ reject+ MCVersion RejectDuplicate "Duplicate version message"+ return $+ misbehaving pid ph minorDoS "Duplicate version message"+ Nothing -> do+ modifyPeerSession pid $ \s ->+ s{ peerSessionVersion = Just v }+ return $ return ()+ $(logDebug) $ formatPid pid ph "Done processing MVersion message"+ MPing (Ping nonce) -> lift $ do+ $(logDebug) $ formatPid pid ph "Processing MPing message"+ -- Just reply to the Ping with a Pong message+ _ <- atomicallyNodeT $ trySendMessage pid $ MPong $ Pong nonce+ return ()+ MPong (Pong nonce) -> lift $ do+ $(logDebug) $ formatPid pid ph "Processing MPong message"+ atomicallyNodeT $ do+ PeerSession{..} <- getPeerSession pid+ -- Add the Pong response time+ lift $ modifyTVar' peerSessionPings (++ [nonce])+ MHeaders h -> lift $ do+ $(logDebug) $ formatPid pid ph "Processing MHeaders message"+ _ <- atomicallyNodeT $ tryPutTMVarS sharedHeaders (pid, h)+ return ()+ MInv inv -> lift $ do+ $(logDebug) $ formatPid pid ph "Processing MInv message"+ processInvMessage pid ph inv+ MGetData (GetData inv) -> do+ $(logDebug) $ formatPid pid ph "Processing MGetData message"+ let txlist = filter ((== InvTx) . invType) inv+ txids = nub $ map (TxHash . invHash) txlist+ $(logDebug) $ formatPid pid ph $ unlines $+ "Received GetData request for transactions"+ : map ((" " ++) . cs . txHashToHex) txids+ -- Add the txids to the GetData request map+ mapTVar <- asks sharedTxGetData+ liftIO . atomically $ modifyTVar' mapTVar $ \datMap ->+ let newMap = M.fromList $ map (\tid -> (tid, [(pid, ph)])) txids+ in M.unionWith (\x -> nub . (x ++)) newMap datMap+ MTx tx -> do+ $(logDebug) $ formatPid pid ph "Processing MTx message"+ PeerSession{..} <- lift . atomicallyNodeT $ getPeerSession pid+ txChan <- lift $ asks sharedTxChan+ get >>= \merkleM -> case merkleM of+ Just (_, mTxs) -> if txHash tx `elem` mTxs+ then do+ $(logDebug) $ formatPid pid ph $ unwords+ [ "Received merkle tx", cs $ txHashToHex $ txHash tx ]+ liftIO . atomically $+ writeTBMChan peerSessionMerkleChan $ Right tx+ else do+ $(logDebug) $ formatPid pid ph $ unwords+ [ "Received tx broadcast (ending a merkle block)"+ , cs $ txHashToHex $ txHash tx+ ]+ endMerkle+ liftIO . atomically $ writeTBMChan txChan (pid, ph, tx)+ _ -> do+ $(logDebug) $ formatPid pid ph $ unwords+ [ "Received tx broadcast", cs $ txHashToHex $ txHash tx ]+ liftIO . atomically $ writeTBMChan txChan (pid, ph, tx)+ MMerkleBlock mb@(MerkleBlock mHead ntx hs fs) -> do+ $(logDebug) $ formatPid pid ph "Processing MMerkleBlock message"+ case extractMatches fs hs (fromIntegral ntx) of+ Left err -> lift $ misbehaving pid ph severeDoS $ unwords+ [ "Received an invalid merkle block:", err ]+ Right (decodedRoot, mTxs) ->+ -- Make sure that the merkle roots match+ if decodedRoot == merkleRoot mHead+ then do+ $(logDebug) $ formatPid pid ph $ unwords+ [ "Received valid merkle block"+ , cs $ blockHashToHex $ headerHash mHead+ ]+ if null mTxs+ -- Deliver the merkle block+ then lift . atomicallyNodeT $ do+ PeerSession{..} <- getPeerSession pid+ lift $ writeTBMChan peerSessionMerkleChan $+ Left (mb, [])+ -- Buffer the merkle block until we received all txs+ else put $ Just (mb, mTxs)+ else lift $ misbehaving pid ph severeDoS+ "Received a merkle block with an invalid merkle root"+ _ -> return () -- Ignore other requests+ where+ checkMerkleEnd = unless (isTxMsg msg) endMerkle+ endMerkle = get >>= \merkleM -> case merkleM of+ Just (mb, mTxs) -> do+ lift . atomicallyNodeT $ do+ PeerSession{..} <- getPeerSession pid+ lift $ writeTBMChan peerSessionMerkleChan $ Left (mb, mTxs)+ put Nothing+ _ -> return ()+ isTxMsg (MTx _) = True+ isTxMsg _ = False++processInvMessage :: (MonadLogger m, MonadIO m, MonadBaseControl IO m)+ => PeerId+ -> PeerHost+ -> Inv+ -> NodeT m ()+processInvMessage pid ph (Inv vs) = case tickleM of+ Just tickle -> do+ $(logDebug) $ formatPid pid ph $ unwords+ [ "Received block tickle", cs $ blockHashToHex tickle ]+ tickleChan <- asks sharedTickleChan+ liftIO $ atomically $ writeTBMChan tickleChan (pid, ph, tickle)+ _ -> do+ unless (null txlist) $ do+ forM_ txlist $ \tid -> $(logDebug) $ formatPid pid ph $ unwords+ [ "Received transaction INV", cs (txHashToHex tid) ]+ -- We simply request the transactions.+ -- TODO: Should we do something more elaborate here?+ atomicallyNodeT $ sendMessage pid $ MGetData $ GetData $+ map (InvVector InvTx . getTxHash) txlist+ unless (null blocklist) $ do+ $(logDebug) $ formatPid pid ph $ unlines $+ "Received block INV"+ : map ((" " ++) . cs . blockHashToHex) blocklist+ -- We ignore block INVs as we do headers-first sync+ return ()+ where+ -- Single blockhash INV is a tickle+ tickleM = case blocklist of+ [h] -> if null txlist then Just h else Nothing+ _ -> Nothing+ txlist :: [TxHash]+ txlist = map (TxHash . invHash) $+ filter ((== InvTx) . invType) vs+ blocklist :: [BlockHash]+ blocklist = map (BlockHash . invHash) $ filter ((== InvBlock) . invType) vs++-- | Encode message that are being sent to the remote host.+encodeMessage :: (MonadIO m, MonadLogger m)+ => Conduit Message (NodeT m) BS.ByteString+encodeMessage = awaitForever $ yield . encode'++peerPing :: (MonadIO m, MonadLogger m, MonadBaseControl IO m)+ => PeerId+ -> PeerHost+ -> NodeT m ()+peerPing pid ph = forever $ do+ nonce <- liftIO randomIO+ (nonceTVar, busy) <- atomicallyNodeT $ do+ busy <- isPeerBusy pid+ PeerSession{..} <- getPeerSession pid+ sendMessage pid $ MPing $ Ping nonce+ return (peerSessionPings, busy)++ $(logDebug) $ formatPid pid ph $ unwords+ [ "Waiting for Ping nonce", show nonce ]+ -- Wait 120 seconds for the pong or time out+ startTime <- liftIO getCurrentTime+ resE <- raceTimeout 120 (killPeer nonce) (waitPong nonce nonceTVar)+ case resE of+ Right _ -> do+ endTime <- liftIO getCurrentTime+ (diff, score) <- atomicallyNodeT $ do+ PeerSession{..} <- getPeerSession pid+ -- Compute the ping time and the new score+ let diff = diffUTCTime endTime startTime+ score = 0.5*diff + 0.5*(fromMaybe diff peerSessionScore)+ -- Save the score in the peer session unless the peer is busy+ unless busy $ modifyPeerSession pid $+ \s -> s{ peerSessionScore = Just score }+ return (diff, score)+ $(logDebug) $ formatPid pid ph $ unwords+ [ "Got response to ping", show nonce+ , "with time", show diff, "and score", show score+ ]+ _ -> return ()++ -- Sleep 30 seconds before sending the next ping+ liftIO $ threadDelay $ 30 * 1000000+ where+ -- Wait for the Pong message of our Ping nonce to arrive+ waitPong nonce nonceTVar = do+ ns <- liftIO . atomically $ do+ ns <- swapTVar nonceTVar []+ if null ns then retry else return ns+ if nonce `elem` ns then return () else waitPong nonce nonceTVar+ killPeer nonce = do+ $(logWarn) $ formatPid pid ph $ concat+ [ "Did not receive a timely reply for Ping ", show nonce+ , ". Reconnecting the peer."+ ]+ disconnectPeer pid ph++peerHandshake :: (MonadIO m, MonadLogger m, MonadBaseControl IO m)+ => PeerId+ -> PeerHost+ -> TBMChan Message+ -> NodeT m ()+peerHandshake pid ph chan = do+ ourVer <- buildVersion+ $(logDebug) $ formatPid pid ph "Sending our version message"+ liftIO . atomically $ writeTBMChan chan $ MVersion ourVer+ -- Wait for the peer version message to arrive+ $(logDebug) $ formatPid pid ph "Waiting for the peers version message..."+ peerVer <- atomicallyNodeT $ waitPeerVersion pid+ $(logInfo) $ formatPid pid ph $ unlines+ [ unwords [ "Connected to peer host"+ , show $ naAddress $ addrSend peerVer+ ]+ , unwords [ " version :", show $ version peerVer ]+ , unwords [ " subVer :", show $ userAgent peerVer ]+ , unwords [ " services :", show $ services peerVer ]+ , unwords [ " time :", show $ timestamp peerVer ]+ , unwords [ " blocks :", show $ startHeight peerVer ]+ ]+ -- Check the protocol version+ if version peerVer < minProtocolVersion+ then misbehaving pid ph severeDoS $ unwords+ [ "Connected to a peer speaking protocol version"+ , show $ version peerVer+ , "but we require at least"+ , show $ minProtocolVersion+ ]+ else do+ atomicallyNodeT $ do+ -- Save the peers height and update the network height+ modifyPeerSession pid $ \s ->+ s{ peerSessionHeight = startHeight peerVer+ , peerSessionConnected = True+ }+ updateNetworkHeight+ -- Reset the reconnection timer (exponential backoff)+ modifyHostSession ph $ \s -> s{ peerHostSessionReconnect = 1 }+ -- ACK the version message+ lift $ writeTBMChan chan MVerAck++ $(logDebug) $ formatPid pid ph "Handshake complete"+ where+ buildVersion = do+ -- TODO: Get our correct IP here+ let add = NetworkAddress 1 $ SockAddrInet 0 0+ ua = VarString haskoinUserAgent+ time <- liftM floor $ liftIO getPOSIXTime+ rdmn <- liftIO randomIO -- nonce+ h <- runHeaderTree bestBlockHeaderHeight+ return Version { version = 70001+ , services = 1+ , timestamp = time+ , addrRecv = add+ , addrSend = add+ , verNonce = rdmn+ , userAgent = ua+ , startHeight = h+ , relay = False+ }++-- Wait for the version message of a peer and return it+waitPeerVersion :: PeerId -> NodeT STM Version+waitPeerVersion pid = do+ PeerSession{..} <- getPeerSession pid+ case peerSessionVersion of+ Just ver -> return ver+ _ -> lift retry++-- Delete the session of a peer and send a kill signal to the peers thread.+-- Unless the peer is banned, the peer will try to reconnect.+disconnectPeer :: (MonadIO m, MonadLogger m)+ => PeerId+ -> PeerHost+ -> NodeT m ()+disconnectPeer pid ph = do+ sessM <- atomicallyNodeT $ tryGetPeerSession pid+ case sessM of+ Just PeerSession{..} -> do+ $(logDebug) $ formatPid pid ph "Killing the peer thread"+ liftIO $ killThread peerSessionThreadId+ _ -> return ()++{- Peer utility functions -}++-- Is the peer busy?+isPeerBusy :: PeerId -> NodeT STM Bool+isPeerBusy pid = do+ hPidM <- readTVarS sharedHeaderPeer+ mPidM <- readTVarS sharedMerklePeer+ return $ Just pid `elem` [hPidM, mPidM]++-- Wait for a non-empty bloom filter to be available+waitBloomFilter :: NodeT STM BloomFilter+waitBloomFilter =+ maybe (lift retry) (return . fst) =<< readTVarS sharedBloomFilter++sendBloomFilter :: BloomFilter -> Int -> NodeT STM ()+sendBloomFilter bloom elems = unless (isBloomEmpty bloom) $ do+ oldBloomM <- readTVarS sharedBloomFilter+ let oldElems = maybe 0 snd oldBloomM+ -- Only update the bloom filter if the number of elements is larger+ when (elems > oldElems) $ do+ writeTVarS sharedBloomFilter $ Just (bloom, elems)+ sendMessageAll $ MFilterLoad $ FilterLoad bloom++-- Returns the median height of all the peers+getMedianHeight :: NodeT STM BlockHeight+getMedianHeight = do+ hs <- liftM (map (peerSessionHeight . snd)) getConnectedPeers+ let (_,ms) = splitAt (length hs `div` 2) $ sort hs+ return $ fromMaybe 0 $ listToMaybe ms++-- Set the network height to the median height of all peers.+updateNetworkHeight :: NodeT STM ()+updateNetworkHeight = writeTVarS sharedNetworkHeight =<< getMedianHeight++getPeers :: NodeT STM [(PeerId, PeerSession)]+getPeers = do+ peerMap <- readTVarS sharedPeerMap+ lift $ mapM f $ M.assocs peerMap+ where+ f (pid, sess) = liftM ((,) pid) $ readTVar sess++getConnectedPeers :: NodeT STM [(PeerId, PeerSession)]+getConnectedPeers = liftM (filter (peerSessionConnected . snd)) getPeers++-- Returns a peer that is connected, at the network height and+-- with the best score.+getPeersAtNetHeight :: NodeT STM [(PeerId, PeerSession)]+getPeersAtNetHeight = do+ -- Find the current network height+ height <- readTVarS sharedNetworkHeight+ getPeersAtHeight (== height)++-- Find the best peer at the given height+getPeersAtHeight :: (BlockHeight -> Bool)+ -> NodeT STM [(PeerId, PeerSession)]+getPeersAtHeight cmpHeight = do+ peers <- liftM (filter f) getPeers+ -- Choose the peer with the best score+ return $ sortBy s peers+ where+ f (_, p) =+ peerSessionConnected p && -- Only connected peers+ isJust (peerSessionScore p) && -- Only peers with scores+ cmpHeight (peerSessionHeight p) -- Only peers at the required height+ s (_,a) (_,b) = peerSessionScore a `compare` peerSessionScore b++-- Send a message to a peer only if it is connected. It returns True on+-- success.+trySendMessage :: PeerId -> Message -> NodeT STM Bool+trySendMessage pid msg = do+ sessM <- tryGetPeerSession pid+ lift $ case sessM of+ Just PeerSession{..} ->+ if peerSessionConnected+ then writeTBMChan peerSessionChan msg >> return True+ else return False -- The peer is not yet connected+ _ -> return False -- The peer does not exist++-- Send a message to a peer only if it is connected. It returns True on+-- success. Throws an exception if the peer does not exist or is not connected.+sendMessage :: PeerId -> Message -> NodeT STM ()+sendMessage pid msg = do+ PeerSession{..} <- getPeerSession pid+ if peerSessionConnected+ then lift $ writeTBMChan peerSessionChan msg+ else throw $ NodeExceptionPeerNotConnected pid++-- Send a message to all connected peers.+sendMessageAll :: Message -> NodeT STM ()+sendMessageAll msg = do+ peerMap <- readTVarS sharedPeerMap+ forM_ (M.keys peerMap) $ \pid -> trySendMessage pid msg++getNetworkHeight :: NodeT STM BlockHeight+getNetworkHeight = readTVarS sharedNetworkHeight++misbehaving :: (MonadIO m, MonadLogger m)+ => PeerId+ -> PeerHost+ -> (PeerHostScore -> PeerHostScore)+ -> String+ -> NodeT m ()+misbehaving pid ph f msg = do+ sessM <- atomicallyNodeT $ do+ modifyHostSession ph $ \s ->+ s{ peerHostSessionScore = f $! peerHostSessionScore s+ , peerHostSessionLog = msg:(peerHostSessionLog s)+ }+ getHostSession ph+ case sessM of+ Just PeerHostSession{..} -> do+ $(logWarn) $ formatPid pid ph $ unlines+ [ "Misbehaving peer"+ , unwords [ " Score:", show peerHostSessionScore ]+ , unwords [ " Reason:", msg ]+ ]+ when (isHostScoreBanned peerHostSessionScore) $+ disconnectPeer pid ph+ _ -> return ()++{- LevelDB function -}++runHeaderTree :: MonadIO m => ReaderT L.DB IO a -> NodeT m a+runHeaderTree action = do+ lock <- asks sharedLevelDBLock+ fp <- asks levelDBFilePath+ opts <- asks levelDBOptions+ liftIO $ Lock.with lock $ L.withDB fp opts $ runReaderT action++{- Utilities -}++raceTimeout :: (MonadIO m, MonadBaseControl IO m)+ => Int+ -- ^ Timeout value in seconds+ -> m a+ -- ^ Action to run if the main action times out+ -> m b+ -- ^ Action to run until the time runs out+ -> m (Either a b)+raceTimeout sec cleanup action = do+ resE <- race (liftIO $ threadDelay (sec * 1000000)) action+ case resE of+ Right res -> return $ Right res+ Left _ -> liftM Left cleanup++formatPid :: PeerId -> PeerHost -> String -> Text+formatPid pid ph str = pack $ concat+ [ "[Peer ", show $ hashUnique pid+ , " | ", peerHostString ph, "] ", str+ ]+
+ Network/Haskoin/Node/STM.hs view
@@ -0,0 +1,408 @@+{-# OPTIONS_GHC -fno-warn-orphans #-}+{-# LANGUAGE TemplateHaskell #-}+module Network.Haskoin.Node.STM where++import Control.Monad (liftM, (<=<))+import Control.Monad.Trans (MonadIO, lift, liftIO)+import Control.Monad.Reader (ReaderT, ask, asks, runReaderT)+import Control.Monad.Trans.Control (MonadBaseControl)+import Control.Monad.Logger (MonadLogger, logDebug)+import Control.Concurrent (ThreadId)+import Control.Concurrent.STM.Lock (Lock)+import Control.DeepSeq (NFData(..))+import qualified Control.Concurrent.STM.Lock as Lock (new)+import Control.Concurrent.STM.TBMChan+ (TBMChan, closeTBMChan, newTBMChan)+import Control.Exception.Lifted+ (SomeException, Exception, fromException, throw, catch)+import Control.Concurrent.STM+ ( STM, TMVar, TVar, atomically, orElse+ , readTVar, writeTVar, newTVarIO, modifyTVar', newTVar+ , tryPutTMVar, takeTMVar, newEmptyTMVarIO, isEmptyTMVar, putTMVar+ , tryReadTMVar, readTMVar+ )++import Data.Time.Clock (NominalDiffTime)+import Data.Unique (Unique, hashUnique)+import Data.Typeable (Typeable)+import Data.Word (Word32, Word64)+import Data.Maybe (isJust)+import Data.Aeson.TH (deriveJSON)+import qualified Data.Map.Strict as M (Map, insert, lookup, delete, empty)++import qualified Database.LevelDB.Base as L (Options(..), withDB)++import Network.Haskoin.Util+import Network.Haskoin.Node+import Network.Haskoin.Block+import Network.Haskoin.Transaction+import Network.Haskoin.Constants+import Network.Haskoin.Node.HeaderTree++{- Type aliases -}++type MerkleTxs = [TxHash]+type NodeT = ReaderT SharedNodeState+type PeerId = Unique+type PeerHostScore = Word32++instance Show PeerId where+ show = show . hashUnique++getNodeState :: (MonadIO m, MonadLogger m)+ => FilePath+ -> L.Options+ -> m SharedNodeState+getNodeState levelDBFilePath levelDBOptions = do+ -- Initialize the HeaderTree+ $(logDebug) "Initializing the HeaderTree and NodeState"+ liftIO $ do+ L.withDB levelDBFilePath levelDBOptions $ runReaderT initHeaderTree+ sharedPeerMap <- newTVarIO M.empty+ sharedHostMap <- newTVarIO M.empty+ sharedNetworkHeight <- newTVarIO 0+ sharedHeaders <- newEmptyTMVarIO+ sharedHeaderPeer <- newTVarIO Nothing+ sharedMerklePeer <- newTVarIO Nothing+ sharedSyncLock <- atomically Lock.new+ sharedTickleChan <- atomically $ newTBMChan 1024+ sharedTxChan <- atomically $ newTBMChan 1024+ sharedTxGetData <- newTVarIO M.empty+ sharedRescan <- newEmptyTMVarIO+ sharedMempool <- newTVarIO False+ sharedLevelDBLock <- atomically Lock.new+ sharedBloomFilter <- newTVarIO Nothing+ -- Find our best node in the HeaderTree+ sharedBestHeader <- newTVarIO =<< L.withDB+ levelDBFilePath+ levelDBOptions+ (runReaderT getBestBlockHeader)+ sharedBestBlock <- newTVarIO $ headerHash genesisHeader+ return SharedNodeState{..}++runNodeT :: Monad m => SharedNodeState -> NodeT m a -> m a+runNodeT state action = runReaderT action state++withNodeT :: (MonadIO m, MonadLogger m)+ => FilePath+ -> L.Options+ -> NodeT m a+ -> m a+withNodeT fp opts action = flip runNodeT action =<< getNodeState fp opts++atomicallyNodeT :: MonadIO m => NodeT STM a -> NodeT m a+atomicallyNodeT action = liftIO . atomically . runReaderT action =<< ask++{- PeerHost Session -}++data PeerHostSession = PeerHostSession+ { peerHostSessionScore :: !PeerHostScore+ , peerHostSessionReconnect :: !Int+ , peerHostSessionLog :: ![String]+ -- ^ Important host log messages that should appear in status command+ }++instance NFData PeerHostSession where+ rnf PeerHostSession{..} =+ rnf peerHostSessionScore `seq`+ rnf peerHostSessionReconnect `seq`+ rnf peerHostSessionLog++{- Shared Peer STM Type -}++data SharedNodeState = SharedNodeState+ { sharedPeerMap :: !(TVar (M.Map PeerId (TVar PeerSession)))+ -- ^ Map of all active peers and their sessions+ , sharedHostMap :: !(TVar (M.Map PeerHost (TVar PeerHostSession)))+ -- ^ The peer that is currently syncing the block headers+ , sharedNetworkHeight :: !(TVar BlockHeight)+ -- ^ The current height of the network+ , sharedHeaders :: !(TMVar (PeerId, Headers))+ -- ^ Block headers sent from a peer+ , sharedHeaderPeer :: !(TVar (Maybe PeerId))+ -- ^ Peer currently syncing headers+ , sharedMerklePeer :: !(TVar (Maybe PeerId))+ -- ^ Peer currently downloading merkle blocks+ , sharedSyncLock :: !Lock+ -- ^ Lock on the header syncing process+ , sharedLevelDBLock :: !Lock+ -- ^ LevelDB lock+ , sharedBestHeader :: !(TVar BlockHeaderNode)+ -- ^ Our best block header+ , sharedBestBlock :: !(TVar BlockHash)+ -- ^ Our best merkle block+ , sharedTxGetData :: !(TVar (M.Map TxHash [(PeerId, PeerHost)]))+ -- ^ List of Tx GetData requests+ , sharedBloomFilter :: !(TVar (Maybe (BloomFilter, Int)))+ -- ^ Bloom filter+ , sharedTickleChan :: !(TBMChan (PeerId, PeerHost, BlockHash))+ -- ^ Channel containing all the block tickles received from peers+ , sharedTxChan :: !(TBMChan (PeerId, PeerHost, Tx))+ -- ^ Transaction channel+ , sharedRescan :: !(TMVar (Either Timestamp BlockHeight))+ -- ^ Rescan requests from a timestamp or from a block height+ , sharedMempool :: !(TVar Bool)+ -- ^ Did we do a Mempool sync ?+ , levelDBFilePath :: !FilePath+ -- ^ LevelDB FilePath+ , levelDBOptions :: !L.Options+ -- ^ LevelDB Options+ }++{- Peer Data -}++type PingNonce = Word64++-- Data stored about a peer+data PeerSession = PeerSession+ { peerSessionConnected :: !Bool+ -- ^ True if the peer is connected (completed the handshake)+ , peerSessionVersion :: !(Maybe Version)+ -- ^ Contains the version message that we received from the peer+ , peerSessionHeight :: !BlockHeight+ -- ^ Current known height of the peer+ , peerSessionChan :: !(TBMChan Message)+ -- ^ Message channel to send messages to the peer+ , peerSessionHost :: !PeerHost+ -- ^ Host to which this peer is connected+ , peerSessionThreadId :: !ThreadId+ -- ^ Peer ThreadId+ , peerSessionMerkleChan :: !(TBMChan (Either (MerkleBlock, MerkleTxs) Tx))+ -- ^ Merkle block/Merkle transaction channel+ , peerSessionPings :: !(TVar [PingNonce])+ -- ^ Time at which we requested pings+ , peerSessionScore :: !(Maybe NominalDiffTime)+ -- ^ Ping scores for this peer (round trip times)+ }++instance NFData PeerSession where+ rnf PeerSession{..} =+ rnf peerSessionConnected `seq`+ rnf peerSessionVersion `seq`+ rnf peerSessionHeight `seq`+ peerSessionChan `seq`+ rnf peerSessionHost `seq`+ peerSessionThreadId `seq` ()++{- Peer Hosts -}++data PeerHost = PeerHost+ { peerHost :: !String+ , peerPort :: !Int+ }+ deriving (Eq, Ord)++$(deriveJSON (dropFieldLabel 4) ''PeerHost)++peerHostString :: PeerHost -> String+peerHostString PeerHost{..} = concat [ peerHost, ":", show peerPort ]++instance NFData PeerHost where+ rnf PeerHost{..} =+ rnf peerHost `seq`+ rnf peerPort++{- Node Status -}++data PeerStatus = PeerStatus+ -- Regular fields+ { peerStatusPeerId :: !Int+ , peerStatusHost :: !PeerHost+ , peerStatusConnected :: !Bool+ , peerStatusHeight :: !BlockHeight+ , peerStatusProtocol :: !(Maybe Word32)+ , peerStatusUserAgent :: !(Maybe String)+ , peerStatusPing :: !(Maybe String)+ , peerStatusDoSScore :: !(Maybe PeerHostScore)+ -- Debug fields+ , peerStatusThreadId :: !String+ , peerStatusHaveMerkles :: !Bool+ , peerStatusHaveMessage :: !Bool+ , peerStatusPingNonces :: ![PingNonce]+ , peerStatusReconnectTimer :: !(Maybe Int)+ , peerStatusLog :: !(Maybe [String])+ }++$(deriveJSON (dropFieldLabel 10) ''PeerStatus)++data NodeStatus = NodeStatus+ -- Regular fields+ { nodeStatusPeers :: ![PeerStatus]+ , nodeStatusNetworkHeight :: !BlockHeight+ , nodeStatusBestHeader :: !BlockHash+ , nodeStatusBestHeaderHeight :: !BlockHeight+ , nodeStatusBestBlock :: !BlockHash+ , nodeStatusBloomSize :: !Int+ -- Debug fields+ , nodeStatusHeaderPeer :: !(Maybe Int)+ , nodeStatusMerklePeer :: !(Maybe Int)+ , nodeStatusHaveHeaders :: !Bool+ , nodeStatusHaveTickles :: !Bool+ , nodeStatusHaveTxs :: !Bool+ , nodeStatusGetData :: ![TxHash]+ , nodeStatusRescan :: !(Maybe (Either Timestamp BlockHeight))+ , nodeStatusMempool :: !Bool+ , nodeStatusSyncLock :: !Bool+ , nodeStatusLevelDBLock :: !Bool+ }++$(deriveJSON (dropFieldLabel 10) ''NodeStatus)++{- Getters / Setters -}++tryGetPeerSession :: PeerId -> NodeT STM (Maybe PeerSession)+tryGetPeerSession pid = do+ peerMap <- readTVarS sharedPeerMap+ case M.lookup pid peerMap of+ Just sessTVar -> liftM Just $ lift $ readTVar sessTVar+ _ -> return Nothing++getPeerSession :: PeerId -> NodeT STM PeerSession+getPeerSession pid = do+ sessM <- tryGetPeerSession pid+ case sessM of+ Just sess -> return sess+ _ -> throw $ NodeExceptionInvalidPeer pid++newPeerSession :: PeerId -> PeerSession -> NodeT STM ()+newPeerSession pid sess = do+ peerMapTVar <- asks sharedPeerMap+ peerMap <- lift $ readTVar peerMapTVar+ case M.lookup pid peerMap of+ Just _ -> return ()+ Nothing -> do+ sessTVar <- lift $ newTVar sess+ let newMap = M.insert pid sessTVar peerMap+ lift $ writeTVar peerMapTVar $! newMap++modifyPeerSession :: PeerId -> (PeerSession -> PeerSession) -> NodeT STM ()+modifyPeerSession pid f = do+ peerMap <- readTVarS sharedPeerMap+ case M.lookup pid peerMap of+ Just sessTVar -> lift $ modifyTVar' sessTVar f+ _ -> return ()++removePeerSession :: PeerId -> NodeT STM (Maybe PeerSession)+removePeerSession pid = do+ peerMapTVar <- asks sharedPeerMap+ peerMap <- lift $ readTVar peerMapTVar+ -- Close the peer TBMChan+ sessM <- case M.lookup pid peerMap of+ Just sessTVar -> lift $ do+ sess@PeerSession{..} <- readTVar sessTVar+ closeTBMChan peerSessionChan+ return $ Just sess+ _ -> return Nothing+ -- Remove the peer from the peerMap+ let newMap = M.delete pid peerMap+ lift $ writeTVar peerMapTVar $! newMap+ return sessM++getHostSession :: PeerHost+ -> NodeT STM (Maybe PeerHostSession)+getHostSession ph = do+ hostMap <- readTVarS sharedHostMap+ lift $ case M.lookup ph hostMap of+ Just hostSessionTVar -> liftM Just $ readTVar hostSessionTVar+ _ -> return Nothing++modifyHostSession :: PeerHost+ -> (PeerHostSession -> PeerHostSession)+ -> NodeT STM ()+modifyHostSession ph f = do+ hostMap <- readTVarS sharedHostMap+ case M.lookup ph hostMap of+ Just hostSessionTVar -> lift $ modifyTVar' hostSessionTVar f+ _ -> newHostSession ph $!+ f PeerHostSession { peerHostSessionScore = 0+ , peerHostSessionReconnect = 1+ , peerHostSessionLog = []+ }++newHostSession :: PeerHost -> PeerHostSession -> NodeT STM ()+newHostSession ph session = do+ hostMapTVar <- asks sharedHostMap+ hostMap <- lift $ readTVar hostMapTVar+ case M.lookup ph hostMap of+ Just _ -> return ()+ Nothing -> lift $ do+ hostSessionTVar <- newTVar session+ let newHostMap = M.insert ph hostSessionTVar hostMap+ writeTVar hostMapTVar $! newHostMap++{- Host DOS Scores -}++bannedScore :: PeerHostScore+bannedScore = 100++minorDoS :: PeerHostScore -> PeerHostScore+minorDoS = (+ 1)++moderateDoS :: PeerHostScore -> PeerHostScore+moderateDoS = (+ 10)++severeDoS :: PeerHostScore -> PeerHostScore+severeDoS = (+ bannedScore)++isHostScoreBanned :: PeerHostScore -> Bool+isHostScoreBanned = (>= bannedScore)++{- STM Utilities -}++orElseNodeT :: NodeT STM a -> NodeT STM a -> NodeT STM a+orElseNodeT a b = do+ s <- ask+ lift $ (runNodeT s a) `orElse` (runNodeT s b)++{- TVar Utilities -}++readTVarS :: (SharedNodeState -> TVar a) -> NodeT STM a+readTVarS = lift . readTVar <=< asks++writeTVarS :: (SharedNodeState -> TVar a) -> a -> NodeT STM ()+writeTVarS f val = lift . (flip writeTVar val) =<< asks f++{- TMVar Utilities -}++takeTMVarS :: (SharedNodeState -> TMVar a) -> NodeT STM a+takeTMVarS = lift . takeTMVar <=< asks++readTMVarS :: (SharedNodeState -> TMVar a) -> NodeT STM a+readTMVarS = lift . readTMVar <=< asks++tryReadTMVarS :: (SharedNodeState -> TMVar a) -> NodeT STM (Maybe a)+tryReadTMVarS = lift . tryReadTMVar <=< asks++putTMVarS :: (SharedNodeState -> TMVar a) -> a -> NodeT STM ()+putTMVarS f val = lift . (flip putTMVar val) =<< asks f++tryPutTMVarS :: (SharedNodeState -> TMVar a) -> a -> NodeT STM Bool+tryPutTMVarS f val = lift . (flip tryPutTMVar val) =<< asks f++swapTMVarS :: (SharedNodeState -> TMVar a) -> a -> NodeT STM ()+swapTMVarS f val = lift . (flip putTMVar val) =<< asks f++isEmptyTMVarS :: (SharedNodeState -> TMVar a) -> NodeT STM Bool+isEmptyTMVarS f = lift . isEmptyTMVar =<< asks f++data NodeException+ = NodeExceptionBanned+ | NodeExceptionConnected+ | NodeExceptionInvalidPeer !PeerId+ | NodeExceptionPeerNotConnected !PeerId+ | NodeException !String+ deriving (Show, Typeable)++instance Exception NodeException++isNodeException :: SomeException -> Bool+isNodeException se = isJust (fromException se :: Maybe NodeException)++catchAny :: MonadBaseControl IO m+ => m a -> (SomeException -> m a) -> m a+catchAny = catch++catchAny_ :: MonadBaseControl IO m+ => m () -> m ()+catchAny_ = flip catchAny $ \_ -> return ()+
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ UNLICENSE view
@@ -0,0 +1,24 @@+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 <http://unlicense.org/>
+ haskoin-node.cabal view
@@ -0,0 +1,99 @@+name: haskoin-node+version: 0.2.0+synopsis:+ Implementation of a Bitoin node.+description:+ haskoin-node provides an implementation of the Bitcoin network protocol+ that allows you to synchronize headers (with SPV validation) and download+ merkle blocks and full blocks. This package can be used to implement+ wallets or other Bitcoin components that require talking to the Bitcoin+ network. It provides the following features:+ .+ * Implementation of the Bitcoin network protocol+ * Headertree implementation with SPV verification+ * Headers-first synchronization+ * Merkle block download from peers with bloom filters+ * Full block download from peers+ .+ A wallet implementation using this package is available in haskoin-wallet.++homepage: http://github.com/haskoin/haskoin+bug-reports: http://github.com/haskoin/haskoin/issues+stability: stable+license: PublicDomain+license-file: UNLICENSE+author: Philippe Laprade, Jean-Pierre Rupp+maintainer: plaprade+hackage@gmail.com+category: Bitcoin, Finance, Network+build-type: Simple+cabal-version: >= 1.9.2+extra-source-files: stack.yaml++source-repository head+ type: git+ location: git://github.com/haskoin/haskoin.git++library+ exposed-modules: Network.Haskoin.Node.HeaderTree+ Network.Haskoin.Node.Checkpoints+ Network.Haskoin.Node.Peer+ Network.Haskoin.Node.BlockChain+ Network.Haskoin.Node.STM++ extensions: OverloadedStrings+ FlexibleInstances+ FlexibleContexts+ RecordWildCards+ DeriveDataTypeable++ build-depends: aeson >= 0.7 && < 0.9+ , async >= 2.0 && < 2.1+ , base >= 4.8 && < 5+ , binary >= 0.7 && < 0.8+ , bytestring >= 0.10 && < 0.11+ , concurrent-extra >= 0.7 && < 0.8+ , conduit >= 1.2 && < 1.3+ , conduit-extra >= 1.1 && < 1.2+ , containers >= 0.5 && < 0.6+ , data-default >= 0.5 && < 0.6+ , deepseq >= 1.4 && < 1.5+ , either >= 4.3 && < 4.5+ , exceptions >= 0.8 && < 0.9+ , haskoin-core >= 0.2 && < 0.3+ , leveldb-haskell >= 0.6 && < 0.7+ , lifted-async >= 0.2 && < 0.8+ , lifted-base >= 0.2 && < 0.3+ , monad-control >= 1.0 && < 1.1+ , monad-logger >= 0.3 && < 0.4+ , mtl >= 2.2 && < 2.3+ , network >= 2.4 && < 2.7+ , random >= 1.0 && < 1.2+ , stm >= 2.4 && < 2.5+ , stm-chans >= 3.0 && < 3.1+ , stm-conduit >= 2.5 && < 2.7+ , string-conversions >= 0.4 && < 0.5+ , text >= 0.11 && < 1.3+ , time >= 1.4 && < 1.6++ ghc-options: -Wall++test-suite test-haskoin-node+ type: exitcode-stdio-1.0+ main-is: Main.hs++ extensions: EmptyDataDecls++ other-modules: Network.Haskoin.Node.Tests+ Network.Haskoin.Node.Units++ build-depends: base >= 4.8 && < 5+ , haskoin-node+ , HUnit >= 1.2 && < 1.3+ , QuickCheck >= 2.6 && < 2.9+ , test-framework >= 0.8 && < 0.9+ , test-framework-quickcheck2 >= 0.3 && < 0.4+ , test-framework-hunit >= 0.3 && < 0.4++ ghc-options: -Wall+ hs-source-dirs: tests+
+ stack.yaml view
@@ -0,0 +1,14 @@+flags: {}+packages:+- '.'+- '../haskoin-core'+- location:+ git: https://github.com/haskoin/secp256k1.git+ commit: 5ee603061b3c1eaf2943e8d2c08e6effe85f38e7+ extra-dep: true+extra-deps:+- leveldb-haskell-0.6.3+- murmur3-1.0.0+- pbkdf-1.1.1.1+- largeword-1.2.4+resolver: lts-3.4
+ tests/Main.hs view
@@ -0,0 +1,13 @@+module Main where++import Test.Framework (defaultMain)++import qualified Network.Haskoin.Node.Tests (tests)+import qualified Network.Haskoin.Node.Units (tests)++main :: IO ()+main = defaultMain+ ( Network.Haskoin.Node.Tests.tests+ ++ Network.Haskoin.Node.Units.tests+ )+
+ tests/Network/Haskoin/Node/Tests.hs view
@@ -0,0 +1,12 @@+module Network.Haskoin.Node.Tests (tests) where++import Test.Framework (Test, testGroup)+-- import Test.Framework.Providers.QuickCheck2 (testProperty)++tests :: [Test]+tests =+ [ testGroup "Serialize & de-serialize haskoin node types to JSON"+ [+ ]+ ]+
+ tests/Network/Haskoin/Node/Units.hs view
@@ -0,0 +1,17 @@+module Network.Haskoin.Node.Units (tests) where++-- import Test.HUnit (Assertion, assertBool)+import Test.Framework (Test, testGroup)+-- import Test.Framework.Providers.HUnit (testCase)++-- TODO: Make sure that evalNewChain for a partially overlapping best chain+-- properly evaluates to BestChain.++tests :: [Test]+tests =+ [ testGroup "Test Group"+ [+ ]+ ]++