haskoin-store 0.47.6 → 0.48.0

raw patch · 5 files changed

+186/−64 lines, 5 filesdep ~haskoin-store-dataPVP ok

version bump matches the API change (PVP)

Dependency ranges changed: haskoin-store-data

API changes (from Hackage documentation)

Files

• CHANGELOG.md +5/−0
• app/Main.hs +48/−0
• haskoin-store.cabal +5/−5
• src/Haskoin/Store/Database/Reader.hs +42/−25
• src/Haskoin/Store/Web.hs +86/−34

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

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

@@ -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 <https://github.com/haskoin/haskoin-store#readme> 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

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

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