haskoin-store 0.26.2 → 0.26.3
raw patch · 7 files changed
+103/−67 lines, 7 filesdep ~haskoin-store-data
Dependency ranges changed: haskoin-store-data
Files
- haskoin-store.cabal +5/−5
- src/Haskoin/Store/Cache.hs +85/−53
- src/Haskoin/Store/Common.hs +6/−1
- src/Haskoin/Store/Database/Reader.hs +2/−2
- src/Haskoin/Store/Database/Writer.hs +2/−2
- src/Haskoin/Store/Logic.hs +2/−2
- src/Haskoin/Store/Web.hs +1/−2
haskoin-store.cabal view
@@ -4,10 +4,10 @@ -- -- see: https://github.com/sol/hpack ----- hash: d8f0a497d75c2e8bea6249b9809c79b72d8b13a39a166553bae1a4e57466d42d+-- hash: aa22c0a2b15ff3fca854ff7a28d2d69f11c446cea2307b746b8823bfbc7f0919 name: haskoin-store-version: 0.26.2+version: 0.26.3 synopsis: Storage and index for Bitcoin and Bitcoin Cash description: Store and index Bitcoin or Bitcoin Cash blocks, transactions, balances and unspent outputs. All data is available via REST API in JSON or binary format.@@ -57,7 +57,7 @@ , hashable >=1.3.0.0 , haskoin-core >=0.13.3 , haskoin-node >=0.13.0- , haskoin-store-data ==0.26.2+ , haskoin-store-data ==0.26.3 , hedis >=0.12.13 , http-types >=0.12.3 , monad-logger >=0.3.32@@ -99,7 +99,7 @@ , haskoin-core >=0.13.3 , haskoin-node >=0.13.0 , haskoin-store- , haskoin-store-data ==0.26.2+ , haskoin-store-data ==0.26.3 , monad-logger >=0.3.32 , mtl >=2.2.2 , nqe >=0.6.1@@ -149,7 +149,7 @@ , hashable >=1.3.0.0 , haskoin-core >=0.13.3 , haskoin-node >=0.13.0- , haskoin-store-data ==0.26.2+ , haskoin-store-data ==0.26.3 , hedis >=0.12.13 , hspec >=2.7.1 , http-types >=0.12.3
src/Haskoin/Store/Cache.hs view
@@ -1,6 +1,7 @@ {-# LANGUAGE ApplicativeDo #-} {-# LANGUAGE DeriveAnyClass #-} {-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE LambdaCase #-} {-# LANGUAGE OverloadedStrings #-}@@ -47,7 +48,7 @@ import Data.String.Conversions (cs) import Data.Text (Text) import Data.Time.Clock.System (getSystemTime, systemSeconds)-import Data.Word (Word64)+import Data.Word (Word32, Word64) import Database.Redis (Connection, Redis, RedisCtx, Reply, checkedConnect, defaultConnectInfo, hgetall, parseConnectInfo, zadd,@@ -78,7 +79,7 @@ XPubBal (..), XPubSpec (..), XPubUnspent (..), nullBalance) import NQE (Inbox, Mailbox, receive, send)-import System.Random (randomRIO)+import System.Random (randomIO, randomRIO) import UnliftIO (Exception, MonadIO, MonadUnliftIO, bracket, liftIO, throwIO) import UnliftIO.Concurrent (threadDelay)@@ -427,7 +428,7 @@ mempoolSetKey = "mempool" lockKey :: ByteString-lockKey = "lock"+lockKey = "xpub" importLockKey :: ByteString importLockKey = "import"@@ -463,22 +464,29 @@ x <- receive inbox cacheWriterReact x -lockIt :: MonadLoggerIO m => ByteString -> CacheT m Bool+lockIt :: MonadLoggerIO m => ByteString -> CacheT m (Maybe Word32) lockIt k = do- go >>= \case- Right Redis.Ok -> return True+ rnd <- liftIO randomIO+ go rnd >>= \case+ Right Redis.Ok -> do+ $(logDebugS) "Cache" $+ "Acquired " <> cs k <> " lock with value " <> cs (show rnd)+ return (Just rnd) Right Redis.Pong -> do- $(logErrorS) "Cache" $ "Got unexpected Pong from Redis"- return False+ $(logErrorS) "Cache" $+ "Unexpected pong when acquiring " <> cs k <> " lock"+ return Nothing Right (Redis.Status s) -> do- $(logErrorS) "Cache" $ "Got unexpected status from Redis: " <> cs s- return False- Left (Redis.Bulk Nothing) -> return False+ $(logErrorS) "Cache" $+ "Unexpected status acquiring " <> cs k <> " lock: " <> cs s+ return Nothing+ Left (Redis.Bulk Nothing) -> return Nothing Left e -> do- $(logErrorS) "Cache" $ "Error when trying to acquire lock"+ $(logErrorS) "Cache" $+ "Error when trying to acquire " <> cs k <> " lock" throwIO (RedisError e) where- go = do+ go rnd = do conn <- asks cacheConn liftIO . Redis.runRedis conn $ do let opts =@@ -487,35 +495,58 @@ , Redis.setMilliseconds = Nothing , Redis.setCondition = Just Redis.Nx }- Redis.setOpts k "l" opts+ Redis.setOpts k (cs (show rnd)) opts -unlockIt :: MonadLoggerIO m => ByteString -> CacheT m ()-unlockIt k = runRedis (Redis.del [k]) >> return ()+unlockIt :: MonadLoggerIO m => ByteString -> Maybe Word32 -> CacheT m ()+unlockIt _ Nothing = return ()+unlockIt k (Just i) =+ runRedis (Redis.get k) >>= \case+ Nothing ->+ $(logErrorS) "Cache" $+ "Not releasing " <> cs k <> " lock with value " <> cs (show i) <>+ ": not locked"+ Just bs ->+ if read (cs bs) == i+ then do+ runRedis (Redis.del [k]) >> return ()+ $(logDebugS) "Cache" $+ "Released " <> cs k <> " lock with value " <>+ cs (show i)+ else do+ $(logErrorS) "Cache" $+ "Could not release " <> cs k <> " lock: value is not " <>+ cs (show i) withLock ::- (MonadLoggerIO m, MonadUnliftIO m) => ByteString -> CacheT m a -> CacheT m (Maybe a)+ (MonadLoggerIO m, MonadUnliftIO m)+ => ByteString+ -> CacheT m a+ -> CacheT m (Maybe a) withLock k f =- bracket (lockIt k) (const (unlockIt k)) $ \case- True -> Just <$> f- False -> return Nothing+ bracket (lockIt k) (unlockIt k) $ \case+ Just _ -> Just <$> f+ Nothing -> return Nothing withLockWait :: (MonadLoggerIO m, MonadUnliftIO m) => ByteString -> CacheT m a -> CacheT m a-withLockWait k f =- withLock k f >>= \case- Just x -> return x- Nothing -> do- r <- liftIO $ randomRIO (500, 10000)- threadDelay r- withLockWait k f+withLockWait k f = do+ $(logDebugS) "Cache" $ "Acquiring " <> cs k <> " lock…"+ go+ where+ go =+ withLock k f >>= \case+ Just x -> return x+ Nothing -> do+ r <- liftIO $ randomRIO (500, 10000)+ threadDelay r+ go pruneDB :: (MonadUnliftIO m, MonadLoggerIO m, StoreRead m) => CacheT m Integer-pruneDB =- withLockWait lockKey $+pruneDB = do isAnythingCached >>= \case True -> do x <- asks cacheMax@@ -528,13 +559,14 @@ flush n = do case min 1000 (n `div` 64) of 0 -> return 0- x -> do- ks <-- fmap (map fst) . runRedis $- getFromSortedSet maxKey Nothing 0 (Just x)- $(logDebugS) "Cache" $- "Pruning " <> cs (show (length ks)) <> " old xpubs"- delXPubKeys ks+ x ->+ withLockWait lockKey $ do+ ks <-+ fmap (map fst) . runRedis $+ getFromSortedSet maxKey Nothing 0 (Just x)+ $(logDebugS) "Cache" $+ "Pruning " <> cs (show (length ks)) <> " old xpubs"+ delXPubKeys ks touchKeys :: MonadLoggerIO m => [XPubSpec] -> CacheT m () touchKeys xpubs = do@@ -616,20 +648,23 @@ return $ b >> c >> d >> e >> return () newBlockC :: (MonadUnliftIO m, MonadLoggerIO m, StoreRead m) => CacheT m ()-newBlockC =- withLockWait importLockKey $- isAnythingCached >>= \case- False -> return ()- True ->- cachePrime >>= \case- Nothing -> return ()- Just cachehead ->- lift getBestBlock >>= \case- Nothing -> return ()- Just newhead -> go newhead cachehead+newBlockC = withLockWait importLockKey f where+ f =+ isAnythingCached >>= \case+ False ->+ $(logDebugS) "Cache" "Not syncing blocks because cache is empty"+ True ->+ cachePrime >>= \case+ Nothing -> return ()+ Just cachehead ->+ lift getBestBlock >>= \case+ Nothing -> return ()+ Just newhead -> go newhead cachehead go newhead cachehead- | cachehead == newhead = syncMempoolC+ | cachehead == newhead = do+ $(logDebugS) "Cache" $ "Blocks in sync: " <> blockHashToHex cachehead+ syncMempoolC | otherwise = asks cacheChain >>= \ch -> chainBlockMain newhead ch >>= \case@@ -648,7 +683,7 @@ $(logWarnS) "Cache" $ "Reverting cache head not in main chain: " <> blockHashToHex cachehead- removeHeadC >> newBlockC+ removeHeadC >> f True -> chainGetBlock newhead ch >>= \case Nothing -> do@@ -673,8 +708,7 @@ $(logErrorS) "Cache" txt throwIO $ LogicError $ cs txt Just newcachenode ->- importBlockC (headerHash (nodeHeader newcachenode)) >>- newBlockC+ importBlockC (headerHash (nodeHeader newcachenode)) >> f importBlockC :: (MonadUnliftIO m, StoreRead m, MonadLoggerIO m) => BlockHash -> CacheT m () importBlockC bh =@@ -743,7 +777,6 @@ "Affected xpub " <> cs (show i) <> "/" <> cs (show (length xpubs)) <> ": " <> xpubtxt- $(logDebugS) "Cache" $ "Acquiring lock…" addrs' <- withLockWait lockKey $ do $(logDebugS) "Cache" $@@ -873,7 +906,6 @@ utxomap <- HashMap.fromListWith (<>) <$> mapM (uncurry getutxo) addrs $(logDebugS) "Cache" $ "Getting transactions…" txmap <- HashMap.fromListWith (<>) <$> mapM (uncurry gettxmap) addrs- $(logDebugS) "Cache" $ "Acquiring lock…" withLockWait lockKey $ do $(logDebugS) "Cache" $ "Running Redis pipeline…" runRedis $ do
src/Haskoin/Store/Common.hs view
@@ -22,6 +22,7 @@ , applyLimitC , applyOffsetLimitC , sortTxs+ , nub ) where import Conduit (ConduitT, dropC, mapC, takeC)@@ -32,9 +33,10 @@ import Control.Monad.Trans.Maybe (MaybeT (..), runMaybeT) import Data.ByteString (ByteString) import Data.Function (on)+import Data.Hashable (Hashable) import qualified Data.HashSet as H import Data.IntMap.Strict (IntMap)-import Data.List (nub, sortBy)+import Data.List (sortBy) import Data.Maybe (listToMaybe) import Data.Serialize (Serialize (..)) import Data.Word (Word32, Word64)@@ -334,3 +336,6 @@ in if orp then go ((i, tx) : orphans) ths xs else (i, tx) : go orphans (txHash tx `H.delete` ths) xs++nub :: (Eq a, Hashable a) => [a] -> [a]+nub = H.toList . H.fromList
src/Haskoin/Store/Database/Reader.hs view
@@ -15,7 +15,7 @@ import Data.Function (on) import Data.IntMap (IntMap) import qualified Data.IntMap.Strict as I-import Data.List (nub, sortBy)+import Data.List (sortBy) import Data.Maybe (fromMaybe) import Data.Word (Word32) import Database.RocksDB (Compression (..), DB,@@ -27,7 +27,7 @@ import Haskoin (Address, BlockHash, BlockHeight, Network, OutPoint (..), TxHash) import Haskoin.Store.Common (Limit, StoreRead (..),- applyLimit, applyLimitC)+ applyLimit, applyLimitC, nub) import Haskoin.Store.Data (Balance, BlockData, BlockRef (..), BlockTx (..), Spender, TxData, Unspent (..),
src/Haskoin/Store/Database/Writer.hs view
@@ -17,14 +17,14 @@ import qualified Data.HashMap.Strict as M import Data.IntMap.Strict (IntMap) import qualified Data.IntMap.Strict as I-import Data.List (nub) import Data.Maybe (fromJust, fromMaybe, isJust, maybeToList) import Database.RocksDB (BatchOp) import Database.RocksDB.Query (deleteOp, insertOp, writeBatch) import Haskoin (Address, BlockHash, BlockHeight, OutPoint (..), TxHash)-import Haskoin.Store.Common (StoreRead (..), StoreWrite (..))+import Haskoin.Store.Common (StoreRead (..), StoreWrite (..),+ nub) import Haskoin.Store.Data (Balance, BlockData, BlockRef (..), BlockTx (..), Spender, TxData, Unspent,
src/Haskoin/Store/Logic.hs view
@@ -22,7 +22,7 @@ import qualified Data.ByteString.Short as B.Short import Data.Either (rights) import qualified Data.IntMap.Strict as I-import Data.List (nub, sort)+import Data.List (sort) import Data.Maybe (fromMaybe, isNothing) import Data.Serialize (encode) import Data.String (fromString)@@ -39,7 +39,7 @@ isGenesis, nullOutPoint, scriptToAddressBS, txHash, txHashToHex)-import Haskoin.Store.Common (StoreRead (..), StoreWrite (..),+import Haskoin.Store.Common (StoreRead (..), StoreWrite (..), nub, sortTxs) import Haskoin.Store.Data (Balance (..), BlockData (..), BlockRef (..), BlockTx (..),
src/Haskoin/Store/Web.hs view
@@ -29,7 +29,6 @@ import qualified Data.ByteString.Lazy.Char8 as C import Data.Char (isSpace) import Data.Default (Default (..))-import Data.List (nub) import Data.Maybe (catMaybes, fromMaybe, isJust, listToMaybe, mapMaybe) import Data.Serialize as Serialize@@ -62,7 +61,7 @@ import Haskoin.Store.Common (Limit, Offset, PubExcept (..), StoreEvent (..), StoreRead (..), applyOffset, blockAtOrBefore,- getTransaction)+ getTransaction, nub) import Haskoin.Store.Data (BlockData (..), BlockRef (..), BlockTx (..), DeriveType (..), Event (..), Except (..),