diff --git a/app/Main.hs b/app/Main.hs --- a/app/Main.hs +++ b/app/Main.hs @@ -7,20 +7,13 @@ {-# LANGUAGE OverloadedRecordDot #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE RecordWildCards #-} -{-# LANGUAGE TemplateHaskell #-} {-# LANGUAGE TupleSections #-} {-# LANGUAGE NoFieldSelectors #-} module Main (main) where -import Control.Monad (forever, void, when) +import Control.Monad (when) import Control.Monad.Cont (ContT (ContT), runContT) -import Control.Monad.Logger - ( LogLevel (..), - filterLogger, - logInfoS, - runStderrLoggingT, - ) import Control.Monad.Trans (lift) import Data.Char (toLower) import Data.Default (Default (..)) @@ -34,7 +27,14 @@ netByName, withContext, ) -import Haskoin.Node (withConnection) +import Haskoin.Node + ( LogLevel (..), + logInfo, + withConnection, + withLogger, + withSource, + ) +import Haskoin.Store.Common (loopForever) import Haskoin.Store ( Store (..), StoreConfig (..), @@ -115,6 +115,7 @@ noMempool :: !Bool, wipeMempool :: !Bool, syncMempool :: !Bool, + blockWindow :: !Int, peerTimeout :: !Int, maxPeerLife :: !Int, mempoolTimeout :: !Int, @@ -132,7 +133,10 @@ noXPub :: !Bool, healthCheckInterval :: !Int, bloom :: !Bool, - noWeb :: !Bool + noWeb :: !Bool, + maxOpenFiles :: !Int, + writeBufferMB :: !Int, + maxWalMB :: !Int } env :: (MonadIO m) => String -> a -> (String -> Maybe a) -> m a @@ -180,6 +184,8 @@ env "WIPE_MEMPOOL" False parseBool syncMempool <- env "SYNC_MEMPOOL" False parseBool + blockWindow <- + env "BLOCK_WINDOW" 64 readMaybe mempoolTimeout <- env "MEMPOOL_TIMEOUT" 14 readMaybe peerTimeout <- @@ -216,6 +222,12 @@ env "BLOOM" False parseBool noWeb <- env "NO_WEB" False parseBool + maxOpenFiles <- + env "MAX_OPEN_FILES" 2000 readMaybe + writeBufferMB <- + env "WRITE_BUFFER_MB" 256 readMaybe + maxWalMB <- + env "MAX_WAL_MB" 1024 readMaybe return Config {version = False, ..} where tickerString = @@ -439,6 +451,13 @@ flag c.syncMempool True $ long "sync-mempool" <> help "Attempt to download peer mempools" + blockWindow <- + option auto $ + metavar "COUNT" + <> long "block-window" + <> help "Blocks to request from the syncing peer at a time" + <> showDefault + <> value c.blockWindow mempoolTimeout <- option auto $ metavar "DAYS" @@ -540,6 +559,27 @@ flag c.bloom True $ long "bloom" <> help "RocksDB Bloom filters" + maxOpenFiles <- + option auto $ + metavar "COUNT" + <> long "max-open-files" + <> help "RocksDB max_open_files; bounds index/filter memory (0 = unlimited)" + <> showDefault + <> value c.maxOpenFiles + writeBufferMB <- + option auto $ + metavar "MB" + <> long "write-buffer-mb" + <> help "RocksDB memtable budget across all column families in MiB (0 = unlimited)" + <> showDefault + <> value c.writeBufferMB + maxWalMB <- + option auto $ + metavar "MB" + <> long "max-wal-mb" + <> help "RocksDB write-ahead log size in MiB that forces flushes so old log files can be deleted (0 = RocksDB default)" + <> showDefault + <> value c.maxWalMB pure Config {..} networkReader :: String -> Either String Network @@ -568,21 +608,21 @@ run :: Config -> IO () run cfg = withContext $ \ctx -> - runStderrLoggingT $ filterLogger l $ flip runContT return $ do + withLogger level $ \logger -> flip runContT return $ do net <- either error return $ networkReader cfg.net - stats <- ContT $ with_stats net + stats <- ContT $ with_stats logger net let dir = cfg.dir net.name - $(logInfoS) "haskoin-store" $ + logInfo (withSource "haskoin-store" logger) $ "Creating working directory (if not found): " <> cs dir createDirectoryIfMissing True dir - store <- ContT $ run_store net ctx stats dir + store <- ContT $ run_store logger net ctx stats dir if cfg.noWeb then lift . withSubscription store.pub $ \sub -> - forever $ void (receive sub) + loopForever (receive sub) else do - lift $ run_web stats store + lift $ run_web logger stats store where - run_store net ctx stats dir = + run_store logger net ctx stats dir = withStore StoreConfig { maxPeers = cfg.maxPeers, @@ -599,15 +639,26 @@ wipeMempool = cfg.wipeMempool, noMempool = cfg.noMempool, syncMempool = cfg.syncMempool, + blockWindow = cfg.blockWindow, mempoolTimeout = cfg.mempoolTimeout, peerTimeout = fromIntegral cfg.peerTimeout, maxPeerLife = fromIntegral cfg.maxPeerLife, connect = withConnection, stats = stats, redisSyncInterval = cfg.redisSyncInterval, - bloom = cfg.bloom + bloom = cfg.bloom, + maxOpenFiles = if cfg.maxOpenFiles > 0 then Just cfg.maxOpenFiles else Nothing, + writeBufferSize = + if cfg.writeBufferMB > 0 + then Just (cfg.writeBufferMB * 1024 * 1024) + else Nothing, + maxWalSize = + if cfg.maxWalMB > 0 + then Just (cfg.maxWalMB * 1024 * 1024) + else Nothing, + logger = logger } - run_web stats store = + run_web logger stats store = runWeb WebConfig { host = cfg.host, @@ -625,11 +676,12 @@ priceHistoryURL = cfg.priceHistoryURL, noXPub = cfg.noXPub, noBlockchainInfo = cfg.noBlockchainInfo, - healthCheckInterval = cfg.healthCheckInterval + healthCheckInterval = cfg.healthCheckInterval, + logger = withSource "Web" logger } - with_stats net go + with_stats logger net go | cfg.statsd = do - $(logInfoS) "Main" $ + logInfo (withSource "Main" logger) $ "Sending stats to " <> T.pack cfg.statsdHost <> ":" @@ -646,6 +698,6 @@ } (go . Just) | otherwise = go Nothing - l _ lvl - | cfg.debug = True - | otherwise = LevelInfo <= lvl + level + | cfg.debug = LevelDebug + | otherwise = LevelInfo diff --git a/haskoin-store.cabal b/haskoin-store.cabal --- a/haskoin-store.cabal +++ b/haskoin-store.cabal @@ -5,7 +5,7 @@ -- see: https://github.com/sol/hpack name: haskoin-store -version: 1.6.3 +version: 1.6.4 synopsis: Storage and index for Bitcoin and Bitcoin Cash description: Please see the README on GitHub at category: Bitcoin, Finance, Network @@ -42,7 +42,7 @@ aeson >=1.4.7.1 , aeson-pretty >=0.8.8 , base >=4.9 && <5 - , base16 >=1.0 + , base16 >=2.0 , bytes >=0.17 , bytestring >=0.10.10.0 , cereal >=0.5.8.1 @@ -52,19 +52,18 @@ , deepseq >=1.4.4.0 , hashable >=1.3.0.0 , hashtables >=1.2.4.2 - , haskoin-core >=1.3.0 - , haskoin-node >=1.4.3 + , haskoin-core >=1.5.0 + , haskoin-node >=2.1.2 , haskoin-store-data ==1.6.3 , hedis >=0.16.1 , http-types >=0.12.3 , lens >=4.18.1 , monad-control >=1.0.2.3 - , monad-logger >=0.3.32 , mtl >=2.2.2 , network >=3.1.1.1 , nqe >=0.6.5 , random >=1.1 - , rocksdb-haskell-jprupp >=2.3.0 + , rocksdb-haskell-jprupp >=2.5.0 , rocksdb-query >=0.6.0 , scotty >=0.22 , statsd-rupp >=0.4 @@ -79,7 +78,7 @@ , wai >=3.2.2.1 , wai-websockets >=3.0.1.2 , warp >=3.3.10 - , websockets >=0.12.4 + , websockets >=0.13 , wreq >=0.5.3.2 default-language: Haskell2010 @@ -89,12 +88,12 @@ Paths_haskoin_store hs-source-dirs: app - ghc-options: -Wall -Wcompat -Widentities -Wincomplete-record-updates -Wincomplete-uni-patterns -Wmissing-export-lists -Wmissing-home-modules -Wpartial-fields -Wredundant-constraints -threaded -rtsopts -with-rtsopts=-N + ghc-options: -Wall -Wcompat -Widentities -Wincomplete-record-updates -Wincomplete-uni-patterns -Wmissing-export-lists -Wmissing-home-modules -Wpartial-fields -Wredundant-constraints -threaded -rtsopts -with-rtsopts=-N4 build-depends: aeson >=1.4.7.1 , aeson-pretty >=0.8.8 , base >=4.9 && <5 - , base16 >=1.0 + , base16 >=2.0 , bytes >=0.17 , bytestring >=0.10.10.0 , cereal >=0.5.8.1 @@ -105,21 +104,20 @@ , filepath , hashable >=1.3.0.0 , hashtables >=1.2.4.2 - , haskoin-core >=1.3.0 - , haskoin-node >=1.4.3 + , haskoin-core >=1.5.0 + , haskoin-node >=2.1.2 , haskoin-store , haskoin-store-data ==1.6.3 , hedis >=0.16.1 , http-types >=0.12.3 , lens >=4.18.1 , monad-control >=1.0.2.3 - , monad-logger >=0.3.32 , mtl >=2.2.2 , network >=3.1.1.1 , nqe >=0.6.5 , optparse-applicative , random >=1.1 - , rocksdb-haskell-jprupp >=2.3.0 + , rocksdb-haskell-jprupp >=2.5.0 , rocksdb-query >=0.6.0 , scotty >=0.22 , statsd-rupp >=0.4 @@ -134,7 +132,7 @@ , wai >=3.2.2.1 , wai-websockets >=3.0.1.2 , warp >=3.3.10 - , websockets >=0.12.4 + , websockets >=0.13 , wreq >=0.5.3.2 default-language: Haskell2010 @@ -142,18 +140,21 @@ type: exitcode-stdio-1.0 main-is: Spec.hs other-modules: + Haskoin.Store.BlockStoreSpec Haskoin.Store.CacheSpec + Haskoin.Store.LogicSpec + Haskoin.Store.TestBlocks Haskoin.StoreSpec Paths_haskoin_store hs-source-dirs: test - ghc-options: -Wall -Wcompat -Widentities -Wincomplete-record-updates -Wincomplete-uni-patterns -Wmissing-export-lists -Wmissing-home-modules -Wpartial-fields -Wredundant-constraints -threaded -rtsopts -with-rtsopts=-N + ghc-options: -Wall -Wcompat -Widentities -Wincomplete-record-updates -Wincomplete-uni-patterns -Wmissing-export-lists -Wmissing-home-modules -Wpartial-fields -Wredundant-constraints -threaded -rtsopts "-with-rtsopts=-N4 -T" build-depends: QuickCheck >=2.13.2 , aeson >=1.4.7.1 , aeson-pretty >=0.8.8 , base >=4.9 && <5 - , base16 >=1.0 + , base16 >=2.0 , base64 >=0.4.1 , bytes >=0.17 , bytestring >=0.10.10.0 @@ -162,10 +163,11 @@ , containers >=0.6.2.1 , data-default >=0.7.1.1 , deepseq >=1.4.4.0 + , fast-logger >=3.0 , hashable >=1.3.0.0 , hashtables >=1.2.4.2 - , haskoin-core >=1.3.0 - , haskoin-node >=1.4.3 + , haskoin-core >=1.5.0 + , haskoin-node >=2.1.2 , haskoin-store , haskoin-store-data ==1.6.3 , hedis >=0.16.1 @@ -173,12 +175,11 @@ , http-types >=0.12.3 , lens >=4.18.1 , monad-control >=1.0.2.3 - , monad-logger >=0.3.32 , mtl >=2.2.2 , network >=3.1.1.1 , nqe >=0.6.5 , random >=1.1 - , rocksdb-haskell-jprupp >=2.3.0 + , rocksdb-haskell-jprupp >=2.5.0 , rocksdb-query >=0.6.0 , scotty >=0.22 , statsd-rupp >=0.4 @@ -193,7 +194,7 @@ , wai >=3.2.2.1 , wai-websockets >=3.0.1.2 , warp >=3.3.10 - , websockets >=0.12.4 + , websockets >=0.13 , wreq >=0.5.3.2 default-language: Haskell2010 build-tool-depends: hspec-discover:hspec-discover 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 @@ -8,7 +8,6 @@ {-# LANGUAGE OverloadedRecordDot #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE RecordWildCards #-} -{-# LANGUAGE TemplateHaskell #-} {-# LANGUAGE TupleSections #-} {-# LANGUAGE NoFieldSelectors #-} {-# OPTIONS_GHC -Wno-ambiguous-fields #-} @@ -34,31 +33,18 @@ blockStoreTxHashSTM, blockStorePendingTxs, blockStorePendingTxsSTM, + needsMoreBlocks, ) where import Control.Monad ( forM, forM_, - forever, mzero, unless, void, when, ) -import Control.Monad.Except - ( ExceptT (..), - MonadError, - catchError, - runExceptT, - ) -import Control.Monad.Logger - ( MonadLoggerIO, - logDebugS, - logErrorS, - logInfoS, - logWarnS, - ) import Control.Monad.Reader ( MonadReader, ReaderT (..), @@ -124,6 +110,8 @@ ) import Haskoin.Node ( Chain, + LogLevel (..), + Logger, OnlinePeer (..), Peer (..), PeerMgr, @@ -134,6 +122,7 @@ chainGetParents, getPeers, killPeer, + logAt, sendMessage, setBusy, setFree, @@ -144,6 +133,7 @@ import Haskoin.Store.Database.Writer import Haskoin.Store.Logic ( ImportException (Orphan), + ImportM, deleteUnconfirmedTx, importBlock, initBest, @@ -173,13 +163,16 @@ TVar, async, atomically, + catch, liftIO, link, modifyTVar, + modifyTVar', newTVarIO, readTVar, readTVarIO, throwIO, + try, withAsync, writeTVar, ) @@ -321,23 +314,38 @@ wipeMempool :: !Bool, -- | sync mempool from peers syncMempool :: !Bool, - -- | disconnect syncing peer if inactive for this long - mempoolTimeout :: !Int, + -- | blocks to request from the syncing peer in one batch + blockWindow :: !Int, -- | delete transactions older than this number of days from mempool + mempoolTimeout :: !Int, + -- | disconnect syncing peer if inactive for this long peerTimeout :: !NominalDiffTime, - stats :: !(Maybe Stats) + stats :: !(Maybe Stats), + -- | log messages here + logger :: !Logger } type BlockT m = ReaderT BlockStore m +logDebug, logInfo, logWarn, logError :: + (MonadIO m, MonadReader BlockStore m) => Text -> m () +logDebug = blockLog LevelDebug +logInfo = blockLog LevelInfo +logWarn = blockLog LevelWarn +logError = blockLog LevelError + +blockLog :: (MonadIO m, MonadReader BlockStore m) => LogLevel -> Text -> m () +blockLog l msg = asks (.config.logger) >>= \lg -> logAt l lg msg + runImport :: - (MonadLoggerIO m) => + (MonadIO m) => Network -> Ctx -> - WriterT (ExceptT ImportException m) a -> + ImportM a -> BlockT m (Either ImportException a) runImport net ctx f = - ReaderT $ \r -> runExceptT $ runWriter net ctx r.config.db f + ReaderT $ \r -> + liftIO . try $ runWriter net ctx r.config.db r.config.logger f runRocksDB :: ReaderT DatabaseReader m a -> BlockT m a runRocksDB f = @@ -393,7 +401,7 @@ -- | Run block store process. withBlockStore :: - (MonadUnliftIO m, MonadLoggerIO m) => + (MonadUnliftIO m) => BlockStoreConfig -> (BlockStore -> m a) -> m a @@ -423,7 +431,7 @@ wipe run inbox del txs = do - $(logInfoS) "BlockStore" $ + logAt LevelInfo cfg.logger $ "Deleting " <> cs (show (length txs)) <> " transactions" forM_ txs $ \(_, th) -> deleteUnconfirmedTx False th wipe_it txs = do @@ -433,7 +441,7 @@ unless (null txs1) $ runImport net ctx (del txs1) >>= \case Left e -> do - $(logErrorS) "BlockStore" $ + logError $ "Could not wipe mempool: " <> cs (show e) throwIO e Right () -> wipe_it txs2 @@ -447,7 +455,7 @@ ctx <- getCtx runImport net ctx initBest >>= \case Left e -> do - $(logErrorS) "BlockStore" $ + logError $ "Could not initialize: " <> cs (show e) throwIO e Right () -> return () @@ -456,20 +464,20 @@ link a >> runBlockStoreLoop inbox runBlockStoreLoop :: - (InChan mbox, MonadUnliftIO m, MonadLoggerIO m) => + (InChan mbox, MonadUnliftIO m) => mbox BlockStoreMessage -> ReaderT BlockStore m b runBlockStoreLoop inbox = - forever $ do - $(logDebugS) "BlockStore" "Waiting for new event..." + loopForever $ do + logDebug "Waiting for new event..." msg <- receive inbox ReaderT $ runReaderT $ processBlockStoreMessage msg -isInSync :: (MonadLoggerIO m) => BlockT m Bool +isInSync :: (MonadIO m) => BlockT m Bool isInSync = getBestBlock >>= \case Nothing -> do - $(logErrorS) "BlockStore" "Block database uninitialized" + logError "Block database uninitialized" throwIO Uninitialized Just bb -> do cb <- asks (.config.chain) >>= liftIO . chainGetBest @@ -477,26 +485,26 @@ then clearSyncingState >> return True else return False -guardSync :: (MonadLoggerIO m) => BlockT m () -> BlockT m () +guardSync :: (MonadIO m) => BlockT m () -> BlockT m () guardSync f = isInSync >>= \s -> when s f guardMempool :: (Monad m) => BlockT m () -> BlockT m () guardMempool f = asks (.config.noMempool) >>= \n -> unless n f -syncMempool :: (MonadUnliftIO m, MonadLoggerIO m) => Peer -> BlockT m () +syncMempool :: (MonadUnliftIO m) => Peer -> BlockT m () syncMempool p = guardMempool . guardSync $ do s <- asks (.config.syncMempool) m <- asks (.syncedMempool) y <- not . HashSet.member p.label <$> readTVarIO m when (s && y) $ do - $(logDebugS) "BlockStore" $ + logDebug $ "Requesting mempool from peer: " <> p.label atomically . modifyTVar m $ HashSet.insert p.label liftIO $ MMempool `sendMessage` p processBlock :: - (MonadUnliftIO m, MonadLoggerIO m) => + (MonadUnliftIO m) => Peer -> Block -> BlockT m () @@ -504,7 +512,7 @@ checkPeer peer >>= \case True -> return () False -> do - $(logErrorS) "BlockStore" $ + logError $ "Non-syncing peer " <> peer.label <> " sent block " @@ -515,14 +523,14 @@ getBlockNode blockhash >>= \case Just b -> return b Nothing -> do - $(logErrorS) "BlockStore" $ + logError $ "Peer " <> peer.label <> " sent unknown block " <> blockHashToHex blockhash killPeer peer mzero - $(logDebugS) "BlockStore" $ + logDebug $ "Processing block " <> blockText node Nothing <> " from peer " @@ -539,7 +547,7 @@ blockhash = headerHash header hexhash = blockHashToHex blockhash success node = do - $(logInfoS) "BlockStore" $ + logInfo $ "Best block: " <> blockText node (Just block) removeSyncingBlock $ headerHash node.header touchPeer @@ -549,7 +557,7 @@ updateOrphans syncMempool peer failure e = do - $(logErrorS) "BlockStore" $ + logError $ "Error importing block " <> hexhash <> " from peer: " @@ -597,13 +605,13 @@ Nothing -> Nothing Just x -> Just (x :: Syncing) {blocks = delete h x.blocks} -checkPeer :: (MonadLoggerIO m, MonadReader BlockStore m) => Peer -> m Bool +checkPeer :: (MonadIO m, MonadReader BlockStore m) => Peer -> m Bool checkPeer p = fmap (fromMaybe False) <$> runMaybeT $ do Syncing {peer} <- MaybeT getSyncingState return $ peer == p getBlockNode :: - (MonadLoggerIO m, MonadReader BlockStore m) => + (MonadIO m, MonadReader BlockStore m) => BlockHash -> m (Maybe BlockNode) getBlockNode blockhash = do @@ -611,13 +619,13 @@ liftIO $ chainGetBlock ch blockhash processNoBlocks :: - (MonadLoggerIO m) => + (MonadIO m) => Peer -> [BlockHash] -> BlockT m () processNoBlocks p hs = do forM_ (zip [(1 :: Int) ..] hs) $ \(i, h) -> - $(logErrorS) "BlockStore" $ + logError $ "Block " <> cs (show i) <> "/" @@ -628,10 +636,10 @@ <> p.label killPeer p -processTx :: (MonadLoggerIO m) => Peer -> Tx -> BlockT m () +processTx :: (MonadIO m) => Peer -> Tx -> BlockT m () processTx p tx = guardMempool $ do t <- liftIO getCurrentTime - $(logDebugS) "BlockManager" $ + logDebug $ "Received tx " <> txHashToHex (txHash tx) <> " by peer: " @@ -642,8 +650,9 @@ pruneOrphans = guardMempool $ do ts <- asks (.txs) now <- liftIO getCurrentTime - atomically . modifyTVar ts . HashMap.filter $ \p -> - now `diffUTCTime` p.time > 3600 + -- keep pending transactions received within the last hour, drop older ones + atomically . modifyTVar' ts . HashMap.filter $ \p -> + now `diffUTCTime` p.time <= 3600 addPendingTx :: (MonadIO m) => PendingTx -> BlockT m () addPendingTx p = do @@ -706,7 +715,7 @@ updateOrphans :: ( StoreReadBase m, - MonadLoggerIO m, + MonadIO m, MonadReader BlockStore m ) => m () @@ -738,13 +747,13 @@ return $ foldl fulfill p unspents newOrphanTx :: - (MonadLoggerIO m) => + (MonadIO m) => BlockStore -> UTCTime -> Tx -> WriterT m () newOrphanTx block_read time tx = do - $(logDebugS) "BlockStore" $ + logAt LevelDebug block_read.config.logger $ "Import tx " <> txHashToHex (txHash tx) <> ": Orphan" @@ -765,16 +774,12 @@ prev_set = HashSet.fromList prevs prevs = map (.outpoint) tx.inputs -importMempoolTx :: - (MonadLoggerIO m, MonadError ImportException m) => - BlockStore -> - UTCTime -> - Tx -> - WriterT m Bool +importMempoolTx :: BlockStore -> UTCTime -> Tx -> ImportM Bool importMempoolTx block_read time tx = - catchError new_mempool_tx handle_error + catch new_mempool_tx handle_error where tx_hash = txHash tx + handle_error :: ImportException -> ImportM Bool handle_error Orphan = do newOrphanTx block_read time tx return False @@ -782,7 +787,7 @@ seconds = floor (utcTimeToPOSIXSeconds time) new_mempool_tx = do t <- newMempoolTx tx seconds - $(logInfoS) "BlockStore" $ + logAt LevelInfo block_read.config.logger $ bool "Already have" "Imported" t <> " tx " <> txHashToHex (txHash tx) @@ -808,11 +813,12 @@ Just b -> HashSet.fromList (map txHash b.txs) Nothing -> HashSet.empty -processMempool :: (MonadLoggerIO m) => BlockT m () -processMempool = guardMempool . notify Nothing $ do +processMempool :: (MonadIO m) => BlockT m () +processMempool = guardMempool $ do txs <- pendingTxs 2000 block_read <- ask - unless (null txs) (import_txs block_read txs) + -- notify reads the whole mempool twice, so only when there is work to do + unless (null txs) $ notify Nothing (import_txs block_read txs) where run_import block_read p = importMempoolTx block_read p.time p.tx @@ -825,13 +831,13 @@ Left e -> report_error e Right _ -> return () report_error e = do - $(logErrorS) "BlockStore" $ + logError $ "Error processing mempool: " <> cs (show e) throwIO e -pruneMempool :: (MonadLoggerIO m) => BlockT m () +pruneMempool :: (MonadIO m) => BlockT m () pruneMempool = - guardMempool . notify Nothing $ do + guardMempool $ do days <- asks (.config.mempoolTimeout) when (days > 0) $ do mempool <- reverse <$> getMempool @@ -840,22 +846,23 @@ txs = take 1000 $ map snd $ takeWhile ((< thresh) . fst) mempool net <- getNetwork ctx <- getCtx - when (length txs > 0) $ do - $(logInfoS) "BlockStore" $ + -- notify reads the whole mempool twice, so only when there is work to do + unless (null txs) . notify Nothing $ do + logInfo $ "Deleting " <> cs (show (length txs)) <> " old transactions from mempool" runImport net ctx (mapM_ (deleteUnconfirmedTx False) txs) >>= \case Left e -> - $(logErrorS) "BlockStore" $ + logError $ "Error pruning mempool: " <> cs (show e) Right () -> return () processTxs :: - (MonadLoggerIO m) => + (MonadIO m) => Peer -> [TxHash] -> BlockT m () processTxs p hs = guardMempool . guardSync $ do - $(logDebugS) "BlockStore" $ + logDebug $ "Received inventory with " <> cs (show (length hs)) <> " transactions from peer: " @@ -868,7 +875,7 @@ process_tx i h = isPending h >>= \case True -> do - $(logDebugS) "BlockStore" $ + logDebug $ "Tx " <> cs (show i) <> "/" @@ -881,7 +888,7 @@ False -> getTxData h >>= \case Just _ -> do - $(logDebugS) "BlockStore" $ + logDebug $ "Tx " <> cs (show i) <> "/" @@ -892,7 +899,7 @@ <> "Already Imported" return Nothing Nothing -> do - $(logDebugS) "BlockStore" $ + logDebug $ "Tx " <> cs (show i) <> "/" @@ -926,7 +933,7 @@ fmap $ \x -> (x :: Syncing) {time = now} -checkTime :: (MonadLoggerIO m) => BlockT m () +checkTime :: (MonadIO m) => BlockT m () checkTime = asks (.peer) >>= readTVarIO >>= \case Nothing -> return () @@ -938,23 +945,23 @@ now <- liftIO getCurrentTime peer_time_out <- asks (.config.peerTimeout) when (now `diffUTCTime` t > peer_time_out) $ do - $(logErrorS) "BlockStore" $ + logError $ "Syncing peer timeout: " <> p.label killPeer p -revertToMainChain :: (MonadLoggerIO m) => BlockT m () +revertToMainChain :: (MonadIO m) => BlockT m () revertToMainChain = do h <- headerHash . (.header) <$> getBest ch <- asks (.config.chain) net <- getNetwork ctx <- getCtx liftIO (chainBlockMain ch h) >>= \x -> unless x $ do - $(logWarnS) "BlockStore" $ + logWarn $ "Reverting best block: " <> blockHashToHex h runImport net ctx (revertBlock h) >>= \case Left e -> do - $(logErrorS) "BlockStore" $ + logError $ "Could not revert block " <> blockHashToHex h <> ": " @@ -963,24 +970,24 @@ Right () -> setSyncingBlocks [] revertToMainChain -getBest :: (MonadLoggerIO m) => BlockT m BlockNode +getBest :: (MonadIO m) => BlockT m BlockNode getBest = do bb <- getBestBlock >>= \case Just b -> return b Nothing -> do - $(logErrorS) "BlockStore" "No best block set" + logError "No best block set" throwIO Uninitialized ch <- asks (.config.chain) liftIO (chainGetBlock ch bb) >>= \case Just x -> return x Nothing -> do - $(logErrorS) "BlockStore" $ + logError $ "Header not found for best block: " <> blockHashToHex bb throwIO (BlockNotInChain bb) -getSyncBest :: (MonadLoggerIO m) => BlockT m BlockNode +getSyncBest :: (MonadIO m) => BlockT m BlockNode getSyncBest = do bb <- getSyncingBlocks >>= \case @@ -988,30 +995,49 @@ getBestBlock >>= \case Just b -> return b Nothing -> do - $(logErrorS) "BlockStore" "No best block set" + logError "No best block set" throwIO Uninitialized hs -> return $ last hs ch <- asks (.config.chain) liftIO (chainGetBlock ch bb) >>= \case Just x -> return x Nothing -> do - $(logErrorS) "BlockStore" $ + logError $ "Header not found for block: " <> blockHashToHex bb throwIO (BlockNotInChain bb) -shouldSync :: (MonadLoggerIO m) => BlockT m (Maybe Peer) +-- | The syncing peer, when it is time to ask it for the next batch of blocks. +shouldSync :: (MonadIO m) => BlockT m (Maybe Peer) shouldSync = isInSync >>= \case True -> return Nothing False -> getSyncingState >>= \case Nothing -> return Nothing - Just Syncing {peer = p, blocks = bs} - | 100 > length bs -> return (Just p) - | otherwise -> return Nothing + Just Syncing {peer = p, blocks = bs} -> do + window <- blockWindow + return $ + if needsMoreBlocks window (length bs) + then Just p + else Nothing -syncMe :: (MonadLoggerIO m) => BlockT m () +-- | Blocks to request from the syncing peer in one batch. Never below one, +-- or the store would never ask for any block. +blockWindow :: (Monad m) => BlockT m Int +blockWindow = asks (max 1 . (.config.blockWindow)) + +-- | Whether to ask for the next batch, given the window size and how many +-- requested blocks are still waiting to be imported. Asking once fewer than +-- a quarter of the window remain keeps the peer busy, while the queue of +-- downloaded blocks never grows much beyond one window. The threshold is at +-- least one, so that a window smaller than four still refills once empty. +needsMoreBlocks :: Int -> Int -> Bool +needsMoreBlocks window outstanding = + outstanding < max 1 (window `div` 4) + +-- | Ask the syncing peer for the next batch of blocks, if the window has room. +syncMe :: (MonadIO m) => BlockT m () syncMe = do revertToMainChain shouldSync >>= \case @@ -1022,7 +1048,7 @@ when (bb /= bh) $ do bns <- sel bb bh iv <- getiv bns - $(logDebugS) "BlockStore" $ + logDebug $ "Requesting " <> fromString (show (length iv)) <> " blocks from peer: " @@ -1037,24 +1063,17 @@ return $ map f bns getbh = liftIO . chainGetBest =<< asks (.config.chain) + -- The blocks right after the last requested one, up to one window or + -- the chain tip, whichever comes first. sel bb bh = do - let sh = geth bb bh - t <- top sh bh + window <- blockWindow + top <- findAncestor (min (bb.height + fromIntegral window) bh.height) bh ch <- asks (.config.chain) - ps <- liftIO $ chainGetParents ch (bb.height + 1) t - return $ - if 500 > length ps - then ps <> [bh] - else ps - geth bb bh = - min (bb.height + 501) bh.height - top sh bh = - if sh == bh.height - then return bh - else findAncestor sh bh + ps <- liftIO $ chainGetParents ch (bb.height + 1) top + return $ ps <> [top] findAncestor :: - (MonadLoggerIO m, MonadReader BlockStore m) => + (MonadIO m, MonadReader BlockStore m) => BlockHeight -> BlockNode -> m BlockNode @@ -1064,7 +1083,7 @@ Just ancestor -> return ancestor Nothing -> do let h = headerHash target.header - $(logErrorS) "BlockStore" $ + logError $ "Could not find header for ancestor of block " <> blockHashToHex h <> " at height " @@ -1072,7 +1091,7 @@ throwIO $ AncestorNotInChain height h finishPeer :: - (MonadLoggerIO m, MonadReader BlockStore m) => + (MonadIO m, MonadReader BlockStore m) => Peer -> m () finishPeer p = do @@ -1085,10 +1104,10 @@ where reset_it box = do atomically $ writeTVar box Nothing - $(logDebugS) "BlockStore" $ "Releasing peer: " <> p.label + logDebug $ "Releasing peer: " <> p.label liftIO $ setFree p -trySetPeer :: (MonadLoggerIO m) => Peer -> BlockT m Bool +trySetPeer :: (MonadIO m) => Peer -> BlockT m Bool trySetPeer p = getSyncingState >>= \case Just _ -> return False @@ -1098,7 +1117,7 @@ liftIO (setBusy p) >>= \case False -> return False True -> do - $(logDebugS) "BlockStore" $ + logDebug $ "Locked peer: " <> p.label box <- asks (.peer) now <- liftIO getCurrentTime @@ -1111,7 +1130,7 @@ } return True -trySyncing :: (MonadLoggerIO m) => BlockT m () +trySyncing :: (MonadIO m) => BlockT m () trySyncing = isInSync >>= \case True -> return () @@ -1130,7 +1149,7 @@ let ps = map (.mailbox) ops recurse ps -trySyncingPeer :: (MonadUnliftIO m, MonadLoggerIO m) => Peer -> BlockT m () +trySyncingPeer :: (MonadUnliftIO m) => Peer -> BlockT m () trySyncingPeer p = isInSync >>= \case True -> syncMempool p @@ -1145,59 +1164,59 @@ readTVarIO =<< asks (.peer) clearSyncingState :: - (MonadLoggerIO m, MonadReader BlockStore m) => m () + (MonadIO m, MonadReader BlockStore m) => m () clearSyncingState = asks (.peer) >>= readTVarIO >>= \case Nothing -> return () Just Syncing {peer = p} -> finishPeer p processBlockStoreMessage :: - (MonadUnliftIO m, MonadLoggerIO m) => + (MonadUnliftIO m) => BlockStoreMessage -> BlockT m () processBlockStoreMessage (BlockNewBest node) = do - $(logDebugS) "BlockStore" $ + logDebug $ "New best block mined at height " <> cs (show node.height) <> ": " <> blockHashToHex (headerHash node.header) trySyncing processBlockStoreMessage (BlockPeerConnect p) = do - $(logDebugS) "BlockStore" $ + logDebug $ "New peer connected: " <> p.label trySyncingPeer p processBlockStoreMessage (BlockPeerDisconnect p) = do - $(logDebugS) "BlockStore" $ + logDebug $ "Peer disconnected: " <> p.label finishPeer p processBlockStoreMessage (BlockReceived p b) = do - $(logDebugS) "BlockStore" $ + logDebug $ "Received block: " <> blockHashToHex (headerHash b.header) processBlock p b processBlockStoreMessage (BlockNotFound p bs) = do - $(logDebugS) "BlockStore" $ + logDebug $ "Blocks not found by peer " <> p.label <> ": " <> T.unwords (map blockHashToHex bs) processNoBlocks p bs processBlockStoreMessage (TxRefReceived p tx) = do - $(logDebugS) "BlockStore" $ + logDebug $ "Transaction received from peer " <> p.label <> ": " <> txHashToHex (txHash tx) processTx p tx processBlockStoreMessage (TxRefAvailable p ts) = do - $(logDebugS) "BlockStore" $ + logDebug $ "Transactions available from peer " <> p.label <> ": " <> T.unwords (map txHashToHex ts) processTxs p ts processBlockStoreMessage (BlockPing r) = do - $(logDebugS) "BlockStore" "Internal clock event" + logDebug "Internal clock event" setStoreHeight setHeadersHeight setPendingTxs @@ -1210,9 +1229,9 @@ checkTime atomically (r ()) -pingMe :: (MonadLoggerIO m) => Mailbox BlockStoreMessage -> m () +pingMe :: (MonadIO m) => Mailbox BlockStoreMessage -> m () pingMe mbox = - forever $ do + loopForever $ do BlockPing `query` mbox delay <- liftIO $ randomRIO (100 * 1000, 1000 * 1000) threadDelay delay diff --git a/src/Haskoin/Store/Cache.hs b/src/Haskoin/Store/Cache.hs --- a/src/Haskoin/Store/Cache.hs +++ b/src/Haskoin/Store/Cache.hs @@ -11,7 +11,6 @@ {-# LANGUAGE OverloadedRecordDot #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE RecordWildCards #-} -{-# LANGUAGE TemplateHaskell #-} {-# LANGUAGE TupleSections #-} {-# LANGUAGE NoFieldSelectors #-} @@ -37,13 +36,7 @@ where import Control.DeepSeq (NFData) -import Control.Monad (forM, forM_, forever, unless, void, when) -import Control.Monad.Logger - ( MonadLoggerIO, - logDebugS, - logErrorS, - logInfoS, - ) +import Control.Monad (forM, forM_, unless, void, when) import Control.Monad.Reader (ReaderT (..), asks) import Control.Monad.Trans (lift) import Data.Bits (shift, (.&.), (.|.)) @@ -111,10 +104,13 @@ ) import Haskoin.Node ( Chain, + LogLevel (..), + Logger, chainGetBest, chainGetBlock, chainGetParents, chainGetSplitBlock, + logAt, ) import Haskoin.Store.Common import Haskoin.Store.Data @@ -123,8 +119,11 @@ ( Inbox, Listen, Mailbox, + mailboxEmptySTM, query, receive, + receiveSTM, + requeueSTM, send, ) import System.Metrics.StatsD @@ -140,17 +139,19 @@ ) import UnliftIO.Concurrent (threadDelay) -runRedis :: (MonadLoggerIO m) => Redis (Either Reply a) -> CacheX m a +runRedis :: (MonadIO m) => Redis (Either Reply a) -> CacheX m a runRedis action = do conn <- asks (.redis) - withRedis conn action + logger <- asks (.logger) + withRedis logger conn action -withRedis :: (MonadLoggerIO m) => Connection -> Redis (Either Reply a) -> m a -withRedis conn action = +withRedis :: + (MonadIO m) => Logger -> Connection -> Redis (Either Reply a) -> m a +withRedis logger conn action = liftIO (Redis.runRedis conn action) >>= \case Right x -> return x Left e -> do - $(logErrorS) "Cache" $ "Got error from Redis: " <> cs (show e) + logAt LevelError logger $ "Got error from Redis: " <> cs (show e) throwIO (RedisError e) data CacheConfig = CacheConfig @@ -158,7 +159,8 @@ minAddrs :: !Int, maxKeys :: !Integer, chain :: !Chain, - metrics :: !(Maybe CacheMetrics) + metrics :: !(Maybe CacheMetrics), + logger :: !Logger } data CacheMetrics = CacheMetrics @@ -176,12 +178,13 @@ } newCacheMetrics :: - (MonadLoggerIO m) => + (MonadIO m) => + Logger -> Stats -> Connection -> DatabaseReader -> m CacheMetrics -newCacheMetrics s r db = do +newCacheMetrics logger s r db = do h <- redisBest hits <- c "cache.hits" misses <- c "cache.misses" @@ -197,7 +200,7 @@ return CacheMetrics {..} where redisBest = - maybe dbBest dbBlock =<< withRedis r redisGetHead + maybe dbBest dbBlock =<< withRedis logger r redisGetHead dbBlock h = maybe 0 (fromIntegral . (.height)) <$> withDB db (getBlock h) dbBest = @@ -230,6 +233,15 @@ type CacheX = ReaderT CacheConfig +logDebug, logInfo, logWarn, logError :: (MonadIO m) => Text -> CacheX m () +logDebug = cacheLog LevelDebug +logInfo = cacheLog LevelInfo +logWarn = cacheLog LevelWarn +logError = cacheLog LevelError + +cacheLog :: (MonadIO m) => LogLevel -> Text -> CacheX m () +cacheLog l msg = asks (.logger) >>= \lg -> logAt l lg msg + data CacheError = RedisError Reply | RedisTxError !String @@ -271,7 +283,7 @@ getMempool = lift getMempool instance - (MonadUnliftIO m, MonadLoggerIO m, StoreReadExtra m) => + (MonadUnliftIO m, StoreReadExtra m) => StoreReadExtra (CacheX m) where getBalances = lift . getBalances @@ -288,7 +300,7 @@ xPubUnspents = getXPubUnspents instance - (MonadUnliftIO m, MonadLoggerIO m, StoreReadExtra m) => + (MonadUnliftIO m, StoreReadExtra m) => StoreReadExtra (CacheT m) where getBalances = lift . getBalances @@ -335,7 +347,7 @@ xpubLock xpub = "xlock-" <> encode xpub getXPubTxs :: - (MonadUnliftIO m, MonadLoggerIO m, StoreReadExtra m) => + (MonadUnliftIO m, StoreReadExtra m) => XPubSpec -> [XPubBal] -> Limits -> @@ -351,7 +363,7 @@ else lift $ xPubTxs xpub xbals limits getXPubTxCount :: - (MonadUnliftIO m, MonadLoggerIO m, StoreReadExtra m) => + (MonadUnliftIO m, StoreReadExtra m) => XPubSpec -> [XPubBal] -> CacheX m Word32 @@ -365,7 +377,7 @@ else lift $ xPubTxCount xpub xbals getXPubUnspents :: - (MonadUnliftIO m, MonadLoggerIO m, StoreReadExtra m) => + (MonadUnliftIO m, StoreReadExtra m) => XPubSpec -> [XPubBal] -> Limits -> @@ -399,7 +411,7 @@ else lift $ xPubUnspents xpub xbals limits getXPubBalances :: - (MonadUnliftIO m, MonadLoggerIO m, StoreReadExtra m) => + (MonadUnliftIO m, StoreReadExtra m) => XPubSpec -> CacheX m [XPubBal] getXPubBalances xpub = do @@ -413,31 +425,39 @@ return xbals else do withMetrics $ \s -> incrementCounter s.misses 1 - withRetryLock (xpubLock xpub) $ do - bals <- lift $ xPubBals xpub - newXPubC xpub bals - return bals + withRetryLock (xpubLock xpub) (load_and_cache xpub) >>= \case + Just bals -> return bals + Nothing -> do + -- Another thread held the lock for too long: serve from the + -- database without caching rather than wait any longer. + logWarn "Could not acquire xpub lock, serving from database" + lift $ xPubBals xpub + where + load_and_cache x = do + bals <- lift $ xPubBals x + newXPubC x bals + return bals -isInCache :: (MonadLoggerIO m) => XPubSpec -> CacheT m Bool +isInCache :: (MonadIO m) => XPubSpec -> CacheT m Bool isInCache xpub = ReaderT $ \case Nothing -> return False Just cfg -> runReaderT (isXPubCached xpub) cfg -isXPubCached :: (MonadLoggerIO m) => XPubSpec -> CacheX m Bool +isXPubCached :: (MonadIO m) => XPubSpec -> CacheX m Bool isXPubCached xpub = runRedis (redisIsXPubCached xpub) redisIsXPubCached :: (RedisCtx m f) => XPubSpec -> m (f Bool) redisIsXPubCached xpub = Redis.exists (balancesPfx <> encode xpub) -cacheGetXPubBalances :: (MonadLoggerIO m) => XPubSpec -> CacheX m [XPubBal] +cacheGetXPubBalances :: (MonadIO m) => XPubSpec -> CacheX m [XPubBal] cacheGetXPubBalances xpub = do bals <- runRedis $ redisGetXPubBalances xpub touchKeys [xpub] return bals cacheGetXPubTxCount :: - (MonadUnliftIO m, MonadLoggerIO m) => + (MonadUnliftIO m) => XPubSpec -> CacheX m Word32 cacheGetXPubTxCount xpub = do @@ -449,7 +469,7 @@ redisGetXPubTxCount xpub = Redis.zcard (txSetPfx <> encode xpub) cacheGetXPubTxs :: - (StoreReadBase m, MonadLoggerIO m) => + (StoreReadBase m, MonadIO m) => XPubSpec -> Limits -> CacheX m [TxRef] @@ -500,7 +520,7 @@ f t s = TxRef {txid = t, block = scoreBlockRef s} cacheGetXPubUnspents :: - (StoreReadBase m, MonadLoggerIO m) => + (StoreReadBase m, MonadIO m) => XPubSpec -> Limits -> CacheX m [(BlockRef, OutPoint)] @@ -638,6 +658,8 @@ data CacheWriterMessage = CacheNewBlock | CacheNewTx !TxHash + | -- | internal: a batch of coalesced 'CacheNewTx' events + CacheNewTxs ![TxHash] | CacheSyncMempool !(Listen ()) type CacheWriterInbox = Inbox CacheWriterMessage @@ -668,7 +690,7 @@ xPubAddrFunction ctx DeriveP2WPKH = xPubWitnessAddr ctx cacheWriter :: - (MonadUnliftIO m, MonadLoggerIO m, StoreReadExtra m) => + (MonadUnliftIO m, StoreReadExtra m) => CacheConfig -> CacheWriterInbox -> m () @@ -678,37 +700,52 @@ go = do newBlockC syncMempoolC - forever $ do - $(logDebugS) "Cache" "Awaiting event..." + loopForever $ do + logDebug "Awaiting event..." x <- receive inbox - cacheWriterReact x + case x of + CacheNewTx txid -> do + -- Coalesce: every queued CacheNewTx is synced in one pass instead + -- of one Redis round-trip sequence per transaction, so the inbox + -- cannot grow without bound when Redis is slower than the mempool. + more <- liftIO (atomically (drainNewTxs [])) + cacheWriterReact (CacheNewTxs (txid : more)) + _ -> cacheWriterReact x + drainNewTxs acc + | length acc >= 1000 = return (reverse acc) + | otherwise = do + empty <- mailboxEmptySTM inbox + if empty + then return (reverse acc) + else + receiveSTM inbox >>= \case + CacheNewTx t -> drainNewTxs (t : acc) + other -> do + requeueSTM other inbox + return (reverse acc) -lockIt :: (MonadLoggerIO m) => ByteString -> CacheX m Bool +lockIt :: (MonadIO m) => ByteString -> CacheX m Bool lockIt l = do go >>= \case Right Redis.Ok -> do - $(logDebugS) "Cache" "Acquired lock" + logDebug "Acquired lock" withMetrics $ \s -> incrementCounter s.lockAcquired 1 return True Right Redis.Pong -> do - $(logErrorS) - "Cache" - "Unexpected pong when acquiring lock" + logError "Unexpected pong when acquiring lock" withMetrics $ \s -> incrementCounter s.lockFailed 1 return False Right (Redis.Status s) -> do - $(logErrorS) "Cache" $ + logError $ "Unexpected status acquiring lock: " <> cs s withMetrics $ \m -> incrementCounter m.lockFailed 1 return False Left (Redis.Bulk Nothing) -> do - $(logDebugS) "Cache" "Lock already taken" + logDebug "Lock already taken" withMetrics $ \s -> incrementCounter s.lockFailed 1 return False Left e -> do - $(logErrorS) - "Cache" - "Error when trying to acquire lock" + logError "Error when trying to acquire lock" withMetrics $ \s -> incrementCounter s.lockFailed 1 throwIO (RedisError e) where @@ -726,7 +763,7 @@ } Redis.setOpts l "locked" opts -refreshLock :: (MonadLoggerIO m) => ByteString -> CacheX m () +refreshLock :: (MonadIO m) => ByteString -> CacheX m () refreshLock l = void . runRedis $ do let opts = Redis.SetOpts @@ -739,11 +776,11 @@ } Redis.setOpts l "locked" opts -unlockIt :: (MonadLoggerIO m) => ByteString -> CacheX m () +unlockIt :: (MonadIO m) => ByteString -> CacheX m () unlockIt l = void . runRedis . Redis.del $ NonEmpty.singleton l withLock :: - (MonadLoggerIO m, MonadUnliftIO m) => + (MonadUnliftIO m) => ByteString -> CacheX m a -> CacheX m (Maybe a) @@ -753,36 +790,44 @@ False -> return Nothing where go = withAsync refresh $ const f - refresh = forever $ do + refresh = loopForever $ do threadDelay (150 * 1000 * 1000) refreshLock l +-- | Run the action holding the lock, retrying with exponential backoff +-- while another thread holds it. The delay between attempts grows from +-- 0.2 s to at most 5 s, and after about 30 s of waiting it gives up and +-- returns 'Nothing' instead of keeping the calling thread blocked forever. withRetryLock :: - (MonadLoggerIO m, MonadUnliftIO m) => + (MonadUnliftIO m) => ByteString -> CacheX m a -> - CacheX m a + CacheX m (Maybe a) withRetryLock l f = - go (200 * 1000) + go 0 (200 * 1000) where - go n = + max_delay = 5 * 1000 * 1000 + max_wait = 30 * 1000 * 1000 + go waited n = withLock l f >>= \case - Just x -> return x - Nothing -> do - threadDelay n - go (n * 2) + Just x -> return (Just x) + Nothing + | waited >= max_wait -> return Nothing + | otherwise -> do + threadDelay n + go (waited + n) (min max_delay (n * 2)) pruneDB :: - (MonadUnliftIO m, MonadLoggerIO m, StoreReadBase m) => CacheX m () + (MonadUnliftIO m, StoreReadBase m) => CacheX m () pruneDB = void . withLock "lock" $ do x <- asks (.maxKeys) s <- runRedis Redis.dbsize - $(logDebugS) "Cache" "Pruning old xpubs" + logDebug "Pruning old xpubs" when (s > x) $ void . delXPubKeys . map fst =<< runRedis (getFromSortedSet maxKey Nothing 0 (fromIntegral (s - x))) -touchKeys :: (MonadLoggerIO m) => [XPubSpec] -> CacheX m () +touchKeys :: (MonadIO m) => [XPubSpec] -> CacheX m () touchKeys xpubs = do now <- round <$> liftIO getPOSIXTime :: (MonadIO m) => m Int runRedis $ redisTouchKeys now xpubs @@ -793,21 +838,23 @@ void <$> Redis.zadd maxKey (map ((realToFrac now,) . encode) xpubs) cacheWriterReact :: - (MonadUnliftIO m, MonadLoggerIO m, StoreReadExtra m) => + (MonadUnliftIO m, StoreReadExtra m) => CacheWriterMessage -> CacheX m () cacheWriterReact CacheNewBlock = do - $(logDebugS) "Cache" "Received new block event" + logDebug "Received new block event" pruneDB newBlockC syncMempoolC -cacheWriterReact (CacheNewTx txid) = do - $(logDebugS) "Cache" $ - "Received new transaction event: " <> txHashToHex txid +cacheWriterReact (CacheNewTx txid) = + cacheWriterReact (CacheNewTxs [txid]) +cacheWriterReact (CacheNewTxs txids) = do + logDebug $ + "Received " <> cs (show (length txids)) <> " new transaction event(s)" pruneDB - syncNewTxC [txid] + syncNewTxC txids cacheWriterReact (CacheSyncMempool l) = do - $(logDebugS) "Cache" "Received sync mempool event" + logDebug "Received sync mempool event" pruneDB newBlockC syncMempoolC @@ -817,7 +864,7 @@ lenNotNull = length . filter (not . nullBalance . (.balance)) newXPubC :: - (MonadUnliftIO m, MonadLoggerIO m, StoreReadExtra m) => + (MonadUnliftIO m, StoreReadExtra m) => XPubSpec -> [XPubBal] -> CacheX m () @@ -826,7 +873,7 @@ bracket set_index unset_index $ \j -> when j $ withTiming (.indexTime) $ do xpubtxt <- xpubText xpub - $(logDebugS) "Cache" $ + logDebug $ "Caching " <> xpubtxt <> ": " @@ -835,14 +882,14 @@ <> cs (show (lenNotNull xbals)) <> " used" utxo <- lift $ xPubUnspents xpub xbals def - $(logDebugS) "Cache" $ + logDebug $ "Caching " <> xpubtxt <> ": " <> cs (show (length utxo)) <> " utxos" xtxs <- lift $ xPubTxs xpub xbals def - $(logDebugS) "Cache" $ + logDebug $ "Caching " <> xpubtxt <> ": " @@ -855,7 +902,7 @@ d <- redisAddXPubUnspents xpub (map op utxo) e <- redisAddXPubTxs xpub xtxs return $ b >> c >> d >> e >> return () - $(logDebugS) "Cache" $ "Cached " <> xpubtxt + logDebug $ "Cached " <> xpubtxt where op XPubUnspent {unspent = u} = (u.outpoint, u.block) should_index = @@ -892,21 +939,21 @@ return $ cb.height > 0 && headerHash cb.header == bb newBlockC :: - (MonadUnliftIO m, MonadLoggerIO m, StoreReadExtra m) => + (MonadUnliftIO m, StoreReadExtra m) => CacheX m () newBlockC = inSync >>= \s -> when s . void . withLock "lock" $ do get_best_block_node >>= \case - Nothing -> $(logErrorS) "Cache" "No best block available" + Nothing -> logError "No best block available" Just best_block_node -> cacheGetHead >>= \case Nothing -> do - $(logInfoS) "Cache" "Initializing best cache block" + logInfo "Initializing best cache block" importBlockC $ headerHash best_block_node.header Just cache_head_hash -> get_block_node cache_head_hash >>= \case Nothing -> do - $(logErrorS) "Cache" $ + logError $ "Could not get best cache block: " <> blockHashToHex cache_head_hash Just cache_head_node -> do @@ -935,7 +982,7 @@ return $ map (headerHash . (.header)) blocks importBlockC :: - (MonadUnliftIO m, StoreReadExtra m, MonadLoggerIO m) => + (MonadUnliftIO m, StoreReadExtra m) => BlockHash -> CacheX m () importBlockC bh = @@ -943,20 +990,20 @@ Just bd -> do let ths = bd.txs tds <- sortTxData . catMaybes <$> mapM (lift . getTxData) ths - $(logDebugS) "Cache" $ + logDebug $ "Importing " <> cs (show (length tds)) <> " transactions from block " <> blockHashToHex bh importMultiTxC tds - $(logDebugS) "Cache" $ + logDebug $ "Done importing " <> cs (show (length tds)) <> " transactions from block " <> blockHashToHex bh cacheSetHead bd Nothing -> do - $(logErrorS) "Cache" $ + logError $ "Could not get block: " <> blockHashToHex bh throwIO . LogicError . cs $ @@ -964,24 +1011,24 @@ <> blockHashToHex bh importMultiTxC :: - (MonadUnliftIO m, StoreReadExtra m, MonadLoggerIO m) => + (MonadUnliftIO m, StoreReadExtra m) => [TxData] -> CacheX m () importMultiTxC txs = do ctx <- lift getCtx - $(logDebugS) "Cache" $ "Processing " <> cs (show (length txs)) <> " txs" - $(logDebugS) "Cache" $ + logDebug $ "Processing " <> cs (show (length txs)) <> " txs" + logDebug $ "Getting address information for " <> cs (show (length (alladdrs ctx))) <> " addresses" addrmap <- getaddrmap ctx let addrs = HashMap.keys addrmap - $(logDebugS) "Cache" $ + logDebug $ "Getting balances for " <> cs (show (HashMap.size addrmap)) <> " addresses" balmap <- getbalances addrs - $(logDebugS) "Cache" $ + logDebug $ "Getting unspent data for " <> cs (show (length (allops ctx))) <> " outputs" @@ -991,7 +1038,7 @@ let xpubs = allxpubsls addrmap forM_ (zip [(1 :: Int) ..] xpubs) $ \(i, xpub) -> do xpubtxt <- xpubText xpub - $(logDebugS) "Cache" $ + logDebug $ "Affected xpub " <> cs (show i) <> "/" @@ -999,19 +1046,19 @@ <> ": " <> xpubtxt addrs' <- do - $(logDebugS) "Cache" $ + logDebug $ "Getting xpub balances for " <> cs (show (length xpubs)) <> " xpubs" xmap <- getxbals xpubs let addrmap' = faddrmap (HashMap.keysSet xmap) addrmap - $(logDebugS) "Cache" "Starting Redis import pipeline" + logDebug "Starting Redis import pipeline" runRedis $ do x <- redisImportMultiTx ctx addrmap' unspentmap txs y <- redisUpdateBalances addrmap' balmap z <- redisTouchKeys now (HashMap.keys xmap) return $ x >> y >> z >> return () - $(logDebugS) "Cache" "Completed Redis pipeline" + logDebug "Completed Redis pipeline" return $ getNewAddrs ctx gap xmap (HashMap.elems addrmap') cacheAddAddresses addrs' where @@ -1126,33 +1173,33 @@ XPubBal {path = ainfo.path, balance = bal} cacheAddAddresses :: - (StoreReadExtra m, MonadUnliftIO m, MonadLoggerIO m) => + (StoreReadExtra m, MonadUnliftIO m) => [(Address, AddressXPub)] -> CacheX m () -cacheAddAddresses [] = $(logDebugS) "Cache" "No further addresses to add" +cacheAddAddresses [] = logDebug "No further addresses to add" cacheAddAddresses addrs = do ctx <- lift getCtx - $(logDebugS) "Cache" $ + logDebug $ "Adding " <> cs (show (length addrs)) <> " new generated addresses" - $(logDebugS) "Cache" "Getting balances" + logDebug "Getting balances" balmap <- HashMap.fromListWith (<>) <$> mapM (uncurry getbal) addrs - $(logDebugS) "Cache" "Getting unspent outputs" + logDebug "Getting unspent outputs" utxomap <- HashMap.fromListWith (<>) <$> mapM (uncurry getutxo) addrs - $(logDebugS) "Cache" "Getting transactions" + logDebug "Getting transactions" txmap <- HashMap.fromListWith (<>) <$> mapM (uncurry gettxmap) addrs - $(logDebugS) "Cache" "Running Redis pipeline" + logDebug "Running Redis pipeline" runRedis $ do a <- forM (HashMap.toList balmap) (uncurry redisAddXPubBalances) b <- forM (HashMap.toList utxomap) (uncurry redisAddXPubUnspents) c <- forM (HashMap.toList txmap) (uncurry redisAddXPubTxs) return $ sequence_ a >> sequence_ b >> sequence_ c - $(logDebugS) "Cache" "Completed Redis pipeline" + logDebug "Completed Redis pipeline" let xpubs = HashSet.toList . HashSet.fromList . map (.spec) $ Map.elems amap - $(logDebugS) "Cache" "Getting xpub balances" + logDebug "Getting xpub balances" xmap <- getbals xpubs gap <- lift getMaxGap let notnulls = getnotnull balmap @@ -1205,7 +1252,7 @@ Just bals -> addrsToAdd ctx gap bals a syncNewTxC :: - (MonadUnliftIO m, MonadLoggerIO m, StoreReadExtra m) => + (MonadUnliftIO m, StoreReadExtra m) => [TxHash] -> CacheX m () syncNewTxC ths = @@ -1213,12 +1260,12 @@ txs <- catMaybes <$> mapM (lift . getTxData) ths unless (null txs) $ do forM_ txs $ \tx -> - $(logDebugS) "Cache" $ + logDebug $ "Synchronizing transaction: " <> txHashToHex (txHash tx.tx) importMultiTxC txs syncMempoolC :: - (MonadUnliftIO m, MonadLoggerIO m, StoreReadExtra m) => + (MonadUnliftIO m, StoreReadExtra m) => CacheX m () syncMempoolC = inSync >>= \s -> when s . void . withLock "lock" $ do @@ -1230,26 +1277,26 @@ let tids = HashSet.toList diffset txs <- catMaybes <$> mapM (lift . getTxData) tids unless (null txs) $ do - $(logDebugS) "Cache" $ + logDebug $ "Synchronizing " <> cs (show (length txs)) <> " mempool transactions" importMultiTxC txs -cacheGetMempool :: (MonadLoggerIO m) => CacheX m [(UnixTime, TxHash)] +cacheGetMempool :: (MonadIO m) => CacheX m [(UnixTime, TxHash)] cacheGetMempool = runRedis redisGetMempool -cacheGetHead :: (MonadLoggerIO m) => CacheX m (Maybe BlockHash) +cacheGetHead :: (MonadIO m) => CacheX m (Maybe BlockHash) cacheGetHead = runRedis redisGetHead -cacheSetHead :: (MonadLoggerIO m, StoreReadBase m) => BlockData -> CacheX m () +cacheSetHead :: (MonadIO m) => BlockData -> CacheX m () cacheSetHead bd = do - $(logDebugS) "Cache" $ "Cache head set to: " <> blockHashToHex bh + logDebug $ "Cache head set to: " <> blockHashToHex bh void $ runRedis (redisSetHead bh) withMetrics $ \s -> setGauge s.height (fromIntegral bd.height) where bh = headerHash bd.header cacheGetAddrsInfo :: - (MonadLoggerIO m) => [Address] -> CacheX m [Maybe AddressXPub] + (MonadIO m) => [Address] -> CacheX m [Maybe AddressXPub] cacheGetAddrsInfo as = runRedis (redisGetAddrsInfo as) redisAddToMempool :: (Applicative f, RedisCtx m f) => [TxRef] -> m (f Integer) @@ -1272,7 +1319,7 @@ redisSetAddrInfo a i = void <$> Redis.set (addrPfx <> encode a) (encode i) cacheDelXPubs :: - (MonadLoggerIO m, StoreReadBase m) => + (MonadIO m, StoreReadBase m) => [XPubSpec] -> CacheT m Integer cacheDelXPubs xpubs = ReaderT $ \case @@ -1280,14 +1327,14 @@ Nothing -> return 0 delXPubKeys :: - (MonadLoggerIO m, StoreReadBase m) => + (MonadIO m, StoreReadBase m) => [XPubSpec] -> CacheX m Integer delXPubKeys [] = return 0 delXPubKeys xpubs = do forM_ xpubs $ \x -> do xtxt <- xpubText x - $(logDebugS) "Cache" $ "Deleting xpub: " <> xtxt + logDebug $ "Deleting xpub: " <> xtxt xbals <- runRedis . fmap sequence . forM xpubs $ \xpub -> do bs <- redisGetXPubBalances xpub diff --git a/src/Haskoin/Store/Common.hs b/src/Haskoin/Store/Common.hs --- a/src/Haskoin/Store/Common.hs +++ b/src/Haskoin/Store/Common.hs @@ -36,6 +36,7 @@ microseconds, streamThings, joinDescStreams, + loopForever, ) where @@ -306,6 +307,18 @@ nub' :: (Hashable a) => [a] -> [a] nub' = H.toList . H.fromList + +-- | A 'forever' that stays a flat loop in any monad. 'forever' sequences +-- with @*>@; monad-logger's @LoggingT@ leaves @*>@ at its default +-- @(id <$ a) <*> b@ and @ReaderT@ defines @>>@ as @*>@, so a loop that is +-- polymorphic in its monad and not specialised by GHC keeps an @<*>@ +-- continuation frame and two thunks per iteration for its whole lifetime. +-- Sequencing with @>>=@ only is a tail call in all these stacks. +loopForever :: (Monad m) => m a -> m b +loopForever action = go + where + go = action >>= \_ -> go +{-# INLINE loopForever #-} microseconds :: (MonadIO m) => m Integer microseconds = diff --git a/src/Haskoin/Store/Database/Reader.hs b/src/Haskoin/Store/Database/Reader.hs --- a/src/Haskoin/Store/Database/Reader.hs +++ b/src/Haskoin/Store/Database/Reader.hs @@ -16,6 +16,7 @@ withDB, createDataMetrics, withDatabaseReader, + memoryReport, addrTxCF, addrOutCF, txCF, @@ -23,21 +24,25 @@ blockCF, heightCF, balanceCF, + mempoolCF, ) where import Conduit -import Control.Monad (forM_) +import Control.Monad (forM, forM_) import Control.Monad.Reader import Control.Monad.Trans.Maybe (MaybeT (..), runMaybeT) import Data.Bits ((.&.)) import Data.ByteString qualified as BS +import Data.ByteString.Char8 qualified as BS8 import Data.Default (def) import Data.IntMap.Strict qualified as IntMap import Data.List (sortOn) -import Data.Maybe (fromMaybe) +import Data.Maybe (fromMaybe, mapMaybe) import Data.Ord (Down (..)) import Data.Serialize (encode) +import Data.Text (Text) +import Data.Text qualified as T import Data.Word (Word32) import Database.RocksDB hiding (withDB) import Database.RocksDB.Query @@ -46,6 +51,7 @@ import Haskoin.Store.Data import Haskoin.Store.Database.Types import System.Metrics.StatsD +import Text.Read (readMaybe) import UnliftIO type DatabaseReaderT = ReaderT DatabaseReader @@ -96,7 +102,7 @@ withMetrics go = asks (.metrics) >>= mapM_ go dataVersion :: Word32 -dataVersion = 18 +dataVersion = 19 withDB :: DatabaseReader -> DatabaseReaderT m a -> m a withDB = flip runReaderT @@ -109,10 +115,13 @@ Word32 -> FilePath -> Bool -> + Maybe Int -> + Maybe Int -> + Maybe Int -> Maybe DataMetrics -> DatabaseReaderT m a -> m a -withDatabaseReader net ctx igap gap dir bloom stats m = +withDatabaseReader net ctx igap gap dir bloom maxOpenFiles writeBuffers maxWal stats m = withRunInIO $ \run -> withDBCF dir cfg (columnFamilyConfig bloom) $ \db -> do let bdb = @@ -127,8 +136,47 @@ initRocksDB bdb run $ runReaderT m bdb where - cfg = def {createIfMissing = True} + cfg = + def + { createIfMissing = True, + createMissingColumnFamilies = True, + maxFiles = maxOpenFiles, + dbWriteBufferSize = writeBuffers, + maxTotalWalSize = maxWal + } +-- | Memory that RocksDB uses outside the GHC heap, per column family: +-- memtables and table readers (index and filter blocks of open files), +-- then the block cache shared by all of them. Values are in bytes. +memoryReport :: DB -> IO Text +memoryReport db = do + rows <- forM cfs $ \(name, cf) -> do + mem <- prop cf CurSizeAllMemTables + readers <- prop cf EstimateTableReadersMem + return (name, mem, readers) + cache <- prop Nothing BlockCacheUsage + let total f = sum (mapMaybe f rows) + line (name, mem, readers) = + T.justifyLeft 10 ' ' name + <> T.justifyRight 16 ' ' (num mem) + <> T.justifyRight 16 ' ' (num readers) + return . T.unlines $ + ["** RocksDB memory (bytes) **", T.justifyLeft 10 ' ' "family" <> T.justifyRight 16 ' ' "memtables" <> T.justifyRight 16 ' ' "table readers"] + <> map line rows + <> [ line ("total", Just (total (\(_, m, _) -> m)), Just (total (\(_, _, r) -> r))), + "block cache: " <> num cache, + "" + ] + where + cfs = + ("default", Nothing) + : zip (map (T.pack . fst) (columnFamilyConfig False)) (map Just (columnFamilies db)) + prop cf p = do + v <- maybe (getProperty db p) (\c -> getPropertyCF db c p) cf + return (v >>= readMaybe . BS8.unpack) + num :: Maybe Integer -> Text + num = maybe "?" (T.pack . show) + columnFamilyConfig :: Bool -> [(String, Config)] columnFamilyConfig bloom = [ ("addr-tx", def {prefixLength = Just 22, bloomFilter = bloom}), @@ -138,7 +186,8 @@ ("unspent", def {prefixLength = Just 37, bloomFilter = bloom}), ("block", def {prefixLength = Just 33, bloomFilter = bloom}), ("height", def {prefixLength = Nothing, bloomFilter = bloom}), - ("balance", def {prefixLength = Just 22, bloomFilter = bloom}) + ("balance", def {prefixLength = Just 22, bloomFilter = bloom}), + ("mempool", def {prefixLength = Nothing, bloomFilter = bloom}) ] addrTxCF :: DB -> ColumnFamily @@ -162,14 +211,28 @@ balanceCF :: DB -> ColumnFamily balanceCF db = columnFamilies db !! 7 +mempoolCF :: DB -> ColumnFamily +mempoolCF db = columnFamilies db !! 8 + initRocksDB :: DatabaseReader -> IO () initRocksDB DatabaseReader {db = db} = retrieve db VersionKey >>= \case Just v | v == dataVersion -> return () + | v == 18 -> migrateMempool db | otherwise -> error "Incorrect RocksDB database version" Nothing -> setInitRocksDB db +-- | Version 18 kept the whole mempool as one list under 'MemKey'. Move each +-- transaction to its own key in the mempool column family, in the same +-- batch that sets the new version. +migrateMempool :: DB -> IO () +migrateMempool db = do + mempool <- fromMaybe [] <$> (retrieve db MemKey :: IO (Maybe [(UnixTime, TxHash)])) + writeBatch db $ + [insertOpCF (mempoolCF db) (MemTxKey h) t | (t, h) <- mempool] + <> [deleteOp MemKey, insertOp VersionKey dataVersion] + setInitRocksDB :: DB -> IO () setInitRocksDB db = insert db VersionKey dataVersion @@ -323,7 +386,11 @@ getMempool = do db <- asks (.db) withMetrics $ \s -> incrementCounter s.dataMempoolCount 1 - fromMaybe [] <$> liftIO (retrieve db MemKey) + sortOn Down . mapMaybe entry + <$> liftIO (matchingAsListCF db (mempoolCF db) MemTxKeyS) + where + entry (MemTxKey h, t) = Just (t, h) + entry (MemTxKeyS, _) = Nothing getBestBlock = do withMetrics $ \s -> incrementCounter s.dataBestCount 1 diff --git a/src/Haskoin/Store/Database/Types.hs b/src/Haskoin/Store/Database/Types.hs --- a/src/Haskoin/Store/Database/Types.hs +++ b/src/Haskoin/Store/Database/Types.hs @@ -16,6 +16,7 @@ BalKey (..), HeightKey (..), MemKey (..), + MemTxKey (..), TxKey (..), decodeTxKey, UnspentKey (..), @@ -243,6 +244,29 @@ instance Key MemKey instance KeyValue MemKey [(UnixTime, TxHash)] + +-- | Key of one mempool transaction in the mempool column family, with the +-- time it entered the mempool as its value. Replaces 'MemKey', which kept +-- the whole mempool in a single value. +data MemTxKey + = MemTxKey !TxHash + | MemTxKeyS + deriving (Show, Read, Eq, Ord, Generic, Hashable) + +instance Serialize MemTxKey where + -- 0x08 ยท TxHash + put (MemTxKey h) = do + put MemTxKeyS + put h + -- 0x08 + put MemTxKeyS = putWord8 0x08 + get = do + guard . (== 0x08) =<< getWord8 + MemTxKey <$> get + +instance Key MemTxKey + +instance KeyValue MemTxKey UnixTime -- | Block entry database key. newtype BlockKey = BlockKey BlockHash diff --git a/src/Haskoin/Store/Database/Writer.hs b/src/Haskoin/Store/Database/Writer.hs --- a/src/Haskoin/Store/Database/Writer.hs +++ b/src/Haskoin/Store/Database/Writer.hs @@ -8,15 +8,15 @@ {-# LANGUAGE RecordWildCards #-} {-# LANGUAGE NoFieldSelectors #-} -module Haskoin.Store.Database.Writer (WriterT, runWriter) where +module Haskoin.Store.Database.Writer (WriterT, runWriter, writerLogger) where import Control.Monad.Reader (ReaderT (..)) import Control.Monad.Reader qualified as R +import Data.HashMap.Strict qualified as HashMap import Data.HashTable.IO qualified as H import Data.IntMap.Strict qualified as IntMap import Data.List (sortOn) import Data.Ord (Down (..)) -import Data.Tuple (swap) import Database.RocksDB (BatchOp, DB) import Database.RocksDB.Query ( deleteOp, @@ -36,6 +36,7 @@ headerHash, txHash, ) +import Haskoin.Node (Logger) import Haskoin.Store.Common import Haskoin.Store.Data import Haskoin.Store.Database.Reader @@ -51,11 +52,16 @@ data Writer = Writer { reader :: !DatabaseReader, - memory :: !Memory + memory :: !Memory, + logger :: !Logger } type WriterT = ReaderT Writer +-- | Logger passed to 'runWriter'. +writerLogger :: (Monad m) => WriterT m Logger +writerLogger = R.asks (.logger) + instance (MonadIO m) => StoreReadBase (WriterT m) where getNetwork = R.asks (.reader.net) getCtx = R.asks (.reader.ctx) @@ -84,7 +90,8 @@ type AddrOutTable = H.BasicHashTable (Address, BlockRef, OutPoint) (Maybe OutVal) -type MempoolTable = H.BasicHashTable TxHash UnixTime +-- | Mempool changes in this batch: 'Nothing' for a deletion. +type MempoolTable = H.BasicHashTable TxHash (Maybe UnixTime) data Memory = Memory { net :: !Network, @@ -132,10 +139,10 @@ k = (a, u.block, u.outpoint) addToMempool x t = ReaderT $ \Writer {memory = Memory {..}} -> - liftIO $ H.insert mempoolTable x t + liftIO $ H.insert mempoolTable x (Just t) deleteFromMempool x = ReaderT $ \Writer {memory = Memory {..}} -> - liftIO $ H.delete mempoolTable x + liftIO $ H.insert mempoolTable x Nothing setBalance b = ReaderT $ \Writer {memory = Memory {..}} -> liftIO $ H.insert balanceTable b.address (Just b) @@ -162,12 +169,12 @@ Network -> Ctx -> DatabaseReader -> + Logger -> WriterT m a -> m a -runWriter net ctx bdb@DatabaseReader {db} f = do - mempool <- runReaderT getMempool bdb - hm <- newMemory net ctx mempool - x <- R.runReaderT f Writer {reader = bdb, memory = hm} +runWriter net ctx bdb@DatabaseReader {db} logger f = do + hm <- newMemory net ctx + x <- R.runReaderT f Writer {reader = bdb, memory = hm, logger} ops <- hashMapOps db hm liftIO $ writeBatch db ops return x @@ -183,7 +190,7 @@ balOps db balanceTable, addrTxOps db addressTable, addrOutOps db outputTable, - mempoolOp mempoolTable, + mempoolOps db mempoolTable, unspentOps db unspentTable ] @@ -236,9 +243,11 @@ (addrOutCF db) (AddrOutKey a b p) -mempoolOp :: (MonadIO m) => MempoolTable -> m [BatchOp] -mempoolOp t = - return . insertOp MemKey . sortOn Down . map swap <$> liftIO (H.toList t) +mempoolOps :: (MonadIO m) => DB -> MempoolTable -> m [BatchOp] +mempoolOps db t = map (uncurry f) <$> liftIO (H.toList t) + where + f h (Just u) = insertOpCF (mempoolCF db) (MemTxKey h) u + f h Nothing = deleteOpCF (mempoolCF db) (MemTxKey h) unspentOps :: (MonadIO m) => DB -> UnspentTable -> m [BatchOp] unspentOps db t = map (uncurry f) <$> liftIO (H.toList t) @@ -270,18 +279,24 @@ getUnspentI :: (MonadIO m) => OutPoint -> WriterT m (Maybe Unspent) getUnspentI op = getLayered (getUnspentH op) (getUnspent op) +-- | The mempool in the database with the changes of this batch applied, +-- newest first. getMempoolI :: (MonadIO m) => WriterT m [(UnixTime, TxHash)] getMempoolI = - ReaderT $ \Writer {memory = Memory {..}} -> - liftIO $ map swap <$> H.toList mempoolTable + ReaderT $ \Writer {reader = db, memory = Memory {..}} -> do + stored <- runReaderT getMempool db + changes <- liftIO $ H.toList mempoolTable + let changed = HashMap.fromList changes + kept = filter (not . (`HashMap.member` changed) . snd) stored + added = [(t, h) | (h, Just t) <- changes] + return $ sortOn Down (kept <> added) newMemory :: (MonadIO m) => Network -> Ctx -> - [(UnixTime, TxHash)] -> m Memory -newMemory net ctx mempool = do +newMemory net ctx = do best <- newIORef Nothing blockTable <- liftIO H.new heightTable <- liftIO H.new @@ -290,7 +305,7 @@ balanceTable <- liftIO H.new addressTable <- liftIO H.new outputTable <- liftIO H.new - mempoolTable <- liftIO $ H.fromList (map swap mempool) + mempoolTable <- liftIO H.new return Memory {..} getBestBlockH :: Memory -> IO (Maybe (Maybe BlockHash)) diff --git a/src/Haskoin/Store/Logic.hs b/src/Haskoin/Store/Logic.hs --- a/src/Haskoin/Store/Logic.hs +++ b/src/Haskoin/Store/Logic.hs @@ -6,14 +6,13 @@ {-# LANGUAGE OverloadedRecordDot #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE RecordWildCards #-} -{-# LANGUAGE TemplateHaskell #-} {-# LANGUAGE TupleSections #-} {-# LANGUAGE NoFieldSelectors #-} {-# OPTIONS_GHC -Wno-ambiguous-fields #-} module Haskoin.Store.Logic ( ImportException (..), - MonadImport, + ImportM, initBest, revertBlock, importBlock, @@ -31,12 +30,6 @@ when, zipWithM_, ) -import Control.Monad.Except (MonadError, throwError) -import Control.Monad.Logger - ( MonadLoggerIO (..), - logDebugS, - logErrorS, - ) import qualified Data.ByteString as B import Data.Either (rights) import qualified Data.HashMap.Strict as M @@ -51,6 +44,7 @@ ) import Data.Serialize (encode) import Data.String.Conversions (cs) +import Data.Text (Text) import Data.Word (Word32, Word64) import Haskoin ( Address, @@ -77,17 +71,21 @@ txHash, txHashToHex, ) +import Haskoin.Node (LogLevel (..), logAt) import Haskoin.Store.Common import Haskoin.Store.Data -import UnliftIO (Exception) +import Haskoin.Store.Database.Writer (WriterT, writerLogger) +import UnliftIO (Exception, throwIO) -type MonadImport m = - ( MonadError ImportException m, - MonadLoggerIO m, - StoreReadBase m, - StoreWrite m - ) +-- | Block and transaction import. Failures are thrown as 'ImportException'. +-- Writes are buffered and committed together when 'runWriter' finishes, so a +-- failed batch commits nothing. +type ImportM = WriterT IO +logDebug, logError :: Text -> ImportM () +logDebug msg = writerLogger >>= \l -> logAt LevelDebug l msg +logError msg = writerLogger >>= \l -> logAt LevelError l msg + data ImportException = PrevBlockNotBest | Orphan @@ -115,94 +113,96 @@ show DuplicatePrevOutput = "Duplicate previous output" show TxSpent = "Transaction is spent" -initBest :: (MonadImport m) => m () +initBest :: ImportM () initBest = do - ctx <- getCtx - $(logDebugS) "BlockStore" "Initializing best block" + logDebug "Initializing best block" net <- getNetwork m <- getBestBlock when (isNothing m) . void $ do - $(logDebugS) "BlockStore" "Importing Genesis block" - importBlock (genesisBlock net ctx) (genesisNode net) + logDebug "Importing Genesis block" + importBlock (genesisBlock net) (genesisNode net) -newMempoolTx :: (MonadImport m) => Tx -> UnixTime -> m Bool +newMempoolTx :: Tx -> UnixTime -> ImportM Bool newMempoolTx tx w = getActiveTxData (txHash tx) >>= \case Just _ -> return False Nothing -> do - freeOutputs True True tx + -- Check everything before writing, so that a rejected transaction + -- leaves the mempool as it was. + replaced <- getConflicts True True tx + checkNewTx (map txHash replaced) tx + mapM_ (deleteSingleTx . txHash) replaced rbf <- isRBF (MemRef w) tx - checkNewTx tx _ <- importTx (MemRef w) w rbf tx return True -bestBlockData :: (MonadImport m) => m BlockData +bestBlockData :: ImportM BlockData bestBlockData = do h <- getBestBlock >>= \case Nothing -> do - $(logErrorS) "BlockStore" "Best block unknown" - throwError BestBlockNotFound + logError "Best block unknown" + throwIO BestBlockNotFound Just h -> return h getBlock h >>= \case Nothing -> do - $(logErrorS) "BlockStore" "Best block not found" - throwError BestBlockNotFound + logError "Best block not found" + throwIO BestBlockNotFound Just b -> return b -revertBlock :: (MonadImport m) => BlockHash -> m () +revertBlock :: BlockHash -> ImportM () revertBlock bh = do bd <- bestBlockData >>= \b -> if headerHash b.header == bh then return b else do - $(logErrorS) "BlockStore" $ + logError $ "Cannot revert non-head block: " <> blockHashToHex bh - throwError BlockNotBest - $(logDebugS) "BlockStore" $ + throwIO BlockNotBest + logDebug $ "Obtained block data for " <> blockHashToHex bh tds <- mapM getImportTxData bd.txs - $(logDebugS) "BlockStore" $ + logDebug $ "Obtained import tx data for block " <> blockHashToHex bh setBest bd.header.prev - $(logDebugS) "BlockStore" $ + logDebug $ "Set parent as best block " <> blockHashToHex bd.header.prev insertBlock bd {main = False} - $(logDebugS) "BlockStore" $ + logDebug $ "Updated as not in main chain: " <> blockHashToHex bh forM_ (drop 1 tds) unConfirmTx - $(logDebugS) "BlockStore" $ + logDebug $ "Unconfirmed " <> cs (show (length tds)) <> " transactions" deleteConfirmedTx (txHash (tds !! 0).tx) - $(logDebugS) "BlockStore" $ + logDebug $ "Deleted coinbase: " <> txHashToHex (txHash (tds !! 0).tx) -checkNewBlock :: (MonadImport m) => Block -> BlockNode -> m () +checkNewBlock :: Block -> BlockNode -> ImportM () checkNewBlock b n = getBestBlock >>= \case Nothing | isGenesis n -> return () | otherwise -> do - $(logErrorS) "BlockStore" $ + logError $ "Cannot import non-genesis block: " <> blockHashToHex (headerHash b.header) - throwError BestBlockNotFound + throwIO BestBlockNotFound Just h | b.header.prev == h -> return () | otherwise -> do - $(logErrorS) "BlockStore" $ + logError $ "Block does not build on head: " <> blockHashToHex (headerHash b.header) - throwError PrevBlockNotBest + throwIO PrevBlockNotBest -importOrConfirm :: (MonadImport m) => BlockNode -> [Tx] -> m [TxData] +importOrConfirm :: BlockNode -> [Tx] -> ImportM [TxData] importOrConfirm bn txns = do - $(logDebugS) "BlockStore" "Freeing outputs..." + logDebug "Freeing outputs..." mapM_ (freeOutputs True False . snd) (reverse txs) - $(logDebugS) "BlockStore" "Outputs freed" + logDebug "Outputs freed" mapM (uncurry action) txs where txs = sortTxs txns @@ -215,31 +215,31 @@ confirm_it i tx = getActiveTxData (txHash tx) >>= \case Just t -> do - $(logDebugS) "BlockStore" $ + logDebug $ "Confirming tx: " <> txHashToHex (txHash tx) confirmTx t (br i) Nothing -> do - $(logErrorS) "BlockStore" $ + logError $ "Cannot find tx to confirm: " <> txHashToHex (txHash tx) - throwError TxNotFound + throwIO TxNotFound import_it i tx = do - $(logDebugS) "BlockStore" $ + logDebug $ "Importing tx: " <> txHashToHex (txHash tx) importTx (br i) bn_time False tx -importBlock :: (MonadImport m) => Block -> BlockNode -> m (BlockData, [TxData]) +importBlock :: Block -> BlockNode -> ImportM (BlockData, [TxData]) importBlock b n = do - $(logDebugS) "BlockStore" $ + logDebug $ "Checking new block: " <> blockHashToHex (headerHash n.header) checkNewBlock b n - $(logDebugS) "BlockStore" "Passed check" + logDebug "Passed check" net <- getNetwork let subsidy = computeSubsidy net n.height bs <- getBlocksAtHeight n.height - $(logDebugS) "BlockStore" $ + logDebug $ "Inserting block entries for: " <> blockHashToHex (headerHash n.header) setBlocksAtHeight @@ -261,7 +261,7 @@ outputs = ts_out_val } insertBlock bd - $(logDebugS) "BlockStore" $ + logDebug $ "Finished importing block: " <> blockHashToHex (headerHash n.header) return (bd, tds) @@ -275,33 +275,35 @@ s = B.length (encode b') in fromIntegral $ s * 3 + x -checkNewTx :: (MonadImport m) => Tx -> m () -checkNewTx tx = do +-- | Check a transaction that will replace these ones. Outputs that they spend +-- count as unspent, and outputs that they create as missing. +checkNewTx :: [TxHash] -> Tx -> ImportM () +checkNewTx replaced tx = do when (unique_inputs < length tx.inputs) $ do - $(logErrorS) "BlockStore" $ + logError $ "Transaction spends same output twice: " <> txHashToHex (txHash tx) - throwError DuplicatePrevOutput - us <- getUnspentOutputs tx - when (any isNothing us) $ do - $(logErrorS) "BlockStore" $ + throwIO DuplicatePrevOutput + prevs <- mapM prev_output (prevOuts tx) + when (any isNothing prevs) $ do + logError $ "Orphan: " <> txHashToHex (txHash tx) - throwError Orphan + throwIO Orphan when (isCoinbaseTx tx) $ do - $(logErrorS) "BlockStore" $ + logError $ "Coinbase cannot be imported into mempool: " <> txHashToHex (txHash tx) - throwError UnexpectedCoinbase - when (length (prevOuts tx) > length us) $ do - $(logErrorS) "BlockStore" $ - "Orphan: " <> txHashToHex (txHash tx) - throwError Orphan - when (outputs > unspents us) $ do - $(logErrorS) "BlockStore" $ + throwIO UnexpectedCoinbase + when (outputs > sum (map (.value) (catMaybes prevs))) $ do + logError $ "Insufficient funds for tx: " <> txHashToHex (txHash tx) - throwError InsufficientFunds + throwIO InsufficientFunds where - unspents = sum . map (.value) . catMaybes + prev_output op + | op.hash `elem` replaced = return Nothing + | otherwise = do + parent <- getActiveTxData op.hash + return (parent >>= getTxOut op.index . (.tx)) outputs = sum (map (.value) tx.outputs) unique_inputs = length (nub' (map (.outpoint) tx.inputs)) @@ -324,34 +326,33 @@ ps = I.fromList $ zip [0 ..] $ map mkprv us importTx :: - (MonadImport m) => BlockRef -> -- | unix time Word64 -> -- | RBF Bool -> Tx -> - m TxData + ImportM TxData importTx br tt rbf tx = do mus <- getUnspentOutputs tx us <- forM mus $ \case Nothing -> do - $(logErrorS) "BlockStore" $ + logError $ "Attempted to import a tx missing UTXO: " <> txHashToHex (txHash tx) - throwError Orphan + throwIO Orphan Just u -> return u let td = prepareTxData rbf br tt tx us commitAddTx td return td -unConfirmTx :: (MonadImport m) => TxData -> m TxData +unConfirmTx :: TxData -> ImportM TxData unConfirmTx t = confTx t Nothing -confirmTx :: (MonadImport m) => TxData -> BlockRef -> m TxData +confirmTx :: TxData -> BlockRef -> ImportM TxData confirmTx t br = confTx t (Just br) -replaceAddressTx :: (MonadImport m) => TxData -> BlockRef -> m () +replaceAddressTx :: TxData -> BlockRef -> ImportM () replaceAddressTx t new = do ctx <- getCtx forM_ (txDataAddresses ctx t) $ \a -> do @@ -369,12 +370,11 @@ } adjustAddressOutput :: - (MonadImport m) => OutPoint -> TxOut -> BlockRef -> BlockRef -> - m () + ImportM () adjustAddressOutput op o old new = do let pk = o.script getUnspent op >>= \case @@ -420,7 +420,7 @@ decreaseBalance (confirmed old) a o.value increaseBalance (confirmed new) a o.value -confTx :: (MonadImport m) => TxData -> Maybe BlockRef -> m TxData +confTx :: TxData -> Maybe BlockRef -> ImportM TxData confTx t mbr = do replaceAddressTx t new forM_ (zip [0 ..] t.tx.outputs) $ \(n, o) -> do @@ -430,51 +430,72 @@ let td = (t :: TxData) {block = new, rbf = rbf} insertTx td updateMempool td + updateSpendersRBF td return td where new = fromMaybe (MemRef t.timestamp) mbr old = t.block -freeOutputs :: - (MonadImport m) => +-- | Spenders inherit BIP125 signalling from unconfirmed parents only, so their +-- flag can change when a parent is confirmed or unconfirmed. +updateSpendersRBF :: TxData -> ImportM () +updateSpendersRBF td = + forM_ (nub' (map (.txid) (I.elems td.spenders))) $ \h -> + getActiveTxData h >>= mapM_ update + where + update s = do + rbf <- isRBF s.block s.tx + when (rbf /= s.rbf) $ do + let s' = (s :: TxData) {rbf = rbf} + insertTx s' + updateSpendersRBF s' + +freeOutputs :: Bool -> Bool -> Tx -> ImportM () +freeOutputs memonly rbfcheck tx = + getConflicts memonly rbfcheck tx >>= mapM_ (deleteSingleTx . txHash) + +-- | Transactions that spend the same outputs as this one, with their +-- descendants, in an order in which they can be deleted. +getConflicts :: -- | only delete transaction if unconfirmed Bool -> - -- | only delete RBF + -- | only delete if network has RBF Bool -> Tx -> - m () -freeOutputs memonly rbfcheck tx = do - $(logDebugS) "BlockStore" $ - "Freeing outputs for tx " <> txHashToHex (txHash tx) <> "..." + ImportM [Tx] +getConflicts memonly rbfcheck tx = do + logDebug $ + "Getting conflicts for tx " <> txHashToHex (txHash tx) <> "..." let prevs = prevOuts tx unspents <- mapM getUnspent prevs let spents = [p | (p, Nothing) <- zip prevs unspents] spndrs <- catMaybes <$> mapM getSpender spents let txids = H.fromList $ filter (/= txHash tx) $ map (.txid) spndrs - mapM_ (deleteTx memonly rbfcheck) $ H.toList txids + chains <- mapM (getChain memonly rbfcheck) (H.toList txids) + return . reverse . map snd . sortTxs . M.elems $ + M.fromList [(txHash t, t) | t <- concat chains] -deleteConfirmedTx :: (MonadImport m) => TxHash -> m () +deleteConfirmedTx :: TxHash -> ImportM () deleteConfirmedTx = deleteTx False False -deleteUnconfirmedTx :: (MonadImport m) => Bool -> TxHash -> m () +deleteUnconfirmedTx :: Bool -> TxHash -> ImportM () deleteUnconfirmedTx rbfcheck th = getActiveTxData th >>= \case Just _ -> deleteTx True rbfcheck th Nothing -> - $(logDebugS) "BlockStore" $ + logDebug $ "Not found or already deleted: " <> txHashToHex th deleteTx :: - (MonadImport m) => -- | only delete transaction if unconfirmed Bool -> - -- | only delete RBF + -- | only delete if network has RBF Bool -> TxHash -> - m () + ImportM () deleteTx memonly rbfcheck th = do chain <- getChain memonly rbfcheck th - $(logDebugS) "BlockStore" $ + logDebug $ "Deleting " <> cs (show (length chain)) <> " txs from chain leading to " @@ -482,15 +503,14 @@ mapM_ (\t -> let h = txHash t in deleteSingleTx h >> return h) chain getChain :: - (MonadImport m) => -- | only delete transaction if unconfirmed Bool -> - -- | only delete RBF + -- | only delete if network has RBF Bool -> TxHash -> - m [Tx] + ImportM [Tx] getChain memonly rbfcheck th' = do - $(logDebugS) "BlockStore" $ + logDebug $ "Getting chain for tx " <> txHashToHex th' sort_clean <$> go M.empty (H.singleton th') where @@ -498,23 +518,24 @@ get_tx th = getActiveTxData th >>= \case Nothing -> do - $(logDebugS) "BlockStore" $ + logDebug $ "Transaction not found: " <> txHashToHex th return Nothing Just td | memonly && confirmed td.block -> do - $(logErrorS) "BlockStore" $ + logError $ "Transaction already confirmed: " <> txHashToHex th - throwError TxConfirmed + throwIO TxConfirmed | rbfcheck -> - isRBF td.block td.tx >>= \case - True -> return $ Just td - False -> do - $(logErrorS) "BlockStore" $ - "Double-spending transaction: " - <> txHashToHex th - throwError DoubleSpend + getNetwork >>= \net -> + if net.replaceByFee + then return $ Just td + else do + logError $ + "Double-spending transaction: " + <> txHashToHex th + throwIO DoubleSpend | otherwise -> return $ Just td go txm pdg = do let ths = filter (not . (`M.member` txm)) (H.toList pdg) @@ -525,7 +546,7 @@ txm' = txmn <> txm if H.null pdg' then do - $(logDebugS) "BlockStore" $ + logDebug $ "Chain for tx " <> txHashToHex th' <> " contains " @@ -534,32 +555,32 @@ return $ M.elems txm' else go txm' pdg' -deleteSingleTx :: (MonadImport m) => TxHash -> m () +deleteSingleTx :: TxHash -> ImportM () deleteSingleTx th = getActiveTxData th >>= \case Nothing -> do - $(logErrorS) "BlockStore" $ + logError $ "Already deleted: " <> txHashToHex th - throwError TxNotFound + throwIO TxNotFound Just td -> if I.null td.spenders then do - $(logDebugS) "BlockStore" $ + logDebug $ "Deleting tx: " <> txHashToHex th commitDelTx td else do - $(logErrorS) "BlockStore" $ + logError $ "Tried to delete spent tx: " <> txHashToHex th - throwError TxSpent + throwIO TxSpent -commitDelTx :: (MonadImport m) => TxData -> m () +commitDelTx :: TxData -> ImportM () commitDelTx = commitModTx False -commitAddTx :: (MonadImport m) => TxData -> m () +commitAddTx :: TxData -> ImportM () commitAddTx = commitModTx True -commitModTx :: (MonadImport m) => Bool -> TxData -> m () +commitModTx :: Bool -> TxData -> ImportM () commitModTx add tx_data = do ctx <- getCtx mapM_ mod_addr_tx (txDataAddresses ctx td) @@ -586,7 +607,7 @@ | add = addOutputs br tx | otherwise = delOutputs br tx -updateMempool :: (MonadImport m) => TxData -> m () +updateMempool :: TxData -> ImportM () updateMempool td@TxData {deleted = True} = deleteFromMempool (txHash td.tx) updateMempool td@TxData {block = MemRef t} = @@ -594,11 +615,11 @@ updateMempool td@TxData {block = BlockRef {}} = deleteFromMempool (txHash td.tx) -spendOutputs :: (MonadImport m) => Tx -> m () +spendOutputs :: Tx -> ImportM () spendOutputs tx = zipWithM_ (spendOutput (txHash tx)) [0 ..] (prevOuts tx) -addOutputs :: (MonadImport m) => BlockRef -> Tx -> m () +addOutputs :: BlockRef -> Tx -> ImportM () addOutputs br tx = zipWithM_ (addOutput br . OutPoint (txHash tx)) [0 ..] tx.outputs @@ -630,19 +651,18 @@ | otherwise -> ck hs' in ck hs -addOutput :: (MonadImport m) => BlockRef -> OutPoint -> TxOut -> m () +addOutput :: BlockRef -> OutPoint -> TxOut -> ImportM () addOutput = modOutput True -delOutput :: (MonadImport m) => BlockRef -> OutPoint -> TxOut -> m () +delOutput :: BlockRef -> OutPoint -> TxOut -> ImportM () delOutput = modOutput False modOutput :: - (MonadImport m) => Bool -> BlockRef -> OutPoint -> TxOut -> - m () + ImportM () modOutput add br op o = do ctx <- getCtx mod_unspent ctx @@ -670,18 +690,18 @@ | add = insertAddrUnspent | otherwise = deleteAddrUnspent -delOutputs :: (MonadImport m) => BlockRef -> Tx -> m () +delOutputs :: BlockRef -> Tx -> ImportM () delOutputs br tx = forM_ (zip [0 ..] tx.outputs) $ \(i, o) -> do let op = OutPoint (txHash tx) i delOutput br op o -getImportTxData :: (MonadImport m) => TxHash -> m TxData +getImportTxData :: TxHash -> ImportM TxData getImportTxData th = getActiveTxData th >>= \case Nothing -> do - $(logDebugS) "BlockStore" $ "Tx not found: " <> txHashToHex th - throwError TxNotFound + logDebug $ "Tx not found: " <> txHashToHex th + throwIO TxNotFound Just d -> return d getTxOut :: Word32 -> Tx -> Maybe TxOut @@ -689,7 +709,7 @@ guard (fromIntegral i < length tx.outputs) return $ tx.outputs !! fromIntegral i -insertSpender :: (MonadImport m) => OutPoint -> Spender -> m () +insertSpender :: OutPoint -> Spender -> ImportM () insertSpender op s = do td <- getImportTxData op.hash let p = td.spenders @@ -697,7 +717,7 @@ td' = (td :: TxData) {spenders = p'} insertTx td' -deleteSpender :: (MonadImport m) => OutPoint -> m () +deleteSpender :: OutPoint -> ImportM () deleteSpender op = do td <- getImportTxData op.hash let p = td.spenders @@ -705,7 +725,7 @@ td' = (td :: TxData) {spenders = p'} insertTx td' -spendOutput :: (MonadImport m) => TxHash -> Word32 -> OutPoint -> m () +spendOutput :: TxHash -> Word32 -> OutPoint -> ImportM () spendOutput th ix op = do u <- getUnspent op >>= \case @@ -719,10 +739,10 @@ decreaseBalance (confirmed u.block) a u.value deleteAddrUnspent a u -unspendOutputs :: (MonadImport m) => Tx -> m () +unspendOutputs :: Tx -> ImportM () unspendOutputs = mapM_ unspendOutput . prevOuts -unspendOutput :: (MonadImport m) => OutPoint -> m () +unspendOutput :: OutPoint -> ImportM () unspendOutput op = do ctx <- getCtx t <- @@ -748,26 +768,25 @@ insertAddrUnspent a u increaseBalance (confirmed u.block) a o.value -modifyReceived :: (MonadImport m) => Address -> (Word64 -> Word64) -> m () +modifyReceived :: Address -> (Word64 -> Word64) -> ImportM () modifyReceived a f = do b <- getDefaultBalance a setBalance b {received = f b.received} -decreaseBalance :: (MonadImport m) => Bool -> Address -> Word64 -> m () +decreaseBalance :: Bool -> Address -> Word64 -> ImportM () decreaseBalance conf = modBalance conf False -increaseBalance :: (MonadImport m) => Bool -> Address -> Word64 -> m () +increaseBalance :: Bool -> Address -> Word64 -> ImportM () increaseBalance conf = modBalance conf True modBalance :: - (MonadImport m) => -- | confirmed Bool -> -- | add Bool -> Address -> Word64 -> - m () + ImportM () modBalance conf add a val = do b <- getDefaultBalance a setBalance $ (g . f) b @@ -780,7 +799,7 @@ | add = (+) | otherwise = subtract -modAddressCount :: (MonadImport m) => Bool -> Address -> m () +modAddressCount :: Bool -> Address -> ImportM () modAddressCount add a = do b <- getDefaultBalance a setBalance b {txs = f b.txs} 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 @@ -2,6 +2,7 @@ {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE NamedFieldPuns #-} {-# LANGUAGE OverloadedRecordDot #-} +{-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE NoFieldSelectors #-} module Haskoin.Store.Manager @@ -11,15 +12,13 @@ ) where -import Control.Monad (forever, unless) +import Control.Monad (unless) import Control.Monad.Cont ( ContT (..), runContT, ) -import Control.Monad.Logger (MonadLoggerIO) import Control.Monad.Reader (ReaderT (ReaderT), runReaderT) import Control.Monad.Trans (lift) -import Data.Serialize (decode) import Data.Time.Clock (NominalDiffTime) import Data.Word (Word32) import Haskoin @@ -42,6 +41,7 @@ import Haskoin.Node ( Chain, ChainEvent (..), + Logger, Node (..), NodeConfig (..), NodeEvent (..), @@ -49,6 +49,7 @@ PeerMgr, WithConnection, withNode, + withSource, ) import Haskoin.Store.BlockStore ( BlockStore, @@ -74,6 +75,7 @@ ) import Haskoin.Store.Common ( StoreEvent (..), + loopForever, ) import Haskoin.Store.Database.Reader ( DatabaseReader (..), @@ -146,9 +148,11 @@ wipeMempool :: !Bool, -- | sync mempool from peers syncMempool :: !Bool, - -- | disconnect peer if message not received for this many seconds - mempoolTimeout :: !Int, + -- | blocks to request from the syncing peer in one batch + blockWindow :: !Int, -- | delete mempool transactions older than this number of days + mempoolTimeout :: !Int, + -- | disconnect peer if message not received for this many seconds peerTimeout :: !NominalDiffTime, -- | disconnect peer if it has been connected this long maxPeerLife :: !NominalDiffTime, @@ -159,11 +163,22 @@ -- | sync mempool against cache every this many seconds redisSyncInterval :: !Int, -- | database Bloom filters - bloom :: !Bool + bloom :: !Bool, + -- | RocksDB max_open_files ('Nothing' = unlimited: every SST file's index + -- and filter blocks stay resident for the life of the process) + maxOpenFiles :: !(Maybe Int), + -- | bytes of RocksDB memtables across all column families ('Nothing' = + -- unlimited: each column family keeps up to two 64 MiB memtables) + writeBufferSize :: !(Maybe Int), + -- | bytes of write-ahead log after which the column families holding the + -- oldest log file are flushed ('Nothing' = RocksDB default) + maxWalSize :: !(Maybe Int), + -- | log messages here + logger :: !Logger } withStore :: - (MonadLoggerIO m, MonadUnliftIO m) => + (MonadUnliftIO m) => StoreConfig -> (Store -> m a) -> m a @@ -205,6 +220,9 @@ cfg.gap cfg.db cfg.bloom + cfg.maxOpenFiles + cfg.writeBufferSize + cfg.maxWalSize stats f @@ -225,9 +243,11 @@ noMempool = cfg.noMempool, wipeMempool = cfg.wipeMempool, syncMempool = cfg.syncMempool, + blockWindow = cfg.blockWindow, mempoolTimeout = cfg.mempoolTimeout, peerTimeout = cfg.peerTimeout, - stats = cfg.stats + stats = cfg.stats, + logger = withSource "BlockStore" cfg.logger } nodeCfg :: @@ -250,11 +270,12 @@ net = cfg.net, timeout = cfg.peerTimeout, maxPeerLife = cfg.maxPeerLife, - connect = cfg.connect + connect = cfg.connect, + logger = cfg.logger } withCache :: - (MonadUnliftIO m, MonadLoggerIO m) => + (MonadUnliftIO m) => StoreConfig -> Chain -> DatabaseReader -> @@ -267,7 +288,7 @@ action Nothing Just redisurl -> do conn <- connectRedis redisurl - metrics <- mapM (\s -> newCacheMetrics s conn db) cfg.stats + metrics <- mapM (\s -> newCacheMetrics logger s conn db) cfg.stats withSubscription pub $ \evts -> let conf = c conn metrics in withProcess (f conf) $ \p -> @@ -278,13 +299,15 @@ (action (Just conf)) where f conf cwinbox = runReaderT (cacheWriter conf cwinbox) db + logger = withSource "Cache" cfg.logger c conn metrics = CacheConfig { redis = conn, minAddrs = cfg.redisMinAddrs, chain = chain, maxKeys = cfg.redisMaxKeys, - metrics = metrics + metrics = metrics, + logger = logger } cacheWriterProcesses :: @@ -300,11 +323,11 @@ cacheWriterEvents :: (MonadUnliftIO m) => Int -> Inbox StoreEvent -> CacheWriter -> m () cacheWriterEvents interval evts cwm = withAsync mempool . const $ - forever $ + loopForever $ receive evts >>= \e -> e `cacheWriterDispatch` cwm where - mempool = forever $ do + mempool = loopForever $ do threadDelay (interval * 1000 * 1000) cacheSyncMempool cwm @@ -321,7 +344,7 @@ Inbox NodeEvent -> m () nodeForwarder b pub sub = - forever $ receive sub >>= atomically . storeDispatch b pub + loopForever $ receive sub >>= atomically . storeDispatch b pub -- | Dispatcher of node events. storeDispatch :: @@ -360,17 +383,10 @@ ] publishSTM (StoreTxAnnounce p txs) pub unless (null txs) $ blockStoreTxHashSTM p txs b - MReject Reject {message = MCTx, extra, code, reason} -> - case decode extra of - Left _ -> return () - Right th -> - let reject = - StoreTxReject - p - th - code - reason.get - in publishSTM reject pub + -- A reject that names no transaction hash cannot be attributed to a + -- transaction and falls through to the catch-all below. + MReject Reject {message = MCTx, extra = Just h, code, reason} -> + publishSTM (StoreTxReject p (TxHash h) code reason.get) pub _ -> return () storeDispatch b _ (ChainEvent (ChainBestBlock bn)) = blockStoreHeadSTM bn 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 @@ -10,7 +10,6 @@ {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE RecordWildCards #-} {-# LANGUAGE ScopedTypeVariables #-} -{-# LANGUAGE TemplateHaskell #-} {-# LANGUAGE TupleSections #-} {-# LANGUAGE NoFieldSelectors #-} {-# OPTIONS_GHC -Wno-deprecations #-} @@ -44,17 +43,11 @@ import Control.Lens ((.~), (^.)) import Control.Monad ( forM_, - forever, unless, void, when, (<=<), ) -import Control.Monad.Logger - ( MonadLoggerIO, - logDebugS, - logErrorS, - ) import Control.Monad.Reader ( MonadReader, ReaderT, @@ -87,8 +80,7 @@ ( encodingToLazyByteString, list, ) -import Data.Base16.Types (assertBase16) -import Data.ByteString.Base16 (decodeBase16, isBase16) +import Data.ByteString.Base16 (decodeBase16, parseBase16) import Data.ByteString.Builder (lazyByteString) import Data.ByteString.Char8 qualified as C import Data.ByteString.Lazy qualified as L @@ -132,12 +124,16 @@ import Haskoin.Network import Haskoin.Node ( Chain, + Logger, OnlinePeer (..), PeerMgr, chainGetAncestor, chainGetBest, getPeers, + logDebug, + logError, sendMessage, + withSource, ) import Haskoin.Script import Haskoin.Store.BlockStore @@ -150,9 +146,13 @@ import Haskoin.Transaction import Haskoin.Util import NQE - ( Inbox, - receive, - withSubscription, + ( Publisher, + mailboxFullSTM, + newBoundedInbox, + receiveSTM, + sendSTM, + subscribe, + unsubscribe, ) import Network.HTTP.Types ( Status (..), @@ -186,9 +186,15 @@ sendTextData, ) import Network.WebSockets qualified as WebSockets +import Network.WebSockets.Connection.PingPong + ( PingPongOptions (..), + defaultPingPongOptions, + withPingPong, + ) import Network.Wreq qualified as Wreq import Network.Wreq.Session as Wreq (Session) import Network.Wreq.Session qualified as Wreq.Session +import Numeric.Natural (Natural) import System.Metrics.StatsD import UnliftIO ( MonadIO, @@ -199,9 +205,13 @@ atomically, bracket, bracket_, + checkSTM, handleAny, liftIO, newTVarIO, + orElse, + race_, + readTVar, readTVarIO, withAsync, writeTVar, @@ -254,7 +264,8 @@ priceHistoryURL :: !String, noXPub :: !Bool, noBlockchainInfo :: !Bool, - healthCheckInterval :: !Int + healthCheckInterval :: !Int, + logger :: !Logger } data WebState = WebState @@ -450,7 +461,7 @@ getMempool = runInWebReader getMempool instance - (MonadUnliftIO m, MonadLoggerIO m) => + (MonadUnliftIO m) => StoreReadExtra (ReaderT WebState m) where getMaxGap = runInWebReader getMaxGap @@ -478,7 +489,7 @@ getBalance = lift . getBalance getMempool = lift getMempool -instance (MonadUnliftIO m, MonadLoggerIO m) => StoreReadExtra (ActionT m) where +instance (MonadUnliftIO m) => StoreReadExtra (ActionT m) where getBalances = lift . getBalances getAddressesTxs as = lift . getAddressesTxs as getAddressTxs a = lift . getAddressTxs a @@ -496,7 +507,7 @@ -- Path Handlers -- ------------------- -runWeb :: (MonadUnliftIO m, MonadLoggerIO m) => WebConfig -> m () +runWeb :: (MonadUnliftIO m) => WebConfig -> m () runWeb config = do ticker <- newTVarIO HashMap.empty metrics <- mapM createMetrics config.stats @@ -505,37 +516,38 @@ let state = WebState {..} withAsync (priceUpdater session ticker) $ \_a1 -> withAsync (healthCheckLoop health) $ \_a2 -> do - logger <- logIt runner <- askRunInIO S.scottyOptsT opts (runner . flip runReaderT state) $ do S.middleware $ webSocketEvents state - S.middleware logger + S.middleware (logIt config.logger) handlePaths config S.notFound $ raise ThingNotFound where priceUpdater session = unless config.noBlockchainInfo . price + (withSource "Price" config.logger) config.store.net session config.tickerURL config.tickerRefresh runHealthCheck = runReaderT (healthCheck config) config.store.db - healthCheckLoop v = forever $ do + healthCheckLoop v = loopForever $ do threadDelay (config.healthCheckInterval * 1000 * 1000) runHealthCheck >>= atomically . writeTVar v opts = S.defaultOptions {S.settings = settings defaultSettings} settings = setPort config.port . setHost (fromString config.host) getRates :: - (MonadUnliftIO m, MonadLoggerIO m) => + (MonadUnliftIO m) => + Logger -> Network -> Wreq.Session -> String -> Text -> [Word64] -> m [BinfoRate] -getRates net session url currency times = do +getRates logger net session url currency times = do handleAny err $ do r <- liftIO $ @@ -544,7 +556,7 @@ return $ r ^. Wreq.responseBody where err _ = do - $(logErrorS) "Web" "Could not get historic prices" + logError logger "Could not get historic prices" return [] body = toJSON times base = @@ -553,15 +565,16 @@ opts = base & Wreq.param "quote" .~ [currency] price :: - (MonadUnliftIO m, MonadLoggerIO m) => + (MonadUnliftIO m) => + Logger -> Network -> Wreq.Session -> String -> Int -> TVar (HashMap Text BinfoTicker) -> m () -price net session url pget v = forM_ purl $ \u -> forever $ do - let err e = $(logErrorS) "Price" $ cs (show e) +price logger net session url pget v = forM_ purl $ \u -> loopForever $ do + let err e = logError logger $ cs (show e) handleAny err $ do r <- liftIO $ Wreq.asJSON =<< Wreq.Session.get session u atomically . writeTVar v $ r ^. Wreq.responseBody @@ -601,7 +614,7 @@ errStatus RequestTooLarge = status413 handlePaths :: - (MonadUnliftIO m, MonadLoggerIO m) => + (MonadUnliftIO m) => WebConfig -> S.ScottyT (ReaderT WebState m) () handlePaths cfg = do @@ -1021,7 +1034,7 @@ -- GET BlockTime / BlockTimeRaw -- scottyBlockTime :: - (MonadUnliftIO m, MonadLoggerIO m) => + (MonadUnliftIO m) => GetBlockTime -> ActionT m BlockData scottyBlockTime (GetBlockTime (TimeParam t) (NoTx notx)) = @@ -1032,7 +1045,7 @@ Just b -> return $ pruneTx notx b scottyBlockMTP :: - (MonadUnliftIO m, MonadLoggerIO m) => + (MonadUnliftIO m) => GetBlockMTP -> ActionT m BlockData scottyBlockMTP (GetBlockMTP (TimeParam t) (NoTx notx)) = @@ -1043,7 +1056,7 @@ Just b -> return $ pruneTx notx b scottyBlockTimeRaw :: - (MonadUnliftIO m, MonadLoggerIO m) => + (MonadUnliftIO m) => GetBlockTimeRaw -> ActionT m (RawResult H.Block) scottyBlockTimeRaw (GetBlockTimeRaw (TimeParam t)) = @@ -1054,7 +1067,7 @@ Just b -> RawResult <$> lift (toRawBlock b) scottyBlockMTPRaw :: - (MonadUnliftIO m, MonadLoggerIO m) => + (MonadUnliftIO m) => GetBlockMTPRaw -> ActionT m (RawResult H.Block) scottyBlockMTPRaw (GetBlockMTPRaw (TimeParam t)) = @@ -1209,19 +1222,30 @@ websocketsOr defaultConnectionOptions (wrap . events) where pub = s.config.store.pub + pingPong = defaultPingPongOptions {pingInterval = 30, pongTimeout = 60} gauge = (.events) <$> s.metrics wrap f = case gauge of Nothing -> f Just g -> withGaugeIO g f - events pending = withSubscription pub $ \sub -> do + events pending = do let path = requestPath $ pendingRequest pending if path == "/events" - then do + then withEventSubscription pub $ \next -> do conn <- acceptRequest pending - forever $ - receiveEvent sub >>= \case - Nothing -> return () - Just event -> sendTextData conn (A.encode event) + -- Drop clients that stop answering pings, and clients that fall + -- so far behind that their event queue fills up. + -- websockets only sees pongs while reading, hence the reader. + withPingPong pingPong conn $ \c -> + let send_events = + next >>= \case + Nothing -> + WebSockets.sendClose c ("Event queue full" :: Text) + Just event -> do + sendTextData c (A.encode event) + send_events + in race_ + (loopForever (WebSockets.receiveDataMessage c)) + send_events else rejectRequestWith pending @@ -1243,12 +1267,14 @@ proto <- setupContentType False pub <- askl (.config.store.pub) S.stream $ \io flush' -> - withSubscription pub $ \sub -> - forever $ do - flush' - receiveEvent sub >>= \case - Nothing -> return () - Just msg -> io (serial proto msg) + withEventSubscription pub $ \next -> + -- ends the response when the client falls too far behind + let send_events = do + flush' + next >>= \case + Nothing -> return () + Just msg -> io (serial proto msg) >> send_events + in send_events where serial proto e = lazyByteString $ @@ -1257,11 +1283,32 @@ newLine SerialAsJSON = "\n" newLine SerialAsPrettyJSON = mempty -receiveEvent :: Inbox StoreEvent -> IO (Maybe Event) -receiveEvent sub = - go <$> receive sub +-- | Most events queued for one /events client. A client that does not keep +-- up is disconnected once this many are waiting, instead of the queue +-- growing without bound. +eventQueueSize :: Natural +eventQueueSize = 10000 + +-- | Subscribe to the store events that /events clients see. The action gets +-- a blocking read that returns the next event, or 'Nothing' once the queue +-- has overflowed, after which the client must be disconnected. Other store +-- events are dropped by the publisher instead of being queued. +withEventSubscription :: + Publisher StoreEvent -> (IO (Maybe Event) -> IO a) -> IO a +withEventSubscription pub f = do + inbox <- newBoundedInbox eventQueueSize + overflow <- newTVarIO False + let listen e = forM_ (toEvent e) $ \ev -> + mailboxFullSTM inbox >>= \case + True -> writeTVar overflow True + False -> sendSTM ev inbox + next = + atomically $ + (readTVar overflow >>= checkSTM >> return Nothing) + `orElse` (Just <$> receiveSTM inbox) + bracket (subscribe pub listen) (unsubscribe pub) $ \_ -> f next where - go = \case + toEvent = \case StoreBestBlock b -> Just (EventBlock b) StoreMempoolNew t -> Just (EventTx t) StoreMempoolDelete t -> Just (EventTx t) @@ -1270,19 +1317,19 @@ -- GET Address Transactions -- scottyAddrTxs :: - (MonadUnliftIO m, MonadLoggerIO m) => GetAddrTxs -> ActionT m [TxRef] + (MonadUnliftIO m) => GetAddrTxs -> ActionT m [TxRef] scottyAddrTxs (GetAddrTxs addr pLimits) = withMetrics (.addressTx) $ getAddressTxs addr =<< paramToLimits False pLimits scottyAddrsTxs :: - (MonadUnliftIO m, MonadLoggerIO m) => GetAddrsTxs -> ActionT m [TxRef] + (MonadUnliftIO m) => GetAddrsTxs -> ActionT m [TxRef] scottyAddrsTxs (GetAddrsTxs addrs pLimits) = withMetrics (.addressTx) $ getAddressesTxs addrs =<< paramToLimits False pLimits scottyAddrTxsFull :: - (MonadUnliftIO m, MonadLoggerIO m) => + (MonadUnliftIO m) => GetAddrTxsFull -> ActionT m [Transaction] scottyAddrTxsFull (GetAddrTxsFull addr pLimits) = @@ -1291,7 +1338,7 @@ catMaybes <$> mapM (getTransaction . (.txid)) txs scottyAddrsTxsFull :: - (MonadUnliftIO m, MonadLoggerIO m) => + (MonadUnliftIO m) => GetAddrsTxsFull -> ActionT m [Transaction] scottyAddrsTxsFull (GetAddrsTxsFull addrs pLimits) = @@ -1304,18 +1351,18 @@ withMetrics (.addressBalance) $ getDefaultBalance addr scottyAddrsBalance :: - (MonadUnliftIO m, MonadLoggerIO m) => GetAddrsBalance -> ActionT m [Balance] + (MonadUnliftIO m) => GetAddrsBalance -> ActionT m [Balance] scottyAddrsBalance (GetAddrsBalance addrs) = withMetrics (.addressBalance) $ getBalances addrs scottyAddrUnspent :: - (MonadUnliftIO m, MonadLoggerIO m) => GetAddrUnspent -> ActionT m [Unspent] + (MonadUnliftIO m) => GetAddrUnspent -> ActionT m [Unspent] scottyAddrUnspent (GetAddrUnspent addr pLimits) = withMetrics (.addressUnspent) $ getAddressUnspents addr =<< paramToLimits False pLimits scottyAddrsUnspent :: - (MonadUnliftIO m, MonadLoggerIO m) => GetAddrsUnspent -> ActionT m [Unspent] + (MonadUnliftIO m) => GetAddrsUnspent -> ActionT m [Unspent] scottyAddrsUnspent (GetAddrsUnspent addrs pLimits) = withMetrics (.addressUnspent) $ getAddressesUnspents addrs =<< paramToLimits False pLimits @@ -1323,14 +1370,14 @@ -- GET XPubs -- scottyXPub :: - (MonadUnliftIO m, MonadLoggerIO m) => GetXPub -> ActionT m XPubSummary + (MonadUnliftIO m) => GetXPub -> ActionT m XPubSummary scottyXPub (GetXPub xpub deriv (NoCache noCache)) = withMetrics (.xpub) $ let xs = XPubSpec xpub deriv in xPubSummary xs <$> lift (runNoCache noCache $ xPubBals xs) scottyDelXPub :: - (MonadUnliftIO m, MonadLoggerIO m) => + (MonadUnliftIO m) => DelCachedXPub -> ActionT m (GenericResult Bool) scottyDelXPub (DelCachedXPub xpub deriv) = @@ -1340,7 +1387,7 @@ return $ GenericResult (n > 0) getXPubTxs :: - (MonadUnliftIO m, MonadLoggerIO m) => + (MonadUnliftIO m) => XPubKey -> DeriveType -> LimitsParam -> @@ -1353,13 +1400,13 @@ lift $ runNoCache nocache $ xPubTxs xs xbals limits scottyXPubTxs :: - (MonadUnliftIO m, MonadLoggerIO m) => GetXPubTxs -> ActionT m [TxRef] + (MonadUnliftIO m) => GetXPubTxs -> ActionT m [TxRef] scottyXPubTxs (GetXPubTxs xpub deriv plimits (NoCache nocache)) = withMetrics (.xpubTx) $ getXPubTxs xpub deriv plimits nocache scottyXPubTxsFull :: - (MonadUnliftIO m, MonadLoggerIO m) => + (MonadUnliftIO m) => GetXPubTxsFull -> ActionT m [Transaction] scottyXPubTxsFull (GetXPubTxsFull xpub deriv plimits (NoCache nocache)) = @@ -1369,7 +1416,7 @@ =<< getXPubTxs xpub deriv plimits nocache scottyXPubBalances :: - (MonadUnliftIO m, MonadLoggerIO m) => GetXPubBalances -> ActionT m [XPubBal] + (MonadUnliftIO m) => GetXPubBalances -> ActionT m [XPubBal] scottyXPubBalances (GetXPubBalances xpub deriv (NoCache noCache)) = withMetrics (.xpubBalance) $ lift $ @@ -1377,7 +1424,7 @@ xPubBals (XPubSpec xpub deriv) scottyXPubUnspent :: - (MonadUnliftIO m, MonadLoggerIO m) => + (MonadUnliftIO m) => GetXPubUnspent -> ActionT m [XPubUnspent] scottyXPubUnspent (GetXPubUnspent xpub deriv pLimits (NoCache noCache)) = @@ -1482,7 +1529,7 @@ getChainHeight = fmap (.height) $ liftIO . chainGetBest =<< askl (.config.store.chain) -scottyBinfoUnspent :: (MonadUnliftIO m, MonadLoggerIO m) => ActionT m () +scottyBinfoUnspent :: (MonadUnliftIO m) => ActionT m () scottyBinfoUnspent = withMetrics (.binfoUnspent) $ do (xspecs, addrs) <- getBinfoActive @@ -1661,7 +1708,7 @@ xpub = BinfoXpub <$> xPubImport net ctx txt maybe S.next return (addr <|> xpub) -scottyBinfoHistory :: (MonadUnliftIO m, MonadLoggerIO m) => ActionT m () +scottyBinfoHistory :: (MonadUnliftIO m) => ActionT m () scottyBinfoHistory = withMetrics (.binfoExportHistory) $ do (xspecs, addrs) <- getBinfoActive @@ -1683,7 +1730,8 @@ net <- askl (.config.store.net) url <- askl (.config.priceHistoryURL) session <- askl (.session) - rates <- map (.price) <$> lift (getRates net session url code times) + logger <- askl (.config.logger) + rates <- map (.price) <$> lift (getRates logger net session url code times) let hs = zipWith (convert cur aaddrs) txs (rates <> repeat 0.0) setHeaders streamEncoding $ toEncoding hs @@ -1749,7 +1797,7 @@ getSymbol :: (MonadUnliftIO m) => ActionT m BinfoSymbol getSymbol = uncurry binfoTickerToSymbol <$> getPrice -scottyBinfoBlocksDay :: (MonadUnliftIO m, MonadLoggerIO m) => ActionT m () +scottyBinfoBlocksDay :: (MonadUnliftIO m) => ActionT m () scottyBinfoBlocksDay = withMetrics (.binfoBlock) $ do t <- min h . (`div` 1000) <$> S.captureParam "milliseconds" @@ -1768,7 +1816,7 @@ b' <- getBlock b.header.prev (b :) <$> go t b' -scottyMultiAddr :: (MonadUnliftIO m, MonadLoggerIO m) => ActionT m () +scottyMultiAddr :: (MonadUnliftIO m) => ActionT m () scottyMultiAddr = withMetrics (.binfoMultiaddr) $ do (addrs', _, saddrs, sxpubs, xspecs) <- get_addrs @@ -1947,7 +1995,7 @@ "offset exceeded: " <> show o <> " > " <> show x return (fromIntegral o :: Int) -scottyRawAddr :: (MonadUnliftIO m, MonadLoggerIO m) => ActionT m () +scottyRawAddr :: (MonadUnliftIO m) => ActionT m () scottyRawAddr = withMetrics (.binfoAddressRaw) $ getBinfoAddr "addr" >>= \case @@ -2066,7 +2114,7 @@ setHeaders S.text $ cs $ show $ b.confirmed + b.unconfirmed -scottyFirstSeen :: (MonadUnliftIO m, MonadLoggerIO m) => ActionT m () +scottyFirstSeen :: (MonadUnliftIO m) => ActionT m () scottyFirstSeen = withMetrics (.binfoQaddressfirstseen) $ do a <- getAddress =<< S.captureParam "addr" @@ -2094,7 +2142,8 @@ liftIO (chainGetAncestor ch h bb) >>= \case Just b -> return b.header.timestamp Nothing -> do - lift . $(logErrorS) "Web" $ + logger <- askl (.config.logger) + logError logger $ "Could not get ancestor at height " <> cs (show h) <> " for block " @@ -2104,7 +2153,7 @@ let l = Limits 1 0 (Just (AtBlock h)) not . null <$> getAddressTxs a l -scottyShortBal :: (MonadUnliftIO m, MonadLoggerIO m) => ActionT m () +scottyShortBal :: (MonadUnliftIO m) => ActionT m () scottyShortBal = withMetrics (.binfoBalance) $ do (xspecs, addrs) <- getBinfoActive @@ -2219,7 +2268,7 @@ streamEncoding $ marshalEncoding (net, ctx) y getBinfoTx :: - (MonadLoggerIO m, MonadUnliftIO m) => + (MonadUnliftIO m) => BinfoTxId -> ActionT m (Either Except Transaction) getBinfoTx txid = do @@ -2233,7 +2282,7 @@ let tids = map (txHash . transactionData) ts in return $ Left (TxIndexConflict tids) -scottyBinfoTx :: (MonadUnliftIO m, MonadLoggerIO m) => ActionT m () +scottyBinfoTx :: (MonadUnliftIO m) => ActionT m () scottyBinfoTx = withMetrics (.binfoTxRaw) $ do numtxid <- getNumTxId @@ -2254,7 +2303,7 @@ setHeaders S.text . encodeHexLazy . runPutL . serialize $ transactionData t -scottyBinfoTotalOut :: (MonadUnliftIO m, MonadLoggerIO m) => ActionT m () +scottyBinfoTotalOut :: (MonadUnliftIO m) => ActionT m () scottyBinfoTotalOut = withMetrics (.binfoQtxtotalbtcoutput) $ do txid <- S.captureParam "txid" @@ -2264,7 +2313,7 @@ Left e -> raise e S.text $ cs $ show $ sum $ map (.value) tx.outputs -scottyBinfoTxFees :: (MonadUnliftIO m, MonadLoggerIO m) => ActionT m () +scottyBinfoTxFees :: (MonadUnliftIO m) => ActionT m () scottyBinfoTxFees = withMetrics (.binfoQtxfee) $ do txid <- S.captureParam "txid" @@ -2279,7 +2328,7 @@ f StoreInput {} = True f StoreCoinbase {} = False -scottyBinfoTxResult :: (MonadUnliftIO m, MonadLoggerIO m) => ActionT m () +scottyBinfoTxResult :: (MonadUnliftIO m) => ActionT m () scottyBinfoTxResult = withMetrics (.binfoQtxresult) $ do txid <- S.captureParam "txid" @@ -2297,7 +2346,7 @@ g addr StoreOutput {address = Just a} = a == addr g _ _ = False -scottyBinfoTotalInput :: (MonadUnliftIO m, MonadLoggerIO m) => ActionT m () +scottyBinfoTotalInput :: (MonadUnliftIO m) => ActionT m () scottyBinfoTotalInput = withMetrics (.binfoQtxtotalbtcinput) $ do txid <- S.captureParam "txid" @@ -2392,7 +2441,7 @@ Nothing -> raise ThingNotFound Just a -> S.text $ TL.fromStrict a -scottyBinfoPubKeyAddr :: (MonadUnliftIO m, MonadLoggerIO m) => ActionT m () +scottyBinfoPubKeyAddr :: (MonadUnliftIO m) => ActionT m () scottyBinfoPubKeyAddr = withMetrics (.binfoQpubkeyaddr) $ do addr <- getAddress =<< S.captureParam "addr" @@ -2459,7 +2508,7 @@ -- GET Network Information -- scottyPeers :: - (MonadUnliftIO m, MonadLoggerIO m) => + (MonadUnliftIO m) => GetPeers -> ActionT m [PeerInfo] scottyPeers _ = @@ -2468,7 +2517,7 @@ -- | Obtain information about connected peers from peer manager process. getPeersInformation :: - (MonadLoggerIO m) => PeerMgr -> m [PeerInfo] + (MonadIO m) => PeerMgr -> m [PeerInfo] getPeersInformation mgr = mapMaybe toInfo <$> liftIO (getPeers mgr) where @@ -2588,8 +2637,9 @@ withMetrics (.db) $ do setHeaders db <- askl (.config.store.db.db) + memory <- liftIO (memoryReport db) statsM <- liftIO (getProperty db Stats) - S.text $ maybe "Could not get stats" cs statsM + S.text . cs $ memory <> maybe "Could not get stats" cs statsM ----------------------- -- Parameter Parsing -- @@ -2641,9 +2691,9 @@ bin = runGetS deserialize hex b = let ns = C.filter (not . isSpace) b - in if isBase16 ns - then bin . decodeBase16 $ assertBase16 ns - else Left "Invalid hex input" + in case parseBase16 ns of + Right b16 -> bin $ decodeBase16 b16 + Left e -> Left (cs e) parseOffset :: (MonadUnliftIO m) => ActionT m OffsetParam parseOffset = do @@ -2655,7 +2705,7 @@ return res parseStart :: - (MonadUnliftIO m, MonadLoggerIO m) => + (MonadUnliftIO m) => Maybe StartParam -> ActionT m (Maybe Start) parseStart Nothing = return Nothing @@ -2682,7 +2732,7 @@ parseLimits = LimitsParam <$> paramOptional <*> parseOffset <*> paramOptional paramToLimits :: - (MonadUnliftIO m, MonadLoggerIO m) => + (MonadUnliftIO m) => Bool -> LimitsParam -> ActionT m Limits @@ -2724,19 +2774,15 @@ runNoCache True = local $ \s -> s {config = s.config {store = s.config.store {cache = Nothing}}} -logIt :: - (MonadUnliftIO m, MonadLoggerIO m) => - m Middleware -logIt = do - runner <- askRunInIO - return $ \app req respond -> - app req $ \res -> do - let s = responseStatus res - msg = fmtReq req <> ": " <> fmtStatus s - if statusIsSuccessful s - then runner $ $(logDebugS) "Web" msg - else runner $ $(logErrorS) "Web" msg - respond res +logIt :: Logger -> Middleware +logIt logger app req respond = + app req $ \res -> do + let s = responseStatus res + msg = fmtReq req <> ": " <> fmtStatus s + if statusIsSuccessful s + then logDebug logger msg + else logError logger msg + respond res fmtReq :: Request -> Text fmtReq req = diff --git a/test/Haskoin/Store/BlockStoreSpec.hs b/test/Haskoin/Store/BlockStoreSpec.hs new file mode 100644 --- /dev/null +++ b/test/Haskoin/Store/BlockStoreSpec.hs @@ -0,0 +1,16 @@ +module Haskoin.Store.BlockStoreSpec (spec) where + +import Haskoin.Store.BlockStore (needsMoreBlocks) +import Test.Hspec + +spec :: Spec +spec = + describe "needsMoreBlocks" $ do + it "asks for more blocks once fewer than a quarter of the window remain" $ do + -- 64 / 4 = 16 + needsMoreBlocks 64 16 `shouldBe` False + needsMoreBlocks 64 15 `shouldBe` True + it "asks for more blocks once none remain, even with a window below four" $ do + needsMoreBlocks 1 1 `shouldBe` False + needsMoreBlocks 1 0 `shouldBe` True + needsMoreBlocks 3 0 `shouldBe` True diff --git a/test/Haskoin/Store/LogicSpec.hs b/test/Haskoin/Store/LogicSpec.hs new file mode 100644 --- /dev/null +++ b/test/Haskoin/Store/LogicSpec.hs @@ -0,0 +1,400 @@ +{-# LANGUAGE DuplicateRecordFields #-} +{-# LANGUAGE ImportQualifiedPost #-} +{-# LANGUAGE OverloadedRecordDot #-} +{-# LANGUAGE OverloadedStrings #-} + +module Haskoin.Store.LogicSpec (spec) where + +import Control.Monad (forM_, void) +import Control.Monad.Reader (ask, liftIO, runReaderT) +import Data.ByteString qualified as B +import Data.List (sort) +import Data.Text qualified as T +import Data.Word (Word32, Word64) +import Database.RocksDB.Query (insert, retrieve) +import Haskoin + ( Block (..), + BlockHash, + BlockHeader (..), + BlockNode (..), + Ctx, + Network, + OutPoint (..), + Tx (..), + TxHash (..), + sha256, + TxIn (..), + TxOut (..), + bchRegTest, + btcRegTest, + genesisNode, + headerHash, + prepareContext, + scriptToAddressBS, + txHash, + ) +import Haskoin.Node (LogLevel (..), Logger (..)) +import Haskoin.Store.Common (StoreReadBase (..), getActiveTxData, getDefaultBalance) +import Haskoin.Store.Data + ( Balance (..), + BlockData (..), + BlockRef (..), + Spender (..), + TxData (..), + Unspent (..), + ) +import Haskoin.Store.Database.Reader (DatabaseReader (..), DatabaseReaderT, memoryReport, withDatabaseReader) +import Haskoin.Store.Database.Types (MemKey (..), VersionKey (..)) +import Haskoin.Store.Database.Writer (runWriter) +import Haskoin.Store.Logic + ( ImportException (..), + ImportM, + deleteUnconfirmedTx, + importBlock, + initBest, + newMempoolTx, + revertBlock, + ) +import Haskoin.Store.TestBlocks (coinbaseTx, nextBlock, script) +import System.Log.FastLogger (fromLogStr) +import Test.Hspec +import UnliftIO (catch, evaluate, try, withSystemTempDirectory) + +spec :: Spec +spec = prepareContext $ \ctx -> do + describe "initBest" $ + it "imports the genesis block once" $ + withDB btcRegTest ctx $ \db -> do + importing db initBest `shouldReturn` Right () + importing db initBest `shouldReturn` Right () + reading db getBestBlock `shouldReturn` Just genesis + + describe "memoryReport" $ + it "lists the memory of every column family" $ + withCoin btcRegTest ctx $ \db _ -> do + report <- memoryReport db.db + forM_ ["default", "addr-tx", "unspent", "balance", "total", "block cache"] $ \name -> + report `shouldSatisfy` T.isInfixOf name + report `shouldNotSatisfy` T.isInfixOf "?" + + describe "data version 18" $ + it "moves the mempool into its column family when opened" $ + withSystemTempDirectory "haskoin-store-migrate-" $ \dir -> do + let open :: DatabaseReaderT IO () -> IO () + open = withDatabaseReader btcRegTest ctx 20 32 dir False Nothing Nothing Nothing Nothing + h1 = TxHash (sha256 ("one" :: B.ByteString)) + h2 = TxHash (sha256 ("two" :: B.ByteString)) + old = [(200, h2), (100, h1)] :: [(Word64, TxHash)] + open $ ask >>= \r -> liftIO $ do + insert r.db VersionKey (18 :: Word32) + insert r.db MemKey old + open $ ask >>= \r -> liftIO $ do + reading r getMempool `shouldReturn` old + (retrieve r.db MemKey :: IO (Maybe [(Word64, TxHash)])) `shouldReturn` Nothing + retrieve r.db VersionKey `shouldReturn` Just (19 :: Word32) + + describe "newMempoolTx" $ do + it "imports a transaction and spends its input" $ + withCoin btcRegTest ctx $ \db coin -> do + let tx = spend [coin] [1000, 2000] + importing db (newMempoolTx tx 1700000000) `shouldReturn` Right True + reading db getMempool `shouldReturn` [(1700000000, txHash tx)] + reading db (getSpender coin) `shouldReturn` Just (Spender (txHash tx) 0) + fmap (.value) <$> reading db (getUnspent (out tx 1)) `shouldReturn` Just 2000 + + it "ignores a transaction it already has" $ + withCoin btcRegTest ctx $ \db coin -> do + let tx = spend [coin] [1000] + mapM (addTx db) [tx, tx] `shouldReturn` [Right True, Right False] + _ <- mine db [tx] + addTx db tx `shouldReturn` Right False + fmap (.block) <$> txData db tx `shouldReturn` Just (BlockRef 2 1) + + it "rejects invalid transactions" $ + withCoin btcRegTest ctx $ \db coin -> do + addTx db (spend [out (coinbaseTx 9) 0] [1]) `shouldReturn` Left Orphan + addTx db (spend [coin, coin] [1]) `shouldReturn` Left DuplicatePrevOutput + addTx db (spend [coin] [5000000001]) `shouldReturn` Left InsufficientFunds + addTx db (coinbaseTx 9) `shouldReturn` Left UnexpectedCoinbase + mempool db `shouldReturn` [] + + it "flags BIP125 signalling transactions and their descendants" $ + withCoin btcRegTest ctx $ \db coin -> do + let parent = spendRBF [coin] [1000] + child = spend [out parent 0] [900] + importing db (mapM (`newMempoolTx` 0) [parent, child]) + `shouldReturn` Right [True, True] + mapM (rbfFlag db) [parent, child] `shouldReturn` [Just True, Just True] + + it "does not flag other transactions" $ do + withCoin btcRegTest ctx $ \db coin -> do + let parent = spend [coin] [1000] + child = spend [out parent 0] [900] + mapM (addTx db) [parent, child] `shouldReturn` [Right True, Right True] + mapM (rbfFlag db) [parent, child] `shouldReturn` [Just False, Just False] + withCoin bchRegTest ctx $ \db coin -> do + let tx = spendRBF [coin] [1000] + addTx db tx `shouldReturn` Right True + rbfFlag db tx `shouldReturn` Just False + + it "replaces conflicting transactions and their descendants" $ + withCoin btcRegTest ctx $ \db coin -> do + let original = spend [coin] [1000] + child = spend [out original 0] [900] + replacement = spend [coin] [2000] + mapM (addTx db) [original, child, replacement] + `shouldReturn` [Right True, Right True, Right True] + mempool db `shouldReturn` [txHash replacement] + reading db (getSpender coin) `shouldReturn` Just (Spender (txHash replacement) 0) + fmap (.deleted) <$> reading db (getTxData (txHash original)) `shouldReturn` Just True + + it "replaces a transaction and its descendant when both conflict" $ + withCoin btcRegTest ctx $ \db coin1 -> do + _ <- mine db [] + let coin2 = out (coinbaseTx 2) 0 + parent = spend [coin1] [1000] + child = spend [out parent 0, coin2] [900] + replacement = spend [coin1, coin2] [2000] + mapM (addTx db) [parent, child, replacement] + `shouldReturn` [Right True, Right True, Right True] + mempool db `shouldReturn` [txHash replacement] + + it "leaves the mempool unchanged when it rejects a replacement" $ + withCoin btcRegTest ctx $ \db coin1 -> do + _ <- mine db [] + let coin2 = out (coinbaseTx 2) 0 + original = spend [coin1] [1000] + _ <- mine db [spend [coin2] [1000]] + addTx db original `shouldReturn` Right True + -- The block store imports mempool transactions in batches and skips + -- the ones that are rejected. + let tryAdd tx = newMempoolTx tx 0 `catch` skip + skip :: ImportException -> ImportM Bool + skip _ = return False + unfunded = spend [coin1] [5000000001] + confirmedConflict = spend [coin1, coin2] [2000] + spendsReplaced = spend [coin1, out original 0] [900] + importing db (mapM tryAdd [unfunded, confirmedConflict, spendsReplaced]) + `shouldReturn` Right [False, False, False] + mempool db `shouldReturn` [txHash original] + + it "rejects double spends on networks without RBF" $ + withCoin bchRegTest ctx $ \db coin -> do + let original = spendRBF [coin] [1000] + mapM (addTx db) [original, spend [coin] [2000]] + `shouldReturn` [Right True, Left DoubleSpend] + mempool db `shouldReturn` [txHash original] + + it "rejects double spends of confirmed transactions" $ + withCoin btcRegTest ctx $ \db coin -> do + _ <- mine db [spend [coin] [1000]] + addTx db (spend [coin] [2000]) `shouldReturn` Left TxConfirmed + + it "commits nothing from a batch that fails" $ + withCoin btcRegTest ctx $ \db coin -> do + let good = spend [coin] [1000] + orphan = spend [out (coinbaseTx 9) 0] [1] + importing db (newMempoolTx good 0 >> newMempoolTx orphan 0) + `shouldReturn` Left Orphan + mempool db `shouldReturn` [] + reading db (getSpender coin) `shouldReturn` Nothing + + describe "importBlock" $ do + it "confirms mempool transactions" $ + withCoin btcRegTest ctx $ \db coin -> do + let tx = spendRBF [coin] [1000] + addTx db tx `shouldReturn` Right True + _ <- mine db [tx] + mempool db `shouldReturn` [] + fmap (\t -> (t.block, t.rbf)) <$> txData db tx `shouldReturn` Just (BlockRef 2 1, False) + fmap (.block) <$> reading db (getUnspent (out tx 0)) `shouldReturn` Just (BlockRef 2 1) + + it "confirms a transaction whose outputs are spent in the mempool" $ + withCoin btcRegTest ctx $ \db coin -> do + let parent = spend [coin] [1000] + child = spend [out parent 0] [900] + mapM (addTx db) [parent, child] `shouldReturn` [Right True, Right True] + _ <- mine db [parent] + mempool db `shouldReturn` [txHash child] + fmap (.block) <$> txData db parent `shouldReturn` Just (BlockRef 2 1) + reading db (getSpender (out parent 0)) `shouldReturn` Just (Spender (txHash child) 0) + + it "stops flagging a descendant once its signalling ancestor confirms" $ + withCoin btcRegTest ctx $ \db coin -> do + let parent = spendRBF [coin] [1000] + child = spend [out parent 0] [900] + mapM (addTx db) [parent, child] `shouldReturn` [Right True, Right True] + _ <- mine db [parent] + rbfFlag db child `shouldReturn` Just False + + it "imports transactions in any order and records the block" $ + withCoin btcRegTest ctx $ \db coin -> do + let parent = spend [coin] [4000000000] + child = spend [out parent 0] [3999990000] + block <- mine db [child, parent] + Just bd <- reading db (getBlock (headerHash block.header)) + (bd.height, bd.main, bd.txs) `shouldBe` (2, True, map txHash block.txs) + (bd.subsidy, bd.fee, bd.outputs) `shouldBe` (5000000000, 1000010000, 7999990000) + fmap (.block) <$> txData db child `shouldReturn` Just (BlockRef 2 1) + + it "evicts conflicting mempool transactions and their descendants" $ + forM_ [btcRegTest, bchRegTest] $ \net -> + withCoin net ctx $ \db coin -> do + let original = spend [coin] [1000] + child = spend [out original 0] [900] + mapM (addTx db) [original, child] `shouldReturn` [Right True, Right True] + _ <- mine db [spend [coin] [2000]] + mempool db `shouldReturn` [] + mapM (txData db) [original, child] `shouldReturn` [Nothing, Nothing] + + it "rejects blocks that spend unknown outputs, changing nothing" $ + withCoin btcRegTest ctx $ \db _ -> do + best <- reading db getBestBlock + tryMine db [spend [out (coinbaseTx 9) 0] [1]] `shouldReturn` Left Orphan + reading db getBestBlock `shouldReturn` best + + it "rejects blocks that do not build on the best block" $ + withDB btcRegTest ctx $ \db -> do + let (block1, node1) = nextBlock btcRegTest 0 (genesisNode btcRegTest).header [] + (block2, node2) = nextBlock btcRegTest 1 node1.header [] + importing db (void (importBlock block1 node1)) `shouldReturn` Left BestBlockNotFound + importing db initBest `shouldReturn` Right () + importing db (void (importBlock block2 node2)) `shouldReturn` Left PrevBlockNotBest + + describe "revertBlock" $ do + it "returns the block's transactions to the mempool" $ + withCoin btcRegTest ctx $ \db coin -> do + let tx = spendRBF [coin] [1000] + block <- mine db [tx] + let h = headerHash block.header + importing db (revertBlock h) `shouldReturn` Right () + reading db getBestBlock `shouldReturn` Just block.header.prev + fmap (.main) <$> reading db (getBlock h) `shouldReturn` Just False + reading db getMempool `shouldReturn` [(1296689802, txHash tx)] + rbfFlag db tx `shouldReturn` Just True + txData db (coinbaseTx 2) `shouldReturn` Nothing + + it "flags mempool descendants of the signalling transactions it returns" $ + withCoin btcRegTest ctx $ \db coin -> do + let parent = spendRBF [coin] [1000] + child = spend [out parent 0] [900] + block <- mine db [parent] + addTx db child `shouldReturn` Right True + rbfFlag db child `shouldReturn` Just False + importing db (revertBlock (headerHash block.header)) `shouldReturn` Right () + rbfFlag db child `shouldReturn` Just True + + it "only reverts the best block" $ + withDB btcRegTest ctx $ \db -> do + importing db (revertBlock genesis) `shouldReturn` Left BestBlockNotFound + importing db initBest `shouldReturn` Right () + _ <- mine db [] + importing db (revertBlock genesis) `shouldReturn` Left BlockNotBest + + describe "deleteUnconfirmedTx" $ do + it "deletes a transaction and its descendants" $ + withCoin btcRegTest ctx $ \db coin -> do + let parent = spend [coin] [1000] + child = spend [out parent 0] [900] + mapM (addTx db) [parent, child] `shouldReturn` [Right True, Right True] + importing db (deleteUnconfirmedTx False (txHash parent)) `shouldReturn` Right () + mempool db `shouldReturn` [] + fmap (.value) <$> reading db (getUnspent coin) `shouldReturn` Just 5000000000 + + it "ignores unknown transactions" $ + withCoin btcRegTest ctx $ \db _ -> + importing db (deleteUnconfirmedTx False (txHash (coinbaseTx 9))) + `shouldReturn` Right () + + it "refuses to delete confirmed transactions" $ + withCoin btcRegTest ctx $ \db _ -> + importing db (deleteUnconfirmedTx False (txHash (coinbaseTx 1))) + `shouldReturn` Left TxConfirmed + + describe "address balances" $ + it "follow transactions through the mempool, a block and its reversal" $ + withCoin btcRegTest ctx $ \db coin -> do + a <- either fail return (scriptToAddressBS ctx script) + let tx = spend [coin] [1000, 2000] + -- confirmed, unconfirmed, outputs, transactions, received + balance = do + b <- reading db (getDefaultBalance a) + return (b.confirmed, b.unconfirmed, b.utxo, b.txs, b.received) + balance `shouldReturn` (5000000000, 0, 1, 1, 5000000000) + addTx db tx `shouldReturn` Right True + balance `shouldReturn` (0, 3000, 2, 2, 5000003000) + block <- mine db [tx] + balance `shouldReturn` (5000003000, 0, 3, 3, 10000003000) + importing db (revertBlock (headerHash block.header)) `shouldReturn` Right () + balance `shouldReturn` (0, 3000, 2, 2, 5000003000) + +-- | Run with an empty database in a temporary directory. +withDB :: Network -> Ctx -> (DatabaseReader -> IO a) -> IO a +withDB net ctx go = + withSystemTempDirectory "haskoin-store-logic-" $ \dir -> + withDatabaseReader net ctx 20 32 dir False Nothing Nothing Nothing Nothing (ask >>= liftIO . go) + +-- | Run with the genesis block and one mined block, passing the output of +-- that block's coinbase, worth 5000000000. +withCoin :: Network -> Ctx -> (DatabaseReader -> OutPoint -> IO a) -> IO a +withCoin net ctx go = + withDB net ctx $ \db -> do + importing db initBest `shouldReturn` Right () + _ <- mine db [] + go db (out (coinbaseTx 1) 0) + +importing :: DatabaseReader -> ImportM a -> IO (Either ImportException a) +importing db = try . runWriter db.net db.ctx db quiet + where + -- Build every message, so that log lines are tested too, but print none. + quiet = + Logger + { level = LevelDebug, + source = "", + push = void . evaluate . B.length . fromLogStr + } + +reading :: DatabaseReader -> DatabaseReaderT IO a -> IO a +reading db f = runReaderT f db + +addTx :: DatabaseReader -> Tx -> IO (Either ImportException Bool) +addTx db tx = importing db (newMempoolTx tx 0) + +txData :: DatabaseReader -> Tx -> IO (Maybe TxData) +txData db = reading db . getActiveTxData . txHash + +rbfFlag :: DatabaseReader -> Tx -> IO (Maybe Bool) +rbfFlag db tx = fmap (.rbf) <$> txData db tx + +mempool :: DatabaseReader -> IO [TxHash] +mempool db = sort . map snd <$> reading db getMempool + +-- | Mine a block with these transactions on top of the best block. +tryMine :: DatabaseReader -> [Tx] -> IO (Either ImportException Block) +tryMine db txs = do + Just h <- reading db getBestBlock + Just parent <- reading db (getBlock h) + let (block, node) = nextBlock db.net parent.height parent.header txs + fmap (const block) <$> importing db (importBlock block node) + +mine :: DatabaseReader -> [Tx] -> IO Block +mine db txs = tryMine db txs >>= either (fail . ("Block rejected: " <>) . show) return + +-- | Spend outputs into new ones of the given values. +spend :: [OutPoint] -> [Word64] -> Tx +spend = spendWith maxBound + +-- | Like 'spend', but signal BIP125 replaceability. +spendRBF :: [OutPoint] -> [Word64] -> Tx +spendRBF = spendWith 0 + +spendWith :: Word32 -> [OutPoint] -> [Word64] -> Tx +spendWith s ops vs = + Tx 2 [TxIn op "" s | op <- ops] [TxOut v script | v <- vs] [] 0 + +out :: Tx -> Word32 -> OutPoint +out tx = OutPoint (txHash tx) + +-- | Bitcoin regtest genesis block. +genesis :: BlockHash +genesis = "0f9188f13cb7b2c71f2a335e3a4fc328bf5beb436012afca590b1a11466e2206" diff --git a/test/Haskoin/Store/TestBlocks.hs b/test/Haskoin/Store/TestBlocks.hs new file mode 100644 --- /dev/null +++ b/test/Haskoin/Store/TestBlocks.hs @@ -0,0 +1,59 @@ +{-# LANGUAGE ImportQualifiedPost #-} +{-# LANGUAGE OverloadedRecordDot #-} +{-# LANGUAGE OverloadedStrings #-} + +-- | Regtest blocks for tests. Their headers carry valid proof of work, so +-- they pass haskoin-node's header checks as well as the store's import. +module Haskoin.Store.TestBlocks + ( nextBlock, + coinbaseTx, + script, + ) +where + +import Data.ByteString (ByteString) +import Data.ByteString qualified as B +import Data.List (find) +import Data.Maybe (fromMaybe) +import Data.Serialize (encode) +import Haskoin + ( Block (..), + BlockHeader (..), + BlockHeight, + BlockNode (..), + Network, + Tx (..), + TxIn (..), + TxOut (..), + buildMerkleRoot, + headerHash, + isValidPOW, + nullOutPoint, + txHash, + ) + +-- | The next block, 600 seconds after its parent, with a coinbase in front +-- of the given transactions. +nextBlock :: Network -> BlockHeight -> BlockHeader -> [Tx] -> (Block, BlockNode) +nextBlock net height parent txs = + (Block header all_txs, BlockNode header (height + 1) 0 prev) + where + prev = headerHash parent + all_txs = coinbaseTx (height + 1) : txs + -- The list always holds the coinbase, so the root cannot fail for an + -- empty block. + merkle = either error id (buildMerkleRoot (map txHash all_txs)) + candidate = BlockHeader 0x20000000 prev merkle (parent.timestamp + 600) parent.bits + -- Regtest accepts about half of all hashes, so the nonce search is short. + header = + fromMaybe (error "No nonce satisfies regtest proof of work") $ + find (isValidPOW net) (map candidate [0 ..]) + +-- | Coinbase of the block at this height. +coinbaseTx :: BlockHeight -> Tx +coinbaseTx h = Tx 1 [TxIn nullOutPoint (encode h) maxBound] [TxOut 5000000000 script] [] 0 + +-- | Every output pays to this pay-to-public-key-hash script, so that address +-- balances are kept for all of them. +script :: ByteString +script = "\x76\xa9\x14" <> B.replicate 20 0x11 <> "\x88\xac" diff --git a/test/Haskoin/StoreSpec.hs b/test/Haskoin/StoreSpec.hs --- a/test/Haskoin/StoreSpec.hs +++ b/test/Haskoin/StoreSpec.hs @@ -11,20 +11,22 @@ import Conduit import Control.Monad -import Control.Monad.Logger import Control.Monad.Reader import Data.ByteString (ByteString) import Data.ByteString qualified as B import Data.ByteString.Base64 import Data.Either -import Data.List (find) +import Data.HashMap.Strict qualified as HashMap import Data.Maybe import Data.Serialize import Data.Time.Clock.POSIX import Data.Word +import GHC.Stats (GCDetails (..), RTSStats (..), getRTSStats, getRTSStatsEnabled) +import System.Mem (performMajorGC) import Haskoin import Haskoin.Node import Haskoin.Store +import Haskoin.Store.TestBlocks (nextBlock) import NQE import Network.Socket import System.Random @@ -35,7 +37,9 @@ { testStoreDB :: !DatabaseReader, testStoreBlockStore :: !BlockStore, testStoreChain :: !Chain, - testStoreEvents :: !(Inbox StoreEvent) + testStoreEvents :: !(Inbox StoreEvent), + -- | How many blocks each getdata sent to the mock peer asked for, oldest first. + testStoreBlockRequests :: !(TVar [Int]) } spec :: Spec @@ -72,7 +76,46 @@ length bd.txs `shouldBe` 1 listToMaybe bd.txs `shouldBe` Just h1 fmap (txHash . transactionData) t1 `shouldBe` Just h1 + it "downloads 80 blocks with a request for 64 and one for the remaining 16" $ + withTestStore bchRegTest ctx "block-window" $ \TestStore {..} -> do + bs <- replicateM 80 . receiveMatch testStoreEvents $ \case + StoreBestBlock b -> Just b + _ -> Nothing + bestNodeM <- liftIO $ chainGetBlock testStoreChain (last bs) + fmap (.height) bestNodeM `shouldBe` Just 80 + readTVarIO testStoreBlockRequests `shouldReturn` [64, 16] + describe "Memory" $ + it "keeps the heap flat while syncing" $ do + enabled <- getRTSStatsEnabled + unless enabled $ pendingWith "run with +RTS -T to measure the heap" + -- Each block passes messages through the peer, node and store loops. + -- When forever went through a dictionary, each of six loops kept 33 to + -- 107 bytes per message, a few hundred bytes per block in all: over + -- the 5000 blocks measured, megabytes. The heap normally grows by about + -- 330 kB here, and a leak of 500 bytes per block fails the test. The + -- length is a multiple of 2000, so the last headers message the mock + -- peer sends is empty, which haskoin-node before 2.1.2 did not handle. + let blocks = chainOf 6000 + withTestStoreBlocks blocks bchRegTest ctx "heap" $ \TestStore {..} -> do + let wait_for height = + receive testStoreEvents >>= \case + StoreBestBlock b -> + chainGetBlock testStoreChain b >>= \case + Just n | n.height >= height -> return () + _ -> wait_for height + _ -> wait_for height + live = do + performMajorGC + (.gc.gcdetails_live_bytes) <$> getRTSStats + wait_for 1000 + live_before <- live + wait_for 6000 + live_after <- live + let growth = fromIntegral live_after - fromIntegral live_before :: Integer + putStrLn $ "heap growth over 5000 blocks: " <> show growth <> " bytes" + growth `shouldSatisfy` (< 1024 * 1024) + withTestStore :: (MonadUnliftIO m) => Network -> @@ -80,60 +123,90 @@ String -> (TestStore -> m a) -> m a -withTestStore net ctx t f = - withSystemTempDirectory ("haskoin-store-test-" <> t <> "-") $ \w -> - runNoLoggingT $ do - let ad = - NetworkAddress - nodeNetwork - (sockToHostAddress (SockAddrInet 0 0)) - cfg = - StoreConfig - { maxPeers = 20, - initPeers = [], - discover = True, - db = w, - net = net, - ctx = ctx, - redis = Nothing, - gap = gap, - initGap = 20, - redisMinAddrs = 100, - redisMaxKeys = 100 * 1000 * 1000, - noMempool = False, - wipeMempool = False, - syncMempool = False, - mempoolTimeout = 14, - peerTimeout = 60, - maxPeerLife = 48 * 3600, - connect = dummyPeerConnect net ad, - stats = Nothing, - redisSyncInterval = 30, - bloom = True - } - withStore cfg $ \store -> - withSubscription store.pub $ \sub -> - lift $ - f - TestStore - { testStoreDB = store.db, - testStoreBlockStore = store.block, - testStoreChain = store.chain, - testStoreEvents = sub - } +withTestStore = withTestStoreBlocks allBlocks +-- | Like 'withTestStore', with the mock peer serving these blocks. +withTestStoreBlocks :: + (MonadUnliftIO m) => + [Block] -> + Network -> + Ctx -> + String -> + (TestStore -> m a) -> + m a +withTestStoreBlocks blocks net ctx t f = + withSystemTempDirectory ("haskoin-store-test-" <> t <> "-") $ \w -> do + requests <- newTVarIO [] + let ad = + NetworkAddress + nodeNetwork + (sockToHostAddress (SockAddrInet 0 0)) + cfg = + StoreConfig + { maxPeers = 20, + initPeers = [], + discover = True, + db = w, + net = net, + ctx = ctx, + redis = Nothing, + gap = gap, + initGap = 20, + redisMinAddrs = 100, + redisMaxKeys = 100 * 1000 * 1000, + noMempool = False, + wipeMempool = False, + syncMempool = False, + blockWindow = 64, + mempoolTimeout = 14, + peerTimeout = 60, + maxPeerLife = 48 * 3600, + connect = dummyPeerConnect blocks requests net ad, + stats = Nothing, + redisSyncInterval = 30, + bloom = True, + maxOpenFiles = Nothing, + writeBufferSize = Nothing, + maxWalSize = Nothing, + logger = Logger {level = LevelError, source = "", push = \_ -> return ()} + } + withStore cfg $ \store -> + withSubscription store.pub $ \sub -> + f + TestStore + { testStoreDB = store.db, + testStoreBlockStore = store.block, + testStoreChain = store.chain, + testStoreEvents = sub, + testStoreBlockRequests = requests + } + gap :: Word32 gap = 32 +-- | The 15 canned blocks followed by 65 mined on top of them: more than one +-- 64-block window, so that a download needs a second request. allBlocks :: [Block] -allBlocks = +allBlocks = chainOf 80 + +-- | The canned blocks followed by blocks mined on top of them, this many in +-- all. +chainOf :: Int -> [Block] +chainOf n = take n (cannedBlocks <> mined (length cannedBlocks) (last cannedBlocks)) + where + mined height parent = block : mined (height + 1) block + where + (block, _) = nextBlock bchRegTest (fromIntegral height) parent.header [] + +cannedBlocks :: [Block] +cannedBlocks = fromRight (error "Could not decode blocks") $ - runGet f (decodeBase64Lenient allBlocksBase64) + runGet f (decodeBase64Lenient cannedBlocksBase64) where f = mapM (const get) [(1 :: Int) .. 15] -allBlocksBase64 :: ByteString -allBlocksBase64 = +cannedBlocksBase64 :: ByteString +cannedBlocksBase64 = "AAAAIAYibkYRGgtZyq8SYEPrW78ow086XjMqH8eytzzxiJEPakRJalmWTFwdvzNuH8fHLZEjn+4N\ \FNMANdB7ez2M4a3TFbNe//9/IAMAAAABAgAAAAEAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA\ \AAAAAP////8MUQEBCC9FQjMyLjAv/////wEA8gUqAQAAACMhAwTspkCjMezKs47BPpafou1jjsHf\ @@ -185,8 +258,8 @@ \AAAAAAAAAAAAAAAAAAD/////DF8BAQgvRUIzMi4wL/////8BAPIFKgEAAAAjIQME7KZAozHsyrOO\ \wT6Wn6LtY47B39Th44JKsZ8BGJCvc6wAAAAA" -dummyPeerConnect :: Network -> NetworkAddress -> SockAddr -> WithConnection -dummyPeerConnect net ad sa f = do +dummyPeerConnect :: [Block] -> TVar [Int] -> Network -> NetworkAddress -> SockAddr -> WithConnection +dummyPeerConnect blocks requests net ad sa f = do r <- newInbox s <- newInbox let s' = inboxToMailbox s @@ -205,7 +278,8 @@ runConduit $ forever (receive r >>= yield) .| inc - .| concatMapC mockPeerReact + .| iterMC (recordBlockRequest requests) + .| concatMapC (mockPeerReact blocks) .| outc .| awaitForever (`send` s) outc = mapMC $ \msg' -> return $ runPut (putMessage net msg') @@ -222,16 +296,26 @@ error $ "Dummy peer could not decode payload: " <> show e -mockPeerReact :: Message -> [Message] -mockPeerReact (MPing (Ping n)) = [MPong (Pong n)] -mockPeerReact (MVersion _) = [MVerAck] -mockPeerReact (MGetHeaders (GetHeaders _ _hs _)) = [MHeaders (Headers hs')] +-- | Note how many blocks a getdata asks for. +recordBlockRequest :: TVar [Int] -> Message -> IO () +recordBlockRequest requests (MGetData (GetData ivs)) = + atomically $ modifyTVar' requests (<> [length [() | InvVector InvBlock _ <- ivs]]) +recordBlockRequest _ _ = return () + +mockPeerReact :: [Block] -> Message -> [Message] +mockPeerReact _ (MPing (Ping n)) = [MPong (Pong n)] +mockPeerReact _ (MVersion _) = [MVerAck] +mockPeerReact blocks (MGetHeaders (GetHeaders _ locator _)) = [MHeaders (Headers hs')] where f b = (b.header, VarInt $ fromIntegral $ length b.txs) - hs' = map f allBlocks -mockPeerReact (MGetData (GetData ivs)) = mapMaybe f ivs + -- Up to 2000 headers after the first block of the locator that the + -- peer has, like a real peer; from the start if it has none of them. + heights = HashMap.fromList (zip (map (headerHash . (.header)) blocks) [1 :: Int ..]) + start = fromMaybe 0 (listToMaybe (mapMaybe (`HashMap.lookup` heights) locator)) + hs' = map f (take 2000 (drop start blocks)) +mockPeerReact blocks (MGetData (GetData ivs)) = mapMaybe f ivs where - f (InvVector InvBlock h) = MBlock <$> find (l h) allBlocks + f (InvVector InvBlock h) = MBlock <$> HashMap.lookup (BlockHash h) byHash f _ = Nothing - l h b = headerHash b.header == BlockHash h -mockPeerReact _ = [] + byHash = HashMap.fromList [(headerHash b.header, b) | b <- blocks] +mockPeerReact _ _ = []