haskoin-store 0.65.8 → 0.65.9

raw patch · 6 files changed

+101/−55 lines, 6 filesPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

API changes (from Hackage documentation)

Files

• CHANGELOG.md +5/−0
• app/Main.hs +21/−4
• haskoin-store.cabal +1/−1
• src/Haskoin/Store/Cache.hs +56/−39
• src/Haskoin/Store/Manager.hs +16/−10
• test/Haskoin/StoreSpec.hs +2/−1

@@ -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.

@@ -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

@@ -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 <https://github.com/haskoin/haskoin-store#readme> category:       Bitcoin, Finance, Network

@@ -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

@@ -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

@@ -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 ->