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: f1615ec7d8469f9d8c159a8a4c0a62904bfdd0ea8143e0d6c38f4cc0f7da7ad9 +-- hash: cb7f884a089f07246be3cb4f0347f2fd4fc9064384e62b4570a1c3bfe6a3fdcb name: haskoin-store -version: 0.35.0 +version: 0.35.1 synopsis: Storage and index for Bitcoin and Bitcoin Cash description: Please see the README on GitHub at category: Bitcoin, Finance, Network @@ -53,7 +53,7 @@ , hashable >=1.3.0.0 , haskoin-core >=0.13.6 , haskoin-node >=0.14.1 - , haskoin-store-data ==0.35.0 + , haskoin-store-data ==0.35.1 , hedis >=0.12.13 , http-types >=0.12.3 , monad-logger >=0.3.32 @@ -96,7 +96,7 @@ , haskoin-core >=0.13.6 , haskoin-node >=0.14.1 , haskoin-store - , haskoin-store-data ==0.35.0 + , haskoin-store-data ==0.35.1 , monad-logger >=0.3.32 , mtl >=2.2.2 , nqe >=0.6.1 @@ -135,8 +135,8 @@ , hashable >=1.3.0.0 , haskoin-core >=0.13.6 , haskoin-node >=0.14.1 - , haskoin-store ==0.35.0 - , haskoin-store-data ==0.35.0 + , haskoin-store ==0.35.1 + , haskoin-store-data ==0.35.1 , 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 @@ -79,10 +79,9 @@ XPubSpec (..), XPubUnspent (..), nullBalance) import NQE (Inbox, Mailbox, receive, send) -import System.Random (randomIO, randomRIO) +import System.Random (randomIO) import UnliftIO (Exception, MonadIO, MonadUnliftIO, bracket, liftIO, throwIO) -import UnliftIO.Concurrent (threadDelay) runRedis :: MonadLoggerIO m => Redis (Either Reply a) -> CacheX m a runRedis action = @@ -180,7 +179,7 @@ return txs False -> do $(logDebugS) "Cache" $ "Caching new xpub " <> xpubtxt - newXPubC xpub >>= \(bals, t) -> + newXPubC xpub >>= \(t, bals) -> if t then do $(logDebugS) "Cache" $ @@ -202,7 +201,7 @@ bals <- cacheGetXPubBalances xpub process bals False -> - newXPubC xpub >>= \(bals, t) -> + newXPubC xpub >>= \(t, bals) -> if t then do $(logDebugS) "Cache" $ @@ -254,7 +253,7 @@ return bals False -> do $(logDebugS) "Cache" $ "Caching balances for xpub " <> xpubtxt - fst <$> newXPubC xpub + snd <$> newXPubC xpub isInCache :: MonadLoggerIO m => XPubSpec -> CacheT m Bool isInCache xpub = @@ -451,7 +450,6 @@ go = do newBlockC forever $ do - pruneDB x <- receive inbox cacheWriterReact x @@ -518,29 +516,6 @@ Just _ -> Just <$> f Nothing -> return Nothing -withLockWait :: - (MonadLoggerIO m, MonadUnliftIO m) - => Integer - -> CacheX m a - -> CacheX m (Maybe a) -withLockWait n f = do - $(logDebugS) "Cache" "Acquiring lock" - go n - where - go n' - | n' <= 0 = do - $(logDebugS) "Cache" "Failed to acquire lock" - return Nothing - | otherwise = withLock f >>= \case - Just x -> return $ Just x - Nothing -> do - $(logDebugS) "Cache" $ - "Retrying acquisition of lock " <> - cs (show n') <> " times" - r <- liftIO $ randomRIO (5000, 10000) - threadDelay r - go (n' - 1) - pruneDB :: (MonadUnliftIO m, MonadLoggerIO m, StoreReadBase m) => CacheX m Integer pruneDB = @@ -557,7 +532,7 @@ case min 1000 (n `div` 64) of 0 -> return 0 x -> do - m <- withLockWait 20 $ do + m <- withLock $ do ks <- fmap (map fst) . runRedis $ getFromSortedSet maxKey Nothing 0 (fromIntegral x) @@ -583,33 +558,29 @@ => CacheWriterMessage -> CacheX m () cacheWriterReact CacheNewBlock = newBlockC cacheWriterReact (CacheNewTx th) = newTxC th -cacheWriterReact CachePing = newBlockC >> syncMempoolC +cacheWriterReact CachePing = pruneDB >> newBlockC >> syncMempoolC lenNotNull :: [XPubBal] -> Int lenNotNull bals = length $ filter (not . nullBalance . xPubBal) bals newXPubC :: (MonadUnliftIO m, MonadLoggerIO m, StoreReadExtra m) - => XPubSpec -> CacheX m ([XPubBal], Bool) + => XPubSpec -> CacheX m (Bool, [XPubBal]) newXPubC xpub = cachePrime >>= \case - Nothing -> do - bals <- lift $ xPubBals xpub - return (bals, False) + Nothing -> (False,) <$> lift (xPubBals xpub) Just _ -> do bals <- lift $ xPubBals xpub x <- asks cacheMin t <- xpubText xpub let n = lenNotNull bals if x <= n - then go bals >>= \case - Just _ -> return (bals, True) - Nothing -> return (bals, False) + then go bals else do $(logDebugS) "Cache" $ "Not caching xpub with " <> cs (show n) <> " used addresses: " <> t - return (bals, False) + return (False, bals) where op XPubUnspent {xPubUnspent = u} = (unspentPoint u, unspentBlock u) go bals = do @@ -629,21 +600,21 @@ "Caching xpub with " <> cs (show (length xtxs)) <> " txs: " <> xpubtxt now <- systemSeconds <$> liftIO getSystemTime - withLockWait 20 $ do - $(logDebugS) "Cache" $ - "Running Redis pipeline to cache xpub: " <> xpubtxt <> "" - runRedis $ do - b <- redisTouchKeys now [xpub] - c <- redisAddXPubBalances xpub bals - d <- redisAddXPubUnspents xpub (map op utxo) - e <- redisAddXPubTxs xpub xtxs - return $ b >> c >> d >> e >> return () - $(logDebugS) "Cache" $ "Done caching xpub: " <> xpubtxt + $(logDebugS) "Cache" $ + "Running Redis pipeline to cache xpub: " <> xpubtxt <> "" + runRedis $ do + b <- redisTouchKeys now [xpub] + c <- redisAddXPubBalances xpub bals + d <- redisAddXPubUnspents xpub (map op utxo) + e <- redisAddXPubTxs xpub xtxs + return $ b >> c >> d >> e >> return () + $(logDebugS) "Cache" $ "Done caching xpub: " <> xpubtxt + return (True, bals) newTxC :: (MonadUnliftIO m, MonadLoggerIO m, StoreReadExtra m) => TxHash -> CacheX m () newTxC th = do - m <- withLockWait 20 $ isAnythingCached >>= \case + m <- withLock $ isAnythingCached >>= \case False -> $(logDebugS) "Cache" $ "Not importing tx " <> txHashToHex th <> " because cache is empty" @@ -668,7 +639,7 @@ newBlockC :: (MonadUnliftIO m, MonadLoggerIO m, StoreReadExtra m) => CacheX m () newBlockC = - withLockWait 20 f >>= \m -> + withLock f >>= \m -> when (isNothing m) $ $(logErrorS) "Cache" "Could not get lock to add add block to cache" where @@ -1024,7 +995,7 @@ chbest <- chainGetBest ch if headerHash (nodeHeader chbest) == newhead then do - m <- withLockWait 20 $ do + m <- withLock $ do $(logInfoS) "Cache" "Priming cache" mem <- lift getMempool runRedis $ do