-- | Bounded, process-local timing state. Replayed events seed identities, never
-- invent start times or increment process counters.
module Hydra.Logging.Monitoring.Accounting where

import Hydra.Prelude hiding (state)

import Data.Map.Strict qualified as Map

data SnapshotKey head = SnapshotKey
  { forall head. SnapshotKey head -> head
headIdentity :: head
  , forall head. SnapshotKey head -> Natural
version :: Natural
  , forall head. SnapshotKey head -> Natural
number :: Natural
  }
  deriving stock (SnapshotKey head -> SnapshotKey head -> Bool
(SnapshotKey head -> SnapshotKey head -> Bool)
-> (SnapshotKey head -> SnapshotKey head -> Bool)
-> Eq (SnapshotKey head)
forall head.
Eq head =>
SnapshotKey head -> SnapshotKey head -> Bool
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: forall head.
Eq head =>
SnapshotKey head -> SnapshotKey head -> Bool
== :: SnapshotKey head -> SnapshotKey head -> Bool
$c/= :: forall head.
Eq head =>
SnapshotKey head -> SnapshotKey head -> Bool
/= :: SnapshotKey head -> SnapshotKey head -> Bool
Eq, Int -> SnapshotKey head -> ShowS
[SnapshotKey head] -> ShowS
SnapshotKey head -> String
(Int -> SnapshotKey head -> ShowS)
-> (SnapshotKey head -> String)
-> ([SnapshotKey head] -> ShowS)
-> Show (SnapshotKey head)
forall head. Show head => Int -> SnapshotKey head -> ShowS
forall head. Show head => [SnapshotKey head] -> ShowS
forall head. Show head => SnapshotKey head -> String
forall a.
(Int -> a -> ShowS) -> (a -> String) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: forall head. Show head => Int -> SnapshotKey head -> ShowS
showsPrec :: Int -> SnapshotKey head -> ShowS
$cshow :: forall head. Show head => SnapshotKey head -> String
show :: SnapshotKey head -> String
$cshowList :: forall head. Show head => [SnapshotKey head] -> ShowS
showList :: [SnapshotKey head] -> ShowS
Show)

data Accounting head txid = Accounting
  { forall head txid. Accounting head txid -> Map txid Double
timers :: Map txid Double
  , forall head txid. Accounting head txid -> Map (Double, txid) ()
orderedTimers :: Map (Double, txid) ()
  , forall head txid. Accounting head txid -> Maybe Word64
latestInput :: Maybe Word64
  , forall head txid. Accounting head txid -> Maybe (SnapshotKey head)
confirmed :: Maybe (SnapshotKey head)
  , forall head txid.
Accounting head txid
-> Maybe (SnapshotKey head, Maybe Double, [txid])
inFlight :: Maybe (SnapshotKey head, Maybe Double, [txid])
  }
  deriving stock (Accounting head txid -> Accounting head txid -> Bool
(Accounting head txid -> Accounting head txid -> Bool)
-> (Accounting head txid -> Accounting head txid -> Bool)
-> Eq (Accounting head txid)
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
forall head txid.
(Eq txid, Eq head) =>
Accounting head txid -> Accounting head txid -> Bool
$c== :: forall head txid.
(Eq txid, Eq head) =>
Accounting head txid -> Accounting head txid -> Bool
== :: Accounting head txid -> Accounting head txid -> Bool
$c/= :: forall head txid.
(Eq txid, Eq head) =>
Accounting head txid -> Accounting head txid -> Bool
/= :: Accounting head txid -> Accounting head txid -> Bool
Eq, Int -> Accounting head txid -> ShowS
[Accounting head txid] -> ShowS
Accounting head txid -> String
(Int -> Accounting head txid -> ShowS)
-> (Accounting head txid -> String)
-> ([Accounting head txid] -> ShowS)
-> Show (Accounting head txid)
forall a.
(Int -> a -> ShowS) -> (a -> String) -> ([a] -> ShowS) -> Show a
forall head txid.
(Show txid, Show head) =>
Int -> Accounting head txid -> ShowS
forall head txid.
(Show txid, Show head) =>
[Accounting head txid] -> ShowS
forall head txid.
(Show txid, Show head) =>
Accounting head txid -> String
$cshowsPrec :: forall head txid.
(Show txid, Show head) =>
Int -> Accounting head txid -> ShowS
showsPrec :: Int -> Accounting head txid -> ShowS
$cshow :: forall head txid.
(Show txid, Show head) =>
Accounting head txid -> String
show :: Accounting head txid -> String
$cshowList :: forall head txid.
(Show txid, Show head) =>
[Accounting head txid] -> ShowS
showList :: [Accounting head txid] -> ShowS
Show)

empty :: Accounting head txid
empty :: forall head txid. Accounting head txid
empty = Map txid Double
-> Map (Double, txid) ()
-> Maybe Word64
-> Maybe (SnapshotKey head)
-> Maybe (SnapshotKey head, Maybe Double, [txid])
-> Accounting head txid
forall head txid.
Map txid Double
-> Map (Double, txid) ()
-> Maybe Word64
-> Maybe (SnapshotKey head)
-> Maybe (SnapshotKey head, Maybe Double, [txid])
-> Accounting head txid
Accounting Map txid Double
forall k a. Map k a
Map.empty Map (Double, txid) ()
forall k a. Map k a
Map.empty Maybe Word64
forall a. Maybe a
Nothing Maybe (SnapshotKey head)
forall a. Maybe a
Nothing Maybe (SnapshotKey head, Maybe Double, [txid])
forall a. Maybe a
Nothing

-- | Admission order is monotone. Retries and parked inputs retain their ID,
-- so one high-water mark distinguishes first processing without a set of IDs.
request :: Ord txid => Int -> Double -> Double -> Word64 -> txid -> Accounting head txid -> (Bool, Int, Accounting head txid)
request :: forall txid head.
Ord txid =>
Int
-> Double
-> Double
-> Word64
-> txid
-> Accounting head txid
-> (Bool, Int, Accounting head txid)
request Int
capacity Double
lifetime Double
now Word64
inputId txid
txid Accounting head txid
state =
  let fresh :: Bool
fresh = Bool -> (Word64 -> Bool) -> Maybe Word64 -> Bool
forall b a. b -> (a -> b) -> Maybe a -> b
maybe Bool
True (Word64
inputId >) (Accounting head txid -> Maybe Word64
forall head txid. Accounting head txid -> Maybe Word64
latestInput Accounting head txid
state)
      (Int
expired, Accounting head txid
pruned) = Int
-> Double
-> Double
-> Accounting head txid
-> (Int, Accounting head txid)
forall txid head.
Ord txid =>
Int
-> Double
-> Double
-> Accounting head txid
-> (Int, Accounting head txid)
prune Int
capacity Double
lifetime Double
now Accounting head txid
state
      inserted :: Accounting head txid
inserted =
        if Bool
fresh Bool -> Bool -> Bool
&& txid -> Map txid Double -> Bool
forall k a. Ord k => k -> Map k a -> Bool
Map.notMember txid
txid (Accounting head txid -> Map txid Double
forall head txid. Accounting head txid -> Map txid Double
timers Accounting head txid
pruned)
          then Accounting head txid
pruned{timers = Map.insert txid now (timers pruned), orderedTimers = Map.insert (now, txid) () (orderedTimers pruned)}
          else Accounting head txid
pruned
      (Int
overflow, Accounting head txid
bounded) = Int
-> Double
-> Double
-> Accounting head txid
-> (Int, Accounting head txid)
forall txid head.
Ord txid =>
Int
-> Double
-> Double
-> Accounting head txid
-> (Int, Accounting head txid)
prune Int
capacity Double
lifetime Double
now Accounting head txid
inserted
   in (Bool
fresh, Int
expired Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Int
overflow, Accounting head txid
bounded{latestInput = max (Just inputId) (latestInput bounded)})

prune :: Ord txid => Int -> Double -> Double -> Accounting head txid -> (Int, Accounting head txid)
prune :: forall txid head.
Ord txid =>
Int
-> Double
-> Double
-> Accounting head txid
-> (Int, Accounting head txid)
prune Int
capacity Double
lifetime Double
now = Int -> Accounting head txid -> (Int, Accounting head txid)
go Int
0
 where
  go :: Int -> Accounting head txid -> (Int, Accounting head txid)
go Int
removed Accounting head txid
state = case Map (Double, txid) () -> Maybe ((Double, txid), ())
forall k a. Map k a -> Maybe (k, a)
Map.lookupMin (Accounting head txid -> Map (Double, txid) ()
forall head txid. Accounting head txid -> Map (Double, txid) ()
orderedTimers Accounting head txid
state) of
    Just ((Double
start, txid
txid), ())
      | Map txid Double -> Int
forall k a. Map k a -> Int
Map.size (Accounting head txid -> Map txid Double
forall head txid. Accounting head txid -> Map txid Double
timers Accounting head txid
state) Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
> Int
capacity Bool -> Bool -> Bool
|| Double
now Double -> Double -> Double
forall a. Num a => a -> a -> a
- Double
start Double -> Double -> Bool
forall a. Ord a => a -> a -> Bool
>= Double
lifetime ->
          Int -> Accounting head txid -> (Int, Accounting head txid)
go (Int
removed Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Int
1) (txid -> Accounting head txid -> Accounting head txid
forall txid head.
Ord txid =>
txid -> Accounting head txid -> Accounting head txid
forget txid
txid Accounting head txid
state)
    Maybe ((Double, txid), ())
_ -> (Int
removed, Accounting head txid
state)

forget :: Ord txid => txid -> Accounting head txid -> Accounting head txid
forget :: forall txid head.
Ord txid =>
txid -> Accounting head txid -> Accounting head txid
forget txid
txid state :: Accounting head txid
state@Accounting{Map txid Double
$sel:timers:Accounting :: forall head txid. Accounting head txid -> Map txid Double
timers :: Map txid Double
timers, Map (Double, txid) ()
$sel:orderedTimers:Accounting :: forall head txid. Accounting head txid -> Map (Double, txid) ()
orderedTimers :: Map (Double, txid) ()
orderedTimers} =
  Accounting head txid
state
    { timers = Map.delete txid timers
    , orderedTimers = maybe orderedTimers (\Double
start -> (Double, txid) -> Map (Double, txid) () -> Map (Double, txid) ()
forall k a. Ord k => k -> Map k a -> Map k a
Map.delete (Double
start, txid
txid) Map (Double, txid) ()
orderedTimers) (Map.lookup txid timers)
    }

clearPending :: Accounting head txid -> Accounting head txid
clearPending :: forall head txid. Accounting head txid -> Accounting head txid
clearPending Accounting head txid
state = Accounting head txid
state{timers = Map.empty, orderedTimers = Map.empty, inFlight = Nothing}

startSnapshot :: Eq head => SnapshotKey head -> Maybe Double -> [txid] -> Accounting head txid -> Accounting head txid
startSnapshot :: forall head txid.
Eq head =>
SnapshotKey head
-> Maybe Double
-> [txid]
-> Accounting head txid
-> Accounting head txid
startSnapshot SnapshotKey head
key Maybe Double
start [txid]
txids Accounting head txid
state = case Accounting head txid
-> Maybe (SnapshotKey head, Maybe Double, [txid])
forall head txid.
Accounting head txid
-> Maybe (SnapshotKey head, Maybe Double, [txid])
inFlight Accounting head txid
state of
  Just (SnapshotKey head
existing, Maybe Double
_, [txid]
_) | SnapshotKey head
existing SnapshotKey head -> SnapshotKey head -> Bool
forall a. Eq a => a -> a -> Bool
== SnapshotKey head
key -> Accounting head txid
state
  Maybe (SnapshotKey head, Maybe Double, [txid])
_ -> Accounting head txid
state{inFlight = Just (key, start, txids)}

-- | Returns transactions and durations only when the confirmed number advances
-- within this head. A version bump at the same number does not confirm the
-- transactions again. An unrelated adoption cannot consume a pending timer.
confirmSnapshot :: (Eq head, Ord txid) => Double -> SnapshotKey head -> [txid] -> Accounting head txid -> (Bool, Maybe Double, [Double], Accounting head txid)
confirmSnapshot :: forall head txid.
(Eq head, Ord txid) =>
Double
-> SnapshotKey head
-> [txid]
-> Accounting head txid
-> (Bool, Maybe Double, [Double], Accounting head txid)
confirmSnapshot Double
now SnapshotKey head
key [txid]
txids Accounting head txid
state =
  let advances :: Bool
advances = case Accounting head txid -> Maybe (SnapshotKey head)
forall head txid. Accounting head txid -> Maybe (SnapshotKey head)
confirmed Accounting head txid
state of
        Maybe (SnapshotKey head)
Nothing -> Bool
True
        Just SnapshotKey head
old -> SnapshotKey head -> head
forall head. SnapshotKey head -> head
headIdentity SnapshotKey head
old head -> head -> Bool
forall a. Eq a => a -> a -> Bool
/= SnapshotKey head -> head
forall head. SnapshotKey head -> head
headIdentity SnapshotKey head
key Bool -> Bool -> Bool
|| SnapshotKey head -> Natural
forall head. SnapshotKey head -> Natural
number SnapshotKey head
key Natural -> Natural -> Bool
forall a. Ord a => a -> a -> Bool
> SnapshotKey head -> Natural
forall head. SnapshotKey head -> Natural
number SnapshotKey head
old
      matching :: Maybe (Maybe Double)
matching = case Accounting head txid
-> Maybe (SnapshotKey head, Maybe Double, [txid])
forall head txid.
Accounting head txid
-> Maybe (SnapshotKey head, Maybe Double, [txid])
inFlight Accounting head txid
state of
        Just (SnapshotKey head
pending, Maybe Double
start, [txid]
_) | SnapshotKey head
pending SnapshotKey head -> SnapshotKey head -> Bool
forall a. Eq a => a -> a -> Bool
== SnapshotKey head
key -> Maybe Double -> Maybe (Maybe Double)
forall a. a -> Maybe a
Just Maybe Double
start
        Maybe (SnapshotKey head, Maybe Double, [txid])
_ -> Maybe (Maybe Double)
forall a. Maybe a
Nothing
      duration :: Maybe Double
duration = if Bool
advances then (Double -> Double) -> Maybe Double -> Maybe Double
forall a b. (a -> b) -> Maybe a -> Maybe b
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
fmap (Double -> Double -> Double
forall a. Ord a => a -> a -> a
max Double
0 (Double -> Double) -> (Double -> Double) -> Double -> Double
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (Double
now -)) (Maybe (Maybe Double) -> Maybe Double
forall (m :: * -> *) a. Monad m => m (m a) -> m a
join Maybe (Maybe Double)
matching) else Maybe Double
forall a. Maybe a
Nothing
      latencies :: [Double]
latencies = if Bool
advances then (txid -> Maybe Double) -> [txid] -> [Double]
forall a b. (a -> Maybe b) -> [a] -> [b]
mapMaybe (\txid
txid -> Double -> Double -> Double
forall a. Ord a => a -> a -> a
max Double
0 (Double -> Double) -> (Double -> Double) -> Double -> Double
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (Double
now -) (Double -> Double) -> Maybe Double -> Maybe Double
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> txid -> Map txid Double -> Maybe Double
forall k a. Ord k => k -> Map k a -> Maybe a
Map.lookup txid
txid (Accounting head txid -> Map txid Double
forall head txid. Accounting head txid -> Map txid Double
timers Accounting head txid
state)) [txid]
txids else []
      next :: Accounting head txid
next = (Accounting head txid -> txid -> Accounting head txid)
-> Accounting head txid -> [txid] -> Accounting head txid
forall b a. (b -> a -> b) -> b -> [a] -> b
forall (t :: * -> *) b a.
Foldable t =>
(b -> a -> b) -> b -> t a -> b
foldl' ((txid -> Accounting head txid -> Accounting head txid)
-> Accounting head txid -> txid -> Accounting head txid
forall a b c. (a -> b -> c) -> b -> a -> c
flip txid -> Accounting head txid -> Accounting head txid
forall txid head.
Ord txid =>
txid -> Accounting head txid -> Accounting head txid
forget) Accounting head txid
state [txid]
txids
      watermark :: Maybe (SnapshotKey head)
watermark = case Accounting head txid -> Maybe (SnapshotKey head)
forall head txid. Accounting head txid -> Maybe (SnapshotKey head)
confirmed Accounting head txid
state of
        Just SnapshotKey head
old | SnapshotKey head -> head
forall head. SnapshotKey head -> head
headIdentity SnapshotKey head
old head -> head -> Bool
forall a. Eq a => a -> a -> Bool
== SnapshotKey head -> head
forall head. SnapshotKey head -> head
headIdentity SnapshotKey head
key Bool -> Bool -> Bool
&& (SnapshotKey head -> Natural
forall head. SnapshotKey head -> Natural
number SnapshotKey head
old, SnapshotKey head -> Natural
forall head. SnapshotKey head -> Natural
version SnapshotKey head
old) (Natural, Natural) -> (Natural, Natural) -> Bool
forall a. Ord a => a -> a -> Bool
> (SnapshotKey head -> Natural
forall head. SnapshotKey head -> Natural
number SnapshotKey head
key, SnapshotKey head -> Natural
forall head. SnapshotKey head -> Natural
version SnapshotKey head
key) -> SnapshotKey head -> Maybe (SnapshotKey head)
forall a. a -> Maybe a
Just SnapshotKey head
old
        Maybe (SnapshotKey head)
_ -> SnapshotKey head -> Maybe (SnapshotKey head)
forall a. a -> Maybe a
Just SnapshotKey head
key
   in (Bool
advances, Maybe Double
duration, [Double]
latencies, Accounting head txid
next{confirmed = watermark, inFlight = if isJust matching then Nothing else inFlight state})