diff --git a/app/Main.hs b/app/Main.hs --- a/app/Main.hs +++ b/app/Main.hs @@ -46,57 +46,61 @@ #endif data Config = Config - { configDir :: !FilePath - , configHost :: !String - , configPort :: !Int - , configNetwork :: !Network - , configAsert :: !Word32 - , configDiscover :: !Bool - , configPeers :: ![(String, Maybe Int)] - , configVersion :: !Bool - , configDebug :: !Bool - , configMaxPending :: !Int - , configWebLimits :: !WebLimits - , configWebTimeouts :: !WebTimeouts - , configRedis :: !Bool - , configRedisURL :: !String - , configRedisMin :: !Int - , configRedisMax :: !Integer - , configWipeMempool :: !Bool - , configNoMempool :: !Bool - , configPeerTimeout :: !Int - , configPeerMaxLife :: !Int - , configMaxPeers :: !Int - , configMaxDiff :: !Int - , configCacheRefresh :: !Int - , configNumTxId :: !Bool + { configDir :: !FilePath + , configHost :: !String + , configPort :: !Int + , configNetwork :: !Network + , configAsert :: !Word32 + , configDiscover :: !Bool + , configPeers :: ![(String, Maybe Int)] + , configVersion :: !Bool + , configDebug :: !Bool + , configMaxPending :: !Int + , configWebLimits :: !WebLimits + , configWebTimeouts :: !WebTimeouts + , configRedis :: !Bool + , configRedisURL :: !String + , configRedisMin :: !Int + , configRedisMax :: !Integer + , configWipeMempool :: !Bool + , configNoMempool :: !Bool + , configPeerTimeout :: !Int + , configPeerMaxLife :: !Int + , configMaxPeers :: !Int + , configMaxDiff :: !Int + , configCacheRefresh :: !Int + , configCacheRetries :: !Int + , configCacheRetryDelay :: Int + , configNumTxId :: !Bool } instance Default Config where - def = Config { configDir = defDirectory - , configHost = defHost - , configPort = defPort - , configNetwork = defNetwork - , configAsert = defAsert - , configDiscover = defDiscover - , configPeers = defPeers - , configVersion = False - , configDebug = defDebug - , configMaxPending = defMaxPending - , configWebLimits = defWebLimits - , configWebTimeouts = defWebTimeouts - , configRedis = defRedis - , configRedisURL = defRedisURL - , configRedisMin = defRedisMin - , configRedisMax = defRedisMax - , configWipeMempool = defWipeMempool - , configNoMempool = defNoMempool - , configPeerTimeout = defPeerTimeout - , configPeerMaxLife = defPeerMaxLife - , configMaxPeers = defMaxPeers - , configMaxDiff = defMaxDiff - , configCacheRefresh = defCacheRefresh - , configNumTxId = defNumTxId + def = Config { configDir = defDirectory + , configHost = defHost + , configPort = defPort + , configNetwork = defNetwork + , configAsert = defAsert + , configDiscover = defDiscover + , configPeers = defPeers + , configVersion = False + , configDebug = defDebug + , configMaxPending = defMaxPending + , configWebLimits = defWebLimits + , configWebTimeouts = defWebTimeouts + , configRedis = defRedis + , configRedisURL = defRedisURL + , configRedisMin = defRedisMin + , configRedisMax = defRedisMax + , configWipeMempool = defWipeMempool + , configNoMempool = defNoMempool + , configPeerTimeout = defPeerTimeout + , configPeerMaxLife = defPeerMaxLife + , configMaxPeers = defMaxPeers + , configMaxDiff = defMaxDiff + , configCacheRefresh = defCacheRefresh + , configCacheRetries = defCacheRetries + , configCacheRetryDelay = defCacheRetryDelay + , configNumTxId = defNumTxId } defEnv :: MonadIO m => String -> a -> (String -> Maybe a) -> m a @@ -109,6 +113,16 @@ defEnv "CACHE_REFRESH" 750 readMaybe {-# NOINLINE defCacheRefresh #-} +defCacheRetries :: Int +defCacheRetries = unsafePerformIO $ + defEnv "CACHE_RETRIES" 100 readMaybe +{-# NOINLINE defCacheRetries #-} + +defCacheRetryDelay :: Int +defCacheRetryDelay = unsafePerformIO $ + defEnv "CACHE_RETRY_DELAY" 100000 readMaybe +{-# NOINLINE defCacheRetryDelay #-} + defMaxPending :: Int defMaxPending = unsafePerformIO $ defEnv "MAX_PENDING_TXS" 100 readMaybe @@ -425,6 +439,20 @@ <> help "Refresh cache this frequently" <> showDefault <> value (configCacheRefresh def) + configCacheRetries <- + option auto $ + metavar "INT" + <> long "cache-retries" + <> help "Retry getting cache lock to index xpub" + <> showDefault + <> value (configCacheRetries def) + configCacheRetryDelay <- + option auto $ + metavar "MICROSECONDS" + <> long "cache-retry-delay" + <> help "Delay to retry getting cache lock to index xpub" + <> showDefault + <> value (configCacheRetryDelay def) configNoMempool <- flag (configNoMempool def) True $ long "no-mempool" @@ -438,6 +466,7 @@ metavar "INT" <> long "max-diff" <> help "Maximum difference between headers and blocks" + <> showDefault <> value (configMaxDiff def) configNumTxId <- flag (configNumTxId def) True $ @@ -514,6 +543,8 @@ , configMaxDiff = maxdiff , configNoMempool = nomem , configCacheRefresh = crefresh + , configCacheRetries = cretries + , configCacheRetryDelay = cretrydelay , configNumTxId = numtxid } = runStderrLoggingT . filterLogger l $ do @@ -542,6 +573,8 @@ , storeConfPeerMaxLife = fromIntegral peerlife , storeConfConnect = withConnection , storeConfCacheRefresh = crefresh + , storeConfCacheRetries = cretries + , storeConfCacheRetryDelay = cretrydelay } 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.40.17 +version: 0.40.18 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 @@ -77,8 +77,8 @@ send) import System.Random (randomIO, randomRIO) import UnliftIO (Exception, MonadIO, MonadUnliftIO, - atomically, bracket, liftIO, link, - throwIO, withAsync) + async, atomically, bracket, liftIO, + link, throwIO, withAsync) import UnliftIO.Concurrent (threadDelay) runRedis :: MonadLoggerIO m => Redis (Either Reply a) -> CacheX m a @@ -91,11 +91,13 @@ throwIO (RedisError e) data CacheConfig = CacheConfig - { cacheConn :: !Connection - , cacheMin :: !Int - , cacheMax :: !Integer - , cacheChain :: !Chain - , cacheRefresh :: !Int + { cacheConn :: !Connection + , cacheMin :: !Int + , cacheMax :: !Integer + , cacheChain :: !Chain + , cacheRefresh :: !Int -- millisenconds + , cacheRetries :: !Int + , cacheRetryDelay :: !Int -- microseconds } type CacheT = ReaderT (Maybe CacheConfig) @@ -172,12 +174,9 @@ txs <- cacheGetXPubTxs xpub limits return txs False -> do - newXPubC xpub >>= \(t, bals) -> - if t - then do - cacheGetXPubTxs xpub limits - else do - lift $ xPubBalsTxs bals limits + bals <- lift $ xPubBals xpub + newXPubC xpub bals + lift $ xPubBalsTxs bals limits getXPubUnspents :: (MonadUnliftIO m, MonadLoggerIO m, StoreReadExtra m) @@ -188,13 +187,10 @@ True -> do bals <- cacheGetXPubBalances xpub process bals - False -> - newXPubC xpub >>= \(t, bals) -> - if t - then do - process bals - else do - lift $ xPubBalsUnspents bals limits + False -> do + bals <- lift $ xPubBals xpub + newXPubC xpub bals + lift $ xPubBalsUnspents bals limits where process bals = do ops <- map snd <$> cacheGetXPubUnspents xpub limits @@ -228,7 +224,9 @@ bals <- cacheGetXPubBalances xpub return bals False -> do - snd <$> newXPubC xpub + bals <- lift $ xPubBals xpub + newXPubC xpub bals + return bals isInCache :: MonadLoggerIO m => XPubSpec -> CacheT m Bool isInCache xpub = @@ -494,8 +492,12 @@ Just _ -> Just <$> f Nothing -> return Nothing -smallDelay :: MonadUnliftIO m => m () -smallDelay = threadDelay =<< liftIO (randomRIO (50 * 1000, 250 * 1000)) +smallDelay :: MonadUnliftIO m => CacheX m () +smallDelay = do + delay <- asks cacheRetryDelay + let delayMin = delay `div` 2 + let delayMax = delay * 3 `div` 2 + threadDelay =<< liftIO (randomRIO (delayMin, delayMax)) withLockForever :: (MonadLoggerIO m, MonadUnliftIO m) @@ -511,7 +513,7 @@ withLockRetry :: (MonadLoggerIO m, MonadUnliftIO m) - => Integer + => Int -> CacheX m a -> CacheX m (Maybe a) withLockRetry i f @@ -567,43 +569,42 @@ 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 inSync >>= \s -> if s then go bals else return (False, bals) - else return (False, bals) + => XPubSpec -> [XPubBal] -> CacheX m () +newXPubC xpub bals = + should_index >>= \i -> when i $ + asks cacheRetries >>= \r -> + void . async . withLockRetry r $ + isXPubCached xpub >>= \c -> when (not c) index where op XPubUnspent {xPubUnspent = u} = (unspentPoint u, unspentBlock u) - go bals = do - m <- withLockRetry 10 $ do - xpubtxt <- xpubText xpub - $(logDebugS) "Cache" $ - "Caching " <> xpubtxt <> ": " <> cs (show (length bals)) <> - " addresses / " <> cs (show (lenNotNull bals)) <> - " used" - utxo <- lift $ xPubUnspents xpub def - $(logDebugS) "Cache" $ - "Caching " <> xpubtxt <> ": " <> cs (show (length utxo)) <> - " utxos" - xtxs <- lift $ xPubTxs xpub def - $(logDebugS) "Cache" $ - "Caching " <> xpubtxt <> ": " <> cs (show (length xtxs)) <> - " txs" - now <- systemSeconds <$> liftIO getSystemTime - runRedis $ do - b <- redisTouchKeys now [xpub] - c <- redisAddXPubBalances xpub bals - d <- redisAddXPubUnspents xpub (map op utxo) - e <- redisAddXPubTxs xpub xtxs - return $ b >> c >> d >> e >> return () - $(logDebugS) "Cache" $ "Cached " <> xpubtxt - case m of - Nothing -> return (False, bals) - Just () -> return (True, bals) + should_index = do + x <- asks cacheMin + if x <= lenNotNull bals + then inSync + else return False + index = do + bals <- lift (xPubBals xpub) + xpubtxt <- xpubText xpub + $(logDebugS) "Cache" $ + "Caching " <> xpubtxt <> ": " <> cs (show (length bals)) <> + " addresses / " <> cs (show (lenNotNull bals)) <> + " used" + utxo <- lift $ xPubUnspents xpub def + $(logDebugS) "Cache" $ + "Caching " <> xpubtxt <> ": " <> cs (show (length utxo)) <> + " utxos" + xtxs <- lift $ xPubTxs xpub def + $(logDebugS) "Cache" $ + "Caching " <> xpubtxt <> ": " <> cs (show (length xtxs)) <> + " txs" + now <- systemSeconds <$> liftIO getSystemTime + runRedis $ do + b <- redisTouchKeys now [xpub] + c <- redisAddXPubBalances xpub bals + d <- redisAddXPubUnspents xpub (map op utxo) + e <- redisAddXPubTxs xpub xtxs + return $ b >> c >> d >> e >> return () + $(logDebugS) "Cache" $ "Cached " <> xpubtxt inSync :: (MonadUnliftIO m, MonadLoggerIO m, StoreReadExtra m) => CacheX m Bool @@ -969,8 +970,7 @@ nodepool <- HashSet.fromList . map snd <$> lift getMempool cachepool <- HashSet.fromList . map snd <$> cacheGetMempool getem (HashSet.difference nodepool cachepool) - withCool "prune" (refresh * 90 `div` 10) $ - getem (HashSet.difference cachepool nodepool) + getem (HashSet.difference cachepool nodepool) where getem tset = do let tids = HashSet.toList tset 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 @@ -65,38 +65,42 @@ -- | Configuration for a 'Store'. data StoreConfig = StoreConfig - { storeConfMaxPeers :: !Int + { storeConfMaxPeers :: !Int -- ^ max peers to connect to - , storeConfInitPeers :: ![HostPort] + , storeConfInitPeers :: ![HostPort] -- ^ static set of peers to connect to - , storeConfDiscover :: !Bool + , storeConfDiscover :: !Bool -- ^ discover new peers - , storeConfDB :: !FilePath + , storeConfDB :: !FilePath -- ^ RocksDB database path - , storeConfNetwork :: !Network + , storeConfNetwork :: !Network -- ^ network constants - , storeConfCache :: !(Maybe String) + , storeConfCache :: !(Maybe String) -- ^ Redis cache configuration - , storeConfInitialGap :: !Word32 + , storeConfInitialGap :: !Word32 -- ^ gap on extended public key with no transactions - , storeConfGap :: !Word32 + , storeConfGap :: !Word32 -- ^ gap for extended public keys - , storeConfCacheMin :: !Int + , storeConfCacheMin :: !Int -- ^ cache xpubs with more than this many used addresses - , storeConfMaxKeys :: !Integer + , storeConfMaxKeys :: !Integer -- ^ maximum number of keys in Redis cache - , storeConfNoMempool :: !Bool + , storeConfNoMempool :: !Bool -- ^ do not index new mempool transactions - , storeConfWipeMempool :: !Bool + , storeConfWipeMempool :: !Bool -- ^ wipe mempool when starting - , storeConfPeerTimeout :: !NominalDiffTime + , storeConfPeerTimeout :: !NominalDiffTime -- ^ disconnect peer if message not received for this many seconds - , storeConfPeerMaxLife :: !NominalDiffTime + , storeConfPeerMaxLife :: !NominalDiffTime -- ^ disconnect peer if it has been connected this long - , storeConfConnect :: !(SockAddr -> WithConnection) + , storeConfConnect :: !(SockAddr -> WithConnection) -- ^ connect to peers using the function 'withConnection' - , storeConfCacheRefresh :: !Int + , storeConfCacheRefresh :: !Int -- ^ refresh the cache this often (milliseconds) + , storeConfCacheRetries :: !Int + -- ^ retry count for getting cache lock to index xpub + , storeConfCacheRetryDelay :: !Int + -- ^ delay in microseconds to retry getting cache lock } withStore :: (MonadLoggerIO m, MonadUnliftIO m) @@ -194,6 +198,8 @@ , cacheChain = chain , cacheMax = storeConfMaxKeys cfg , cacheRefresh = storeConfCacheRefresh cfg + , cacheRetries = storeConfCacheRetries cfg + , cacheRetryDelay = storeConfCacheRetryDelay cfg } cacheWriterProcesses :: MonadUnliftIO m diff --git a/test/Haskoin/StoreSpec.hs b/test/Haskoin/StoreSpec.hs --- a/test/Haskoin/StoreSpec.hs +++ b/test/Haskoin/StoreSpec.hs @@ -97,6 +97,8 @@ , storeConfPeerMaxLife = 48 * 3600 , storeConfConnect = dummyPeerConnect net ad , storeConfCacheRefresh = 750 + , storeConfCacheRetries = 100 + , storeConfCacheRetryDelay = 100000 } withStore cfg $ \Store {..} -> withSubscription storePublisher $ \sub ->