haskoin-store 0.23.3 → 0.23.4
raw patch · 4 files changed
+316/−236 lines, 4 files
Files
- CHANGELOG.md +10/−0
- haskoin-store.cabal +2/−2
- src/Haskoin/Store/Cache.hs +272/−201
- src/Haskoin/Store/Web.hs +32/−33
CHANGELOG.md view
@@ -4,6 +4,16 @@ 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.23.4+### Added+- Add extra debug logging for cache code.++### Fixed+- Fix a bug with xpub growing algorithm.++### Changed+- Use locks instead of transactions to update cache.+ ## 0.23.3 ### Fixed - Reduce contention when many instances of Haskoin Store share a cache.
haskoin-store.cabal view
@@ -4,10 +4,10 @@ -- -- see: https://github.com/sol/hpack ----- hash: ab4c33eb49a6c0ff991802e63a9a5ef7bf56cf2cb8c2b73b518b7ae7658c2f97+-- hash: 1e674b620cac18d9bc1185ecf77f9c600b3b0c953483d8f0e4fe18fe4d3fe02e name: haskoin-store-version: 0.23.3+version: 0.23.4 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.
src/Haskoin/Store/Cache.hs view
@@ -24,9 +24,9 @@ ) where import Control.DeepSeq (NFData)-import Control.Monad (forM, forever, void)-import Control.Monad.Logger (MonadLoggerIO, logErrorS, logInfoS,- logWarnS)+import Control.Monad (forM, forM_, forever, void)+import Control.Monad.Logger (MonadLoggerIO, logDebugS, logErrorS,+ logInfoS, logWarnS) import Control.Monad.Reader (ReaderT (..), asks) import Control.Monad.Trans (lift) import Control.Monad.Trans.Maybe (MaybeT (..), runMaybeT)@@ -37,19 +37,19 @@ import Data.HashMap.Strict (HashMap) import qualified Data.HashMap.Strict as HashMap import qualified Data.IntMap.Strict as IntMap-import Data.List (nub, sort)+import Data.List (nub, sort, (\\)) import qualified Data.Map.Strict as Map import Data.Maybe (catMaybes, mapMaybe) import Data.Serialize (decode, encode) import Data.Serialize (Serialize) import Data.String.Conversions (cs)+import Data.Text (Text) import Data.Time.Clock.System (getSystemTime, systemSeconds) import Data.Word (Word64)-import Database.Redis (Connection, Queued, Redis, RedisCtx,- RedisTx, Reply, TxResult (..),+import Database.Redis (Connection, Redis, RedisCtx, Reply, checkedConnect, defaultConnectInfo,- hgetall, parseConnectInfo, watch,- zadd, zrangeWithscores,+ hgetall, parseConnectInfo, zadd,+ zrangeWithscores, zrangebyscoreWithscoresLimit, zrem) import qualified Database.Redis as Redis import GHC.Generics (Generic)@@ -62,7 +62,7 @@ eitherToMaybe, headerHash, pathToList, scriptToAddressBS, txHash, txHashToHex, xPubAddr,- xPubCompatWitnessAddr,+ xPubCompatWitnessAddr, xPubExport, xPubWitnessAddr) import Haskoin.Node (Chain, chainGetAncestor, chainGetBlock, chainGetSplitBlock)@@ -77,8 +77,10 @@ xPubBalsTxs, xPubBalsUnspents, xPubTxs) import NQE (Inbox, Mailbox, receive)+import System.Random (randomIO, randomRIO) import UnliftIO (Exception, MonadIO, MonadUnliftIO,- liftIO, throwIO)+ bracket, liftIO, throwIO)+import UnliftIO.Concurrent (threadDelay) runRedis :: MonadIO m => Redis (Either Reply a) -> CacheT m a runRedis action =@@ -87,23 +89,6 @@ Right x -> return x Left e -> throwIO (RedisError e) -runRedisTx ::- MonadIO m- => Redis (Either Reply b)- -> (b -> RedisTx (Queued a))- -> CacheT m (Maybe a)-runRedisTx pre action =- asks cacheConn >>= \conn ->- liftIO (Redis.runRedis conn go) >>= \case- TxSuccess x -> return (Just x)- TxAborted -> return Nothing- TxError e -> throwIO (RedisTxError e)- where- go =- pre >>= \case- Left e -> throwIO (RedisError e)- Right y -> Redis.multiExec (action y)- data CacheConfig = CacheConfig { cacheConn :: !Connection@@ -130,7 +115,8 @@ Right r -> return r liftIO (checkedConnect conninfo) -instance (MonadLoggerIO m, StoreRead m) => StoreRead (CacheT m) where+instance (MonadUnliftIO m, MonadLoggerIO m, StoreRead m) =>+ StoreRead (CacheT m) where getNetwork = lift getNetwork getBestBlock = lift getBestBlock getBlocksAtHeight = lift . getBlocksAtHeight@@ -167,31 +153,46 @@ utxoPfx = "u" getXPubTxs ::- (MonadLoggerIO m, StoreRead m)+ (MonadUnliftIO m, MonadLoggerIO m, StoreRead m) => XPubSpec -> Maybe BlockRef -> Offset -> Maybe Limit -> CacheT m [BlockTx]-getXPubTxs xpub start offset limit =+getXPubTxs xpub start offset limit = do+ xpubtxt <- xpubText xpub+ $(logDebugS) "Cache" $ "Getting xpub txs for " <> xpubtxt isXPubCached xpub >>= \case True -> do txs <- cacheGetXPubTxs xpub start offset limit+ $(logDebugS) "Cache" $+ "Returning " <> cs (show (length txs)) <>+ " transactions for cached xpub: " <>+ xpubtxt return txs False -> do+ $(logDebugS) "Cache" $ "Caching new xpub " <> xpubtxt newXPubC xpub >>= \(bals, t) -> if t- then cacheGetXPubTxs xpub start offset limit- else xPubBalsTxs bals start offset limit+ then do+ $(logDebugS) "Cache" $+ "Successfully cached xpub " <> xpubtxt+ cacheGetXPubTxs xpub start offset limit+ else do+ $(logDebugS) "Cache" $+ "Using DB to return txs for xpub " <> xpubtxt+ xPubBalsTxs bals start offset limit getXPubUnspents ::- (MonadLoggerIO m, StoreRead m)+ (MonadUnliftIO m, MonadLoggerIO m, StoreRead m) => XPubSpec -> Maybe BlockRef -> Offset -> Maybe Limit -> CacheT m [XPubUnspent]-getXPubUnspents xpub start offset limit =+getXPubUnspents xpub start offset limit = do+ xpubtxt <- xpubText xpub+ $(logDebugS) "Cache" $ "Getting utxo for xpub " <> xpubtxt isXPubCached xpub >>= \case True -> do bals <- cacheGetXPubBalances xpub@@ -199,8 +200,14 @@ False -> do newXPubC xpub >>= \(bals, t) -> if t- then process bals- else xPubBalsUnspents bals start offset limit+ then do+ $(logDebugS) "Cache" $+ "Successfully cached xpub " <> xpubtxt+ process bals+ else do+ $(logDebugS) "Cache" $+ "Using DB to return utxo for xpub " <> xpubtxt+ xPubBalsUnspents bals start offset limit where process bals = do ops <- map snd <$> cacheGetXPubUnspents xpub start offset limit@@ -221,18 +228,29 @@ mapMaybe (\(a, u) -> (\p -> XPubUnspent p u) <$> Map.lookup a addrmap) addrutxo+ xpubtxt <- xpubText xpub+ $(logDebugS) "Cache" $+ "Returning " <> cs (show (length xpubutxo)) <>+ " utxos for cached xpub: " <>+ xpubtxt return xpubutxo getXPubBalances ::- (MonadLoggerIO m, StoreRead m)+ (MonadUnliftIO m, MonadLoggerIO m, StoreRead m) => XPubSpec -> CacheT m [XPubBal]-getXPubBalances xpub =+getXPubBalances xpub = do+ xpubtxt <- xpubText xpub+ $(logDebugS) "Cache" $ "Getting balances for xpub " <> xpubtxt isXPubCached xpub >>= \case True -> do bals <- cacheGetXPubBalances xpub+ $(logDebugS) "Cache" $+ "Returning " <> cs (show (length bals)) <> " balances for xpub " <>+ xpubtxt return bals False -> do+ $(logDebugS) "Cache" $ "Caching balances for xpub " <> xpubtxt fst <$> newXPubC xpub isXPubCached :: MonadIO m => XPubSpec -> CacheT m Bool@@ -241,13 +259,6 @@ redisIsXPubCached :: RedisCtx m f => XPubSpec -> m (f Bool) redisIsXPubCached xpub = Redis.exists (balancesPfx <> encode xpub) -redisWatchXPubCached :: [XPubSpec] -> Redis (Either Reply [Bool])-redisWatchXPubCached [] = return (pure [])-redisWatchXPubCached xpubs = do- w <- watch $ map ((balancesPfx <>) . encode) xpubs- ys <- sequence <$> forM xpubs redisIsXPubCached- return $ w >> ys- cacheGetXPubBalances :: MonadIO m => XPubSpec -> CacheT m [XPubBal] cacheGetXPubBalances xpub = do bals <- runRedis $ redisGetXPubBalances xpub@@ -416,6 +427,9 @@ mempoolSetKey :: ByteString mempoolSetKey = "mempool" +deleteLockKey :: ByteString+deleteLockKey = "deleted"+ addrPfx :: ByteString addrPfx = "a" @@ -444,39 +458,76 @@ x <- receive inbox cacheWriterReact x -pruneDB :: (MonadLoggerIO m, StoreRead m) => CacheT m Integer-pruneDB = do- x <- asks cacheMax- s <- runRedis Redis.dbsize- if s > x- then flush (s - x)- else return 0+lockIt :: MonadIO m => CacheT m Bool+lockIt = do+ r <- liftIO randomIO+ mi <-+ runRedis $ do+ let opts =+ Redis.SetOpts+ { Redis.setSeconds = Just 30+ , Redis.setMilliseconds = Nothing+ , Redis.setCondition = Just Redis.Nx+ }+ a <- Redis.setOpts deleteLockKey (cs (show (r :: Integer))) opts+ b <- Redis.get deleteLockKey+ return $ a >> fmap (read . cs) <$> b+ case mi of+ Just i+ | i == r -> return True+ _ -> return False+++unlockIt :: MonadIO m => CacheT m ()+unlockIt = runRedis (Redis.del [deleteLockKey]) >> return ()++withLock :: MonadUnliftIO m => CacheT m a -> CacheT m (Maybe a)+withLock f =+ bracket lockIt (const unlockIt) $ \case+ True -> Just <$> f+ False -> return Nothing++withLockWait :: MonadUnliftIO m => CacheT m a -> CacheT m a+withLockWait f =+ withLock f >>= \case+ Just x -> return x+ Nothing -> do+ r <- liftIO $ randomRIO (500, 10000)+ threadDelay r+ withLockWait f++pruneDB :: (MonadUnliftIO m, MonadLoggerIO m, StoreRead m) => CacheT m Integer+pruneDB =+ withLock go >>= \case+ Nothing -> return 0+ Just n -> return n where- flush n =+ go = do+ x <- asks cacheMax+ s <- runRedis Redis.dbsize+ if s > x+ then flush (s - x)+ else return 0+ 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 -touchKeys :: MonadIO m => [XPubSpec] -> CacheT m Integer+touchKeys :: MonadIO m => [XPubSpec] -> CacheT m () touchKeys xpubs = do now <- systemSeconds <$> liftIO getSystemTime- m <-- runRedisTx (redisWatchXPubCached xpubs) $ \ys -> do- let xpubs' = map fst . filter snd $ zip xpubs ys- redisTouchKeys now xpubs'- case m of- Nothing -> touchKeys xpubs- Just x -> return x+ runRedis $ redisTouchKeys now xpubs -redisTouchKeys ::- (Applicative f, RedisCtx m f, Real a) => a -> [XPubSpec] -> m (f Integer)-redisTouchKeys _ [] = return (pure 0)+redisTouchKeys :: (Monad f, RedisCtx m f, Real a) => a -> [XPubSpec] -> m (f ())+redisTouchKeys _ [] = return $ return () redisTouchKeys now xpubs =- Redis.zadd maxKey $ map ((realToFrac now, ) . encode) xpubs+ void <$> Redis.zadd maxKey (map ((realToFrac now, ) . encode) xpubs) cacheWriterReact :: (MonadUnliftIO m, MonadLoggerIO m, StoreRead m)@@ -490,37 +541,54 @@ lenNotNull bals = length $ filter (not . nullBalance . xPubBal) bals newXPubC ::- (MonadLoggerIO m, StoreRead m)+ (MonadUnliftIO m, MonadLoggerIO m, StoreRead m) => XPubSpec -> CacheT m ([XPubBal], Bool)-newXPubC xpub = do- bals <- lift $ xPubBals xpub- x <- asks cacheMin- let n = lenNotNull bals- if x <= n- then do- go bals- return (bals, True)- else do- return (bals, False)+newXPubC xpub =+ withLockWait $ do+ bals <- lift $ xPubBals xpub+ x <- asks cacheMin+ xpubtxt <- xpubText xpub+ let n = lenNotNull bals+ if x <= n+ then do+ go bals+ return (bals, True)+ else do+ $(logDebugS) "Cache" $+ "Not caching xpub with " <> cs (show n) <>+ " used addresses: " <>+ xpubtxt+ return (bals, False) where op XPubUnspent {xPubUnspent = u} = (unspentPoint u, unspentBlock u) go bals = do+ xpubtxt <- xpubText xpub+ $(logDebugS) "Cache" $+ "Caching xpub with " <> cs (show (length bals)) <>+ " addresses (used: " <>+ cs (show (lenNotNull bals)) <>+ "): " <>+ xpubtxt utxo <- lift $ xPubUnspents xpub Nothing 0 Nothing+ $(logDebugS) "Cache" $+ "Caching xpub with " <> cs (show (length utxo)) <> " utxos: " <>+ xpubtxt xtxs <- lift $ xPubTxs xpub Nothing 0 Nothing+ $(logDebugS) "Cache" $+ "Caching xpub with " <> cs (show (length xtxs)) <> " txs: " <>+ xpubtxt now <- systemSeconds <$> liftIO getSystemTime- runRedisTx (redisWatchXPubCached [xpub]) $ \ts ->- if and ts- then return (pure ())- else do- x <- redisAddXPubBalances xpub bals- y <- redisAddXPubUnspents xpub (map op utxo)- z <- redisAddXPubTxs xpub xtxs- t <- redisTouchKeys now [xpub]- return $ x >> y >> z >> t >> pure ()+ 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 () -newBlockC :: (MonadLoggerIO m, StoreRead m) => CacheT m ()+newBlockC :: (MonadUnliftIO m, MonadLoggerIO m, StoreRead m) => CacheT m () newBlockC =+ withLockWait $ lift getBestBlock >>= \case Nothing -> $(logErrorS) "Cache" "Best block not set yet" Just newhead -> do@@ -578,17 +646,17 @@ importBlockC (headerHash (nodeHeader a)) newBlockC -newTxC :: (MonadLoggerIO m, StoreRead m) => TxHash -> CacheT m ()+newTxC :: (MonadUnliftIO m, MonadLoggerIO m, StoreRead m) => TxHash -> CacheT m () newTxC th =+ withLockWait $ lift (getTxData th) >>= \case- Just txd ->- importMultiTxC [txd] >>= \case- False -> newTxC th- True -> return ()+ Just txd -> do+ $(logDebugS) "Cache" $ "Importing transaction: " <> txHashToHex th+ importMultiTxC [txd] Nothing -> $(logErrorS) "Cache" $ "Transaction not found: " <> txHashToHex th -importBlockC :: (StoreRead m, MonadLoggerIO m) => BlockHash -> CacheT m ()+importBlockC :: (MonadUnliftIO m, StoreRead m, MonadLoggerIO m) => BlockHash -> CacheT m () importBlockC bh = lift (getBlock bh) >>= \case Nothing -> do@@ -596,17 +664,19 @@ throwIO . LogicError . cs $ "Could not get block: " <> blockHashToHex bh Just bd -> do- $(logInfoS) "Cache" $ "Importing block: " <> blockHashToHex bh go bd where go bd = do let ths = blockDataTxs bd tds <- sortTxData . catMaybes <$> mapM (lift . getTxData) ths- importMultiTxC tds >>= \case- False -> return ()- True -> cacheSetHead bh+ $(logInfoS) "Cache" $+ "Importing " <> cs (show (length tds)) <>+ " transactions from block " <>+ blockHashToHex bh+ importMultiTxC tds+ cacheSetHead bh -removeHeadC :: (StoreRead m, MonadLoggerIO m) => CacheT m ()+removeHeadC :: (StoreRead m, MonadUnliftIO m, MonadLoggerIO m) => CacheT m () removeHeadC = void . runMaybeT $ do bh <- MaybeT cacheGetHead@@ -616,45 +686,56 @@ sortTxData . catMaybes <$> mapM (lift . getTxData) (blockDataTxs bd) $(logWarnS) "Cache" $ "Reverting head: " <> blockHashToHex bh- importMultiTxC tds >>= \case- True -> cacheSetHead (prevBlock (blockDataHeader bd))- False -> return ()+ importMultiTxC tds+ cacheSetHead (prevBlock (blockDataHeader bd)) -importMultiTxC :: (StoreRead m, MonadLoggerIO m) => [TxData] -> CacheT m Bool+importMultiTxC ::+ (MonadUnliftIO m, StoreRead m, MonadLoggerIO m)+ => [TxData]+ -> CacheT m () importMultiTxC txs = do- addrmap <-- HashMap.fromList . catMaybes . zipWith (\a -> fmap (a, )) alladdrs <$>- cacheGetAddrsInfo alladdrs+ forM_ (zip [(1 :: Int) ..] txs) $ \(i, tx) ->+ $(logDebugS) "Cache" $+ "Import tx " <> cs (show i) <> "/" <> cs (show (length txs)) <> ": " <>+ txHashToHex (txHash (txData tx))+ addrmap <- getaddrmap let addrs = HashMap.keys addrmap- balmap <-- HashMap.fromList . zipWith (,) addrs <$> mapM (lift . getBalance) addrs- unspentmap <-- HashMap.fromList . catMaybes . zipWith (\p -> fmap (p, )) allops <$>- lift (mapM getUnspent allops)+ balmap <- getbalances addrs+ unspentmap <- getunspents gap <- getMaxGap now <- systemSeconds <$> liftIO getSystemTime- ms <-- runRedisTx (w (allxpubs addrmap)) $ \bals -> do- let xmap = HashMap.filter (not . null) (HashMap.fromList bals)- addrmap' = faddrmap (HashMap.keysSet xmap) addrmap- x <-- if not (HashMap.null xmap)- then do- x <- redisImportMultiTx addrmap' unspentmap txs- y <- redisUpdateBalances addrmap' balmap- z <- redisTouchKeys now (allxpubs addrmap')- return $ x >> y >> z >> return ()- else return (pure ())- return $ x >> return (getNewAddrs gap xmap (HashMap.elems addrmap'))- case ms of- Nothing -> return False- Just addrs' -> cacheAddAddresses addrs' >> return True+ let xpubs = allxpubs addrmap+ forM_ (zip [(1 :: Int) ..] xpubs) $ \(i, xpub) -> do+ xpubtxt <- xpubText xpub+ $(logDebugS) "Cache" $+ "Affected xpub " <> cs (show i) <> "/" <> cs (show (length xpubs)) <>+ ": " <>+ xpubtxt+ bals <- getxbals xpubs+ let xmap = HashMap.filter (not . null) (HashMap.fromList bals)+ addrs' = getNewAddrs gap xmap (HashMap.elems addrmap)+ runRedis $+ if not (HashMap.null xmap)+ then do+ x <- redisImportMultiTx addrmap unspentmap txs+ y <- redisUpdateBalances addrmap balmap+ z <- redisTouchKeys now xpubs+ return $ x >> y >> z >> return ()+ else return (return ())+ cacheAddAddresses addrs' where- w xpubs = do- bals <- watchRedisXPubBalances xpubs- addrs <- watchRedisXPubAddrs alladdrs- return $ addrs >> bals- faddrmap xpubs = HashMap.filter ((`elem` xpubs) . addressXPubSpec)+ getaddrmap =+ HashMap.fromList . catMaybes . zipWith (\a -> fmap (a, )) alladdrs <$>+ cacheGetAddrsInfo alladdrs+ getunspents =+ HashMap.fromList . catMaybes . zipWith (\p -> fmap (p, )) allops <$>+ lift (mapM getUnspent allops)+ getbalances addrs =+ HashMap.fromList . zipWith (,) addrs <$> mapM (lift . getBalance) addrs+ getxbals xpubs =+ runRedis . fmap sequence . forM xpubs $ \xpub -> do+ bs <- redisGetXPubBalances xpub+ return $ (xpub, ) <$> bs allops = map snd $ concatMap txInputs txs <> concatMap txOutputs txs alladdrs = nub . map fst $ concatMap txInputs txs <> concatMap txOutputs txs allxpubs addrmap = nub . map addressXPubSpec $ HashMap.elems addrmap@@ -702,8 +783,8 @@ y <- redisRemFromMempool (txHash (txData tx)) return $ sequence x >> y >> return () else do- tops <- mapM (uncurry (addtx tx)) (txaddrops tx)- mem <-+ a <- sequence <$> mapM (uncurry (addtx tx)) (txaddrops tx)+ b <- case txDataBlock tx of b@MemRef {} -> redisAddToMempool@@ -712,7 +793,7 @@ , blockTxBlock = b } _ -> redisRemFromMempool (txHash (txData tx))- return $ sequence tops >> mem >> return ()+ return $ a >> b >> return () txaddrops td = spnts td <> utxos td spnts td = txInputs td utxos td = txOutputs td@@ -733,41 +814,34 @@ XPubBal {xPubBalPath = addressXPubPath ainfo, xPubBal = bal} cacheAddAddresses ::- (StoreRead m, MonadLoggerIO m)+ (StoreRead m, MonadUnliftIO m, MonadLoggerIO m) => [(Address, AddressXPub)] -> CacheT m () cacheAddAddresses [] = return () cacheAddAddresses addrs = do+ $(logDebugS) "Cache" $+ "Adding " <> cs (show (length addrs)) <> " new generated addresses"+ balmap <- HashMap.fromListWith (<>) <$> mapM (uncurry getbal) addrs+ utxomap <- HashMap.fromListWith (<>) <$> mapM (uncurry getutxo) addrs+ txmap <- HashMap.fromListWith (<>) <$> mapM (uncurry gettxmap) addrs+ runRedis $ do+ a <- forM (HashMap.toList balmap) (uncurry redisAddXPubBalances)+ b <- forM (HashMap.toList utxomap) (uncurry redisAddXPubUnspents)+ c <- forM (HashMap.toList txmap) (uncurry redisAddXPubTxs)+ return $ sequence a >> sequence b >> sequence c >> return ()+ let xpubs = nub (map addressXPubSpec (Map.elems amap))+ xmap <- getbals xpubs gap <- getMaxGap- balmap' <- HashMap.fromListWith (<>) <$> mapM (uncurry getbal) addrs- utxomap' <- HashMap.fromListWith (<>) <$> mapM (uncurry getutxo) addrs- txmap' <- HashMap.fromListWith (<>) <$> mapM (uncurry gettxmap) addrs- let xpubs' = nub (map addressXPubSpec (Map.elems amap))- ms <-- runRedisTx (watchRedisXPubBalances xpubs') $ \bals -> do- let xmap = HashMap.filter (not . null) (HashMap.fromList bals)- xpubs = HashMap.keysSet xmap- balmap = HashMap.filterWithKey (\k _ -> k `elem` xpubs) balmap'- utxomap =- HashMap.filterWithKey (\k _ -> k `elem` xpubs) utxomap'- txmap = HashMap.filterWithKey (\k _ -> k `elem` xpubs) txmap'- notnulls = getnotnull balmap- x' <-- forM (HashMap.toList balmap) $ \(x, bs) ->- redisAddXPubBalances x bs- y' <-- forM (HashMap.toList utxomap) $ \(x, us) ->- redisAddXPubUnspents x us- z' <- forM (HashMap.toList txmap) $ \(x, ts) -> redisAddXPubTxs x ts- return $ do- _ <- sequence x'- _ <- sequence y'- _ <- sequence z'- return $ getNewAddrs gap xmap notnulls- case ms of- Nothing -> cacheAddAddresses addrs- Just addrs' -> cacheAddAddresses addrs'+ let notnulls = getnotnull balmap+ addrs' = getNewAddrs gap xmap notnulls+ cacheAddAddresses addrs' where+ getbals xpubs =+ runRedis $ do+ bs <- sequence <$> forM xpubs redisGetXPubBalances+ return $+ HashMap.filter (not . null) . HashMap.fromList . zip xpubs <$>+ bs amap = Map.fromList addrs getnotnull = let f xpub =@@ -805,14 +879,17 @@ Nothing -> [] Just bals -> addrsToAdd gap bals a -syncMempoolC :: (MonadLoggerIO m, StoreRead m) => CacheT m ()+syncMempoolC :: (MonadUnliftIO m, MonadLoggerIO m, StoreRead m) => CacheT m () syncMempoolC = do nodepool <- map blockTxHash <$> lift getMempool cachepool <- map blockTxHash <$> cacheGetMempool- txs <- catMaybes <$> mapM (lift . getTxData) (nodepool <> cachepool)- importMultiTxC txs >>= \case- True -> return ()- False -> syncMempoolC+ txs <- catMaybes <$> mapM (lift . getTxData) (nodepool \\ cachepool)+ if null txs+ then $(logDebugS) "Cache" "Cache mempool in sync"+ else do+ $(logDebugS) "Cache" $+ "Importing " <> cs (show (length txs)) <> " mempool transactions"+ importMultiTxC txs cacheGetMempool :: MonadIO m => CacheT m [BlockTx] cacheGetMempool = runRedis redisGetMempool@@ -820,8 +897,10 @@ cacheGetHead :: MonadIO m => CacheT m (Maybe BlockHash) cacheGetHead = runRedis redisGetHead -cacheSetHead :: MonadIO m => BlockHash -> CacheT m ()-cacheSetHead bh = runRedis (redisSetHead bh) >> return ()+cacheSetHead :: (MonadLoggerIO m, StoreRead m) => BlockHash -> CacheT m ()+cacheSetHead bh = do+ $(logDebugS) "Cache" $ "Cache head set to: " <> blockHashToHex bh+ runRedis (redisSetHead bh) >> return () cacheGetAddrsInfo :: MonadIO m => [Address] -> CacheT m [Maybe AddressXPub]@@ -836,37 +915,24 @@ redisRemFromMempool :: RedisCtx m f => TxHash -> m (f Integer) redisRemFromMempool th = zrem mempoolSetKey [encode th] -redisSetAddrInfo :: RedisCtx m f => Address -> AddressXPub -> m (f Redis.Status)-redisSetAddrInfo a i = Redis.set (addrPfx <> encode a) (encode i)+redisSetAddrInfo ::+ (Functor f, RedisCtx m f) => Address -> AddressXPub -> m (f ())+redisSetAddrInfo a i = void <$> Redis.set (addrPfx <> encode a) (encode i) -delXPubKeys :: MonadIO m => [XPubSpec] -> CacheT m Integer+delXPubKeys ::+ (MonadUnliftIO m, MonadLoggerIO m, StoreRead m)+ => [XPubSpec]+ -> CacheT m Integer+delXPubKeys [] = return 0 delXPubKeys xpubs = do- ms <-- runRedisTx (watchRedisXPubBalances xpubs) $ \xbals -> do- xs <- forM xbals $ uncurry redisDelXPubKeys- return $ sequence xs- case ms of- Nothing -> delXPubKeys xpubs- Just is -> return $ sum is--watchRedisXPubBalances ::- [XPubSpec] -> Redis (Either Reply [(XPubSpec, [XPubBal])])-watchRedisXPubBalances [] = return (pure [])-watchRedisXPubBalances xpubs = do- watch $ map (\xpub -> balancesPfx <> encode xpub) xpubs- bals' <- mapM redisGetXPubBalances xpubs- return $ do- bals <- sequence bals'- return $ zip xpubs bals--watchRedisXPubAddrs :: [Address] -> Redis (Either Reply [Maybe AddressXPub])-watchRedisXPubAddrs [] = return (pure [])-watchRedisXPubAddrs as = do- x <- watch $ map (\a -> addrPfx <> encode a) as- i <- redisGetAddrsInfo as- return $ x >> i--+ forM_ xpubs $ \x -> do+ xtxt <- xpubText x+ $(logDebugS) "Cache" $ "Deleting xpub: " <> xtxt+ xbals <-+ runRedis . fmap sequence . forM xpubs $ \xpub -> do+ bs <- redisGetXPubBalances xpub+ return $ (xpub, ) <$> bs+ runRedis $ fmap sum . sequence <$> forM xbals (uncurry redisDelXPubKeys) redisDelXPubKeys :: (Monad f, RedisCtx m f) => XPubSpec -> [XPubBal] -> m (f Integer)@@ -874,9 +940,9 @@ where go addrs = do addrcount <-- if null addrs- then return (pure 0)- else Redis.del (map ((addrPfx <>) . encode) addrs)+ case addrs of+ [] -> return (pure 0)+ _ -> Redis.del (map ((addrPfx <>) . encode) addrs) txsetcount <- Redis.del [txSetPfx <> encode xpub] utxocount <- Redis.del [utxoPfx <> encode xpub] balcount <- Redis.del [balancesPfx <> encode xpub]@@ -1004,3 +1070,8 @@ return $ map (uncurry f) ys where f t s = BlockTx {blockTxBlock = scoreBlockRef s, blockTxHash = t}++xpubText :: (MonadUnliftIO m, MonadLoggerIO m, StoreRead m) => XPubSpec -> CacheT m Text+xpubText xpub = do+ net <- getNetwork+ return . cs $ xPubExport net (xPubSpecKey xpub)
src/Haskoin/Store/Web.hs view
@@ -224,7 +224,8 @@ bdb <- asks (storeDB . webStore) lift $ withDatabaseReader bdb f -instance MonadLoggerIO m => StoreRead (ReaderT WebConfig m) where+instance (MonadUnliftIO m, MonadLoggerIO m) =>+ StoreRead (ReaderT WebConfig m) where getMaxGap = runInWebReader getMaxGap getMaxGap getInitialGap = runInWebReader getInitialGap getInitialGap getNetwork = runInWebReader getNetwork getNetwork@@ -250,8 +251,7 @@ (getAddressesUnspents addrs start limit) getOrphans = runInWebReader getOrphans getOrphans xPubBals xpub = runInWebReader (xPubBals xpub) (xPubBals xpub)- xPubSummary xpub =- runInWebReader (xPubSummary xpub) (xPubSummary xpub)+ xPubSummary xpub = runInWebReader (xPubSummary xpub) (xPubSummary xpub) xPubUnspents xpub start offset limit = runInWebReader (xPubUnspents xpub start offset limit)@@ -261,7 +261,7 @@ (xPubTxs xpub start offset limit) (xPubTxs xpub start offset limit) -instance MonadLoggerIO m => StoreRead (WebT m) where+instance (MonadUnliftIO m, MonadLoggerIO m) => StoreRead (WebT m) where getNetwork = lift getNetwork getBestBlock = lift getBestBlock getBlocksAtHeight = lift . getBlocksAtHeight@@ -335,7 +335,7 @@ S.raw (serialAnyNet net proto x) scottyBestBlock ::- (MonadLoggerIO m, MonadIO m) => Bool -> WebT m ()+ (MonadUnliftIO m, MonadLoggerIO m, MonadIO m) => Bool -> WebT m () scottyBestBlock raw = do limits <- lift $ asks webMaxLimits setHeaders@@ -355,7 +355,7 @@ rawBlock b >>= protoSerial proto else protoSerial proto (pruneTx n b) -scottyBlock :: MonadLoggerIO m => Bool -> WebT m ()+scottyBlock :: (MonadUnliftIO m, MonadLoggerIO m) => Bool -> WebT m () scottyBlock raw = do limits <- lift $ asks webMaxLimits setHeaders@@ -372,8 +372,7 @@ rawBlock b >>= protoSerial proto else protoSerial proto (pruneTx n b) -scottyBlockHeight ::- MonadLoggerIO m => Bool -> WebT m ()+scottyBlockHeight :: (MonadUnliftIO m, MonadLoggerIO m) => Bool -> WebT m () scottyBlockHeight raw = do limits <- lift $ asks webMaxLimits setHeaders@@ -392,7 +391,7 @@ let blocks' = map (pruneTx n) blocks protoSerial proto blocks' -scottyBlockTime :: MonadLoggerIO m => Bool -> WebT m ()+scottyBlockTime :: (MonadUnliftIO m, MonadLoggerIO m) => Bool -> WebT m () scottyBlockTime raw = do limits <- lift $ asks webMaxLimits setHeaders@@ -409,7 +408,7 @@ Just <$> rawBlock d else maybeSerial proto m -scottyBlockHeights :: MonadLoggerIO m => WebT m ()+scottyBlockHeights :: (MonadUnliftIO m, MonadLoggerIO m) => WebT m () scottyBlockHeights = do setHeaders heights <- S.param "heights"@@ -419,7 +418,7 @@ blocks <- map (pruneTx n) . catMaybes <$> mapM getBlock bhs protoSerial proto blocks -scottyBlockLatest :: MonadLoggerIO m => WebT m ()+scottyBlockLatest :: (MonadUnliftIO m, MonadLoggerIO m) => WebT m () scottyBlockLatest = do setHeaders n <- parseNoTx@@ -443,7 +442,7 @@ else (b' :) <$> go i' n prev -scottyBlocks :: MonadLoggerIO m => WebT m ()+scottyBlocks :: (MonadUnliftIO m, MonadLoggerIO m) => WebT m () scottyBlocks = do setHeaders bhs <- map (\(MyBlockHash h) -> h) <$> S.param "blocks"@@ -452,14 +451,14 @@ bks <- map (pruneTx n) . catMaybes <$> mapM getBlock (nub bhs) protoSerial proto bks -scottyMempool :: MonadLoggerIO m => WebT m ()+scottyMempool :: (MonadUnliftIO m, MonadLoggerIO m) => WebT m () scottyMempool = do setHeaders proto <- setupBin txs <- map blockTxHash <$> getMempool protoSerial proto txs -scottyTransaction :: MonadLoggerIO m => WebT m ()+scottyTransaction :: (MonadUnliftIO m, MonadLoggerIO m) => WebT m () scottyTransaction = do setHeaders MyTxHash txid <- S.param "txid"@@ -467,7 +466,7 @@ res <- getTransaction txid maybeSerialNet proto res -scottyRawTransaction :: MonadLoggerIO m => WebT m ()+scottyRawTransaction :: (MonadUnliftIO m, MonadLoggerIO m) => WebT m () scottyRawTransaction = do setHeaders MyTxHash txid <- S.param "txid"@@ -475,7 +474,7 @@ res <- fmap transactionData <$> getTransaction txid maybeSerial proto res -scottyTxAfterHeight :: MonadLoggerIO m => WebT m ()+scottyTxAfterHeight :: (MonadUnliftIO m, MonadLoggerIO m) => WebT m () scottyTxAfterHeight = do setHeaders MyTxHash txid <- S.param "txid"@@ -484,7 +483,7 @@ res <- cbAfterHeight 10000 height txid protoSerial proto res -scottyTransactions :: MonadLoggerIO m => WebT m ()+scottyTransactions :: (MonadUnliftIO m, MonadLoggerIO m) => WebT m () scottyTransactions = do setHeaders txids <- map (\(MyTxHash h) -> h) <$> S.param "txids"@@ -492,7 +491,7 @@ txs <- catMaybes <$> mapM getTransaction (nub txids) protoSerialNet proto txs -scottyBlockTransactions :: MonadLoggerIO m => WebT m ()+scottyBlockTransactions :: (MonadUnliftIO m, MonadLoggerIO m) => WebT m () scottyBlockTransactions = do limits <- lift $ asks webMaxLimits setHeaders@@ -507,7 +506,7 @@ Nothing -> S.raise ThingNotFound scottyRawTransactions ::- MonadLoggerIO m => WebT m ()+ (MonadUnliftIO m, MonadLoggerIO m) => WebT m () scottyRawTransactions = do setHeaders txids <- map (\(MyTxHash h) -> h) <$> S.param "txids"@@ -523,7 +522,7 @@ return Block {blockHeader = h, blockTxns = txs} scottyRawBlockTransactions ::- MonadLoggerIO m => WebT m ()+ (MonadUnliftIO m, MonadLoggerIO m) => WebT m () scottyRawBlockTransactions = do limits <- lift $ asks webMaxLimits setHeaders@@ -537,7 +536,7 @@ protoSerial proto txs Nothing -> S.raise ThingNotFound -scottyAddressTxs :: MonadLoggerIO m => Bool -> WebT m ()+scottyAddressTxs :: (MonadUnliftIO m, MonadLoggerIO m) => Bool -> WebT m () scottyAddressTxs full = do setHeaders a <- parseAddress@@ -552,7 +551,7 @@ getAddressTxsLimit o l s a >>= protoSerial proto scottyAddressesTxs ::- MonadLoggerIO m => Bool -> WebT m ()+ (MonadUnliftIO m, MonadLoggerIO m) => Bool -> WebT m () scottyAddressesTxs full = do setHeaders as <- parseAddresses@@ -563,7 +562,7 @@ then getAddressesTxsFull l s as >>= protoSerialNet proto else getAddressesTxsLimit l s as >>= protoSerial proto -scottyAddressUnspent :: MonadLoggerIO m => WebT m ()+scottyAddressUnspent :: (MonadUnliftIO m, MonadLoggerIO m) => WebT m () scottyAddressUnspent = do setHeaders a <- parseAddress@@ -574,7 +573,7 @@ uns <- getAddressUnspentsLimit o l s a protoSerial proto uns -scottyAddressesUnspent :: MonadLoggerIO m => WebT m ()+scottyAddressesUnspent :: (MonadUnliftIO m, MonadLoggerIO m) => WebT m () scottyAddressesUnspent = do setHeaders as <- parseAddresses@@ -584,7 +583,7 @@ uns <- getAddressesUnspentsLimit l s as protoSerial proto uns -scottyAddressBalance :: MonadLoggerIO m => WebT m ()+scottyAddressBalance :: (MonadUnliftIO m, MonadLoggerIO m) => WebT m () scottyAddressBalance = do setHeaders a <- parseAddress@@ -592,7 +591,7 @@ res <- getBalance a protoSerialNet proto res -scottyAddressesBalances :: MonadLoggerIO m => WebT m ()+scottyAddressesBalances :: (MonadUnliftIO m, MonadLoggerIO m) => WebT m () scottyAddressesBalances = do setHeaders as <- parseAddresses@@ -600,7 +599,7 @@ res <- getBalances as protoSerialNet proto res -scottyXpubBalances :: MonadLoggerIO m => WebT m ()+scottyXpubBalances :: (MonadUnliftIO m, MonadLoggerIO m) => WebT m () scottyXpubBalances = do setHeaders nocache <- parseNoCache@@ -613,7 +612,7 @@ else xPubBals xpub protoSerialNet proto res -scottyXpubTxs :: MonadLoggerIO m => Bool -> WebT m ()+scottyXpubTxs :: (MonadUnliftIO m, MonadLoggerIO m) => Bool -> WebT m () scottyXpubTxs full = do setHeaders nocache <- parseNoCache@@ -632,7 +631,7 @@ protoSerialNet proto txs' else protoSerial proto txs -scottyXpubEvict :: MonadLoggerIO m => WebT m ()+scottyXpubEvict :: (MonadUnliftIO m, MonadLoggerIO m) => WebT m () scottyXpubEvict = lift (asks (storeCache . webStore)) >>= \case Nothing -> S.raise ThingNotFound@@ -640,11 +639,11 @@ setHeaders xpub <- parseXpub proto <- setupBin- n <- withCache cache (delXPubKeys [xpub])+ n <- lift $ withCache cache (delXPubKeys [xpub]) protoSerial proto (GenericResult (n > 0)) -scottyXpubUnspents :: MonadLoggerIO m => WebT m ()+scottyXpubUnspents :: (MonadUnliftIO m, MonadLoggerIO m) => WebT m () scottyXpubUnspents = do setHeaders nocache <- parseNoCache@@ -658,7 +657,7 @@ else xPubUnspents xpub start 0 limit protoSerial proto uns -scottyXpubSummary :: MonadLoggerIO m => WebT m ()+scottyXpubSummary :: (MonadUnliftIO m, MonadLoggerIO m) => WebT m () scottyXpubSummary = do setHeaders nocache <- parseNoCache@@ -809,7 +808,7 @@ S.get "/health" scottyHealth S.notFound $ S.raise ThingNotFound -getStart :: MonadLoggerIO m => WebT m (Maybe BlockRef)+getStart :: (MonadUnliftIO m, MonadLoggerIO m) => WebT m (Maybe BlockRef) getStart = runMaybeT $ do s <- MaybeT $ (Just <$> S.param "height") `S.rescue` const (return Nothing)