diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,10 @@ 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.65.8 +### Changed +- Perform health check in separate thread. + ## 0.65.7 ### Changed - Do not sync mempool to Redis if node is not up-to-date. diff --git a/app/Main.hs b/app/Main.hs --- a/app/Main.hs +++ b/app/Main.hs @@ -119,7 +119,8 @@ configWebTickerURL :: !String, configWebHistoryURL :: !String, configWebNoBlockchain :: !Bool, - configWebNoSlow :: !Bool + configWebNoSlow :: !Bool, + configHealthCheckInterval :: !Int } instance Default Config where @@ -155,7 +156,8 @@ configWebTickerURL = defWebTickerURL, configWebHistoryURL = defWebHistoryURL, configWebNoBlockchain = defWebNoBlockchain, - configWebNoSlow = defWebNoSlow + configWebNoSlow = defWebNoSlow, + configHealthCheckInterval = defHealthCheckInterval } defEnv :: MonadIO m => String -> a -> (String -> Maybe a) -> m a @@ -374,6 +376,12 @@ return $ "app." <> task <> "." <> service {-# NOINLINE defStatsdPrefix #-} +defHealthCheckInterval :: Int +defHealthCheckInterval = + unsafePerformIO $ + defEnv "HEALTH_CHECK_INTERVAL" 30 readMaybe +{-# NOINLINE defHealthCheckInterval #-} + netNames :: String netNames = intercalate "|" (map getNetworkName allNets) @@ -637,6 +645,13 @@ flag (configWebNoSlow def) False $ long "no-slow" <> help "Disable non-scalable API calls (recommend also --no-blockchain)" + configHealthCheckInterval <- + option auto $ + metavar "SECONDS" + <> long "health-check-interval" + <> help "Background check update interval" + <> showDefault + <> value (configHealthCheckInterval def) pure Config { configWebLimits = WebLimits {..}, @@ -708,7 +723,8 @@ configWebTickerURL = wturl, configWebHistoryURL = whurl, configWebNoSlow = no_slow, - configWebNoBlockchain = no_blockchain + configWebNoBlockchain = no_blockchain, + configHealthCheckInterval = health_check_interval } = runStderrLoggingT . filterLogger l . with_stats $ \stats -> do net <- case networkReader net_str of @@ -757,7 +773,8 @@ webTickerURL = wturl, webHistoryURL = whurl, webSlow = not no_slow, - webBlockchain = not no_blockchain + webBlockchain = not no_blockchain, + webHealthCheckInterval = health_check_interval } where with_stats go 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.65.7 +version: 0.65.8 synopsis: Storage and index for Bitcoin and Bitcoin Cash description: Please see the README on GitHub at category: Bitcoin, Finance, Network diff --git a/src/Haskoin/Store/BlockStore.hs b/src/Haskoin/Store/BlockStore.hs --- a/src/Haskoin/Store/BlockStore.hs +++ b/src/Haskoin/Store/BlockStore.hs @@ -6,6 +6,7 @@ {-# LANGUAGE RecordWildCards #-} {-# LANGUAGE TemplateHaskell #-} {-# LANGUAGE TupleSections #-} +{-# LANGUAGE ImportQualifiedPost #-} module Haskoin.Store.BlockStore ( -- * Block Store @@ -61,11 +62,11 @@ ) import Control.Monad.Trans (lift) import Control.Monad.Trans.Maybe (runMaybeT) -import qualified Data.ByteString as B +import Data.ByteString qualified as B import Data.HashMap.Strict (HashMap) -import qualified Data.HashMap.Strict as HashMap +import Data.HashMap.Strict qualified as HashMap import Data.HashSet (HashSet) -import qualified Data.HashSet as HashSet +import Data.HashSet qualified as HashSet import Data.List (delete) import Data.Maybe ( catMaybes, @@ -90,7 +91,6 @@ import Data.Time.Format ( defaultTimeLocale, formatTime, - iso8601DateFormat, ) import Haskoin ( Block (..), @@ -155,9 +155,9 @@ send, sendSTM, ) -import qualified System.Metrics as Metrics -import qualified System.Metrics.Gauge as Metrics (Gauge) -import qualified System.Metrics.Gauge as Metrics.Gauge +import System.Metrics qualified as Metrics +import System.Metrics.Gauge qualified as Metrics (Gauge) +import System.Metrics.Gauge qualified as Metrics.Gauge import System.Random (randomRIO) import UnliftIO ( Exception, @@ -411,9 +411,9 @@ Right () -> wipe_it txs2 wipe | blockConfWipeMempool cfg = - getMempool >>= wipe_it + getMempool >>= wipe_it | otherwise = - return () + return () ini = runImport initBest >>= \case Left e -> do @@ -468,7 +468,8 @@ True -> return () False -> do $(logErrorS) "BlockStore" $ - "Non-syncing peer " <> peerText peer + "Non-syncing peer " + <> peerText peer <> " sent me a block: " <> blockHashToHex blockhash PeerMisbehaving "Sent unexpected block" `killPeer` peer @@ -478,13 +479,15 @@ Just b -> return b Nothing -> do $(logErrorS) "BlockStore" $ - "Peer " <> peerText peer + "Peer " + <> peerText peer <> " sent unknown block: " <> blockHashToHex blockhash PeerMisbehaving "Sent unknown block" `killPeer` peer mzero $(logDebugS) "BlockStore" $ - "Processing block: " <> blockText node Nothing + "Processing block: " + <> blockText node Nothing <> " from peer: " <> peerText peer lift . notify (Just block) $ @@ -507,7 +510,8 @@ mempool peer failure e = do $(logErrorS) "BlockStore" $ - "Error importing block " <> hexhash + "Error importing block " + <> hexhash <> " from peer: " <> peerText peer <> ": " @@ -588,7 +592,8 @@ processTx p tx = guardMempool $ do t <- liftIO getCurrentTime $(logDebugS) "BlockManager" $ - "Received tx " <> txHashToHex (txHash tx) + "Received tx " + <> txHashToHex (txHash tx) <> " by peer: " <> peerText p addPendingTx $ PendingTx t tx HashSet.empty @@ -740,13 +745,15 @@ newMempoolTx tx seconds >>= \case True -> do $(logInfoS) "BlockStore" $ - "Import tx " <> txHashToHex (txHash tx) + "Import tx " + <> txHashToHex (txHash tx) <> ": OK" fulfillOrphans block_read tx_hash return True False -> do $(logDebugS) "BlockStore" $ - "Import tx " <> txHashToHex (txHash tx) + "Import tx " + <> txHashToHex (txHash tx) <> ": Already imported" return False @@ -811,7 +818,10 @@ isPending h >>= \case True -> do $(logDebugS) "BlockStore" $ - "Tx " <> cs (show i) <> "/" <> cs (show len) + "Tx " + <> cs (show i) + <> "/" + <> cs (show len) <> " " <> txHashToHex h <> ": " @@ -821,7 +831,10 @@ getActiveTxData h >>= \case Just _ -> do $(logDebugS) "BlockStore" $ - "Tx " <> cs (show i) <> "/" <> cs (show len) + "Tx " + <> cs (show i) + <> "/" + <> cs (show len) <> " " <> txHashToHex h <> ": " @@ -829,7 +842,10 @@ return Nothing Nothing -> do $(logDebugS) "BlockStore" $ - "Tx " <> cs (show i) <> "/" <> cs (show len) + "Tx " + <> cs (show i) + <> "/" + <> cs (show len) <> " " <> txHashToHex h <> ": " @@ -856,7 +872,8 @@ now <- liftIO getCurrentTime atomically $ modifyTVar box $ - fmap $ \x -> x {syncingTime = now} + fmap $ + \x -> x {syncingTime = now} checkTime :: MonadLoggerIO m => BlockT m () checkTime = @@ -1207,22 +1224,13 @@ blockText :: BlockNode -> Maybe Block -> Text blockText bn mblock = case mblock of Nothing -> - height <> sep <> time <> sep <> hash + height <> sep <> t <> sep <> hash Just block -> - height <> sep <> time <> sep <> hash <> sep <> size block + height <> sep <> t <> sep <> hash <> sep <> size block where height = cs $ show (nodeHeight bn) - systime = - posixSecondsToUTCTime $ - fromIntegral $ - blockTimestamp $ - nodeHeader bn - time = - cs $ - formatTime - defaultTimeLocale - (iso8601DateFormat (Just "%H:%M")) - systime + b = posixSecondsToUTCTime . fromIntegral . blockTimestamp $ nodeHeader bn + t = cs $ formatTime defaultTimeLocale "%FT%T" b hash = blockHashToHex (headerHash (nodeHeader bn)) sep = " | " size = (<> " bytes") . cs . show . B.length . encode 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 @@ -9,6 +9,7 @@ {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE TemplateHaskell #-} {-# LANGUAGE TupleSections #-} +{-# LANGUAGE ImportQualifiedPost #-} {-# OPTIONS_GHC -Wno-deprecations #-} module Haskoin.Store.Web @@ -48,6 +49,7 @@ unless, when, (<=<), + (>=>), ) import Control.Monad.Logger ( MonadLoggerIO, @@ -72,7 +74,7 @@ ToJSON (..), Value, ) -import qualified Data.Aeson as A +import Data.Aeson qualified as A import Data.Aeson.Encode.Pretty ( Config (..), defConfig, @@ -84,11 +86,11 @@ ) import Data.Aeson.Text (encodeToLazyText) import Data.ByteString (ByteString) -import qualified Data.ByteString as B -import qualified Data.ByteString.Base16 as B16 +import Data.ByteString qualified as B +import Data.ByteString.Base16 qualified as B16 import Data.ByteString.Builder (lazyByteString) -import qualified Data.ByteString.Char8 as C -import qualified Data.ByteString.Lazy as L +import Data.ByteString.Char8 qualified as C +import Data.ByteString.Lazy qualified as L import Data.Bytes.Get import Data.Bytes.Put import Data.Bytes.Serial @@ -96,9 +98,9 @@ import Data.Default (Default (..)) import Data.Function ((&)) import Data.HashMap.Strict (HashMap) -import qualified Data.HashMap.Strict as HashMap +import Data.HashMap.Strict qualified as HashMap import Data.HashSet (HashSet) -import qualified Data.HashSet as HashSet +import Data.HashSet qualified as HashSet import Data.Int (Int64) import Data.List (nub) import Data.Maybe @@ -114,24 +116,25 @@ import Data.String (fromString) import Data.String.Conversions (cs) import Data.Text (Text) -import qualified Data.Text as T -import qualified Data.Text.Encoding as T +import Data.Text qualified as T +import Data.Text.Encoding qualified as T import Data.Text.Lazy (toStrict) -import qualified Data.Text.Lazy as TL +import Data.Text.Lazy qualified as TL import Data.Time.Clock (diffUTCTime) import Data.Time.Clock.System ( getSystemTime, systemSeconds, systemToUTCTime, ) -import qualified Data.Vault.Lazy as V +import Data.Vault.Lazy qualified as V import Data.Word (Word32, Word64) import Database.RocksDB ( Property (..), getProperty, ) +import GHC.RTS.Flags (ConcFlags (ConcFlags)) import Haskoin.Address -import qualified Haskoin.Block as H +import Haskoin.Block qualified as H import Haskoin.Constants import Haskoin.Data import Haskoin.Keys @@ -201,19 +204,20 @@ requestPath, sendTextData, ) -import qualified Network.WebSockets as WebSockets -import qualified Network.Wreq as Wreq +import Network.WebSockets qualified as WebSockets +import Network.Wreq qualified as Wreq import Network.Wreq.Session as Wreq (Session) -import qualified Network.Wreq.Session as Wreq.Session +import Network.Wreq.Session qualified as Wreq.Session import System.IO.Unsafe (unsafeInterleaveIO) -import qualified System.Metrics as Metrics -import qualified System.Metrics.Gauge as Metrics (Gauge) -import qualified System.Metrics.Gauge as Metrics.Gauge +import System.Metrics qualified as Metrics +import System.Metrics.Gauge qualified as Metrics (Gauge) +import System.Metrics.Gauge qualified as Metrics.Gauge import UnliftIO ( MonadIO, MonadUnliftIO, TVar, askRunInIO, + async, atomically, bracket, bracket_, @@ -229,7 +233,7 @@ ) import UnliftIO.Concurrent (threadDelay) import Web.Scotty.Internal.Types (ActionT) -import qualified Web.Scotty.Trans as S +import Web.Scotty.Trans qualified as S type WebT m = ActionT Except (ReaderT WebState m) @@ -273,14 +277,16 @@ webTickerURL :: !String, webHistoryURL :: !String, webSlow :: !Bool, - webBlockchain :: !Bool + webBlockchain :: !Bool, + webHealthCheckInterval :: !Int } data WebState = WebState { webConfig :: !WebConfig, webTicker :: !(TVar (HashMap Text BinfoTicker)), webMetrics :: !(Maybe WebMetrics), - webWreqSession :: !Wreq.Session + webWreqSession :: !Wreq.Session, + webHealthCheck :: !(TVar HealthCheck) } data WebMetrics = WebMetrics @@ -455,12 +461,11 @@ addItemCount :: MonadUnliftIO m => Int -> WebT m () addItemCount i = - asks webMetrics >>= mapM_ \m -> + asks webMetrics >>= mapM_ \m -> do addStatItems (statAll m) (fromIntegral i) - >> S.request >>= \req -> - forM_ (V.lookup (statKey m) (vault req)) \t -> - readTVarIO t >>= mapM_ \s -> - addStatItems (s m) (fromIntegral i) + req <- S.request + forM_ (V.lookup (statKey m) (vault req)) $ + readTVarIO >=> mapM_ \s -> addStatItems (s m) (fromIntegral i) getItemCounter :: (MonadIO m, MonadIO n) => WebT m (Int -> n ()) getItemCounter = @@ -554,21 +559,27 @@ webTickerURL = turl, webMaxLimits = WebLimits {..}, webSlow = slow, - webBlockchain = blockchain + webBlockchain = blockchain, + webHealthCheckInterval = healthint } = do ticker <- newTVarIO HashMap.empty metrics <- mapM createMetrics stats session <- liftIO Wreq.Session.newAPISession + health' <- hcheck >>= newTVarIO let st = WebState { webConfig = cfg, webTicker = ticker, webMetrics = metrics, - webWreqSession = session + webWreqSession = session, + webHealthCheck = health' } net = storeNetwork store' - withAsync (when blockchain $ price net session turl pget ticker) $ - const $ do + withAsync (when blockchain $ price net session turl pget ticker) + . const + . withAsync (health health') + . const + $ do reqLogger <- logIt metrics runner <- askRunInIO S.scottyOptsT opts (runner . (`runReaderT` st)) $ do @@ -580,6 +591,10 @@ handlePaths slow blockchain S.notFound $ raise ThingNotFound where + hcheck = runReaderT (healthCheck cfg) (storeDB store') + health v = forever $ do + threadDelay (healthint * 1000 * 1000) + hcheck >>= atomically . writeTVar v opts = def {S.settings = settings defaultSettings} settings = setPort port . setHost (fromString host) @@ -644,15 +659,15 @@ let status = errStatus err if | statusIsClientError status -> - liftIO $ do - addClientError (statAll m) - forM_ mM $ \f -> addClientError (f m) + liftIO $ do + addClientError (statAll m) + forM_ mM $ \f -> addClientError (f m) | statusIsServerError status -> - liftIO $ do - addServerError (statAll m) - forM_ mM $ \f -> addServerError (f m) + liftIO $ do + addServerError (statAll m) + forM_ mM $ \f -> addServerError (f m) | otherwise -> - return () + return () S.raise err errStatus :: Except -> Status @@ -1130,8 +1145,8 @@ | blockDataHeight b <= 0 = return $ reverse acc | length acc == 99 = return . reverse $ pruneTx noTx b : acc | otherwise = do - let prev = H.prevBlock (blockDataHeader b) - go (pruneTx noTx b : acc) =<< getBlock prev + let prev = H.prevBlock (blockDataHeader b) + go (pruneTx noTx b : acc) =<< getBlock prev -- GET BlockHeight / BlockHeights / BlockHeightRaw -- @@ -1345,11 +1360,11 @@ Nothing -> return (Nothing, n - i) Just tx | height_check tx -> - if cb_check tx - then return (Just True, n - i + 1) - else - let ns' = HashSet.union (ins tx) ns - in inputs (i - 1) is ns' ts + if cb_check tx + then return (Just True, n - i + 1) + else + let ns' = HashSet.union (ins tx) ns + 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 @@ -1402,10 +1417,10 @@ f p s | webNoMempool cfg = return $ Right () | otherwise = - liftIO (UnliftIO.timeout t (g p s)) >>= \case - Nothing -> return $ Left PubTimeout - Just (Left e) -> return $ Left e - Just (Right ()) -> return $ Right () + liftIO (UnliftIO.timeout t (g p s)) >>= \case + Nothing -> return $ Left PubTimeout + Just (Left e) -> return $ Left e + Just (Right ()) -> return $ Right () g p s = receive s >>= \case StoreTxReject p' h' c _ @@ -1644,32 +1659,32 @@ netBinfoSymbol :: Network -> BinfoSymbol netBinfoSymbol net | net == btc = - BinfoSymbol - { getBinfoSymbolCode = "BTC", - getBinfoSymbolString = "BTC", - getBinfoSymbolName = "Bitcoin", - getBinfoSymbolConversion = 100 * 1000 * 1000, - getBinfoSymbolAfter = True, - getBinfoSymbolLocal = False - } + BinfoSymbol + { getBinfoSymbolCode = "BTC", + getBinfoSymbolString = "BTC", + getBinfoSymbolName = "Bitcoin", + getBinfoSymbolConversion = 100 * 1000 * 1000, + getBinfoSymbolAfter = True, + getBinfoSymbolLocal = False + } | net == bch = - BinfoSymbol - { getBinfoSymbolCode = "BCH", - getBinfoSymbolString = "BCH", - getBinfoSymbolName = "Bitcoin Cash", - getBinfoSymbolConversion = 100 * 1000 * 1000, - getBinfoSymbolAfter = True, - getBinfoSymbolLocal = False - } + BinfoSymbol + { getBinfoSymbolCode = "BCH", + getBinfoSymbolString = "BCH", + getBinfoSymbolName = "Bitcoin Cash", + getBinfoSymbolConversion = 100 * 1000 * 1000, + getBinfoSymbolAfter = True, + getBinfoSymbolLocal = False + } | otherwise = - BinfoSymbol - { getBinfoSymbolCode = "XTS", - getBinfoSymbolString = "¤", - getBinfoSymbolName = "Test", - getBinfoSymbolConversion = 100 * 1000 * 1000, - getBinfoSymbolAfter = False, - getBinfoSymbolLocal = False - } + BinfoSymbol + { getBinfoSymbolCode = "XTS", + getBinfoSymbolString = "¤", + getBinfoSymbolName = "Test", + getBinfoSymbolConversion = 100 * 1000 * 1000, + getBinfoSymbolAfter = False, + getBinfoSymbolLocal = False + } binfoTickerToSymbol :: Text -> BinfoTicker -> BinfoSymbol binfoTickerToSymbol code BinfoTicker {..} = @@ -1891,7 +1906,7 @@ Nothing -> 0 Just a | a `HashSet.member` baddrs -> - if b then val else negate val + if b then val else negate val | otherwise -> 0 in sum $ map (f False) ins <> map (f True) out go b = @@ -2063,10 +2078,10 @@ go _ Nothing = return [] go t (Just b) | H.blockTimestamp (blockDataHeader b) <= fromIntegral t = - return [] + return [] | otherwise = do - b' <- getBlock (H.prevBlock (blockDataHeader b)) - (b :) <$> go t b' + b' <- getBlock (H.prevBlock (blockDataHeader b)) + (b :) <$> go t b' scottyMultiAddr :: (MonadUnliftIO m, MonadLoggerIO m) => WebT m () scottyMultiAddr = do @@ -2249,13 +2264,15 @@ o <- S.param "offset" `S.rescue` const (return 0) when (o > x) $ raise $ - UserError $ "offset exceeded: " <> show o <> " > " <> show x + UserError $ + "offset exceeded: " <> show o <> " > " <> show x return (fromIntegral o :: Int) scottyRawAddr :: (MonadUnliftIO m, MonadLoggerIO m) => WebT m () scottyRawAddr = setMetrics statBlockchainRawaddr - >> getBinfoAddr "addr" >>= \case + >> getBinfoAddr "addr" + >>= \case BinfoAddr addr -> do_addr addr BinfoXpub xpub -> do_xpub xpub where @@ -2493,7 +2510,8 @@ get_tx th = withRunInIO $ \run -> unsafeInterleaveIO $ - run $ fromJust <$> getTransaction th + run $ + fromJust <$> getTransaction th get_binfo_blocks numtxid next_block_headers block_header = do let my_hash = H.headerHash (blockDataHeader block_header) get_prev = H.prevBlock . blockDataHeader @@ -2545,7 +2563,8 @@ get_tx th = withRunInIO $ \run -> unsafeInterleaveIO $ - run $ fromJust <$> getTransaction th + run $ + fromJust <$> getTransaction th go numtxid hex bh = getBlock bh >>= \case Nothing -> raise ThingNotFound @@ -2862,7 +2881,7 @@ (MonadUnliftIO m, MonadLoggerIO m) => GetHealth -> WebT m HealthCheck scottyHealth _ = do setMetrics statHealth - h <- lift $ asks webConfig >>= healthCheck + h <- asks webHealthCheck >>= readTVarIO unless (isOK h) $ S.status status503 addItemCount 1 return h