Struct KeyedJobSpawner 

pub struct KeyedJobSpawner<Config> { /* private fields */ }

A handle for spawning keyed jobs of a specific type.

Returned by crate::Jobs::add_keyed_initializer.

Implementations

impl<Config> KeyedJobSpawner<Config>

where Config: Serialize + Send + Sync,

pub fn job_type(&self) -> &JobType

Returns the job type this spawner creates.

pub async fn spawn( &self, key: impl Into<String> + Send + Debug, config: Config, ) -> Result<JobHandle, JobError>

Create a keyed job, or resolve to the LIVE one if key already holds one.

While a job is LIVE (pending/running) under key, spawning again is a no-op: returns a JobHandle for the persisted job, so a double-spawning caller always observes the job that actually runs. Once that job reaches a terminal state the key becomes respawnable — the next call creates a NEW job (new internally-generated id, new config). The job’s id is generated internally — a keyed job is identified by its (job_type, key), not a caller-chosen id — so read the id back from the returned handle every time.

Keys are opaque to the crate — a sharded consumer can spawn one singleton per shard and later enumerate them all via crate::Jobs::keyed_handles, pairing each handle with JobHandle::load’s execution state to answer “caught up?” per shard. Does not consume the spawner: many keys of one type are expected.

Errors

Spawning against a held key is not an error — it resolves to the holder. This has no failure mode of its own beyond the underlying database errors.

pub async fn spawn_in_op( &self, op: &mut (impl AtomicOperation + ?Sized), key: impl Into<String> + Send + Debug, config: Config, ) -> Result<KeyedSpawn, JobError>

Self::spawn as part of an existing atomic operation, so a keyed job is created in the same transaction as whatever prompted it.

Same semantics as Self::spawn — create, or resolve to the LIVE holder — but returns KeyedSpawn rather than a bare handle, since in-op callers usually need to know whether they are the ones who created it before writing their own side of the transaction. Use KeyedSpawn::into_handle when they don’t.

Two calls on the SAME op for the same key are safe and resolve the second to the first’s job: the execution row is inserted before this returns, and a transaction sees its own uncommitted writes, so the second call’s live-check finds it. See Self::spawn_all_in_op.

pub async fn spawn_all( &self, specs: Vec<KeyedJobSpec<Config>>, ) -> Result<Vec<KeyedSpawn>, JobError>

Create or resolve many keys of this type in a single atomic operation.

Outcomes are returned in the order of specs, one per spec. Every key yields a KeyedSpawn — none are silently dropped — so this can be zipped straight back against the inputs.

pub async fn spawn_all_in_op( &self, op: &mut (impl AtomicOperation + ?Sized), specs: Vec<KeyedJobSpec<Config>>, ) -> Result<Vec<KeyedSpawn>, JobError>

Self::spawn_all as part of an existing atomic operation. The core every other spawn* method on this spawner delegates to.

How a key is claimed

JobRepo::lock_and_check_live_keys_in_op takes a transaction-scoped advisory lock per key and THEN reports each key’s live holder. Every writer of a keyed key goes through it, so once the check reports a key free, no other transaction can claim it before this one ends — the subsequent insert cannot conflict, and needs no ON CONFLICT clause. A unique violation on idx_job_executions_job_type_unique_key from here would mean a writer bypassed the lock, which is a bug worth failing loudly on rather than absorbing.

Resolving liveness BEFORE creating any jobs row is what keeps a resolved-to-holder spec from leaving an orphan jobs row behind (see JobRepo::lock_and_check_live_keys_in_op).

Why the insert is inline rather than deferred to ExecutionInsertHook

Because it makes duplicate keys within one op self-checking. The hook batches inserts to commit time, so a sibling call’s row would not exist yet when the next call runs its live-check; inserting here means the live-check — which reads inside this transaction, and so sees this transaction’s own uncommitted writes — is the single mechanism resolving same-op, same-transaction and cross-transaction collisions alike. The batching that would buy is small in exchange: keyed rows always have queue_id = NULL, so the hook’s queue parking/promotion machinery is inert for them, and bulk callers already get one statement per call from here.

seen covers the remaining case the live-check cannot: two specs sharing a key WITHIN this call, neither inserted yet.

Waking a live holder (KeyedJobSpec::force_reschedule)

A spec that resolves to a holder normally has no effect whatsoever. With the flag set it additionally runs pull_forward_in_op — execute_at = LEAST(execute_at, target) for the holder’s row, where the target is the spec’s own schedule_at or now — turning the respawn into “run no later than the time I am asking for” for a job scheduled beyond it.

Two things make that safe:

• It never shortens a backoff. The row must carry attempt_index <= 1. RetryPolicy retries carry the NEXT attempt index (>1, finalizer.rs’s retry write), while every deliberate reschedule resets it to 1 — so an exponential backoff is invisible to this path. Without that guard a keyed job spawned on every upstream event (lana’s price-shock sweeps are the live example) would have its backoff erased by the next event and hot-loop at the event rate precisely while it is failing.
• It only ever moves execute_at EARLIER, so it is monotone and idempotent: repeated respawns of a due row are no-ops, and no respawn can ever delay a job.

Only keys that were ALREADY live when the call started reach that statement. A key repeated within one call — first spec creates it far ahead, second spec asks for a wake — is resolved in memory instead: on a row this call is itself inserting every guard holds by construction, so the wake is just a lower execute_at on the insert. Same row, same reported pulled_forward, one statement fewer.

Rows actually moved are then reported through the SAME two signals a creation uses (the ExecutionReady notify and the local poller’s claim demand). Firing neither would leave the wake to be discovered by the next ordinary poll tick, which is exactly the latency this mechanism exists to remove.

Trait Implementations

impl<Config: Clone> Clone for KeyedJobSpawner<Config>

fn clone(&self) -> KeyedJobSpawner<Config>

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.