diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,11 @@ 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) +## Unreleased +### Changed +- Be more aggressive caching individual transactions. +- Introduce a periodic cache mempool sync task. + ## 0.65.8 ### Changed - Perform health check in separate thread. diff --git a/app/Main.hs b/app/Main.hs --- a/app/Main.hs +++ b/app/Main.hs @@ -120,7 +120,8 @@ configWebHistoryURL :: !String, configWebNoBlockchain :: !Bool, configWebNoSlow :: !Bool, - configHealthCheckInterval :: !Int + configHealthCheckInterval :: !Int, + configCacheMempoolSync :: !Int } instance Default Config where @@ -157,7 +158,8 @@ configWebHistoryURL = defWebHistoryURL, configWebNoBlockchain = defWebNoBlockchain, configWebNoSlow = defWebNoSlow, - configHealthCheckInterval = defHealthCheckInterval + configHealthCheckInterval = defHealthCheckInterval, + configCacheMempoolSync = defCacheMempoolSync } defEnv :: MonadIO m => String -> a -> (String -> Maybe a) -> m a @@ -382,6 +384,12 @@ defEnv "HEALTH_CHECK_INTERVAL" 30 readMaybe {-# NOINLINE defHealthCheckInterval #-} +defCacheMempoolSync :: Int +defCacheMempoolSync = + unsafePerformIO $ + defEnv "CACHE_MEMPOOL_SYNC" 30 readMaybe +{-# NOINLINE defCacheMempoolSync #-} + netNames :: String netNames = intercalate "|" (map getNetworkName allNets) @@ -652,6 +660,13 @@ <> help "Background check update interval" <> showDefault <> value (configHealthCheckInterval def) + configCacheMempoolSync <- + option auto $ + metavar "SECONDS" + <> long "cache-mempool-sync" + <> help "Sync mempool to cache interval" + <> showDefault + <> value (configCacheMempoolSync def) pure Config { configWebLimits = WebLimits {..}, @@ -724,7 +739,8 @@ configWebHistoryURL = whurl, configWebNoSlow = no_slow, configWebNoBlockchain = no_blockchain, - configHealthCheckInterval = health_check_interval + configHealthCheckInterval = health_check_interval, + configCacheMempoolSync = cachesync } = runStderrLoggingT . filterLogger l . with_stats $ \stats -> do net <- case networkReader net_str of @@ -754,7 +770,8 @@ storeConfPeerTimeout = fromIntegral peertimeout, storeConfPeerMaxLife = fromIntegral peerlife, storeConfConnect = withConnection, - storeConfStats = stats + storeConfStats = stats, + storeConfCacheMempoolSync = cachesync } withStore scfg $ \st -> runWeb diff --git a/haskoin-store.cabal b/haskoin-store.cabal --- a/haskoin-store.cabal +++ b/haskoin-store.cabal @@ -5,7 +5,7 @@ -- see: https://github.com/sol/hpack name: haskoin-store -version: 0.65.8 +version: 0.65.9 synopsis: Storage and index for Bitcoin and Bitcoin Cash description: Please see the README on GitHub at category: Bitcoin, Finance, Network 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 @@ -4,6 +4,7 @@ {-# LANGUAGE DeriveGeneric #-} {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE FlexibleInstances #-} +{-# LANGUAGE ImportQualifiedPost #-} {-# LANGUAGE LambdaCase #-} {-# LANGUAGE MultiWayIf #-} {-# LANGUAGE OverloadedStrings #-} @@ -25,6 +26,7 @@ CacheWriterInbox, cacheNewBlock, cacheNewTx, + cacheSyncMempool, cacheWriter, cacheDelXPubs, isInCache, @@ -45,17 +47,17 @@ import Control.Monad.Trans.Maybe (MaybeT (..), runMaybeT) import Data.Bits (complement, shift, (.&.), (.|.)) import Data.ByteString (ByteString) -import qualified Data.ByteString as B +import Data.ByteString qualified as B import Data.Default (def) import Data.Either (fromRight, isRight, rights) import Data.Functor ((<&>)) import Data.HashMap.Strict (HashMap) -import qualified Data.HashMap.Strict as HashMap +import Data.HashMap.Strict qualified as HashMap import Data.HashSet (HashSet) -import qualified Data.HashSet as HashSet -import qualified Data.IntMap.Strict as I +import Data.HashSet qualified as HashSet +import Data.IntMap.Strict qualified as I import Data.List (sort) -import qualified Data.Map.Strict as Map +import Data.Map.Strict qualified as Map import Data.Maybe ( catMaybes, fromMaybe, @@ -87,8 +89,8 @@ zrangebyscoreWithscoresLimit, zrem, ) -import qualified Database.Redis as Redis -import qualified Database.Redis as Reids +import Database.Redis qualified as Redis +import Database.Redis qualified as Reids import GHC.Generics (Generic) import Haskoin ( Address, @@ -135,13 +137,13 @@ receive, send, ) -import qualified System.Metrics as Metrics -import qualified System.Metrics.Counter as Metrics (Counter) -import qualified System.Metrics.Counter as Metrics.Counter -import qualified System.Metrics.Distribution as Metrics (Distribution) -import qualified System.Metrics.Distribution as Metrics.Distribution -import qualified System.Metrics.Gauge as Metrics (Gauge) -import qualified System.Metrics.Gauge as Metrics.Gauge +import System.Metrics qualified as Metrics +import System.Metrics.Counter qualified as Metrics (Counter) +import System.Metrics.Counter qualified as Metrics.Counter +import System.Metrics.Distribution qualified as Metrics (Distribution) +import System.Metrics.Distribution qualified as Metrics.Distribution +import System.Metrics.Gauge qualified as Metrics (Gauge) +import System.Metrics.Gauge qualified as Metrics.Gauge import System.Random (randomIO, randomRIO) import UnliftIO ( Exception, @@ -659,7 +661,8 @@ data CacheWriterMessage = CacheNewBlock - | CacheNewTx TxHash + | CacheNewTx !TxHash + | CacheSyncMempool !(Listen ()) type CacheWriterInbox = Inbox CacheWriterMessage @@ -707,8 +710,7 @@ lockIt = do go >>= \case Right Redis.Ok -> do - $(logDebugS) "Cache" $ - "Acquired lock" + $(logDebugS) "Cache" "Acquired lock" incrementCounter cacheLockAcquired 1 return True Right Redis.Pong -> do @@ -764,8 +766,13 @@ CacheX m (Maybe a) withLock f = bracket lockIt unlockIt $ \case - True -> Just <$> f + True -> Just <$> go False -> return Nothing + where + go = withAsync refresh $ const f + refresh = forever $ do + threadDelay (150 * 1000 * 1000) + refreshLock isFull :: (MonadUnliftIO m, MonadLoggerIO m, StoreReadBase m) => @@ -808,17 +815,14 @@ (MonadUnliftIO m, MonadLoggerIO m, StoreReadExtra m) => CacheWriterMessage -> CacheX m () -cacheWriterReact CacheNewBlock = - doSync -cacheWriterReact (CacheNewTx txid) = - doSync - -doSync :: - (MonadUnliftIO m, MonadLoggerIO m, StoreReadExtra m) => - CacheX m () -doSync = do +cacheWriterReact CacheNewBlock = do newBlockC syncMempoolC +cacheWriterReact (CacheNewTx txid) = + syncNewTxC [txid] +cacheWriterReact (CacheSyncMempool l) = do + syncMempoolC + atomically $ l () lenNotNull :: [XPubBal] -> Int lenNotNull = length . filter (not . nullBalance . xPubBal) @@ -1231,6 +1235,19 @@ Nothing -> [] Just bals -> addrsToAdd gap bals a +syncNewTxC :: + (MonadUnliftIO m, MonadLoggerIO m, StoreReadExtra m) => + [TxHash] -> + CacheX m () +syncNewTxC ths = + inSync >>= \s -> when s . void . withLock $ do + txs <- catMaybes <$> mapM (lift . getTxData) ths + unless (null txs) $ do + forM_ txs $ \tx -> + $(logDebugS) "Cache" $ + "Synchronizing transaction: " <> txHashToHex (txHash (txData tx)) + importMultiTxC txs + syncMempoolC :: (MonadUnliftIO m, MonadLoggerIO m, StoreReadExtra m) => CacheX m () @@ -1238,18 +1255,15 @@ inSync >>= \s -> when s . void . withLock $ do nodepool <- HashSet.fromList . map snd <$> lift getMempool cachepool <- HashSet.fromList . map snd <$> cacheGetMempool - getem (HashSet.difference nodepool cachepool) - refreshLock - getem (HashSet.difference cachepool nodepool) - refreshLock - where - getem tset = do - let tids = HashSet.toList tset - txs <- catMaybes <$> mapM (lift . getTxData) tids - unless (null txs) $ do - $(logDebugS) "Cache" $ - "Importing mempool transactions: " <> cs (show (length txs)) - importMultiTxC txs + let diff1 = HashSet.difference nodepool cachepool + let diff2 = HashSet.difference cachepool nodepool + let diffset = diff1 <> diff2 + let tids = HashSet.toList diffset + txs <- catMaybes <$> mapM (lift . getTxData) tids + unless (null txs) $ do + $(logDebugS) "Cache" $ + "Synchronizing " <> cs (show (length txs)) <> " mempool transactions" + importMultiTxC txs cacheGetMempool :: MonadLoggerIO m => CacheX m [(UnixTime, TxHash)] cacheGetMempool = runRedis redisGetMempool @@ -1479,3 +1493,6 @@ cacheNewTx :: MonadIO m => TxHash -> CacheWriter -> m () cacheNewTx = send . CacheNewTx + +cacheSyncMempool :: MonadIO m => CacheWriter -> m () +cacheSyncMempool = query CacheSyncMempool 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 @@ -59,7 +59,7 @@ cacheNewTx, cacheWriter, connectRedis, - newCacheMetrics, + newCacheMetrics, cacheSyncMempool, ) import Haskoin.Store.Common ( StoreEvent (..), @@ -138,7 +138,9 @@ -- | connect to peers using the function 'withConnection' storeConfConnect :: !(SockAddr -> WithConnection), -- | stats store - storeConfStats :: !(Maybe Metrics.Store) + storeConfStats :: !(Maybe Metrics.Store), + -- | sync mempool against cache every this many seconds + storeConfCacheMempoolSync :: !Int } withStore :: @@ -240,9 +242,10 @@ withSubscription pub $ \evts -> let conf = c conn metrics in withProcess (f conf) $ \p -> - cacheWriterProcesses evts (getProcessMailbox p) $ do + cacheWriterProcesses interval evts (getProcessMailbox p) $ do action (Just conf) where + interval = storeConfCacheMempoolSync cfg f conf cwinbox = runReaderT (cacheWriter conf cwinbox) db c conn metrics = CacheConfig @@ -255,20 +258,23 @@ cacheWriterProcesses :: MonadUnliftIO m => + Int -> Inbox StoreEvent -> CacheWriter -> m a -> m a -cacheWriterProcesses evts cwm action = - withAsync events $ \a1 -> link a1 >> action - where - events = cacheWriterEvents evts cwm +cacheWriterProcesses interval evts cwm action = + withAsync (cacheWriterEvents interval evts cwm) $ \a1 -> link a1 >> action -cacheWriterEvents :: MonadIO m => Inbox StoreEvent -> CacheWriter -> m () -cacheWriterEvents evts cwm = - forever $ +cacheWriterEvents :: MonadUnliftIO m => Int -> Inbox StoreEvent -> CacheWriter -> m () +cacheWriterEvents interval evts cwm = + withAsync mempool . const $ forever $ receive evts >>= \e -> e `cacheWriterDispatch` cwm + where + mempool = forever $ do + threadDelay (interval * 1000 * 1000) + cacheSyncMempool cwm cacheWriterDispatch :: MonadIO m => StoreEvent -> CacheWriter -> m () cacheWriterDispatch (StoreBestBlock _) = cacheNewBlock diff --git a/test/Haskoin/StoreSpec.hs b/test/Haskoin/StoreSpec.hs --- a/test/Haskoin/StoreSpec.hs +++ b/test/Haskoin/StoreSpec.hs @@ -99,7 +99,8 @@ storeConfPeerTimeout = 60, storeConfPeerMaxLife = 48 * 3600, storeConfConnect = dummyPeerConnect net ad, - storeConfStats = Nothing + storeConfStats = Nothing, + storeConfCacheMempoolSync = 30 } withStore cfg $ \Store {..} -> withSubscription storePublisher $ \sub ->