diff --git a/app/Main.hs b/app/Main.hs --- a/app/Main.hs +++ b/app/Main.hs @@ -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 } diff --git a/bench/Main.hs b/bench/Main.hs new file mode 100644 --- /dev/null +++ b/bench/Main.hs @@ -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 _ = [] diff --git a/haskoin-store.cabal b/haskoin-store.cabal --- a/haskoin-store.cabal +++ b/haskoin-store.cabal @@ -5,7 +5,7 @@ -- see: https://github.com/sol/hpack name: haskoin-store -version: 1.7.0 +version: 1.8.0 synopsis: Storage and index for Bitcoin and Bitcoin Cash description: Please see the README on GitHub at 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 diff --git a/src/Haskoin/Store.hs b/src/Haskoin/Store.hs --- a/src/Haskoin/Store.hs +++ b/src/Haskoin/Store.hs @@ -27,6 +27,7 @@ -- * Useful Fuctions getTransaction, + getTransactions, getDefaultBalance, getActiveTxData, blockAtOrBefore, diff --git a/src/Haskoin/Store/BlockStore.hs b/src/Haskoin/Store/BlockStore.hs --- a/src/Haskoin/Store/BlockStore.hs +++ b/src/Haskoin/Store/BlockStore.hs @@ -399,6 +399,8 @@ runRocksDB . getBlock getTxData = runRocksDB . getTxData + getManyTxData = + runRocksDB . getManyTxData getSpender = runRocksDB . getSpender getUnspent = diff --git a/src/Haskoin/Store/Cache.hs b/src/Haskoin/Store/Cache.hs --- a/src/Haskoin/Store/Cache.hs +++ b/src/Haskoin/Store/Cache.hs @@ -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) diff --git a/src/Haskoin/Store/Common.hs b/src/Haskoin/Store/Common.hs --- a/src/Haskoin/Store/Common.hs +++ b/src/Haskoin/Store/Common.hs @@ -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') diff --git a/src/Haskoin/Store/Database/Reader.hs b/src/Haskoin/Store/Database/Reader.hs --- a/src/Haskoin/Store/Database/Reader.hs +++ b/src/Haskoin/Store/Database/Reader.hs @@ -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) diff --git a/src/Haskoin/Store/Manager.hs b/src/Haskoin/Store/Manager.hs --- a/src/Haskoin/Store/Manager.hs +++ b/src/Haskoin/Store/Manager.hs @@ -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 diff --git a/src/Haskoin/Store/Web.hs b/src/Haskoin/Store/Web.hs --- a/src/Haskoin/Store/Web.hs +++ b/src/Haskoin/Store/Web.hs @@ -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 diff --git a/test/Haskoin/Store/CommonSpec.hs b/test/Haskoin/Store/CommonSpec.hs new file mode 100644 --- /dev/null +++ b/test/Haskoin/Store/CommonSpec.hs @@ -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 diff --git a/test/Haskoin/Store/LogicSpec.hs b/test/Haskoin/Store/LogicSpec.hs --- a/test/Haskoin/Store/LogicSpec.hs +++ b/test/Haskoin/Store/LogicSpec.hs @@ -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. diff --git a/test/Haskoin/StoreSpec.hs b/test/Haskoin/StoreSpec.hs --- a/test/Haskoin/StoreSpec.hs +++ b/test/Haskoin/StoreSpec.hs @@ -168,6 +168,7 @@ maxOpenFiles = Nothing, writeBufferSize = Nothing, maxWalSize = Nothing, + blockCache = Nothing, logger = Logger {level = LevelError, source = "", push = \_ -> return ()} } withStore cfg $ \store ->