diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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.19.3 +### Changed +- Add address transactions to cache. +- Improve multi-address transaction retrieval algorithms. + ## 0.19.2 ### Removed - Cache-Control header turned out to be unnecessary. diff --git a/app/Main.hs b/app/Main.hs --- a/app/Main.hs +++ b/app/Main.hs @@ -14,27 +14,15 @@ import Control.Exception () import Control.Monad import Control.Monad.Logger -import Control.Monad.Trans.Maybe -import Data.Aeson.Encoding (encodingToLazyByteString, - fromEncoding) import Data.Bits -import Data.ByteString.Builder -import qualified Data.ByteString.Lazy as L -import qualified Data.ByteString.Lazy.Char8 as C -import Data.Char -import Data.Function import Data.List import Data.Maybe -import Data.Serialize as Serialize 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 import Haskoin.Store -import Network.HTTP.Types import NQE import Options.Applicative import Paths_haskoin_store as P @@ -44,7 +32,6 @@ import Text.Read (readMaybe) import UnliftIO import UnliftIO.Directory -import Web.Scotty.Trans data Config = Config { configDir :: !FilePath @@ -204,7 +191,7 @@ ("haskoin-store version " <> showVersion P.version) cacheDir :: Network -> FilePath -> Maybe FilePath -cacheDir net "" = Nothing +cacheDir _ "" = Nothing cacheDir net ch = Just (ch getNetworkName net "cache") run :: MonadUnliftIO m => Config -> m () @@ -242,7 +229,15 @@ removePathForcibly ch $(logInfoS) "Main" $ "Creating cache directory: " <> cs ch createDirectoryIfMissing True ch - dbh <- open ch R.defaultOptions {createIfMissing = True} + dbh <- + open + ch + R.defaultOptions + { createIfMissing = True + , compression = SnappyCompression + , maxOpenFiles = -1 + , writeBufferSize = 2 `shift` 30 + } return $ Just BlockDB @@ -265,14 +260,14 @@ , storeConfNetwork = net , storeConfListen = (`sendSTM` pub) . Event } - in withStore scfg $ \str -> + in withStore scfg $ \st -> let wcfg = WebConfig { webPort = port , webNetwork = net , webDB = ldb , webPublisher = pub - , webStore = str + , webStore = st , webMaxLimits = limits , webReqLog = reqlog , webTimeouts = tos diff --git a/haskoin-store.cabal b/haskoin-store.cabal --- a/haskoin-store.cabal +++ b/haskoin-store.cabal @@ -4,10 +4,10 @@ -- -- see: https://github.com/sol/hpack -- --- hash: 48d2a767b1724ab8df98a7b585636285acb01533c5f64154c11b922c4fa1a795 +-- hash: a1ecccb0f56324a96c0e1b9b728d04decb450aac987685698592e3618fc1140a name: haskoin-store -version: 0.19.2 +version: 0.19.3 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 @@ -72,7 +72,6 @@ , transformers , unliftio , unordered-containers - , vector , wai , warp default-language: Haskell2010 @@ -114,7 +113,6 @@ , transformers , unliftio , unordered-containers - , vector , wai , warp default-language: Haskell2010 @@ -158,7 +156,6 @@ , transformers , unliftio , unordered-containers - , vector , wai , warp 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 @@ -77,7 +77,6 @@ , healthCheck , withBlockMem , withLayeredDB - , insertNubInSortedBy ) where import Conduit diff --git a/src/Network/Haskoin/Store/Data/Cached.hs b/src/Network/Haskoin/Store/Data/Cached.hs --- a/src/Network/Haskoin/Store/Data/Cached.hs +++ b/src/Network/Haskoin/Store/Data/Cached.hs @@ -4,26 +4,16 @@ module Network.Haskoin.Store.Data.Cached where import Conduit -import Control.Applicative -import Control.Monad.Except -import Control.Monad.Logger -import Control.Monad.Reader (MonadReader, ReaderT) +import Control.Monad.Reader (ReaderT) import qualified Control.Monad.Reader as R -import Control.Monad.Trans.Maybe import qualified Data.ByteString as B -import qualified Data.ByteString.Short as B.Short import Data.IntMap.Strict (IntMap) -import Data.List -import Data.Maybe import Data.Serialize (Serialize, encode) -import Data.String.Conversions (cs) import Database.RocksDB as R -import Database.RocksDB.Query as R import Haskoin import Network.Haskoin.Store.Data import Network.Haskoin.Store.Data.KeyValue import Network.Haskoin.Store.Data.RocksDB -import NQE (query) import UnliftIO newLayeredDB :: MonadUnliftIO m => BlockDB -> Maybe BlockDB -> m LayeredDB @@ -33,6 +23,7 @@ bulkCopy opts db cdb BalKeyS bulkCopy opts db cdb UnspentKeyB bulkCopy opts db cdb MemKey + bulkCopy opts db cdb AddrTxKeyS return LayeredDB {layeredDB = blocks, layeredCache = Just cache} where BlockDB {blockDBopts = opts, blockDB = db} = blocks @@ -108,19 +99,21 @@ -> Maybe BlockRef -> LayeredDB -> ConduitT i BlockTx m () +getAddressTxsC addr mbr LayeredDB {layeredCache = Just db} = + getAddressTxsDB addr mbr db getAddressTxsC addr mbr LayeredDB {layeredDB = db} = getAddressTxsDB addr mbr db bulkCopy :: (Serialize k, MonadUnliftIO m) => ReadOptions -> DB -> DB -> k -> m () -bulkCopy opts db cdb k = +bulkCopy opts db cdb b = runResourceT $ do ch <- newTBQueueIO 1000000 - withAsync (iterate ch) $ \a -> write_batch ch [] 0 + withAsync (iter ch) $ \_ -> write_batch ch [] (0 :: Int) where - iterate ch = + iter ch = withIterator db opts $ \it -> do - iterSeek it (encode k) + iterSeek it (encode b) recurse it ch write_batch ch acc l | l >= 10000 = do @@ -128,16 +121,16 @@ write_batch ch [] 0 | otherwise = atomically (readTBQueue ch) >>= \case - Just (key, val) -> write_batch ch (Put key val : acc) (l + 1) + Just (k, v) -> write_batch ch (Put k v : acc) (l + 1) Nothing -> write cdb defaultWriteOptions acc recurse it ch = iterEntry it >>= \case Nothing -> atomically $ writeTBQueue ch Nothing - Just (key, val) -> - let pfx = B.take (B.length (encode k)) key - in if pfx == encode k + Just (k, v) -> + let pfx = B.take (B.length (encode b)) k + in if pfx == encode b then do - atomically . writeTBQueue ch $ Just (key, val) + atomically . writeTBQueue ch $ Just (k, v) iterNext it recurse it ch else atomically $ writeTBQueue ch Nothing diff --git a/src/Network/Haskoin/Store/Data/ImportDB.hs b/src/Network/Haskoin/Store/Data/ImportDB.hs --- a/src/Network/Haskoin/Store/Data/ImportDB.hs +++ b/src/Network/Haskoin/Store/Data/ImportDB.hs @@ -13,14 +13,12 @@ import Control.Monad.Reader (ReaderT) import qualified Control.Monad.Reader as R import Control.Monad.Trans.Maybe -import qualified Data.ByteString.Short as B.Short import Data.HashMap.Strict (HashMap) import qualified Data.HashMap.Strict as M import Data.IntMap.Strict (IntMap) import qualified Data.IntMap.Strict as I import Data.List import Data.Maybe -import Data.String.Conversions (cs) import Database.RocksDB as R import Database.RocksDB.Query as R import Haskoin @@ -78,6 +76,7 @@ cacheMapOps :: BlockMem -> [BatchOp] cacheMapOps db = balOps (hBalance db) <> maybeToList (mempoolOp <$> hMempool db) <> + addrTxOps (hAddrTx db) <> unspentOps (hUnspent db) bestBlockOp :: Maybe BlockHash -> [BatchOp] @@ -348,7 +347,7 @@ instance MonadIO m => StoreStream (ReaderT ImportDB m) where getOrphans = undefined - getAddressUnspents a m = undefined - getAddressTxs a m = undefined + getAddressUnspents _ _ = undefined + getAddressTxs _ _ = undefined getAddressBalances = undefined getUnspents = undefined diff --git a/src/Network/Haskoin/Store/Data/KeyValue.hs b/src/Network/Haskoin/Store/Data/KeyValue.hs --- a/src/Network/Haskoin/Store/Data/KeyValue.hs +++ b/src/Network/Haskoin/Store/Data/KeyValue.hs @@ -7,7 +7,6 @@ import Control.Monad.Reader import Data.ByteString (ByteString) import qualified Data.ByteString as B -import Data.Default import Data.Hashable import Data.Serialize as S import Data.Word diff --git a/src/Network/Haskoin/Store/Web.hs b/src/Network/Haskoin/Store/Web.hs --- a/src/Network/Haskoin/Store/Web.hs +++ b/src/Network/Haskoin/Store/Web.hs @@ -24,7 +24,6 @@ import qualified Data.ByteString.Lazy as L import qualified Data.ByteString.Lazy.Char8 as C import Data.Char -import Data.Foldable import Data.Function import qualified Data.HashMap.Strict as H import Data.List @@ -37,8 +36,6 @@ import qualified Data.Text.Lazy as T.Lazy import Data.Time.Clock import Data.Time.Clock.System -import Data.Vector (Vector, cons, (!)) -import qualified Data.Vector as V import Data.Word import Database.RocksDB as R import Haskoin @@ -589,17 +586,46 @@ derive <- parseDeriveAddrs net proto <- setupBin db <- askDB - as <- - liftIO . runStream db . runConduit $ - xpubBals limits derive x .| mapC (balanceAddress . xPubBal) .| sinkList S.stream $ \io flush' -> do - runStream db . runConduit $ f proto l s as io + runStream db . runConduit $ f proto l s derive io x flush' where - f proto l s as io - | full = getAddressesTxsFull l s as .| streamAny net proto io - | otherwise = getAddressesTxsLimit l s as .| streamAny net proto io + f proto l s derive io x + | full = + xpubTxs limits s l derive x .| + concatMapMC (getTransaction . blockTxHash) .| + streamAny net proto io + | otherwise = xpubTxs limits s l derive x .| streamAny net proto io +xpubTxs :: + (Monad m, StoreRead m, StoreStream m) + => MaxLimits + -> Maybe BlockRef + -> Maybe Limit + -> DeriveAddrs + -> XPubKey + -> ConduitT i BlockTx m () +xpubTxs max_limits start limit derive xpub = do + ts <- + fmap (nub . sortBy (flip compare `on` blockTxBlock)) $ + (go 0 >> go 1) .| sinkList + forM_ ts yield .| applyLimit limit + where + go m = yieldMany (addrs m) .| txs .| gap + addrs m = map (\(a, _, _) -> a) (derive (pubSubKey xpub m) 0) + txs = + awaitForever $ \a -> do + ts <- getAddressTxs a start .| applyLimit limit .| sinkList + yield ts + gap = + let r 0 = return () + r i = + await >>= \case + Nothing -> return () + Just [] -> r (i - 1) + Just xs -> yieldMany xs >> r (maxLimitGap max_limits) + in r (maxLimitGap max_limits) + scottyXpubUnspents :: (MonadLoggerIO m, MonadUnliftIO m) => Network -> MaxLimits -> WebT m () scottyXpubUnspents net limits = do @@ -1141,56 +1167,6 @@ then return $ Just (e', True) else r e' ns --- Snatched from: --- https://github.com/cblp/conduit-merge/blob/master/src/Data/Conduit/Merge.hs -mergeSourcesBy :: - (Foldable f, Monad m) - => (a -> a -> Ordering) - -> f (ConduitT () a m ()) - -> ConduitT i a m () -mergeSourcesBy f = mergeSealed . fmap sealConduitT . toList - where - mergeSealed sources = do - prefetchedSources <- lift $ traverse ($$++ await) sources - go . V.fromList . nubBy (\a b -> f (fst a) (fst b) == EQ) $ - sortBy (f `on` fst) [(a, s) | (s, Just a) <- prefetchedSources] - go sources - | V.null sources = pure () - | otherwise = do - let (a, src1) = V.head sources - sources1 = V.tail sources - yield a - (src2, mb) <- lift $ src1 $$++ await - let sources2 = - case mb of - Nothing -> sources1 - Just b -> - insertNubInSortedBy (f `on` fst) (b, src2) sources1 - go sources2 - -insertNubInSortedBy :: (a -> a -> Ordering) -> a -> Vector a -> Vector a -insertNubInSortedBy f x xs - | null xs = x `cons` xs - | otherwise = - case find_idx 0 (length xs - 1) of - Nothing -> x `cons` xs - Just i -> - let (xs1, xs2) = V.splitAt i xs - in xs1 <> x `cons` xs2 - where - find_idx a b - | f (xs ! a) x == EQ = Nothing - | f (xs ! b) x == EQ = Nothing - | f (xs ! b) x == LT = Just (b + 1) - | f (xs ! a) x == GT = Just a - | b - a == 1 = Just b - | otherwise = - let c = a + (b - a) `div` 2 - z = xs ! c - in if f z x == GT - then find_idx a c - else find_idx c b - getAddressTxsLimit :: (Monad m, StoreStream m) => Offset @@ -1218,10 +1194,13 @@ -> Maybe BlockRef -> [Address] -> ConduitT i BlockTx m () -getAddressesTxsLimit limit start addrs = - mergeSourcesBy (flip compare `on` blockTxBlock) xs .| applyLimit limit +getAddressesTxsLimit limit start addrs = do + ts <- + fmap (nub . sortBy (flip compare `on` blockTxBlock)) $ + get_txs .| sinkList + forM_ ts yield .| applyLimit limit where - xs = map (`getAddressTxs` start) addrs + get_txs = mapM_ (\a -> getAddressTxs a start .| applyLimit limit) addrs getAddressesTxsFull :: (MonadResource m, MonadUnliftIO m, StoreStream m, StoreRead m) @@ -1249,11 +1228,13 @@ -> Maybe BlockRef -> [Address] -> ConduitT i Unspent m () -getAddressesUnspentsLimit limit start addrs = - mergeSourcesBy - (flip compare `on` unspentBlock) - (map (`getAddressUnspents` start) addrs) .| - applyLimit limit +getAddressesUnspentsLimit limit start addrs = do + uns <- + fmap (nub . sortBy (flip compare `on` unspentBlock)) $ + get_uns .| sinkList + forM_ uns yield .| applyLimit limit + where + get_uns = mapM_ (\a -> getAddressUnspents a start .| applyLimit limit) addrs applyOffsetLimit :: Monad m => Offset -> Maybe Limit -> ConduitT i i m () applyOffsetLimit offset limit = applyOffset offset >> applyLimit limit