haskoin-store 0.36.4 → 0.37.0
raw patch · 8 files changed
+449/−384 lines, 8 filesdep ~haskoin-storedep ~haskoin-store-datadep ~rocksdb-haskell-jprupp
Dependency ranges changed: haskoin-store, haskoin-store-data, rocksdb-haskell-jprupp
Files
- haskoin-store.cabal +8/−8
- src/Haskoin/Store/Cache.hs +294/−228
- src/Haskoin/Store/Common.hs +0/−10
- src/Haskoin/Store/Database/Reader.hs +65/−28
- src/Haskoin/Store/Database/Types.hs +2/−36
- src/Haskoin/Store/Database/Writer.hs +64/−61
- src/Haskoin/Store/Logic.hs +1/−1
- src/Haskoin/Store/Manager.hs +15/−12
haskoin-store.cabal view
@@ -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 <https://github.com/haskoin/haskoin-store#readme> 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
src/Haskoin/Store/Cache.hs view
@@ -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
src/Haskoin/Store/Common.hs view
@@ -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
src/Haskoin/Store/Database/Reader.hs view
@@ -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
src/Haskoin/Store/Database/Types.hs view
@@ -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
src/Haskoin/Store/Database/Writer.hs view
@@ -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} =
src/Haskoin/Store/Logic.hs view
@@ -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
src/Haskoin/Store/Manager.hs view
@@ -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