diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,17 @@ 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) +## 0.64.0 +### Changed +- Improve metrics. + +### Fixed +- Get lock before importing transactions to cache. + +## 0.63.0 +### Fixed +- Make cache updates lightweight. + ## 0.62.1 ### Fixed - Fix idempotent transaction elimination bug. diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -27,10 +27,8 @@ ## Install on Ubuntu or Debian -* Get [Stack](https://haskellstack.org/) - ```sh -apt install git zlib1g-dev libsecp256k1-dev librocksdb-dev pkg-config +apt install git zlib1g-dev libsecp256k1-dev librocksdb-dev pkg-config haskell-stack git clone https://github.com/haskoin/haskoin-store.git cd haskoin-store stack build --copy-bins diff --git a/app/Main.hs b/app/Main.hs --- a/app/Main.hs +++ b/app/Main.hs @@ -73,7 +73,6 @@ , configPeerMaxLife :: !Int , configMaxPeers :: !Int , configMaxDiff :: !Int - , configCacheRefresh :: !Int , configCacheRetryDelay :: !Int , configStatsd :: !Bool , configStatsdHost :: !String @@ -107,7 +106,6 @@ , configPeerMaxLife = defPeerMaxLife , configMaxPeers = defMaxPeers , configMaxDiff = defMaxDiff - , configCacheRefresh = defCacheRefresh , configCacheRetryDelay = defCacheRetryDelay , configStatsd = defStatsd , configStatsdHost = defStatsdHost @@ -123,11 +121,6 @@ ms <- lookupEnv e return $ fromMaybe d $ p =<< ms -defCacheRefresh :: Int -defCacheRefresh = unsafePerformIO $ - defEnv "CACHE_REFRESH" 750 readMaybe -{-# NOINLINE defCacheRefresh #-} - defCacheRetryDelay :: Int defCacheRetryDelay = unsafePerformIO $ defEnv "CACHE_RETRY_DELAY" 100000 readMaybe @@ -486,13 +479,6 @@ <> help "Maximum number of keys in Redis xpub cache" <> showDefault <> value (configRedisMax def) - configCacheRefresh <- - option auto $ - metavar "MILLISECONDS" - <> long "cache-refresh" - <> help "Refresh cache this frequently" - <> showDefault - <> value (configCacheRefresh def) configCacheRetryDelay <- option auto $ metavar "MICROSECONDS" @@ -636,7 +622,6 @@ , configMaxDiff = maxdiff , configNoMempool = nomem , configSyncMempool = syncmem - , configCacheRefresh = crefresh , configCacheRetryDelay = cretrydelay , configStatsd = statsd , configStatsdHost = statsdhost @@ -675,7 +660,6 @@ , storeConfPeerTimeout = fromIntegral peertimeout , storeConfPeerMaxLife = fromIntegral peerlife , storeConfConnect = withConnection - , storeConfCacheRefresh = crefresh , storeConfCacheRetryDelay = cretrydelay , storeConfStats = stats } 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.62.1 +version: 0.64.0 synopsis: Storage and index for Bitcoin and Bitcoin Cash description: Please see the README on GitHub at category: Bitcoin, Finance, Network @@ -60,7 +60,7 @@ , hashable >=1.3.0.0 , haskoin-core >=0.21.1 , haskoin-node >=0.17.0 - , haskoin-store-data ==0.62.1 + , haskoin-store-data ==0.64.0 , hedis >=0.12.13 , http-types >=0.12.3 , lens >=4.18.1 @@ -116,7 +116,7 @@ , haskoin-core >=0.21.1 , haskoin-node >=0.17.0 , haskoin-store - , haskoin-store-data ==0.62.1 + , haskoin-store-data ==0.64.0 , hedis >=0.12.13 , http-types >=0.12.3 , lens >=4.18.1 @@ -177,7 +177,7 @@ , haskoin-core >=0.21.1 , haskoin-node >=0.17.0 , haskoin-store - , haskoin-store-data ==0.62.1 + , haskoin-store-data ==0.64.0 , hedis >=0.12.13 , hspec >=2.7.1 , http-types >=0.12.3 diff --git a/src/Haskoin/Store/BlockStore.hs b/src/Haskoin/Store/BlockStore.hs --- a/src/Haskoin/Store/BlockStore.hs +++ b/src/Haskoin/Store/BlockStore.hs @@ -277,6 +277,16 @@ runRocksDB . getAddressTxs a getNumTxData = runRocksDB . getNumTxData + getBalances = + runRocksDB . getBalances + xPubBals = + runRocksDB . xPubBals + xPubUnspents x l = + runRocksDB . xPubUnspents x l + xPubTxs x l = + runRocksDB . xPubTxs x l + xPubTxCount x = + runRocksDB . xPubTxCount x -- | Run block store process. withBlockStore :: 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 @@ -22,6 +22,7 @@ , CacheWriter , CacheWriterInbox , cacheNewBlock + , cacheNewTx , cacheWriter , cacheDelXPubs , isInCache @@ -47,8 +48,8 @@ import qualified Data.IntMap.Strict as I import Data.List (sort) import qualified Data.Map.Strict as Map -import Data.Maybe (catMaybes, fromMaybe, isNothing, - mapMaybe) +import Data.Maybe (catMaybes, fromMaybe, isJust, + isNothing, mapMaybe) import Data.Serialize (Serialize, decode, encode) import Data.String.Conversions (cs) import Data.Text (Text) @@ -73,7 +74,7 @@ blockHashToHex, derivePubPath, eitherToMaybe, headerHash, pathToList, scriptToAddressBS, - txHash, xPubAddr, + txHash, txHashToHex, xPubAddr, xPubCompatWitnessAddr, xPubExport, xPubWitnessAddr) import Haskoin.Node (Chain, chainBlockMain, @@ -115,7 +116,6 @@ , cacheMin :: !Int , cacheMax :: !Integer , cacheChain :: !Chain - , cacheRefresh :: !Int -- millisenconds , cacheRetryDelay :: !Int -- microseconds , cacheMetrics :: !(Maybe CacheMetrics) } @@ -123,32 +123,34 @@ data CacheMetrics = CacheMetrics { cacheHits :: !Metrics.Counter , cacheMisses :: !Metrics.Counter - , cacheIgnore :: !Metrics.Counter - , cacheRefreshes :: !Metrics.Counter , cacheLockAcquired :: !Metrics.Counter , cacheLockReleased :: !Metrics.Counter , cacheLockFailed :: !Metrics.Counter + , cacheXPubBals :: !Metrics.Counter + , cacheXPubUnspents :: !Metrics.Counter + , cacheXPubTxs :: !Metrics.Counter + , cacheXPubTxCount :: !Metrics.Counter , cacheIndexTime :: !StatDist - , cacheMempoolSyncTime :: !StatDist , cacheBlockSyncTime :: !StatDist } newCacheMetrics :: MonadIO m => Metrics.Store -> m CacheMetrics newCacheMetrics s = liftIO $ do - cacheHits <- c "hits" - cacheMisses <- c "misses" - cacheIgnore <- c "ignore" - cacheRefreshes <- c "refreshes" - cacheLockAcquired <- c "lock_acquired" - cacheLockReleased <- c "lock_released" - cacheLockFailed <- c "lock_failed" - cacheIndexTime <- d "index" - cacheMempoolSyncTime <- d "mempool_sync" - cacheBlockSyncTime <- d "block_sync" + cacheHits <- c "cache.hits" + cacheMisses <- c "cache.misses" + cacheLockAcquired <- c "cache.lock_acquired" + cacheLockReleased <- c "cache.lock_released" + cacheLockFailed <- c "cache.lock_failed" + cacheIndexTime <- d "cache.index" + cacheBlockSyncTime <- d "cache.block_sync" + cacheXPubBals <- c "cache.xpub_bals" + cacheXPubUnspents <- c "cache.xpub_unspents" + cacheXPubTxs <- c "cache.xpub_txs" + cacheXPubTxCount <- c "cache.xpub_tx_count" return CacheMetrics{..} where - c x = Metrics.createCounter ("cache." <> x) s - d x = createStatDist ("cache." <> x) s + c x = Metrics.createCounter x s + d x = createStatDist x s withMetrics :: MonadUnliftIO m => (CacheMetrics -> StatDist) @@ -171,11 +173,12 @@ incrementCounter :: MonadIO m => (CacheMetrics -> Metrics.Counter) + -> Int -> CacheX m () -incrementCounter cf = +incrementCounter f i = asks cacheMetrics >>= \case + Just s -> liftIO $ Metrics.Counter.add (f s) (fromIntegral i) Nothing -> return () - Just m -> liftIO $ Metrics.Counter.inc (cf m) type CacheT = ReaderT (Maybe CacheConfig) type CacheX = ReaderT CacheConfig @@ -264,8 +267,9 @@ where go m = isXPubCached xpub >>= \case True -> do - unless m $ incrementCounter cacheHits - cacheGetXPubTxs xpub limits + txs <- cacheGetXPubTxs xpub limits + incrementCounter cacheXPubTxs (length txs) + return txs False -> case m of True -> lift $ xPubTxs xpub xbals limits @@ -280,7 +284,7 @@ where go t = isXPubCached xpub >>= \case True -> do - unless t $ incrementCounter cacheHits + incrementCounter cacheXPubTxCount 1 cacheGetXPubTxCount xpub False -> if t @@ -299,11 +303,11 @@ in HashMap.fromList $ map f $ filter g xbals go m = isXPubCached xpub >>= \case True -> do - unless m $ incrementCounter cacheHits process False -> case m of - True -> - lift $ xPubUnspents xpub xbals limits + True -> do + us <- lift $ xPubUnspents xpub xbals limits + return us False -> do newXPubC xpub xbals go True @@ -323,14 +327,16 @@ } us = mapMaybe f uns i a u = h u <$> g a + incrementCounter cacheXPubUnspents (length us) return $ mapMaybe (uncurry i) us getXPubBalances :: (MonadUnliftIO m, MonadLoggerIO m, StoreReadExtra m) => XPubSpec -> CacheX m [XPubBal] getXPubBalances xpub = isXPubCached xpub >>= \case True -> do - incrementCounter cacheHits - cacheGetXPubBalances xpub + xbals <- cacheGetXPubBalances xpub + incrementCounter cacheXPubBals (length xbals) + return xbals False -> do bals <- lift $ xPubBals xpub newXPubC xpub bals @@ -343,7 +349,12 @@ Just cfg -> runReaderT (isXPubCached xpub) cfg isXPubCached :: MonadLoggerIO m => XPubSpec -> CacheX m Bool -isXPubCached = runRedis . redisIsXPubCached +isXPubCached xpub = do + cached <- runRedis (redisIsXPubCached xpub) + if cached + then incrementCounter cacheHits 1 + else incrementCounter cacheMisses 1 + return cached redisIsXPubCached :: RedisCtx m f => XPubSpec -> m (f Bool) redisIsXPubCached xpub = Redis.exists (balancesPfx <> encode xpub) @@ -544,7 +555,7 @@ data CacheWriterMessage = CacheNewBlock - | CachePing !(Listen ()) + | CacheNewTx TxHash type CacheWriterInbox = Inbox CacheWriterMessage type CacheWriter = Mailbox CacheWriterMessage @@ -576,14 +587,11 @@ (MonadUnliftIO m, MonadLoggerIO m, StoreReadExtra m) => CacheConfig -> CacheWriterInbox -> m () cacheWriter cfg inbox = - withAsync ping $ \a -> - link a >> runReaderT go cfg + runReaderT go cfg where - ping = forever $ do - threadDelay (cacheRefresh cfg * 1000) - cachePing (inboxToMailbox inbox) go = do newBlockC + syncMempoolC forever $ do x <- receive inbox cacheWriterReact x @@ -595,26 +603,26 @@ Right Redis.Ok -> do $(logDebugS) "Cache" $ "Acquired lock with value " <> cs (show rnd) - incrementCounter cacheLockAcquired + incrementCounter cacheLockAcquired 1 return (Just rnd) Right Redis.Pong -> do $(logErrorS) "Cache" "Unexpected pong when acquiring lock" - incrementCounter cacheLockFailed + incrementCounter cacheLockFailed 1 return Nothing Right (Redis.Status s) -> do $(logErrorS) "Cache" $ "Unexpected status acquiring lock: " <> cs s - incrementCounter cacheLockFailed + incrementCounter cacheLockFailed 1 return Nothing Left (Redis.Bulk Nothing) -> do $(logDebugS) "Cache" "Lock already taken" - incrementCounter cacheLockFailed + incrementCounter cacheLockFailed 1 return Nothing Left e -> do $(logErrorS) "Cache" "Error when trying to acquire lock" - incrementCounter cacheLockFailed + incrementCounter cacheLockFailed 1 throwIO (RedisError e) where go rnd = do @@ -644,7 +652,7 @@ $(logDebugS) "Cache" $ "Released lock with value " <> cs (show i) - incrementCounter cacheLockReleased + incrementCounter cacheLockReleased 1 else $(logErrorS) "Cache" $ "Could not release lock: value is not " <> @@ -726,16 +734,22 @@ => CacheWriterMessage -> CacheX m () cacheWriterReact CacheNewBlock = doSync -cacheWriterReact (CachePing respond) = - doSync >> atomically (respond ()) +cacheWriterReact (CacheNewTx txid) = go + where + f = cacheIsInMempool txid >>= \x -> + unless x $ + lift (getTxData txid) >>= \mtx -> + forM_ mtx $ \tx -> do + $(logDebugS) "Cache" $ + "Importing mempool transaction: " <> txHashToHex txid + importMultiTxC [tx] + go = withLock f >>= \case + Just () -> return () + Nothing -> smallDelay >> go doSync :: (MonadUnliftIO m, MonadLoggerIO m, StoreReadExtra m) => CacheX m () -doSync = - inSync >>= \s -> when s $ do - newBlockC - syncMempoolC - void pruneDB +doSync = newBlockC >> void pruneDB lenNotNull :: [XPubBal] -> Int lenNotNull = length . filter (not . nullBalance . xPubBal) @@ -744,11 +758,7 @@ (MonadUnliftIO m, MonadLoggerIO m, StoreReadExtra m) => XPubSpec -> [XPubBal] -> CacheX m () newXPubC xpub xbals = should_index >>= \i -> - if i - then do - incrementCounter cacheMisses - withMetrics cacheIndexTime index - else incrementCounter cacheIgnore + when i $ withMetrics cacheIndexTime index where op XPubUnspent {xPubUnspent = u} = (unspentPoint u, unspentBlock u) should_index = @@ -803,80 +813,71 @@ newBlockC :: (MonadUnliftIO m, MonadLoggerIO m, StoreReadExtra m) => CacheX m () newBlockC = - inSync >>= \s -> when s $ do - lift getBestBlock >>= \case - Nothing -> $(logWarnS) "Cache" "No best block" - Just bb -> - asks cacheChain >>= \ch -> - chainGetBest ch >>= \cn -> - cacheGetHead >>= \case - Nothing -> do - $(logInfoS) "Cache" "Initializing best cache block" - void $ importBlockC bb - Just hb -> - if hb == headerHash (nodeHeader cn) - then $(logDebugS) "Cache" "Cache in sync" - else sync ch hb bb + inSync >>= \s -> when s $ + asks cacheChain >>= \ch -> + chainGetBest ch >>= \bn -> + cacheGetHead >>= \case + Nothing -> + $(logInfoS) "Cache" "Initializing best cache block" >> + withLock (do_import bn) >>= \case + Nothing -> smallDelay >> newBlockC + Just () -> return () + Just hb -> + if hb == headerHash (nodeHeader bn) + then $(logDebugS) "Cache" "Cache in sync" + else + withLock (sync ch hb bn) >>= \case + Nothing -> smallDelay >> newBlockC + Just () -> return () where - sync ch hb bb = + sync ch hb bn = chainGetBlock hb ch >>= \case - Nothing -> + Nothing -> do $(logErrorS) "Cache" $ - "Cache head block node not found: " <> - blockHashToHex hb + "Cache head block node not found: " <> + blockHashToHex hb + throwIO $ LogicError $ + "Cache head block node not found: " <> + cs (blockHashToHex hb) Just hn -> chainBlockMain hb ch >>= \m -> if m - then - chainGetBlock bb ch >>= \case - Just bn -> next ch bn hn - Nothing -> do - $(logErrorS) "Cache" $ - "Cache head node not found: " - <> blockHashToHex bb - throwIO $ - LogicError $ - "Cache head node not found: " - <> cs (blockHashToHex bb) + then next ch bn hn else do $(logDebugS) "Cache" $ "Reverting cache head not in main chain: " <> blockHashToHex hb - removeHeadC hb >>= \s -> when s newBlockC + removeHeadC hb + cacheGetHead >>= \case + Nothing -> do_import bn + Just hb' -> sync ch hb' bn next ch bn hn = if | prevBlock (nodeHeader bn) == headerHash (nodeHeader hn) -> - void $ importBlockC (headerHash (nodeHeader bn)) + do_import bn | nodeHeight bn > nodeHeight hn -> chainGetAncestor (nodeHeight hn + 1) bn ch >>= \case - Nothing -> - $(logWarnS) "Cache" $ - "Ancestor not found at height " - <> cs (show (nodeHeight hn + 1)) - <> " for block: " - <> blockHashToHex (headerHash (nodeHeader bn)) - Just hn' -> - importBlockC (headerHash (nodeHeader hn')) >>= \s -> - when s newBlockC + Nothing -> do + $(logErrorS) "Cache" $ + "Ancestor not found at height " + <> cs (show (nodeHeight hn + 1)) + <> " for block: " + <> blockHashToHex (headerHash (nodeHeader bn)) + throwIO $ LogicError $ + "Ancestor not found at height " + <> show (nodeHeight hn + 1) + <> " for block: " + <> cs (blockHashToHex (headerHash (nodeHeader bn))) + Just hn' -> do + do_import hn' + next ch bn hn' | otherwise -> $(logInfoS) "Cache" "Cache best block higher than this node's" + do_import = importBlockC . headerHash . nodeHeader importBlockC :: (MonadUnliftIO m, StoreReadExtra m, MonadLoggerIO m) - => BlockHash -> CacheX m Bool + => BlockHash -> CacheX m () importBlockC bh = - fmap (maybe False (const True)) $ withLock $ - withMetrics cacheBlockSyncTime $ - cacheGetHead >>= \case - Nothing -> go - Just cb -> - asks cacheChain >>= \ch -> - chainGetBlock bh ch >>= \case - Nothing -> return () - Just bn -> - if prevBlock (nodeHeader bn) == cb - then go - else return () - where - go = lift (getBlock bh) >>= \case + lift (getBlock bh) >>= \case Just bd -> do let ths = blockDataTxs bd tds <- sortTxData . catMaybes <$> mapM (lift . getTxData) ths @@ -899,9 +900,8 @@ <> blockHashToHex bh removeHeadC :: (StoreReadExtra m, MonadUnliftIO m, MonadLoggerIO m) - => BlockHash -> CacheX m Bool + => BlockHash -> CacheX m () removeHeadC cb = - fmap (maybe False (const True)) $ withLock $ void . runMaybeT $ do bh <- MaybeT cacheGetHead guard (cb == bh) @@ -1121,42 +1121,13 @@ Nothing -> [] Just bals -> addrsToAdd gap bals a -withCool :: (MonadLoggerIO m) - => ByteString - -> Integer - -> CacheX m a - -> CacheX m (Maybe a) -withCool key milliseconds run = - is_cool >>= \c -> - if c - then do - $(logDebugS) "Cache" $ "Cooldown reached for key: " <> cs (show key) - x <- run - cooldown - return (Just x) - else do - $(logDebugS) "Cache" $ "Cooldown NOT reached for key: " <> cs (show key) - return Nothing - where - is_cool = isNothing <$> runRedis (Redis.get key) - cooldown = - let opts = Redis.SetOpts - { Redis.setSeconds = Nothing - , Redis.setMilliseconds = Just milliseconds - , Redis.setCondition = Just Redis.Nx } - in void . runRedis $ Redis.setOpts key "0" opts - syncMempoolC :: (MonadUnliftIO m, MonadLoggerIO m, StoreReadExtra m) => CacheX m () -syncMempoolC = - toInteger <$> asks cacheRefresh >>= \refresh -> - void . withLock . withCool "cool" (refresh * 9 `div` 10) $ - withMetrics cacheMempoolSyncTime $ do +syncMempoolC = void . withLockForever $ do nodepool <- HashSet.fromList . map snd <$> lift getMempool cachepool <- HashSet.fromList . map snd <$> cacheGetMempool getem (HashSet.difference nodepool cachepool) getem (HashSet.difference cachepool nodepool) - incrementCounter cacheRefreshes where getem tset = do let tids = HashSet.toList tset @@ -1166,10 +1137,12 @@ "Importing mempool transactions: " <> cs (show (length txs)) importMultiTxC txs - cacheGetMempool :: MonadLoggerIO m => CacheX m [(UnixTime, TxHash)] cacheGetMempool = runRedis redisGetMempool +cacheIsInMempool :: MonadLoggerIO m => TxHash -> CacheX m Bool +cacheIsInMempool = runRedis . redisIsInMempool + cacheGetHead :: MonadLoggerIO m => CacheX m (Maybe BlockHash) cacheGetHead = runRedis redisGetHead @@ -1189,6 +1162,10 @@ map (\btx -> (blockRefScore (txRefBlock btx), encode (txRefHash btx))) btxs +redisIsInMempool :: (Applicative f, RedisCtx m f) => TxHash -> m (f Bool) +redisIsInMempool txid = + fmap isJust <$> Redis.zrank mempoolSetKey (encode txid) + redisRemFromMempool :: (Applicative f, RedisCtx m f) => [TxHash] -> m (f Integer) redisRemFromMempool [] = return (pure 0) @@ -1203,7 +1180,7 @@ -> CacheT m Integer cacheDelXPubs xpubs = ReaderT $ \case Just cache -> runReaderT (delXPubKeys xpubs) cache - Nothing -> return 0 + Nothing -> return 0 delXPubKeys :: (MonadUnliftIO m, MonadLoggerIO m, StoreReadBase m) @@ -1375,5 +1352,5 @@ cacheNewBlock :: MonadIO m => CacheWriter -> m () cacheNewBlock = send CacheNewBlock -cachePing :: MonadIO m => CacheWriter -> m () -cachePing = query CachePing +cacheNewTx :: MonadIO m => TxHash -> CacheWriter -> m () +cacheNewTx = send . CacheNewTx 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 @@ -14,6 +14,7 @@ , StoreWrite(..) , StoreEvent(..) , PubExcept(..) + , DataMetrics(..) , getActiveBlock , getActiveTxData , getDefaultBalance @@ -23,6 +24,9 @@ , blockAtOrAfter , blockAtOrBefore , blockAtOrAfterMTP + , xPubSummary + , deriveAddresses + , deriveFunction , deOffset , applyLimits , applyLimitsC @@ -33,6 +37,7 @@ , microseconds , streamThings , joinDescStreams + , createDataMetrics ) where import Conduit (ConduitT, await, dropC, mapC, @@ -79,6 +84,9 @@ XPubSummary (..), XPubUnspent (..), nullBalance, toTransaction, zeroBalance) +import qualified System.Metrics as Metrics +import System.Metrics.Counter (Counter) +import qualified System.Metrics.Counter as Counter import UnliftIO (MonadIO, liftIO) type DeriveAddr = XPubKey -> KeyIndex -> Address @@ -115,113 +123,20 @@ getUnspent :: OutPoint -> m (Maybe Unspent) getMempool :: m [(UnixTime, TxHash)] - countBlocks :: Int -> m () - countBlocks _ = return () - - countTxs :: Int -> m () - countTxs _ = return () - - countBalances :: Int -> m () - countBalances _ = return () - - countUnspents :: Int -> m () - countUnspents _ = return () - class StoreReadBase m => StoreReadExtra m where getAddressesTxs :: [Address] -> Limits -> m [TxRef] getAddressesUnspents :: [Address] -> Limits -> m [Unspent] getInitialGap :: m Word32 getMaxGap :: m Word32 getNumTxData :: Word64 -> m [TxData] - getBalances :: [Address] -> m [Balance] - getBalances as = - zipWith f as <$> mapM getBalance as - where - f a Nothing = zeroBalance a - f _ (Just b) = b - getAddressTxs :: Address -> Limits -> m [TxRef] - getAddressTxs a = getAddressesTxs [a] - getAddressUnspents :: Address -> Limits -> m [Unspent] - getAddressUnspents a = getAddressesUnspents [a] - xPubBals :: XPubSpec -> m [XPubBal] - xPubBals xpub = do - igap <- getInitialGap - gap <- getMaxGap - ext1 <- derive_until_gap gap 0 (take (fromIntegral igap) (aderiv 0 0)) - if all (nullBalance . xPubBal) ext1 - then return [] - else do - ext2 <- derive_until_gap gap 0 (aderiv 0 igap) - chg <- derive_until_gap gap 1 (aderiv 1 0) - return (ext1 <> ext2 <> chg) - where - aderiv m = - deriveAddresses - (deriveFunction (xPubDeriveType xpub)) - (pubSubKey (xPubSpecKey xpub) m) - xbalance m b n = XPubBal {xPubBalPath = [m, n], xPubBal = b} - derive_until_gap _ _ [] = return [] - derive_until_gap gap m as = do - let (as1, as2) = splitAt (fromIntegral gap) as - countXPubDerivations (length as1) - bs <- getBalances (map snd as1) - let xbs = zipWith (xbalance m) bs (map fst as1) - if all nullBalance bs - then return xbs - else (xbs <>) <$> derive_until_gap gap m as2 - - xPubSummary :: XPubSpec -> [XPubBal] -> m XPubSummary - xPubSummary _xspec xbals = return - XPubSummary - { xPubSummaryConfirmed = sum (map (balanceAmount . xPubBal) bs) - , xPubSummaryZero = sum (map (balanceZero . xPubBal) bs) - , xPubSummaryReceived = rx - , xPubUnspentCount = uc - , xPubChangeIndex = ch - , xPubExternalIndex = ex - } - where - bs = filter (not . nullBalance . xPubBal) xbals - ex = foldl max 0 [i | XPubBal {xPubBalPath = [0, i]} <- bs] - ch = foldl max 0 [i | XPubBal {xPubBalPath = [1, i]} <- bs] - uc = sum [balanceUnspentCount (xPubBal b) | b <- bs] - xt = [b | b@XPubBal {xPubBalPath = [0, _]} <- bs] - rx = sum [balanceTotalReceived (xPubBal b) | b <- xt] - xPubUnspents :: XPubSpec -> [XPubBal] -> Limits -> m [XPubUnspent] - xPubUnspents _xspec xbals limits = - applyLimits limits . sortOn Down . concat <$> mapM h cs - where - l = deOffset limits - cs = filter ((> 0) . balanceUnspentCount . xPubBal) xbals - i b = do - us <- getAddressUnspents (balanceAddress (xPubBal b)) l - countUnspents (length us) - return us - f b t = XPubUnspent {xPubUnspentPath = xPubBalPath b, xPubUnspent = t} - h b = map (f b) <$> i b - xPubTxs :: XPubSpec -> [XPubBal] -> Limits -> m [TxRef] - xPubTxs _xspec xbals limits = - let as = map balanceAddress $ - filter (not . nullBalance) $ - map xPubBal xbals - in getAddressesTxs as limits - xPubTxCount :: XPubSpec -> [XPubBal] -> m Word32 - xPubTxCount xspec xbals = - fromIntegral . length <$> xPubTxs xspec xbals def - countTxRefs :: Int -> m () - countTxRefs _ = return () - - countXPubDerivations :: Int -> m () - countXPubDerivations _ = return () - class StoreWrite m where setBest :: BlockHash -> m () insertBlock :: BlockData -> m () @@ -271,6 +186,24 @@ deriveFunction DeriveP2SH i = fst . deriveCompatWitnessAddr i deriveFunction DeriveP2WPKH i = fst . deriveWitnessAddr i +xPubSummary :: XPubSpec -> [XPubBal] -> XPubSummary +xPubSummary _xspec xbals = + XPubSummary + { xPubSummaryConfirmed = sum (map (balanceAmount . xPubBal) bs) + , xPubSummaryZero = sum (map (balanceZero . xPubBal) bs) + , xPubSummaryReceived = rx + , xPubUnspentCount = uc + , xPubChangeIndex = ch + , xPubExternalIndex = ex + } + where + bs = filter (not . nullBalance . xPubBal) xbals + ex = foldl max 0 [i | XPubBal {xPubBalPath = [0, i]} <- bs] + ch = foldl max 0 [i | XPubBal {xPubBalPath = [1, i]} <- bs] + uc = sum [balanceUnspentCount (xPubBal b) | b <- bs] + xt = [b | b@XPubBal {xPubBalPath = [0, _]} <- bs] + rx = sum [balanceTotalReceived (xPubBal b) | b <- xt] + getTransaction :: (Monad m, StoreReadBase m) => TxHash -> m (Maybe Transaction) getTransaction h = runMaybeT $ do @@ -332,7 +265,7 @@ | StorePeerConnected !Peer | StorePeerDisconnected !Peer | StorePeerPong !Peer !Word64 - | StoreTxAvailable !Peer ![TxHash] + | StoreTxAnnounce !Peer ![TxHash] | StoreTxReject !Peer !TxHash !RejectCode !ByteString data PubExcept = PubNoPeers @@ -449,3 +382,35 @@ let mp2 = Map.deleteMax mp mp' = Map.unionWith (++) mp1 mp2 go (Just x) mp' + +data DataMetrics = + DataMetrics + { dataBestCount :: !Counter + , dataBlockCount :: !Counter + , dataTxCount :: !Counter + , dataSpenderCount :: !Counter + , dataMempoolCount :: !Counter + , dataBalanceCount :: !Counter + , dataUnspentCount :: !Counter + , dataAddrTxCount :: !Counter + , dataXPubBals :: !Counter + , dataXPubUnspents :: !Counter + , dataXPubTxs :: !Counter + , dataXPubTxCount :: !Counter + } + +createDataMetrics :: MonadIO m => Metrics.Store -> m DataMetrics +createDataMetrics s = liftIO $ do + dataBestCount <- Metrics.createCounter "data.best_block" s + dataBlockCount <- Metrics.createCounter "data.blocks" s + dataTxCount <- Metrics.createCounter "data.txs" s + dataSpenderCount <- Metrics.createCounter "data.spenders" s + dataMempoolCount <- Metrics.createCounter "data.mempool" s + dataBalanceCount <- Metrics.createCounter "data.balances" s + dataUnspentCount <- Metrics.createCounter "data.unspents" s + dataAddrTxCount <- Metrics.createCounter "data.address_txs" s + dataXPubBals <- Metrics.createCounter "data.xpub_balances" s + dataXPubUnspents <- Metrics.createCounter "data.xpub_unspents" s + dataXPubTxs <- Metrics.createCounter "data.xpub_txs" s + dataXPubTxCount <- Metrics.createCounter "data.xpub_tx_count" s + return DataMetrics{..} 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 @@ -7,8 +7,6 @@ ( -- * RocksDB Database Access DatabaseReader (..) , DatabaseReaderT - , DatabaseStats(..) - , createDatabaseStats , withDatabaseReader , addrTxCF , addrOutCF @@ -41,7 +39,7 @@ retrieve, retrieveCF) import Haskoin (Address, BlockHash, BlockHeight, Network, OutPoint (..), TxHash, - txHash) + pubSubKey, txHash) import Haskoin.Store.Common import Haskoin.Store.Data import Haskoin.Store.Database.Types @@ -52,42 +50,21 @@ type DatabaseReaderT = ReaderT DatabaseReader -data DatabaseStats = - DatabaseStats - { databaseBlockCount :: !Counter - , databaseTxCount :: !Counter - , databaseBalanceCount :: !Counter - , databaseUnspentCount :: !Counter - , databaseTxRefCount :: !Counter - , databaseDerivations :: !Counter - } - data DatabaseReader = DatabaseReader { databaseHandle :: !DB , databaseMaxGap :: !Word32 , databaseInitialGap :: !Word32 , databaseNetwork :: !Network - , databaseStats :: !(Maybe DatabaseStats) + , databaseMetrics :: !(Maybe DataMetrics) } -createDatabaseStats :: MonadIO m => Metrics.Store -> m DatabaseStats -createDatabaseStats s = liftIO $ do - databaseBlockCount <- Metrics.createCounter "database.blocks" s - databaseTxCount <- Metrics.createCounter "database.txs" s - databaseBalanceCount <- Metrics.createCounter "database.balances" s - databaseUnspentCount <- Metrics.createCounter "database.unspents" s - databaseTxRefCount <- Metrics.createCounter "database.txrefs" s - databaseDerivations <- Metrics.createCounter "database.derivations" s - return DatabaseStats{..} - incrementCounter :: MonadIO m - => (DatabaseStats -> Counter) + => (DataMetrics -> Counter) -> Int -> ReaderT DatabaseReader m () -incrementCounter f i = do - stats <- asks databaseStats - case stats of +incrementCounter f i = + asks databaseMetrics >>= \case Just s -> liftIO $ Counter.add (f s) (fromIntegral i) Nothing -> return () @@ -99,7 +76,7 @@ -> Word32 -> Word32 -> FilePath - -> Maybe DatabaseStats + -> Maybe DataMetrics -> DatabaseReaderT m a -> m a withDatabaseReader net igap gap dir stats f = @@ -110,7 +87,7 @@ , databaseMaxGap = gap , databaseNetwork = net , databaseInitialGap = igap - , databaseStats = stats + , databaseMetrics = stats } initRocksDB bdb runReaderT f bdb @@ -169,92 +146,6 @@ setInitRocksDB :: MonadIO m => DB -> m () setInitRocksDB db = insert db VersionKey dataVersion -getBestDB :: MonadIO m - => DatabaseReaderT m (Maybe BlockHash) -getBestDB = - asks databaseHandle >>= (`retrieve` BestKey) - -getBlocksAtHeightDB :: MonadIO m - => BlockHeight - -> DatabaseReaderT m [BlockHash] -getBlocksAtHeightDB h = do - db <- asks databaseHandle - retrieveCF db (heightCF db) (HeightKey h) >>= \case - Nothing -> return [] - Just ls -> countBlocks (length ls) >> return ls - -getDatabaseReader :: MonadIO m - => BlockHash - -> DatabaseReaderT m (Maybe BlockData) -getDatabaseReader h = do - db <- asks databaseHandle - retrieveCF db (blockCF db) (BlockKey h) >>= \case - Nothing -> return Nothing - Just b -> countBlocks 1 >> return (Just b) - -getTxDataDB :: MonadIO m - => TxHash -> DatabaseReaderT m (Maybe TxData) -getTxDataDB th = do - db <- asks databaseHandle - retrieveCF db (txCF db) (TxKey th) >>= \case - Nothing -> return Nothing - Just t -> countTxs 1 >> return (Just t) - -getNumTxDataDB :: MonadIO m - => Word64 - -> DatabaseReaderT m [TxData] -getNumTxDataDB i = do - db <- asks databaseHandle - let (sk, w) = decodeTxKey i - ls <- liftIO $ matchingAsListCF db (txCF db) (TxKeyS sk) - let f t = let bs = encode $ txHash (txData t) - b = BS.head (BS.drop 6 bs) - w' = b .&. 0xf8 - in w == w' - txs = filter f $ map snd ls - countTxs (length txs) - return txs - -getSpenderDB :: MonadIO m - => OutPoint - -> DatabaseReaderT m (Maybe Spender) -getSpenderDB op = do - db <- asks databaseHandle - retrieveCF db (spenderCF db) $ SpenderKey op - -getBalanceDB :: MonadIO m - => Address - -> DatabaseReaderT m (Maybe Balance) -getBalanceDB a = do - db <- asks databaseHandle - fmap (valToBalance a) <$> retrieveCF db (balanceCF db) (BalKey a) >>= \case - Nothing -> return Nothing - Just b -> countBalances 1 >> return (Just b) - -getMempoolDB :: MonadIO m - => DatabaseReaderT m [(UnixTime, TxHash)] -getMempoolDB = do - db <- asks databaseHandle - fromMaybe [] <$> retrieve db MemKey - -getAddressesTxsDB :: MonadUnliftIO m - => [Address] - -> Limits - -> DatabaseReaderT m [TxRef] -getAddressesTxsDB addrs limits = do - txs <- applyLimits limits . sortOn Down . concat <$> mapM f addrs - countTxRefs (length txs) - return txs - where - l = deOffset limits - f a = do - db <- asks databaseHandle - withIterCF db (addrTxCF db) $ \it -> - runConduit $ - addressConduit a (start l) it .| - applyLimitC (limit l) .| - sinkList - addressConduit :: MonadUnliftIO m => Address -> Maybe Start @@ -274,7 +165,7 @@ (AddrTxKeyA a) (AddrTxKeyB a (BlockRef bh maxBound)) Just (AtTx txh) -> - lift (getTxDataDB txh) >>= \case + lift (getTxData txh) >>= \case Just TxData {txDataBlock = b@BlockRef{}} -> matchingSkip it (AddrTxKeyA a) (AddrTxKeyB a b) Just TxData {txDataBlock = MemRef{}} -> @@ -286,47 +177,6 @@ (dropWhileC (cond . fst) >> mapC id) Nothing -> return () -getAddressTxsDB :: MonadUnliftIO m - => Address - -> Limits - -> DatabaseReaderT m [TxRef] -getAddressTxsDB a limits = do - db <- asks databaseHandle - txs <- withIterCF db (addrTxCF db) $ \it -> - runConduit $ - addressConduit a (start limits) it .| - applyLimitsC limits .| - sinkList - countTxRefs (length txs) - return txs - -getUnspentDB :: MonadIO m - => OutPoint - -> DatabaseReaderT m (Maybe Unspent) -getUnspentDB p = do - db <- asks databaseHandle - fmap (valToUnspent p) <$> retrieveCF db (unspentCF db) (UnspentKey p) >>= \case - Nothing -> return Nothing - Just u -> countUnspents 1 >> return (Just u) - -getAddressesUnspentsDB :: MonadUnliftIO m - => [Address] - -> Limits - -> DatabaseReaderT m [Unspent] -getAddressesUnspentsDB addrs limits = do - us <- applyLimits limits . sortOn Down . concat <$> mapM f addrs - countUnspents (length us) - return us - where - l = deOffset limits - f a = do - db <- asks databaseHandle - withIterCF db (addrOutCF db) $ \it -> - runConduit $ - unspentConduit a (start l) it .| - applyLimitC (limit l) .| - sinkList - unspentConduit :: MonadUnliftIO m => Address -> Maybe Start @@ -344,7 +194,7 @@ (AddrOutKeyA a) (AddrOutKeyB a (BlockRef h maxBound)) Just (AtTx txh) -> - lift (getTxDataDB txh) >>= \case + lift (getTxData txh) >>= \case Just TxData {txDataBlock = b@BlockRef{}} -> matchingSkip it (AddrOutKeyA a) (AddrOutKeyB a b) Just TxData {txDataBlock = MemRef{}} -> @@ -356,61 +206,202 @@ (dropWhileC (cond . fst) >> mapC id) Nothing -> return () -getAddressUnspentsDB :: MonadUnliftIO m - => Address - -> Limits - -> DatabaseReaderT m [Unspent] -getAddressUnspentsDB a limits = do - db <- asks databaseHandle - us <- withIterCF db (addrOutCF db) $ \it -> runConduit $ - x it .| applyLimitsC limits .| mapC (uncurry toUnspent) .| sinkList - countUnspents (length us) - return us - where - x it = case start limits of - Nothing -> - matching it (AddrOutKeyA a) - Just (AtBlock h) -> - matchingSkip - it - (AddrOutKeyA a) - (AddrOutKeyB a (BlockRef h maxBound)) - Just (AtTx txh) -> - lift (getTxDataDB txh) >>= \case - Just TxData {txDataBlock = b@BlockRef{}} -> - matchingSkip it (AddrOutKeyA a) (AddrOutKeyB a b) - Just TxData {txDataBlock = MemRef{}} -> - let cond (AddrOutKey _a MemRef{} p) = - outPointHash p /= txh - cond (AddrOutKey _a BlockRef{} _p) = - False - in matching it (AddrOutKeyA a) .| - (dropWhileC (cond . fst) >> mapC id) - _ -> matching it (AddrOutKeyA a) - instance MonadIO m => StoreReadBase (DatabaseReaderT m) where getNetwork = asks databaseNetwork - getTxData t = getTxDataDB t - getSpender p = getSpenderDB p - getUnspent a = getUnspentDB a - getBalance a = getBalanceDB a - getMempool = getMempoolDB - getBestBlock = getBestDB - getBlocksAtHeight h = getBlocksAtHeightDB h - getBlock b = getDatabaseReader b - countBlocks = incrementCounter databaseBlockCount - countTxs = incrementCounter databaseTxCount - countBalances = incrementCounter databaseBalanceCount - countUnspents = incrementCounter databaseUnspentCount + getTxData th = do + db <- asks databaseHandle + retrieveCF db (txCF db) (TxKey th) >>= \case + Nothing -> return Nothing + Just t -> do + incrementCounter dataTxCount 1 + return (Just t) + + getSpender op = do + db <- asks databaseHandle + retrieveCF db (spenderCF db) (SpenderKey op) >>= \case + Nothing -> return Nothing + Just s -> do + incrementCounter dataSpenderCount 1 + return (Just s) + + getUnspent p = do + db <- asks databaseHandle + fmap (valToUnspent p) <$> retrieveCF db (unspentCF db) (UnspentKey p) >>= \case + Nothing -> return Nothing + Just u -> do + incrementCounter dataUnspentCount 1 + return (Just u) + + getBalance a = do + db <- asks databaseHandle + incrementCounter dataBalanceCount 1 + fmap (valToBalance a) <$> retrieveCF db (balanceCF db) (BalKey a) + + getMempool = do + db <- asks databaseHandle + incrementCounter dataMempoolCount 1 + fromMaybe [] <$> retrieve db MemKey + + getBestBlock = do + incrementCounter dataBestCount 1 + asks databaseHandle >>= (`retrieve` BestKey) + + getBlocksAtHeight h = do + db <- asks databaseHandle + retrieveCF db (heightCF db) (HeightKey h) >>= \case + Nothing -> return [] + Just ls -> do + incrementCounter dataBlockCount (length ls) + return ls + + getBlock h = do + db <- asks databaseHandle + retrieveCF db (blockCF db) (BlockKey h) >>= \case + Nothing -> return Nothing + Just b -> do + incrementCounter dataBlockCount 1 + return (Just b) + instance MonadUnliftIO m => StoreReadExtra (DatabaseReaderT m) where - getAddressesTxs as limits = getAddressesTxsDB as limits - getAddressesUnspents as limits = getAddressesUnspentsDB as limits - getAddressUnspents a limits = getAddressUnspentsDB a limits - getAddressTxs a limits = getAddressTxsDB a limits + getAddressesTxs addrs limits = do + txs <- applyLimits limits . sortOn Down . concat <$> mapM f addrs + incrementCounter dataAddrTxCount (length txs) + return txs + where + l = deOffset limits + f a = do + db <- asks databaseHandle + withIterCF db (addrTxCF db) $ \it -> + runConduit $ + addressConduit a (start l) it .| + applyLimitC (limit l) .| + sinkList + + getAddressesUnspents addrs limits = do + us <- applyLimits limits . sortOn Down . concat <$> mapM f addrs + incrementCounter dataUnspentCount (length us) + return us + where + l = deOffset limits + f a = do + db <- asks databaseHandle + withIterCF db (addrOutCF db) $ \it -> + runConduit $ + unspentConduit a (start l) it .| + applyLimitC (limit l) .| + sinkList + + getAddressUnspents a limits = do + db <- asks databaseHandle + us <- withIterCF db (addrOutCF db) $ \it -> runConduit $ + x it .| applyLimitsC limits .| mapC (uncurry toUnspent) .| sinkList + incrementCounter dataUnspentCount (length us) + return us + where + x it = case start limits of + Nothing -> + matching it (AddrOutKeyA a) + Just (AtBlock h) -> + matchingSkip + it + (AddrOutKeyA a) + (AddrOutKeyB a (BlockRef h maxBound)) + Just (AtTx txh) -> + lift (getTxData txh) >>= \case + Just TxData {txDataBlock = b@BlockRef{}} -> + matchingSkip it (AddrOutKeyA a) (AddrOutKeyB a b) + Just TxData {txDataBlock = MemRef{}} -> + let cond (AddrOutKey _a MemRef{} p) = + outPointHash p /= txh + cond (AddrOutKey _a BlockRef{} _p) = + False + in matching it (AddrOutKeyA a) .| + (dropWhileC (cond . fst) >> mapC id) + _ -> matching it (AddrOutKeyA a) + + getAddressTxs a limits = do + db <- asks databaseHandle + txs <- withIterCF db (addrTxCF db) $ \it -> + runConduit $ + addressConduit a (start limits) it .| + applyLimitsC limits .| + sinkList + incrementCounter dataAddrTxCount (length txs) + return txs + getMaxGap = asks databaseMaxGap + getInitialGap = asks databaseInitialGap - getNumTxData t = getNumTxDataDB t - countTxRefs = incrementCounter databaseTxRefCount - countXPubDerivations = incrementCounter databaseDerivations + getNumTxData i = do + db <- asks databaseHandle + let (sk, w) = decodeTxKey i + ls <- liftIO $ matchingAsListCF db (txCF db) (TxKeyS sk) + let f t = let bs = encode $ txHash (txData t) + b = BS.head (BS.drop 6 bs) + w' = b .&. 0xf8 + in w == w' + txs = filter f $ map snd ls + incrementCounter dataTxCount (length txs) + return txs + + getBalances as = do + zipWith f as <$> mapM getBalance as + where + f a Nothing = zeroBalance a + f _ (Just b) = b + + xPubBals xpub = do + igap <- getInitialGap + gap <- getMaxGap + ext1 <- derive_until_gap gap 0 (take (fromIntegral igap) (aderiv 0 0)) + if all (nullBalance . xPubBal) ext1 + then do + incrementCounter dataXPubBals (length ext1) + return ext1 + else do + ext2 <- derive_until_gap gap 0 (aderiv 0 igap) + chg <- derive_until_gap gap 1 (aderiv 1 0) + let bals = ext1 <> ext2 <> chg + incrementCounter dataXPubBals (length bals) + return bals + where + aderiv m = + deriveAddresses + (deriveFunction (xPubDeriveType xpub)) + (pubSubKey (xPubSpecKey xpub) m) + xbalance m b n = XPubBal {xPubBalPath = [m, n], xPubBal = b} + derive_until_gap _ _ [] = return [] + derive_until_gap gap m as = do + let (as1, as2) = splitAt (fromIntegral gap) as + bs <- getBalances (map snd as1) + let xbs = zipWith (xbalance m) bs (map fst as1) + if all nullBalance bs + then return xbs + else (xbs <>) <$> derive_until_gap gap m as2 + + xPubUnspents _xspec xbals limits = do + us <- concat <$> mapM h cs + incrementCounter dataXPubUnspents (length us) + return . applyLimits limits $ sortOn Down us + where + l = deOffset limits + cs = filter ((> 0) . balanceUnspentCount . xPubBal) xbals + i b = do + us <- getAddressUnspents (balanceAddress (xPubBal b)) l + return us + f b t = XPubUnspent {xPubUnspentPath = xPubBalPath b, xPubUnspent = t} + h b = map (f b) <$> i b + + xPubTxs _xspec xbals limits = do + let as = map balanceAddress $ + filter (not . nullBalance) $ + map xPubBal xbals + txs <- getAddressesTxs as limits + incrementCounter dataXPubTxs (length txs) + return txs + + xPubTxCount xspec xbals = do + incrementCounter dataXPubTxCount 1 + fromIntegral . length <$> xPubTxs xspec xbals def 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,12 +35,13 @@ blockStoreTxHashSTM, blockStoreTxSTM, withBlockStore) import Haskoin.Store.Cache (CacheConfig (..), CacheWriter, - cacheNewBlock, cacheWriter, - connectRedis, newCacheMetrics) -import Haskoin.Store.Common (StoreEvent (..)) + cacheNewBlock, cacheNewTx, + cacheWriter, connectRedis, + newCacheMetrics) +import Haskoin.Store.Common (StoreEvent (..), + createDataMetrics) import Haskoin.Store.Database.Reader (DatabaseReader (..), DatabaseReaderT, - createDatabaseStats, withDatabaseReader) import NQE (Inbox, Process (..), Publisher, publishSTM, receive, @@ -99,8 +100,6 @@ -- ^ disconnect peer if it has been connected this long , storeConfConnect :: !(SockAddr -> WithConnection) -- ^ connect to peers using the function 'withConnection' - , storeConfCacheRefresh :: !Int - -- ^ refresh the cache this often (milliseconds) , storeConfCacheRetryDelay :: !Int -- ^ delay in microseconds to retry getting cache lock , storeConfStats :: !(Maybe Metrics.Store) @@ -129,7 +128,7 @@ connectDB :: MonadUnliftIO m => StoreConfig -> DatabaseReaderT m a -> m a connectDB cfg f = do - stats <- mapM createDatabaseStats (storeConfStats cfg) + stats <- mapM createDataMetrics (storeConfStats cfg) withDatabaseReader (storeConfNetwork cfg) (storeConfInitialGap cfg) @@ -196,10 +195,9 @@ withSubscription pub $ \evts -> let conf = c conn metrics in withProcess (f conf) $ \p -> - cacheWriterProcesses crefresh evts (getProcessMailbox p) $ do + cacheWriterProcesses evts (getProcessMailbox p) $ do action (Just conf) where - crefresh = storeConfCacheRefresh cfg f conf cwinbox = runReaderT (cacheWriter conf cwinbox) db c conn metrics = CacheConfig @@ -207,18 +205,16 @@ , cacheMin = storeConfCacheMin cfg , cacheChain = chain , cacheMax = storeConfMaxKeys cfg - , cacheRefresh = storeConfCacheRefresh cfg , cacheRetryDelay = storeConfCacheRetryDelay cfg , cacheMetrics = metrics } cacheWriterProcesses :: MonadUnliftIO m - => Int - -> Inbox StoreEvent + => Inbox StoreEvent -> CacheWriter -> m a -> m a -cacheWriterProcesses crefresh evts cwm action = +cacheWriterProcesses evts cwm action = withAsync events $ \a1 -> link a1 >> action where events = cacheWriterEvents evts cwm @@ -230,8 +226,10 @@ e `cacheWriterDispatch` cwm cacheWriterDispatch :: MonadIO m => StoreEvent -> CacheWriter -> m () -cacheWriterDispatch (StoreBestBlock _) = cacheNewBlock -cacheWriterDispatch _ = const (return ()) +cacheWriterDispatch (StoreBestBlock _) = cacheNewBlock +cacheWriterDispatch (StoreMempoolNew t) = cacheNewTx t +cacheWriterDispatch (StoreMempoolDelete t) = cacheNewTx t +cacheWriterDispatch _ = const (return ()) nodeForwarder :: MonadIO m => BlockStore @@ -280,7 +278,7 @@ storeDispatch b pub (PeerMessage p (MInv (Inv is))) = do let txs = [TxHash h | InvVector t h <- is, t == InvTx || t == InvWitnessTx] - publishSTM (StoreTxAvailable p txs) pub + publishSTM (StoreTxAnnounce p txs) pub unless (null txs) $ blockStoreTxHashSTM p txs b storeDispatch _ pub (PeerMessage p (MReject r)) = diff --git a/src/Haskoin/Store/Stats.hs b/src/Haskoin/Store/Stats.hs --- a/src/Haskoin/Store/Stats.hs +++ b/src/Haskoin/Store/Stats.hs @@ -1,4 +1,5 @@ {-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE RecordWildCards #-} module Haskoin.Store.Stats ( StatDist , withStats @@ -7,6 +8,7 @@ , addClientError , addServerError , addStatQuery + , addStatItems ) where import Control.Concurrent.STM.TQueue (TQueue, flushTQueue, @@ -52,6 +54,7 @@ { statTimes :: ![Int64], statQueries :: !Int64, + statItems :: !Int64, statClientErrors :: !Int64, statServerErrors :: !Int64 } @@ -61,29 +64,43 @@ { distQueue :: !(TQueue Int64), distQueries :: !(TVar Int64), + distItems :: !(TVar Int64), distClientErrors :: !(TVar Int64), distServerErrors :: !(TVar Int64) } createStatDist :: MonadIO m => Text -> Store -> m StatDist createStatDist t store = liftIO $ do - q <- newTQueueIO - queries <- newTVarIO 0 - client_errors <- newTVarIO 0 - server_errors <- newTVarIO 0 + distQueue <- newTQueueIO + distQueries <- newTVarIO 0 + distItems <- newTVarIO 0 + distClientErrors <- newTVarIO 0 + distServerErrors <- newTVarIO 0 let metrics = HashMap.fromList - [ (t <> ".query_count", Counter . statQueries) - , (t <> ".errors.client", Counter . statClientErrors) - , (t <> ".errors.server", Counter . statServerErrors) - , (t <> ".mean_ms", Gauge . mean . statTimes) - , (t <> ".avg_ms", Gauge . avg . statTimes) - , (t <> ".max_ms", Gauge . maxi . statTimes) - , (t <> ".min_ms", Gauge . mini . statTimes) - , (t <> ".p90max_ms", Gauge . p90max . statTimes) - , (t <> ".p90avg_ms", Gauge . p90avg . statTimes) - , (t <> ".var_ms", Gauge . var . statTimes) + [ (t <> ".query_count", + Counter . statQueries) + , (t <> ".item_count", + Counter . statItems) + , (t <> ".errors.client", + Counter . statClientErrors) + , (t <> ".errors.server", + Counter . statServerErrors) + , (t <> ".mean_ms", + Gauge . mean . statTimes) + , (t <> ".avg_ms", + Gauge . avg . statTimes) + , (t <> ".max_ms", + Gauge . maxi . statTimes) + , (t <> ".min_ms", + Gauge . mini . statTimes) + , (t <> ".p90max_ms", + Gauge . p90max . statTimes) + , (t <> ".p90avg_ms", + Gauge . p90avg . statTimes) + , (t <> ".var_ms", + Gauge . var . statTimes) ] - let sd = StatDist q queries client_errors server_errors + let sd = StatDist{..} registerGroup metrics (flush sd) store return sd @@ -98,6 +115,10 @@ addStatQuery q = liftIO . atomically $ modifyTVar (distQueries q) (+1) +addStatItems :: MonadIO m => StatDist -> Int64 -> m () +addStatItems q = + liftIO . atomically . modifyTVar (distItems q) . (+) + addClientError :: MonadIO m => StatDist -> m () addClientError q = liftIO . atomically $ modifyTVar (distClientErrors q) (+1) @@ -107,12 +128,13 @@ liftIO . atomically $ modifyTVar (distServerErrors q) (+1) flush :: MonadIO m => StatDist -> m StatData -flush (StatDist q n c s) = atomically $ do - ts <- flushTQueue q - qs <- readTVar n - ce <- readTVar c - se <- readTVar s - return $ StatData ts qs ce se +flush StatDist{..} = atomically $ do + statTimes <- flushTQueue distQueue + statQueries <- readTVar distQueries + statItems <- readTVar distItems + statClientErrors <- readTVar distClientErrors + statServerErrors <- readTVar distServerErrors + return $ StatData{..} average :: Fractional a => L.Fold a a average = (/) <$> L.sum <*> L.genericLength diff --git a/src/Haskoin/Store/Web.hs b/src/Haskoin/Store/Web.hs --- a/src/Haskoin/Store/Web.hs +++ b/src/Haskoin/Store/Web.hs @@ -374,9 +374,7 @@ s <- liftWith $ \run -> withGaugeIO (gf m) (run go) restoreT $ return s -setMetrics :: MonadUnliftIO m - => (WebMetrics -> StatDist) - -> WebT m () +setMetrics :: MonadUnliftIO m => (WebMetrics -> StatDist) -> WebT m () setMetrics df = asks webMetrics >>= mapM_ go where @@ -386,6 +384,14 @@ atomically $ writeTVar t (Just df) e = error "the ways of the warrior are yet to be mastered" +addItemCount :: MonadUnliftIO m => Int -> WebT m () +addItemCount i = + asks webMetrics >>= \mm -> forM_ mm $ \m -> + S.request >>= \req -> + forM_ (V.lookup (statKey m) (vault req)) $ \t -> + readTVarIO t >>= \ms -> forM_ ms $ \s -> + addStatItems (s m) (fromIntegral i) + data WebTimeouts = WebTimeouts { txTimeout :: !Word64 , blockTimeout :: !Word64 @@ -416,9 +422,10 @@ getInitialGap = runInWebReader getInitialGap getBalances as = runInWebReader (getBalances as) getAddressesTxs as = runInWebReader . getAddressesTxs as + getAddressTxs a = runInWebReader . getAddressTxs a + getAddressUnspents a = runInWebReader . getAddressUnspents a getAddressesUnspents as = runInWebReader . getAddressesUnspents as xPubBals = runInWebReader . xPubBals - xPubSummary xpub = runInWebReader . xPubSummary xpub xPubUnspents xpub xbals = runInWebReader . xPubUnspents xpub xbals xPubTxs xpub xbals = runInWebReader . xPubTxs xpub xbals xPubTxCount xpub = runInWebReader . xPubTxCount xpub @@ -438,9 +445,10 @@ instance (MonadUnliftIO m, MonadLoggerIO m) => StoreReadExtra (WebT m) where getBalances = lift . getBalances getAddressesTxs as = lift . getAddressesTxs as + getAddressTxs a = lift . getAddressTxs a + getAddressUnspents a = lift . getAddressUnspents a getAddressesUnspents as = lift . getAddressesUnspents as xPubBals = lift . xPubBals - xPubSummary xpub = lift . xPubSummary xpub xPubUnspents xpub xbals = lift . xPubUnspents xpub xbals xPubTxs xpub xbals = lift . xPubTxs xpub xbals xPubTxCount xpub = lift . xPubTxCount xpub @@ -486,10 +494,17 @@ settings = setPort port . setHost (fromString host) getRates :: (MonadUnliftIO m, MonadLoggerIO m) - => Network -> Wreq.Session -> String -> Text -> [Word64] -> m [BinfoRate] + => Network + -> Wreq.Session + -> String + -> Text + -> [Word64] + -> m [BinfoRate] getRates net session url currency times = do handleAny err $ do - r <- liftIO $ Wreq.asJSON =<< Wreq.Session.postWith opts session url body + r <- liftIO $ + Wreq.asJSON =<< + Wreq.Session.postWith opts session url body return $ r ^. Wreq.responseBody where err _ = do @@ -887,6 +902,7 @@ Nothing -> raise ThingNotFound Just b -> do + addItemCount 1 return $ pruneTx noTx b getBlocks :: (MonadUnliftIO m, MonadLoggerIO m) @@ -901,6 +917,7 @@ scottyBlocks (GetBlocks hs (NoTx notx)) = do setMetrics statBlock bs <- getBlocks hs notx + addItemCount (length bs) return bs pruneTx :: Bool -> BlockData -> BlockData @@ -914,6 +931,7 @@ scottyBlockRaw (GetBlockRaw h) = do setMetrics statBlockRaw b <- getRawBlock h + addItemCount 1 return $ RawResult b getRawBlock :: (MonadUnliftIO m, MonadLoggerIO m) @@ -945,7 +963,8 @@ Just bb -> getBlock bb >>= \case Nothing -> raise ThingNotFound Just b -> do - return $ pruneTx notx b + addItemCount 1 + return $ pruneTx notx b scottyBlockBestRaw :: (MonadUnliftIO m, MonadLoggerIO m) @@ -956,8 +975,9 @@ getBestBlock >>= \case Nothing -> raise ThingNotFound Just bb -> do - b <- getRawBlock bb - return $ RawResult b + b <- getRawBlock bb + addItemCount 1 + return $ RawResult b -- GET BlockLatest -- @@ -970,6 +990,7 @@ blocks <- getBestBlock >>= maybe (raise ThingNotFound) (go [] <=< getBlock) + addItemCount (length blocks) return blocks where go acc Nothing = return $ reverse acc @@ -987,6 +1008,7 @@ scottyBlockHeight (GetBlockHeight h (NoTx notx)) = do setMetrics statBlock blocks <- (`getBlocks` notx) =<< getBlocksAtHeight (fromIntegral h) + addItemCount (length blocks) return blocks scottyBlockHeights :: @@ -997,6 +1019,7 @@ setMetrics statBlock bhs <- concat <$> mapM getBlocksAtHeight (fromIntegral <$> heights) blocks <- getBlocks bhs notx + addItemCount (length blocks) return blocks scottyBlockHeightRaw :: @@ -1006,6 +1029,7 @@ scottyBlockHeightRaw (GetBlockHeightRaw h) = do setMetrics statBlockRaw blocks <- mapM getRawBlock =<< getBlocksAtHeight (fromIntegral h) + addItemCount (length blocks) return $ RawResultList blocks -- GET BlockTime / BlockTimeRaw -- @@ -1018,6 +1042,7 @@ blockAtOrBefore ch t >>= \case Nothing -> raise ThingNotFound Just b -> do + addItemCount 1 return $ pruneTx notx b scottyBlockMTP :: (MonadUnliftIO m, MonadLoggerIO m) @@ -1028,6 +1053,7 @@ blockAtOrAfterMTP ch t >>= \case Nothing -> raise ThingNotFound Just b -> do + addItemCount 1 return $ pruneTx notx b scottyBlockTimeRaw :: (MonadUnliftIO m, MonadLoggerIO m) @@ -1039,6 +1065,7 @@ Nothing -> raise ThingNotFound Just b -> do raw <- lift $ toRawBlock b + addItemCount 1 return $ RawResult raw scottyBlockMTPRaw :: (MonadUnliftIO m, MonadLoggerIO m) @@ -1050,6 +1077,7 @@ Nothing -> raise ThingNotFound Just b -> do raw <- lift $ toRawBlock b + addItemCount 1 return $ RawResult raw -- GET Transactions -- @@ -1060,6 +1088,7 @@ getTransaction txid >>= \case Nothing -> raise ThingNotFound Just tx -> do + addItemCount 1 return tx scottyTxs :: @@ -1067,6 +1096,7 @@ scottyTxs (GetTxs txids) = do setMetrics statTransaction txs <- catMaybes <$> mapM f (nub txids) + addItemCount (length txs) return txs where f x = lift $ withRunInIO $ \run -> @@ -1080,6 +1110,7 @@ getTransaction txid >>= \case Nothing -> raise ThingNotFound Just tx -> do + addItemCount 1 return $ RawResult (transactionData tx) scottyTxsRaw :: @@ -1089,6 +1120,7 @@ scottyTxsRaw (GetTxsRaw txids) = do setMetrics statTransactionRaw txs <- catMaybes <$> mapM f (nub txids) + addItemCount (length txs) return $ RawResultList $ transactionData <$> txs where f x = lift $ withRunInIO $ \run -> @@ -1102,6 +1134,7 @@ Nothing -> raise ThingNotFound Just b -> do txs <- mapM f (blockDataTxs b) + addItemCount (length txs) return txs where f x = lift $ withRunInIO $ \run -> @@ -1115,6 +1148,7 @@ scottyTxsBlock (GetTxsBlock h) = do setMetrics statTransactionsBlock txs <- getTxsBlock h + addItemCount (length txs) return txs scottyTxsBlockRaw :: @@ -1124,6 +1158,7 @@ scottyTxsBlockRaw (GetTxsBlockRaw h) = do setMetrics statTransactionsBlockRaw txs <- fmap transactionData <$> getTxsBlock h + addItemCount (length txs) return $ RawResultList txs -- GET TransactionAfterHeight -- @@ -1134,7 +1169,9 @@ -> WebT m (GenericResult (Maybe Bool)) scottyTxAfter (GetTxAfter txid height) = do setMetrics statTransactionAfter - GenericResult <$> cbAfterHeight (fromIntegral height) txid + (result, count) <- cbAfterHeight (fromIntegral height) txid + addItemCount count + return $ GenericResult result -- | Check if any of the ancestors of this transaction is a coinbase after the -- specified height. Returns 'Nothing' if answer cannot be computed before @@ -1143,25 +1180,27 @@ (MonadIO m, StoreReadBase m) => H.BlockHeight -> TxHash - -> m (Maybe Bool) + -> m (Maybe Bool, Int) cbAfterHeight height txid = - inputs (10000 :: Int) HashSet.empty HashSet.empty [txid] + inputs n HashSet.empty HashSet.empty [txid] where - inputs 0 _ _ [] = return Nothing + n = 10000 + inputs 0 _ _ [] = return (Nothing, 10000) inputs i is ns [] = let is' = HashSet.union is ns ns' = HashSet.empty ts = HashSet.toList (HashSet.difference ns is) in case ts of - [] -> return (Just False) + [] -> return (Just False, n - i) _ -> inputs i is' ns' ts - inputs i is ns (t:ts) = getTransaction t >>= \case - Nothing -> return Nothing + inputs i is ns (t:ts) = + getTransaction t >>= \case + Nothing -> return (Nothing, n - i) Just tx | height_check tx -> if cb_check tx - then return (Just True) + then return (Just True, n - i + 1) else let ns' = HashSet.union (ins tx) ns - in inputs (i - 1) is ns' ts + in inputs (i - 1) is ns' ts | otherwise -> inputs (i - 1) is ns ts cb_check = any isCoinbase . transactionInputs ins = HashSet.fromList . map (outPointHash . inputPoint) . transactionInputs @@ -1175,6 +1214,7 @@ scottyPostTx :: (MonadUnliftIO m, MonadLoggerIO m) => PostTx -> WebT m TxId scottyPostTx (PostTx tx) = do setMetrics statTransactionPost + addItemCount 1 lift (asks webConfig) >>= \cfg -> lift (publishTx cfg tx) >>= \case Right () -> return (TxId (txHash tx)) Left e@(PubReject _) -> raise $ UserError (show e) @@ -1236,6 +1276,7 @@ let wl' = wl { maxLimitCount = 0 } l = Limits (validateLimit wl' False limitM) (fromIntegral o) Nothing ths <- map snd . applyLimits l <$> getMempool + addItemCount (length ths) return ths @@ -1281,14 +1322,14 @@ newLine SerialAsPrettyJSON = mempty receiveEvent :: Inbox StoreEvent -> IO (Maybe Event) -receiveEvent sub = do - se <- receive sub - return $ - case se of - StoreBestBlock b -> Just (EventBlock b) - StoreMempoolNew t -> Just (EventTx t) - StoreMempoolDelete t -> Just (EventTx t) - _ -> Nothing +receiveEvent sub = + go <$> receive sub + where + go = \case + StoreBestBlock b -> Just (EventBlock b) + StoreMempoolNew t -> Just (EventTx t) + StoreMempoolDelete t -> Just (EventTx t) + _ -> Nothing -- GET Address Transactions -- @@ -1297,6 +1338,7 @@ scottyAddrTxs (GetAddrTxs addr pLimits) = do setMetrics statAddressTransactions txs <- getAddressTxs addr =<< paramToLimits False pLimits + addItemCount (length txs) return txs scottyAddrsTxs :: @@ -1304,6 +1346,7 @@ scottyAddrsTxs (GetAddrsTxs addrs pLimits) = do setMetrics statAddressTransactions txs <- getAddressesTxs addrs =<< paramToLimits False pLimits + addItemCount (length txs) return txs scottyAddrTxsFull :: @@ -1314,6 +1357,7 @@ setMetrics statAddressTransactionsFull txs <- getAddressTxs addr =<< paramToLimits True pLimits ts <- catMaybes <$> mapM (getTransaction . txRefHash) txs + addItemCount (length ts) return ts scottyAddrsTxsFull :: (MonadUnliftIO m, MonadLoggerIO m) @@ -1322,12 +1366,14 @@ setMetrics statAddressTransactionsFull txs <- getAddressesTxs addrs =<< paramToLimits True pLimits ts <- catMaybes <$> mapM (getTransaction . txRefHash) txs + addItemCount (length ts) return ts scottyAddrBalance :: (MonadUnliftIO m, MonadLoggerIO m) => GetAddrBalance -> WebT m Balance scottyAddrBalance (GetAddrBalance addr) = do setMetrics statAddressBalance + addItemCount 1 getDefaultBalance addr scottyAddrsBalance :: @@ -1335,6 +1381,7 @@ scottyAddrsBalance (GetAddrsBalance addrs) = do setMetrics statAddressBalance balances <- getBalances addrs + addItemCount (length balances) return balances scottyAddrUnspent :: @@ -1342,6 +1389,7 @@ scottyAddrUnspent (GetAddrUnspent addr pLimits) = do setMetrics statAddressUnspent unspents <- getAddressUnspents addr =<< paramToLimits False pLimits + addItemCount (length unspents) return unspents scottyAddrsUnspent :: @@ -1349,6 +1397,7 @@ scottyAddrsUnspent (GetAddrsUnspent addrs pLimits) = do setMetrics statAddressUnspent unspents <- getAddressesUnspents addrs =<< paramToLimits False pLimits + addItemCount (length unspents) return unspents -- GET XPubs -- @@ -1358,8 +1407,9 @@ scottyXPub (GetXPub xpub deriv (NoCache noCache)) = do setMetrics statXpub let xspec = XPubSpec xpub deriv - xbals <- xPubBals xspec - lift . runNoCache noCache $ xPubSummary xspec xbals + xbals <- lift . runNoCache noCache $ xPubBals xspec + addItemCount (length xbals) + return $ xPubSummary xspec xbals scottyDelXPub :: (MonadUnliftIO m, MonadLoggerIO m) => DelCachedXPub -> WebT m (GenericResult Bool) @@ -1368,6 +1418,7 @@ let xspec = XPubSpec xpub deriv cacheM <- lift (asks (storeCache . webStore . webConfig)) n <- lift $ withCache cacheM (cacheDelXPubs [xspec]) + addItemCount (fromIntegral n) return (GenericResult (n > 0)) getXPubTxs :: (MonadUnliftIO m, MonadLoggerIO m) @@ -1376,13 +1427,16 @@ limits <- paramToLimits False plimits let xspec = XPubSpec xpub deriv xbals <- xPubBals xspec - lift . runNoCache nocache $ xPubTxs xspec xbals limits + txs <- lift . runNoCache nocache $ xPubTxs xspec xbals limits + addItemCount (length txs) + return txs scottyXPubTxs :: (MonadUnliftIO m, MonadLoggerIO m) => GetXPubTxs -> WebT m [TxRef] scottyXPubTxs (GetXPubTxs xpub deriv plimits (NoCache nocache)) = do setMetrics statXpubTransactions txs <- getXPubTxs xpub deriv plimits nocache + addItemCount (length txs) return txs scottyXPubTxsFull :: @@ -1395,6 +1449,7 @@ txs <- fmap catMaybes $ lift . runNoCache nocache $ mapM (getTransaction . txRefHash) refs + addItemCount (length txs) return txs scottyXPubBalances :: @@ -1402,6 +1457,7 @@ scottyXPubBalances (GetXPubBalances xpub deriv (NoCache noCache)) = do setMetrics statXpubBalances balances <- filter f <$> lift (runNoCache noCache (xPubBals spec)) + addItemCount (length balances) return balances where spec = XPubSpec xpub deriv @@ -1417,6 +1473,7 @@ let xspec = XPubSpec xpub deriv xbals <- xPubBals xspec unspents <- lift . runNoCache noCache $ xPubUnspents xspec xbals limits + addItemCount (length unspents) return unspents --------------------------------------- @@ -1522,6 +1579,7 @@ getBinfoUnspents numtxid height xspecs' addrs .| (dropWhileC mn >> takeC limit .| sinkList) setHeaders + addItemCount (length bus) streamEncoding (binfoUnspentsToEncoding net (BinfoUnspents bus)) where get_limit = fmap (min 1000) $ S.param "limit" `S.rescue` const (return 250) @@ -1683,6 +1741,7 @@ rates <- map binfoRatePrice <$> lift (getRates net session url code times) let hs = zipWith (convert cur aaddrs) txs (rates <> repeat 0.0) setHeaders + addItemCount (length hs) streamEncoding $ toEncoding hs where is_newer (Just BlockData{..}) TxRef{txRefBlock = BlockRef{..}} = @@ -1753,6 +1812,7 @@ ch <- lift $ asks (storeChain . webStore . webConfig) m <- blockAtOrBefore ch t bs <- go (d t) m + addItemCount (length bs) streamEncoding $ toEncoding $ map toBinfoBlockInfo bs where h = fromIntegral (maxBound :: H.Timestamp) @@ -1831,6 +1891,7 @@ , getBinfoLatestBlock = block } setHeaders + addItemCount (length abook + length ftxs) streamEncoding $ binfoMultiAddrToEncoding net BinfoMultiAddr { getBinfoMultiAddrAddresses = baddrs @@ -1915,20 +1976,20 @@ getBinfoCount :: (MonadUnliftIO m, MonadLoggerIO m) => TL.Text -> WebT m Int getBinfoCount str = do - d <- lift (asks (maxLimitDefault . webMaxLimits . webConfig)) - x <- lift (asks (maxLimitFull . webMaxLimits . webConfig)) - i <- min x <$> (S.param str `S.rescue` const (return d)) - return (fromIntegral i :: Int) + d <- lift (asks (maxLimitDefault . webMaxLimits . webConfig)) + x <- lift (asks (maxLimitFull . webMaxLimits . webConfig)) + i <- min x <$> (S.param str `S.rescue` const (return d)) + return (fromIntegral i :: Int) getBinfoOffset :: (MonadUnliftIO m, MonadLoggerIO m) => WebT m Int getBinfoOffset = do - x <- lift (asks (maxLimitOffset . webMaxLimits . webConfig)) - o <- S.param "offset" `S.rescue` const (return 0) - when (o > x) $ - raise $ - UserError $ "offset exceeded: " <> show o <> " > " <> show x - return (fromIntegral o :: Int) + x <- lift (asks (maxLimitOffset . webMaxLimits . webConfig)) + o <- S.param "offset" `S.rescue` const (return 0) + when (o > x) $ + raise $ + UserError $ "offset exceeded: " <> show o <> " > " <> show x + return (fromIntegral o :: Int) scottyRawAddr :: (MonadUnliftIO m, MonadLoggerIO m) => WebT m () scottyRawAddr = @@ -1945,8 +2006,8 @@ off <- getBinfoOffset xbals <- xPubBals xspec net <- lift $ asks (storeNetwork . webStore . webConfig) - summary <- xPubSummary xspec xbals - let abook = compute_abook xpub xbals + let summary = xPubSummary xspec xbals + abook = compute_abook xpub xbals xspecs = HashSet.singleton xspec saddrs = HashSet.empty baddrs = HashMap.keysSet abook @@ -1976,6 +2037,7 @@ , binfoRawTxs = txs } setHeaders + addItemCount (length abook + length txs) streamEncoding $ binfoRawAddrToEncoding net ra compute_abook xpub xbals = let f XPubBal{..} = @@ -2019,6 +2081,7 @@ , binfoRawTxs = txs } setHeaders + addItemCount (1 + length txs) streamEncoding $ binfoRawAddrToEncoding net ra scottyBinfoReceived :: (MonadUnliftIO m, MonadLoggerIO m) => WebT m () @@ -2027,6 +2090,7 @@ a <- getAddress "addr" b <- fromMaybe (zeroBalance a) <$> getBalance a setHeaders + addItemCount 1 S.text . cs . show $ balanceTotalReceived b scottyBinfoSent :: (MonadUnliftIO m, MonadLoggerIO m) => WebT m () @@ -2035,6 +2099,7 @@ a <- getAddress "addr" b <- fromMaybe (zeroBalance a) <$> getBalance a setHeaders + addItemCount 1 S.text . cs . show $ balanceTotalReceived b - balanceAmount b - balanceZero b scottyBinfoAddrBalance :: (MonadUnliftIO m, MonadLoggerIO m) => WebT m () @@ -2043,6 +2108,7 @@ a <- getAddress "addr" b <- fromMaybe (zeroBalance a) <$> getBalance a setHeaders + addItemCount 1 S.text . cs . show $ balanceAmount b + balanceZero b scottyFirstSeen :: (MonadUnliftIO m, MonadLoggerIO m) => WebT m () @@ -2055,6 +2121,7 @@ bot = 0 i <- go ch bb a bot top setHeaders + addItemCount 1 S.text . cs $ show i where go ch bb a bot top = do @@ -2087,6 +2154,7 @@ xbals <- mapM (get_xspec_balance net) (HashMap.elems xspecs) let res = HashMap.fromList (abals <> xbals) setHeaders + addItemCount (length abals) streamEncoding $ toEncoding res where to_short_bal Balance{..} = @@ -2123,6 +2191,7 @@ binfoShortBalTxCount = fromIntegral xts, binfoShortBalReceived = rcv } + addItemCount (length xbals) return (xPubExport net (xPubSpecKey xpub), sbl) getBinfoHex :: Monad m => WebT m Bool @@ -2143,6 +2212,7 @@ mapM (get_binfo_blocks numtxid next_block_headers) block_headers setHeaders net <- lift $ asks (storeNetwork . webStore . webConfig) + addItemCount (length binfo_blocks) streamEncoding $ binfoBlocksToEncoding net binfo_blocks where get_tx th = @@ -2154,6 +2224,7 @@ get_prev = H.prevBlock . blockDataHeader get_hash = H.headerHash . blockDataHeader txs <- lift $ mapM get_tx (blockDataTxs block_header) + addItemCount (length txs) let next_blocks = map get_hash $ filter ((== my_hash) . get_prev) next_block_headers @@ -2171,6 +2242,7 @@ binfoHeaderTime = H.blockTimestamp (blockDataHeader best) binfoHeaderIndex = binfoHeaderTime binfoHeaderHeight = blockDataHeight best + addItemCount 1 streamEncoding $ toEncoding BinfoHeader{..} where get_best_block = @@ -2221,6 +2293,7 @@ y = toBinfoBlock b btxs nxt setHeaders net <- lift $ asks (storeNetwork . webStore . webConfig) + addItemCount (length btxs + 1) streamEncoding $ binfoBlockToEncoding net y getBinfoTx :: (MonadLoggerIO m, MonadUnliftIO m) => @@ -2245,6 +2318,7 @@ tx <- getBinfoTx txid >>= \case Right t -> return t Left e -> raise e + addItemCount 1 if hex then hx tx else js numtxid tx where js numtxid t = do @@ -2262,6 +2336,7 @@ tx <- getBinfoTx txid >>= \case Right t -> return t Left e -> raise e + addItemCount 1 S.text . cs . show . sum . map outputAmount $ transactionOutputs tx scottyBinfoTxFees :: (MonadUnliftIO m, MonadLoggerIO m) => WebT m () @@ -2274,6 +2349,7 @@ let i = sum . map inputAmount . filter f $ transactionInputs tx o = sum . map outputAmount $ transactionOutputs tx + addItemCount 1 S.text . cs . show $ i - o where f StoreInput{} = True @@ -2291,6 +2367,7 @@ transactionInputs tx o = toInteger . sum . map outputAmount . filter (g addr) $ transactionOutputs tx + addItemCount 1 S.text . cs . show $ o - i where f addr StoreInput{inputAddress = Just a} = a == addr @@ -2307,6 +2384,7 @@ tx <- getBinfoTx txid >>= \case Right t -> return t Left e -> raise e + addItemCount 1 S.text . cs . show . sum . map inputAmount . filter f $ transactionInputs tx where f StoreInput{} = True @@ -2324,6 +2402,7 @@ net <- lift $ asks (storeNetwork . webStore . webConfig) setHeaders let mem = BinfoMempool $ map (toBinfoTxSimple numtxid) txs + addItemCount (length txs) streamEncoding $ binfoMempoolToEncoding net mem scottyBinfoGetBlockCount :: (MonadUnliftIO m, MonadLoggerIO m) => WebT m () @@ -2332,6 +2411,7 @@ ch <- asks (storeChain . webStore . webConfig) bn <- chainGetBest ch setHeaders + addItemCount 1 S.text . cs . show $ H.nodeHeight bn scottyBinfoLatestHash :: (MonadUnliftIO m, MonadLoggerIO m) => WebT m () @@ -2340,6 +2420,7 @@ ch <- asks (storeChain . webStore . webConfig) bn <- chainGetBest ch setHeaders + addItemCount 1 S.text . TL.fromStrict . H.blockHashToHex . H.headerHash $ H.nodeHeader bn scottyBinfoSubsidy :: (MonadUnliftIO m, MonadLoggerIO m) => WebT m () @@ -2349,6 +2430,7 @@ net <- asks (storeNetwork . webStore . webConfig) bn <- chainGetBest ch setHeaders + addItemCount 1 S.text . cs . show . (/ (100 * 1000 * 1000 :: Double)) . fromIntegral $ H.computeSubsidy net (H.nodeHeight bn + 1) @@ -2357,6 +2439,7 @@ setMetrics statBlockchainQaddresstohash addr <- getAddress "addr" setHeaders + addItemCount 1 S.text . encodeHexLazy . runPutL . serialize $ getAddrHash160 addr scottyBinfoHashToAddr :: (MonadUnliftIO m, MonadLoggerIO m) => WebT m () @@ -2367,6 +2450,7 @@ hash <- either (const S.next) return (decode bs) addr <- maybe S.next return (addrToText net (PubKeyAddress hash)) setHeaders + addItemCount 1 S.text $ TL.fromStrict addr scottyBinfoAddrPubkey :: (MonadUnliftIO m, MonadLoggerIO m) => WebT m () @@ -2379,7 +2463,9 @@ setHeaders case addrToText net pubkey of Nothing -> raise ThingNotFound - Just a -> S.text $ TL.fromStrict a + Just a -> do + addItemCount 1 + S.text $ TL.fromStrict a scottyBinfoPubKeyAddr :: (MonadUnliftIO m, MonadLoggerIO m) => WebT m () scottyBinfoPubKeyAddr = do @@ -2393,6 +2479,7 @@ Left e -> raise $ UserError e Right t -> return t setHeaders + addItemCount 1 S.text $ encodeHexLazy $ L.fromStrict pk where strm addr = runConduit $ @@ -2431,6 +2518,7 @@ Nothing -> raise $ UserError "Could not decode public key" Just pk -> return $ pubKeyAddr pk setHeaders + addItemCount 1 S.text . encodeHexLazy . runPutL . serialize $ getAddrHash160 addr @@ -2441,7 +2529,11 @@ -> WebT m [PeerInformation] scottyPeers _ = do setMetrics statPeers - lift $ getPeersInformation =<< asks (storeManager . webStore . webConfig) + ps <- lift $ + getPeersInformation =<< + asks (storeManager . webStore . webConfig) + addItemCount (length ps) + return ps -- | Obtain information about connected peers from peer manager process. getPeersInformation @@ -2471,6 +2563,7 @@ setMetrics statHealth h <- lift $ asks webConfig >>= healthCheck unless (isOK h) $ S.status status503 + addItemCount 1 return h blockHealthCheck :: (MonadUnliftIO m, MonadLoggerIO m, StoreReadBase m) @@ -2550,6 +2643,7 @@ setHeaders db <- lift $ asks (databaseHandle . storeDB . webStore . webConfig) statsM <- lift (getProperty db Stats) + addItemCount 1 S.text $ maybe "Could not get stats" cs statsM ----------------------- diff --git a/test/Haskoin/StoreSpec.hs b/test/Haskoin/StoreSpec.hs --- a/test/Haskoin/StoreSpec.hs +++ b/test/Haskoin/StoreSpec.hs @@ -97,7 +97,6 @@ , storeConfPeerTimeout = 60 , storeConfPeerMaxLife = 48 * 3600 , storeConfConnect = dummyPeerConnect net ad - , storeConfCacheRefresh = 750 , storeConfCacheRetryDelay = 100000 , storeConfStats = Nothing }