Trait Store 

pub trait Store:
    Send
    + Sync
    + 'static {
    // Required methods
    fn admit<'life0, 'async_trait>(
        &'life0 self,
        req: AdmitRequest,
    ) -> Pin<Box<dyn Future<Output = Result<Vec<AdmissionUnit>, StoreError>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;
    fn ack_attempt_with_actual_weight<'life0, 'life1, 'life2, 'life3, 'async_trait>(
        &'life0 self,
        lease: &'life1 LeaseRef,
        outcome: Outcome,
        err: Option<&'life2 str>,
        delay_ms: Option<i64>,
        logs: &'life3 [String],
        actual_weight: Option<u32>,
    ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait,
             'life2: 'async_trait,
             'life3: 'async_trait;
    fn renew<'life0, 'life1, 'async_trait>(
        &'life0 self,
        leases: &'life1 [LeaseRef],
        lease: Duration,
    ) -> Pin<Box<dyn Future<Output = Result<Vec<String>, StoreError>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn enqueue<'life0, 'life1, 'async_trait>(
        &'life0 self,
        batch: &'life1 [Envelope],
    ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn checkpoint<'life0, 'life1, 'life2, 'async_trait>(
        &'life0 self,
        lease: &'life1 LeaseRef,
        cp: &'life2 Checkpoint,
    ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait,
             'life2: 'async_trait;
    fn reclaim_expired<'life0, 'async_trait>(
        &'life0 self,
        limit: i64,
    ) -> Pin<Box<dyn Future<Output = Result<Vec<Reclaimed>, StoreError>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;
    fn promote_due<'life0, 'async_trait>(
        &'life0 self,
        limit: i64,
    ) -> Pin<Box<dyn Future<Output = Result<u64, StoreError>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;
    fn evict_retained<'life0, 'async_trait>(
        &'life0 self,
        limit: i64,
    ) -> Pin<Box<dyn Future<Output = Result<u64, StoreError>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;
    fn claim_duty<'life0, 'life1, 'life2, 'async_trait>(
        &'life0 self,
        name: &'life1 str,
        holder: &'life2 str,
        lease: Duration,
    ) -> Pin<Box<dyn Future<Output = Result<bool, StoreError>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait,
             'life2: 'async_trait;
    fn release_duty<'life0, 'life1, 'life2, 'async_trait>(
        &'life0 self,
        name: &'life1 str,
        holder: &'life2 str,
    ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait,
             'life2: 'async_trait;
    fn caps(&self) -> Caps;

    // Provided methods
    fn ack<'life0, 'life1, 'life2, 'async_trait>(
        &'life0 self,
        lease: &'life1 LeaseRef,
        outcome: Outcome,
        err: Option<&'life2 str>,
        delay_ms: Option<i64>,
    ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait,
             'life2: 'async_trait { ... }
    fn ack_attempt<'life0, 'life1, 'life2, 'life3, 'async_trait>(
        &'life0 self,
        lease: &'life1 LeaseRef,
        outcome: Outcome,
        err: Option<&'life2 str>,
        delay_ms: Option<i64>,
        logs: &'life3 [String],
    ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait,
             'life2: 'async_trait,
             'life3: 'async_trait { ... }
    fn as_result_store(&self) -> Option<&dyn ResultStore> { ... }
    fn as_output_store(&self) -> Option<&dyn OutputStore> { ... }
    fn as_progress_store(&self) -> Option<&dyn ProgressStore> { ... }
    fn as_transactional(&self) -> Option<&dyn Transactional> { ... }
    fn as_inspect(&self) -> Option<&dyn Inspect> { ... }
    fn as_notifying(&self) -> Option<&dyn Notifying> { ... }
}

The whole port. Coarse on purpose — the admission decision must stay atomic inside the store, so a fine-grained port would force the gate back into the worker.

async_trait rather than RPITIT so Box<dyn Store> works: selecting a backend from a config string needs a trait object, and impl Future in trait position is not dyn-compatible. Store calls are I/O-bound, so the boxed future is noise.

Required Methods

fn admit<'life0, 'async_trait>( &'life0 self, req: AdmitRequest, ) -> Pin<Box<dyn Future<Output = Result<Vec<AdmissionUnit>, StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

The whole admission decision: policy + claim + lease, atomically, store-side.

fn ack_attempt_with_actual_weight<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, lease: &'life1 LeaseRef, outcome: Outcome, err: Option<&'life2 str>, delay_ms: Option<i64>, logs: &'life3 [String], actual_weight: Option<u32>, ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait,

Store::ack_attempt plus surveyed policy behavior cost reconciliation. Admission charges the envelope’s estimated weight; Some(actual) corrects that charge under the same fence and in the same atomic write as the state transition. Some(0) is a real full refund; None means the estimate was exact. The extra method is kept coarse on purpose: a separate reconcile call could commit after a rejected ack.

fn renew<'life0, 'life1, 'async_trait>( &'life0 self, leases: &'life1 [LeaseRef], lease: Duration, ) -> Pin<Box<dyn Future<Output = Result<Vec<String>, StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Extend leases; return the job ids of the leases that were LOST. A worker that lost a lease must be able to stop — silently succeeding here is how asynq stranded jobs in ACTIVE since 2022.

fn enqueue<'life0, 'life1, 'async_trait>( &'life0 self, batch: &'life1 [Envelope], ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

fn checkpoint<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, lease: &'life1 LeaseRef, cp: &'life2 Checkpoint, ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

step replay persist step progress, fence-verified: the write succeeds only while the caller still holds the lease, so it doubles as the step boundary’s lease check. LeaseRejected here means STOP — do not run the next step’s side effects. Durable BEFORE the step runs, never after the worker returns (River’s mistake).

fn reclaim_expired<'life0, 'async_trait>( &'life0 self, limit: i64, ) -> Pin<Box<dyn Future<Output = Result<Vec<Reclaimed>, StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

lease fencing/crash quarantine the lease reclaimer’s sweep. An expired lease is Outcome::LeaseLost, NEVER Retry: it increments crash_attempt and leaves attempt alone. At the crash limit the job parks in quarantined and its fingerprint is registered. Safe under contention; run it via a duty lease to avoid redundant sweeps.

fn promote_due<'life0, 'async_trait>( &'life0 self, limit: i64, ) -> Pin<Box<dyn Future<Output = Result<u64, StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

The schedule_due/backoff_due sweep: due scheduled and retryable jobs become available. Returns how many were promoted.

fn evict_retained<'life0, 'async_trait>( &'life0 self, limit: i64, ) -> Pin<Box<dyn Future<Output = Result<u64, StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

retention and eviction contract the retention sweep: TERMINAL jobs whose finalized_at_ms + retention_ms has lapsed are deleted (the transition table’s completed -> deleted by retention; retention_ms = 0 already deleted at ack time). quarantined is exempt — it parks VISIBLY until an operator acts, never silently expires. Bounded per call; run under the retention duty lease.

fn claim_duty<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, name: &'life1 str, holder: &'life2 str, lease: Duration, ) -> Pin<Box<dyn Future<Output = Result<bool, StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

singleton duties claim (or renew) a singleton duty. Same compare-and-set as claiming a job, on store time — a skewed node cannot steal a duty early. false = someone else holds it; skip this tick, never block on it.

fn release_duty<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, name: &'life1 str, holder: &'life2 str, ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

singleton duties step down by expiring the duty immediately, so takeover is fast. A no-match (not the holder) is fine — release is best-effort on shutdown.

fn caps(&self) -> Caps

Provided Methods

fn ack<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, lease: &'life1 LeaseRef, outcome: Outcome, err: Option<&'life2 str>, delay_ms: Option<i64>, ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Apply the transition table for outcome, write the error history, honour the fence. delay_ms: required for Snooze (must be > 0); for Retry it overrides the store’s default backoff (the retry-policy port computes it caller-side); ignored otherwise. LeaseLost is never acked — it is the reclaimer’s transition. Convenience over Store::ack_attempt with no logs.

fn ack_attempt<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, lease: &'life1 LeaseRef, outcome: Outcome, err: Option<&'life2 str>, delay_ms: Option<i64>, logs: &'life3 [String], ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait,

Store::ack plus attempt-log contract per-attempt execution logs (River’s riverlog): captured handler log lines land INSIDE the attempt’s error-history entry, so the console shows why an attempt failed, not just that it did. Recorded for success / retry / skip / undecodable (a non-empty logs on success writes a success entry — the only time one exists); dropped for snooze / rate_limited / revoke, which by design record no attempt entry.

fn as_result_store(&self) -> Option<&dyn ResultStore>

Optional capability surface for a success transition that records result bytes under the same fence. A backend implementing ResultStore returns it here.

fn as_output_store(&self) -> Option<&dyn OutputStore>

Optional capability for fence-verified mid-run output writes. Unlike a final result this does not transition the job; it succeeds only for the current running lease and returns store-stamped attempt/time metadata.

fn as_progress_store(&self) -> Option<&dyn ProgressStore>

Optional capability for operator-facing progress writes. Like mid-run output, this is a fenced write that does not transition the job.

fn as_transactional(&self) -> Option<&dyn Transactional>

runtime capability boundary Runtime capability upcast. None means genuinely unsupported — never a silent no-op, and never a config knob that does nothing.

fn as_inspect(&self) -> Option<&dyn Inspect>

control plane the inspection/control surface. Same rule as as_transactional.

fn as_notifying(&self) -> Option<&dyn Notifying>

push wakeups push wakeup. MySQL never has this (poll only); PgBouncer in transaction pooling breaks it, which is why poll-only remains a first-class mode.

Dyn Compatibility

This trait is dyn compatible.

In older versions of Rust, dyn compatibility was called "object safety".

Implementors