packages feed

haskoin-store 0.17.1 → 0.17.2

raw patch · 9 files changed

+158/−108 lines, 9 filesdep +warpdep ~rocksdb-query

Dependencies added: warp

Dependency ranges changed: rocksdb-query

Files

CHANGELOG.md view
@@ -4,6 +4,11 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/) and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.html). +## 0.17.2+### Changed+- Stream address balances and unspent outputs.+- Add configurable max value for limit and offset which defaults at 10000.+ ## 0.17.1 ### Changed - When posting a transaction to the network, timeout is now five seconds.
app/Main.hs view
@@ -27,6 +27,7 @@ import           Data.String.Conversions import qualified Data.Text.Lazy             as T import           Data.Version+import           Data.Word                  (Word32) import           Database.RocksDB           as R import           Haskoin import           Haskoin.Node@@ -41,7 +42,7 @@ import           Text.Read                  (readMaybe) import           UnliftIO import           UnliftIO.Directory-import           Web.Scotty.Trans           as S+import           Web.Scotty.Trans  data Config = Config     { configDir      :: !FilePath@@ -52,6 +53,7 @@     , configVersion  :: !Bool     , configCache    :: !FilePath     , configDebug    :: !Bool+    , configMaxCount :: !Word32     }  defPort :: Int@@ -63,6 +65,9 @@ netNames :: String netNames = intercalate "|" (map getNetworkName allNets) +defMaxCount :: Word32+defMaxCount = 10000+ config :: Parser Config config = do     configDir <-@@ -81,9 +86,7 @@         help "Network to connect to" <>         showDefault <>         value defNetwork-    configDiscover <--        switch $-        long "auto" <> short 'a' <> help "Peer discovery"+    configDiscover <- switch $ long "auto" <> short 'a' <> help "Peer discovery"     configPeers <-         many . option (eitherReader peerReader) $         metavar "HOST" <> long "peer" <> short 'p' <>@@ -92,10 +95,14 @@         option str $         long "cache" <> short 'c' <> help "Memory mapped disk for cache" <>         value ""-    configVersion <--        switch $ long "version" <> short 'v' <> help "Show version"-    configDebug <--        switch $ long "debug" <> help "Show debug messages"+    configVersion <- switch $ long "version" <> short 'v' <> help "Show version"+    configDebug <- switch $ long "debug" <> help "Show debug messages"+    configMaxCount <-+        option auto $+        metavar "INT" <> long "max" <> short 'x' <>+        help "Max offset or length (0 for no limit)" <>+        showDefault <>+        value defMaxCount     return Config {..}  networkReader :: String -> Either String Network@@ -154,6 +161,7 @@            , configCache = cache_path            , configDir = db_dir            , configDebug = deb+           , configMaxCount = max_count            } =     runStderrLoggingT . filterLogger l . flip finally clear $ do         $(logInfoS) "Main" $@@ -209,6 +217,7 @@                                 , webDB = ldb                                 , webPublisher = pub                                 , webStore = str+                                , webMaxCount = max_count                                 }                      in runWeb wcfg   where
haskoin-store.cabal view
@@ -4,10 +4,10 @@ -- -- see: https://github.com/sol/hpack ----- hash: 606922c9587a5c632a26585ce6c4e65dfbb82928fd1aad63ace29554f1aa2651+-- hash: fe384b3ea5c3ecda43c35136aaabfb52c82bf6b08e13da5e1d98852039cdd315  name:           haskoin-store-version:        0.17.1+version:        0.17.2 synopsis:       Storage and index for Bitcoin and Bitcoin Cash description:    Store blocks, transactions, and balances for Bitcoin or Bitcoin Cash, and make that information via REST API. category:       Bitcoin, Finance, Network@@ -64,7 +64,7 @@     , random     , resourcet     , rocksdb-haskell-    , rocksdb-query+    , rocksdb-query >=0.3.0     , scotty     , string-conversions     , text@@ -73,6 +73,7 @@     , unliftio     , unordered-containers     , wai+    , warp   default-language: Haskell2010  executable haskoin-store@@ -81,7 +82,7 @@       Paths_haskoin_store   hs-source-dirs:       app-  ghc-options: -threaded -rtsopts -with-rtsopts=-N+  ghc-options: -threaded -rtsopts -with-rtsopts=-N -O2   build-depends:       aeson     , base >=4.9 && <5@@ -104,7 +105,7 @@     , random     , resourcet     , rocksdb-haskell-    , rocksdb-query+    , rocksdb-query >=0.3.0     , scotty     , string-conversions     , text@@ -113,6 +114,7 @@     , unliftio     , unordered-containers     , wai+    , warp   default-language: Haskell2010  test-suite haskoin-store-test@@ -124,7 +126,7 @@       Paths_haskoin_store   hs-source-dirs:       test-  ghc-options: -threaded -rtsopts -with-rtsopts=-N+  ghc-options: -threaded -rtsopts -with-rtsopts=-N -O2   build-depends:       aeson     , base >=4.9 && <5@@ -146,7 +148,7 @@     , random     , resourcet     , rocksdb-haskell-    , rocksdb-query+    , rocksdb-query >=0.3.0     , scotty     , string-conversions     , text@@ -155,5 +157,6 @@     , unliftio     , unordered-containers     , wai+    , warp   default-language: Haskell2010   build-tool-depends: hspec-discover:hspec-discover
src/Network/Haskoin/Store/Data.hs view
@@ -107,12 +107,12 @@     return $ toTransaction d sm  class StoreStream m where-    getMempool :: Maybe UnixTime -> ConduitT () (UnixTime, TxHash) m ()-    getOrphans :: ConduitT () (UnixTime, Tx) m ()-    getAddressUnspents :: Address -> Maybe BlockRef -> ConduitT () Unspent m ()-    getAddressTxs :: Address -> Maybe BlockRef -> ConduitT () BlockTx m ()-    getAddressBalances :: ConduitT () Balance m ()-    getUnspents :: ConduitT () Unspent m ()+    getMempool :: Maybe UnixTime -> ConduitT i (UnixTime, TxHash) m ()+    getOrphans :: ConduitT i (UnixTime, Tx) m ()+    getAddressUnspents :: Address -> Maybe BlockRef -> ConduitT i Unspent m ()+    getAddressTxs :: Address -> Maybe BlockRef -> ConduitT i BlockTx m ()+    getAddressBalances :: ConduitT i Balance m ()+    getUnspents :: ConduitT i Unspent m ()  -- | Serialize such that ordering is inverted. putUnixTime w = putWord64be $ maxBound - w
src/Network/Haskoin/Store/Data/Cached.hs view
@@ -74,27 +74,27 @@ getUnspentC op LayeredDB {layeredDB = db}         = getUnspentDB op db  getUnspentsC ::-       (MonadResource m, MonadIO m) => LayeredDB -> ConduitT () Unspent m ()+       (MonadResource m, MonadIO m) => LayeredDB -> ConduitT i Unspent m () getUnspentsC LayeredDB {layeredDB = db} = getUnspentsDB db  getMempoolC ::        (MonadResource m, MonadUnliftIO m)     => Maybe UnixTime     -> LayeredDB-    -> ConduitT () (UnixTime, TxHash) m ()+    -> ConduitT i (UnixTime, TxHash) m () getMempoolC mpu LayeredDB {layeredCache = Just db} = getMempoolDB mpu db getMempoolC mpu LayeredDB {layeredDB = db}         = getMempoolDB mpu db  getOrphansC ::        (MonadUnliftIO m, MonadResource m)     => LayeredDB-    -> ConduitT () (UnixTime, Tx) m ()+    -> ConduitT i (UnixTime, Tx) m () getOrphansC LayeredDB {layeredDB = db} = getOrphansDB db  getAddressBalancesC ::        (MonadUnliftIO m, MonadResource m)     => LayeredDB-    -> ConduitT () Balance m ()+    -> ConduitT i Balance m () getAddressBalancesC LayeredDB {layeredDB = db} = getAddressBalancesDB db  getAddressUnspentsC ::@@ -102,7 +102,7 @@     => Address     -> Maybe BlockRef     -> LayeredDB-    -> ConduitT () Unspent m ()+    -> ConduitT i Unspent m () getAddressUnspentsC addr mbr LayeredDB {layeredDB = db} =     getAddressUnspentsDB addr mbr db @@ -111,7 +111,7 @@     => Address     -> Maybe BlockRef     -> LayeredDB-    -> ConduitT () BlockTx m ()+    -> ConduitT i BlockTx m () getAddressTxsC addr mbr LayeredDB {layeredDB = db} =     getAddressTxsDB addr mbr db
src/Network/Haskoin/Store/Data/ImportDB.hs view
@@ -318,7 +318,7 @@        MonadIO m     => Maybe UnixTime     -> ImportDB-    -> ConduitT () (UnixTime, TxHash) m ()+    -> ConduitT i (UnixTime, TxHash) m () getMempoolI mpu ImportDB {importHashMap = hm, importLayeredDB = db} = do     h <- hMempool <$> readTVarIO hm     let hmap =
src/Network/Haskoin/Store/Data/Memory.hs view
@@ -87,7 +87,7 @@        Monad m     => Maybe UnixTime     -> BlockMem-    -> ConduitT () (UnixTime, TxHash) m ()+    -> ConduitT i (UnixTime, TxHash) m () getMempoolH mpu db =     let f ts =             case mpu of@@ -99,13 +99,13 @@             hMempool db      in yieldMany [(u, h) | (u, hs) <- ls, h <- hs] -getOrphansH :: Monad m => BlockMem -> ConduitT () (UnixTime, Tx) m ()+getOrphansH :: Monad m => BlockMem -> ConduitT i (UnixTime, Tx) m () getOrphansH = yieldMany . catMaybes . M.elems . hOrphans  getOrphanTxH :: TxHash -> BlockMem -> Maybe (Maybe (UnixTime, Tx)) getOrphanTxH h = M.lookup h . hOrphans -getUnspentsH :: Monad m => BlockMem -> ConduitT () Unspent m ()+getUnspentsH :: Monad m => BlockMem -> ConduitT i Unspent m () getUnspentsH BlockMem {hUnspent = us} =     yieldMany         [ u@@ -133,7 +133,7 @@             Nothing -> False             Just br -> b > br -getAddressBalancesH :: Monad m => BlockMem -> ConduitT () Balance m ()+getAddressBalancesH :: Monad m => BlockMem -> ConduitT i Balance m () getAddressBalancesH BlockMem {hBalance = bm} =     yieldMany (M.toList bm) .| mapC (uncurry balValToBalance)
src/Network/Haskoin/Store/Data/RocksDB.hs view
@@ -70,7 +70,7 @@        (MonadIO m, MonadResource m)     => Maybe UnixTime     -> BlockDB-    -> ConduitT () (UnixTime, TxHash) m ()+    -> ConduitT i (UnixTime, TxHash) m () getMempoolDB mpu BlockDB {blockDBopts = opts, blockDB = db} =     x .| mapC (uncurry f)   where@@ -84,7 +84,7 @@ getOrphansDB ::        (MonadIO m, MonadResource m)     => BlockDB-    -> ConduitT () (UnixTime, Tx) m ()+    -> ConduitT i (UnixTime, Tx) m () getOrphansDB BlockDB {blockDBopts = opts, blockDB = db} =     matching db opts OrphanKeyS .| mapC snd @@ -97,7 +97,7 @@     => Address     -> Maybe BlockRef     -> BlockDB-    -> ConduitT () BlockTx m ()+    -> ConduitT i BlockTx m () getAddressTxsDB a mbr BlockDB {blockDBopts = opts, blockDB = db} =     x .| mapC (uncurry f)   where@@ -111,14 +111,14 @@ getAddressBalancesDB ::        (MonadIO m, MonadResource m)     => BlockDB-    -> ConduitT () Balance m ()+    -> ConduitT i Balance m () getAddressBalancesDB BlockDB {blockDBopts = opts, blockDB = db} =     matching db opts BalKeyS .| mapC (\(BalKey a, b) -> balValToBalance a b)  getUnspentsDB ::        (MonadIO m, MonadResource m)     => BlockDB-    -> ConduitT () Unspent m ()+    -> ConduitT i Unspent m () getUnspentsDB BlockDB {blockDBopts = opts, blockDB = db} =     matching db opts UnspentKeyB .|     mapC (\(UnspentKey k, v) -> unspentFromDB k v)@@ -132,7 +132,7 @@     => Address     -> Maybe BlockRef     -> BlockDB-    -> ConduitT () Unspent m ()+    -> ConduitT i Unspent m () getAddressUnspentsDB a mbr BlockDB {blockDBopts = opts, blockDB = db} =     x .| mapC (uncurry f)   where
src/Network/Haskoin/Store/Web.hs view
@@ -24,6 +24,7 @@ import qualified Data.ByteString.Lazy              as L import qualified Data.ByteString.Lazy.Char8        as C import           Data.Char+import           Data.Default import           Data.Foldable import           Data.Function import qualified Data.HashMap.Strict               as H@@ -46,8 +47,10 @@ import           Network.Haskoin.Store.Messages import           Network.HTTP.Types import           Network.Wai+import           Network.Wai.Handler.Warp import           NQE import qualified Paths_haskoin_store               as P+import           Text.Printf import           Text.Read                         (readMaybe) import           UnliftIO import           UnliftIO.Resource@@ -107,6 +110,7 @@         , webDB        :: !LayeredDB         , webPublisher :: !(Publisher StoreEvent)         , webStore     :: !Store+        , webMaxCount  :: !Word32         }  instance Parsable BlockHash where@@ -261,10 +265,11 @@             streamAny net proto io         flush' -scottyMempool :: (MonadLoggerIO m, MonadUnliftIO m) => Network -> WebT m ()-scottyMempool net = do+scottyMempool ::+       (MonadLoggerIO m, MonadUnliftIO m) => Network -> Word32 -> WebT m ()+scottyMempool net max_count = do     cors-    (l, s) <- parseLimits+    (l, s) <- parseLimits max_count     proto <- setupBin     db <- askDB     stream $ \io flush' -> do@@ -354,11 +359,11 @@             raise ThingNotFound  scottyAddressTxs ::-       (MonadLoggerIO m, MonadUnliftIO m) => Network -> Bool -> WebT m ()-scottyAddressTxs net full = do+       (MonadLoggerIO m, MonadUnliftIO m) => Network -> Word32 -> Bool -> WebT m ()+scottyAddressTxs net max_count full = do     cors     a <- parseAddress net-    (l, s) <- parseLimits+    (l, s) <- parseLimits max_count     proto <- setupBin     db <- askDB     stream $ \io flush' -> do@@ -370,11 +375,11 @@         | otherwise = getAddressTxsLimit l s a .| streamAny net proto io  scottyAddressesTxs ::-       (MonadLoggerIO m, MonadUnliftIO m) => Network -> Bool -> WebT m ()-scottyAddressesTxs net full = do+       (MonadLoggerIO m, MonadUnliftIO m) => Network -> Word32 -> Bool -> WebT m ()+scottyAddressesTxs net max_count full = do     cors     as <- parseAddresses net-    (l, s) <- parseLimits+    (l, s) <- parseLimits max_count     proto <- setupBin     db <- askDB     stream $ \io flush' -> do@@ -386,11 +391,11 @@         | otherwise = getAddressesTxsLimit l s as .| streamAny net proto io  scottyAddressUnspent ::-       (MonadLoggerIO m, MonadUnliftIO m) => Network -> WebT m ()-scottyAddressUnspent net = do+       (MonadLoggerIO m, MonadUnliftIO m) => Network -> Word32 -> WebT m ()+scottyAddressUnspent net max_count = do     cors     a <- parseAddress net-    (l, s) <- parseLimits+    (l, s) <- parseLimits max_count     proto <- setupBin     db <- askDB     stream $ \io flush' -> do@@ -399,11 +404,11 @@         flush'  scottyAddressesUnspent ::-       (MonadLoggerIO m, MonadUnliftIO m) => Network -> WebT m ()-scottyAddressesUnspent net = do+       (MonadLoggerIO m, MonadUnliftIO m) => Network -> Word32 -> WebT m ()+scottyAddressesUnspent net max_count = do     cors     as <- parseAddresses net-    (l, s) <- parseLimits+    (l, s) <- parseLimits max_count     proto <- setupBin     db <- askDB     stream $ \io flush' -> do@@ -436,6 +441,7 @@     cors     as <- parseAddresses net     proto <- setupBin+    db <- askDB     let f a Nothing =             Balance                 { balanceAddress = a@@ -446,8 +452,11 @@                 , balanceTotalReceived = 0                 }         f _ (Just b) = b-    res <- mapM (\a -> f a <$> getBalance a) as-    protoSerial net proto res+    stream $ \io flush' -> do+        runResourceT . withLayeredDB db . runConduit $+            yieldMany as .| mapMC (\a -> f a <$> getBalance a) .|+            streamAny net proto io+        flush'  scottyXpubBalances :: (MonadUnliftIO m, MonadLoggerIO m) => Network -> WebT m () scottyXpubBalances net = do@@ -455,18 +464,26 @@     xpub <- parseXpub net     proto <- setupBin     db <- askDB-    res <- liftIO . runResourceT . withLayeredDB db $ xpubBals xpub-    protoSerial net proto res+    stream $ \io flush' -> do+        runResourceT . withLayeredDB db . runConduit $+            xpubBals xpub .| streamAny net proto io+        flush'  scottyXpubTxs ::-       (MonadLoggerIO m, MonadUnliftIO m) => Network -> Bool -> WebT m ()-scottyXpubTxs net full = do+       (MonadLoggerIO m, MonadUnliftIO m)+    => Network+    -> Word32+    -> Bool+    -> WebT m ()+scottyXpubTxs net max_count full = do     cors     x <- parseXpub net-    (l, s) <- parseLimits+    (l, s) <- parseLimits max_count     proto <- setupBin     db <- askDB-    bs <- liftIO . runResourceT . withLayeredDB db $ xpubBals x+    bs <-+        liftIO . runResourceT . withLayeredDB db $+        runConduit $ xpubBals x .| sinkList     stream $ \io flush' -> do         runResourceT . withLayeredDB db . runConduit $ f proto l s bs io         flush'@@ -479,23 +496,24 @@             getAddressesTxsLimit l s (map (balanceAddress . xPubBal) bs) .|             streamAny net proto io -scottyXpubUnspents :: MonadLoggerIO m => Network -> WebT m ()-scottyXpubUnspents net = do+scottyXpubUnspents :: MonadLoggerIO m => Network -> Word32 -> WebT m ()+scottyXpubUnspents net max_count = do     cors     x <- parseXpub net     proto <- setupBin-    (l, s) <- parseLimits+    (l, s) <- parseLimits max_count     db <- askDB     stream $ \io flush' -> do         runResourceT . withLayeredDB db . runConduit $             xpubUnspentLimit net l s x .| streamAny net proto io         flush' -scottyXpubSummary :: (MonadLoggerIO m, MonadUnliftIO m) => Network -> WebT m ()-scottyXpubSummary net = do+scottyXpubSummary ::+       (MonadLoggerIO m, MonadUnliftIO m) => Network -> Word32 -> WebT m ()+scottyXpubSummary net max_count = do     cors     x <- parseXpub net-    (l, s) <- parseLimits+    (l, s) <- parseLimits max_count     proto <- setupBin     db <- askDB     res <- liftIO . runResourceT . withLayeredDB db $ xpubSummary l s x@@ -590,10 +608,12 @@                  , webNetwork = net                  , webStore = st                  , webPublisher = pub+                 , webMaxCount = max_count                  } = do     runner <- askRunInIO+    logger <- askRunInIO     l <- logIt-    scottyT port (runner . withLayeredDB db) $ do+    scottyOptsT (opts logger) (runner . withLayeredDB db) $ do         middleware l         defaultHandler (defHandler net)         S.get "/block/best" $ scottyBestBlock net@@ -602,7 +622,7 @@         S.get "/block/heights" $ scottyBlockHeights net         S.get "/block/latest" $ scottyBlockLatest net         S.get "/blocks" $ scottyBlocks net-        S.get "/mempool" $ scottyMempool net+        S.get "/mempool" $ scottyMempool net max_count         S.get "/transaction/:txid" $ scottyTransaction net         S.get "/transaction/:txid/raw" $ scottyRawTransaction net         S.get "/transaction/:txid/after/:height" $ scottyTxAfterHeight net@@ -610,28 +630,41 @@         S.get "/transactions/raw" $ scottyRawTransactions net         S.get "/transactions/block/:block" $ scottyBlockTransactions net         S.get "/transactions/block/:block/raw" $ scottyRawBlockTransactions net-        S.get "/address/:address/transactions" $ scottyAddressTxs net False-        S.get "/address/:address/transactions/full" $ scottyAddressTxs net True-        S.get "/address/transactions" $ scottyAddressesTxs net False-        S.get "/address/transactions/full" $ scottyAddressesTxs net True-        S.get "/address/:address/unspent" $ scottyAddressUnspent net-        S.get "/address/unspent" $ scottyAddressesUnspent net+        S.get "/address/:address/transactions" $+            scottyAddressTxs net max_count False+        S.get "/address/:address/transactions/full" $+            scottyAddressTxs net max_count True+        S.get "/address/transactions" $ scottyAddressesTxs net max_count False+        S.get "/address/transactions/full" $+            scottyAddressesTxs net max_count True+        S.get "/address/:address/unspent" $ scottyAddressUnspent net max_count+        S.get "/address/unspent" $ scottyAddressesUnspent net max_count         S.get "/address/:address/balance" $ scottyAddressBalance net         S.get "/address/balances" $ scottyAddressesBalances net         S.get "/xpub/:xpub/balances" $ scottyXpubBalances net-        S.get "/xpub/:xpub/transactions" $ scottyXpubTxs net False-        S.get "/xpub/:xpub/transactions/full" $ scottyXpubTxs net True-        S.get "/xpub/:xpub/unspent" $ scottyXpubUnspents net-        S.get "/xpub/:xpub" $ scottyXpubSummary net+        S.get "/xpub/:xpub/transactions" $ scottyXpubTxs net max_count False+        S.get "/xpub/:xpub/transactions/full" $ scottyXpubTxs net max_count True+        S.get "/xpub/:xpub/unspent" $ scottyXpubUnspents net max_count+        S.get "/xpub/:xpub" $ scottyXpubSummary net max_count         S.post "/transactions" $ scottyPostTx net st pub         S.get "/dbstats" scottyDbStats         S.get "/events" $ scottyEvents net pub         S.get "/peers" $ scottyPeers net st         S.get "/health" $ scottyHealth net st         notFound $ raise ThingNotFound+  where+    opts runner = def {settings = setPort port (setOnOpen f defaultSettings)}+      where+        f s =+            runner $ do+                $(logDebugS) "Web" $ "Incoming connection: " <> cs (show s)+                return True -parseLimits :: (ScottyError e, Monad m) => ActionT e m (Maybe Word32, StartFrom)-parseLimits = do+parseLimits ::+       (ScottyError e, Monad m)+    => Word32+    -> ActionT e m (Maybe Word32, StartFrom)+parseLimits max_count = do     let b = do             height <- param "height"             pos <- param "pos" `rescue` const (return maxBound)@@ -641,8 +674,11 @@             return $ StartMem time         o = do             o <- param "offset" `rescue` const (return 0)-            return $ StartOffset o-    l <- (Just <$> param "limit") `rescue` const (return Nothing)+            let o' = if max_count == 0 then o else min o max_count+            return $ StartOffset o'+    l <- runMaybeT $ do+        l <- MaybeT $ (Just <$> param "limit") `rescue` const (return Nothing)+        return $ if max_count == 0 then l else min l max_count     s <- b <|> m <|> o     return (l, s) @@ -777,12 +813,12 @@                 }  xpubBals ::-       (MonadResource m, MonadUnliftIO m, StoreRead m) => XPubKey -> m [XPubBal]-xpubBals xpub = (<>) <$> go 0 <*> go 1+       (MonadResource m, MonadUnliftIO m, StoreRead m)+    => XPubKey+    -> ConduitT i XPubBal m ()+xpubBals xpub = go 0 >> go 1   where-    go m =-        runConduit $-        yieldMany (addrs m) .| mapMC (uncurry bal) .| gap 20 .| sinkList+    go m = yieldMany (addrs m) .| mapMC (uncurry bal) .| gap 20     bal a p =         getBalance a >>= \case             Nothing -> return Nothing@@ -807,16 +843,13 @@     => Network     -> Maybe BlockRef     -> XPubKey-    -> ConduitT () XPubUnspent m ()-xpubUnspent net mbr xpub = do-    xs <- do-        bs <- lift $ xpubBals xpub-        return $ map g bs-    mergeSourcesBy (flip compare `on` (unspentBlock . xPubUnspent)) xs+    -> ConduitT i XPubUnspent m ()+xpubUnspent net mbr xpub = xpubBals xpub .| go   where-    f p t = XPubUnspent {xPubUnspentPath = p, xPubUnspent = t}-    g XPubBal {xPubBalPath = p, xPubBal = b} =-        getAddressUnspents (balanceAddress b) mbr .| mapC (f p)+    go =+        awaitForever $ \XPubBal {xPubBalPath = p, xPubBal = b} ->+            getAddressUnspents (balanceAddress b) mbr .|+            mapC (\t -> XPubUnspent {xPubUnspentPath = p, xPubUnspent = t})  xpubUnspentLimit ::        ( MonadResource m@@ -828,7 +861,7 @@     -> Maybe Word32     -> StartFrom     -> XPubKey-    -> ConduitT () XPubUnspent m ()+    -> ConduitT i XPubUnspent m () xpubUnspentLimit net l s x =     xpubUnspent net (mbr s) x .| (offset s >> limit l) @@ -839,7 +872,7 @@     -> XPubKey     -> m XPubSummary xpubSummary l s x = do-    bs <- xpubBals x+    bs <- runConduit $ xpubBals x .| sinkList     let f XPubBal {xPubBalPath = p, xPubBal = Balance {balanceAddress = a}} =             (a, p)         pm = H.fromList $ map f bs@@ -946,7 +979,7 @@        (Monad m, StoreStream m)     => Maybe Word32     -> StartFrom-    -> ConduitT () TxHash m ()+    -> ConduitT i TxHash m () getMempoolLimit _ StartBlock {} = return () getMempoolLimit l (StartMem t) =     getMempool (Just t) .| mapC snd .| limit l@@ -958,7 +991,7 @@     => Maybe Word32     -> StartFrom     -> Address-    -> ConduitT () BlockTx m ()+    -> ConduitT i BlockTx m () getAddressTxsLimit l s a =     getAddressTxs a (mbr s) .| (offset s >> limit l) @@ -967,7 +1000,7 @@     => Maybe Word32     -> StartFrom     -> Address-    -> ConduitT () Transaction m ()+    -> ConduitT i Transaction m () getAddressTxsFull l s a =     getAddressTxsLimit l s a .| concatMapMC (getTransaction . blockTxHash) @@ -976,7 +1009,7 @@     => Maybe Word32     -> StartFrom     -> [Address]-    -> ConduitT () BlockTx m ()+    -> ConduitT i BlockTx m () getAddressesTxsLimit l s as =     mergeSourcesBy (flip compare `on` blockTxBlock) xs .| dedup .|     (offset s >> limit l)@@ -988,7 +1021,7 @@     => Maybe Word32     -> StartFrom     -> [Address]-    -> ConduitT () Transaction m ()+    -> ConduitT i Transaction m () getAddressesTxsFull l s as =     getAddressesTxsLimit l s as .| concatMapMC (getTransaction . blockTxHash) @@ -997,7 +1030,7 @@     => Maybe Word32     -> StartFrom     -> Address-    -> ConduitT () Unspent m ()+    -> ConduitT i Unspent m () getAddressUnspentsLimit l s a =     getAddressUnspents a (mbr s) .| (offset s >> limit l) @@ -1006,7 +1039,7 @@     => Maybe Word32     -> StartFrom     -> [Address]-    -> ConduitT () Unspent m ()+    -> ConduitT i Unspent m () getAddressesUnspentsLimit l s as =     mergeSourcesBy         (flip compare `on` unspentBlock)@@ -1032,7 +1065,7 @@         Just x -> atomically (writeTBQueue q (Just x)) >> conduitToQueue q         Nothing -> atomically $ writeTBQueue q Nothing -queueToConduit :: MonadIO m => TBQueue (Maybe a) -> ConduitT () a m ()+queueToConduit :: MonadIO m => TBQueue (Maybe a) -> ConduitT i a m () queueToConduit q =     atomically (readTBQueue q) >>= \case         Just x -> yield x >> queueToConduit q@@ -1123,7 +1156,7 @@         method <> " " <> path <> query <> " " <> cs (show version)  fmtDiff :: NominalDiffTime -> Text-fmtDiff d = cs (show (round (d * 1000) :: Integer)) <> " ms"+fmtDiff d = cs (printf "%0.3f" (realToFrac (d * 1000) :: Double) :: String) <> " ms"  fmtStatus :: Status -> Text fmtStatus s = cs (show (statusCode s)) <> " " <> cs (statusMessage s)