haskoin-store 1.6.3 → 1.6.4

raw patch · 15 files changed

+1510/−632 lines, 15 filesdep +fast-loggerdep −monad-loggerdep ~base16dep ~haskoin-coredep ~haskoin-nodePVP: major bump suggested

API removals or changes: PVP suggests a major version bump

Dependencies added: fast-logger

Dependencies removed: monad-logger

Dependency ranges changed: base16, haskoin-core, haskoin-node, rocksdb-haskell-jprupp, websockets

API changes (from Hackage documentation)

Files

• app/Main.hs +77/−25
• haskoin-store.cabal +22/−21
• src/Haskoin/Store/BlockStore.hs +141/−122
• src/Haskoin/Store/Cache.hs +159/−112
• src/Haskoin/Store/Common.hs +13/−0
• src/Haskoin/Store/Database/Reader.hs +74/−7
• src/Haskoin/Store/Database/Types.hs +24/−0
• src/Haskoin/Store/Database/Writer.hs +34/−19
• src/Haskoin/Store/Logic.hs +162/−143
• src/Haskoin/Store/Manager.hs +42/−26
• src/Haskoin/Store/Web.hs +143/−97
• test/Haskoin/Store/BlockStoreSpec.hs +16/−0
• test/Haskoin/Store/LogicSpec.hs +400/−0
• test/Haskoin/Store/TestBlocks.hs +59/−0
• test/Haskoin/StoreSpec.hs +144/−60

@@ -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

@@ -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 <https://github.com/jprupp/haskoin-store#readme> 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

@@ -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

@@ -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

@@ -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 =

@@ -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

@@ -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

@@ -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))

@@ -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}

@@ -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

@@ -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 =

@@ -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

@@ -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"

@@ -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"

@@ -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 _ _ = []