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: 6b4f5ffd53487d5c48dfefb572b8181be964d07da949370f1fd9d5fd51146fe3 +-- hash: 233badef6c274ea380714d9bc0492c96784d8cceb6012c6f476caaf9f45a8352 name: haskoin-store -version: 0.36.4 +version: 0.37.0 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.36.4 + , haskoin-store-data ==0.37.0 , hedis >=0.12.13 , http-types >=0.12.3 , monad-logger >=0.3.32 @@ -61,7 +61,7 @@ , network >=3.1.1.1 , nqe >=0.6.1 , random >=1.1 - , rocksdb-haskell-jprupp >=2.0.0 + , rocksdb-haskell-jprupp >=2.1.3 , rocksdb-query >=0.4.0 , scotty >=0.11.5 , string-conversions >=0.4.0.1 @@ -96,7 +96,7 @@ , haskoin-core >=0.13.6 , haskoin-node >=0.14.1 , haskoin-store - , haskoin-store-data ==0.36.4 + , haskoin-store-data ==0.37.0 , 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.36.4 - , haskoin-store-data ==0.36.4 + , haskoin-store ==0.37.0 + , haskoin-store-data ==0.37.0 , hedis >=0.12.13 , hspec >=2.7.1 , http-types >=0.12.3 @@ -145,7 +145,7 @@ , network >=3.1.1.1 , nqe >=0.6.1 , random >=1.1 - , rocksdb-haskell-jprupp >=2.0.0 + , rocksdb-haskell-jprupp >=2.1.3 , rocksdb-query >=0.4.0 , scotty >=0.11.5 , string-conversions >=0.4.0.1 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 @@ -18,6 +18,7 @@ , CacheWriter , CacheWriterInbox , cacheNewBlock + , cacheNewTx , cachePing , cacheWriter , isInCache @@ -28,7 +29,7 @@ import Control.Monad (forM, forM_, forever, unless, void, when) import Control.Monad.Logger (MonadLoggerIO, logDebugS, logErrorS, - logWarnS) + logInfoS, logWarnS) import Control.Monad.Reader (ReaderT (..), ask, asks) import Control.Monad.Trans (lift) import Control.Monad.Trans.Maybe (MaybeT (..), runMaybeT) @@ -36,14 +37,14 @@ import Data.ByteString (ByteString) import qualified Data.ByteString.Short as BSS import Data.Default (def) -import Data.Either (rights) +import Data.Either (lefts, rights) import Data.HashMap.Strict (HashMap) import qualified Data.HashMap.Strict as HashMap import qualified Data.HashSet as HashSet import qualified Data.IntMap.Strict as I import Data.List (sort) import qualified Data.Map.Strict as Map -import Data.Maybe (catMaybes, mapMaybe) +import Data.Maybe (catMaybes, isNothing, mapMaybe) import Data.Serialize (Serialize, decode, encode) import Data.String.Conversions (cs) import Data.Text (Text) @@ -64,7 +65,7 @@ blockHashToHex, derivePubPath, eitherToMaybe, headerHash, pathToList, scriptToAddressBS, - txHash, xPubAddr, + txHash, txHashToHex, xPubAddr, xPubCompatWitnessAddr, xPubExport, xPubWitnessAddr) import Haskoin.Node (Chain, chainBlockMain, @@ -85,11 +86,11 @@ runRedis :: MonadLoggerIO m => Redis (Either Reply a) -> CacheX m a runRedis action = asks cacheConn >>= \conn -> - liftIO (Redis.runRedis conn action) >>= \case - Right x -> return x - Left e -> do - $(logErrorS) "Cache" $ "Got error from Redis: " <> cs (show e) - throwIO (RedisError e) + liftIO (Redis.runRedis conn action) >>= \case + Right x -> return x + Left e -> do + $(logErrorS) "Cache" $ "Got error from Redis: " <> cs (show e) + throwIO (RedisError e) data CacheConfig = CacheConfig { cacheConn :: !Connection @@ -409,6 +410,7 @@ data CacheWriterMessage = CacheNewBlock + | CacheNewTx !TxHash | CachePing type CacheWriterInbox = Inbox CacheWriterMessage @@ -432,6 +434,9 @@ maxKey :: ByteString maxKey = "max" +isAnythingCached :: MonadLoggerIO m => CacheX m Bool +isAnythingCached = runRedis $ Redis.exists maxKey + xPubAddrFunction :: DeriveType -> XPubKey -> Address xPubAddrFunction DeriveNormal = xPubAddr xPubAddrFunction DeriveP2SH = xPubCompatWitnessAddr @@ -475,7 +480,7 @@ liftIO . Redis.runRedis conn $ do let opts = Redis.SetOpts - { Redis.setSeconds = Just 300 + { Redis.setSeconds = Just 30 , Redis.setMilliseconds = Nothing , Redis.setCondition = Just Redis.Nx } @@ -492,15 +497,15 @@ ": not locked" Just bs -> if read (cs bs) == i - then do - void $ runRedis (Redis.del ["lock"]) - $(logDebugS) "Cache" $ - "Released lock with value " <> - cs (show i) - else - $(logErrorS) "Cache" $ - "Could not release lock: value is not " <> - cs (show i) + then do + void $ runRedis (Redis.del ["lock"]) + $(logDebugS) "Cache" $ + "Released lock with value " <> + cs (show i) + else + $(logErrorS) "Cache" $ + "Could not release lock: value is not " <> + cs (show i) withLock :: (MonadLoggerIO m, MonadUnliftIO m) @@ -513,24 +518,30 @@ pruneDB :: (MonadUnliftIO m, MonadLoggerIO m, StoreReadBase m) => CacheX m Integer -pruneDB = do - x <- asks cacheMax - s <- runRedis Redis.dbsize - if s > x then flush (s - x) else return 0 +pruneDB = + isAnythingCached >>= \case + True -> do + x <- asks cacheMax + s <- runRedis Redis.dbsize + if s > x + then flush (s - x) + else return 0 + False -> return 0 where flush n = case min 1000 (n `div` 64) of - 0 -> return 0 - x -> do - m <- withLock $ do - ks <- fmap (map fst) . runRedis $ - getFromSortedSet maxKey Nothing 0 (fromIntegral x) - $(logDebugS) "Cache" $ - "Pruning " <> cs (show (length ks)) <> " old xpubs" - delXPubKeys ks - case m of - Nothing -> return 0 - Just y -> return y + 0 -> return 0 + x -> do + m <- withLock $ do + ks <- + fmap (map fst) . runRedis $ + getFromSortedSet maxKey Nothing 0 (fromIntegral x) + $(logDebugS) "Cache" $ + "Pruning " <> cs (show (length ks)) <> " old xpubs" + delXPubKeys ks + case m of + Nothing -> return 0 + Just y -> return y touchKeys :: MonadLoggerIO m => [XPubSpec] -> CacheX m () touchKeys xpubs = do @@ -545,13 +556,9 @@ cacheWriterReact :: (MonadUnliftIO m, MonadLoggerIO m, StoreReadExtra m) => CacheWriterMessage -> CacheX m () -cacheWriterReact CacheNewBlock = do - newBlockC - syncMempoolC -cacheWriterReact CachePing = do - pruneDB - newBlockC - syncMempoolC +cacheWriterReact CacheNewBlock = newBlockC +cacheWriterReact (CacheNewTx th) = newTxC th +cacheWriterReact CachePing = pruneDB >> newBlockC >> syncMempoolC lenNotNull :: [XPubBal] -> Int lenNotNull bals = length $ filter (not . nullBalance . xPubBal) bals @@ -559,16 +566,21 @@ newXPubC :: (MonadUnliftIO m, MonadLoggerIO m, StoreReadExtra m) => XPubSpec -> CacheX m (Bool, [XPubBal]) -newXPubC xpub = do - bals <- lift $ xPubBals xpub - x <- asks cacheMin - t <- xpubText xpub - let n = lenNotNull bals - if x <= n - then go bals - else do - $(logDebugS) "Cache" $ "Not caching xpub: " <> t - return (False, bals) +newXPubC xpub = + cachePrime >>= \case + 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 + else do + $(logDebugS) "Cache" $ + "Not caching xpub with " <> cs (show n) <> + " used addresses: " <> t + return (False, bals) where op XPubUnspent {xPubUnspent = u} = (unspentPoint u, unspentBlock u) go bals = do @@ -599,73 +611,102 @@ $(logDebugS) "Cache" $ "Done caching xpub: " <> xpubtxt return (True, bals) +newTxC :: (MonadUnliftIO m, MonadLoggerIO m, StoreReadExtra m) + => TxHash -> CacheX m () +newTxC th = do + m <- withLock $ isAnythingCached >>= \case + False -> + $(logDebugS) "Cache" $ + "Not importing tx " <> txHashToHex th <> " because cache is empty" + True -> + runRedis (Redis.zscore mempoolSetKey (encode th)) >>= \case + Nothing -> + lift (getTxData th) >>= \case + Just td -> do + importMultiTxC [td] + $(logDebugS) "Cache" $ + "Updated cache with tx: " <> txHashToHex th + Nothing -> + $(logErrorS) "Cache" $ + "Tx not found in db: " <> txHashToHex th + Just _ -> + $(logDebugS) "Cache" $ + "Tx already in cache: " <> txHashToHex th + when (isNothing m) $ + $(logErrorS) "Cache" $ + "Could not get lock to cache tx: " <> txHashToHex th + newBlockC :: (MonadUnliftIO m, MonadLoggerIO m, StoreReadExtra m) => CacheX m () newBlockC = - lift getBestBlock >>= \m -> - forM_ m $ \bb -> - in_sync bb >>= \s -> - when s $ void $ withLock $ f bb + withLock f >>= \m -> + when (isNothing m) $ + $(logErrorS) "Cache" "Could not get lock to add add block to cache" where - f bb = cacheGetHead >>= go bb - in_sync bb = - asks cacheChain >>= \ch -> - chainGetBest ch >>= \cb -> - return $ headerHash (nodeHeader cb) == bb - go bb Nothing = - importBlockC bb - go bb (Just cb) - | cb == bb = - $(logDebugS) "Cache" "Cache in sync" - | otherwise = - sync bb cb - sync bb cb = - asks cacheChain >>= \ch -> - chainBlockMain bb ch >>= \case + f = isAnythingCached >>= \case False -> - $(logErrorS) "Cache" $ - "New head not in main chain: " <> blockHashToHex bb - True -> - chainGetBlock cb ch >>= \case - Nothing -> - $(logErrorS) "Cache" $ - "Cache head block node not found: " <> - blockHashToHex cb - Just cn -> - chainBlockMain cb ch >>= \case - False -> do - $(logDebugS) "Cache" $ - "Reverting cache head not in main chain: " <> - blockHashToHex cb - removeHeadC >> f bb - True -> - chainGetBlock bb ch >>= \case - Just nn -> next bb nn cn - Nothing -> do + $(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 = + $(logDebugS) "Cache" $ + "Blocks in sync: " <> blockHashToHex cachehead + | otherwise = do + ch <- asks cacheChain + chainBlockMain newhead ch >>= \case + False -> + $(logErrorS) "Cache" $ + "New head not in main chain: " <> blockHashToHex newhead + True -> chainGetBlock cachehead ch >>= \case + Nothing -> $(logErrorS) "Cache" $ - "Cache head node not found: " - <> blockHashToHex bb - throwIO $ - LogicError $ - "Cache head node not found: " - <> cs (blockHashToHex bb) - next bb nn cn = - asks cacheChain >>= \ch -> - chainGetAncestor (nodeHeight cn + 1) nn ch >>= \case - Nothing -> - $(logWarnS) "Cache" $ - "Ancestor not found at height " - <> cs (show (nodeHeight cn + 1)) - <> " for block: " - <> blockHashToHex (headerHash (nodeHeader nn)) - Just cn' -> - importBlockC (headerHash (nodeHeader cn')) >> f bb + "Cache head block node not found: " <> + blockHashToHex cachehead + Just cacheheadnode -> chainBlockMain cachehead ch >>= \case + False -> do + $(logDebugS) "Cache" $ + "Reverting cache head not in main chain: " <> + blockHashToHex cachehead + removeHeadC >> f + True -> chainGetBlock newhead ch >>= \case + Nothing -> do + $(logErrorS) "Cache" $ + "Cache head node not found: " <> + blockHashToHex newhead + throwIO $ + LogicError $ + "Cache head node not found: " <> + cs (blockHashToHex newhead) + Just newheadnode -> + next newheadnode cacheheadnode + next newheadnode cacheheadnode = do + ch <- asks cacheChain + chainGetAncestor (nodeHeight cacheheadnode + 1) newheadnode ch >>= \case + Nothing -> + $(logWarnS) "Cache" $ + "Ancestor not found at height " <> + cs (show (nodeHeight cacheheadnode + 1)) <> + " for block: " <> + blockHashToHex (headerHash (nodeHeader newheadnode)) + Just newcachenode -> + importBlockC (headerHash (nodeHeader newcachenode)) >> f importBlockC :: (MonadUnliftIO m, StoreReadExtra m, MonadLoggerIO m) => BlockHash -> CacheX m () importBlockC bh = lift (getBlock bh) >>= \case - Just bd -> do + Nothing -> do + $(logErrorS) "Cache" $ "Could not get block: " <> blockHashToHex bh + throwIO . LogicError . cs $ + "Could not get block: " <> blockHashToHex bh + Just bd -> + go bd + where + go bd = do let ths = blockDataTxs bd tds <- sortTxData . catMaybes <$> mapM (lift . getTxData) ths $(logDebugS) "Cache" $ @@ -678,31 +719,23 @@ " transactions from block " <> blockHashToHex bh cacheSetHead bh - Nothing -> do - $(logErrorS) "Cache" $ - "Could not get block: " - <> blockHashToHex bh - throwIO . LogicError . cs $ - "Could not get block: " - <> blockHashToHex bh removeHeadC :: (StoreReadExtra m, MonadUnliftIO m, MonadLoggerIO m) => CacheX m () removeHeadC = void . runMaybeT $ do - bh <- MaybeT cacheGetHead - bd <- MaybeT (lift (getBlock bh)) - lift $ do - tds <- sortTxData . catMaybes <$> - mapM (lift . getTxData) (blockDataTxs bd) - $(logDebugS) "Cache" $ "Reverting head: " <> blockHashToHex bh - importMultiTxC tds - $(logWarnS) "Cache" $ - "Reverted block head " - <> blockHashToHex bh - <> " to parent " - <> blockHashToHex (prevBlock (blockDataHeader bd)) - cacheSetHead (prevBlock (blockDataHeader bd)) + bh <- MaybeT cacheGetHead + bd <- MaybeT (lift (getBlock bh)) + lift $ do + tds <- + sortTxData . catMaybes <$> + mapM (lift . getTxData) (blockDataTxs bd) + $(logDebugS) "Cache" $ "Reverting head: " <> blockHashToHex bh + importMultiTxC tds + $(logWarnS) "Cache" $ + "Reverted block head " <> blockHashToHex bh <> " to parent " <> + blockHashToHex (prevBlock (blockDataHeader bd)) + cacheSetHead (prevBlock (blockDataHeader bd)) importMultiTxC :: (MonadUnliftIO m, StoreReadExtra m, MonadLoggerIO m) @@ -710,20 +743,16 @@ importMultiTxC txs = do $(logDebugS) "Cache" $ "Processing " <> cs (show (length txs)) <> " txs" $(logDebugS) "Cache" $ - "Getting address information for " - <> cs (show (length alladdrs)) - <> " addresses" + "Getting address information for " <> cs (show (length alladdrs)) <> + " addresses" addrmap <- getaddrmap let addrs = HashMap.keys addrmap $(logDebugS) "Cache" $ - "Getting balances for " - <> cs (show (HashMap.size addrmap)) - <> " addresses" + "Getting balances for " <> cs (show (HashMap.size addrmap)) <> + " addresses" balmap <- getbalances addrs $(logDebugS) "Cache" $ - "Getting unspent data for " - <> cs (show (length allops)) - <> " outputs" + "Getting unspent data for " <> cs (show (length allops)) <> " outputs" unspentmap <- getunspents gap <- lift getMaxGap now <- systemSeconds <$> liftIO getSystemTime @@ -731,13 +760,13 @@ forM_ (zip [(1 :: Int) ..] xpubs) $ \(i, xpub) -> do xpubtxt <- xpubText xpub $(logDebugS) "Cache" $ - "Affected xpub " - <> cs (show i) <> "/" <> cs (show (length xpubs)) - <> ": " <> xpubtxt + "Affected xpub " <> cs (show i) <> "/" <> cs (show (length xpubs)) <> + ": " <> + xpubtxt addrs' <- do $(logDebugS) "Cache" $ - "Getting xpub balances for " - <> cs (show (length xpubs)) <> " xpubs" + "Getting xpub balances for " <> cs (show (length xpubs)) <> + " xpubs" xmap <- getxbals xpubs let addrmap' = faddrmap (HashMap.keysSet xmap) addrmap $(logDebugS) "Cache" "Starting Redis import pipeline" @@ -761,9 +790,10 @@ getbalances addrs = HashMap.fromList . zip addrs <$> mapM (lift . getDefaultBalance) addrs getxbals xpubs = do - bals <- runRedis . fmap sequence . forM xpubs $ \xpub -> do - bs <- redisGetXPubBalances xpub - return $ (xpub, ) <$> bs + bals <- + runRedis . fmap sequence . forM xpubs $ \xpub -> do + bs <- redisGetXPubBalances xpub + return $ (xpub, ) <$> bs return $ HashMap.filter (not . null) (HashMap.fromList bals) allops = map snd $ concatMap txInputs txs <> concatMap txOutputs txs alladdrs = @@ -785,43 +815,49 @@ where uns p i = case HashMap.lookup p unspentmap of - Just u -> - redisAddXPubUnspents (addressXPubSpec i) [(p, unspentBlock u)] - Nothing -> redisRemXPubUnspents (addressXPubSpec i) [p] + Just u -> + redisAddXPubUnspents (addressXPubSpec i) [(p, unspentBlock u)] + Nothing -> redisRemXPubUnspents (addressXPubSpec i) [p] addtx tx a p = case HashMap.lookup a addrmap of - Just i -> do - let tr = TxRef { txRefHash = txHash (txData tx) - , txRefBlock = txDataBlock tx - } - x <- redisAddXPubTxs (addressXPubSpec i) [tr] - y <- uns p i - return $ x >> y >> return () - Nothing -> return (pure ()) + Just i -> do + x <- + redisAddXPubTxs + (addressXPubSpec i) + [ TxRef + { txRefHash = txHash (txData tx) + , txRefBlock = txDataBlock tx + } + ] + y <- uns p i + return $ x >> y >> return () + Nothing -> return (pure ()) remtx tx a p = case HashMap.lookup a addrmap of - Just i -> do - x <- redisRemXPubTxs (addressXPubSpec i) [txHash (txData tx)] - y <- uns p i - return $ x >> y >> return () - Nothing -> return (pure ()) + Just i -> do + x <- redisRemXPubTxs (addressXPubSpec i) [txHash (txData tx)] + y <- uns p i + return $ x >> y >> return () + Nothing -> return (pure ()) importtxentries tx = if txDataDeleted tx - then do - x <- mapM (uncurry (remtx tx)) (txaddrops tx) - y <- redisRemFromMempool [txHash (txData tx)] - return $ sequence_ x >> void y - else do - a <- sequence <$> mapM (uncurry (addtx tx)) (txaddrops tx) - b <- - case txDataBlock tx of - b@MemRef {} -> - let tr = TxRef { txRefHash = txHash (txData tx) - , txRefBlock = b - } - in redisAddToMempool [tr] - _ -> redisRemFromMempool [txHash (txData tx)] - return $ a >> b >> return () + then do + x <- mapM (uncurry (remtx tx)) (txaddrops tx) + y <- redisRemFromMempool [txHash (txData tx)] + return $ sequence_ x >> void y + else do + a <- sequence <$> mapM (uncurry (addtx tx)) (txaddrops tx) + b <- + case txDataBlock tx of + b@MemRef {} -> + redisAddToMempool $ + (: []) + TxRef + { txRefHash = txHash (txData tx) + , txRefBlock = b + } + _ -> redisRemFromMempool [txHash (txData tx)] + return $ a >> b >> return () txaddrops td = txInputs td <> txOutputs td redisUpdateBalances :: @@ -831,10 +867,10 @@ -> m (f ()) redisUpdateBalances addrmap balmap = fmap (fmap mconcat . sequence) . forM (HashMap.keys addrmap) $ \a -> - case (HashMap.lookup a addrmap, HashMap.lookup a balmap) of - (Just ainfo, Just bal) -> - redisAddXPubBalances (addressXPubSpec ainfo) [xpubbal ainfo bal] - _ -> return (pure ()) + case (HashMap.lookup a addrmap, HashMap.lookup a balmap) of + (Just ainfo, Just bal) -> + redisAddXPubBalances (addressXPubSpec ainfo) [xpubbal ainfo bal] + _ -> return (pure ()) where xpubbal ainfo bal = XPubBal {xPubBalPath = addressXPubPath ainfo, xPubBal = bal} @@ -860,10 +896,7 @@ c <- forM (HashMap.toList txmap) (uncurry redisAddXPubTxs) return $ sequence_ a >> sequence_ b >> sequence_ c $(logDebugS) "Cache" "Completed Redis pipeline" - let xpubs = HashSet.toList - . HashSet.fromList - . map addressXPubSpec - $ Map.elems amap + let xpubs = HashSet.toList . HashSet.fromList . map addressXPubSpec $ Map.elems amap $(logDebugS) "Cache" "Getting xpub balances" xmap <- getbals xpubs gap <- lift getMaxGap @@ -871,10 +904,12 @@ 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 + 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 = @@ -886,66 +921,57 @@ g = filter (not . nullBalance . xPubBal) in concatMap (uncurry f) . HashMap.toList . HashMap.map g getbal a i = - let f b = ( addressXPubSpec i - , [XPubBal {xPubBal = b, xPubBalPath = addressXPubPath i}] ) + let f b = + ( addressXPubSpec i + , [XPubBal {xPubBal = b, xPubBalPath = addressXPubPath i}]) in f <$> lift (getDefaultBalance a) getutxo a i = - let f us = ( addressXPubSpec i - , map (\u -> (unspentPoint u, unspentBlock u)) us ) + let f us = + ( addressXPubSpec i + , map (\u -> (unspentPoint u, unspentBlock u)) us) in f <$> lift (getAddressUnspents a def) gettxmap a i = let f ts = (addressXPubSpec i, ts) in f <$> lift (getAddressTxs a def) -getNewAddrs :: KeyIndex - -> HashMap XPubSpec [XPubBal] - -> [AddressXPub] - -> [(Address, AddressXPub)] -getNewAddrs gap xpubs = - concatMap $ \a -> - case HashMap.lookup (addressXPubSpec a) xpubs of - Nothing -> [] - Just bals -> addrsToAdd gap bals a - -cacheCoolKey :: ByteString -cacheCoolKey = "cooldown" - -isCool :: (MonadUnliftIO m, MonadLoggerIO m) => CacheX m Bool -isCool = - runRedis (Redis.get cacheCoolKey) >>= \case - Nothing -> return True - Just bs -> do - let t = read (cs bs) - now <- microseconds - return (cooldown <= now - t) +getNewAddrs :: + KeyIndex + -> HashMap XPubSpec [XPubBal] + -> [AddressXPub] + -> [(Address, AddressXPub)] +getNewAddrs gap xpubs = concatMap f where - cooldown = 250000 - -startCooldown :: (MonadUnliftIO m, MonadLoggerIO m) => CacheX m () -startCooldown = do - now <- microseconds - void $ runRedis (Redis.set cacheCoolKey (cs (show now))) + f a = + case HashMap.lookup (addressXPubSpec a) xpubs of + Nothing -> [] + Just bals -> addrsToAdd gap bals a syncMempoolC :: (MonadUnliftIO m, MonadLoggerIO m, StoreReadExtra m) => CacheX m () -syncMempoolC = - void . withLock $ isCool >>= \cool -> when cool $ do +syncMempoolC = do nodepool <- HashSet.fromList . map txRefHash <$> lift getMempool cachepool <- HashSet.fromList . map txRefHash <$> cacheGetMempool - txs <- mapM getit . HashSet.toList $ - mappend - (HashSet.difference nodepool cachepool) - (HashSet.difference cachepool nodepool) + txs <- + mapM getit . HashSet.toList $ + mappend + (HashSet.difference nodepool cachepool) + (HashSet.difference cachepool nodepool) unless (null txs) $ do - $(logDebugS) "Cache" "Importing mempool transactions" + $(logDebugS) "Cache" $ + "Importing " <> cs (show (length txs)) + <> " mempool transactions" importMultiTxC (rights txs) - startCooldown + forM_ (zip [(1 :: Int) ..] (lefts txs)) $ \(i, h) -> + $(logDebugS) "Cache" $ + "Ignoring cache mempool tx " <> cs (show i) <> "/" + <> cs (show (length txs)) <> ": " + <> txHashToHex h where getit th = lift (getTxData th) >>= \case - Nothing -> return (Left th) - Just tx -> return (Right tx) + Nothing -> return (Left th) + Just tx -> return (Right tx) cacheGetMempool :: MonadLoggerIO m => CacheX m [TxRef] cacheGetMempool = runRedis redisGetMempool @@ -953,6 +979,43 @@ cacheGetHead :: MonadLoggerIO m => CacheX m (Maybe BlockHash) cacheGetHead = runRedis redisGetHead +cachePrime :: + (StoreReadBase m, MonadUnliftIO m, MonadLoggerIO m) + => CacheX m (Maybe BlockHash) +cachePrime = cacheGetHead >>= \case + Nothing -> do + $(logDebugS) "Cache" "Cache has no best block set" + lift getBestBlock >>= \case + Nothing -> do + $(logDebugS) "Cache" "Best block not set yet" + return Nothing + Just newhead -> do + ch <- asks cacheChain + chbest <- chainGetBest ch + if headerHash (nodeHeader chbest) == newhead + then do + m <- withLock $ do + $(logInfoS) "Cache" "Priming cache" + mem <- lift getMempool + runRedis $ do + a <- redisAddToMempool mem + b <- redisSetHead newhead + return $ a >> b + $(logDebugS) "Cache" "Primed" + return newhead + case m of + Nothing -> do + $(logErrorS) "Cache" + "Could not get lock to prime cache" + return Nothing + Just x -> return (Just x) + else do + $(logDebugS) + "Cache" + "Not priming cache because not in sync" + return Nothing + Just cachehead -> return (Just cachehead) + cacheSetHead :: (MonadLoggerIO m, StoreReadBase m) => BlockHash -> CacheX m () cacheSetHead bh = do $(logDebugS) "Cache" $ "Cache head set to: " <> blockHashToHex bh @@ -1146,6 +1209,9 @@ cacheNewBlock :: MonadIO m => CacheWriter -> m () cacheNewBlock = send CacheNewBlock + +cacheNewTx :: MonadIO m => TxHash -> CacheWriter -> m () +cacheNewTx th = send (CacheNewTx th) cachePing :: MonadIO m => CacheWriter -> m () cachePing = send CachePing 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 @@ -27,7 +27,6 @@ , applyLimitsC , sortTxs , nub' - , microseconds ) where import Conduit (ConduitT, dropC, mapC, takeC) @@ -46,8 +45,6 @@ import Data.List (sortBy) import Data.Maybe (catMaybes, listToMaybe) import Data.Serialize (Serialize (..)) -import Data.Time.Clock.System (getSystemTime, systemNanoseconds, - systemSeconds) import Data.Word (Word32, Word64) import GHC.Generics (Generic) import Haskoin (Address, BlockHash, @@ -68,7 +65,6 @@ XPubSummary (..), XPubUnspent (..), nullBalance, toTransaction, zeroBalance) -import UnliftIO (MonadIO, liftIO) type DeriveAddr = XPubKey -> KeyIndex -> Address @@ -374,9 +370,3 @@ nub' :: (Eq a, Hashable a) => [a] -> [a] nub' = H.toList . H.fromList - -microseconds :: MonadIO m => m Integer -microseconds = - let f t = toInteger (systemSeconds t) * 1000000 - + toInteger (systemNanoseconds t) `div` 1000 - in liftIO $ f <$> getSystemTime 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 @@ -6,6 +6,14 @@ DatabaseReader (..) , DatabaseReaderT , withDatabaseReader + , addrTxCF + , addrOutCF + , txCF + , spenderCF + , unspentCF + , blockCF + , heightCF + , balanceCF ) where import Conduit (mapC, runConduit, sinkList, (.|)) @@ -16,10 +24,10 @@ import Data.List (sortBy) import Data.Maybe (fromMaybe) import Data.Word (Word32) -import Database.RocksDB (Config (..), DB, withDB, - withIter) -import Database.RocksDB.Query (insert, matching, matchingAsList, - matchingSkip, retrieve) +import Database.RocksDB (ColumnFamily, Config (..), + DB (..), withDBCF, withIterCF) +import Database.RocksDB.Query (insert, matching, matchingSkip, + retrieve, retrieveCF) import Haskoin (Address, BlockHash, BlockHeight, Network, OutPoint (..), TxHash) import Haskoin.Store.Common @@ -30,11 +38,10 @@ import Haskoin.Store.Database.Types (AddrOutKey (..), AddrTxKey (..), BalKey (..), BestKey (..), BlockKey (..), HeightKey (..), - MemKey (..), OldMemKey (..), - SpenderKey (..), TxKey (..), - UnspentKey (..), VersionKey (..), - toUnspent, valToBalance, - valToUnspent) + MemKey (..), SpenderKey (..), + TxKey (..), UnspentKey (..), + VersionKey (..), toUnspent, + valToBalance, valToUnspent) import UnliftIO (MonadIO, MonadUnliftIO, liftIO) type DatabaseReaderT = ReaderT DatabaseReader @@ -48,7 +55,7 @@ } dataVersion :: Word32 -dataVersion = 16 +dataVersion = 17 withDatabaseReader :: MonadUnliftIO m => Network @@ -58,7 +65,7 @@ -> DatabaseReaderT m a -> m a withDatabaseReader net igap gap dir f = - withDB dir def{createIfMissing = True, maxFiles = Just (-1)} $ \db -> do + withDBCF dir cfg columnFamilyConfig $ \db -> do let bdb = DatabaseReader { databaseHandle = db @@ -68,28 +75,58 @@ } initRocksDB bdb runReaderT f bdb + where + cfg = def{createIfMissing = True, maxFiles = Just (-1)} +columnFamilyConfig :: [(String, Config)] +columnFamilyConfig = + [ ("addr-tx", def{prefixLength = Just 22, bloomFilter = True}) + , ("addr-out", def{prefixLength = Just 22, bloomFilter = True}) + , ("tx", def{prefixLength = Just 33, bloomFilter = True}) + , ("spender", def{prefixLength = Just 33, bloomFilter = True}) + , ("unspent", def{prefixLength = Just 37, bloomFilter = True}) + , ("block", def{prefixLength = Just 33, bloomFilter = True}) + , ("height", def{prefixLength = Nothing, bloomFilter = True}) + , ("balance", def{prefixLength = Just 22, bloomFilter = True}) + ] + +addrTxCF :: DB -> ColumnFamily +addrTxCF = head . columnFamilies + +addrOutCF :: DB -> ColumnFamily +addrOutCF db = columnFamilies db !! 1 + +txCF :: DB -> ColumnFamily +txCF db = columnFamilies db !! 2 + +spenderCF :: DB -> ColumnFamily +spenderCF db = columnFamilies db !! 3 + +unspentCF :: DB -> ColumnFamily +unspentCF db = columnFamilies db !! 4 + +blockCF :: DB -> ColumnFamily +blockCF db = columnFamilies db !! 5 + +heightCF :: DB -> ColumnFamily +heightCF db = columnFamilies db !! 6 + +balanceCF :: DB -> ColumnFamily +balanceCF db = columnFamilies db !! 7 + initRocksDB :: MonadIO m => DatabaseReader -> m () -initRocksDB bdb@DatabaseReader{databaseHandle = db} = do +initRocksDB DatabaseReader{databaseHandle = db} = do e <- runExceptT $ retrieve db VersionKey >>= \case Just v | v == dataVersion -> return () - | v == 15 -> migrate15to16 bdb >> initRocksDB bdb | otherwise -> throwError "Incorrect RocksDB database version" Nothing -> setInitRocksDB db case e of Left s -> error s Right () -> return () -migrate15to16 :: MonadIO m => DatabaseReader -> m () -migrate15to16 DatabaseReader{databaseHandle = db} = do - xs <- liftIO $ matchingAsList db OldMemKeyS - let ys = map (\(OldMemKey t h, ()) -> (t, h)) xs - insert db MemKey ys - insert db VersionKey (16 :: Word32) - setInitRocksDB :: MonadIO m => DB -> m () setInitRocksDB db = insert db VersionKey dataVersion @@ -99,26 +136,26 @@ getBlocksAtHeightDB :: MonadIO m => BlockHeight -> DatabaseReader -> m [BlockHash] getBlocksAtHeightDB h DatabaseReader{databaseHandle = db} = - retrieve db (HeightKey h) >>= \case + retrieveCF db (heightCF db) (HeightKey h) >>= \case Nothing -> return [] Just ls -> return ls getDatabaseReader :: MonadIO m => BlockHash -> DatabaseReader -> m (Maybe BlockData) getDatabaseReader h DatabaseReader{databaseHandle = db} = - retrieve db (BlockKey h) + retrieveCF db (blockCF db) (BlockKey h) getTxDataDB :: MonadIO m => TxHash -> DatabaseReader -> m (Maybe TxData) getTxDataDB th DatabaseReader{databaseHandle = db} = - retrieve db (TxKey th) + retrieveCF db (txCF db) (TxKey th) getSpenderDB :: MonadIO m => OutPoint -> DatabaseReader -> m (Maybe Spender) getSpenderDB op DatabaseReader{databaseHandle = db} = - retrieve db $ SpenderKey op + retrieveCF db (spenderCF db) $ SpenderKey op getBalanceDB :: MonadIO m => Address -> DatabaseReader -> m (Maybe Balance) getBalanceDB a DatabaseReader{databaseHandle = db} = - fmap (valToBalance a) <$> retrieve db (BalKey a) + fmap (valToBalance a) <$> retrieveCF db (balanceCF db) (BalKey a) getMempoolDB :: MonadIO m => DatabaseReader -> m [TxRef] getMempoolDB DatabaseReader{databaseHandle = db} = @@ -144,7 +181,7 @@ -> DatabaseReader -> m [TxRef] getAddressTxsDB a limits bdb@DatabaseReader{databaseHandle = db} = - liftIO $ withIter db $ \it -> runConduit $ + liftIO $ withIterCF db (addrTxCF db) $ \it -> runConduit $ x it .| applyLimitsC limits .| mapC (uncurry f) .| sinkList where x it = @@ -165,7 +202,7 @@ getUnspentDB :: MonadIO m => OutPoint -> DatabaseReader -> m (Maybe Unspent) getUnspentDB p DatabaseReader{databaseHandle = db} = - fmap (valToUnspent p) <$> retrieve db (UnspentKey p) + fmap (valToUnspent p) <$> retrieveCF db (unspentCF db) (UnspentKey p) getAddressesUnspentsDB :: MonadIO m @@ -187,7 +224,7 @@ -> DatabaseReader -> m [Unspent] getAddressUnspentsDB a limits bdb@DatabaseReader{databaseHandle = db} = - liftIO $ withIter db $ \it -> runConduit $ + liftIO $ withIterCF db (unspentCF db) $ \it -> runConduit $ x it .| applyLimitsC limits .| mapC (uncurry toUnspent) .| sinkList where x it = case start limits of diff --git a/src/Haskoin/Store/Database/Types.hs b/src/Haskoin/Store/Database/Types.hs --- a/src/Haskoin/Store/Database/Types.hs +++ b/src/Haskoin/Store/Database/Types.hs @@ -10,7 +10,6 @@ , BalKey(..) , HeightKey(..) , MemKey(..) - , OldMemKey(..) , SpenderKey(..) , TxKey(..) , UnspentKey(..) @@ -42,9 +41,8 @@ OutPoint (..), TxHash, eitherToMaybe, scriptToAddressBS) import Haskoin.Store.Data (Balance (..), BlockData, BlockRef, - TxRef (..), Spender, TxData, - UnixTime, Unspent (..), getUnixTime, - putUnixTime) + Spender, TxData, TxRef (..), UnixTime, + Unspent (..)) -- | Database key for an address transaction. data AddrTxKey @@ -311,38 +309,6 @@ instance Key VersionKey instance KeyValue VersionKey Word32 - - --- | Old mempool transaction database key. -data OldMemKey - = OldMemKey - { memTime :: !UnixTime - , memKey :: !TxHash - } - | OldMemKeyT - { memTime :: !UnixTime - } - | OldMemKeyS - deriving (Show, Read, Eq, Ord, Generic, Hashable) - -instance Serialize OldMemKey where - -- 0x07 · UnixTime · TxHash - put (OldMemKey t h) = do - putWord8 0x07 - putUnixTime t - put h - -- 0x07 · UnixTime - put (OldMemKeyT t) = do - putWord8 0x07 - putUnixTime t - -- 0x07 - put OldMemKeyS = putWord8 0x07 - get = do - guard . (== 0x07) =<< getWord8 - OldMemKey <$> getUnixTime <*> get - -instance Key OldMemKey -instance KeyValue OldMemKey () data BalVal = BalVal { balValAmount :: !Word64 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,8 +17,9 @@ import qualified Data.HashMap.Strict as M import Data.List (sortOn) import Data.Ord (Down (..)) -import Database.RocksDB (BatchOp) -import Database.RocksDB.Query (deleteOp, insertOp, writeBatch) +import Database.RocksDB (BatchOp, DB) +import Database.RocksDB.Query (deleteOp, deleteOpCF, insertOp, + insertOpCF, writeBatch) import GHC.Generics (Generic) import Haskoin (Address, BlockHash, BlockHeight, Network, OutPoint (..), TxHash, @@ -160,92 +161,94 @@ (mem, best) <- runReaderT ((,) <$> getMempool <*> getBestBlock) bdb hm <- newTVarIO (emptyMemory net best mem) x <- R.runReaderT f Writer {getReader = bdb, getState = hm} - ops <- hashMapOps <$> readTVarIO hm + ops <- hashMapOps db <$> readTVarIO hm writeBatch db ops return x -hashMapOps :: Memory -> [BatchOp] -hashMapOps db = - bestBlockOp (hBest db) <> - blockHashOps (hBlock db) <> - blockHeightOps (hHeight db) <> - txOps (hTx db) <> - spenderOps (hSpender db) <> - balOps (hBalance db) <> - addrTxOps (hAddrTx db) <> - addrOutOps (hAddrOut db) <> - mempoolOp (hMempool db) <> - unspentOps (hUnspent db) +hashMapOps :: DB -> Memory -> [BatchOp] +hashMapOps db mem = + bestBlockOp db (hBest mem) <> + blockHashOps db (hBlock mem) <> + blockHeightOps db (hHeight mem) <> + txOps db (hTx mem) <> + spenderOps db (hSpender mem) <> + balOps db (hBalance mem) <> + addrTxOps db (hAddrTx mem) <> + addrOutOps db (hAddrOut mem) <> + mempoolOp db (hMempool mem) <> + unspentOps db (hUnspent mem) -bestBlockOp :: Maybe BlockHash -> [BatchOp] -bestBlockOp Nothing = [deleteOp BestKey] -bestBlockOp (Just b) = [insertOp BestKey b] +bestBlockOp :: DB -> Maybe BlockHash -> [BatchOp] +bestBlockOp _ Nothing = [deleteOp BestKey] +bestBlockOp _ (Just b) = [insertOp BestKey b] -blockHashOps :: HashMap BlockHash BlockData -> [BatchOp] -blockHashOps = map (uncurry f) . M.toList +blockHashOps :: DB -> HashMap BlockHash BlockData -> [BatchOp] +blockHashOps db = map (uncurry f) . M.toList where - f = insertOp . BlockKey + f = insertOpCF (blockCF db) . BlockKey -blockHeightOps :: HashMap BlockHeight [BlockHash] -> [BatchOp] -blockHeightOps = map (uncurry f) . M.toList +blockHeightOps :: DB -> HashMap BlockHeight [BlockHash] -> [BatchOp] +blockHeightOps db = map (uncurry f) . M.toList where - f = insertOp . HeightKey + f = insertOpCF (heightCF db) . HeightKey -txOps :: HashMap TxHash TxData -> [BatchOp] -txOps = map (uncurry f) . M.toList +txOps :: DB -> HashMap TxHash TxData -> [BatchOp] +txOps db = map (uncurry f) . M.toList where - f = insertOp . TxKey + f = insertOpCF (txCF db) . TxKey -spenderOps :: HashMap OutPoint (Dirty Spender) - -> [BatchOp] -spenderOps = map (uncurry f) . M.toList +spenderOps :: DB -> HashMap OutPoint (Dirty Spender) -> [BatchOp] +spenderOps db = map (uncurry f) . M.toList where f o (Modified s) = - insertOp (SpenderKey o) s + insertOpCF (spenderCF db) (SpenderKey o) s f o Deleted = - deleteOp (SpenderKey o) + deleteOpCF (spenderCF db) (SpenderKey o) -balOps :: HashMap Address BalVal -> [BatchOp] -balOps = map (uncurry f) . M.toList +balOps :: DB -> HashMap Address BalVal -> [BatchOp] +balOps db = map (uncurry f) . M.toList where - f = insertOp . BalKey + f = insertOpCF (balanceCF db) . BalKey -addrTxOps :: HashMap (Address, TxRef) (Dirty ()) -> [BatchOp] -addrTxOps = map (uncurry f) . M.toList +addrTxOps :: DB -> HashMap (Address, TxRef) (Dirty ()) -> [BatchOp] +addrTxOps db = map (uncurry f) . M.toList where - f (a, t) (Modified ()) = insertOp (AddrTxKey a t) () - f (a, t) Deleted = deleteOp (AddrTxKey a t) + f (a, t) (Modified ()) = insertOpCF (addrTxCF db) (AddrTxKey a t) () + f (a, t) Deleted = deleteOpCF (addrTxCF db) (AddrTxKey a t) -addrOutOps - :: HashMap (Address, BlockRef, OutPoint) (Dirty OutVal) -> [BatchOp] -addrOutOps = map (uncurry f) . M.toList +addrOutOps :: DB + -> HashMap (Address, BlockRef, OutPoint) (Dirty OutVal) + -> [BatchOp] +addrOutOps db = map (uncurry f) . M.toList where f (a, b, p) (Modified l) = - insertOp - (AddrOutKey { addrOutKeyA = a - , addrOutKeyB = b - , addrOutKeyP = p }) - l + insertOpCF (addrOutCF db) + ( AddrOutKey { addrOutKeyA = a + , addrOutKeyB = b + , addrOutKeyP = p + } + ) l f (a, b, p) Deleted = - deleteOp AddrOutKey { addrOutKeyA = a - , addrOutKeyB = b - , addrOutKeyP = p } + deleteOpCF (addrOutCF db) + AddrOutKey { addrOutKeyA = a + , addrOutKeyB = b + , addrOutKeyP = p + } -mempoolOp :: HashMap TxHash UnixTime -> [BatchOp] -mempoolOp = (: []) - . insertOp MemKey - . sortOn Down - . map (\(h, t) -> (t, h)) - . M.toList +mempoolOp :: DB -> HashMap TxHash UnixTime -> [BatchOp] +mempoolOp _ = (: []) + . insertOp MemKey + . sortOn Down + . map (\(h, t) -> (t, h)) + . M.toList -unspentOps :: HashMap OutPoint (Dirty UnspentVal) - -> [BatchOp] -unspentOps = map (uncurry f) . M.toList +unspentOps :: DB -> HashMap OutPoint (Dirty UnspentVal) -> [BatchOp] +unspentOps db = map (uncurry f) . M.toList where f p (Modified u) = - insertOp (UnspentKey p) u + insertOpCF (unspentCF db) (UnspentKey p) u f p Deleted = - deleteOp (UnspentKey p) + deleteOpCF (unspentCF db) (UnspentKey p) getNetworkI :: MonadIO m => Writer -> m Network getNetworkI Writer {getState = hm} = 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 @@ -290,7 +290,7 @@ $(logErrorS) "BlockStore" $ "Attempted to import a tx missing UTXO: " <> txHashToHex (txHash tx) - -- throwError Orphan + throwError Orphan let us' = catMaybes us td = prepareTxData rbf br tt us' tx commitAddTx us' td diff --git a/src/Haskoin/Store/Manager.hs b/src/Haskoin/Store/Manager.hs --- a/src/Haskoin/Store/Manager.hs +++ b/src/Haskoin/Store/Manager.hs @@ -35,8 +35,9 @@ blockStoreTxHashSTM, blockStoreTxSTM, withBlockStore) import Haskoin.Store.Cache (CacheConfig (..), CacheWriter, - cacheNewBlock, cachePing, - cacheWriter, connectRedis) + cacheNewBlock, cacheNewTx, + cachePing, cacheWriter, + connectRedis) import Haskoin.Store.Common (StoreEvent (..)) import Haskoin.Store.Database.Reader (DatabaseReader (..), DatabaseReaderT, @@ -149,6 +150,7 @@ NodeConfig { nodeConfMaxPeers = storeConfMaxPeers cfg , nodeConfDB = databaseHandle db + , nodeConfColumnFamily = Nothing , nodeConfPeers = storeConfInitPeers cfg , nodeConfDiscover = storeConfDiscover cfg , nodeConfEvents = pub @@ -182,13 +184,12 @@ where f conn cwinbox = runReaderT (cacheWriter (c conn) cwinbox) db - c conn = - CacheConfig - { cacheConn = conn - , cacheMin = storeConfCacheMin cfg - , cacheChain = chain - , cacheMax = storeConfMaxKeys cfg - } + c conn = CacheConfig + { cacheConn = conn + , cacheMin = storeConfCacheMin cfg + , cacheChain = chain + , cacheMax = storeConfMaxKeys cfg + } cacheWriterProcesses :: MonadUnliftIO m => Inbox StoreEvent @@ -202,9 +203,10 @@ where events = cacheWriterEvents evts cwm ping = forever $ do - time <- liftIO $ randomRIO (600000, 1200000) + time <- liftIO $ randomRIO (300 * second, 600 * second) threadDelay time cachePing cwm + second = 1000000 cacheWriterEvents :: MonadIO m => Inbox StoreEvent -> CacheWriter -> m () cacheWriterEvents evts cwm = @@ -213,8 +215,9 @@ e `cacheWriterDispatch` cwm cacheWriterDispatch :: MonadIO m => StoreEvent -> CacheWriter -> m () -cacheWriterDispatch (StoreBestBlock _) = cacheNewBlock -cacheWriterDispatch _ = const (return ()) +cacheWriterDispatch (StoreBestBlock _) = cacheNewBlock +cacheWriterDispatch (StoreMempoolNew th) = cacheNewTx th +cacheWriterDispatch _ = const (return ()) nodeForwarder :: MonadIO m => BlockStore