{-# LANGUAGE OverloadedStrings #-}
{-# OPTIONS_HADDOCK not-home #-}

-- | Internal to the arbiter packages. Not covered by the PVP.
--
-- Batch settlement, outcome reporting, and force-cancel finalization.
--
-- Each 'Effects' statement runs as one transaction or one hook.
module Arbiter.Worker.Batch
  ( -- * The pool's side
    Effects (..)
  , Report (..)
  , Failure
  , FailureKind (..)
  , Outcome (..)
  , cancelsTree
  , Mode (..)
  , Callbacks (..)

    -- * Running a batch
  , Handoff
  , newHandoff
  , runBatch
  ) where

import Arbiter.Core.Exceptions
  ( BranchCancelException (..)
  , JobDeadlineExceeded (..)
  , JobException (..)
  , JobForceCancelled (..)
  , JobGoneException (..)
  , JobNackException (..)
  , JobPermanentException (..)
  , JobRetryableException (..)
  , JobScopedFailure (..)
  , ParsingException (..)
  , TreeCancelException (..)
  , displayEx
  , namedJobIds
  )
import Arbiter.Core.Job.Types (JobId)
import Control.Applicative ((<|>))
import Control.Concurrent.Class.MonadSTM
  ( MonadSTM
  , TVar
  , atomically
  , modifyTVar'
  , newTVarIO
  , readTVarIO
  , stateTVar
  )
import Control.Monad (unless, void, when)
import Control.Monad.Class.MonadFork (MonadFork)
import Control.Monad.Class.MonadThrow (Exception (..), MonadCatch (..), MonadMask (..), MonadThrow (..), SomeException)
import Control.Monad.Class.MonadTime.SI (MonadMonotonicTime, MonadTime (..), UTCTime)
import Data.Foldable (toList, traverse_)
import Data.List (partition, sortOn)
import Data.List.NonEmpty (NonEmpty (..))
import Data.Map.Strict qualified as Map
import Data.Maybe (isNothing)
import Data.Ord (Down (..))
import Data.Set (Set)
import Data.Set qualified as Set
import Data.Text (Text)
import Data.Time (NominalDiffTime)

import Arbiter.Worker.Heartbeat.Guard (Batch (..), HeartbeatGuard, guardBatch, guardKey, trySync)
import Arbiter.Worker.Logger (LogLevel (..))

-- ---------------------------------------------------------------------------
-- The pool's side
-- ---------------------------------------------------------------------------

-- | A hook the lifecycle fires.
data Report job
  = -- | The job was claimed. The batch start time.
    Claimed job UTCTime
  | -- | The job was acked. Its start and end times.
    Succeeded job UTCTime UTCTime
  | -- | A handler failure, written as the outcome says.
    Failed job Text UTCTime UTCTime Outcome
  | -- | A force-cancel or a tree cancel took the job, for the reason given.
    Cancelled job Text
  | -- | This worker can no longer act on the job.
    Unavailable job Text

-- | How a handler failure is written.
data FailureKind
  = -- | Retry with backoff, or move to the DLQ when attempts are exhausted.
    RetryFailure
  | -- | Retry after this delay in seconds, or move to the DLQ when attempts are exhausted.
    RetryAfterFailure NominalDiffTime
  | -- | Move to the DLQ now.
    PermanentFailure
  | -- | Delete the whole job tree from its root.
    TreeCancelFailure
  | -- | Delete the job's parent, or the job when it has none, with all descendants.
    BranchCancelFailure
  deriving stock (FailureKind -> FailureKind -> Bool
(FailureKind -> FailureKind -> Bool)
-> (FailureKind -> FailureKind -> Bool) -> Eq FailureKind
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: FailureKind -> FailureKind -> Bool
== :: FailureKind -> FailureKind -> Bool
$c/= :: FailureKind -> FailureKind -> Bool
/= :: FailureKind -> FailureKind -> Bool
Eq, Int -> FailureKind -> ShowS
[FailureKind] -> ShowS
FailureKind -> String
(Int -> FailureKind -> ShowS)
-> (FailureKind -> String)
-> ([FailureKind] -> ShowS)
-> Show FailureKind
forall a.
(Int -> a -> ShowS) -> (a -> String) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> FailureKind -> ShowS
showsPrec :: Int -> FailureKind -> ShowS
$cshow :: FailureKind -> String
show :: FailureKind -> String
$cshowList :: [FailureKind] -> ShowS
showList :: [FailureKind] -> ShowS
Show)

-- | A handler failure: its message and disposition.
type Failure = (Text, FailureKind)

-- | Where a failure write left the job. A tree cancel carries the ids it deleted.
data Outcome
  = -- | Scheduled for retry after this backoff delay.
    Retrying NominalDiffTime
  | -- | Moved to the DLQ.
    DeadLettered
  | -- | Deleted by a tree or branch cancel.
    TreeCancelled [JobId]
  deriving stock (Outcome -> Outcome -> Bool
(Outcome -> Outcome -> Bool)
-> (Outcome -> Outcome -> Bool) -> Eq Outcome
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: Outcome -> Outcome -> Bool
== :: Outcome -> Outcome -> Bool
$c/= :: Outcome -> Outcome -> Bool
/= :: Outcome -> Outcome -> Bool
Eq, Int -> Outcome -> ShowS
[Outcome] -> ShowS
Outcome -> String
(Int -> Outcome -> ShowS)
-> (Outcome -> String) -> ([Outcome] -> ShowS) -> Show Outcome
forall a.
(Int -> a -> ShowS) -> (a -> String) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> Outcome -> ShowS
showsPrec :: Int -> Outcome -> ShowS
$cshow :: Outcome -> String
show :: Outcome -> String
$cshowList :: [Outcome] -> ShowS
showList :: [Outcome] -> ShowS
Show)

-- | Batch statements and hooks. Callbacks use the handler's context and join
-- its active transaction. Each statement effect is one transaction or one hook.
data Effects n ctx job kids stored = Effects
  { forall (n :: * -> *) ctx job kids stored.
Effects n ctx job kids stored -> ctx
effectAmbient :: ctx
  -- ^ The context the batch runs in, outside the handler.
  , forall (n :: * -> *) ctx job kids stored.
Effects n ctx job kids stored -> ((n () -> n ()) -> n ()) -> n ()
effectSpan :: ((n () -> n ()) -> n ()) -> n ()
  -- ^ Runs the batch under its consumer span, handing it the capture that runs
  -- the heartbeat hooks under that span.
  , forall (n :: * -> *) ctx job kids stored.
Effects n ctx job kids stored -> ctx -> job -> Maybe stored -> n ()
effectAck :: ctx -> job -> Maybe stored -> n ()
  -- ^ Ack one job and store its result. Throws 'JobGoneException' for a job held elsewhere.
  , forall (n :: * -> *) ctx job kids stored.
Effects n ctx job kids stored
-> ctx -> [(job, Maybe stored)] -> n ([job], [job])
effectAckAll :: ctx -> [(job, Maybe stored)] -> n ([job], [job])
  -- ^ Ack in bulk. The jobs acked, and the jobs found held elsewhere.
  , forall (n :: * -> *) ctx job kids stored.
Effects n ctx job kids stored
-> ctx -> Failure -> job -> n (Either Text Outcome)
effectFail :: ctx -> Failure -> job -> n (Either Text Outcome)
  -- ^ Write one job's failure. 'Left' why no row was written.
  , forall (n :: * -> *) ctx job kids stored.
Effects n ctx job kids stored
-> ctx
-> Failure
-> [job]
-> [job]
-> n [(job, Either Text Outcome)]
effectFailAll :: ctx -> Failure -> [job] -> [job] -> n [(job, Either Text Outcome)]
  -- ^ Write the failures of the jobs left unfinalized, and cancel the jobs held
  -- elsewhere when the failure cancels a tree, in one transaction.
  , forall (n :: * -> *) ctx job kids stored.
Effects n ctx job kids stored -> ctx -> [job] -> n (Set JobId)
effectDeleteCancelled :: ctx -> [job] -> n (Set JobId)
  -- ^ Delete the jobs a force-cancel flagged. The ids deleted.
  , forall (n :: * -> *) ctx job kids stored.
Effects n ctx job kids stored
-> ctx -> Maybe NominalDiffTime -> [job] -> n (Set JobId)
effectRelease :: ctx -> Maybe NominalDiffTime -> [job] -> n (Set JobId)
  -- ^ Hand back the attempt the claim consumed, parking the jobs for the delay when
  -- there is one. The ids released.
  , forall (n :: * -> *) ctx job kids stored.
Effects n ctx job kids stored -> ctx -> job -> kids -> n ()
effectSpawn :: ctx -> job -> kids -> n ()
  -- ^ Attach children to one job and settle it suspended, in one transaction.
  , forall (n :: * -> *) ctx job kids stored.
Effects n ctx job kids stored -> ctx -> Report job -> n ()
effectReport :: ctx -> Report job -> n ()
  -- ^ Fire the hook for one report.
  , forall (n :: * -> *) ctx job kids stored.
Effects n ctx job kids stored -> LogLevel -> [job] -> Text -> n ()
effectLog :: LogLevel -> [job] -> Text -> n ()
  -- ^ The pool log, with the jobs in context.
  }

-- | How the pool runs a batch's handler.
data Mode n ctx job kids stored
  = -- | One transaction that runs the handler, acks the job and stores its result.
    SingleMode (job -> n ())
  | -- | The handler settles each job through the callbacks, in its context at the call.
    BatchedMode (NonEmpty job -> Callbacks n ctx job kids stored -> n ())

-- | The settle operations a batch handler drives its jobs through.
data Callbacks n ctx job kids stored = Callbacks
  { forall (n :: * -> *) ctx job kids stored.
Callbacks n ctx job kids stored
-> ctx -> job -> Maybe stored -> n ()
callbackAck :: ctx -> job -> Maybe stored -> n ()
  -- ^ Ack one job and store its result.
  , forall (n :: * -> *) ctx job kids stored.
Callbacks n ctx job kids stored
-> ctx -> [(job, Maybe stored)] -> n ()
callbackAckAll :: ctx -> [(job, Maybe stored)] -> n ()
  -- ^ Ack in bulk, storing each result.
  , forall (n :: * -> *) ctx job kids stored.
Callbacks n ctx job kids stored -> ctx -> Failure -> job -> n ()
callbackFail :: ctx -> Failure -> job -> n ()
  -- ^ Write one job's failure.
  , forall (n :: * -> *) ctx job kids stored.
Callbacks n ctx job kids stored
-> ctx -> Maybe NominalDiffTime -> job -> n ()
callbackNack :: ctx -> Maybe NominalDiffTime -> job -> n ()
  -- ^ Release one job for reprocessing, after the delay when there is one.
  , forall (n :: * -> *) ctx job kids stored.
Callbacks n ctx job kids stored -> ctx -> job -> kids -> n ()
callbackSpawn :: ctx -> job -> kids -> n ()
  -- ^ Attach children to one job and suspend it.
  }

-- ---------------------------------------------------------------------------
-- The handoff
-- ---------------------------------------------------------------------------

-- | Batch settlement progress.
data Progress n = Progress
  { forall (n :: * -> *). Progress n -> Set JobId
progressHandled :: !(Set JobId)
  -- ^ Jobs whose outcome has been recorded.
  , forall (n :: * -> *). Progress n -> Set JobId
progressUnowned :: !(Set JobId)
  -- ^ Jobs a batch ack found under another claim.
  , forall (n :: * -> *). Progress n -> Set JobId
progressCancelled :: !(Set JobId)
  -- ^ Jobs a force-cancel accounted for.
  , forall (n :: * -> *). Progress n -> Set JobId
progressClaimed :: !(Set JobId)
  -- ^ Jobs whose claim report fired.
  , forall (n :: * -> *). Progress n -> Set JobId
progressReported :: !(Set JobId)
  -- ^ Jobs whose terminal report fired.
  , forall (n :: * -> *). Progress n -> Maybe (n ())
progressDeferred :: !(Maybe (n ()))
  -- ^ The hooks of a settle a signal can interrupt.
  }

-- | Finalization state shared by the handler and the force-cancel finalizer, with
-- the batch's job key and start time.
data Handoff n job = Handoff
  { forall (n :: * -> *) job. Handoff n job -> job -> JobId
handoffKey :: job -> JobId
  , forall (n :: * -> *) job. Handoff n job -> UTCTime
handoffStart :: UTCTime
  , forall (n :: * -> *) job. Handoff n job -> TVar n (Progress n)
handoffVar :: TVar n (Progress n)
  }

-- | A handoff keyed the way the guard keys its jobs, starting now.
newHandoff :: (MonadSTM n, MonadTime n) => HeartbeatGuard n job -> n (Handoff n job)
{-# SPECIALIZE newHandoff :: HeartbeatGuard IO job -> IO (Handoff IO job) #-}
newHandoff :: forall (n :: * -> *) job.
(MonadSTM n, MonadTime n) =>
HeartbeatGuard n job -> n (Handoff n job)
newHandoff HeartbeatGuard n job
guard = (job -> JobId) -> UTCTime -> TVar n (Progress n) -> Handoff n job
forall (n :: * -> *) job.
(job -> JobId) -> UTCTime -> TVar n (Progress n) -> Handoff n job
Handoff (HeartbeatGuard n job -> job -> JobId
forall (n :: * -> *) job. HeartbeatGuard n job -> job -> JobId
guardKey HeartbeatGuard n job
guard) (UTCTime -> TVar n (Progress n) -> Handoff n job)
-> n UTCTime -> n (TVar n (Progress n) -> Handoff n job)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> n UTCTime
forall (m :: * -> *). MonadTime m => m UTCTime
getCurrentTime n (TVar n (Progress n) -> Handoff n job)
-> n (TVar n (Progress n)) -> n (Handoff n job)
forall a b. n (a -> b) -> n a -> n b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Progress n -> n (TVar n (Progress n))
forall a. a -> n (TVar n a)
forall (m :: * -> *) a. MonadSTM m => a -> m (TVar m a)
newTVarIO Progress n
forall (n :: * -> *). Progress n
emptyProgress

-- | Initial settlement state.
emptyProgress :: Progress n
emptyProgress :: forall (n :: * -> *). Progress n
emptyProgress =
  Progress
    { progressHandled :: Set JobId
progressHandled = Set JobId
forall a. Monoid a => a
mempty
    , progressUnowned :: Set JobId
progressUnowned = Set JobId
forall a. Monoid a => a
mempty
    , progressCancelled :: Set JobId
progressCancelled = Set JobId
forall a. Monoid a => a
mempty
    , progressClaimed :: Set JobId
progressClaimed = Set JobId
forall a. Monoid a => a
mempty
    , progressReported :: Set JobId
progressReported = Set JobId
forall a. Monoid a => a
mempty
    , progressDeferred :: Maybe (n ())
progressDeferred = Maybe (n ())
forall a. Maybe a
Nothing
    }

readProgress :: (MonadSTM n) => Handoff n job -> n (Progress n)
readProgress :: forall (n :: * -> *) job.
MonadSTM n =>
Handoff n job -> n (Progress n)
readProgress = TVar n (Progress n) -> n (Progress n)
forall a. TVar n a -> n a
forall (m :: * -> *) a. MonadSTM m => TVar m a -> m a
readTVarIO (TVar n (Progress n) -> n (Progress n))
-> (Handoff n job -> TVar n (Progress n))
-> Handoff n job
-> n (Progress n)
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Handoff n job -> TVar n (Progress n)
forall (n :: * -> *) job. Handoff n job -> TVar n (Progress n)
handoffVar

alterProgress :: (MonadSTM n) => Handoff n job -> (Progress n -> Progress n) -> n ()
alterProgress :: forall (n :: * -> *) job.
MonadSTM n =>
Handoff n job -> (Progress n -> Progress n) -> n ()
alterProgress Handoff n job
handoff = STM n () -> n ()
forall a. HasCallStack => STM n a -> n a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM n () -> n ())
-> ((Progress n -> Progress n) -> STM n ())
-> (Progress n -> Progress n)
-> n ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. TVar n (Progress n) -> (Progress n -> Progress n) -> STM n ()
forall a. TVar n a -> (a -> a) -> STM n ()
forall (m :: * -> *) a.
MonadSTM m =>
TVar m a -> (a -> a) -> STM m ()
modifyTVar' (Handoff n job -> TVar n (Progress n)
forall (n :: * -> *) job. Handoff n job -> TVar n (Progress n)
handoffVar Handoff n job
handoff)

onProgress :: (MonadSTM n) => Handoff n job -> (Progress n -> (a, Progress n)) -> n a
onProgress :: forall (n :: * -> *) job a.
MonadSTM n =>
Handoff n job -> (Progress n -> (a, Progress n)) -> n a
onProgress Handoff n job
handoff = STM n a -> n a
forall a. HasCallStack => STM n a -> n a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM n a -> n a)
-> ((Progress n -> (a, Progress n)) -> STM n a)
-> (Progress n -> (a, Progress n))
-> n a
forall b c a. (b -> c) -> (a -> b) -> a -> c
. TVar n (Progress n) -> (Progress n -> (a, Progress n)) -> STM n a
forall s a. TVar n s -> (s -> (a, s)) -> STM n a
forall (m :: * -> *) s a.
MonadSTM m =>
TVar m s -> (s -> (a, s)) -> STM m a
stateTVar (Handoff n job -> TVar n (Progress n)
forall (n :: * -> *) job. Handoff n job -> TVar n (Progress n)
handoffVar Handoff n job
handoff)

-- | Finalized jobs and jobs held under another claim.
data Settled job = Settled [job] [job]

finalized :: [job] -> Settled job
finalized :: forall job. [job] -> Settled job
finalized [job]
jobs = [job] -> [job] -> Settled job
forall job. [job] -> [job] -> Settled job
Settled [job]
jobs []

-- | Record a finalization. Jobs owned by another claim also count as handled.
recorded :: Handoff n job -> Settled job -> Progress n -> Progress n
recorded :: forall (n :: * -> *) job.
Handoff n job -> Settled job -> Progress n -> Progress n
recorded Handoff n job
handoff (Settled [job]
handled [job]
unowned) Progress n
progress =
  Progress n
progress
    { progressHandled = progressHandled progress <> ids handled <> gone
    , progressUnowned = progressUnowned progress <> gone
    }
  where
    gone :: Set JobId
gone = [job] -> Set JobId
ids [job]
unowned
    ids :: [job] -> Set JobId
ids = [JobId] -> Set JobId
forall a. Ord a => [a] -> Set a
Set.fromList ([JobId] -> Set JobId) -> ([job] -> [JobId]) -> [job] -> Set JobId
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (job -> JobId) -> [job] -> [JobId]
forall a b. (a -> b) -> [a] -> [b]
map (Handoff n job -> job -> JobId
forall (n :: * -> *) job. Handoff n job -> job -> JobId
handoffKey Handoff n job
handoff)

record :: (MonadSTM n) => Handoff n job -> Settled job -> n ()
record :: forall (n :: * -> *) job.
MonadSTM n =>
Handoff n job -> Settled job -> n ()
record Handoff n job
handoff = Handoff n job -> (Progress n -> Progress n) -> n ()
forall (n :: * -> *) job.
MonadSTM n =>
Handoff n job -> (Progress n -> Progress n) -> n ()
alterProgress Handoff n job
handoff ((Progress n -> Progress n) -> n ())
-> (Settled job -> Progress n -> Progress n) -> Settled job -> n ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Handoff n job -> Settled job -> Progress n -> Progress n
forall (n :: * -> *) job.
Handoff n job -> Settled job -> Progress n -> Progress n
recorded Handoff n job
handoff

-- | Whether a set of job ids names this job.
hasIdIn :: Handoff n job -> Set JobId -> job -> Bool
hasIdIn :: forall (n :: * -> *) job. Handoff n job -> Set JobId -> job -> Bool
hasIdIn Handoff n job
handoff Set JobId
ids job
job = JobId -> Set JobId -> Bool
forall a. Ord a => a -> Set a -> Bool
Set.member (Handoff n job -> job -> JobId
forall (n :: * -> *) job. Handoff n job -> job -> JobId
handoffKey Handoff n job
handoff job
job) Set JobId
ids

-- | Sort jobs by descending id for consistent row-lock ordering.
byIdDesc :: Handoff n job -> (job -> Bool) -> NonEmpty job -> [job]
byIdDesc :: forall (n :: * -> *) job.
Handoff n job -> (job -> Bool) -> NonEmpty job -> [job]
byIdDesc Handoff n job
handoff job -> Bool
keep = (job -> Down JobId) -> [job] -> [job]
forall b a. Ord b => (a -> b) -> [a] -> [a]
sortOn (JobId -> Down JobId
forall a. a -> Down a
Down (JobId -> Down JobId) -> (job -> JobId) -> job -> Down JobId
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Handoff n job -> job -> JobId
forall (n :: * -> *) job. Handoff n job -> job -> JobId
handoffKey Handoff n job
handoff) ([job] -> [job])
-> (NonEmpty job -> [job]) -> NonEmpty job -> [job]
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (job -> Bool) -> [job] -> [job]
forall a. (a -> Bool) -> [a] -> [a]
filter job -> Bool
keep ([job] -> [job])
-> (NonEmpty job -> [job]) -> NonEmpty job -> [job]
forall b c a. (b -> c) -> (a -> b) -> a -> c
. NonEmpty job -> [job]
forall a. NonEmpty a -> [a]
forall (t :: * -> *) a. Foldable t => t a -> [a]
toList

-- | Jobs in the batch by membership in one of the recorded sets.
jobsBy :: (MonadSTM n) => Handoff n job -> (Progress n -> Set JobId) -> (Bool -> Bool) -> NonEmpty job -> n [job]
jobsBy :: forall (n :: * -> *) job.
MonadSTM n =>
Handoff n job
-> (Progress n -> Set JobId)
-> (Bool -> Bool)
-> NonEmpty job
-> n [job]
jobsBy Handoff n job
handoff Progress n -> Set JobId
recordedSet Bool -> Bool
member NonEmpty job
jobs = do
  progress <- Handoff n job -> n (Progress n)
forall (n :: * -> *) job.
MonadSTM n =>
Handoff n job -> n (Progress n)
readProgress Handoff n job
handoff
  pure (byIdDesc handoff (member . hasIdIn handoff (recordedSet progress)) jobs)

-- | Jobs in the batch that have no recorded outcome.
pendingJobs :: (MonadSTM n) => Handoff n job -> NonEmpty job -> n [job]
pendingJobs :: forall (n :: * -> *) job.
MonadSTM n =>
Handoff n job -> NonEmpty job -> n [job]
pendingJobs Handoff n job
handoff = Handoff n job
-> (Progress n -> Set JobId)
-> (Bool -> Bool)
-> NonEmpty job
-> n [job]
forall (n :: * -> *) job.
MonadSTM n =>
Handoff n job
-> (Progress n -> Set JobId)
-> (Bool -> Bool)
-> NonEmpty job
-> n [job]
jobsBy Handoff n job
handoff Progress n -> Set JobId
forall (n :: * -> *). Progress n -> Set JobId
progressHandled Bool -> Bool
not

-- | Jobs found under another claim.
unownedJobs :: (MonadSTM n) => Handoff n job -> NonEmpty job -> n [job]
unownedJobs :: forall (n :: * -> *) job.
MonadSTM n =>
Handoff n job -> NonEmpty job -> n [job]
unownedJobs Handoff n job
handoff = Handoff n job
-> (Progress n -> Set JobId)
-> (Bool -> Bool)
-> NonEmpty job
-> n [job]
forall (n :: * -> *) job.
MonadSTM n =>
Handoff n job
-> (Progress n -> Set JobId)
-> (Bool -> Bool)
-> NonEmpty job
-> n [job]
jobsBy Handoff n job
handoff Progress n -> Set JobId
forall (n :: * -> *). Progress n -> Set JobId
progressUnowned Bool -> Bool
forall a. a -> a
id

-- | Record force-cancelled jobs and return all recorded ids.
recordCancelled :: (MonadSTM n) => Handoff n job -> Set JobId -> n (Set JobId)
recordCancelled :: forall (n :: * -> *) job.
MonadSTM n =>
Handoff n job -> Set JobId -> n (Set JobId)
recordCancelled Handoff n job
handoff Set JobId
ids =
  Handoff n job
-> (Progress n -> (Set JobId, Progress n)) -> n (Set JobId)
forall (n :: * -> *) job a.
MonadSTM n =>
Handoff n job -> (Progress n -> (a, Progress n)) -> n a
onProgress Handoff n job
handoff ((Progress n -> (Set JobId, Progress n)) -> n (Set JobId))
-> (Progress n -> (Set JobId, Progress n)) -> n (Set JobId)
forall a b. (a -> b) -> a -> b
$ \Progress n
progress ->
    let cancelled :: Set JobId
cancelled = Progress n -> Set JobId
forall (n :: * -> *). Progress n -> Set JobId
progressCancelled Progress n
progress Set JobId -> Set JobId -> Set JobId
forall a. Semigroup a => a -> a -> a
<> Set JobId
ids
     in (Set JobId
cancelled, Progress n
progress {progressCancelled = cancelled})

-- | Drop the hooks the batch's next report would finish.
clearDeferred :: (MonadSTM n) => Handoff n job -> n ()
clearDeferred :: forall (n :: * -> *) job. MonadSTM n => Handoff n job -> n ()
clearDeferred Handoff n job
handoff = Handoff n job -> (Progress n -> Progress n) -> n ()
forall (n :: * -> *) job.
MonadSTM n =>
Handoff n job -> (Progress n -> Progress n) -> n ()
alterProgress Handoff n job
handoff ((Progress n -> Progress n) -> n ())
-> (Progress n -> Progress n) -> n ()
forall a b. (a -> b) -> a -> b
$ \Progress n
progress -> Progress n
progress {progressDeferred = Nothing}

-- | Run the hooks a signal interrupted.
finishDeferred :: (MonadSTM n) => Handoff n job -> n ()
finishDeferred :: forall (n :: * -> *) job. MonadSTM n => Handoff n job -> n ()
finishDeferred Handoff n job
handoff = Handoff n job -> n (Progress n)
forall (n :: * -> *) job.
MonadSTM n =>
Handoff n job -> n (Progress n)
readProgress Handoff n job
handoff n (Progress n) -> (Progress n -> n ()) -> n ()
forall a b. n a -> (a -> n b) -> n b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= (n () -> n ()) -> Maybe (n ()) -> n ()
forall (t :: * -> *) (f :: * -> *) a b.
(Foldable t, Applicative f) =>
(a -> f b) -> t a -> f ()
traverse_ (n () -> n () -> n ()
forall a b. n a -> n b -> n b
forall (f :: * -> *) a b. Applicative f => f a -> f b -> f b
*> Handoff n job -> n ()
forall (n :: * -> *) job. MonadSTM n => Handoff n job -> n ()
clearDeferred Handoff n job
handoff) (Maybe (n ()) -> n ())
-> (Progress n -> Maybe (n ())) -> Progress n -> n ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Progress n -> Maybe (n ())
forall (n :: * -> *). Progress n -> Maybe (n ())
progressDeferred

-- ---------------------------------------------------------------------------
-- Running a batch
-- ---------------------------------------------------------------------------

-- | What every settle path works with: the effects, the context to run them in,
-- and the handoff. A callback rebuilds it with the handler's context.
data Run n ctx job kids stored = Run
  { forall (n :: * -> *) ctx job kids stored.
Run n ctx job kids stored -> Effects n ctx job kids stored
runEffects :: Effects n ctx job kids stored
  , forall (n :: * -> *) ctx job kids stored.
Run n ctx job kids stored -> ctx
runContext :: ctx
  , forall (n :: * -> *) ctx job kids stored.
Run n ctx job kids stored -> Handoff n job
runHandoff :: Handoff n job
  }

-- | The batch's run outside the handler.
ambientRun :: Effects n ctx job kids stored -> Handoff n job -> Run n ctx job kids stored
ambientRun :: forall (n :: * -> *) ctx job kids stored.
Effects n ctx job kids stored
-> Handoff n job -> Run n ctx job kids stored
ambientRun Effects n ctx job kids stored
effects Handoff n job
handoff = Run {runEffects :: Effects n ctx job kids stored
runEffects = Effects n ctx job kids stored
effects, runContext :: ctx
runContext = Effects n ctx job kids stored -> ctx
forall (n :: * -> *) ctx job kids stored.
Effects n ctx job kids stored -> ctx
effectAmbient Effects n ctx job kids stored
effects, runHandoff :: Handoff n job
runHandoff = Handoff n job
handoff}

-- | Run the batch on the calling thread: the claim hooks, the handler under the
-- guard, the outcome report, and the force-cancel finalizer.
runBatch
  :: (MonadFork n, MonadMask n, MonadMonotonicTime n, MonadSTM n, MonadTime n)
  => Effects n ctx job kids stored
  -> HeartbeatGuard n job
  -> Mode n ctx job kids stored
  -> Handoff n job
  -> NonEmpty job
  -> n ()
{-# SPECIALIZE runBatch ::
  Effects IO ctx job kids stored
  -> HeartbeatGuard IO job
  -> Mode IO ctx job kids stored
  -> Handoff IO job
  -> NonEmpty job
  -> IO ()
  #-}
runBatch :: forall (n :: * -> *) ctx job kids stored.
(MonadFork n, MonadMask n, MonadMonotonicTime n, MonadSTM n,
 MonadTime n) =>
Effects n ctx job kids stored
-> HeartbeatGuard n job
-> Mode n ctx job kids stored
-> Handoff n job
-> NonEmpty job
-> n ()
runBatch Effects n ctx job kids stored
effects HeartbeatGuard n job
guard Mode n ctx job kids stored
mode Handoff n job
handoff NonEmpty job
jobs = do
  let startTime :: UTCTime
startTime = Handoff n job -> UTCTime
forall (n :: * -> *) job. Handoff n job -> UTCTime
handoffStart Handoff n job
handoff
      run :: Run n ctx job kids stored
run = Effects n ctx job kids stored
-> Handoff n job -> Run n ctx job kids stored
forall (n :: * -> *) ctx job kids stored.
Effects n ctx job kids stored
-> Handoff n job -> Run n ctx job kids stored
ambientRun Effects n ctx job kids stored
effects Handoff n job
handoff
      (job
firstJob :| [job]
_) = NonEmpty job
jobs
      -- Rethrown as it came.
      onForceCancel :: JobForceCancelled -> n ()
onForceCancel exc :: JobForceCancelled
exc@(JobForceCancelled [JobId]
cancelledIds [JobId]
goneIds) = do
        Run n ctx job kids stored
-> NonEmpty job -> [JobId] -> [JobId] -> n ()
forall (n :: * -> *) ctx job kids stored.
(MonadMask n, MonadSTM n) =>
Run n ctx job kids stored
-> NonEmpty job -> [JobId] -> [JobId] -> n ()
finalizeForceCancelled Run n ctx job kids stored
run NonEmpty job
jobs [JobId]
cancelledIds [JobId]
goneIds
        JobForceCancelled -> n ()
forall e a. Exception e => e -> n a
forall (m :: * -> *) e a. (MonadThrow m, Exception e) => e -> m a
throwIO JobForceCancelled
exc
  -- The span covers the claim hooks, the outcome report and the force-cancel
  -- finalizer.
  Effects n ctx job kids stored -> ((n () -> n ()) -> n ()) -> n ()
forall (n :: * -> *) ctx job kids stored.
Effects n ctx job kids stored -> ((n () -> n ()) -> n ()) -> n ()
effectSpan Effects n ctx job kids stored
effects (((n () -> n ()) -> n ()) -> n ())
-> ((n () -> n ()) -> n ()) -> n ()
forall a b. (a -> b) -> a -> b
$ \n () -> n ()
inherit -> (n () -> (JobForceCancelled -> n ()) -> n ()
forall e a. Exception e => n a -> (e -> n a) -> n a
forall (m :: * -> *) e a.
(MonadCatch m, Exception e) =>
m a -> (e -> m a) -> m a
`catch` JobForceCancelled -> n ()
onForceCancel) (n () -> n ()) -> n () -> n ()
forall a b. (a -> b) -> a -> b
$ do
    result <-
      n () -> n (Either SomeException ())
forall (n :: * -> *) a.
MonadCatch n =>
n a -> n (Either SomeException a)
trySync
        (n () -> n (Either SomeException ()))
-> n () -> n (Either SomeException ())
forall a b. (a -> b) -> a -> b
$ HeartbeatGuard n job -> Batch n job -> n () -> n ()
forall (n :: * -> *) job a.
(MonadFork n, MonadMask n, MonadMonotonicTime n, MonadSTM n,
 MonadTime n) =>
HeartbeatGuard n job -> Batch n job -> n a -> n a
guardBatch
          HeartbeatGuard n job
guard
          Batch {batchJobs :: NonEmpty job
batchJobs = NonEmpty job
jobs, batchPending :: n [job]
batchPending = Handoff n job -> NonEmpty job -> n [job]
forall (n :: * -> *) job.
MonadSTM n =>
Handoff n job -> NonEmpty job -> n [job]
pendingJobs Handoff n job
handoff NonEmpty job
jobs, batchStart :: UTCTime
batchStart = UTCTime
startTime, batchInherit :: n () -> n ()
batchInherit = n () -> n ()
inherit}
        (n () -> n ()) -> n () -> n ()
forall a b. (a -> b) -> a -> b
$ do
          (job -> n ()) -> NonEmpty job -> n ()
forall (t :: * -> *) (f :: * -> *) a b.
(Foldable t, Applicative f) =>
(a -> f b) -> t a -> f ()
traverse_ (Run n ctx job kids stored -> job -> n ()
forall (n :: * -> *) ctx job kids stored.
MonadSTM n =>
Run n ctx job kids stored -> job -> n ()
reportClaimed Run n ctx job kids stored
run) NonEmpty job
jobs
          case Mode n ctx job kids stored
mode of
            -- The commit stays interruptible. The transaction contains the handler.
            SingleMode job -> n ()
transaction ->
              Run n ctx job kids stored
-> ((forall x. n x -> n x) -> n ())
-> (() -> Settled job)
-> (Run n ctx job kids stored -> () -> n ())
-> n ()
forall (n :: * -> *) ctx job kids stored a.
(MonadMask n, MonadSTM n) =>
Run n ctx job kids stored
-> ((forall x. n x -> n x) -> n a)
-> (a -> Settled job)
-> (Run n ctx job kids stored -> a -> n ())
-> n a
settleWith Run n ctx job kids stored
run (\forall x. n x -> n x
restore -> n () -> n ()
forall x. n x -> n x
restore (job -> n ()
transaction job
firstJob)) (Settled job -> () -> Settled job
forall a b. a -> b -> a
const ([job] -> Settled job
forall job. [job] -> Settled job
finalized [job
firstJob])) ((Run n ctx job kids stored -> () -> n ()) -> n ())
-> (Run n ctx job kids stored -> () -> n ()) -> n ()
forall a b. (a -> b) -> a -> b
$ \Run n ctx job kids stored
at () ->
                Run n ctx job kids stored -> UTCTime -> job -> n ()
forall (n :: * -> *) ctx job kids stored.
(MonadMask n, MonadSTM n, MonadTime n) =>
Run n ctx job kids stored -> UTCTime -> job -> n ()
reportSuccess Run n ctx job kids stored
at UTCTime
startTime job
firstJob
            BatchedMode NonEmpty job -> Callbacks n ctx job kids stored -> n ()
handler -> NonEmpty job -> Callbacks n ctx job kids stored -> n ()
handler NonEmpty job
jobs (Run n ctx job kids stored
-> NonEmpty job -> UTCTime -> Callbacks n ctx job kids stored
forall (n :: * -> *) ctx job kids stored.
(MonadMask n, MonadSTM n, MonadTime n) =>
Run n ctx job kids stored
-> NonEmpty job -> UTCTime -> Callbacks n ctx job kids stored
callbacks Run n ctx job kids stored
run NonEmpty job
jobs UTCTime
startTime)
    endTime <- getCurrentTime
    reportBatchOutcome run jobs startTime endTime result

-- ---------------------------------------------------------------------------
-- Settling
--
-- Settle: commit a statement, record what it accounted for, then fire its hooks.
-- Report: fire hooks only. Release: hand back the attempt, the nack. Finalize:
-- the force-cancel path.
-- ---------------------------------------------------------------------------

-- | An effect, in the run's context.
effect :: (Effects n ctx job kids stored -> ctx -> a) -> Run n ctx job kids stored -> a
effect :: forall (n :: * -> *) ctx job kids stored a.
(Effects n ctx job kids stored -> ctx -> a)
-> Run n ctx job kids stored -> a
effect Effects n ctx job kids stored -> ctx -> a
field Run n ctx job kids stored
run = Effects n ctx job kids stored -> ctx -> a
field (Run n ctx job kids stored -> Effects n ctx job kids stored
forall (n :: * -> *) ctx job kids stored.
Run n ctx job kids stored -> Effects n ctx job kids stored
runEffects Run n ctx job kids stored
run) (Run n ctx job kids stored -> ctx
forall (n :: * -> *) ctx job kids stored.
Run n ctx job kids stored -> ctx
runContext Run n ctx job kids stored
run)

report :: Run n ctx job kids stored -> Report job -> n ()
report :: forall (n :: * -> *) ctx job kids stored.
Run n ctx job kids stored -> Report job -> n ()
report = (Effects n ctx job kids stored -> ctx -> Report job -> n ())
-> Run n ctx job kids stored -> Report job -> n ()
forall (n :: * -> *) ctx job kids stored a.
(Effects n ctx job kids stored -> ctx -> a)
-> Run n ctx job kids stored -> a
effect Effects n ctx job kids stored -> ctx -> Report job -> n ()
forall (n :: * -> *) ctx job kids stored.
Effects n ctx job kids stored -> ctx -> Report job -> n ()
effectReport

logAt :: Run n ctx job kids stored -> LogLevel -> [job] -> Text -> n ()
logAt :: forall (n :: * -> *) ctx job kids stored.
Run n ctx job kids stored -> LogLevel -> [job] -> Text -> n ()
logAt Run n ctx job kids stored
run = Effects n ctx job kids stored -> LogLevel -> [job] -> Text -> n ()
forall (n :: * -> *) ctx job kids stored.
Effects n ctx job kids stored -> LogLevel -> [job] -> Text -> n ()
effectLog (Run n ctx job kids stored -> Effects n ctx job kids stored
forall (n :: * -> *) ctx job kids stored.
Run n ctx job kids stored -> Effects n ctx job kids stored
runEffects Run n ctx job kids stored
run)

-- | Commit a settle, record it, then run its hooks. The mask covers the commit,
-- the record and the deferral, so a signal that interrupts the hooks leaves them
-- to the batch's next report, under the ambient context. The commit may lift the
-- mask for its own run. A nested settle keeps the outermost deferral. Returns
-- what the commit did.
settleWith
  :: (MonadMask n, MonadSTM n)
  => Run n ctx job kids stored
  -> ((forall x. n x -> n x) -> n a)
  -> (a -> Settled job)
  -> (Run n ctx job kids stored -> a -> n ())
  -> n a
settleWith :: forall (n :: * -> *) ctx job kids stored a.
(MonadMask n, MonadSTM n) =>
Run n ctx job kids stored
-> ((forall x. n x -> n x) -> n a)
-> (a -> Settled job)
-> (Run n ctx job kids stored -> a -> n ())
-> n a
settleWith Run n ctx job kids stored
run (forall x. n x -> n x) -> n a
commit a -> Settled job
accounts Run n ctx job kids stored -> a -> n ()
hooks = do
  (result, outermost) <- ((forall x. n x -> n x) -> n (a, Bool)) -> n (a, Bool)
forall b. ((forall x. n x -> n x) -> n b) -> n b
forall (m :: * -> *) b.
MonadMask m =>
((forall a. m a -> m a) -> m b) -> m b
mask (((forall x. n x -> n x) -> n (a, Bool)) -> n (a, Bool))
-> ((forall x. n x -> n x) -> n (a, Bool)) -> n (a, Bool)
forall a b. (a -> b) -> a -> b
$ \forall x. n x -> n x
restore -> do
    result <- (forall x. n x -> n x) -> n a
commit n x -> n x
forall x. n x -> n x
restore
    let later = Run n ctx job kids stored -> a -> n ()
hooks (Effects n ctx job kids stored
-> Handoff n job -> Run n ctx job kids stored
forall (n :: * -> *) ctx job kids stored.
Effects n ctx job kids stored
-> Handoff n job -> Run n ctx job kids stored
ambientRun (Run n ctx job kids stored -> Effects n ctx job kids stored
forall (n :: * -> *) ctx job kids stored.
Run n ctx job kids stored -> Effects n ctx job kids stored
runEffects Run n ctx job kids stored
run) Handoff n job
handoff) a
result
        settled Progress n
progress =
          ( Maybe (n ()) -> Bool
forall a. Maybe a -> Bool
isNothing (Progress n -> Maybe (n ())
forall (n :: * -> *). Progress n -> Maybe (n ())
progressDeferred Progress n
progress)
          , (Handoff n job -> Settled job -> Progress n -> Progress n
forall (n :: * -> *) job.
Handoff n job -> Settled job -> Progress n -> Progress n
recorded Handoff n job
handoff (a -> Settled job
accounts a
result) Progress n
progress)
              { progressDeferred = progressDeferred progress <|> Just later
              }
          )
    (,) result <$> onProgress handoff settled
  hooks run result
  result <$ when outermost (clearDeferred handoff)
  where
    handoff :: Handoff n job
handoff = Run n ctx job kids stored -> Handoff n job
forall (n :: * -> *) ctx job kids stored.
Run n ctx job kids stored -> Handoff n job
runHandoff Run n ctx job kids stored
run

-- | 'settleWith' for a set known before the commit.
settle
  :: (MonadMask n, MonadSTM n)
  => Run n ctx job kids stored -> Settled job -> n a -> (Run n ctx job kids stored -> a -> n ()) -> n a
settle :: forall (n :: * -> *) ctx job kids stored a.
(MonadMask n, MonadSTM n) =>
Run n ctx job kids stored
-> Settled job
-> n a
-> (Run n ctx job kids stored -> a -> n ())
-> n a
settle Run n ctx job kids stored
run Settled job
settled n a
commit = Run n ctx job kids stored
-> ((forall x. n x -> n x) -> n a)
-> (a -> Settled job)
-> (Run n ctx job kids stored -> a -> n ())
-> n a
forall (n :: * -> *) ctx job kids stored a.
(MonadMask n, MonadSTM n) =>
Run n ctx job kids stored
-> ((forall x. n x -> n x) -> n a)
-> (a -> Settled job)
-> (Run n ctx job kids stored -> a -> n ())
-> n a
settleWith Run n ctx job kids stored
run (\forall x. n x -> n x
_ -> n a
commit) (Settled job -> a -> Settled job
forall a b. a -> b -> a
const Settled job
settled)

-- | Mark a job in one of the progress sets. Whether it was unmarked.
markOnce
  :: (MonadSTM n)
  => Handoff n job -> (Progress n -> Set JobId) -> (Set JobId -> Progress n -> Progress n) -> job -> n Bool
markOnce :: forall (n :: * -> *) job.
MonadSTM n =>
Handoff n job
-> (Progress n -> Set JobId)
-> (Set JobId -> Progress n -> Progress n)
-> job
-> n Bool
markOnce Handoff n job
handoff Progress n -> Set JobId
get Set JobId -> Progress n -> Progress n
put job
job = Handoff n job -> (Progress n -> (Bool, Progress n)) -> n Bool
forall (n :: * -> *) job a.
MonadSTM n =>
Handoff n job -> (Progress n -> (a, Progress n)) -> n a
onProgress Handoff n job
handoff ((Progress n -> (Bool, Progress n)) -> n Bool)
-> (Progress n -> (Bool, Progress n)) -> n Bool
forall a b. (a -> b) -> a -> b
$ \Progress n
progress ->
  (Bool -> Bool
not (JobId -> Set JobId -> Bool
forall a. Ord a => a -> Set a -> Bool
Set.member JobId
key (Progress n -> Set JobId
get Progress n
progress)), Set JobId -> Progress n -> Progress n
put (JobId -> Set JobId -> Set JobId
forall a. Ord a => a -> Set a -> Set a
Set.insert JobId
key (Progress n -> Set JobId
get Progress n
progress)) Progress n
progress)
  where
    key :: JobId
key = Handoff n job -> job -> JobId
forall (n :: * -> *) job. Handoff n job -> job -> JobId
handoffKey Handoff n job
handoff job
job

-- | Fire a job's claim report, once per batch. Left unmasked so the guard can
-- interrupt a blocked claim hook.
reportClaimed :: (MonadSTM n) => Run n ctx job kids stored -> job -> n ()
reportClaimed :: forall (n :: * -> *) ctx job kids stored.
MonadSTM n =>
Run n ctx job kids stored -> job -> n ()
reportClaimed Run n ctx job kids stored
run job
job = do
  fresh <- Handoff n job
-> (Progress n -> Set JobId)
-> (Set JobId -> Progress n -> Progress n)
-> job
-> n Bool
forall (n :: * -> *) job.
MonadSTM n =>
Handoff n job
-> (Progress n -> Set JobId)
-> (Set JobId -> Progress n -> Progress n)
-> job
-> n Bool
markOnce Handoff n job
handoff Progress n -> Set JobId
forall (n :: * -> *). Progress n -> Set JobId
progressClaimed (\Set JobId
ids Progress n
progress -> Progress n
progress {progressClaimed = ids}) job
job
  when fresh (report run (Claimed job (handoffStart handoff)))
  where
    handoff :: Handoff n job
handoff = Run n ctx job kids stored -> Handoff n job
forall (n :: * -> *) ctx job kids stored.
Run n ctx job kids stored -> Handoff n job
runHandoff Run n ctx job kids stored
run

-- | Fire a job's terminal report, once per batch, after its claim report. A batch
-- the guard refused, or interrupted before every claim hook ran, still pairs the
-- two. Marking and firing are one unit against a signal.
reportOnce :: (MonadMask n, MonadSTM n) => Run n ctx job kids stored -> job -> n () -> n ()
reportOnce :: forall (n :: * -> *) ctx job kids stored.
(MonadMask n, MonadSTM n) =>
Run n ctx job kids stored -> job -> n () -> n ()
reportOnce Run n ctx job kids stored
run job
job n ()
fire = n () -> n ()
forall a. n a -> n a
forall (m :: * -> *) a. MonadMask m => m a -> m a
mask_ (n () -> n ()) -> n () -> n ()
forall a b. (a -> b) -> a -> b
$ do
  fresh <- Handoff n job
-> (Progress n -> Set JobId)
-> (Set JobId -> Progress n -> Progress n)
-> job
-> n Bool
forall (n :: * -> *) job.
MonadSTM n =>
Handoff n job
-> (Progress n -> Set JobId)
-> (Set JobId -> Progress n -> Progress n)
-> job
-> n Bool
markOnce (Run n ctx job kids stored -> Handoff n job
forall (n :: * -> *) ctx job kids stored.
Run n ctx job kids stored -> Handoff n job
runHandoff Run n ctx job kids stored
run) Progress n -> Set JobId
forall (n :: * -> *). Progress n -> Set JobId
progressReported (\Set JobId
ids Progress n
progress -> Progress n
progress {progressReported = ids}) job
job
  when fresh (reportClaimed run job *> fire)

reportSuccess :: (MonadMask n, MonadSTM n, MonadTime n) => Run n ctx job kids stored -> UTCTime -> job -> n ()
reportSuccess :: forall (n :: * -> *) ctx job kids stored.
(MonadMask n, MonadSTM n, MonadTime n) =>
Run n ctx job kids stored -> UTCTime -> job -> n ()
reportSuccess Run n ctx job kids stored
run UTCTime
startTime job
job = Run n ctx job kids stored -> job -> n () -> n ()
forall (n :: * -> *) ctx job kids stored.
(MonadMask n, MonadSTM n) =>
Run n ctx job kids stored -> job -> n () -> n ()
reportOnce Run n ctx job kids stored
run job
job (n UTCTime
forall (m :: * -> *). MonadTime m => m UTCTime
getCurrentTime n UTCTime -> (UTCTime -> n ()) -> n ()
forall a b. n a -> (a -> n b) -> n b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= Run n ctx job kids stored -> Report job -> n ()
forall (n :: * -> *) ctx job kids stored.
Run n ctx job kids stored -> Report job -> n ()
report Run n ctx job kids stored
run (Report job -> n ()) -> (UTCTime -> Report job) -> UTCTime -> n ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. job -> UTCTime -> UTCTime -> Report job
forall job. job -> UTCTime -> UTCTime -> Report job
Succeeded job
job UTCTime
startTime)

-- | Report a failure's write.
reportFailed
  :: (MonadMask n, MonadSTM n)
  => Run n ctx job kids stored -> Text -> UTCTime -> UTCTime -> job -> Either Text Outcome -> n ()
reportFailed :: forall (n :: * -> *) ctx job kids stored.
(MonadMask n, MonadSTM n) =>
Run n ctx job kids stored
-> Text -> UTCTime -> UTCTime -> job -> Either Text Outcome -> n ()
reportFailed Run n ctx job kids stored
run Text
msg UTCTime
startTime UTCTime
endTime job
job = (Outcome -> n ()) -> Either Text Outcome -> n ()
forall (t :: * -> *) (f :: * -> *) a b.
(Foldable t, Applicative f) =>
(a -> f b) -> t a -> f ()
traverse_ (Run n ctx job kids stored -> job -> n () -> n ()
forall (n :: * -> *) ctx job kids stored.
(MonadMask n, MonadSTM n) =>
Run n ctx job kids stored -> job -> n () -> n ()
reportOnce Run n ctx job kids stored
run job
job (n () -> n ()) -> (Outcome -> n ()) -> Outcome -> n ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Run n ctx job kids stored -> Report job -> n ()
forall (n :: * -> *) ctx job kids stored.
Run n ctx job kids stored -> Report job -> n ()
report Run n ctx job kids stored
run (Report job -> n ()) -> (Outcome -> Report job) -> Outcome -> n ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Outcome -> Report job
toReport)
  where
    toReport :: Outcome -> Report job
toReport (TreeCancelled [JobId]
_) = job -> Text -> Report job
forall job. job -> Text -> Report job
Cancelled job
job Text
msg
    toReport Outcome
outcome = job -> Text -> UTCTime -> UTCTime -> Outcome -> Report job
forall job.
job -> Text -> UTCTime -> UTCTime -> Outcome -> Report job
Failed job
job Text
msg UTCTime
startTime UTCTime
endTime Outcome
outcome

-- | The settle operations a batch handler drives its jobs through, each from the
-- handler's context at the call.
callbacks
  :: (MonadMask n, MonadSTM n, MonadTime n)
  => Run n ctx job kids stored -> NonEmpty job -> UTCTime -> Callbacks n ctx job kids stored
callbacks :: forall (n :: * -> *) ctx job kids stored.
(MonadMask n, MonadSTM n, MonadTime n) =>
Run n ctx job kids stored
-> NonEmpty job -> UTCTime -> Callbacks n ctx job kids stored
callbacks Run n ctx job kids stored
base NonEmpty job
jobs UTCTime
startTime =
  Callbacks
    { callbackAck :: ctx -> job -> Maybe stored -> n ()
callbackAck = \ctx
ctx job
job Maybe stored
stored -> Text -> job -> n () -> n ()
unlessHandled Text
"ack" job
job (n () -> n ()) -> n () -> n ()
forall a b. (a -> b) -> a -> b
$ ctx -> (Run n ctx job kids stored -> n ()) -> n ()
within ctx
ctx ((Run n ctx job kids stored -> n ()) -> n ())
-> (Run n ctx job kids stored -> n ()) -> n ()
forall a b. (a -> b) -> a -> b
$ \Run n ctx job kids stored
run ->
        Run n ctx job kids stored
-> Settled job
-> n ()
-> (Run n ctx job kids stored -> () -> n ())
-> n ()
forall (n :: * -> *) ctx job kids stored a.
(MonadMask n, MonadSTM n) =>
Run n ctx job kids stored
-> Settled job
-> n a
-> (Run n ctx job kids stored -> a -> n ())
-> n a
settle Run n ctx job kids stored
run ([job] -> Settled job
forall job. [job] -> Settled job
finalized [job
job]) ((Effects n ctx job kids stored
 -> ctx -> job -> Maybe stored -> n ())
-> Run n ctx job kids stored -> job -> Maybe stored -> n ()
forall (n :: * -> *) ctx job kids stored a.
(Effects n ctx job kids stored -> ctx -> a)
-> Run n ctx job kids stored -> a
effect Effects n ctx job kids stored -> ctx -> job -> Maybe stored -> n ()
forall (n :: * -> *) ctx job kids stored.
Effects n ctx job kids stored -> ctx -> job -> Maybe stored -> n ()
effectAck Run n ctx job kids stored
run job
job Maybe stored
stored) ((Run n ctx job kids stored -> () -> n ()) -> n ())
-> (Run n ctx job kids stored -> () -> n ()) -> n ()
forall a b. (a -> b) -> a -> b
$ \Run n ctx job kids stored
at () ->
          Run n ctx job kids stored -> UTCTime -> job -> n ()
forall (n :: * -> *) ctx job kids stored.
(MonadMask n, MonadSTM n, MonadTime n) =>
Run n ctx job kids stored -> UTCTime -> job -> n ()
reportSuccess Run n ctx job kids stored
at UTCTime
startTime job
job
    , callbackAckAll :: ctx -> [(job, Maybe stored)] -> n ()
callbackAckAll = \ctx
ctx [(job, Maybe stored)]
pairs -> ctx -> (Run n ctx job kids stored -> n ()) -> n ()
within ctx
ctx ((Run n ctx job kids stored -> n ()) -> n ())
-> (Run n ctx job kids stored -> n ()) -> n ()
forall a b. (a -> b) -> a -> b
$ \Run n ctx job kids stored
run -> [(job, Maybe stored)] -> n [(job, Maybe stored)]
dropHandled [(job, Maybe stored)]
pairs n [(job, Maybe stored)] -> ([(job, Maybe stored)] -> n ()) -> n ()
forall a b. n a -> (a -> n b) -> n b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= Run n ctx job kids stored
-> UTCTime -> [(job, Maybe stored)] -> n ()
forall (n :: * -> *) ctx job kids stored.
(MonadMask n, MonadSTM n, MonadTime n) =>
Run n ctx job kids stored
-> UTCTime -> [(job, Maybe stored)] -> n ()
ackJobs Run n ctx job kids stored
run UTCTime
startTime
    , callbackFail :: ctx -> Failure -> job -> n ()
callbackFail = \ctx
ctx Failure
failure job
job -> Text -> job -> n () -> n ()
unlessHandled Text
"failure" job
job (n () -> n ()) -> n () -> n ()
forall a b. (a -> b) -> a -> b
$ ctx -> (Run n ctx job kids stored -> n ()) -> n ()
within ctx
ctx ((Run n ctx job kids stored -> n ()) -> n ())
-> (Run n ctx job kids stored -> n ()) -> n ()
forall a b. (a -> b) -> a -> b
$ \Run n ctx job kids stored
run ->
        Run n ctx job kids stored
-> NonEmpty job -> UTCTime -> Failure -> job -> n ()
forall (n :: * -> *) ctx job kids stored.
(MonadMask n, MonadSTM n, MonadTime n) =>
Run n ctx job kids stored
-> NonEmpty job -> UTCTime -> Failure -> job -> n ()
failJob Run n ctx job kids stored
run NonEmpty job
jobs UTCTime
startTime Failure
failure job
job
    , callbackNack :: ctx -> Maybe NominalDiffTime -> job -> n ()
callbackNack = \ctx
ctx Maybe NominalDiffTime
delay job
job -> Text -> job -> n () -> n ()
unlessHandled Text
"nack" job
job (n () -> n ()) -> n () -> n ()
forall a b. (a -> b) -> a -> b
$ ctx -> (Run n ctx job kids stored -> n ()) -> n ()
within ctx
ctx ((Run n ctx job kids stored -> n ()) -> n ())
-> (Run n ctx job kids stored -> n ()) -> n ()
forall a b. (a -> b) -> a -> b
$ \Run n ctx job kids stored
run -> Run n ctx job kids stored -> Maybe NominalDiffTime -> [job] -> n ()
forall (n :: * -> *) ctx job kids stored.
(MonadMask n, MonadSTM n) =>
Run n ctx job kids stored -> Maybe NominalDiffTime -> [job] -> n ()
releaseJobs Run n ctx job kids stored
run Maybe NominalDiffTime
delay [job
job]
    , callbackSpawn :: ctx -> job -> kids -> n ()
callbackSpawn = \ctx
ctx job
job kids
kids -> Text -> job -> n () -> n ()
unlessHandled Text
"spawn" job
job (n () -> n ()) -> n () -> n ()
forall a b. (a -> b) -> a -> b
$ ctx -> (Run n ctx job kids stored -> n ()) -> n ()
within ctx
ctx ((Run n ctx job kids stored -> n ()) -> n ())
-> (Run n ctx job kids stored -> n ()) -> n ()
forall a b. (a -> b) -> a -> b
$ \Run n ctx job kids stored
run ->
        Run n ctx job kids stored
-> Settled job
-> n ()
-> (Run n ctx job kids stored -> () -> n ())
-> n ()
forall (n :: * -> *) ctx job kids stored a.
(MonadMask n, MonadSTM n) =>
Run n ctx job kids stored
-> Settled job
-> n a
-> (Run n ctx job kids stored -> a -> n ())
-> n a
settle Run n ctx job kids stored
run ([job] -> Settled job
forall job. [job] -> Settled job
finalized [job
job]) ((Effects n ctx job kids stored -> ctx -> job -> kids -> n ())
-> Run n ctx job kids stored -> job -> kids -> n ()
forall (n :: * -> *) ctx job kids stored a.
(Effects n ctx job kids stored -> ctx -> a)
-> Run n ctx job kids stored -> a
effect Effects n ctx job kids stored -> ctx -> job -> kids -> n ()
forall (n :: * -> *) ctx job kids stored.
Effects n ctx job kids stored -> ctx -> job -> kids -> n ()
effectSpawn Run n ctx job kids stored
run job
job kids
kids) ((Run n ctx job kids stored -> () -> n ()) -> n ())
-> (Run n ctx job kids stored -> () -> n ()) -> n ()
forall a b. (a -> b) -> a -> b
$ \Run n ctx job kids stored
at () ->
          Run n ctx job kids stored -> UTCTime -> job -> n ()
forall (n :: * -> *) ctx job kids stored.
(MonadMask n, MonadSTM n, MonadTime n) =>
Run n ctx job kids stored -> UTCTime -> job -> n ()
reportSuccess Run n ctx job kids stored
at UTCTime
startTime job
job
    }
  where
    within :: ctx -> (Run n ctx job kids stored -> n ()) -> n ()
within ctx
ctx Run n ctx job kids stored -> n ()
body = Run n ctx job kids stored -> n ()
body Run n ctx job kids stored
base {runContext = ctx}
    handledIds :: n (Set JobId)
handledIds = Progress n -> Set JobId
forall (n :: * -> *). Progress n -> Set JobId
progressHandled (Progress n -> Set JobId) -> n (Progress n) -> n (Set JobId)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Handoff n job -> n (Progress n)
forall (n :: * -> *) job.
MonadSTM n =>
Handoff n job -> n (Progress n)
readProgress (Run n ctx job kids stored -> Handoff n job
forall (n :: * -> *) ctx job kids stored.
Run n ctx job kids stored -> Handoff n job
runHandoff Run n ctx job kids stored
base)
    unlessHandled :: Text -> job -> n () -> n ()
unlessHandled Text
label job
job n ()
act = do
      handled <- n (Set JobId)
handledIds
      if hasIdIn (runHandoff base) handled job
        then logAt base Warning [job] ("Ignoring " <> label <> " on a job its handler already finalized")
        else act
    dropHandled :: [(job, Maybe stored)] -> n [(job, Maybe stored)]
dropHandled [(job, Maybe stored)]
pairs = do
      handled <- n (Set JobId)
handledIds
      let (ignored, fresh) = partition (hasIdIn (runHandoff base) handled . fst) pairs
      unless (null ignored) $
        logAt base Warning (map fst ignored) "Ignoring bulk completion on jobs their handler already finalized"
      pure fresh

-- | Ack a bulk completion's fresh jobs in one statement. Only the commit knows which
-- jobs it acked and which had moved.
ackJobs
  :: (MonadMask n, MonadSTM n, MonadTime n) => Run n ctx job kids stored -> UTCTime -> [(job, Maybe stored)] -> n ()
ackJobs :: forall (n :: * -> *) ctx job kids stored.
(MonadMask n, MonadSTM n, MonadTime n) =>
Run n ctx job kids stored
-> UTCTime -> [(job, Maybe stored)] -> n ()
ackJobs Run n ctx job kids stored
_ UTCTime
_ [] = () -> n ()
forall a. a -> n a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
ackJobs Run n ctx job kids stored
run UTCTime
startTime [(job, Maybe stored)]
fresh =
  n ([job], [job]) -> n ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (n ([job], [job]) -> n ()) -> n ([job], [job]) -> n ()
forall a b. (a -> b) -> a -> b
$
    Run n ctx job kids stored
-> ((forall x. n x -> n x) -> n ([job], [job]))
-> (([job], [job]) -> Settled job)
-> (Run n ctx job kids stored -> ([job], [job]) -> n ())
-> n ([job], [job])
forall (n :: * -> *) ctx job kids stored a.
(MonadMask n, MonadSTM n) =>
Run n ctx job kids stored
-> ((forall x. n x -> n x) -> n a)
-> (a -> Settled job)
-> (Run n ctx job kids stored -> a -> n ())
-> n a
settleWith Run n ctx job kids stored
run (\forall x. n x -> n x
_ -> (Effects n ctx job kids stored
 -> ctx -> [(job, Maybe stored)] -> n ([job], [job]))
-> Run n ctx job kids stored
-> [(job, Maybe stored)]
-> n ([job], [job])
forall (n :: * -> *) ctx job kids stored a.
(Effects n ctx job kids stored -> ctx -> a)
-> Run n ctx job kids stored -> a
effect Effects n ctx job kids stored
-> ctx -> [(job, Maybe stored)] -> n ([job], [job])
forall (n :: * -> *) ctx job kids stored.
Effects n ctx job kids stored
-> ctx -> [(job, Maybe stored)] -> n ([job], [job])
effectAckAll Run n ctx job kids stored
run [(job, Maybe stored)]
fresh) (([job] -> [job] -> Settled job) -> ([job], [job]) -> Settled job
forall a b c. (a -> b -> c) -> (a, b) -> c
uncurry [job] -> [job] -> Settled job
forall job. [job] -> [job] -> Settled job
Settled) ((Run n ctx job kids stored -> ([job], [job]) -> n ())
 -> n ([job], [job]))
-> (Run n ctx job kids stored -> ([job], [job]) -> n ())
-> n ([job], [job])
forall a b. (a -> b) -> a -> b
$
      \Run n ctx job kids stored
at ([job]
done, [job]
reclaimed) -> do
        (job -> n ()) -> [job] -> n ()
forall (t :: * -> *) (f :: * -> *) a b.
(Foldable t, Applicative f) =>
(a -> f b) -> t a -> f ()
traverse_ (Run n ctx job kids stored -> UTCTime -> job -> n ()
forall (n :: * -> *) ctx job kids stored.
(MonadMask n, MonadSTM n, MonadTime n) =>
Run n ctx job kids stored -> UTCTime -> job -> n ()
reportSuccess Run n ctx job kids stored
at UTCTime
startTime) [job]
done
        Bool -> n () -> n ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
unless ([job] -> Bool
forall a. [a] -> Bool
forall (t :: * -> *) a. Foldable t => t a -> Bool
null [job]
reclaimed) (n () -> n ()) -> n () -> n ()
forall a b. (a -> b) -> a -> b
$ do
          Run n ctx job kids stored -> LogLevel -> [job] -> Text -> n ()
forall (n :: * -> *) ctx job kids stored.
Run n ctx job kids stored -> LogLevel -> [job] -> Text -> n ()
logAt Run n ctx job kids stored
at LogLevel
Info [job]
reclaimed (Text
"Jobs " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
unownedReason Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" during bulk completion, skipped")
          n (Set JobId) -> n ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (Run n ctx job kids stored -> Text -> [job] -> n (Set JobId)
forall (n :: * -> *) ctx job kids stored.
(MonadMask n, MonadSTM n) =>
Run n ctx job kids stored -> Text -> [job] -> n (Set JobId)
settleGoneJobs Run n ctx job kids stored
at Text
unownedReason [job]
reclaimed)

-- | Write one job's failure and report it with the batch siblings a tree cancel deleted.
failJob
  :: (MonadMask n, MonadSTM n, MonadTime n)
  => Run n ctx job kids stored -> NonEmpty job -> UTCTime -> Failure -> job -> n ()
failJob :: forall (n :: * -> *) ctx job kids stored.
(MonadMask n, MonadSTM n, MonadTime n) =>
Run n ctx job kids stored
-> NonEmpty job -> UTCTime -> Failure -> job -> n ()
failJob Run n ctx job kids stored
run NonEmpty job
jobs UTCTime
startTime failure :: Failure
failure@(Text
msg, FailureKind
_) job
job = do
  endTime <- n UTCTime
forall (m :: * -> *). MonadTime m => m UTCTime
getCurrentTime
  pending <- pendingJobs (runHandoff run) jobs
  let siblings = Handoff n job -> [job] -> job -> Either Text Outcome -> [job]
forall (n :: * -> *) job.
Handoff n job -> [job] -> job -> Either Text Outcome -> [job]
cancelledSiblings (Run n ctx job kids stored -> Handoff n job
forall (n :: * -> *) ctx job kids stored.
Run n ctx job kids stored -> Handoff n job
runHandoff Run n ctx job kids stored
run) [job]
pending job
job
  void $
    settleWith run (\forall x. n x -> n x
_ -> (Effects n ctx job kids stored
 -> ctx -> Failure -> job -> n (Either Text Outcome))
-> Run n ctx job kids stored
-> Failure
-> job
-> n (Either Text Outcome)
forall (n :: * -> *) ctx job kids stored a.
(Effects n ctx job kids stored -> ctx -> a)
-> Run n ctx job kids stored -> a
effect Effects n ctx job kids stored
-> ctx -> Failure -> job -> n (Either Text Outcome)
forall (n :: * -> *) ctx job kids stored.
Effects n ctx job kids stored
-> ctx -> Failure -> job -> n (Either Text Outcome)
effectFail Run n ctx job kids stored
run Failure
failure job
job) (finalized . (job :) . siblings) $
      \Run n ctx job kids stored
at Either Text Outcome
outcome -> do
        Run n ctx job kids stored
-> Text -> UTCTime -> UTCTime -> job -> Either Text Outcome -> n ()
forall (n :: * -> *) ctx job kids stored.
(MonadMask n, MonadSTM n) =>
Run n ctx job kids stored
-> Text -> UTCTime -> UTCTime -> job -> Either Text Outcome -> n ()
reportFailed Run n ctx job kids stored
at Text
msg UTCTime
startTime UTCTime
endTime job
job Either Text Outcome
outcome
        (job -> n ()) -> [job] -> n ()
forall (t :: * -> *) (f :: * -> *) a b.
(Foldable t, Applicative f) =>
(a -> f b) -> t a -> f ()
traverse_ (\job
sibling -> Run n ctx job kids stored -> job -> n () -> n ()
forall (n :: * -> *) ctx job kids stored.
(MonadMask n, MonadSTM n) =>
Run n ctx job kids stored -> job -> n () -> n ()
reportOnce Run n ctx job kids stored
at job
sibling (Run n ctx job kids stored -> Report job -> n ()
forall (n :: * -> *) ctx job kids stored.
Run n ctx job kids stored -> Report job -> n ()
report Run n ctx job kids stored
at (job -> Text -> Report job
forall job. job -> Text -> Report job
Cancelled job
sibling Text
msg))) (Either Text Outcome -> [job]
siblings Either Text Outcome
outcome)
        Run n ctx job kids stored -> [(job, Either Text Outcome)] -> n ()
forall (n :: * -> *) ctx job kids stored outcome.
(MonadMask n, MonadSTM n) =>
Run n ctx job kids stored -> [(job, Either Text outcome)] -> n ()
settleUnwritten Run n ctx job kids stored
at [(job
job, Either Text Outcome
outcome)]

-- | The pending jobs a tree cancel deleted, other than the one it failed.
cancelledSiblings :: Handoff n job -> [job] -> job -> Either Text Outcome -> [job]
cancelledSiblings :: forall (n :: * -> *) job.
Handoff n job -> [job] -> job -> Either Text Outcome -> [job]
cancelledSiblings Handoff n job
handoff [job]
pending job
job = \case
  Right (TreeCancelled [JobId]
deleted) ->
    (job -> Bool) -> [job] -> [job]
forall a. (a -> Bool) -> [a] -> [a]
filter (\job
sibling -> Handoff n job -> Set JobId -> job -> Bool
forall (n :: * -> *) job. Handoff n job -> Set JobId -> job -> Bool
hasIdIn Handoff n job
handoff ([JobId] -> Set JobId
forall a. Ord a => [a] -> Set a
Set.fromList [JobId]
deleted) job
sibling Bool -> Bool -> Bool
&& job -> JobId
key job
sibling JobId -> JobId -> Bool
forall a. Eq a => a -> a -> Bool
/= job -> JobId
key job
job) [job]
pending
  Either Text Outcome
_ -> []
  where
    key :: job -> JobId
key = Handoff n job -> job -> JobId
forall (n :: * -> *) job. Handoff n job -> job -> JobId
handoffKey Handoff n job
handoff

unownedReason :: Text
unownedReason :: Text
unownedReason = Text
"no longer claimed by this worker"

interruptedSiblingWarning :: Text
interruptedSiblingWarning :: Text
interruptedSiblingWarning = Text
"Releasing an interrupted batch sibling failed"

forceCancelledLog :: Text
forceCancelledLog :: Text
forceCancelledLog = Text
"Job(s) force-cancelled"

-- | The reason a force-cancelled job reports.
forceCancelledReason :: Text
forceCancelledReason :: Text
forceCancelledReason = Text
"force-cancelled"

-- | Delete the jobs a force-cancel flagged, report them, and hand back the attempt
-- the claim consumed for the batch siblings it interrupted.
finalizeForceCancelled
  :: (MonadMask n, MonadSTM n)
  => Run n ctx job kids stored
  -> NonEmpty job
  -> [JobId]
  -- ^ The jobs the cancel named.
  -> [JobId]
  -- ^ Jobs the same signal found unavailable. Reported without a nack.
  -> n ()
finalizeForceCancelled :: forall (n :: * -> *) ctx job kids stored.
(MonadMask n, MonadSTM n) =>
Run n ctx job kids stored
-> NonEmpty job -> [JobId] -> [JobId] -> n ()
finalizeForceCancelled Run n ctx job kids stored
run NonEmpty job
jobs [JobId]
cancelledIds [JobId]
goneIds = do
  Handoff n job -> n ()
forall (n :: * -> *) job. MonadSTM n => Handoff n job -> n ()
finishDeferred Handoff n job
handoff
  Run n ctx job kids stored -> LogLevel -> [job] -> Text -> n ()
forall (n :: * -> *) ctx job kids stored.
Run n ctx job kids stored -> LogLevel -> [job] -> Text -> n ()
logAt Run n ctx job kids stored
run LogLevel
Info (NonEmpty job -> [job]
forall a. NonEmpty a -> [a]
forall (t :: * -> *) a. Foldable t => t a -> [a]
toList NonEmpty job
jobs) Text
forceCancelledLog
  pending <- Handoff n job -> NonEmpty job -> n [job]
forall (n :: * -> *) job.
MonadSTM n =>
Handoff n job -> NonEmpty job -> n [job]
pendingJobs Handoff n job
handoff NonEmpty job
jobs
  -- The cancel can name a job the handler finalized after the cancel took effect.
  let settling = Handoff n job -> (job -> Bool) -> NonEmpty job -> [job]
forall (n :: * -> *) job.
Handoff n job -> (job -> Bool) -> NonEmpty job -> [job]
byIdDesc Handoff n job
handoff (Handoff n job -> Set JobId -> job -> Bool
forall (n :: * -> *) job. Handoff n job -> Set JobId -> job -> Bool
hasIdIn Handoff n job
handoff ([JobId] -> Set JobId
forall a. Ord a => [a] -> Set a
Set.fromList ([JobId]
cancelledIds [JobId] -> [JobId] -> [JobId]
forall a. Semigroup a => a -> a -> a
<> (job -> JobId) -> [job] -> [JobId]
forall a b. (a -> b) -> [a] -> [b]
map (Handoff n job -> job -> JobId
forall (n :: * -> *) job. Handoff n job -> job -> JobId
handoffKey Handoff n job
handoff) [job]
pending))) NonEmpty job
jobs
      goneSet = [JobId] -> Set JobId
forall a. Ord a => [a] -> Set a
Set.fromList [JobId]
goneIds
      deletable = (job -> Bool) -> [job] -> [job]
forall a. (a -> Bool) -> [a] -> [a]
filter (Bool -> Bool
not (Bool -> Bool) -> (job -> Bool) -> job -> Bool
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Handoff n job -> Set JobId -> job -> Bool
forall (n :: * -> *) job. Handoff n job -> Set JobId -> job -> Bool
hasIdIn Handoff n job
handoff Set JobId
goneSet) [job]
settling
  deleted <- deleteCancelled run deletable
  cancelled <- recordCancelled handoff (deleted <> Set.fromList cancelledIds)
  let (gone, interrupted) = partition (hasIdIn handoff cancelled) settling
      (unavailable, siblings) = partition (hasIdIn handoff goneSet) interrupted
  unless (null unavailable) $ logAt run Info unavailable ("Job(s) " <> unownedReason <> ", skipping retry")
  reportGoneJobs run unownedReason (gone <> unavailable)
  releaseOrWarn run "Releasing a force-cancel batch sibling failed" Nothing siblings
  where
    handoff :: Handoff n job
handoff = Run n ctx job kids stored -> Handoff n job
forall (n :: * -> *) ctx job kids stored.
Run n ctx job kids stored -> Handoff n job
runHandoff Run n ctx job kids stored
run

-- | Hand back the attempt the claim consumed for the jobs left unfinalized, in one
-- statement, and report whichever of them the release found under another claim.
-- A delay parks the jobs for that long.
releaseJobs :: (MonadMask n, MonadSTM n) => Run n ctx job kids stored -> Maybe NominalDiffTime -> [job] -> n ()
releaseJobs :: forall (n :: * -> *) ctx job kids stored.
(MonadMask n, MonadSTM n) =>
Run n ctx job kids stored -> Maybe NominalDiffTime -> [job] -> n ()
releaseJobs Run n ctx job kids stored
_ Maybe NominalDiffTime
_ [] = () -> n ()
forall a. a -> n a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
releaseJobs Run n ctx job kids stored
run Maybe NominalDiffTime
delay [job]
jobs =
  n (Set JobId) -> n ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (n (Set JobId) -> n ()) -> n (Set JobId) -> n ()
forall a b. (a -> b) -> a -> b
$ Run n ctx job kids stored
-> Settled job
-> n (Set JobId)
-> (Run n ctx job kids stored -> Set JobId -> n ())
-> n (Set JobId)
forall (n :: * -> *) ctx job kids stored a.
(MonadMask n, MonadSTM n) =>
Run n ctx job kids stored
-> Settled job
-> n a
-> (Run n ctx job kids stored -> a -> n ())
-> n a
settle Run n ctx job kids stored
run ([job] -> Settled job
forall job. [job] -> Settled job
finalized [job]
jobs) ((Effects n ctx job kids stored
 -> ctx -> Maybe NominalDiffTime -> [job] -> n (Set JobId))
-> Run n ctx job kids stored
-> Maybe NominalDiffTime
-> [job]
-> n (Set JobId)
forall (n :: * -> *) ctx job kids stored a.
(Effects n ctx job kids stored -> ctx -> a)
-> Run n ctx job kids stored -> a
effect Effects n ctx job kids stored
-> ctx -> Maybe NominalDiffTime -> [job] -> n (Set JobId)
forall (n :: * -> *) ctx job kids stored.
Effects n ctx job kids stored
-> ctx -> Maybe NominalDiffTime -> [job] -> n (Set JobId)
effectRelease Run n ctx job kids stored
run Maybe NominalDiffTime
delay [job]
jobs) ((Run n ctx job kids stored -> Set JobId -> n ()) -> n (Set JobId))
-> (Run n ctx job kids stored -> Set JobId -> n ())
-> n (Set JobId)
forall a b. (a -> b) -> a -> b
$ \Run n ctx job kids stored
at Set JobId
released ->
    Run n ctx job kids stored
-> [(job, Either Text (ZonkAny 0))] -> n ()
forall (n :: * -> *) ctx job kids stored outcome.
(MonadMask n, MonadSTM n) =>
Run n ctx job kids stored -> [(job, Either Text outcome)] -> n ()
settleUnwritten Run n ctx job kids stored
at [(job
job, Text -> Either Text (ZonkAny 0)
forall a b. a -> Either a b
Left Text
unownedReason) | job
job <- [job]
jobs, Bool -> Bool
not (Handoff n job -> Set JobId -> job -> Bool
forall (n :: * -> *) job. Handoff n job -> Set JobId -> job -> Bool
hasIdIn (Run n ctx job kids stored -> Handoff n job
forall (n :: * -> *) ctx job kids stored.
Run n ctx job kids stored -> Handoff n job
runHandoff Run n ctx job kids stored
run) Set JobId
released job
job)]

-- | 'releaseJobs' on an unwinding path. A failure is logged as a warning.
releaseOrWarn
  :: (MonadMask n, MonadSTM n) => Run n ctx job kids stored -> Text -> Maybe NominalDiffTime -> [job] -> n ()
releaseOrWarn :: forall (n :: * -> *) ctx job kids stored.
(MonadMask n, MonadSTM n) =>
Run n ctx job kids stored
-> Text -> Maybe NominalDiffTime -> [job] -> n ()
releaseOrWarn Run n ctx job kids stored
run Text
warning Maybe NominalDiffTime
delay [job]
jobs = Run n ctx job kids stored -> [job] -> Text -> () -> n () -> n ()
forall (n :: * -> *) ctx job kids stored a.
MonadCatch n =>
Run n ctx job kids stored -> [job] -> Text -> a -> n a -> n a
warnOn Run n ctx job kids stored
run [job]
jobs Text
warning () (Run n ctx job kids stored -> Maybe NominalDiffTime -> [job] -> n ()
forall (n :: * -> *) ctx job kids stored.
(MonadMask n, MonadSTM n) =>
Run n ctx job kids stored -> Maybe NominalDiffTime -> [job] -> n ()
releaseJobs Run n ctx job kids stored
run Maybe NominalDiffTime
delay [job]
jobs)

-- | Run an action, logging a warning and returning @fallback@ when it fails.
warnOn :: (MonadCatch n) => Run n ctx job kids stored -> [job] -> Text -> a -> n a -> n a
warnOn :: forall (n :: * -> *) ctx job kids stored a.
MonadCatch n =>
Run n ctx job kids stored -> [job] -> Text -> a -> n a -> n a
warnOn Run n ctx job kids stored
run [job]
jobs Text
label a
fallback n a
act =
  n a -> n (Either SomeException a)
forall (n :: * -> *) a.
MonadCatch n =>
n a -> n (Either SomeException a)
trySync n a
act
    n (Either SomeException a)
-> (Either SomeException a -> n a) -> n a
forall a b. n a -> (a -> n b) -> n b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= (SomeException -> n a)
-> (a -> n a) -> Either SomeException a -> n a
forall a c b. (a -> c) -> (b -> c) -> Either a b -> c
either (\SomeException
exception -> a
fallback a -> n () -> n a
forall a b. a -> n b -> n a
forall (f :: * -> *) a b. Functor f => a -> f b -> f a
<$ Run n ctx job kids stored -> LogLevel -> [job] -> Text -> n ()
forall (n :: * -> *) ctx job kids stored.
Run n ctx job kids stored -> LogLevel -> [job] -> Text -> n ()
logAt Run n ctx job kids stored
run LogLevel
Warning [job]
jobs (Text
label Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
": " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> SomeException -> Text
displayEx SomeException
exception)) a -> n a
forall a. a -> n a
forall (f :: * -> *) a. Applicative f => a -> f a
pure

-- | Delete whichever of @jobs@ a force-cancel flagged, returning the ids deleted. A
-- failure is logged as a warning.
deleteCancelled :: (MonadCatch n) => Run n ctx job kids stored -> [job] -> n (Set JobId)
deleteCancelled :: forall (n :: * -> *) ctx job kids stored.
MonadCatch n =>
Run n ctx job kids stored -> [job] -> n (Set JobId)
deleteCancelled Run n ctx job kids stored
run [job]
jobs =
  Run n ctx job kids stored
-> [job] -> Text -> Set JobId -> n (Set JobId) -> n (Set JobId)
forall (n :: * -> *) ctx job kids stored a.
MonadCatch n =>
Run n ctx job kids stored -> [job] -> Text -> a -> n a -> n a
warnOn Run n ctx job kids stored
run [job]
jobs Text
"Deleting force-cancelled jobs failed" Set JobId
forall a. Monoid a => a
mempty ((Effects n ctx job kids stored -> ctx -> [job] -> n (Set JobId))
-> Run n ctx job kids stored -> [job] -> n (Set JobId)
forall (n :: * -> *) ctx job kids stored a.
(Effects n ctx job kids stored -> ctx -> a)
-> Run n ctx job kids stored -> a
effect Effects n ctx job kids stored -> ctx -> [job] -> n (Set JobId)
forall (n :: * -> *) ctx job kids stored.
Effects n ctx job kids stored -> ctx -> [job] -> n (Set JobId)
effectDeleteCancelled Run n ctx job kids stored
run [job]
jobs)

-- | Settle the jobs a failure could not be written for, each under its own reason.
settleUnwritten :: (MonadMask n, MonadSTM n) => Run n ctx job kids stored -> [(job, Either Text outcome)] -> n ()
settleUnwritten :: forall (n :: * -> *) ctx job kids stored outcome.
(MonadMask n, MonadSTM n) =>
Run n ctx job kids stored -> [(job, Either Text outcome)] -> n ()
settleUnwritten Run n ctx job kids stored
run [(job, Either Text outcome)]
pairs =
  ((Text, [job]) -> n ()) -> [(Text, [job])] -> n ()
forall (t :: * -> *) (f :: * -> *) a b.
(Foldable t, Applicative f) =>
(a -> f b) -> t a -> f ()
traverse_ (Text, [job]) -> n ()
settleOne (Map Text [job] -> [(Text, [job])]
forall k a. Map k a -> [(k, a)]
Map.toList (([job] -> [job] -> [job]) -> [(Text, [job])] -> Map Text [job]
forall k a. Ord k => (a -> a -> a) -> [(k, a)] -> Map k a
Map.fromListWith [job] -> [job] -> [job]
forall a. Semigroup a => a -> a -> a
(<>) [(Text
reason, [job
job]) | (job
job, Left Text
reason) <- [(job, Either Text outcome)]
pairs]))
  where
    settleOne :: (Text, [job]) -> n ()
settleOne (Text
reason, [job]
jobs) = do
      cancelled <- Run n ctx job kids stored -> Text -> [job] -> n (Set JobId)
forall (n :: * -> *) ctx job kids stored.
(MonadMask n, MonadSTM n) =>
Run n ctx job kids stored -> Text -> [job] -> n (Set JobId)
settleGoneJobs Run n ctx job kids stored
run Text
reason [job]
jobs
      let (forced, unavailable) = partition (hasIdIn (runHandoff run) cancelled) jobs
      unless (null forced) $ logAt run Info forced forceCancelledLog
      unless (null unavailable) $ logAt run Warning unavailable ("Job(s) " <> reason)

-- | Delete whichever of @jobs@ a force-cancel flagged and report them all, as one
-- settle. Returns the ids deleted.
settleGoneJobs :: (MonadMask n, MonadSTM n) => Run n ctx job kids stored -> Text -> [job] -> n (Set JobId)
settleGoneJobs :: forall (n :: * -> *) ctx job kids stored.
(MonadMask n, MonadSTM n) =>
Run n ctx job kids stored -> Text -> [job] -> n (Set JobId)
settleGoneJobs Run n ctx job kids stored
_ Text
_ [] = Set JobId -> n (Set JobId)
forall a. a -> n a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Set JobId
forall a. Monoid a => a
mempty
settleGoneJobs Run n ctx job kids stored
run Text
reason [job]
jobs =
  Run n ctx job kids stored
-> ((forall x. n x -> n x) -> n (Set JobId))
-> (Set JobId -> Settled job)
-> (Run n ctx job kids stored -> Set JobId -> n ())
-> n (Set JobId)
forall (n :: * -> *) ctx job kids stored a.
(MonadMask n, MonadSTM n) =>
Run n ctx job kids stored
-> ((forall x. n x -> n x) -> n a)
-> (a -> Settled job)
-> (Run n ctx job kids stored -> a -> n ())
-> n a
settleWith Run n ctx job kids stored
run (\forall x. n x -> n x
_ -> Run n ctx job kids stored -> [job] -> n (Set JobId)
forall (n :: * -> *) ctx job kids stored.
MonadCatch n =>
Run n ctx job kids stored -> [job] -> n (Set JobId)
deleteCancelled Run n ctx job kids stored
run [job]
jobs) (Settled job -> Set JobId -> Settled job
forall a b. a -> b -> a
const ([job] -> Settled job
forall job. [job] -> Settled job
finalized [job]
jobs)) ((Run n ctx job kids stored -> Set JobId -> n ()) -> n (Set JobId))
-> (Run n ctx job kids stored -> Set JobId -> n ())
-> n (Set JobId)
forall a b. (a -> b) -> a -> b
$ \Run n ctx job kids stored
at Set JobId
cancelled -> do
    n (Set JobId) -> n ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (Handoff n job -> Set JobId -> n (Set JobId)
forall (n :: * -> *) job.
MonadSTM n =>
Handoff n job -> Set JobId -> n (Set JobId)
recordCancelled (Run n ctx job kids stored -> Handoff n job
forall (n :: * -> *) ctx job kids stored.
Run n ctx job kids stored -> Handoff n job
runHandoff Run n ctx job kids stored
at) Set JobId
cancelled)
    Run n ctx job kids stored -> Text -> [job] -> n ()
forall (n :: * -> *) ctx job kids stored.
(MonadMask n, MonadSTM n) =>
Run n ctx job kids stored -> Text -> [job] -> n ()
reportGoneJobs Run n ctx job kids stored
at Text
reason [job]
jobs

-- | Report jobs this worker can no longer act on. A job a force-cancel deleted is
-- reported as cancelled. Each is recorded against the handoff.
reportGoneJobs :: (MonadMask n, MonadSTM n) => Run n ctx job kids stored -> Text -> [job] -> n ()
reportGoneJobs :: forall (n :: * -> *) ctx job kids stored.
(MonadMask n, MonadSTM n) =>
Run n ctx job kids stored -> Text -> [job] -> n ()
reportGoneJobs Run n ctx job kids stored
run Text
reason [job]
jobs = do
  Handoff n job -> Settled job -> n ()
forall (n :: * -> *) job.
MonadSTM n =>
Handoff n job -> Settled job -> n ()
record Handoff n job
handoff ([job] -> Settled job
forall job. [job] -> Settled job
finalized [job]
jobs)
  cancelled <- Progress n -> Set JobId
forall (n :: * -> *). Progress n -> Set JobId
progressCancelled (Progress n -> Set JobId) -> n (Progress n) -> n (Set JobId)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Handoff n job -> n (Progress n)
forall (n :: * -> *) job.
MonadSTM n =>
Handoff n job -> n (Progress n)
readProgress Handoff n job
handoff
  traverse_ (reportOne cancelled) jobs
  where
    handoff :: Handoff n job
handoff = Run n ctx job kids stored -> Handoff n job
forall (n :: * -> *) ctx job kids stored.
Run n ctx job kids stored -> Handoff n job
runHandoff Run n ctx job kids stored
run
    reportOne :: Set JobId -> job -> n ()
reportOne Set JobId
cancelled job
job =
      Run n ctx job kids stored -> job -> n () -> n ()
forall (n :: * -> *) ctx job kids stored.
(MonadMask n, MonadSTM n) =>
Run n ctx job kids stored -> job -> n () -> n ()
reportOnce Run n ctx job kids stored
run job
job (n () -> n ()) -> (Report job -> n ()) -> Report job -> n ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Run n ctx job kids stored -> Report job -> n ()
forall (n :: * -> *) ctx job kids stored.
Run n ctx job kids stored -> Report job -> n ()
report Run n ctx job kids stored
run (Report job -> n ()) -> Report job -> n ()
forall a b. (a -> b) -> a -> b
$
        if Handoff n job -> Set JobId -> job -> Bool
forall (n :: * -> *) job. Handoff n job -> Set JobId -> job -> Bool
hasIdIn Handoff n job
handoff Set JobId
cancelled job
job then job -> Text -> Report job
forall job. job -> Text -> Report job
Cancelled job
job Text
forceCancelledReason else job -> Text -> Report job
forall job. job -> Text -> Report job
Unavailable job
job Text
reason

-- | Interpret a finished batch. Warn when the handler left jobs unfinalized. Skip
-- retry for gone or nacked jobs. Fail the rest.
reportBatchOutcome
  :: (MonadMask n, MonadSTM n)
  => Run n ctx job kids stored
  -> NonEmpty job
  -> UTCTime
  -> UTCTime
  -> Either SomeException ()
  -> n ()
reportBatchOutcome :: forall (n :: * -> *) ctx job kids stored.
(MonadMask n, MonadSTM n) =>
Run n ctx job kids stored
-> NonEmpty job
-> UTCTime
-> UTCTime
-> Either SomeException ()
-> n ()
reportBatchOutcome Run n ctx job kids stored
run NonEmpty job
jobs UTCTime
startTime UTCTime
endTime Either SomeException ()
outcome = do
  Handoff n job -> n ()
forall (n :: * -> *) job. MonadSTM n => Handoff n job -> n ()
finishDeferred Handoff n job
handoff
  unhandled <- Handoff n job -> NonEmpty job -> n [job]
forall (n :: * -> *) job.
MonadSTM n =>
Handoff n job -> NonEmpty job -> n [job]
pendingJobs Handoff n job
handoff NonEmpty job
jobs
  let splitNamed [JobId]
ids =
        let named :: Set JobId
named = [JobId] -> Set JobId
forall a. Ord a => [a] -> Set a
Set.fromList [JobId]
ids
            ([job]
theirs, [job]
siblings) = (job -> Bool) -> [job] -> ([job], [job])
forall a. (a -> Bool) -> [a] -> ([a], [a])
partition (Handoff n job -> Set JobId -> job -> Bool
forall (n :: * -> *) job. Handoff n job -> Set JobId -> job -> Bool
hasIdIn Handoff n job
handoff Set JobId
named) [job]
unhandled
         in ((job -> Bool) -> NonEmpty job -> Bool
forall (t :: * -> *) a. Foldable t => (a -> Bool) -> t a -> Bool
any (Handoff n job -> Set JobId -> job -> Bool
forall (n :: * -> *) job. Handoff n job -> Set JobId -> job -> Bool
hasIdIn Handoff n job
handoff Set JobId
named) NonEmpty job
jobs, [job]
theirs, [job]
siblings)
  case outcome of
    Right () ->
      Bool -> n () -> n ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
unless ([job] -> Bool
forall a. [a] -> Bool
forall (t :: * -> *) a. Foldable t => t a -> Bool
null [job]
unhandled) (n () -> n ()) -> n () -> n ()
forall a b. (a -> b) -> a -> b
$
        Run n ctx job kids stored -> LogLevel -> [job] -> Text -> n ()
forall (n :: * -> *) ctx job kids stored.
Run n ctx job kids stored -> LogLevel -> [job] -> Text -> n ()
logAt Run n ctx job kids stored
run LogLevel
Warning [job]
unhandled Text
"Handler left jobs unfinalized, will reprocess"
    Left SomeException
exc
      | Just (JobGoneException Text
reason [JobId]
gone) <- SomeException -> Maybe JobGoneException
forall e. Exception e => SomeException -> Maybe e
fromException SomeException
exc -> do
          Run n ctx job kids stored -> LogLevel -> [job] -> Text -> n ()
forall (n :: * -> *) ctx job kids stored.
Run n ctx job kids stored -> LogLevel -> [job] -> Text -> n ()
logAt Run n ctx job kids stored
run LogLevel
Info (NonEmpty job -> [job]
forall a. NonEmpty a -> [a]
forall (t :: * -> *) a. Foldable t => t a -> [a]
toList NonEmpty job
jobs) (Text
"Job(s) " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
reason Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
", skipping retry" Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> [JobId] -> Text
namedJobIds [JobId]
gone)
          -- An exception naming no job speaks for none of them. The remainder keeps
          -- its attempt.
          let (Bool
inBatch, [job]
jobsGone, [job]
siblings) = [JobId] -> (Bool, [job], [job])
splitNamed [JobId]
gone
          n (Set JobId) -> n ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (Run n ctx job kids stored -> Text -> [job] -> n (Set JobId)
forall (n :: * -> *) ctx job kids stored.
(MonadMask n, MonadSTM n) =>
Run n ctx job kids stored -> Text -> [job] -> n (Set JobId)
settleGoneJobs Run n ctx job kids stored
run Text
reason [job]
jobsGone)
          Bool -> n () -> n ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when Bool
inBatch (n () -> n ()) -> n () -> n ()
forall a b. (a -> b) -> a -> b
$ Run n ctx job kids stored
-> Text -> Maybe NominalDiffTime -> [job] -> n ()
forall (n :: * -> *) ctx job kids stored.
(MonadMask n, MonadSTM n) =>
Run n ctx job kids stored
-> Text -> Maybe NominalDiffTime -> [job] -> n ()
releaseOrWarn Run n ctx job kids stored
run Text
interruptedSiblingWarning Maybe NominalDiffTime
forall a. Maybe a
Nothing [job]
siblings
      | Just (JobScopedFailure JobException
inner [JobId]
scoped) <- SomeException -> Maybe JobScopedFailure
forall e. Exception e => SomeException -> Maybe e
fromException SomeException
exc -> do
          let (Bool
inBatch, [job]
theirs, [job]
siblings) = [JobId] -> (Bool, [job], [job])
splitNamed [JobId]
scoped
              failure :: Failure
failure = SomeException -> Failure
classifyException (JobException -> SomeException
forall e. Exception e => e -> SomeException
toException JobException
inner)
          if Bool
inBatch
            then do
              Failure -> [job] -> n ()
failJobs Failure
failure [job]
theirs
              Run n ctx job kids stored
-> Text -> Maybe NominalDiffTime -> [job] -> n ()
forall (n :: * -> *) ctx job kids stored.
(MonadMask n, MonadSTM n) =>
Run n ctx job kids stored
-> Text -> Maybe NominalDiffTime -> [job] -> n ()
releaseOrWarn Run n ctx job kids stored
run Text
interruptedSiblingWarning Maybe NominalDiffTime
forall a. Maybe a
Nothing [job]
siblings
            else Failure -> [job] -> n ()
failJobs Failure
failure [job]
unhandled
      | Just (JobNackException Maybe NominalDiffTime
delay) <- SomeException -> Maybe JobNackException
forall e. Exception e => SomeException -> Maybe e
fromException SomeException
exc -> do
          -- Hand back the attempt the claim consumed for every job the handler
          -- left unfinalized.
          Run n ctx job kids stored
-> Text -> Maybe NominalDiffTime -> [job] -> n ()
forall (n :: * -> *) ctx job kids stored.
(MonadMask n, MonadSTM n) =>
Run n ctx job kids stored
-> Text -> Maybe NominalDiffTime -> [job] -> n ()
releaseOrWarn Run n ctx job kids stored
run Text
"Handing back a nacked batch's attempt failed" Maybe NominalDiffTime
delay [job]
unhandled
          Run n ctx job kids stored -> LogLevel -> [job] -> Text -> n ()
forall (n :: * -> *) ctx job kids stored.
Run n ctx job kids stored -> LogLevel -> [job] -> Text -> n ()
logAt Run n ctx job kids stored
run LogLevel
Info (NonEmpty job -> [job]
forall a. NonEmpty a -> [a]
forall (t :: * -> *) a. Foldable t => t a -> [a]
toList NonEmpty job
jobs) Text
"Job(s) nacked, will be reprocessed"
      | Bool
otherwise -> do
          -- Fail the jobs the handler did not finalize, in a separate transaction.
          let failure :: Failure
failure@(Text
reason, FailureKind
_) = SomeException -> Failure
classifyException SomeException
exc
          Bool -> n () -> n ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when ([job] -> Bool
forall a. [a] -> Bool
forall (t :: * -> *) a. Foldable t => t a -> Bool
null [job]
unhandled) (n () -> n ()) -> n () -> n ()
forall a b. (a -> b) -> a -> b
$
            Run n ctx job kids stored -> LogLevel -> [job] -> Text -> n ()
forall (n :: * -> *) ctx job kids stored.
Run n ctx job kids stored -> LogLevel -> [job] -> Text -> n ()
logAt Run n ctx job kids stored
run LogLevel
Warning (NonEmpty job -> [job]
forall a. NonEmpty a -> [a]
forall (t :: * -> *) a. Foldable t => t a -> [a]
toList NonEmpty job
jobs) (Text
"Handler stopped with its jobs already finalized: " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
reason)
          Failure -> [job] -> n ()
failJobs Failure
failure [job]
unhandled
  where
    handoff :: Handoff n job
handoff = Run n ctx job kids stored -> Handoff n job
forall (n :: * -> *) ctx job kids stored.
Run n ctx job kids stored -> Handoff n job
runHandoff Run n ctx job kids stored
run
    failJobs :: Failure -> [job] -> n ()
failJobs failure :: Failure
failure@(Text
reason, FailureKind
kind) [job]
targets = do
      -- A tree or branch cancel acts on the whole tree.
      unowned <- if FailureKind -> Bool
cancelsTree FailureKind
kind then Handoff n job -> NonEmpty job -> n [job]
forall (n :: * -> *) job.
MonadSTM n =>
Handoff n job -> NonEmpty job -> n [job]
unownedJobs Handoff n job
handoff NonEmpty job
jobs else [job] -> n [job]
forall a. a -> n a
forall (f :: * -> *) a. Applicative f => a -> f a
pure []
      void
        $ settle
          run
          (finalized targets)
          (effect effectFailAll run failure targets unowned)
        $ \Run n ctx job kids stored
committed [(job, Either Text Outcome)]
outcomes -> do
          ((job, Either Text Outcome) -> n ())
-> [(job, Either Text Outcome)] -> n ()
forall (t :: * -> *) (f :: * -> *) a b.
(Foldable t, Applicative f) =>
(a -> f b) -> t a -> f ()
traverse_ (Run n ctx job kids stored
-> Text -> (job, Either Text Outcome) -> n ()
reportOne Run n ctx job kids stored
committed Text
reason) [(job, Either Text Outcome)]
outcomes
          Run n ctx job kids stored -> [(job, Either Text Outcome)] -> n ()
forall (n :: * -> *) ctx job kids stored outcome.
(MonadMask n, MonadSTM n) =>
Run n ctx job kids stored -> [(job, Either Text outcome)] -> n ()
settleUnwritten Run n ctx job kids stored
committed [(job, Either Text Outcome)]
outcomes
    reportOne :: Run n ctx job kids stored
-> Text -> (job, Either Text Outcome) -> n ()
reportOne Run n ctx job kids stored
at Text
reason (job
job, Either Text Outcome
write) =
      Run n ctx job kids stored -> [job] -> Text -> () -> n () -> n ()
forall (n :: * -> *) ctx job kids stored a.
MonadCatch n =>
Run n ctx job kids stored -> [job] -> Text -> a -> n a -> n a
warnOn Run n ctx job kids stored
at [job
job] Text
"Reporting a job's failure failed" () (Run n ctx job kids stored
-> Text -> UTCTime -> UTCTime -> job -> Either Text Outcome -> n ()
forall (n :: * -> *) ctx job kids stored.
(MonadMask n, MonadSTM n) =>
Run n ctx job kids stored
-> Text -> UTCTime -> UTCTime -> job -> Either Text Outcome -> n ()
reportFailed Run n ctx job kids stored
at Text
reason UTCTime
startTime UTCTime
endTime job
job Either Text Outcome
write)

-- | Classify a handler exception into an error message and failure disposition.
classifyException :: SomeException -> Failure
classifyException :: SomeException -> Failure
classifyException SomeException
exception
  | Just (Retryable (JobRetryableException Text
msg Maybe NominalDiffTime
delay)) <- SomeException -> Maybe JobException
forall e. Exception e => SomeException -> Maybe e
fromException SomeException
exception =
      (Text
msg, FailureKind
-> (NominalDiffTime -> FailureKind)
-> Maybe NominalDiffTime
-> FailureKind
forall b a. b -> (a -> b) -> Maybe a -> b
maybe FailureKind
RetryFailure NominalDiffTime -> FailureKind
RetryAfterFailure Maybe NominalDiffTime
delay)
  | Just (Permanent (JobPermanentException Text
msg)) <- SomeException -> Maybe JobException
forall e. Exception e => SomeException -> Maybe e
fromException SomeException
exception = (Text
msg, FailureKind
PermanentFailure)
  | Just (TreeCancel (TreeCancelException Text
msg)) <- SomeException -> Maybe JobException
forall e. Exception e => SomeException -> Maybe e
fromException SomeException
exception = (Text
msg, FailureKind
TreeCancelFailure)
  | Just (BranchCancel (BranchCancelException Text
msg)) <- SomeException -> Maybe JobException
forall e. Exception e => SomeException -> Maybe e
fromException SomeException
exception = (Text
msg, FailureKind
BranchCancelFailure)
  | Just (ParsingException Text
msg) <- SomeException -> Maybe ParsingException
forall e. Exception e => SomeException -> Maybe e
fromException SomeException
exception = (Text
msg, FailureKind
PermanentFailure)
  | Just (JobDeadlineExceeded Text
msg) <- SomeException -> Maybe JobDeadlineExceeded
forall e. Exception e => SomeException -> Maybe e
fromException SomeException
exception = (Text
msg, FailureKind
RetryFailure)
  | Bool
otherwise = (SomeException -> Text
displayEx SomeException
exception, FailureKind
RetryFailure)

-- | Whether a failure deletes a job tree or updates the job claim.
cancelsTree :: FailureKind -> Bool
cancelsTree :: FailureKind -> Bool
cancelsTree FailureKind
kind = FailureKind
kind FailureKind -> [FailureKind] -> Bool
forall a. Eq a => a -> [a] -> Bool
forall (t :: * -> *) a. (Foldable t, Eq a) => a -> t a -> Bool
`elem` [FailureKind
TreeCancelFailure, FailureKind
BranchCancelFailure]