diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,11 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/) and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.html) +## 0.48.0 +### Added +- Timeouts and token bucket for web requests. +- Configurable price fetch interval. + ## 0.47.6 ### Fixed - Fix web.all stat. diff --git a/app/Main.hs b/app/Main.hs --- a/app/Main.hs +++ b/app/Main.hs @@ -79,6 +79,9 @@ , configStatsdHost :: !String , configStatsdPort :: !Int , configStatsdPrefix :: !String + , configWebRequests :: !Int + , configWebReqTimeout :: !Int + , configWebPriceGet :: !Int } instance Default Config where @@ -110,6 +113,9 @@ , configStatsdHost = defStatsdHost , configStatsdPort = defStatsdPort , configStatsdPrefix = defStatsdPrefix + , configWebRequests = defWebRequests + , configWebReqTimeout = defWebReqTimeout + , configWebPriceGet = defWebPriceGet } defEnv :: MonadIO m => String -> a -> (String -> Maybe a) -> m a @@ -264,6 +270,21 @@ defEnv "STATSD_PORT" 8125 readMaybe {-# NOINLINE defStatsdPort #-} +defWebRequests :: Int +defWebRequests = unsafePerformIO $ + defEnv "WEB_REQUESTS" 32 readMaybe +{-# NOINLINE defWebRequests #-} + +defWebReqTimeout :: Int +defWebReqTimeout = unsafePerformIO $ + defEnv "WEB_REQ_TIMEOUT" (45 * 1000 * 1000) readMaybe +{-# NOINLINE defWebReqTimeout #-} + +defWebPriceGet :: Int +defWebPriceGet = unsafePerformIO $ + defEnv "WEB_PRICE_GET" (90 * 1000 * 1000) readMaybe +{-# NOINLINE defWebPriceGet #-} + defStatsdPrefix :: String defStatsdPrefix = unsafePerformIO $ @@ -515,6 +536,27 @@ <> help "Prefix for statsd metrics" <> showDefault <> value (configStatsdPrefix def) + configWebRequests <- + option auto $ + metavar "INT" + <> long "web-requests" + <> help "Simultaneous web requests" + <> showDefault + <> value (configWebRequests def) + configWebReqTimeout <- + option auto $ + metavar "MICROSECONDS" + <> long "web-req-timeout" + <> help "Web request server timeout" + <> showDefault + <> value (configWebReqTimeout def) + configWebPriceGet <- + option auto $ + metavar "MICROSECONDS" + <> long "web-price-get" + <> help "How often to retrieve price information" + <> showDefault + <> value (configWebPriceGet def) pure Config { configWebLimits = WebLimits {..} @@ -591,6 +633,9 @@ , configStatsdHost = statsdhost , configStatsdPort = statsdport , configStatsdPrefix = statsdpfx + , configWebRequests = wreqs + , configWebReqTimeout = wtimeout + , configWebPriceGet = wpget } = runStderrLoggingT . filterLogger l . with_stats $ \stats -> do $(logInfoS) "Main" $ @@ -634,6 +679,9 @@ , webMaxDiff = maxdiff , webNoMempool = nomem , webStats = stats + , webRequests = wreqs + , webReqTimeout = wtimeout + , webPriceGet = wpget } where with_stats go diff --git a/haskoin-store.cabal b/haskoin-store.cabal --- a/haskoin-store.cabal +++ b/haskoin-store.cabal @@ -4,10 +4,10 @@ -- -- see: https://github.com/sol/hpack -- --- hash: c65aded9c375ca3950e2be61403d36a62bb6e34dc746a54fc6ca460b09b14df5 +-- hash: 386ac449dae5732a7ab6b6f5a3aa3f54779b3acc90fda516a2560304d449327b name: haskoin-store -version: 0.47.6 +version: 0.48.0 synopsis: Storage and index for Bitcoin and Bitcoin Cash description: Please see the README on GitHub at category: Bitcoin, Finance, Network @@ -61,7 +61,7 @@ , hashable >=1.3.0.0 , haskoin-core >=0.19.0 , haskoin-node >=0.17.0 - , haskoin-store-data ==0.47.4 + , haskoin-store-data ==0.48.0 , hedis >=0.12.13 , http-types >=0.12.3 , lens >=4.18.1 @@ -113,7 +113,7 @@ , haskoin-core >=0.19.0 , haskoin-node >=0.17.0 , haskoin-store - , haskoin-store-data ==0.47.4 + , haskoin-store-data ==0.48.0 , hedis >=0.12.13 , http-types >=0.12.3 , lens >=4.18.1 @@ -170,7 +170,7 @@ , haskoin-core >=0.19.0 , haskoin-node >=0.17.0 , haskoin-store - , haskoin-store-data ==0.47.4 + , haskoin-store-data ==0.48.0 , hedis >=0.12.13 , hspec >=2.7.1 , http-types >=0.12.3 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 @@ -16,8 +16,8 @@ , balanceCF ) where -import Conduit (lift, mapC, runConduit, sinkList, - (.|)) +import Conduit (ConduitT, lift, mapC, runConduit, + sinkList, (.|)) import Control.Monad.Except (runExceptT, throwError) import Control.Monad.Reader (ReaderT, ask, asks, runReaderT) import Data.Bits ((.&.)) @@ -29,7 +29,8 @@ import Data.Serialize (encode) import Data.Word (Word32, Word64) import Database.RocksDB (ColumnFamily, Config (..), - DB (..), withDBCF, withIterCF) + DB (..), Iterator, withDBCF, + withIterCF) import Database.RocksDB.Query (insert, matching, matchingAsListCF, matchingSkip, retrieve, retrieveCF) @@ -39,6 +40,7 @@ import Haskoin.Store.Common import Haskoin.Store.Data import Haskoin.Store.Database.Types +import Haskoin.Store.Logic (joinStreams) import UnliftIO (MonadIO, MonadUnliftIO, liftIO) type DatabaseReaderT = ReaderT DatabaseReader @@ -173,11 +175,39 @@ -> Limits -> DatabaseReader -> m [TxRef] -getAddressesTxsDB addrs limits db = do - ts <- concat <$> mapM (\a -> getAddressTxsDB a (deOffset limits) db) addrs - let ts' = sortBy (flip compare `on` txRefBlock) (nub' ts) - return $ applyLimits limits ts' +getAddressesTxsDB addrs limits bdb@DatabaseReader{databaseHandle = db} = + liftIO $ iters addrs [] $ \cs -> + runConduit $ + joinStreams (flip compare) cs .| applyLimitsC limits .| sinkList + where + iters [] acc f = f acc + iters (a : as) acc f = + withIterCF db (addrTxCF db) $ \it -> + iters as (addressConduit a bdb (start limits) it : acc) f +addressConduit :: MonadUnliftIO m + => Address + -> DatabaseReader + -> Maybe Start + -> Iterator + -> ConduitT i TxRef m () +addressConduit a bdb s it = + x .| mapC (uncurry f) + where + f (AddrTxKey _ t) () = t + f _ _ = undefined + x = case s of + Nothing -> matching it (AddrTxKeyA a) + Just (AtTx txh) -> lift (getTxDataDB txh bdb) >>= \case + Just TxData {txDataBlock = b@BlockRef{}} -> + matchingSkip it (AddrTxKeyA a) (AddrTxKeyB a b) + _ -> matching it (AddrTxKeyA a) + Just (AtBlock bh) -> + matchingSkip + it + (AddrTxKeyA a) + (AddrTxKeyB a (BlockRef bh maxBound)) + getAddressTxsDB :: MonadIO m => Address @@ -185,24 +215,11 @@ -> DatabaseReader -> m [TxRef] getAddressTxsDB a limits bdb@DatabaseReader{databaseHandle = db} = - liftIO $ withIterCF db (addrTxCF db) $ \it -> runConduit $ - x it .| applyLimitsC limits .| mapC (uncurry f) .| sinkList - where - x it = - case start limits of - Nothing -> matching it (AddrTxKeyA a) - Just (AtTx txh) -> - lift (getTxDataDB txh bdb) >>= \case - Just TxData {txDataBlock = b@BlockRef {}} -> - matchingSkip it (AddrTxKeyA a) (AddrTxKeyB a b) - _ -> matching it (AddrTxKeyA a) - Just (AtBlock bh) -> - matchingSkip - it - (AddrTxKeyA a) - (AddrTxKeyB a (BlockRef bh maxBound)) - f AddrTxKey {addrTxKeyT = t} () = t - f _ _ = undefined + liftIO . withIterCF db (addrTxCF db) $ \it -> + runConduit $ + addressConduit a bdb (start limits) it .| + applyLimitsC limits .| + sinkList getUnspentDB :: MonadIO m => OutPoint -> DatabaseReader -> m (Maybe Unspent) getUnspentDB p DatabaseReader{databaseHandle = db} = 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 @@ -22,6 +22,7 @@ sinkList, takeC, yield, (.|)) import Control.Applicative ((<|>)) import Control.Arrow (second) +import Control.Concurrent.STM (check) import Control.Lens import Control.Monad (forM, forever, unless, when, (<=<)) @@ -113,9 +114,9 @@ import UnliftIO (MonadIO, MonadUnliftIO, TVar, askRunInIO, atomically, bracket, bracket_, handleAny, liftIO, - newTVarIO, readTVarIO, timeout, - withAsync, withRunInIO, - writeTVar) + modifyTVar, newTVarIO, readTVar, + readTVarIO, timeout, withAsync, + withRunInIO, writeTVar) import UnliftIO.Concurrent (threadDelay) import Web.Scotty.Internal.Types (ActionT) import Web.Scotty.Trans (Parsable) @@ -155,40 +156,45 @@ , webVersion :: !String , webNoMempool :: !Bool , webStats :: !(Maybe Metrics.Store) + , webRequests :: !Int + , webReqTimeout :: !Int + , webPriceGet :: !Int } data WebState = WebState { webConfig :: !WebConfig , webTicker :: !(TVar (HashMap Text BinfoTicker)) , webMetrics :: !(Maybe WebMetrics) + , webTokens :: !(TVar Int) } data WebMetrics = WebMetrics - { everyStat :: !StatDist - , blockStat :: !StatDist - , rawBlockStat :: !StatDist - , txStat :: !StatDist - , txsBlockStat :: !StatDist - , txAfterStat :: !StatDist - , postTxStat :: !StatDist - , mempoolStat :: !StatDist - , addrTxStat :: !StatDist - , addrTxFullStat :: !StatDist - , addrBalanceStat :: !StatDist - , addrUnspentStat :: !StatDist - , xPubStat :: !StatDist - , xPubTxStat :: !StatDist - , xPubTxFullStat :: !StatDist - , xPubUnspentStat :: !StatDist - , multiaddrStat :: !StatDist - , rawaddrStat :: !StatDist - , balanceStat :: !StatDist - , unspentStat :: !StatDist - , rawtxStat :: !StatDist - , peerStat :: !StatDist - , healthStat :: !StatDist - , dbStatsStat :: !StatDist - , eventsConnected :: !Metrics.Gauge + { everyStat :: !StatDist + , blockStat :: !StatDist + , rawBlockStat :: !StatDist + , txStat :: !StatDist + , txsBlockStat :: !StatDist + , txAfterStat :: !StatDist + , postTxStat :: !StatDist + , mempoolStat :: !StatDist + , addrTxStat :: !StatDist + , addrTxFullStat :: !StatDist + , addrBalanceStat :: !StatDist + , addrUnspentStat :: !StatDist + , xPubStat :: !StatDist + , xPubTxStat :: !StatDist + , xPubTxFullStat :: !StatDist + , xPubUnspentStat :: !StatDist + , multiaddrStat :: !StatDist + , rawaddrStat :: !StatDist + , balanceStat :: !StatDist + , unspentStat :: !StatDist + , rawtxStat :: !StatDist + , peerStat :: !StatDist + , healthStat :: !StatDist + , dbStatsStat :: !StatDist + , eventsConnected :: !Metrics.Gauge + , inFlightRequests :: !Metrics.Gauge } createMetrics :: MonadIO m => Metrics.Store -> m WebMetrics @@ -218,6 +224,7 @@ healthStat <- d "health" dbStatsStat <- d "dbstats" eventsConnected <- g "events.connected" + inFlightRequests <- g "inflight" return WebMetrics{..} where d x = createStatDist ("web." <> x) s @@ -241,6 +248,38 @@ start m = liftIO $ Metrics.Gauge.inc (gf m) end m = liftIO $ Metrics.Gauge.dec (gf m) +withTimeout :: MonadUnliftIO m + => (WebMetrics -> StatDist) + -> WebT m a + -> WebT m a +withTimeout metric go = do + to <- lift $ asks (webReqTimeout . webConfig) + m <- liftWith $ \run -> timeout to (run go) + case m of + Nothing -> raise metric ServerTimeout + Just x -> restoreT $ return x + +withToken :: MonadUnliftIO m => WebT m a -> WebT m a +withToken go = do + tok <- lift $ asks webTokens + minf <- lift $ asks (fmap inFlightRequests . webMetrics) + max_req <- lift $ asks (webRequests . webConfig) + x <- liftWith $ \run -> + bracket (t tok) (p tok) $ \rem -> do + let i = fromIntegral $ max_req - rem + case minf of + Nothing -> return () + Just inf -> liftIO $ Metrics.Gauge.set inf i + run go + restoreT $ return x + where + t tok = atomically $ do + ts <- readTVar tok + check (0 < ts) + modifyTVar tok (subtract 1) + readTVar tok + p tok _ = atomically $ modifyTVar tok (+1) + withMetrics :: MonadUnliftIO m => (WebMetrics -> StatDist) -> Int @@ -248,16 +287,17 @@ -> WebT m a withMetrics df i go = lift (asks webMetrics) >>= \case - Nothing -> go + Nothing -> go' Just m -> do x <- liftWith $ \run -> bracket (systemToUTCTime <$> liftIO getSystemTime) (end m) - (const (run go)) + (const (run go')) restoreT $ return x where + go' = withTimeout df $ withToken go end metrics t1 = do t2 <- systemToUTCTime <$> liftIO getSystemTime let diff = round $ diffUTCTime t2 t1 * 1000 @@ -337,11 +377,21 @@ , webPort = port , webStore = store , webStats = stats + , webPriceGet = pget + , webRequests = toks } = do ticker <- newTVarIO HashMap.empty metrics <- mapM createMetrics stats - let st = WebState { webConfig = cfg, webTicker = ticker, webMetrics = metrics } - withAsync (price (storeNetwork store) ticker) $ \_ -> do + tok <- newTVarIO toks + let st = WebState + { + webConfig = cfg, + webTicker = ticker, + webMetrics = metrics, + webTokens = tok + } + net = storeNetwork store + withAsync (price net pget ticker) $ \_ -> do reqLogger <- logIt runner <- askRunInIO S.scottyOptsT opts (runner . (`runReaderT` st)) $ do @@ -355,9 +405,10 @@ price :: (MonadUnliftIO m, MonadLoggerIO m) => Network + -> Int -> TVar (HashMap Text BinfoTicker) -> m () -price net v = +price net pget v = case code of Nothing -> return () Just s -> go s @@ -372,7 +423,7 @@ handleAny err $ do r <- liftIO $ Wreq.asJSON =<< Wreq.get url atomically . writeTVar v $ r ^. Wreq.responseBody - threadDelay $ 5 * 60 * 1000 * 1000 -- five minutes + threadDelay pget raise_ :: MonadIO m => Except -> WebT m a raise_ err = @@ -413,6 +464,7 @@ errStatus StringError{} = status400 errStatus ServerError = status500 errStatus TxIndexConflict{} = status409 +errStatus ServerTimeout = status500 defHandler :: Monad m => Except -> WebT m () defHandler e = do