haskoin-store 0.34.0 → 0.34.1
raw patch · 5 files changed
+132/−102 lines, 5 filesdep ~haskoin-storedep ~haskoin-store-data
Dependency ranges changed: haskoin-store, haskoin-store-data
Files
- app/Main.hs +17/−1
- haskoin-store.cabal +6/−6
- src/Haskoin/Store/BlockStore.hs +80/−61
- src/Haskoin/Store/Manager.hs +5/−7
- src/Haskoin/Store/Web.hs +24/−27
app/Main.hs view
@@ -54,6 +54,7 @@ , configVersion :: !Bool , configDebug :: !Bool , configReqLog :: !Bool+ , configMaxPending :: !Int , configWebLimits :: !WebLimits , configWebTimeouts :: !WebTimeouts , configRedis :: !Bool@@ -76,6 +77,7 @@ , configVersion = False , configDebug = defDebug , configReqLog = defReqLog+ , configMaxPending = defMaxPending , configWebLimits = defWebLimits , configWebTimeouts = defWebTimeouts , configRedis = defRedis@@ -93,6 +95,11 @@ ms <- lookupEnv e return $ fromMaybe d $ p =<< ms +defMaxPending :: Int+defMaxPending = unsafePerformIO $+ defEnv "MAX_PENDING_TXS" 100 readMaybe+{-# NOINLINE defMaxPending #-}+ defMaxPeers :: Int defMaxPeers = unsafePerformIO $ defEnv "MAX_PEERS" 20 readMaybe@@ -348,6 +355,13 @@ <> help "Disconnect peers older than this" <> showDefault <> value (configPeerMaxLife def)+ configMaxPending <-+ option auto $+ metavar "INT"+ <> long "max-pending-txs"+ <> help "Maximum pending txs to fail health check"+ <> showDefault+ <> value (configMaxPending def) configRedis <- flag (configRedis def) True $ long "cache"@@ -432,6 +446,7 @@ , configPeers = peers , configDir = db_dir , configDebug = deb+ , configMaxPending = pend , configWebLimits = limits , configReqLog = reqlog , configWebTimeouts = tos@@ -477,7 +492,8 @@ , webStore = st , webMaxLimits = limits , webReqLog = reqlog- , webWebTimeouts = tos+ , webTimeouts = tos+ , webMaxPending = pend , webVersion = version } where
haskoin-store.cabal view
@@ -4,10 +4,10 @@ -- -- see: https://github.com/sol/hpack ----- hash: 4d387f3124649772fe3bf688cdd281364146d3f1f944ff4e02e433cd147379f1+-- hash: 7194d483a3a97de5193b1e79e587534fa3b778342599612d287289ce31d79799 name: haskoin-store-version: 0.34.0+version: 0.34.1 synopsis: Storage and index for Bitcoin and Bitcoin Cash description: Please see the README on GitHub at <https://github.com/haskoin/haskoin-store#readme> category: Bitcoin, Finance, Network@@ -53,7 +53,7 @@ , hashable >=1.3.0.0 , haskoin-core >=0.13.6 , haskoin-node >=0.14.1- , haskoin-store-data ==0.34.0+ , haskoin-store-data ==0.34.1 , hedis >=0.12.13 , http-types >=0.12.3 , monad-logger >=0.3.32@@ -96,7 +96,7 @@ , haskoin-core >=0.13.6 , haskoin-node >=0.14.1 , haskoin-store- , haskoin-store-data ==0.34.0+ , haskoin-store-data ==0.34.1 , monad-logger >=0.3.32 , mtl >=2.2.2 , nqe >=0.6.1@@ -135,8 +135,8 @@ , hashable >=1.3.0.0 , haskoin-core >=0.13.6 , haskoin-node >=0.14.1- , haskoin-store ==0.34.0- , haskoin-store-data ==0.34.0+ , haskoin-store ==0.34.1+ , haskoin-store-data ==0.34.1 , hedis >=0.12.13 , hspec >=2.7.1 , http-types >=0.12.3
src/Haskoin/Store/BlockStore.hs view
@@ -3,15 +3,14 @@ {-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE LambdaCase #-} {-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-} {-# LANGUAGE TemplateHaskell #-} {-# LANGUAGE TupleSections #-} module Haskoin.Store.BlockStore ( -- * Block Store BlockStore- , BlockStoreMessage- , BlockStoreInbox , BlockStoreConfig(..)- , blockStore+ , withBlockStore , blockStorePeerConnect , blockStorePeerConnectSTM , blockStorePeerDisconnect@@ -26,6 +25,8 @@ , blockStoreTxSTM , blockStoreTxHash , blockStoreTxHashSTM+ , blockStorePendingTxs+ , blockStorePendingTxsSTM ) where import Control.Monad (forM, forM_, forever, mzero,@@ -82,16 +83,16 @@ getOldMempool, importBlock, initBest, newMempoolTx, revertBlock)-import NQE (Inbox, Listen, Mailbox,- Publisher, inboxToMailbox,- publish, query, receive, send,- sendSTM)+import NQE (Listen, Mailbox, Publisher,+ newMailbox, publish, query,+ receive, send, sendSTM) import System.Random (randomRIO) import UnliftIO (Exception, MonadIO, MonadUnliftIO, STM, TVar, async,- atomically, liftIO, modifyTVar,- newTVarIO, readTVar, readTVarIO,- throwIO, withAsync, writeTVar)+ atomically, liftIO, link,+ modifyTVar, newTVarIO, readTVar,+ readTVarIO, throwIO, withAsync,+ writeTVar) import UnliftIO.Concurrent (threadDelay) data BlockStoreMessage@@ -104,9 +105,6 @@ | TxRefAvailable !Peer ![TxHash] | BlockPing !(Listen ()) -type BlockStoreInbox = Inbox BlockStoreMessage-type BlockStore = Mailbox BlockStoreMessage- data BlockException = BlockNotInChain !BlockHash | Uninitialized@@ -131,9 +129,9 @@ deriving (Show, Eq, Ord) -- | Block store process state.-data BlockRead =- BlockRead- { mySelf :: !BlockStore+data BlockStore =+ BlockStore+ { myMailbox :: !(Mailbox BlockStoreMessage) , myConfig :: !BlockStoreConfig , myPeer :: !(TVar (Maybe Syncing)) , myTxs :: !(TVar (HashMap TxHash PendingTx))@@ -159,7 +157,7 @@ -- ^ disconnect syncing peer if inactive for this long } -type BlockT m = ReaderT BlockRead m+type BlockT m = ReaderT BlockStore m runImport :: MonadLoggerIO m => WriterT (ExceptT ImportException m) a@@ -206,24 +204,30 @@ runRocksDB . getAddressTxs a -- | Run block store process.-blockStore ::+withBlockStore :: (MonadUnliftIO m, MonadLoggerIO m) => BlockStoreConfig- -> BlockStoreInbox- -> m ()-blockStore cfg inbox = do+ -> (BlockStore -> m a)+ -> m a+withBlockStore cfg action = do pb <- newTVarIO Nothing ts <- newTVarIO HashMap.empty rq <- newTVarIO HashSet.empty- runReaderT- (ini >> wipe >> run)- BlockRead { mySelf = inboxToMailbox inbox- , myConfig = cfg- , myPeer = pb- , myTxs = ts- , requested = rq- }+ (inbox, mbox) <- newMailbox+ let r = BlockStore { myMailbox = mbox+ , myConfig = cfg+ , myPeer = pb+ , myTxs = ts+ , requested = rq+ }+ withAsync (runReaderT (go inbox mbox) r) $ \a -> do+ link a+ action r where+ go inbox mbox = do+ ini+ wipe+ run inbox mbox del txs = do $(logInfoS) "BlockStore" $ "Deleting " <> cs (show (length txs)) <> " transactions"@@ -249,7 +253,7 @@ "Could not initialize: " <> cs (show e) throwIO e Right () -> return ()- run = withAsync (pingMe (inboxToMailbox inbox))+ run inbox mbox = withAsync (pingMe mbox) $ const $ forever $ receive inbox >>=@@ -324,7 +328,7 @@ listener <- asks (blockConfListener . myConfig) publish (StoreBestBlock blockhash) listener -setSyncingBlocks :: (MonadReader BlockRead m, MonadIO m)+setSyncingBlocks :: (MonadReader BlockStore m, MonadIO m) => [BlockHash] -> m () setSyncingBlocks hs = asks myPeer >>= \box ->@@ -332,13 +336,13 @@ Nothing -> Nothing Just x -> Just x { syncingBlocks = hs } -getSyncingBlocks :: (MonadReader BlockRead m, MonadIO m) => m [BlockHash]+getSyncingBlocks :: (MonadReader BlockStore m, MonadIO m) => m [BlockHash] getSyncingBlocks = asks myPeer >>= readTVarIO >>= \case Nothing -> return [] Just x -> return $ syncingBlocks x -addSyncingBlocks :: (MonadReader BlockRead m, MonadIO m)+addSyncingBlocks :: (MonadReader BlockStore m, MonadIO m) => [BlockHash] -> m () addSyncingBlocks hs = asks myPeer >>= \box ->@@ -346,7 +350,7 @@ Nothing -> Nothing Just x -> Just x { syncingBlocks = syncingBlocks x <> hs } -removeSyncingBlock :: (MonadReader BlockRead m, MonadIO m)+removeSyncingBlock :: (MonadReader BlockStore m, MonadIO m) => BlockHash -> m () removeSyncingBlock h = do box <- asks myPeer@@ -354,13 +358,13 @@ Nothing -> Nothing Just x -> Just x { syncingBlocks = delete h (syncingBlocks x) } -checkPeer :: (MonadLoggerIO m, MonadReader BlockRead m) => Peer -> m Bool+checkPeer :: (MonadLoggerIO m, MonadReader BlockStore m) => Peer -> m Bool checkPeer p = fmap syncingPeer <$> getSyncingState >>= \case Nothing -> return False Just p' -> return $ p == p' -getBlockNode :: (MonadLoggerIO m, MonadReader BlockRead m)+getBlockNode :: (MonadLoggerIO m, MonadReader BlockStore m) => BlockHash -> m (Maybe BlockNode) getBlockNode blockhash = chainGetBlock blockhash =<< asks (blockConfChain . myConfig)@@ -399,7 +403,10 @@ addPendingTx :: MonadIO m => PendingTx -> BlockT m () addPendingTx p = do ts <- asks myTxs- atomically $ modifyTVar ts $ HashMap.insert th p+ rq <- asks requested+ atomically $ do+ modifyTVar ts $ HashMap.insert th p+ modifyTVar rq $ HashSet.delete th where th = txHash (pendingTx p) @@ -439,7 +446,7 @@ txids = map (txHash . snd) sorted in mapMaybe (`HashMap.lookup` m) txids -fulfillOrphans :: MonadIO m => BlockRead -> TxHash -> m ()+fulfillOrphans :: MonadIO m => BlockStore -> TxHash -> m () fulfillOrphans block_read th = atomically $ modifyTVar box (HashMap.map fulfill) where@@ -449,7 +456,7 @@ updateOrphans :: ( StoreReadBase m , MonadLoggerIO m- , MonadReader BlockRead m+ , MonadReader BlockStore m ) => m () updateOrphans = do@@ -479,7 +486,7 @@ return $ foldl fulfill p unspents newOrphanTx :: MonadLoggerIO m- => BlockRead+ => BlockStore -> UTCTime -> Tx -> WriterT m ()@@ -506,7 +513,7 @@ importMempoolTx :: (MonadLoggerIO m, MonadError ImportException m)- => BlockRead+ => BlockStore -> UTCTime -> Tx -> WriterT m Bool@@ -608,7 +615,7 @@ msg `sendMessage` p touchPeer :: ( MonadIO m- , MonadReader BlockRead m+ , MonadReader BlockStore m ) => m () touchPeer =@@ -761,7 +768,7 @@ then return bh else findAncestor sh bh -findAncestor :: (MonadLoggerIO m, MonadReader BlockRead m)+findAncestor :: (MonadLoggerIO m, MonadReader BlockStore m) => BlockHeight -> BlockNode -> m BlockNode findAncestor height target = do ch <- asks (blockConfChain . myConfig)@@ -775,7 +782,7 @@ <> cs (show (nodeHeight target)) throwIO $ AncestorNotInChain height h -finishPeer :: (MonadLoggerIO m, MonadReader BlockRead m)+finishPeer :: (MonadLoggerIO m, MonadReader BlockStore m) => Peer -> m () finishPeer p = do box <- asks myPeer@@ -836,12 +843,12 @@ True -> syncMe getSyncingState- :: (MonadIO m, MonadReader BlockRead m) => m (Maybe Syncing)+ :: (MonadIO m, MonadReader BlockStore m) => m (Maybe Syncing) getSyncingState = readTVarIO =<< asks myPeer clearSyncingState- :: (MonadLoggerIO m, MonadReader BlockRead m) => m ()+ :: (MonadLoggerIO m, MonadReader BlockStore m) => m () clearSyncingState = asks myPeer >>= readTVarIO >>= \case Nothing -> return ()@@ -879,7 +886,7 @@ pruneMempool atomically (r ()) -pingMe :: MonadLoggerIO m => BlockStore -> m ()+pingMe :: MonadLoggerIO m => Mailbox BlockStoreMessage -> m () pingMe mbox = forever $ do delay <- liftIO $@@ -890,72 +897,84 @@ blockStorePeerConnect :: MonadIO m => Peer -> BlockStore -> m () blockStorePeerConnect peer store =- BlockPeerConnect peer `send` store+ BlockPeerConnect peer `send` myMailbox store blockStorePeerDisconnect :: MonadIO m => Peer -> BlockStore -> m () blockStorePeerDisconnect peer store =- BlockPeerDisconnect peer `send` store+ BlockPeerDisconnect peer `send` myMailbox store blockStoreHead :: MonadIO m => BlockNode -> BlockStore -> m () blockStoreHead node store =- BlockNewBest node `send` store+ BlockNewBest node `send` myMailbox store blockStoreBlock :: MonadIO m => Peer -> Block -> BlockStore -> m () blockStoreBlock peer block store =- BlockReceived peer block `send` store+ BlockReceived peer block `send` myMailbox store blockStoreNotFound :: MonadIO m => Peer -> [BlockHash] -> BlockStore -> m () blockStoreNotFound peer blocks store =- BlockNotFound peer blocks `send` store+ BlockNotFound peer blocks `send` myMailbox store blockStoreTx :: MonadIO m => Peer -> Tx -> BlockStore -> m () blockStoreTx peer tx store =- TxRefReceived peer tx `send` store+ TxRefReceived peer tx `send` myMailbox store blockStoreTxHash :: MonadIO m => Peer -> [TxHash] -> BlockStore -> m () blockStoreTxHash peer txhashes store =- TxRefAvailable peer txhashes `send` store+ TxRefAvailable peer txhashes `send` myMailbox store blockStorePeerConnectSTM :: Peer -> BlockStore -> STM () blockStorePeerConnectSTM peer store =- BlockPeerConnect peer `sendSTM` store+ BlockPeerConnect peer `sendSTM` myMailbox store blockStorePeerDisconnectSTM :: Peer -> BlockStore -> STM () blockStorePeerDisconnectSTM peer store =- BlockPeerDisconnect peer `sendSTM` store+ BlockPeerDisconnect peer `sendSTM` myMailbox store blockStoreHeadSTM :: BlockNode -> BlockStore -> STM () blockStoreHeadSTM node store =- BlockNewBest node `sendSTM` store+ BlockNewBest node `sendSTM` myMailbox store blockStoreBlockSTM :: Peer -> Block -> BlockStore -> STM () blockStoreBlockSTM peer block store =- BlockReceived peer block `sendSTM` store+ BlockReceived peer block `sendSTM` myMailbox store blockStoreNotFoundSTM :: Peer -> [BlockHash] -> BlockStore -> STM () blockStoreNotFoundSTM peer blocks store =- BlockNotFound peer blocks `sendSTM` store+ BlockNotFound peer blocks `sendSTM` myMailbox store blockStoreTxSTM :: Peer -> Tx -> BlockStore -> STM () blockStoreTxSTM peer tx store =- TxRefReceived peer tx `sendSTM` store+ TxRefReceived peer tx `sendSTM` myMailbox store blockStoreTxHashSTM :: Peer -> [TxHash] -> BlockStore -> STM () blockStoreTxHashSTM peer txhashes store =- TxRefAvailable peer txhashes `sendSTM` store+ TxRefAvailable peer txhashes `sendSTM` myMailbox store++blockStorePendingTxs+ :: MonadIO m => BlockStore -> m Int+blockStorePendingTxs =+ atomically . blockStorePendingTxsSTM++blockStorePendingTxsSTM+ :: BlockStore -> STM Int+blockStorePendingTxsSTM BlockStore {..} = do+ x <- HashMap.keysSet <$> readTVar myTxs+ y <- readTVar requested+ return $ HashSet.size $ x `HashSet.union` y blockText :: BlockNode -> Maybe Block -> Text blockText bn mblock = case mblock of
src/Haskoin/Store/Manager.hs view
@@ -26,13 +26,13 @@ WithConnection, withNode) import Haskoin.Store.BlockStore (BlockStore, BlockStoreConfig (..),- blockStore, blockStoreBlockSTM,+ blockStoreBlockSTM, blockStoreHeadSTM, blockStoreNotFoundSTM, blockStorePeerConnectSTM, blockStorePeerDisconnectSTM, blockStoreTxHashSTM,- blockStoreTxSTM)+ blockStoreTxSTM, withBlockStore) import Haskoin.Store.Cache (CacheConfig (..), CacheWriter, cacheNewBlock, cacheNewTx, cachePing, cacheWriter,@@ -43,7 +43,7 @@ withDatabaseReader) import Network.Socket (SockAddr (..)) import NQE (Inbox, Process (..), Publisher,- newMailbox, publishSTM, receive,+ publishSTM, receive, withProcess, withPublisher, withSubscription) import System.Random (randomRIO)@@ -101,15 +101,13 @@ => StoreConfig -> (Store -> m a) -> m a withStore cfg action = connectDB cfg >>= \db ->- newMailbox >>= \(bi, b) -> withPublisher $ \pub -> withPublisher $ \node_pub -> withSubscription node_pub $ \node_sub -> withNode (nodeCfg cfg db node_pub) $ \node -> withCache cfg (nodeChain node) db pub $ \mcache ->- withAsync (nodeForwarder b pub node_sub) $ \a1 ->- withAsync (blockStore (blockStoreCfg cfg node pub db) bi) $ \a2 ->- link a1 >> link a2 >>+ withBlockStore (blockStoreCfg cfg node pub db) $ \b ->+ withAsync (nodeForwarder b pub node_sub) $ \a1 -> link a1 >> action Store { storeManager = nodeManager node , storeChain = nodeChain node , storeBlock = b
src/Haskoin/Store/Web.hs view
@@ -19,7 +19,7 @@ import Control.Applicative ((<|>)) import Control.Monad (forever, unless, when, (<=<)) import Control.Monad.Logger (MonadLoggerIO, logInfoS)-import Control.Monad.Reader (ReaderT, asks, local,+import Control.Monad.Reader (ReaderT, ask, asks, local, runReaderT) import Control.Monad.Trans (lift) import Control.Monad.Trans.Maybe (MaybeT (..), runMaybeT)@@ -53,14 +53,12 @@ import Haskoin.Node (Chain, OnlinePeer (..), PeerManager, chainGetBest, getPeers, sendMessage)-import Haskoin.Store.Cache (CacheT, evictFromCache,- withCache)+import Haskoin.Store.BlockStore+import Haskoin.Store.Cache import Haskoin.Store.Common import Haskoin.Store.Data-import Haskoin.Store.Database.Reader (DatabaseReader (..),- DatabaseReaderT,- withDatabaseReader)-import Haskoin.Store.Manager (Store (..))+import Haskoin.Store.Database.Reader+import Haskoin.Store.Manager import Haskoin.Store.WebCommon import Haskoin.Transaction import Haskoin.Util@@ -106,9 +104,10 @@ { webHost :: !String , webPort :: !Int , webStore :: !Store+ , webMaxPending :: !Int , webMaxLimits :: !WebLimits , webReqLog :: !Bool- , webWebTimeouts :: !WebTimeouts+ , webTimeouts :: !WebTimeouts , webVersion :: !String } @@ -869,12 +868,7 @@ scottyHealth :: (MonadUnliftIO m, MonadLoggerIO m) => GetHealth -> WebT m HealthCheck scottyHealth _ = do- net <- lift $ asks (storeNetwork . webStore)- mgr <- lift $ asks (storeManager . webStore)- chn <- lift $ asks (storeChain . webStore)- tos <- lift $ asks webWebTimeouts- ver <- lift $ asks webVersion- h <- lift $ healthCheck net mgr chn tos ver+ h <- lift $ ask >>= healthCheck unless (isOK h) $ S.status status503 return h @@ -911,6 +905,13 @@ timeHealthMax = fromIntegral $ txTimeout tos return TimeHealth {..} +pendingTxsHealthCheck :: (MonadUnliftIO m, MonadLoggerIO m, StoreReadBase m)+ => WebConfig -> m MaxHealth+pendingTxsHealthCheck cfg = do+ let maxHealthMax = webMaxPending cfg+ maxHealthNum <- blockStorePendingTxs (storeBlock (webStore cfg))+ return MaxHealth {..}+ peerHealthCheck :: (MonadUnliftIO m, MonadLoggerIO m, StoreReadBase m) => PeerManager -> m CountHealth peerHealthCheck mgr = do@@ -919,19 +920,15 @@ return CountHealth {..} healthCheck :: (MonadUnliftIO m, MonadLoggerIO m, StoreReadBase m)- => Network- -> PeerManager- -> Chain- -> WebTimeouts- -> String- -> m HealthCheck-healthCheck net mgr ch tos ver = do- healthBlocks <- blockHealthCheck ch- healthLastBlock <- lastBlockHealthCheck ch tos- healthLastTx <- lastTxHealthCheck ch tos- healthPeers <- peerHealthCheck mgr- let healthNetwork = getNetworkName net- healthVersion = ver+ => WebConfig -> m HealthCheck+healthCheck cfg@WebConfig {..} = do+ healthBlocks <- blockHealthCheck (storeChain webStore)+ healthLastBlock <- lastBlockHealthCheck (storeChain webStore) webTimeouts+ healthLastTx <- lastTxHealthCheck (storeChain webStore) webTimeouts+ healthPendingTxs <- pendingTxsHealthCheck cfg+ healthPeers <- peerHealthCheck (storeManager webStore)+ let healthNetwork = getNetworkName (storeNetwork webStore)+ healthVersion = webVersion return HealthCheck {..} scottyDbStats :: MonadLoggerIO m => WebT m ()