haskoin-store 0.19.2 → 0.19.3
raw patch · 8 files changed
+84/−116 lines, 8 filesdep −vectornew-uploader
Dependencies removed: vector
Files
- CHANGELOG.md +5/−0
- app/Main.hs +12/−17
- haskoin-store.cabal +2/−5
- src/Haskoin/Store.hs +0/−1
- src/Network/Haskoin/Store/Data/Cached.hs +13/−20
- src/Network/Haskoin/Store/Data/ImportDB.hs +3/−4
- src/Network/Haskoin/Store/Data/KeyValue.hs +0/−1
- src/Network/Haskoin/Store/Web.hs +49/−68
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.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.
app/Main.hs view
@@ -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
haskoin-store.cabal view
@@ -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
src/Haskoin/Store.hs view
@@ -77,7 +77,6 @@ , healthCheck , withBlockMem , withLayeredDB- , insertNubInSortedBy ) where import Conduit
src/Network/Haskoin/Store/Data/Cached.hs view
@@ -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
src/Network/Haskoin/Store/Data/ImportDB.hs view
@@ -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
src/Network/Haskoin/Store/Data/KeyValue.hs view
@@ -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
src/Network/Haskoin/Store/Web.hs view
@@ -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