{-# LANGUAGE RecordWildCards #-}

-- | Provides Prometheus-based metrics server based on `Tracer` collection.
--
-- To add a new metric, one needs to:
--
--  * Add a field to 'Metrics' and register it in 'registerMetrics',
--  * Update the 'monitor' function to handle relevant 'HydraLog' entries and
--    update the underlying Prometheus metric. Nested helpers are provided to
--    increase a 'Counter' by one (@tick@) and to 'observe' some value in a
--    'Histogram'.
--
-- The metric handles are typed, so a metric can only be updated in the way it
-- was registered and a name can only be referred to if it was registered.
module Hydra.Logging.Monitoring (
  withMonitoring,
) where

import Hydra.Prelude

import Control.Concurrent.Class.MonadSTM (modifyTVar', readTVarIO, writeTVar)
import Control.Tracer (Tracer (Tracer))
import Data.Map.Strict as Map
import Data.Time.Clock.POSIX (utcTimeToPOSIXSeconds)
import GHC.Stats (RTSStats (..), getRTSStats, getRTSStatsEnabled)
import Hydra.HeadLogic (
  Input (NetworkInput),
 )
import Hydra.HeadLogic.Outcome (Outcome (..), StateChanged (..))
import Hydra.Logging.Messages (HydraLog (..))
import Hydra.Network (PortNumber)
import Hydra.Network.Message (Message (ReqTx), NetworkEvent (..))
import Hydra.Node (HydraNodeLog (..))
import Hydra.Tx (IsTx (TxIdType), Snapshot (..), SnapshotNumber, txId)
import System.Metrics.Prometheus.Concurrent.Registry (
  Registry,
  new,
  registerCounter,
  registerGauge,
  registerHistogram,
  sample,
 )
import System.Metrics.Prometheus.Http.Scrape (serveMetrics)
import System.Metrics.Prometheus.Metric.Counter (Counter, add, inc)
import System.Metrics.Prometheus.Metric.Gauge (Gauge)
import System.Metrics.Prometheus.Metric.Gauge qualified as Gauge
import System.Metrics.Prometheus.Metric.Histogram (Histogram, observe)
import System.Metrics.Prometheus.MetricId (Name (Name))

-- | Handles to all metrics hydra-node exposes.
--
-- NOTE: The 'Name's below are a public interface: they are scraped by
-- Prometheus and referenced by the Grafana dashboards in @demo/grafana@, so
-- renaming one is a breaking change for operators.
data Metrics = Metrics
  { Metrics -> Counter
headInputs :: Counter
  , Metrics -> Counter
headRequestedTx :: Counter
  , Metrics -> Counter
headConfirmedTx :: Counter
  , Metrics -> Histogram
txConfirmationTime :: Histogram
  , Metrics -> Histogram
snapshotConfirmationTime :: Histogram
  , Metrics -> Gauge
peersConnected :: Gauge
  , Metrics -> Gauge
chainDriftSeconds :: Gauge
  , Metrics -> Gauge
chainLastBlockTimestampSeconds :: Gauge
  }

-- | Wraps a monadic action using a `Tracer` and capture metrics based on traces.
-- Given a `portNumber`, this wrapper starts a Prometheus-compliant server on this port.
-- This is a no-op if given `Nothing`. This function is not polymorphic over the type of
-- messages because it needs to understand them in order to provide meaningful metrics.
withMonitoring ::
  (MonadIO m, MonadAsync m, IsTx tx, MonadMonotonicTime m, MonadTime m, MonadLabelledSTM m) =>
  Maybe PortNumber ->
  Tracer m (HydraLog tx) ->
  (Tracer m (HydraLog tx) -> m ()) ->
  m ()
withMonitoring :: forall (m :: * -> *) tx.
(MonadIO m, MonadAsync m, IsTx tx, MonadMonotonicTime m,
 MonadTime m, MonadLabelledSTM m) =>
Maybe PortNumber
-> Tracer m (HydraLog tx)
-> (Tracer m (HydraLog tx) -> m ())
-> m ()
withMonitoring Maybe PortNumber
Nothing Tracer m (HydraLog tx)
tracer Tracer m (HydraLog tx) -> m ()
action = Tracer m (HydraLog tx) -> m ()
action Tracer m (HydraLog tx)
tracer
withMonitoring (Just PortNumber
monitoringPort) (Tracer HydraLog tx -> m ()
tracer) Tracer m (HydraLog tx) -> m ()
action = do
  (HydraLog tx -> m ()
traceMetric, Registry
registry) <- m (HydraLog tx -> m (), Registry)
forall (m :: * -> *) tx.
(MonadIO m, MonadMonotonicTime m, MonadTime m, IsTx tx,
 MonadLabelledSTM m) =>
m (HydraLog tx -> m (), Registry)
prepareRegistry
  IO ()
refreshRts <- IO (IO ()) -> m (IO ())
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO (IO ()) -> m (IO ())) -> IO (IO ()) -> m (IO ())
forall a b. (a -> b) -> a -> b
$ Registry -> IO (IO ())
registerRtsMetrics Registry
registry
  (String, m ()) -> (Async m () -> m ()) -> m ()
forall (m :: * -> *) a b.
MonadAsync m =>
(String, m a) -> (Async m a -> m b) -> m b
withAsyncLabelled
    (String
"monitoring-serveMetrics", Port -> Path -> IO RegistrySample -> m ()
forall (m :: * -> *).
MonadIO m =>
Port -> Path -> IO RegistrySample -> m ()
serveMetrics (PortNumber -> Port
forall a b. (Integral a, Num b) => a -> b
fromIntegral PortNumber
monitoringPort) [Text
"metrics"] (IO ()
refreshRts IO () -> IO RegistrySample -> IO RegistrySample
forall a b. IO a -> IO b -> IO b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> Registry -> IO RegistrySample
sample Registry
registry))
    ((Async m () -> m ()) -> m ()) -> (Async m () -> m ()) -> m ()
forall a b. (a -> b) -> a -> b
$ \Async m ()
_ ->
      let wrappedTracer :: Tracer m (HydraLog tx)
wrappedTracer = (HydraLog tx -> m ()) -> Tracer m (HydraLog tx)
forall (m :: * -> *) a. (a -> m ()) -> Tracer m a
Tracer ((HydraLog tx -> m ()) -> Tracer m (HydraLog tx))
-> (HydraLog tx -> m ()) -> Tracer m (HydraLog tx)
forall a b. (a -> b) -> a -> b
$ \HydraLog tx
msg -> do
            HydraLog tx -> m ()
traceMetric HydraLog tx
msg
            HydraLog tx -> m ()
tracer HydraLog tx
msg
       in Tracer m (HydraLog tx) -> m ()
action Tracer m (HydraLog tx)
wrappedTracer

-- | Register GHC RTS work counters when the runtime collects them (process
-- started with '+RTS -T'); returns a refresh action run before each scrape.
-- Without -T nothing is registered, so the endpoint output is identical to
-- before these metrics existed.
registerRtsMetrics :: Registry -> IO (IO ())
registerRtsMetrics :: Registry -> IO (IO ())
registerRtsMetrics Registry
registry = do
  Bool
enabled <- IO Bool
getRTSStatsEnabled
  if Bool -> Bool
not Bool
enabled
    then IO () -> IO (IO ())
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (() -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ())
    else do
      Gauge
allocated <- Text -> IO Gauge
rtsGauge Text
"hydra_rts_allocated_bytes"
      Gauge
mutatorCpu <- Text -> IO Gauge
rtsGauge Text
"hydra_rts_mutator_cpu_seconds"
      Gauge
gcCpu <- Text -> IO Gauge
rtsGauge Text
"hydra_rts_gc_cpu_seconds"
      Gauge
maxLive <- Text -> IO Gauge
rtsGauge Text
"hydra_rts_max_live_bytes"
      Gauge
majorGcs <- Text -> IO Gauge
rtsGauge Text
"hydra_rts_major_gcs"
      IO () -> IO (IO ())
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (IO () -> IO (IO ())) -> IO () -> IO (IO ())
forall a b. (a -> b) -> a -> b
$ do
        RTSStats
stats <- IO RTSStats
getRTSStats
        Double -> Gauge -> IO ()
Gauge.set (Word64 -> Double
forall a b. (Integral a, Num b) => a -> b
fromIntegral (RTSStats -> Word64
allocated_bytes RTSStats
stats)) Gauge
allocated
        Double -> Gauge -> IO ()
Gauge.set (RtsTime -> Double
forall a b. (Integral a, Num b) => a -> b
fromIntegral (RTSStats -> RtsTime
mutator_cpu_ns RTSStats
stats) Double -> Double -> Double
forall a. Fractional a => a -> a -> a
/ Double
1.0e9) Gauge
mutatorCpu
        Double -> Gauge -> IO ()
Gauge.set (RtsTime -> Double
forall a b. (Integral a, Num b) => a -> b
fromIntegral (RTSStats -> RtsTime
gc_cpu_ns RTSStats
stats) Double -> Double -> Double
forall a. Fractional a => a -> a -> a
/ Double
1.0e9) Gauge
gcCpu
        Double -> Gauge -> IO ()
Gauge.set (Word64 -> Double
forall a b. (Integral a, Num b) => a -> b
fromIntegral (RTSStats -> Word64
max_live_bytes RTSStats
stats)) Gauge
maxLive
        Double -> Gauge -> IO ()
Gauge.set (Word32 -> Double
forall a b. (Integral a, Num b) => a -> b
fromIntegral (RTSStats -> Word32
major_gcs RTSStats
stats)) Gauge
majorGcs
 where
  rtsGauge :: Text -> IO Gauge
rtsGauge Text
name = Name -> Labels -> Registry -> IO Gauge
registerGauge (Text -> Name
Name Text
name) Labels
forall a. Monoid a => a
mempty Registry
registry

-- | Register all relevant metrics.
-- Returns the `Registry` which is needed to `serveMetrics` or any other form of publication
-- of metrics, whether push or pull, and a function for updating metrics given some trace event.
prepareRegistry :: forall m tx. (MonadIO m, MonadMonotonicTime m, MonadTime m, IsTx tx, MonadLabelledSTM m) => m (HydraLog tx -> m (), Registry)
prepareRegistry :: forall (m :: * -> *) tx.
(MonadIO m, MonadMonotonicTime m, MonadTime m, IsTx tx,
 MonadLabelledSTM m) =>
m (HydraLog tx -> m (), Registry)
prepareRegistry = do
  TVar m (Map (TxIdType tx) Time)
transactionsMap <- String
-> Map (TxIdType tx) Time -> m (TVar m (Map (TxIdType tx) Time))
forall (m :: * -> *) a.
MonadLabelledSTM m =>
String -> a -> m (TVar m a)
newLabelledTVarIO String
"monitoring-txs-map-registry" Map (TxIdType tx) Time
forall a. Monoid a => a
mempty
  TVar m (Map SnapshotNumber (Time, [TxIdType tx]))
snapshotsMap <- String
-> Map SnapshotNumber (Time, [TxIdType tx])
-> m (TVar m (Map SnapshotNumber (Time, [TxIdType tx])))
forall (m :: * -> *) a.
MonadLabelledSTM m =>
String -> a -> m (TVar m a)
newLabelledTVarIO String
"monitoring-snapshots-map-registry" Map SnapshotNumber (Time, [TxIdType tx])
forall a. Monoid a => a
mempty
  Registry
registry <- IO Registry -> m Registry
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO IO Registry
new
  Metrics
metrics <- Registry -> m Metrics
forall (m :: * -> *). MonadIO m => Registry -> m Metrics
registerMetrics Registry
registry
  (HydraLog tx -> m (), Registry)
-> m (HydraLog tx -> m (), Registry)
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (TVar m (Map (TxIdType tx) Time)
-> TVar m (Map SnapshotNumber (Time, [TxIdType tx]))
-> Metrics
-> HydraLog tx
-> m ()
forall (m :: * -> *) tx.
(MonadIO m, MonadSTM m, MonadMonotonicTime m, MonadTime m,
 IsTx tx) =>
TVar m (Map (TxIdType tx) Time)
-> TVar m (Map SnapshotNumber (Time, [TxIdType tx]))
-> Metrics
-> HydraLog tx
-> m ()
monitor TVar m (Map (TxIdType tx) Time)
transactionsMap TVar m (Map SnapshotNumber (Time, [TxIdType tx]))
snapshotsMap Metrics
metrics, Registry
registry)

registerMetrics :: MonadIO m => Registry -> m Metrics
registerMetrics :: forall (m :: * -> *). MonadIO m => Registry -> m Metrics
registerMetrics Registry
registry = IO Metrics -> m Metrics
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO Metrics -> m Metrics) -> IO Metrics -> m Metrics
forall a b. (a -> b) -> a -> b
$ do
  Counter
headInputs <- Text -> IO Counter
counter Text
"hydra_head_inputs"
  Counter
headRequestedTx <- Text -> IO Counter
counter Text
"hydra_head_requested_tx"
  Counter
headConfirmedTx <- Text -> IO Counter
counter Text
"hydra_head_confirmed_tx"
  Histogram
txConfirmationTime <-
    Text -> [Double] -> IO Histogram
histogram Text
"hydra_head_tx_confirmation_time_ms" [Double
5, Double
10, Double
50, Double
100, Double
1000]
  Histogram
snapshotConfirmationTime <-
    Text -> [Double] -> IO Histogram
histogram Text
"hydra_head_snapshot_confirmation_time_ms" [Double
5, Double
10, Double
50, Double
100, Double
500, Double
1000, Double
5000, Double
10000, Double
30000]
  Gauge
peersConnected <- Text -> IO Gauge
gaugeMetric Text
"hydra_head_peers_connected"
  Gauge
chainDriftSeconds <- Text -> IO Gauge
gaugeMetric Text
"hydra_chain_drift_seconds"
  Gauge
chainLastBlockTimestampSeconds <- Text -> IO Gauge
gaugeMetric Text
"hydra_chain_last_block_timestamp_seconds"
  Metrics -> IO Metrics
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Metrics{Counter
Gauge
Histogram
$sel:headInputs:Metrics :: Counter
$sel:headRequestedTx:Metrics :: Counter
$sel:headConfirmedTx:Metrics :: Counter
$sel:txConfirmationTime:Metrics :: Histogram
$sel:snapshotConfirmationTime:Metrics :: Histogram
$sel:peersConnected:Metrics :: Gauge
$sel:chainDriftSeconds:Metrics :: Gauge
$sel:chainLastBlockTimestampSeconds:Metrics :: Gauge
headInputs :: Counter
headRequestedTx :: Counter
headConfirmedTx :: Counter
txConfirmationTime :: Histogram
snapshotConfirmationTime :: Histogram
peersConnected :: Gauge
chainDriftSeconds :: Gauge
chainLastBlockTimestampSeconds :: Gauge
..}
 where
  counter :: Text -> IO Counter
counter Text
name = Name -> Labels -> Registry -> IO Counter
registerCounter (Text -> Name
Name Text
name) Labels
forall a. Monoid a => a
mempty Registry
registry
  gaugeMetric :: Text -> IO Gauge
gaugeMetric Text
name = Name -> Labels -> Registry -> IO Gauge
registerGauge (Text -> Name
Name Text
name) Labels
forall a. Monoid a => a
mempty Registry
registry
  histogram :: Text -> [Double] -> IO Histogram
histogram Text
name [Double]
buckets = Name -> Labels -> [Double] -> Registry -> IO Histogram
registerHistogram (Text -> Name
Name Text
name) Labels
forall a. Monoid a => a
mempty [Double]
buckets Registry
registry

-- | Main monitoring function that updates metrics store given some log entries.
monitor ::
  forall m tx.
  (MonadIO m, MonadSTM m, MonadMonotonicTime m, MonadTime m, IsTx tx) =>
  TVar m (Map (TxIdType tx) Time) ->
  TVar m (Map SnapshotNumber (Time, [TxIdType tx])) ->
  Metrics ->
  HydraLog tx ->
  m ()
monitor :: forall (m :: * -> *) tx.
(MonadIO m, MonadSTM m, MonadMonotonicTime m, MonadTime m,
 IsTx tx) =>
TVar m (Map (TxIdType tx) Time)
-> TVar m (Map SnapshotNumber (Time, [TxIdType tx]))
-> Metrics
-> HydraLog tx
-> m ()
monitor TVar m (Map (TxIdType tx) Time)
transactionsMap TVar m (Map SnapshotNumber (Time, [TxIdType tx]))
snapshotsMap Metrics{Counter
Gauge
Histogram
$sel:headInputs:Metrics :: Metrics -> Counter
$sel:headRequestedTx:Metrics :: Metrics -> Counter
$sel:headConfirmedTx:Metrics :: Metrics -> Counter
$sel:txConfirmationTime:Metrics :: Metrics -> Histogram
$sel:snapshotConfirmationTime:Metrics :: Metrics -> Histogram
$sel:peersConnected:Metrics :: Metrics -> Gauge
$sel:chainDriftSeconds:Metrics :: Metrics -> Gauge
$sel:chainLastBlockTimestampSeconds:Metrics :: Metrics -> Gauge
headInputs :: Counter
headRequestedTx :: Counter
headConfirmedTx :: Counter
txConfirmationTime :: Histogram
snapshotConfirmationTime :: Histogram
peersConnected :: Gauge
chainDriftSeconds :: Gauge
chainLastBlockTimestampSeconds :: Gauge
..} = \case
  (Node BeginInput{$sel:input:BeginInput :: forall tx. HydraNodeLog tx -> Input tx
input = NetworkInput TTL
_ (ReceivedMessage{$sel:msg:ConnectivityEvent :: forall msg. NetworkEvent msg -> msg
msg = ReqTx tx
tx})}) -> do
    Time
t <- m Time
forall (m :: * -> *). MonadMonotonicTime m => m Time
getMonotonicTime
    -- NOTE: If a requested transaction never gets confirmed, it might stick
    -- forever in the transactions map which could lead to unbounded growth and
    -- memory leak. We might want to have a 'cleaner' thread run that will remove
    -- transactions after some timeout expires
    STM m () -> m ()
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM m () -> m ()) -> STM m () -> m ()
forall a b. (a -> b) -> a -> b
$ TVar m (Map (TxIdType tx) Time)
-> (Map (TxIdType tx) Time -> Map (TxIdType tx) Time) -> STM m ()
forall a. TVar m a -> (a -> a) -> STM m ()
forall (m :: * -> *) a.
MonadSTM m =>
TVar m a -> (a -> a) -> STM m ()
modifyTVar' TVar m (Map (TxIdType tx) Time)
transactionsMap (TxIdType tx
-> Time -> Map (TxIdType tx) Time -> Map (TxIdType tx) Time
forall k a. Ord k => k -> a -> Map k a -> Map k a
Map.insert (tx -> TxIdType tx
forall tx. IsTx tx => tx -> TxIdType tx
txId tx
tx) Time
t)
    Counter -> m ()
tick Counter
headRequestedTx
  (Node LogicOutcome{$sel:outcome:BeginInput :: forall tx. HydraNodeLog tx -> Outcome tx
outcome = Continue{[StateChanged tx]
stateChanges :: [StateChanged tx]
$sel:stateChanges:Continue :: forall tx. Outcome tx -> [StateChanged tx]
stateChanges}}) -> do
    [StateChanged tx] -> (StateChanged tx -> m ()) -> m ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ [StateChanged tx]
stateChanges ((StateChanged tx -> m ()) -> m ())
-> (StateChanged tx -> m ()) -> m ()
forall a b. (a -> b) -> a -> b
$ \case
      PeerConnected{} -> (Gauge -> IO ()) -> Gauge -> m ()
gauge Gauge -> IO ()
Gauge.inc Gauge
peersConnected
      PeerDisconnected{} -> (Gauge -> IO ()) -> Gauge -> m ()
gauge Gauge -> IO ()
Gauge.dec Gauge
peersConnected
      NetworkDisconnected{} -> Gauge -> Double -> m ()
gaugeN Gauge
peersConnected Double
0
      -- On every observed tick, report how far behind the chain we are and when
      -- we last heard from the backend. The latter is a wall-clock timestamp, so
      -- monitoring can alert on `time() - hydra_chain_last_block_timestamp_seconds`
      -- and detect a stalled backend even while the drift gauge is frozen (#2749).
      TickObserved{UTCTime
chainTime :: UTCTime
$sel:chainTime:NetworkConnected :: forall tx. StateChanged tx -> UTCTime
chainTime} -> do
        UTCTime
now <- m UTCTime
forall (m :: * -> *). MonadTime m => m UTCTime
getCurrentTime
        Gauge -> Double -> m ()
gaugeN Gauge
chainDriftSeconds (NominalDiffTime -> Double
forall a b. (Real a, Fractional b) => a -> b
realToFrac (UTCTime
now UTCTime -> UTCTime -> NominalDiffTime
`diffUTCTime` UTCTime
chainTime))
        Gauge -> Double -> m ()
gaugeN Gauge
chainLastBlockTimestampSeconds (NominalDiffTime -> Double
forall a b. (Real a, Fractional b) => a -> b
realToFrac (UTCTime -> NominalDiffTime
utcTimeToPOSIXSeconds UTCTime
now))
      SnapshotRequested{$sel:requestedSnapshot:NetworkConnected :: forall tx. StateChanged tx -> Snapshot tx
requestedSnapshot = Snapshot{SnapshotNumber
number :: SnapshotNumber
$sel:number:Snapshot :: forall tx. Snapshot tx -> SnapshotNumber
number, [tx]
confirmed :: [tx]
$sel:confirmed:Snapshot :: forall tx. Snapshot tx -> [tx]
confirmed}} -> do
        Time
t <- m Time
forall (m :: * -> *). MonadMonotonicTime m => m Time
getMonotonicTime
        STM m () -> m ()
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM m () -> m ()) -> STM m () -> m ()
forall a b. (a -> b) -> a -> b
$ TVar m (Map SnapshotNumber (Time, [TxIdType tx]))
-> (Map SnapshotNumber (Time, [TxIdType tx])
    -> Map SnapshotNumber (Time, [TxIdType tx]))
-> STM m ()
forall a. TVar m a -> (a -> a) -> STM m ()
forall (m :: * -> *) a.
MonadSTM m =>
TVar m a -> (a -> a) -> STM m ()
modifyTVar' TVar m (Map SnapshotNumber (Time, [TxIdType tx]))
snapshotsMap (SnapshotNumber
-> (Time, [TxIdType tx])
-> Map SnapshotNumber (Time, [TxIdType tx])
-> Map SnapshotNumber (Time, [TxIdType tx])
forall k a. Ord k => k -> a -> Map k a -> Map k a
Map.insert SnapshotNumber
number (Time
t, tx -> TxIdType tx
forall tx. IsTx tx => tx -> TxIdType tx
txId (tx -> TxIdType tx) -> [tx] -> [TxIdType tx]
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> [tx]
confirmed))
      SnapshotConfirmed{$sel:snapshot:NetworkConnected :: forall tx. StateChanged tx -> Maybe (Snapshot tx)
snapshot = Maybe (Snapshot tx)
mSnapshot} -> do
        Time
t <- m Time
forall (m :: * -> *). MonadMonotonicTime m => m Time
getMonotonicTime
        -- On the normal signing path the event carries no snapshot; the
        -- protocol confirms snapshots strictly in order, so the oldest
        -- in-flight request is the confirmed one.
        Maybe (Time, [TxIdType tx])
mEntry <- STM m (Maybe (Time, [TxIdType tx]))
-> m (Maybe (Time, [TxIdType tx]))
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM m (Maybe (Time, [TxIdType tx]))
 -> m (Maybe (Time, [TxIdType tx])))
-> STM m (Maybe (Time, [TxIdType tx]))
-> m (Maybe (Time, [TxIdType tx]))
forall a b. (a -> b) -> a -> b
$ do
          Map SnapshotNumber (Time, [TxIdType tx])
inFlight <- TVar m (Map SnapshotNumber (Time, [TxIdType tx]))
-> STM m (Map SnapshotNumber (Time, [TxIdType tx]))
forall a. TVar m a -> STM m a
forall (m :: * -> *) a. MonadSTM m => TVar m a -> STM m a
readTVar TVar m (Map SnapshotNumber (Time, [TxIdType tx]))
snapshotsMap
          Maybe (SnapshotNumber, (Time, [TxIdType tx]))
-> ((SnapshotNumber, (Time, [TxIdType tx]))
    -> STM m (Time, [TxIdType tx]))
-> STM m (Maybe (Time, [TxIdType tx]))
forall (t :: * -> *) (m :: * -> *) a b.
(Traversable t, Monad m) =>
t a -> (a -> m b) -> m (t b)
forM (Map SnapshotNumber (Time, [TxIdType tx])
-> Maybe (SnapshotNumber, (Time, [TxIdType tx]))
forall k a. Map k a -> Maybe (k, a)
Map.lookupMin Map SnapshotNumber (Time, [TxIdType tx])
inFlight) (((SnapshotNumber, (Time, [TxIdType tx]))
  -> STM m (Time, [TxIdType tx]))
 -> STM m (Maybe (Time, [TxIdType tx])))
-> ((SnapshotNumber, (Time, [TxIdType tx]))
    -> STM m (Time, [TxIdType tx]))
-> STM m (Maybe (Time, [TxIdType tx]))
forall a b. (a -> b) -> a -> b
$ \(SnapshotNumber
number, (Time, [TxIdType tx])
entry) -> do
            TVar m (Map SnapshotNumber (Time, [TxIdType tx]))
-> Map SnapshotNumber (Time, [TxIdType tx]) -> STM m ()
forall a. TVar m a -> a -> STM m ()
forall (m :: * -> *) a. MonadSTM m => TVar m a -> a -> STM m ()
writeTVar TVar m (Map SnapshotNumber (Time, [TxIdType tx]))
snapshotsMap (SnapshotNumber
-> Map SnapshotNumber (Time, [TxIdType tx])
-> Map SnapshotNumber (Time, [TxIdType tx])
forall k a. Ord k => k -> Map k a -> Map k a
Map.delete SnapshotNumber
number Map SnapshotNumber (Time, [TxIdType tx])
inFlight)
            (Time, [TxIdType tx]) -> STM m (Time, [TxIdType tx])
forall a. a -> STM m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Time, [TxIdType tx])
entry
        [TxIdType tx]
confirmedIds <- case (Maybe (Snapshot tx)
mSnapshot, Maybe (Time, [TxIdType tx])
mEntry) of
          -- Side-load path: no preceding SnapshotRequested, but the event
          -- carries the snapshot.
          (Just Snapshot{[tx]
$sel:confirmed:Snapshot :: forall tx. Snapshot tx -> [tx]
confirmed :: [tx]
confirmed}, Maybe (Time, [TxIdType tx])
_) -> [TxIdType tx] -> m [TxIdType tx]
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (tx -> TxIdType tx
forall tx. IsTx tx => tx -> TxIdType tx
txId (tx -> TxIdType tx) -> [tx] -> [TxIdType tx]
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> [tx]
confirmed)
          (Maybe (Snapshot tx)
Nothing, Just (Time
_, [TxIdType tx]
txIds)) -> [TxIdType tx] -> m [TxIdType tx]
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure [TxIdType tx]
txIds
          (Maybe (Snapshot tx)
Nothing, Maybe (Time, [TxIdType tx])
Nothing) -> [TxIdType tx] -> m [TxIdType tx]
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure []
        Maybe (Time, [TxIdType tx])
-> ((Time, [TxIdType tx]) -> m ()) -> m ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ Maybe (Time, [TxIdType tx])
mEntry (((Time, [TxIdType tx]) -> m ()) -> m ())
-> ((Time, [TxIdType tx]) -> m ()) -> m ()
forall a b. (a -> b) -> a -> b
$ \(Time
start, [TxIdType tx]
_) ->
          Histogram -> DiffTime -> m ()
histo Histogram
snapshotConfirmationTime (Time -> Time -> DiffTime
diffTime Time
t Time
start)
        Counter -> Port -> m ()
tickN Counter
headConfirmedTx ([TxIdType tx] -> Port
forall a. [a] -> Port
forall (t :: * -> *) a. Foldable t => t a -> Port
length [TxIdType tx]
confirmedIds)
        [TxIdType tx] -> (TxIdType tx -> m ()) -> m ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ [TxIdType tx]
confirmedIds ((TxIdType tx -> m ()) -> m ()) -> (TxIdType tx -> m ()) -> m ()
forall a b. (a -> b) -> a -> b
$ \TxIdType tx
i -> do
          Map (TxIdType tx) Time
txsStartTime <- TVar m (Map (TxIdType tx) Time) -> m (Map (TxIdType tx) Time)
forall a. TVar m a -> m a
forall (m :: * -> *) a. MonadSTM m => TVar m a -> m a
readTVarIO TVar m (Map (TxIdType tx) Time)
transactionsMap
          case TxIdType tx -> Map (TxIdType tx) Time -> Maybe Time
forall k a. Ord k => k -> Map k a -> Maybe a
Map.lookup TxIdType tx
i Map (TxIdType tx) Time
txsStartTime of
            Just Time
start -> do
              STM m () -> m ()
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM m () -> m ()) -> STM m () -> m ()
forall a b. (a -> b) -> a -> b
$ TVar m (Map (TxIdType tx) Time)
-> (Map (TxIdType tx) Time -> Map (TxIdType tx) Time) -> STM m ()
forall a. TVar m a -> (a -> a) -> STM m ()
forall (m :: * -> *) a.
MonadSTM m =>
TVar m a -> (a -> a) -> STM m ()
modifyTVar' TVar m (Map (TxIdType tx) Time)
transactionsMap ((Map (TxIdType tx) Time -> Map (TxIdType tx) Time) -> STM m ())
-> (Map (TxIdType tx) Time -> Map (TxIdType tx) Time) -> STM m ()
forall a b. (a -> b) -> a -> b
$ TxIdType tx -> Map (TxIdType tx) Time -> Map (TxIdType tx) Time
forall k a. Ord k => k -> Map k a -> Map k a
Map.delete TxIdType tx
i
              Histogram -> DiffTime -> m ()
histo Histogram
txConfirmationTime (Time -> Time -> DiffTime
diffTime Time
t Time
start)
            Maybe Time
Nothing -> () -> m ()
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
      StateChanged tx
_ -> () -> m ()
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
  (Node (EndInput Party
_ Word64
_)) ->
    Counter -> m ()
tick Counter
headInputs
  HydraLog tx
_ -> () -> m ()
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
 where
  gaugeN :: Gauge -> Double -> m ()
  gaugeN :: Gauge -> Double -> m ()
gaugeN Gauge
g Double
num = IO () -> m ()
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO () -> m ()) -> IO () -> m ()
forall a b. (a -> b) -> a -> b
$ Double -> Gauge -> IO ()
Gauge.set Double
num Gauge
g

  gauge :: (Gauge -> IO ()) -> Gauge -> m ()
  gauge :: (Gauge -> IO ()) -> Gauge -> m ()
gauge Gauge -> IO ()
f Gauge
g = IO () -> m ()
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO () -> m ()) -> IO () -> m ()
forall a b. (a -> b) -> a -> b
$ Gauge -> IO ()
f Gauge
g

  tick :: Counter -> m ()
  tick :: Counter -> m ()
tick Counter
c = IO () -> m ()
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO () -> m ()) -> IO () -> m ()
forall a b. (a -> b) -> a -> b
$ Counter -> IO ()
inc Counter
c

  tickN :: Counter -> Int -> m ()
  tickN :: Counter -> Port -> m ()
tickN Counter
c Port
num = IO () -> m ()
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO () -> m ()) -> IO () -> m ()
forall a b. (a -> b) -> a -> b
$ Port -> Counter -> IO ()
add Port
num Counter
c

  histo :: Histogram -> DiffTime -> m ()
  histo :: Histogram -> DiffTime -> m ()
histo Histogram
h DiffTime
time = IO () -> m ()
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO () -> m ()) -> IO () -> m ()
forall a b. (a -> b) -> a -> b
$ Double -> Histogram -> IO ()
observe (Rational -> Double
forall a. Fractional a => Rational -> a
fromRational (Rational -> Double) -> Rational -> Double
forall a b. (a -> b) -> a -> b
$ DiffTime -> Rational
forall a. Real a => a -> Rational
toRational (DiffTime -> Rational) -> DiffTime -> Rational
forall a b. (a -> b) -> a -> b
$ DiffTime
time DiffTime -> DiffTime -> DiffTime
forall a. Num a => a -> a -> a
* DiffTime
1000) Histogram
h