haskoin-store 1.7.0 → 1.8.0

raw patch · 13 files changed

+820/−113 lines, 13 filesdep +vectordep ~haskoin-coredep ~haskoin-store-datadep ~rocksdb-haskell-jpruppPVP ok

version bump matches the API change (PVP)

Dependencies added: vector

Dependency ranges changed: haskoin-core, haskoin-store-data, rocksdb-haskell-jprupp

API changes (from Hackage documentation)

Files

• app/Main.hs +15/−1
• bench/Main.hs +306/−0
• haskoin-store.cabal +67/−12
• src/Haskoin/Store.hs +1/−0
• src/Haskoin/Store/BlockStore.hs +2/−0
• src/Haskoin/Store/Cache.hs +79/−39
• src/Haskoin/Store/Common.hs +67/−2
• src/Haskoin/Store/Database/Reader.hs +166/−46
• src/Haskoin/Store/Manager.hs +5/−0
• src/Haskoin/Store/Web.hs +12/−10
• test/Haskoin/Store/CommonSpec.hs +96/−0
• test/Haskoin/Store/LogicSpec.hs +3/−3
• test/Haskoin/StoreSpec.hs +1/−0

@@ -136,7 +136,8 @@     noWeb :: !Bool,     maxOpenFiles :: !Int,     writeBufferMB :: !Int,-    maxWalMB :: !Int+    maxWalMB :: !Int,+    blockCacheMB :: !Int   }  env :: (MonadIO m) => String -> a -> (String -> Maybe a) -> m a@@ -228,6 +229,8 @@     env "WRITE_BUFFER_MB" 256 readMaybe   maxWalMB <-     env "MAX_WAL_MB" 1024 readMaybe+  blockCacheMB <-+    env "BLOCK_CACHE_MB" 1024 readMaybe   return Config {version = False, ..}   where     tickerString =@@ -580,6 +583,13 @@         <> 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+  blockCacheMB <-+    option auto $+      metavar "MB"+        <> long "block-cache-mb"+        <> help "RocksDB block cache in MiB, shared by all column families, holding index and filter blocks too (0 = RocksDB default of 32 MiB per column family)"+        <> showDefault+        <> value c.blockCacheMB   pure Config {..}  networkReader :: String -> Either String Network@@ -655,6 +665,10 @@             maxWalSize =               if cfg.maxWalMB > 0                 then Just (cfg.maxWalMB * 1024 * 1024)+                else Nothing,+            blockCache =+              if cfg.blockCacheMB > 0+                then Just (cfg.blockCacheMB * 1024 * 1024)                 else Nothing,             logger = logger           }

@@ -0,0 +1,306 @@+{-# LANGUAGE DuplicateRecordFields #-}+{-# LANGUAGE ImportQualifiedPost #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE OverloadedRecordDot #-}+{-# LANGUAGE OverloadedStrings #-}++-- | Load-test server for the web API.+--+-- Syncs a synthetic regtest chain from an in-process mock peer into a fresh+-- database, then serves the web API on it until killed, so that a load+-- generator such as wrk can be pointed at it. See @scripts/bench-web.sh@.+--+-- > haskoin-store-bench PORT BLOCKS [REDIS_URL]+--+-- Every block holds a coinbase and ten transactions. The coinbase and one+-- output of each transaction pay the "hot" address, so it gains eleven+-- transactions per block. Nine outputs per block go to the first 100+-- external and change addresses of the benchmark xpub, in turn. Both are+-- printed once the chain is synced.+module Main (main) where++import Conduit+import Control.Monad (forever, unless)+import Control.Monad.Reader (runReaderT)+import Data.ByteString (ByteString)+import Data.ByteString qualified as B+import Data.HashMap.Strict qualified as HashMap+import Data.List (find)+import Control.DeepSeq (NFData, force)+import Data.Aeson (toEncoding)+import Data.Aeson.Encoding (encodingToLazyByteString)+import Data.Default (def)+import Data.IntMap.Strict qualified as IntMap+import Data.Vector qualified as V+import GHC.Clock (getMonotonicTimeNSec)+import Haskoin.Store.Data+import Data.Maybe (fromMaybe, listToMaybe, mapMaybe)+import Data.Serialize (decode, encode, runGet, runPut)+import Data.Text qualified as T+import Data.Text.IO qualified as T+import Data.Time.Clock.POSIX (getPOSIXTime)+import Data.Word (Word32)+import Haskoin+import Haskoin.Node+import Haskoin.Store+import NQE+import Network.Socket (SockAddr (..))+import System.Environment (getArgs)+import System.Random (randomIO)+import UnliftIO++main :: IO ()+main = do+  args <- getArgs+  case args of+    ["micro"] -> micro+    _ -> server args++server :: [String] -> IO ()+server args = do+  let (port, nblocks, redis) = case args of+        [p, n] -> (read p, read n, Nothing)+        [p, n, r] -> (read p, read n, Just r)+        _ -> error "usage: haskoin-store-bench PORT BLOCKS [REDIS_URL]"+  createContext >>= \ctx ->+    withSystemTempDirectory "haskoin-store-bench-" $ \dir -> do+      let net = bchRegTest+          blocks = chain ctx net nblocks+      requests <- newTVarIO []+      withStore (storeConfig ctx net dir redis blocks requests) $ \store -> do+        withSubscription store.pub $ \sub -> waitHeight store sub nblocks+        T.putStrLn $ "hot address: " <> addrText net (hotAddr ctx)+        T.putStrLn $ "xpub: " <> xPubExport net ctx (benchXPub ctx)+        T.putStrLn $ "xpub address: " <> addrText net (xpubAddr ctx 0 0)+        bh <- runReaderT getBestBlock store.db+        T.putStrLn $ "best block: " <> maybe "" blockHashToHex bh+        putStrLn $ "serving on port " <> show port+        hFlush stdout+        runWeb (webConfig store port)++addrText :: Network -> Address -> T.Text+addrText net = fromMaybe (error "address") . addrToText net++waitHeight :: Store -> Inbox StoreEvent -> Int -> IO ()+waitHeight store sub n = do+  h <- maybe 0 (.height) <$> runReaderT (getBestBlock >>= maybe (return Nothing) getBlock) store.db+  unless (fromIntegral h >= n) $ do+    _ <- receiveMatch sub $ \case+      StoreBestBlock b -> Just b+      _ -> Nothing+    waitHeight store sub n++storeConfig ::+  Ctx -> Network -> FilePath -> Maybe String -> [Block] -> TVar [Int] -> StoreConfig+storeConfig ctx net dir redis blocks requests =+  StoreConfig+    { maxPeers = 20,+      initPeers = [],+      discover = True,+      db = dir,+      net = net,+      ctx = ctx,+      redis = redis,+      gap = 32,+      initGap = 20,+      redisMinAddrs = 20,+      redisMaxKeys = 100 * 1000 * 1000,+      noMempool = False,+      wipeMempool = False,+      syncMempool = False,+      blockWindow = 64,+      mempoolTimeout = 14,+      peerTimeout = 60,+      maxPeerLife = 48 * 3600,+      connect = mockPeer blocks requests net ad,+      stats = Nothing,+      redisSyncInterval = 30,+      bloom = True,+      maxOpenFiles = Nothing,+      writeBufferSize = Nothing,+      maxWalSize = Nothing,+      blockCache = Just (1024 * 1024 * 1024),+      logger = quiet+    }+  where+    ad = NetworkAddress nodeNetwork (sockToHostAddress (SockAddrInet 0 0))++quiet :: Logger+quiet = Logger {level = LevelError, source = "", push = \_ -> return ()}++webConfig :: Store -> Int -> WebConfig+webConfig store port =+  WebConfig+    { host = "127.0.0.1",+      port = port,+      store = store,+      maxLaggingBlocks = 1000000,+      maxPendingTxs = 1000000,+      minPeers = 0,+      limits = def,+      version = "bench",+      noMempool = False,+      stats = Nothing,+      tickerRefresh = 90 * 1000 * 1000,+      tickerURL = "",+      priceHistoryURL = "",+      noXPub = False,+      noBlockchainInfo = False,+      healthCheckInterval = 30,+      logger = quiet+    }++---------------------+-- Synthetic chain --+---------------------++benchXPub :: Ctx -> XPubKey+benchXPub ctx = deriveXPubKey ctx (makeXPrvKey "haskoin-store benchmark seed")++xpubAddr :: Ctx -> KeyIndex -> KeyIndex -> Address+xpubAddr ctx m i = fst (deriveAddr ctx (pubSubKey ctx (benchXPub ctx) m) i)++hotAddr :: Ctx -> Address+hotAddr ctx = pubKeyAddr ctx (derivePublicKey ctx (wrapSecKey True (fromMaybe (error "key") (secKey (B.replicate 32 0x42)))))++scriptOf :: Ctx -> Address -> ByteString+scriptOf ctx = fromMaybe (error "script") . addressToScriptBS ctx++-- | Blocks at heights 1 to n on top of the regtest genesis.+chain :: Ctx -> Network -> Int -> [Block]+chain ctx net n = go 1 net.genesisHeader Nothing+  where+    hot = scriptOf ctx (hotAddr ctx)+    go h parent prevcb+      | h > n = []+      | otherwise =+          let cb = Tx 1 [TxIn nullOutPoint (encode (fromIntegral h :: Word32)) maxBound] [TxOut 5000000000 hot] [] 0+              txs = maybe [] (spends h) prevcb+              block = mine net parent (cb : txs)+           in block : go (h + 1) block.header (Just (txHash cb))+    spends h cb =+      let t1 = Tx 1 [TxIn (OutPoint cb 0) "" maxBound] (replicate 9 (TxOut 500000000 hot)) [] 0+          tk k =+            let j = fromIntegral (h * 9 + k) :: Word32+                a = xpubAddr ctx (j `mod` 2) ((j `div` 2) `mod` 50)+             in Tx+                  1+                  [TxIn (OutPoint (txHash t1) (fromIntegral k)) "" maxBound]+                  [TxOut 400000000 (scriptOf ctx a), TxOut 99000000 hot]+                  []+                  0+       in t1 : map tk [0 .. 8]++mine :: Network -> BlockHeader -> [Tx] -> Block+mine net parent txs = Block header txs+  where+    merkle = either error id (buildMerkleRoot (map txHash txs))+    candidate = BlockHeader 0x20000000 (headerHash parent) merkle (parent.timestamp + 600) parent.bits+    header =+      fromMaybe (error "no nonce") $+        find (isValidPOW net) (map candidate [0 ..])++-----------------------+-- Microbenchmarks --+-----------------------++-- | Time per call of the pure functions on the encoding path of a response.+-- Each call gets one of 64 different inputs, so that GHC cannot share the+-- result between calls.+micro :: IO ()+micro = do+  ctx <- createContext+  let net = bchRegTest+      blocks = chain ctx net 66+      txs = V.fromList [b.txs !! 2 | b <- drop 2 blocks]+      addrs = V.generate 64 (xpubAddr ctx 0 . fromIntegral)+      hot = scriptOf ctx (hotAddr ctx)+      txdata tx =+        TxData+          { block = BlockRef 3 2,+            tx = tx,+            prevs = IntMap.fromList [(0, Prev hot 500000000)],+            deleted = False,+            rbf = False,+            timestamp = 1700000000,+            spenders = IntMap.fromList [(0, Spender (txHash tx) 0)]+          }+      tds = V.map txdata txs+      tdbytes = V.map encode tds+      ts = V.map (toTransaction net ctx) tds+      pick :: V.Vector a -> Int -> a+      pick v i = v V.! (i `mod` 64)+  _ <- evaluate (force (tds, tdbytes, ts, addrs))+  bench "addrToText cashaddr" (addrToText net . pick addrs)+  bench "addrToText base58" (addrToText btc . pick addrs)+  bench "Address JSON (cashaddr)" (encodingToLazyByteString . marshalEncoding net . pick addrs)+  bench "txHash" (txHash . pick txs)+  bench "TxData decode" (\i -> either error id (decode (pick tdbytes i)) :: TxData)+  bench "toTransaction" (toTransaction net ctx . pick tds)+  bench "Transaction JSON" (encodingToLazyByteString . marshalEncoding net . pick ts)+  bench "TxHash JSON" (encodingToLazyByteString . toEncoding . txHash . pick txs)+  bench "BlockRef JSON" (\i -> encodingToLazyByteString (toEncoding (BlockRef (fromIntegral i) 1)))+  bench "StoreOutput JSON" (encodingToLazyByteString . marshalEncoding net . firstOf . (.outputs) . pick ts)+  bench "StoreInput JSON" (encodingToLazyByteString . marshalEncoding net . firstOf . (.inputs) . pick ts)+  where+    n = 200000+    firstOf = fromMaybe (error "empty") . listToMaybe+    bench :: (NFData a) => String -> (Int -> a) -> IO ()+    bench name f = do+      _ <- evaluate (force (f 0))+      t0 <- getMonotonicTimeNSec+      let go i+            | i >= n = return ()+            | otherwise = evaluate (force (f i)) >> go (i + 1)+      go 0+      t1 <- getMonotonicTimeNSec+      putStrLn $ name <> ": " <> show (fromIntegral (t1 - t0) / fromIntegral n / 1000 :: Double) <> " us"++---------------+-- Mock peer --+---------------++mockPeer :: [Block] -> TVar [Int] -> Network -> NetworkAddress -> SockAddr -> WithConnection+mockPeer blocks _requests net ad sa f = do+  r <- newInbox+  s <- newInbox+  let s' = inboxToMailbox s+  withAsync (go r s') $ \_ -> do+    let o = awaitForever (`send` r)+        i = forever (receive s >>= yield)+    f (Conduits i o) :: IO ()+  where+    heights = HashMap.fromList (zip (map (headerHash . (.header)) blocks) [1 :: Int ..])+    byHash = HashMap.fromList [(headerHash b.header, b) | b <- blocks]+    go :: Inbox ByteString -> Mailbox ByteString -> IO ()+    go r s = do+      nonce <- randomIO+      now <- round <$> liftIO getPOSIXTime+      let rmt = NetworkAddress 0 (sockToHostAddress sa)+          ver = buildVersion net nonce 0 ad rmt now+      runPut (putMessage net (MVersion ver)) `send` s+      runConduit $+        forever (receive r >>= yield)+          .| inc+          .| concatMapC react+          .| mapC (runPut . putMessage net)+          .| awaitForever (`send` s)+    inc = forever $ do+      x <- takeCE 24 .| foldC+      case decode x of+        Left _ -> error "mock peer: bad message header"+        Right (MessageHeader _ _ len _) -> do+          y <- takeCE (fromIntegral len) .| foldC+          case runGet (getMessage net) (x <> y) of+            Right m -> yield m+            Left e -> error ("mock peer: " <> e)+    react (MPing (Ping n)) = [MPong (Pong n)]+    react (MVersion _) = [MVerAck]+    react (MGetHeaders (GetHeaders _ locator _)) =+      let start = fromMaybe 0 (listToMaybe (mapMaybe (`HashMap.lookup` heights) locator))+          hdr :: Block -> (BlockHeader, VarInt)+          hdr b = (b.header, VarInt (fromIntegral (length b.txs)))+       in [MHeaders (Headers (map hdr (take 2000 (drop start blocks))))]+    react (MGetData (GetData ivs)) =+      [MBlock b | InvVector InvBlock h <- ivs, Just b <- [HashMap.lookup (BlockHash h) byHash]]+    react _ = []

@@ -5,7 +5,7 @@ -- see: https://github.com/sol/hpack  name:           haskoin-store-version:        1.7.0+version:        1.8.0 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@@ -52,9 +52,9 @@     , deepseq >=1.4.4.0     , hashable >=1.3.0.0     , hashtables >=1.2.4.2-    , haskoin-core >=1.6.0+    , haskoin-core >=1.6.1     , haskoin-node >=2.3.0-    , haskoin-store-data ==1.7.0+    , haskoin-store-data ==1.8.0     , hedis >=0.16.1     , http-types >=0.12.3     , lens >=4.18.1@@ -63,7 +63,7 @@     , network >=3.1.1.1     , nqe >=0.6.5     , random >=1.1-    , rocksdb-haskell-jprupp >=2.5.0+    , rocksdb-haskell-jprupp >=2.6.0     , rocksdb-query >=0.6.0     , scotty >=0.22     , statsd-rupp >=0.4@@ -88,7 +88,7 @@       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=-N4+  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   build-depends:       aeson >=1.4.7.1     , aeson-pretty >=0.8.8@@ -104,10 +104,10 @@     , filepath     , hashable >=1.3.0.0     , hashtables >=1.2.4.2-    , haskoin-core >=1.6.0+    , haskoin-core >=1.6.1     , haskoin-node >=2.3.0     , haskoin-store-    , haskoin-store-data ==1.7.0+    , haskoin-store-data ==1.8.0     , hedis >=0.16.1     , http-types >=0.12.3     , lens >=4.18.1@@ -117,7 +117,7 @@     , nqe >=0.6.5     , optparse-applicative     , random >=1.1-    , rocksdb-haskell-jprupp >=2.5.0+    , rocksdb-haskell-jprupp >=2.6.0     , rocksdb-query >=0.6.0     , scotty >=0.22     , statsd-rupp >=0.4@@ -142,6 +142,7 @@   other-modules:       Haskoin.Store.BlockStoreSpec       Haskoin.Store.CacheSpec+      Haskoin.Store.CommonSpec       Haskoin.Store.LogicSpec       Haskoin.Store.TestBlocks       Haskoin.Store.WebSpec@@ -149,7 +150,7 @@       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=-N4 -T"+  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 -T"   build-depends:       QuickCheck >=2.13.2     , aeson >=1.4.7.1@@ -167,10 +168,10 @@     , fast-logger >=3.0     , hashable >=1.3.0.0     , hashtables >=1.2.4.2-    , haskoin-core >=1.6.0+    , haskoin-core >=1.6.1     , haskoin-node >=2.3.0     , haskoin-store-    , haskoin-store-data ==1.7.0+    , haskoin-store-data ==1.8.0     , hedis >=0.16.1     , hspec >=2.7.1     , http-types >=0.12.3@@ -180,7 +181,7 @@     , network >=3.1.1.1     , nqe >=0.6.5     , random >=1.1-    , rocksdb-haskell-jprupp >=2.5.0+    , rocksdb-haskell-jprupp >=2.6.0     , rocksdb-query >=0.6.0     , scotty >=0.22     , statsd-rupp >=0.4@@ -199,3 +200,57 @@     , wreq >=0.5.3.2   default-language: Haskell2010   build-tool-depends: hspec-discover:hspec-discover++benchmark haskoin-store-bench+  type: exitcode-stdio-1.0+  main-is: Main.hs+  other-modules:+      Paths_haskoin_store+  hs-source-dirs:+      bench+  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+  build-depends:+      aeson >=1.4.7.1+    , aeson-pretty >=0.8.8+    , base >=4.9 && <5+    , base16 >=2.0+    , bytes >=0.17+    , bytestring >=0.10.10.0+    , cereal >=0.5.8.1+    , conduit >=1.3.2+    , containers >=0.6.2.1+    , data-default >=0.7.1.1+    , deepseq >=1.4.4.0+    , hashable >=1.3.0.0+    , hashtables >=1.2.4.2+    , haskoin-core >=1.6.1+    , haskoin-node >=2.3.0+    , haskoin-store+    , haskoin-store-data ==1.8.0+    , hedis >=0.16.1+    , http-types >=0.12.3+    , lens >=4.18.1+    , monad-control >=1.0.2.3+    , mtl >=2.2.2+    , network >=3.1.1.1+    , nqe >=0.6.5+    , random >=1.1+    , rocksdb-haskell-jprupp >=2.6.0+    , rocksdb-query >=0.6.0+    , scotty >=0.22+    , statsd-rupp >=0.4+    , stm >=2.5.0.0+    , string-conversions >=0.4.0.1+    , text >=1.2.3.0+    , time >=1.9.3+    , transformers >=0.5.6.2+    , unliftio >=0.2.12.1+    , unordered-containers >=0.2.10.0+    , vault >=0.3.1.5+    , vector+    , wai >=3.2.2.1+    , wai-websockets >=3.0.1.2+    , warp >=3.3.10+    , websockets >=0.13+    , wreq >=0.5.3.2+  default-language: Haskell2010

@@ -27,6 +27,7 @@      -- * Useful Fuctions     getTransaction,+    getTransactions,     getDefaultBalance,     getActiveTxData,     blockAtOrBefore,

@@ -399,6 +399,8 @@     runRocksDB . getBlock   getTxData =     runRocksDB . getTxData+  getManyTxData =+    runRocksDB . getManyTxData   getSpender =     runRocksDB . getSpender   getUnspent =

@@ -266,6 +266,7 @@   getBlocksAtHeight = lift . getBlocksAtHeight   getBlock = lift . getBlock   getTxData = lift . getTxData+  getManyTxData = lift . getManyTxData   getSpender = lift . getSpender   getBalance = lift . getBalance   getUnspent = lift . getUnspent@@ -279,6 +280,7 @@   getBlocksAtHeight = lift . getBlocksAtHeight   getBlock = lift . getBlock   getTxData = lift . getTxData+  getManyTxData = lift . getManyTxData   getSpender = lift . getSpender   getBalance = lift . getBalance   getUnspent = lift . getUnspent@@ -349,35 +351,74 @@ xpubLock :: XPubSpec -> ByteString xpubLock xpub = "xlock-" <> encode xpub +-- | Run a read of the cache entries of an xpub in the same round trip to+-- Redis as the check that the xpub is cached. Redis pipelines the two+-- commands, since the reply of the first is not read before the second is+-- sent. The read is 'Nothing' if the xpub is not cached; it then found no+-- entries, as the keys of an xpub are all deleted together.+whenXPubCached ::+  (MonadIO m) =>+  XPubSpec ->+  Redis (Either Reply a) ->+  CacheX m (Maybe a)+whenXPubCached xpub action = do+  (cached, x) <- runRedis $ do+    e <- redisIsXPubCached xpub+    r <- action+    return $ (,) <$> e <*> r+  return $ if cached then Just x else Nothing++-- | The score to start reading a sorted set of an xpub from, when it does+-- not need a database lookup.+startScore :: Maybe Start -> Maybe (Maybe Double)+startScore Nothing = Just Nothing+startScore (Just (AtBlock h)) = Just (Just (blockRefScore (BlockRef h maxBound)))+startScore (Just AtTx {}) = Nothing+ getXPubTxs ::   (MonadUnliftIO m, StoreReadExtra m) =>   XPubSpec ->   [XPubBal] ->   Limits ->   CacheX m [TxRef]-getXPubTxs xpub xbals limits = do-  cached <- isXPubCached xpub-  if cached-    then do-      txs <- cacheGetXPubTxs xpub limits+getXPubTxs xpub xbals limits =+  case startScore limits.start of+    Just score ->+      whenXPubCached xpub (get_txs score) >>= \case+        Just xs -> do+          touchKeys [xpub]+          hit (map (\(t, s) -> TxRef {txid = t, block = scoreBlockRef s}) xs)+        Nothing -> lift $ xPubTxs xpub xbals limits+    Nothing -> do+      cached <- isXPubCached xpub+      if cached+        then hit =<< cacheGetXPubTxs xpub limits+        else lift $ xPubTxs xpub xbals limits+  where+    get_txs score =+      getFromSortedSet+        (txSetPfx <> encode xpub)+        score+        limits.offset+        limits.limit+    hit txs = do       withMetrics $ \s ->         incrementCounter s.xTxs (length txs)       return txs-    else lift $ xPubTxs xpub xbals limits  getXPubTxCount ::   (MonadUnliftIO m, StoreReadExtra m) =>   XPubSpec ->   [XPubBal] ->   CacheX m Word32-getXPubTxCount xpub xbals = do-  cached <- isXPubCached xpub-  if cached-    then do+getXPubTxCount xpub xbals =+  whenXPubCached xpub (redisGetXPubTxCount xpub) >>= \case+    Just count -> do       withMetrics $ \s ->         incrementCounter s.xTxCount 1-      cacheGetXPubTxCount xpub-    else lift $ xPubTxCount xpub xbals+      touchKeys [xpub]+      return (fromInteger count)+    Nothing -> lift $ xPubTxCount xpub xbals  getXPubUnspents ::   (MonadUnliftIO m, StoreReadExtra m) =>@@ -385,11 +426,27 @@   [XPubBal] ->   Limits ->   CacheX m [XPubUnspent]-getXPubUnspents xpub xbals limits = do-  cached <- isXPubCached xpub-  if cached-    then do-      ops <- map snd <$> cacheGetXPubUnspents xpub limits+getXPubUnspents xpub xbals limits =+  case startScore limits.start of+    Just score ->+      whenXPubCached xpub (get_ops score) >>= \case+        Just xs -> do+          touchKeys [xpub]+          hit (map fst xs)+        Nothing -> lift $ xPubUnspents xpub xbals limits+    Nothing -> do+      cached <- isXPubCached xpub+      if cached+        then hit . map snd =<< cacheGetXPubUnspents xpub limits+        else lift $ xPubUnspents xpub xbals limits+  where+    get_ops score =+      getFromSortedSet+        (utxoPfx <> encode xpub)+        score+        limits.offset+        limits.limit+    hit ops = do       uns <- catMaybes <$> lift (mapM getUnspent ops)       net <- lift getNetwork       ctx <- lift getCtx@@ -412,22 +469,20 @@           i a u = h u <$> g a       withMetrics $ \s -> incrementCounter s.xUnspents (length us)       return $ mapMaybe (uncurry i) us-    else lift $ xPubUnspents xpub xbals limits  getXPubBalances ::   (MonadUnliftIO m, StoreReadExtra m) =>   XPubSpec ->   CacheX m [XPubBal]-getXPubBalances xpub = do-  cached <- isXPubCached xpub-  if cached-    then do-      xbals <- cacheGetXPubBalances xpub+getXPubBalances xpub =+  whenXPubCached xpub (redisGetXPubBalances xpub) >>= \case+    Just xbals -> do+      touchKeys [xpub]       withMetrics $ \s -> do         incrementCounter s.hits 1         incrementCounter s.xBalances (length xbals)       return xbals-    else do+    Nothing -> do       withMetrics $ \s -> incrementCounter s.misses 1       withRetryLock (xpubLock xpub) (load_and_cache xpub) >>= \case         Just bals -> return bals@@ -453,21 +508,6 @@  redisIsXPubCached :: (RedisCtx m f) => XPubSpec -> m (f Bool) redisIsXPubCached xpub = Redis.exists (balancesPfx <> encode xpub)--cacheGetXPubBalances :: (MonadIO m) => XPubSpec -> CacheX m [XPubBal]-cacheGetXPubBalances xpub = do-  bals <- runRedis $ redisGetXPubBalances xpub-  touchKeys [xpub]-  return bals--cacheGetXPubTxCount ::-  (MonadUnliftIO m) =>-  XPubSpec ->-  CacheX m Word32-cacheGetXPubTxCount xpub = do-  count <- fromInteger <$> runRedis (redisGetXPubTxCount xpub)-  touchKeys [xpub]-  return count  redisGetXPubTxCount :: (RedisCtx m f) => XPubSpec -> m (f Integer) redisGetXPubTxCount xpub = Redis.zcard (txSetPfx <> encode xpub)

@@ -19,6 +19,7 @@     getActiveTxData,     getDefaultBalance,     getTransaction,+    getTransactions,     getNumTransaction,     blockAtOrAfter,     blockAtOrBefore,@@ -36,6 +37,7 @@     microseconds,     streamThings,     joinDescStreams,+    mergeDescGroups,     loopForever,   ) where@@ -60,7 +62,9 @@ import Data.HashSet qualified as H import Data.Hashable (Hashable) import Data.Map.Strict qualified as Map-import Data.Maybe (mapMaybe)+import Data.List (sortOn)+import Data.Maybe (catMaybes, mapMaybe)+import Data.Ord (Down (..)) import Data.Time.Clock.System   ( getSystemTime,     systemNanoseconds,@@ -120,6 +124,12 @@   getBlocksAtHeight :: BlockHeight -> m [BlockHash]   getBlock :: BlockHash -> m (Maybe BlockData)   getTxData :: TxHash -> m (Maybe TxData)++  -- | Data of many transactions, in the order of their hashes. Stores that+  -- can read many keys at once read them together.+  getManyTxData :: [TxHash] -> m [Maybe TxData]+  getManyTxData = mapM getTxData+   getSpender :: OutPoint -> m (Maybe Spender)   getBalance :: Address -> m (Maybe Balance)   getUnspent :: OutPoint -> m (Maybe Unspent)@@ -212,6 +222,14 @@   ctx <- getCtx   fmap (toTransaction net ctx) <$> getTxData h +-- | Many transactions, in the order of their hashes, read together.+getTransactions ::+  (StoreReadBase m) => [TxHash] -> m [Maybe Transaction]+getTransactions hs = do+  net <- getNetwork+  ctx <- getCtx+  map (fmap (toTransaction net ctx)) <$> getManyTxData hs+ getNumTransaction ::   (StoreReadExtra m) => Word64 -> m [Transaction] getNumTransaction i = do@@ -347,11 +365,17 @@     [] -> return ()     ls -> mapM_ yield ls >> go limits (last ls)   where+    -- Without a transaction to start the next page from, the next page+    -- starts at an offset, which the store reads and drops. Double the+    -- page each time, up to 'maxPage', so that the rows read and dropped+    -- stay in proportion to the rows streamed, instead of growing with+    -- their square.     h l x = case gettx of       Just g -> Just l {offset = 1, start = Just (AtTx (g x))}       Nothing -> case l.limit of         0 -> Nothing-        _ -> Just l {offset = l.offset + l.limit}+        _ -> Just l {offset = l.offset + l.limit, limit = max l.limit (min maxPage (2 * l.limit))}+    maxPage = 1024     go l x = case h l x of       Nothing -> return ()       Just l' ->@@ -383,3 +407,44 @@         let mp2 = Map.deleteMax mp             mp' = Map.unionWith (++) mp1 mp2         go (Just x) mp'++-- | Merge streams whose keys never increase, taking at most the given number+-- of items from each stream ('Nothing': all of them). The items come out in+-- groups of equal key, in decreasing order of key, and each group is sorted+-- in decreasing order of the items themselves. So the output is the same as+-- sorting the first items of every stream in decreasing order, but a stream+-- is read only as far as the consumer needs: a consumer that stops after+-- the first n items makes each stream be read only a little past its first+-- n items, instead of to the end of its quota.+mergeDescGroups ::+  (Monad m, Ord a, Ord k) =>+  (a -> k) ->+  Maybe Word32 ->+  [ConduitT () a m ()] ->+  ConduitT () a m ()+mergeDescGroups key quota streams = do+  heads <- lift $ mapM (next quota . sealConduitT) streams+  go (insertAll (catMaybes heads) Map.empty)+  where+    insertAll hs mp =+      foldr (\h@(x, _, _) -> Map.insertWith (<>) (Down (key x)) [h]) mp hs+    -- the next item of a stream, if any is left within its quota+    next (Just 0) _ = return Nothing+    next q s = do+      (s', mx) <- s $$++ await+      return $ (\x -> (x, subtract 1 <$> q, s')) <$> mx+    -- all the items of a stream with this key, and the first one after them+    group k (x, q, s) = do+      n <- next q s+      case n of+        Just h@(y, _, _)+          | key y == k -> do+              (ys, rest) <- group k h+              return (x : ys, rest)+        _ -> return ([x], n)+    go mp = case Map.minViewWithKey mp of+      Nothing -> return ()+      Just ((Down k, hs), mp') -> do+        gs <- lift $ mapM (group k) hs+        mapM_ yield (sortOn Down (concatMap fst gs))+        go (insertAll (mapMaybe snd gs) mp')

@@ -1,3 +1,4 @@+{-# LANGUAGE BangPatterns #-} {-# LANGUAGE DuplicateRecordFields #-} {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE FlexibleInstances #-}@@ -31,18 +32,20 @@ where  import Conduit-import Control.Monad (forM, forM_)+import Control.Monad (forM, forM_, replicateM, (>=>)) 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.HashMap.Strict (HashMap)+import Data.HashMap.Strict qualified as HashMap import Data.IntMap.Strict qualified as IntMap-import Data.List (sortOn)-import Data.Maybe (fromMaybe, mapMaybe)-import Data.Ord (Down (..))-import Data.Serialize (decode, encode)+import Data.Functor ((<&>))+import Data.Map.Strict qualified as Map+import Data.Maybe (catMaybes, fromMaybe, isJust, mapMaybe)+import Data.Serialize (Serialize, decode, encode) import Data.Text (Text) import Data.Text qualified as T import Data.Word (Word32)@@ -64,9 +67,52 @@     initGap :: !Word32,     net :: !Network,     metrics :: !(Maybe DataMetrics),-    ctx :: !Ctx+    ctx :: !Ctx,+    -- | addresses derived from recently queried xpubs+    derived :: !(IORef DerivedCache)   } +-- | Addresses of an xpub, external then change, in order of index. Each+-- list is infinite and lazy: it is derived as far as queries have needed,+-- and shared by later queries of the same xpub. Deriving an address costs+-- an elliptic curve operation and two hashes, which made up most of the+-- time of an xpub query.+data Derived = Derived ![(Word32, Address)] ![(Word32, Address)]++-- | Recently used xpubs and their addresses. When the newer generation is+-- full it becomes the older one, so the cache holds at most twice+-- 'derivedCacheSize' xpubs, and an xpub stays while it is in use.+data DerivedCache = DerivedCache+  { newer :: !(HashMap XPubSpec Derived),+    older :: !(HashMap XPubSpec Derived)+  }++derivedCacheSize :: Int+derivedCacheSize = 512++-- | Addresses of an xpub, from the cache or a new lazy derivation.+derivedAddrs :: (MonadIO m) => XPubSpec -> DatabaseReaderT m Derived+derivedAddrs xpub = do+  ref <- asks (.derived)+  ctx <- asks (.ctx)+  let fresh =+        let addrs m =+              deriveAddresses+                (deriveFunction ctx xpub.deriv)+                (pubSubKey ctx xpub.key m)+                0+         in Derived (addrs 0) (addrs 1)+  liftIO . atomicModifyIORef' ref $ \c ->+    case HashMap.lookup xpub c.newer of+      Just d -> (c, d)+      Nothing ->+        let d = fromMaybe fresh (HashMap.lookup xpub c.older)+            c'+              | HashMap.size c.newer >= derivedCacheSize =+                  DerivedCache {newer = HashMap.singleton xpub d, older = c.newer}+              | otherwise = c {newer = HashMap.insert xpub d c.newer}+         in (c', d)+ data DataMetrics = DataMetrics   { dataBestCount :: !StatCounter,     dataBlockCount :: !StatCounter,@@ -120,12 +166,14 @@   Maybe Int ->   Maybe Int ->   Maybe Int ->+  Maybe Int ->   Maybe DataMetrics ->   DatabaseReaderT m a ->   m a-withDatabaseReader net ctx igap gap dir bloom maxOpenFiles writeBuffers maxWal stats m =+withDatabaseReader net ctx igap gap dir bloom maxOpenFiles writeBuffers maxWal blockCache stats m =   withRunInIO $ \run ->-    withDBCF dir cfg (columnFamilyConfig bloom) $ \db -> do+    withDBCF dir cfg (columnFamilyConfig bloom (isJust blockCache)) $ \db -> do+      derived <- newIORef (DerivedCache HashMap.empty HashMap.empty)       let bdb =             DatabaseReader               { db = db,@@ -133,7 +181,8 @@                 net = net,                 initGap = igap,                 metrics = stats,-                ctx = ctx+                ctx = ctx,+                derived = derived               }       initRocksDB bdb       run $ runReaderT m bdb@@ -144,7 +193,9 @@           createMissingColumnFamilies = True,           maxFiles = maxOpenFiles,           dbWriteBufferSize = writeBuffers,-          maxTotalWalSize = maxWal+          maxTotalWalSize = maxWal,+          blockCacheSize = blockCache,+          cacheIndexAndFilterBlocks = isJust blockCache         }  -- | Memory that RocksDB uses outside the GHC heap, per column family:@@ -172,25 +223,29 @@   where     cfs =       ("default", Nothing)-        : zip (map (T.pack . fst) (columnFamilyConfig False)) (map Just (columnFamilies db))+        : zip (map (T.pack . fst) (columnFamilyConfig False 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}),-    ("addr-out", def {prefixLength = Just 22, bloomFilter = bloom}),-    ("tx", def {prefixLength = Just 33, bloomFilter = bloom}),-    ("spender", def {prefixLength = Just 33, bloomFilter = bloom}), -- unused-    ("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}),-    ("mempool", def {prefixLength = Nothing, bloomFilter = bloom})+-- | Column families, with Bloom filters or not, and with their index and+-- filter blocks in the block cache or not.+columnFamilyConfig :: Bool -> Bool -> [(String, Config)]+columnFamilyConfig bloom cached =+  [ ("addr-tx", c (Just 22)),+    ("addr-out", c (Just 22)),+    ("tx", c (Just 33)),+    ("spender", c (Just 33)), -- unused+    ("unspent", c (Just 37)),+    ("block", c (Just 33)),+    ("height", c Nothing),+    ("balance", c (Just 22)),+    ("mempool", c Nothing)   ]+  where+    c p = def {prefixLength = p, bloomFilter = bloom, cacheIndexAndFilterBlocks = cached}  addrTxCF :: DB -> ColumnFamily addrTxCF db = columnFamilies db !! 0@@ -394,6 +449,11 @@     go = withRunInIO $ \run ->       withIterCF db cf $ \it -> run (f it) +-- | Run an action with this many iterators on a column family. They are+-- opened and closed in a single bracket rather than one bracket each: each+-- bracket leaves a frame on the stack, and every call into RocksDB through+-- an iterator walks the stack, so with one bracket per address an xpub of a+-- hundred addresses made each step of each iterator a hundred times slower. withManyIters ::   (MonadUnliftIO m) =>   DB ->@@ -401,11 +461,31 @@   Int ->   ([Iterator] -> DatabaseReaderT m a) ->   DatabaseReaderT m a-withManyIters db cf i f = go [] i+withManyIters db cf i f = do+  m <- asks (.metrics)+  bracket_ (gauge m i) (gauge m (negate i)) $+    withRunInIO $ \run ->+      bracket (newIORef []) (readIORef >=> mapM_ destroyIterator) $ \ref -> do+        its <- replicateM i . mask_ $ do+          it <- createIterator db (Just cf)+          modifyIORef' ref (it :)+          return it+        run (f its)   where-    go acc 0 = f acc-    go acc n = withIterCFStat db cf $ \it -> go (it : acc) (n - 1)+    gauge (Just m) n = incrementGauge m.dataIters n+    gauge Nothing _ = return () +-- | Values of many keys of a column family, read with one multi-get, in+-- the order of the keys.+retrieveManyCF ::+  (KeyValue key value, Serialize key, Serialize value) =>+  DB ->+  ColumnFamily ->+  [key] ->+  IO [Maybe value]+retrieveManyCF db cf keys =+  map (fmap (either error id . decode)) <$> multiGetCF db cf (map encode keys)+ joinConduits ::   (Monad m, Ord o) =>   [ConduitT () o m ()] ->@@ -426,6 +506,12 @@         withMetrics $ \s -> incrementCounter s.dataTxCount 1         return (Just t) +  getManyTxData ths = do+    db <- asks (.db)+    tds <- liftIO (retrieveManyCF db (txCF db) (map TxKey ths))+    withMetrics $ \s -> incrementCounter s.dataTxCount (length (catMaybes tds))+    return tds+   getSpender op = runMaybeT $ do     td <- MaybeT $ getTxData op.hash     let i = fromIntegral op.index@@ -539,33 +625,33 @@     forM_ r.metrics $ \s -> incrementCounter s.dataTxCount (length txs)     return txs +  getBalances [] = return []   getBalances as = do-    zipWith f as <$> mapM getBalance as+    db <- asks (.db)+    withMetrics $ \s -> incrementCounter s.dataBalanceCount (length as)+    vals <- liftIO (retrieveManyCF db (balanceCF db) (map BalKey as))+    return (zipWith f as vals)     where       f a Nothing = zeroBalance a-      f _ (Just b) = b+      f a (Just v) = valToBalance a v    xPubBals xpub = do-    ctx <- asks (.ctx)     igap <- asks (.initGap)     gap <- asks (.maxGap)-    ext1 <- derive_until_gap gap 0 (take (fromIntegral igap) (aderiv ctx 0 0))+    Derived ext chg <- derivedAddrs xpub+    ext1 <- derive_until_gap gap 0 (take (fromIntegral igap) ext)     if all nullxb ext1       then do         withMetrics $ \s -> incrementCounter s.dataXPubBals (length ext1)         return ext1       else do-        ext2 <- derive_until_gap gap 0 (aderiv ctx 0 igap)-        chg <- derive_until_gap gap 1 (aderiv ctx 1 0)-        let bals = ext1 <> ext2 <> chg+        ext2 <- derive_until_gap gap 0 (drop (fromIntegral igap) ext)+        chg' <- derive_until_gap gap 1 chg+        let bals = ext1 <> ext2 <> chg'         withMetrics $ \s -> incrementCounter s.dataXPubBals (length bals)         return bals     where       nullxb b = nullBalance b.balance-      aderiv ctx m =-        deriveAddresses-          (deriveFunction ctx xpub.deriv)-          (pubSubKey ctx xpub.key m)       derive_until_gap gap m adrs =         let xb b n = XPubBal {path = [m, n], balance = b}             ig = fromIntegral gap@@ -585,16 +671,29 @@                   go (xs <> acc) as''          in go [] adrs +  -- The first unspent outputs of each address, at most offset plus limit+  -- of them, sorted together. Read the addresses in step, so that each is+  -- read only as far as the page needs, rather than decoding the whole+  -- quota of every address to keep one page of it.   xPubUnspents _xspec xbals limits = do-    us <- concat <$> mapM h cs+    db <- asks (.db)+    net <- asks (.net)+    ctx <- asks (.ctx)+    let quota = case (deOffset limits).limit of+          0 -> Nothing+          l -> Just l+        stream b it =+          unspentConduit net ctx b.balance.address limits.start it+            .| mapC (\u -> XPubUnspent {path = b.path, unspent = u})+    us <- withManyIters db (addrOutCF db) (length cs) $ \its ->+      runConduit $+        mergeDescGroups (.unspent.block) quota (zipWith stream cs its)+          .| applyLimitsC limits+          .| sinkList     withMetrics $ \s -> incrementCounter s.dataXPubUnspents (length us)-    return . applyLimits limits $ sortOn Down us+    return us     where-      l = deOffset limits       cs = filter ((> 0) . (.balance.utxo)) xbals-      i b = getAddressUnspents b.balance.address l-      f b t = XPubUnspent {path = b.path, unspent = t}-      h b = map (f b) <$> i b    xPubTxs _xspec xbals limits = do     let as =@@ -605,8 +704,11 @@     withMetrics $ \s -> incrementCounter s.dataXPubTxs (length txs)     return txs -  -- Count the merged streams instead of building the list of every-  -- transaction of the xpub only to take its length.+  -- Count the transactions of all the addresses, each once even when it+  -- touches several of them. The entries of an address are sorted, and the+  -- same transaction has the same key after the address prefix in every+  -- address, so merge the addresses by those raw key bytes and count the+  -- distinct ones, without decoding any key or reading any value.   xPubTxCount _xspec xbals = do     withMetrics $ \s -> incrementCounter s.dataXPubTxCount 1     db <- asks (.db)@@ -615,5 +717,23 @@             filter (not . nullBalance) $               map (.balance) xbals     withManyIters db (addrTxCF db) (length as) $ \its ->-      runConduit (joinDescStreams (zipWith (`addressConduit` Nothing) as its) .| lengthC)-        >>= \n -> return (fromIntegral (n :: Int))+      liftIO $ fromIntegral <$> countDistinctSuffixes (zip (map (encode . AddrTxKeyA) as) its)++-- | Number of distinct keys, after their prefix, in a set of iterators that+-- each walk the keys with one prefix, in order.+countDistinctSuffixes :: [(BS.ByteString, Iterator)] -> IO Int+countDistinctSuffixes streams = do+  heads <- mapM start streams+  go 0 (Map.fromListWith (<>) [(k, [x]) | (k, x) <- catMaybes heads])+  where+    start x@(pfx, it) = iterSeek it pfx >> current x+    current x@(pfx, it) =+      iterKey it <&> \case+        Just k | pfx `BS.isPrefixOf` k -> Just (BS.drop (BS.length pfx) k, x)+        _ -> Nothing+    advance x@(_, it) = iterNext it >> current x+    go !n m = case Map.minViewWithKey m of+      Nothing -> return n+      Just ((_, xs), m') -> do+        nexts <- catMaybes <$> mapM advance xs+        go (n + 1) (foldr (\(k, x) -> Map.insertWith (<>) k [x]) m' nexts)

@@ -173,6 +173,10 @@     -- | bytes of write-ahead log after which the column families holding the     -- oldest log file are flushed ('Nothing' = RocksDB default)     maxWalSize :: !(Maybe Int),+    -- | bytes of RocksDB block cache shared by all column families, which+    -- also holds their index and filter blocks ('Nothing' = RocksDB default:+    -- 32 MiB per column family, index and filter blocks outside it)+    blockCache :: !(Maybe Int),     -- | log messages here     logger :: !Logger   }@@ -223,6 +227,7 @@     cfg.maxOpenFiles     cfg.writeBufferSize     cfg.maxWalSize+    cfg.blockCache     stats     f 

@@ -458,6 +458,7 @@   getBlocksAtHeight height = runInWebReader (getBlocksAtHeight height)   getBlock bh = runInWebReader (getBlock bh)   getTxData th = runInWebReader (getTxData th)+  getManyTxData ths = runInWebReader (getManyTxData ths)   getSpender op = runInWebReader (getSpender op)   getUnspent op = runInWebReader (getUnspent op)   getBalance a = runInWebReader (getBalance a)@@ -488,6 +489,7 @@   getBlocksAtHeight = lift . getBlocksAtHeight   getBlock = lift . getBlock   getTxData = lift . getTxData+  getManyTxData = lift . getManyTxData   getSpender = lift . getSpender   getUnspent = lift . getUnspent   getBalance = lift . getBalance@@ -972,7 +974,7 @@  toRawBlock :: (StoreReadBase m) => BlockData -> m H.Block toRawBlock b = do-  txs <- map transactionData . catMaybes <$> mapM getTransaction b.txs+  txs <- map transactionData . catMaybes <$> getTransactions b.txs   return H.Block {header = b.header, txs}  -- GET BlockBest / BlockBestRaw --@@ -1093,7 +1095,7 @@   (MonadUnliftIO m) => GetTxs -> ActionT m [Transaction] scottyTxs (GetTxs txids) =   withMetrics (.tx) $-    catMaybes <$> mapM getTransaction (nub txids)+    catMaybes <$> getTransactions (nub txids)  scottyTxRaw ::   (MonadUnliftIO m) => GetTxRaw -> ActionT m (RawResult Tx)@@ -1111,7 +1113,7 @@ scottyTxsRaw (GetTxsRaw txids) =   withMetrics (.txRaw) $     RawResultList . map transactionData . catMaybes-      <$> mapM getTransaction (nub txids)+      <$> getTransactions (nub txids)  getTxsBlock ::   (MonadUnliftIO m) =>@@ -1120,7 +1122,7 @@ getTxsBlock h =   getBlock h >>= \case     Nothing -> raise ThingNotFound-    Just b -> catMaybes <$> mapM getTransaction b.txs+    Just b -> catMaybes <$> getTransactions b.txs  scottyTxsBlock ::   (MonadUnliftIO m) =>@@ -1340,7 +1342,7 @@ scottyAddrTxsFull (GetAddrTxsFull addr pLimits) =   withMetrics (.addressTxFull) $ do     txs <- getAddressTxs addr =<< paramToLimits True pLimits-    catMaybes <$> mapM (getTransaction . (.txid)) txs+    catMaybes <$> getTransactions (map (.txid) txs)  scottyAddrsTxsFull ::   (MonadUnliftIO m) =>@@ -1349,7 +1351,7 @@ scottyAddrsTxsFull (GetAddrsTxsFull addrs pLimits) =   withMetrics (.addressTxFull) $ do     txs <- getAddressesTxs addrs =<< paramToLimits True pLimits-    catMaybes <$> mapM (getTransaction . (.txid)) txs+    catMaybes <$> getTransactions (map (.txid) txs)  scottyAddrBalance :: (MonadUnliftIO m) => GetAddrBalance -> ActionT m Balance scottyAddrBalance (GetAddrBalance addr) =@@ -1417,7 +1419,7 @@ scottyXPubTxsFull (GetXPubTxsFull xpub deriv plimits (NoCache nocache)) =   withMetrics (.xpubTxFull) $     fmap catMaybes $-      lift . runNoCache nocache . mapM (getTransaction . (.txid))+      lift . runNoCache nocache . getTransactions . map (.txid)         =<< getXPubTxs xpub deriv plimits nocache  scottyXPubBalances ::@@ -2219,7 +2221,7 @@     streamEncoding $ marshalEncoding (net, ctx) is   where     get_binfo_blocks numtxid ns b = do-      txs <- catMaybes <$> mapM getTransaction b.txs+      txs <- catMaybes <$> getTransactions b.txs       let h x = H.headerHash x.header           nbs = [h n | n <- ns, n.header.prev == h b]           bts = map (toBinfoTxSimple numtxid) txs@@ -2255,7 +2257,7 @@   where     go numtxid hex bh = do       b <- maybe (raise ThingNotFound) return =<< getBlock bh-      txs <- catMaybes <$> mapM getTransaction b.txs+      txs <- catMaybes <$> getTransactions b.txs       nhs <- fmap catMaybes $ getBlocksAtHeight (b.height + 1) >>= mapM getBlock       let h x = H.headerHash x.header           nxt = [h n | n <- nhs, n.header.prev == h b]@@ -2377,7 +2379,7 @@         else           map snd             <$> getMempoolPage (Limits (fromIntegral n) (fromIntegral offset) Nothing)-    txs <- catMaybes <$> mapM getTransaction txids+    txs <- catMaybes <$> getTransactions txids     net <- askl (.config.store.net)     setHeaders     let mem = BinfoMempool $ map (toBinfoTxSimple numtxid) txs

@@ -0,0 +1,96 @@+{-# LANGUAGE OverloadedRecordDot #-}++module Haskoin.Store.CommonSpec (spec) where++import Conduit (runConduit, runConduitPure, sinkList, takeC, yieldMany, (.|))+import Control.Monad.State.Strict (State, modify', runState)+import Data.List (sortOn)+import Data.Ord (Down (..))+import Data.Word (Word32)+import Haskoin.Store.Common+  ( Limits (..),+    applyLimit,+    applyLimits,+    applyLimitsC,+    deOffset,+    mergeDescGroups,+    streamThings,+  )+import Test.Hspec+import Test.QuickCheck++spec :: Spec+spec = do+  describe "streamThings" $ do+    it "streams every item in order when paging by offset" $+      property $ \(NonNegative n) (Positive page) ->+        let xs = [1 .. n :: Int]+            (ys, _) = paged (fromIntegral (page `mod` 50 + 1 :: Int)) xs Nothing+         in ys === xs+    it "reads a number of rows in proportion to the rows taken" $+      property $ \(NonNegative n) ->+        let xs = [1 .. 100000 :: Int]+            (ys, reads) = paged 16 xs (Just n)+         in ys === take n xs .&&. counterexample (show reads) (reads <= 4 * n + 2048)+  describe "mergeDescGroups" $ do+    it "pages like sorting the first offset plus limit items of every stream" $+      property $ \(Streams ss) ->+        forAll (choose (0, 8)) $ \l ->+          forAll (choose (0, 8)) $ \o ->+            let limits = Limits l o Nothing+             in merged limits ss === reference limits ss+    it "stops reading once the page is full" $+      -- an undefined item after the first key of each stream would throw+      let ss = [[(9, 1), (8, 2)] <> undefined, [(9, 3), (7, 4)] <> undefined]+          limits = Limits 2 0 Nothing+       in merged limits ss `shouldBe` [(9, 3), (9, 1)]++-- | The paging of xpub unspent outputs before 'mergeDescGroups': the first+-- offset plus limit items of each stream, all sorted, then the page.+reference :: Limits -> [[(Int, Int)]] -> [(Int, Int)]+reference limits =+  applyLimits limits . sortOn Down . concatMap (applyLimit (deOffset limits).limit)++merged :: Limits -> [[(Int, Int)]] -> [(Int, Int)]+merged limits ss =+  runConduitPure $+    mergeDescGroups fst quota (map yieldMany ss)+      .| applyLimitsC limits+      .| sinkList+  where+    quota = case (deOffset limits).limit of+      0 -> Nothing+      q -> Just (q :: Word32)++-- | Streams of (key, item) whose keys never increase, with many equal keys+-- and the items of each key in any order, like the unspent outputs of the+-- addresses of an xpub: keyed by block, in key order within a block.+-- Items are unique across all streams, as outputs are.+newtype Streams = Streams [[(Int, Int)]]+  deriving (Show)++instance Arbitrary Streams where+  arbitrary = do+    n <- choose (0, 6)+    ss <- vectorOf n $ do+      keys <- sortOn Down <$> listOf (choose (0, 5))+      items <- shuffle [0 .. length keys - 1]+      return (zip keys items)+    -- number the items so that they are unique across streams+    return . Streams $+      zipWith (\i -> map (\(k, x) -> (k, x * 10 + i))) [0 ..] ss++-- | Stream a list through 'streamThings' paging by offset, taking at most+-- the given number of items. Also gives the rows that the store read,+-- counting those it skipped to reach the offset of each page.+paged :: Word32 -> [Int] -> Maybe Int -> ([Int], Int)+paged page xs n =+  flip runState 0 . runConduit $+    streamThings getit Nothing (Limits page 0 Nothing)+      .| maybe sinkList (\k -> takeC k .| sinkList) n+  where+    getit :: Limits -> State Int [Int]+    getit l = do+      let ys = applyLimits l xs+      modify' (+ (fromIntegral l.offset + length ys))+      return ys

@@ -82,7 +82,7 @@     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+            open = withDatabaseReader btcRegTest ctx 20 32 dir False Nothing Nothing Nothing Nothing Nothing             h1 = TxHash (sha256 ("one" :: B.ByteString))             h2 = TxHash (sha256 ("two" :: B.ByteString))             old = [(200, h2), (100, h1)] :: [(Word64, TxHash)]@@ -99,7 +99,7 @@     it "indexes the mempool by time 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+            open = withDatabaseReader btcRegTest ctx 20 32 dir False Nothing Nothing Nothing Nothing Nothing             old = sortOn Down (zip [300, 100, 200] (hashes 3))         open $ ask >>= \r -> liftIO $ do           insert r.db VersionKey (19 :: Word32)@@ -398,7 +398,7 @@ 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)+    withDatabaseReader net ctx 20 32 dir False Nothing 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.

@@ -168,6 +168,7 @@               maxOpenFiles = Nothing,               writeBufferSize = Nothing,               maxWalSize = Nothing,+              blockCache = Nothing,               logger = Logger {level = LevelError, source = "", push = \_ -> return ()}             }     withStore cfg $ \store ->