Struct CurrentBatchedJob 

pub struct CurrentBatchedJob<C> { /* private fields */ }

Context handed to a BatchedJobRunner for one batch.

Implementations

impl<C> CurrentBatchedJob<C>

pub fn job_type(&self) -> &JobType

pub fn items(&self) -> &[BatchedJobItem<C>]

The jobs in this batch, ordered by queue_id (job id when unqueued) so concurrent batches take domain locks in a consistent order.

Never contains two jobs with the same queue_id.

pub fn items_mut(&mut self) -> &mut [BatchedJobItem<C>]

Mutable access, for BatchedJobItem::update_execution_state_in_op.

pub fn into_items(self) -> Vec<BatchedJobItem<C>>

Consume the context and take ownership of its items.

pub fn len(&self) -> usize

Number of jobs in this batch.

pub fn is_empty(&self) -> bool

Whether the batch is empty. Never true as dispatched.

pub fn ids(&self) -> Vec<JobId>

Every job id in this batch, in item order.

pub fn pool(&self) -> &PgPool

pub fn clock(&self) -> &ClockHandle

The clock configured on the job service.

pub async fn begin_op(&self) -> Result<DbOp<'static>, JobError>

Begin a new database operation using the job service’s clock.

pub async fn run_isolated<E>( &self, op: &mut impl SavepointOperation, f: impl AsyncFn(&mut SavepointOp<'_>, &BatchedJobItem<C>) -> Result<BatchItemOutcome, E> + Clone + Sync, ) -> Result<BatchOutcomes, Error>

where E: Display,

Run f once per item, each inside its own SAVEPOINT on op, and collect a BatchOutcomes — the batch-loop counterpart of es_entity::DbOp::with_savepoint.

A failing item is rolled back to its own savepoint without poisoning op: earlier and later items still commit in the same transaction. f’s Ok(outcome) is recorded as-is — usually BatchItemOutcome::Complete, though a successfully-released savepoint can still resolve to RescheduleIn/RescheduleAt if the item wants to run again later. f’s Err is translated to BatchItemOutcome::Fail (via E’s Display), so the item is retried per the type’s RetrySettings like any other failure.

Typical use, feeding the result into JobBatchCompletion::WithOutcomesWithOp:

let mut op = current_batch.begin_op().await?;
let outcomes = current_batch
    .run_isolated(&mut op, |op, item| async move {
        self.execute_in_op(op, item.config()).await?;
        Ok(BatchItemOutcome::Complete)
    })
    .await?;
Ok(JobBatchCompletion::WithOutcomesWithOp(op, outcomes))

Errors

The returned Err is the outer layer of with_savepoint: the savepoint machinery itself failed (e.g. a dead connection), not any one item’s domain logic. That failure isn’t attributable to a single item — propagate it with ? and let the whole batch retry, exactly like a whole-batch Err returned from run_batch today.

Use only *_in_op methods against the savepoint inside f. The pool-backed BatchedJobItem::update_execution_state / BatchedJobItem::set_result commit in their own transaction outside op entirely, so they do not unwind if this item’s savepoint rolls back.

A thin proxy over es_entity::BatchIsolation::run_isolated, which owns the actual per-item savepoint loop; this method just adapts it to BatchOutcomes so existing callers see no change in shape.

pub async fn run_bisected<E>( &self, op: &mut impl SavepointOperation, f: impl AsyncFn(&mut SavepointOp<'_>, &[BatchedJobItem<C>]) -> Result<(), E> + Clone + Sync, ) -> Result<BatchOutcomes, Error>

where E: Error + 'static,

Run the whole batch as one unit inside a single SAVEPOINT; on failure, auto-bisect: roll back and probe smaller contiguous sub-ranges — largest pending range first — until each culprit is isolated (or BisectBudget runs out). Clean items resolve to BatchItemOutcome::Complete in this very dispatch; each culprit to BatchItemOutcome::Fail (via E’s Display, from its own singleton probe — so the message is attributable).

The counterpart of run_isolated for true batch implementations: a batch-load/mutate/batch-persist runner has no per-item loop for run_isolated to wrap, so its only failure shape was previously “every item fails, all N retry solo”. f here takes a slice of items — probe them with one statement, the same shape a true batch already uses.

Equivalent to run_bisected_with with BisectBudget::default() (Auto).

Cost

Happy path: one extra SAVEPOINT/RELEASE pair for the whole batch. With k culprits in N items: roughly 2k·log₂(N/k) probes. If a type’s dominant failure mode is “everything fails together” (a shared upstream precondition), returning Err from run_batch directly is still cheaper — this helper targets k ≪ N.

Contract on f

• Must tolerate re-execution over arbitrary contiguous sub-slices, in non-positional order. Each probe’s DB writes unwind with its savepoint on failure, but host-side state (counters, caches) does not — derive everything f needs from (sp, slice). In the worst case f runs 2N-1 times; any single item is re-probed at most ⌈log₂ N⌉+1 times.
• No external calls (HTTP, mail, RPC) — already the rule for shared batch transactions; bisection would repeat them per probe.
• Only *_in_op methods against sp. The pool-backed BatchedJobItem::update_execution_state / BatchedJobItem::set_result commit outside this transaction and will NOT unwind with a rolled-back probe.
• Set semantics, not sequence semantics. A bisected batch gives up the positional “equivalent to running in order” guarantee the module documents for run_isolated/plain loops: probes run largest-range-first, so two items with a real dependency between them (a shared unique key, a read-your-write coupling) resolve deterministically by probe order, not item order. Only use this for batches whose items are safe to probe in any grouping — i.e. the module’s existing entity-disjointness guidance (queue_id is the serialization unit).
• The search does not run against a frozen snapshot. A failed probe’s rollback releases the locks it took, so rows it touched are free for other transactions to change before the next probe reaches them. A probe’s result is a fact about the moment it ran, not a standing one.
• Locks a successful probe takes are held until the shared transaction commits, so bisecting extends how long they are held — by the probes still to run — without widening the set. The set is the rows that were going to be committed anyway.

Errors

An outer Err means the batch could not be dispositioned here at all: either the savepoint machinery itself failed (a dead connection), or probes kept losing lock conflicts. Propagate it with ? and let the whole batch retry, exactly like run_isolated.

A 40P01 deadlock or 40001 serialization failure never drives the search. It is not attributable to any item, so splitting on one cannot isolate anything — and splitting can provoke it, since there are lock sets that succeed taken at once and deadlock taken in halves. Such a probe is re-run against the same range instead: its rollback released its locks, and the partner that won the cycle has moved on, so a retry usually succeeds and settles the batch in this dispatch. Only after they keep recurring does the bisect give up — which bounds the total deadlock_timeout one batch can pay, since Postgres makes every conflicting probe wait one out before reporting it.

let mut op = current_batch.begin_op().await?;
let outcomes = current_batch
    .run_bisected(&mut op, |sp, slice| async move {
        self.execute_many_in_op(sp, slice).await
    })
    .await?;
Ok(JobBatchCompletion::WithOutcomesWithOp(op, outcomes))

pub async fn run_bisected_with<E>( &self, op: &mut impl SavepointOperation, budget: BisectBudget, f: impl AsyncFn(&mut SavepointOp<'_>, &[BatchedJobItem<C>]) -> Result<(), E> + Clone + Sync, ) -> Result<BatchOutcomes, Error>

where E: Error + 'static,

run_bisected with an explicit BisectBudget instead of the default Auto. See run_bisected’s doc comment for the full contract; this is the same helper, just with the probe budget spelled out at the call site.

A thin proxy over es_entity::BatchIsolation::run_bisected, which owns the actual search (largest-pending-range-first, with the transient-conflict re-probe and refund); this method just adapts its es_entity::BisectOutcomes to BatchOutcomes so existing callers see no change in shape or search behavior.

pub fn outcomes_for_each( &self, f: impl FnMut(&BatchedJobItem<C>) -> BatchItemOutcome, ) -> BatchOutcomes

Build a BatchOutcomes by deciding an outcome for every item.

Applying the closure to each item makes it impossible to leave a job undispositioned, which the dispatcher would otherwise reject.

pub async fn shutdown_requested(&mut self) -> bool

Wait for a shutdown signal. Returns true if shutdown was requested.

pub fn is_shutdown_requested(&mut self) -> bool

Non-blocking check for a shutdown request.