diff --git a/app/Main.hs b/app/Main.hs --- a/app/Main.hs +++ b/app/Main.hs @@ -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 diff --git a/haskoin-store.cabal b/haskoin-store.cabal --- a/haskoin-store.cabal +++ b/haskoin-store.cabal @@ -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 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 diff --git a/src/Haskoin/Store/BlockStore.hs b/src/Haskoin/Store/BlockStore.hs --- a/src/Haskoin/Store/BlockStore.hs +++ b/src/Haskoin/Store/BlockStore.hs @@ -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 diff --git a/src/Haskoin/Store/Manager.hs b/src/Haskoin/Store/Manager.hs --- a/src/Haskoin/Store/Manager.hs +++ b/src/Haskoin/Store/Manager.hs @@ -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 diff --git a/src/Haskoin/Store/Web.hs b/src/Haskoin/Store/Web.hs --- a/src/Haskoin/Store/Web.hs +++ b/src/Haskoin/Store/Web.hs @@ -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 ()