Safe Haskell	None
Language	GHC2024

Arbiter.Worker.Batch

Contents

The pool's side
Running a batch

Description

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.

Synopsis

data Effects (n :: Type -> Type) ctx job kids stored = Effects {
- effectAmbient :: ctx
- effectSpan :: ((n () -> n ()) -> n ()) -> n ()
- effectAck :: ctx -> job -> Maybe stored -> n ()
- effectAckAll :: ctx -> [(job, Maybe stored)] -> n ([job], [job])
- effectFail :: ctx -> Failure -> job -> n (Either Text Outcome)
- effectFailAll :: ctx -> Failure -> [job] -> [job] -> n [(job, Either Text Outcome)]
- effectDeleteCancelled :: ctx -> [job] -> n (Set JobId)
- effectRelease :: ctx -> Maybe NominalDiffTime -> [job] -> n (Set JobId)
- effectSpawn :: ctx -> job -> kids -> n ()
- effectReport :: ctx -> Report job -> n ()
- effectLog :: LogLevel -> [job] -> Text -> n ()
}
data Report job
- = Claimed job UTCTime
- | Succeeded job UTCTime UTCTime
- | Failed job Text UTCTime UTCTime Outcome
- | Cancelled job Text
- | Unavailable job Text
type Failure = (Text, FailureKind)
data FailureKind
data Outcome
- = Retrying NominalDiffTime
- | DeadLettered
- | TreeCancelled [JobId]
cancelsTree :: FailureKind -> Bool
data Mode (n :: Type -> Type) ctx job kids stored
- = SingleMode (job -> n ())
- | BatchedMode (NonEmpty job -> Callbacks n ctx job kids stored -> n ())
data Callbacks (n :: Type -> Type) ctx job kids stored = Callbacks {
- callbackAck :: ctx -> job -> Maybe stored -> n ()
- callbackAckAll :: ctx -> [(job, Maybe stored)] -> n ()
- callbackFail :: ctx -> Failure -> job -> n ()
- callbackNack :: ctx -> Maybe NominalDiffTime -> job -> n ()
- callbackSpawn :: ctx -> job -> kids -> n ()
}
data Handoff (n :: Type -> Type) job
newHandoff :: (MonadSTM n, MonadTime n) => HeartbeatGuard n job -> n (Handoff n job)
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 ()

The pool's side

data Effects (n :: Type -> Type) ctx job kids storedSource #

Batch statements and hooks. Callbacks use the handler's context and join its active transaction. Each statement effect is one transaction or one hook.

Constructors

Effects

Fields

effectAmbient :: ctx
The context the batch runs in, outside the handler.
effectSpan :: ((n () -> n ()) -> n ()) -> n ()
Runs the batch under its consumer span, handing it the capture that runs the heartbeat hooks under that span.
effectAck :: ctx -> job -> Maybe stored -> n ()
Ack one job and store its result. Throws JobGoneException for a job held elsewhere.
effectAckAll :: ctx -> [(job, Maybe stored)] -> n ([job], [job])
Ack in bulk. The jobs acked, and the jobs found held elsewhere.
effectFail :: ctx -> Failure -> job -> n (Either Text Outcome)
Write one job's failure. Left why no row was written.
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.
effectDeleteCancelled :: ctx -> [job] -> n (Set JobId)
Delete the jobs a force-cancel flagged. The ids deleted.
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.
effectSpawn :: ctx -> job -> kids -> n ()
Attach children to one job and settle it suspended, in one transaction.
effectReport :: ctx -> Report job -> n ()
Fire the hook for one report.
effectLog :: LogLevel -> [job] -> Text -> n ()
The pool log, with the jobs in context.

data Report jobSource #

A hook the lifecycle fires.

Constructors

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

type Failure = (Text, FailureKind)Source #

A handler failure: its message and disposition.

data FailureKindSource #

How a handler failure is written.

Constructors

RetryFailure	Retry with backoff, or move to the DLQ when attempts are exhausted.
RetryAfterFailure NominalDiffTime	Retry after this delay in seconds, or move to the DLQ when attempts are exhausted.
PermanentFailure	Move to the DLQ now.
TreeCancelFailure	Delete the whole job tree from its root.
BranchCancelFailure	Delete the job's parent, or the job when it has none, with all descendants.

Instances

Instances details

Eq FailureKind Source #
Instance details Defined in Arbiter.Worker.Batch Methods (==) :: FailureKind -> FailureKind -> Bool# (/=) :: FailureKind -> FailureKind -> Bool#
Show FailureKind Source #
Instance details Defined in Arbiter.Worker.Batch Methods showsPrec :: Int -> FailureKind -> ShowS# show :: FailureKind -> String# showList :: [FailureKind] -> ShowS#

data OutcomeSource #

Where a failure write left the job. A tree cancel carries the ids it deleted.

Constructors

Retrying NominalDiffTime	Scheduled for retry after this backoff delay.
DeadLettered	Moved to the DLQ.
TreeCancelled [JobId]	Deleted by a tree or branch cancel.

Instances

Instances details

Eq Outcome Source #
Instance details Defined in Arbiter.Worker.Batch Methods (==) :: Outcome -> Outcome -> Bool# (/=) :: Outcome -> Outcome -> Bool#
Show Outcome Source #
Instance details Defined in Arbiter.Worker.Batch Methods showsPrec :: Int -> Outcome -> ShowS# show :: Outcome -> String# showList :: [Outcome] -> ShowS#

cancelsTree :: FailureKind -> BoolSource #

Whether a failure deletes a job tree or updates the job claim.

data Mode (n :: Type -> Type) ctx job kids storedSource #

How the pool runs a batch's handler.

Constructors

SingleMode (job -> n ())	One transaction that runs the handler, acks the job and stores its result.
BatchedMode (NonEmpty job -> Callbacks n ctx job kids stored -> n ())	The handler settles each job through the callbacks, in its context at the call.

data Callbacks (n :: Type -> Type) ctx job kids storedSource #

The settle operations a batch handler drives its jobs through.

Constructors

Callbacks

Fields

callbackAck :: ctx -> job -> Maybe stored -> n ()
Ack one job and store its result.
callbackAckAll :: ctx -> [(job, Maybe stored)] -> n ()
Ack in bulk, storing each result.
callbackFail :: ctx -> Failure -> job -> n ()
Write one job's failure.
callbackNack :: ctx -> Maybe NominalDiffTime -> job -> n ()
Release one job for reprocessing, after the delay when there is one.
callbackSpawn :: ctx -> job -> kids -> n ()
Attach children to one job and suspend it.

Running a batch

data Handoff (n :: Type -> Type) jobSource #

Finalization state shared by the handler and the force-cancel finalizer, with the batch's job key and start time.

newHandoff :: (MonadSTM n, MonadTime n) => HeartbeatGuard n job -> n (Handoff n job)Source #

A handoff keyed the way the guard keys its jobs, starting now.

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 ()Source #

Run the batch on the calling thread: the claim hooks, the handler under the guard, the outcome report, and the force-cancel finalizer.