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: 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 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 @@ -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 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 @@ -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 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 @@ -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 (..), diff --git a/src/Haskoin/Store/Database/Writer.hs b/src/Haskoin/Store/Database/Writer.hs --- a/src/Haskoin/Store/Database/Writer.hs +++ b/src/Haskoin/Store/Database/Writer.hs @@ -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, diff --git a/src/Haskoin/Store/Logic.hs b/src/Haskoin/Store/Logic.hs --- a/src/Haskoin/Store/Logic.hs +++ b/src/Haskoin/Store/Logic.hs @@ -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 (..), 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 @@ -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 (..),