-- SPDX-FileCopyrightText: 2026 Alexandra de Wit
--
-- SPDX-License-Identifier: MIT

{- | Holding every received receipt for as long as the worker needs it.

A backend hides a delivery for one window ("Ecluse.Core.Queue.Lease"). A batch runs
sequentially, so a receipt waiting its turn and a receipt whose job runs long both outlive that
window and would be redelivered to a second consumer. The controller renews each one from
receipt until its disposition, and a renewal and a disposition never overlap on the same
receipt. A receipt whose lease cannot be kept is dropped alone and left unacknowledged.
-}
module Ecluse.Core.Worker.Lease (
    -- * The controller
    LeasedReceipt,
    leasedMessage,
    withLeasedBatch,
    whileLeased,
    disposing,

    -- * The transport, clock, and pacing it runs on
    LeaseOps (..),
    queueLeaseOps,

    -- * Renewal arithmetic
    leaseRenewAt,
    leaseRetryUntil,
    leaseRequest,
) where

import Control.Concurrent.STM (check, orElse)
import Control.Retry (retrying)
import Katip (KatipContext, Severity (WarningS), logFM, ls)
import UnliftIO (MonadUnliftIO)
import UnliftIO.Async (waitSTM, withAsync)
import UnliftIO.Exception (finally, tryAny)
import UnliftIO.MVar qualified as MVar

import Ecluse.Core.Clock (waitUntilMonotonic)
import Ecluse.Core.Fault (TransportCause (TransportProtocol), TransportFault, tfCause, tfDetail, transportFault, transportRetryable)
import Ecluse.Core.Queue (MirrorQueue (extendVisibility), QueueMessage (msgLease, msgReceipt), ReceiptHandle)
import Ecluse.Core.Queue.Lease (
    MonoTime,
    ReceiptLease (rlCeilingAt, rlExpiresAt, rlWindow),
    Seconds (Seconds),
    monoAfter,
    monoSecondsBetween,
    monotonicNow,
 )
import Ecluse.Core.Supervision (delayListPolicy)
import Ecluse.Core.Text (displayExceptionT)

{- | The transport, clock, and pacing a lease runs on, injected so a test drives renewal on a
clock it controls rather than on real time.
-}
data LeaseOps = LeaseOps
    { LeaseOps
-> ReceiptHandle -> Seconds -> IO (Either TransportFault ())
loRenew :: ReceiptHandle -> Seconds -> IO (Either TransportFault ())
    -- ^ Reset one receipt's visibility window to the given duration.
    , LeaseOps -> IO MonoTime
loNow :: IO MonoTime
    -- ^ The monotonic clock every lease deadline is measured on.
    , LeaseOps -> MonoTime -> IO ()
loWaitUntil :: MonoTime -> IO ()
    {- ^ Wait until the given instant. It takes the instant, not a duration, so several waits
    running at once cannot each push a shared test clock on by their own full pause.
    -}
    , LeaseOps -> [Int]
loRetryDelays :: [Int]
    {- ^ The pacing between renewal attempts, in microseconds. Its length is the retry budget,
    and a retry stops early once the receipt's own margin runs out.
    -}
    }

-- | The lease transport over a live queue handle, at the shipped clock and pacing.
queueLeaseOps :: MirrorQueue -> LeaseOps
queueLeaseOps :: MirrorQueue -> LeaseOps
queueLeaseOps MirrorQueue
queue =
    LeaseOps
        { loRenew :: ReceiptHandle -> Seconds -> IO (Either TransportFault ())
loRenew = MirrorQueue
-> ReceiptHandle -> Seconds -> IO (Either TransportFault ())
extendVisibility MirrorQueue
queue
        , loNow :: IO MonoTime
loNow = IO MonoTime
monotonicNow
        , loWaitUntil :: MonoTime -> IO ()
loWaitUntil = MonoTime -> IO ()
waitUntilMonotonic
        , loRetryDelays :: [Int]
loRetryDelays = [Int]
leaseRetryDelays
        }

{- | The renewal retry pacing: three further attempts inside about two seconds. Each delay is
spent after the margin check, so the budget fits the fixed thirty-second SQS window alone.
-}
leaseRetryDelays :: [Int]
leaseRetryDelays :: [Int]
leaseRetryDelays = [Int
200_000, Int
500_000, Int
1_000_000]

{- | One received receipt and the lease held over it. A renewal task runs per leased receipt,
so a batch runs at most ten of them beside its single artifact task.
-}
data LeasedReceipt = LeasedReceipt
    { LeasedReceipt -> QueueMessage
lrMessage :: QueueMessage
    , -- When the current window lapses, or Nothing once the receipt is disposed or dropped.
      -- Taking it is what stops a renewal and a disposition from overlapping.
      LeasedReceipt -> MVar (Maybe MonoTime)
lrHeld :: MVar (Maybe MonoTime)
    , -- Set once renewal has given up, so the receipt's job is cancelled or skipped.
      LeasedReceipt -> TVar Bool
lrDropped :: TVar Bool
    }

-- | The message this receipt delivered.
leasedMessage :: LeasedReceipt -> QueueMessage
leasedMessage :: LeasedReceipt -> QueueMessage
leasedMessage = LeasedReceipt -> QueueMessage
lrMessage

{- | Lease every receipt in a batch for the body's whole run, renewing each continually.
Leaving the body cancels every renewal, so an unfinished receipt stays unacknowledged.
-}
withLeasedBatch :: (MonadUnliftIO m, KatipContext m) => LeaseOps -> [QueueMessage] -> ([LeasedReceipt] -> m a) -> m a
withLeasedBatch :: forall (m :: * -> *) a.
(MonadUnliftIO m, KatipContext m) =>
LeaseOps -> [QueueMessage] -> ([LeasedReceipt] -> m a) -> m a
withLeasedBatch LeaseOps
ops [QueueMessage]
messages [LeasedReceipt] -> m a
body = do
    leased <- (QueueMessage -> m LeasedReceipt)
-> [QueueMessage] -> m [LeasedReceipt]
forall (t :: * -> *) (f :: * -> *) a b.
(Traversable t, Applicative f) =>
(a -> f b) -> t a -> f (t b)
forall (f :: * -> *) a b.
Applicative f =>
(a -> f b) -> [a] -> f [b]
traverse QueueMessage -> m LeasedReceipt
forall (m :: * -> *). MonadIO m => QueueMessage -> m LeasedReceipt
newLeasedReceipt [QueueMessage]
messages
    withRenewals ops leased (body leased)

newLeasedReceipt :: (MonadIO m) => QueueMessage -> m LeasedReceipt
newLeasedReceipt :: forall (m :: * -> *). MonadIO m => QueueMessage -> m LeasedReceipt
newLeasedReceipt QueueMessage
message = do
    held <- Maybe MonoTime -> m (MVar (Maybe MonoTime))
forall (m :: * -> *) a. MonadIO m => a -> m (MVar a)
MVar.newMVar (ReceiptLease -> MonoTime
rlExpiresAt (ReceiptLease -> MonoTime) -> Maybe ReceiptLease -> Maybe MonoTime
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> QueueMessage -> Maybe ReceiptLease
msgLease QueueMessage
message)
    dropped <- newTVarIO False
    pure LeasedReceipt{lrMessage = message, lrHeld = held, lrDropped = dropped}

{- Nest one renewal task per receipt that carries a lease, so leaving the scope cancels every
one of them. A backend that never expires a delivery grants no lease and gets no task. -}
withRenewals :: (MonadUnliftIO m, KatipContext m) => LeaseOps -> [LeasedReceipt] -> m a -> m a
withRenewals :: forall (m :: * -> *) a.
(MonadUnliftIO m, KatipContext m) =>
LeaseOps -> [LeasedReceipt] -> m a -> m a
withRenewals LeaseOps
ops [LeasedReceipt]
leased m a
inner = (LeasedReceipt -> m a -> m a) -> m a -> [LeasedReceipt] -> m a
forall a b. (a -> b -> b) -> b -> [a] -> b
forall (t :: * -> *) a b.
Foldable t =>
(a -> b -> b) -> b -> t a -> b
foldr LeasedReceipt -> m a -> m a
forall {m :: * -> *} {b}.
(MonadUnliftIO m, KatipContext m) =>
LeasedReceipt -> m b -> m b
renewing m a
inner [LeasedReceipt]
leased
  where
    renewing :: LeasedReceipt -> m b -> m b
renewing LeasedReceipt
entry m b
rest = case QueueMessage -> Maybe ReceiptLease
msgLease (LeasedReceipt -> QueueMessage
lrMessage LeasedReceipt
entry) of
        Maybe ReceiptLease
Nothing -> m b
rest
        Just ReceiptLease
lease -> m () -> (Async () -> m b) -> m b
forall (m :: * -> *) a b.
MonadUnliftIO m =>
m a -> (Async a -> m b) -> m b
withAsync (LeaseOps -> ReceiptLease -> LeasedReceipt -> m ()
forall (m :: * -> *).
(MonadUnliftIO m, KatipContext m) =>
LeaseOps -> ReceiptLease -> LeasedReceipt -> m ()
renewalLoop LeaseOps
ops ReceiptLease
lease LeasedReceipt
entry) (m b -> Async () -> m b
forall a b. a -> b -> a
const m b
rest)

{- | Run this receipt's work while its lease holds, cancelling the work the moment the lease is
dropped. 'Nothing' says the receipt was dropped, so nothing was decided for it.
-}
whileLeased :: (MonadUnliftIO m) => LeasedReceipt -> m a -> m (Maybe a)
whileLeased :: forall (m :: * -> *) a.
MonadUnliftIO m =>
LeasedReceipt -> m a -> m (Maybe a)
whileLeased LeasedReceipt
leased m a
work = do
    dropped <- TVar Bool -> m Bool
forall (m :: * -> *) a. MonadIO m => TVar a -> m a
readTVarIO (LeasedReceipt -> TVar Bool
lrDropped LeasedReceipt
leased)
    if dropped
        then pure Nothing
        else withAsync work $ \Async a
running ->
            STM (Maybe a) -> m (Maybe a)
forall (m :: * -> *) a. MonadIO m => STM a -> m a
atomically ((a -> Maybe a
forall a. a -> Maybe a
Just (a -> Maybe a) -> STM a -> STM (Maybe a)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Async a -> STM a
forall a. Async a -> STM a
waitSTM Async a
running) STM (Maybe a) -> STM (Maybe a) -> STM (Maybe a)
forall a. STM a -> STM a -> STM a
`orElse` (Maybe a
forall a. Maybe a
Nothing Maybe a -> STM () -> STM (Maybe a)
forall a b. a -> STM b -> STM a
forall (f :: * -> *) a b. Functor f => a -> f b -> f a
<$ LeasedReceipt -> STM ()
awaitDropped LeasedReceipt
leased))

-- Block until this receipt's renewal has given up on it.
awaitDropped :: LeasedReceipt -> STM ()
awaitDropped :: LeasedReceipt -> STM ()
awaitDropped LeasedReceipt
leased = TVar Bool -> STM Bool
forall a. TVar a -> STM a
readTVar (LeasedReceipt -> TVar Bool
lrDropped LeasedReceipt
leased) STM Bool -> (Bool -> STM ()) -> STM ()
forall a b. STM a -> (a -> STM b) -> STM b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= Bool -> STM ()
check

{- | Realise a receipt's disposition with its renewal stopped first, so none can follow an
acknowledgement, a release for retry, or a terminal backoff.
-}
disposing :: (MonadUnliftIO m) => LeasedReceipt -> m a -> m a
disposing :: forall (m :: * -> *) a.
MonadUnliftIO m =>
LeasedReceipt -> m a -> m a
disposing LeasedReceipt
leased m a
act = MVar (Maybe MonoTime)
-> (Maybe MonoTime -> m (Maybe MonoTime, a)) -> m a
forall (m :: * -> *) a b.
MonadUnliftIO m =>
MVar a -> (a -> m (a, b)) -> m b
MVar.modifyMVar (LeasedReceipt -> MVar (Maybe MonoTime)
lrHeld LeasedReceipt
leased) (m (Maybe MonoTime, a) -> Maybe MonoTime -> m (Maybe MonoTime, a)
forall a b. a -> b -> a
const ((Maybe MonoTime
forall a. Maybe a
Nothing,) (a -> (Maybe MonoTime, a)) -> m a -> m (Maybe MonoTime, a)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> m a
act))

-- What one renewal attempt settled. A disposed receipt ends the loop, a transient fault may
-- retry inside the margin, and a refusal is the ceiling or a fault no retry could clear.
data RenewalStep
    = RenewalKept
    | RenewalStopped
    | RenewalFailed Text
    | RenewalRefused Text

{- Renew one receipt until it is disposed, or until a renewal it cannot keep drops it. However
the task ends, residue included, its exit is what marks the receipt dropped. -}
renewalLoop :: (MonadUnliftIO m, KatipContext m) => LeaseOps -> ReceiptLease -> LeasedReceipt -> m ()
renewalLoop :: forall (m :: * -> *).
(MonadUnliftIO m, KatipContext m) =>
LeaseOps -> ReceiptLease -> LeasedReceipt -> m ()
renewalLoop LeaseOps
ops ReceiptLease
lease LeasedReceipt
leased = m ()
go m () -> m () -> m ()
forall (m :: * -> *) a b. MonadUnliftIO m => m a -> m b -> m a
`finally` STM () -> m ()
forall (m :: * -> *) a. MonadIO m => STM a -> m a
atomically (TVar Bool -> Bool -> STM ()
forall a. TVar a -> a -> STM ()
writeTVar (LeasedReceipt -> TVar Bool
lrDropped LeasedReceipt
leased) Bool
True)
  where
    go :: m ()
go =
        MVar (Maybe MonoTime) -> m (Maybe MonoTime)
forall (m :: * -> *) a. MonadIO m => MVar a -> m a
MVar.readMVar (LeasedReceipt -> MVar (Maybe MonoTime)
lrHeld LeasedReceipt
leased) m (Maybe MonoTime) -> (Maybe MonoTime -> m ()) -> m ()
forall a b. m a -> (a -> m b) -> m b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \case
            Maybe MonoTime
Nothing -> m ()
forall (f :: * -> *). Applicative f => f ()
pass
            Just MonoTime
deadline -> do
                LeaseOps -> MonoTime -> m ()
forall (m :: * -> *). MonadIO m => LeaseOps -> MonoTime -> m ()
waitForRenewal LeaseOps
ops MonoTime
deadline
                LeaseOps
-> ReceiptLease -> LeasedReceipt -> MonoTime -> m RenewalStep
forall (m :: * -> *).
MonadUnliftIO m =>
LeaseOps
-> ReceiptLease -> LeasedReceipt -> MonoTime -> m RenewalStep
attemptRenewal LeaseOps
ops ReceiptLease
lease LeasedReceipt
leased MonoTime
deadline m RenewalStep -> (RenewalStep -> m ()) -> m ()
forall a b. m a -> (a -> m b) -> m b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \case
                    RenewalStep
RenewalKept -> m ()
go
                    RenewalStep
RenewalStopped -> m ()
forall (f :: * -> *). Applicative f => f ()
pass
                    RenewalFailed Text
detail -> LeasedReceipt -> Text -> m ()
forall (m :: * -> *).
(MonadUnliftIO m, KatipContext m) =>
LeasedReceipt -> Text -> m ()
dropReceipt LeasedReceipt
leased Text
detail
                    RenewalRefused Text
detail -> LeasedReceipt -> Text -> m ()
forall (m :: * -> *).
(MonadUnliftIO m, KatipContext m) =>
LeasedReceipt -> Text -> m ()
dropReceipt LeasedReceipt
leased Text
detail

-- Wait until the renewal falls due, which a disposition in the meantime simply outlasts.
waitForRenewal :: (MonadIO m) => LeaseOps -> MonoTime -> m ()
waitForRenewal :: forall (m :: * -> *). MonadIO m => LeaseOps -> MonoTime -> m ()
waitForRenewal LeaseOps
ops MonoTime
deadline = do
    now <- IO MonoTime -> m MonoTime
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (LeaseOps -> IO MonoTime
loNow LeaseOps
ops)
    liftIO (loWaitUntil ops (leaseRenewAt now deadline))

{- Ask for another window, retrying a transient fault while this receipt's own margin lasts.
The shared delay-list policy paces the attempts, and the margin check ends them early. -}
attemptRenewal :: (MonadUnliftIO m) => LeaseOps -> ReceiptLease -> LeasedReceipt -> MonoTime -> m RenewalStep
attemptRenewal :: forall (m :: * -> *).
MonadUnliftIO m =>
LeaseOps
-> ReceiptLease -> LeasedReceipt -> MonoTime -> m RenewalStep
attemptRenewal LeaseOps
ops ReceiptLease
lease LeasedReceipt
leased MonoTime
deadline =
    RetryPolicyM m
-> (RetryStatus -> RenewalStep -> m Bool)
-> (RetryStatus -> m RenewalStep)
-> m RenewalStep
forall (m :: * -> *) b.
MonadIO m =>
RetryPolicyM m
-> (RetryStatus -> b -> m Bool) -> (RetryStatus -> m b) -> m b
retrying ([Int] -> RetryPolicyM m
forall (m :: * -> *). Monad m => [Int] -> RetryPolicyM m
delayListPolicy (LeaseOps -> [Int]
loRetryDelays LeaseOps
ops)) RetryStatus -> RenewalStep -> m Bool
forall {m :: * -> *} {p}. MonadIO m => p -> RenewalStep -> m Bool
withinMargin (m RenewalStep -> RetryStatus -> m RenewalStep
forall a b. a -> b -> a
const (LeaseOps -> ReceiptLease -> LeasedReceipt -> m RenewalStep
forall (m :: * -> *).
MonadUnliftIO m =>
LeaseOps -> ReceiptLease -> LeasedReceipt -> m RenewalStep
renewOnce LeaseOps
ops ReceiptLease
lease LeasedReceipt
leased))
  where
    withinMargin :: p -> RenewalStep -> m Bool
withinMargin p
_ = \case
        RenewalFailed Text
_ -> do
            now <- IO MonoTime -> m MonoTime
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (LeaseOps -> IO MonoTime
loNow LeaseOps
ops)
            pure (now < leaseRetryUntil (rlWindow lease) deadline)
        RenewalStep
_ -> Bool -> m Bool
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Bool
False

-- One attempt under the receipt's own lock, so it can never overlap that receipt's disposition.
renewOnce :: (MonadUnliftIO m) => LeaseOps -> ReceiptLease -> LeasedReceipt -> m RenewalStep
renewOnce :: forall (m :: * -> *).
MonadUnliftIO m =>
LeaseOps -> ReceiptLease -> LeasedReceipt -> m RenewalStep
renewOnce LeaseOps
ops ReceiptLease
lease LeasedReceipt
leased =
    MVar (Maybe MonoTime)
-> (Maybe MonoTime -> m (Maybe MonoTime, RenewalStep))
-> m RenewalStep
forall (m :: * -> *) a b.
MonadUnliftIO m =>
MVar a -> (a -> m (a, b)) -> m b
MVar.modifyMVar (LeasedReceipt -> MVar (Maybe MonoTime)
lrHeld LeasedReceipt
leased) ((Maybe MonoTime -> m (Maybe MonoTime, RenewalStep))
 -> m RenewalStep)
-> (Maybe MonoTime -> m (Maybe MonoTime, RenewalStep))
-> m RenewalStep
forall a b. (a -> b) -> a -> b
$ \case
        Maybe MonoTime
Nothing -> (Maybe MonoTime, RenewalStep) -> m (Maybe MonoTime, RenewalStep)
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Maybe MonoTime
forall a. Maybe a
Nothing, RenewalStep
RenewalStopped)
        Just MonoTime
deadline -> do
            -- Read before the request, so the extended deadline never claims more than it won.
            now <- IO MonoTime -> m MonoTime
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (LeaseOps -> IO MonoTime
loNow LeaseOps
ops)
            case leaseRequest lease now of
                Maybe Seconds
Nothing -> (Maybe MonoTime, RenewalStep) -> m (Maybe MonoTime, RenewalStep)
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (MonoTime -> Maybe MonoTime
forall a. a -> Maybe a
Just MonoTime
deadline, Text -> RenewalStep
RenewalRefused Text
ceilingReason)
                Just Seconds
window -> MonoTime
-> MonoTime
-> Seconds
-> Either TransportFault ()
-> (Maybe MonoTime, RenewalStep)
settle MonoTime
now MonoTime
deadline Seconds
window (Either TransportFault () -> (Maybe MonoTime, RenewalStep))
-> m (Either TransportFault ()) -> m (Maybe MonoTime, RenewalStep)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> IO (Either TransportFault ()) -> m (Either TransportFault ())
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (LeaseOps
-> ReceiptHandle -> Seconds -> IO (Either TransportFault ())
renewOrResidue LeaseOps
ops (QueueMessage -> ReceiptHandle
msgReceipt (LeasedReceipt -> QueueMessage
lrMessage LeasedReceipt
leased)) Seconds
window)

-- A kept window moves the deadline on, a retryable fault leaves it for another attempt, and
-- anything else ends the lease: the typed cause alone splits retry from refusal.
settle :: MonoTime -> MonoTime -> Seconds -> Either TransportFault () -> (Maybe MonoTime, RenewalStep)
settle :: MonoTime
-> MonoTime
-> Seconds
-> Either TransportFault ()
-> (Maybe MonoTime, RenewalStep)
settle MonoTime
now MonoTime
deadline (Seconds Int
window) = \case
    Right () -> (MonoTime -> Maybe MonoTime
forall a. a -> Maybe a
Just (MonoTime -> Double -> MonoTime
monoAfter MonoTime
now (Int -> Double
forall a b. (Integral a, Num b) => a -> b
fromIntegral Int
window)), RenewalStep
RenewalKept)
    Left TransportFault
fault
        | TransportCause -> Bool
transportRetryable (TransportFault -> TransportCause
tfCause TransportFault
fault) -> (MonoTime -> Maybe MonoTime
forall a. a -> Maybe a
Just MonoTime
deadline, Text -> RenewalStep
RenewalFailed (TransportFault -> Text
tfDetail TransportFault
fault))
        | Bool
otherwise -> (MonoTime -> Maybe MonoTime
forall a. a -> Maybe a
Just MonoTime
deadline, Text -> RenewalStep
RenewalRefused (TransportFault -> Text
tfDetail TransportFault
fault))

{- The handle reports every backend failure as a value, so an exception is an invariant break.
Contain it here: a renewal task that died would let its lease lapse unnoticed. -}
renewOrResidue :: LeaseOps -> ReceiptHandle -> Seconds -> IO (Either TransportFault ())
renewOrResidue :: LeaseOps
-> ReceiptHandle -> Seconds -> IO (Either TransportFault ())
renewOrResidue LeaseOps
ops ReceiptHandle
receipt Seconds
window =
    (SomeException -> Either TransportFault ())
-> (Either TransportFault () -> Either TransportFault ())
-> Either SomeException (Either TransportFault ())
-> Either TransportFault ()
forall a c b. (a -> c) -> (b -> c) -> Either a b -> c
either (TransportFault -> Either TransportFault ()
forall a b. a -> Either a b
Left (TransportFault -> Either TransportFault ())
-> (SomeException -> TransportFault)
-> SomeException
-> Either TransportFault ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. SomeException -> TransportFault
forall {e}. Exception e => e -> TransportFault
residue) Either TransportFault () -> Either TransportFault ()
forall a. a -> a
id (Either SomeException (Either TransportFault ())
 -> Either TransportFault ())
-> IO (Either SomeException (Either TransportFault ()))
-> IO (Either TransportFault ())
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> IO (Either TransportFault ())
-> IO (Either SomeException (Either TransportFault ()))
forall (m :: * -> *) a.
MonadUnliftIO m =>
m a -> m (Either SomeException a)
tryAny (LeaseOps
-> ReceiptHandle -> Seconds -> IO (Either TransportFault ())
loRenew LeaseOps
ops ReceiptHandle
receipt Seconds
window)
  where
    residue :: e -> TransportFault
residue e
e = TransportCause -> Text -> TransportFault
transportFault TransportCause
TransportProtocol (Text
"visibility renewal escaped its typed contract: " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> e -> Text
forall e. Exception e => e -> Text
displayExceptionT e
e)

{- Give up on one receipt: stop renewing it and leave it unacknowledged, so the backend
redelivers it once the window lapses. -}
dropReceipt :: (MonadUnliftIO m, KatipContext m) => LeasedReceipt -> Text -> m ()
dropReceipt :: forall (m :: * -> *).
(MonadUnliftIO m, KatipContext m) =>
LeasedReceipt -> Text -> m ()
dropReceipt LeasedReceipt
leased Text
detail = do
    Severity -> LogStr -> m ()
forall (m :: * -> *).
(Applicative m, KatipContext m) =>
Severity -> LogStr -> m ()
logFM Severity
WarningS (Text -> LogStr
forall a. StringConv a Text => a -> LogStr
ls (Text
"dropping a mirror receipt whose visibility could not be renewed: " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
detail))
    MVar (Maybe MonoTime)
-> (Maybe MonoTime -> m (Maybe MonoTime)) -> m ()
forall (m :: * -> *) a.
MonadUnliftIO m =>
MVar a -> (a -> m a) -> m ()
MVar.modifyMVar_ (LeasedReceipt -> MVar (Maybe MonoTime)
lrHeld LeasedReceipt
leased) (m (Maybe MonoTime) -> Maybe MonoTime -> m (Maybe MonoTime)
forall a b. a -> b -> a
const (Maybe MonoTime -> m (Maybe MonoTime)
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Maybe MonoTime
forall a. Maybe a
Nothing))

ceilingReason :: Text
ceilingReason :: Text
ceilingReason = Text
"the queue's own maximum time in flight from receipt is spent"

{- | When a held lease is renewed: a third of the way into what is left of it, so two renewals
can fail before the window lapses.
-}
leaseRenewAt :: MonoTime -> MonoTime -> MonoTime
leaseRenewAt :: MonoTime -> MonoTime -> MonoTime
leaseRenewAt MonoTime
now MonoTime
deadline = MonoTime -> Double -> MonoTime
monoAfter MonoTime
now (Double -> Double -> Double
forall a. Ord a => a -> a -> a
max Double
0 (MonoTime -> MonoTime -> Double
monoSecondsBetween MonoTime
now MonoTime
deadline) Double -> Double -> Double
forall a. Fractional a => a -> a -> a
/ Double
3)

{- | The last instant a renewal may be sent: a tenth of the window before the deadline, the
transport margin a request needs to land while the lease still holds.
-}
leaseRetryUntil :: Seconds -> MonoTime -> MonoTime
leaseRetryUntil :: Seconds -> MonoTime -> MonoTime
leaseRetryUntil (Seconds Int
window) MonoTime
deadline = MonoTime -> Double -> MonoTime
monoAfter MonoTime
deadline (Double -> Double
forall a. Num a => a -> a
negate (Int -> Double
forall a b. (Integral a, Num b) => a -> b
fromIntegral Int
window Double -> Double -> Double
forall a. Fractional a => a -> a -> a
/ Double
10))

{- | The window a renewal asks for: the one the backend granted, clipped to what is left before
its ceiling from receipt. 'Nothing' once under a second is left, which drops the receipt.
-}
leaseRequest :: ReceiptLease -> MonoTime -> Maybe Seconds
leaseRequest :: ReceiptLease -> MonoTime -> Maybe Seconds
leaseRequest ReceiptLease
lease MonoTime
now
    | Double
remaining Double -> Double -> Bool
forall a. Ord a => a -> a -> Bool
< Double
1 = Maybe Seconds
forall a. Maybe a
Nothing
    | Bool
otherwise = Seconds -> Maybe Seconds
forall a. a -> Maybe a
Just (Int -> Seconds
Seconds (Int -> Int -> Int
forall a. Ord a => a -> a -> a
min Int
window (Double -> Int
forall b. Integral b => Double -> b
forall a b. (RealFrac a, Integral b) => a -> b
floor Double
remaining)))
  where
    Seconds Int
window = ReceiptLease -> Seconds
rlWindow ReceiptLease
lease
    remaining :: Double
remaining = MonoTime -> MonoTime -> Double
monoSecondsBetween MonoTime
now (ReceiptLease -> MonoTime
rlCeilingAt ReceiptLease
lease)