Struct MemStore 

pub struct MemStore {
    pub crash_limit: u32,
    pub retry_base_ms: i64,
    pub retry_cap_ms: i64,
    /* private fields */
}

Fields

crash_limit: u32

crash quarantine quarantine threshold.

retry_base_ms: i64

retry_cap_ms: i64

Implementations

impl MemStore

pub fn new() -> Self

pub fn freeze_clock_at(&self, ms: i64)

Freeze the STORE clock (boundary validation: store-supplied time, even here) at ms.

pub fn advance_clock(&self, by_ms: i64)

Step a frozen clock forward — deterministic backoff/retention tests, no sleeps. Freezes at system-now first if the clock was live.

pub fn unfreeze_clock(&self)

pub fn job_state(&self, id: &str) -> Option<(Envelope, String)>

(envelope snapshot, state) — None if the job does not exist (deleted counts).

pub fn errors(&self, id: &str) -> Vec<String>

The per-attempt error history recorded for a job.

pub fn counts(&self, queue: Option<&str>) -> HashMap<String, usize>

state -> count for one queue (None = all queues).

pub fn set_queue_paused(&self, queue: &str, paused: bool)

pub fn set_rate_limit(&self, name: &str, limit: i64, window_ms: i64, burst: i64)

Configure a fleet token bucket. Unconfigured classes are unlimited here.

impl MemStore

pub async fn enqueue_without_uniqueness( &self, batch: &[Envelope], ) -> Result<(), StoreError>

Test-only, call-scoped uniqueness bypass. IDs remain strict and no mutable flag can leak into another parallel test.

Trait Implementations

impl Default for MemStore

fn default() -> Self

Returns the “default value” for a type.

impl EnqueuedJobs for MemStore

fn all_enqueued(&self) -> Vec<Envelope>

Every job the store currently holds, id-ordered. A job DELETED (retention policy ephemeral retention-0, retention and eviction contract eviction, revoke) is gone from here, which is the honest answer: “was enqueued” is only observable while the row exists.

impl OutputInspect for MemStore

fn get_job_output<'life0, 'life1, 'async_trait>( &'life0 self, id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Option<JobOutput>, StoreError>> + Send + 'async_trait>>

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

Explicit output access. A previous attempt’s latest output may remain visible until the current holder replaces it; JobOutput::fence identifies its author.

impl OutputStore for MemStore

fn write_job_output<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, lease: &'life1 LeaseRef, output: &'life2 JobResult, ) -> Pin<Box<dyn Future<Output = Result<JobOutput, StoreError>> + Send + 'async_trait>>

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

impl ProgressInspect for MemStore

fn get_job_progress<'life0, 'life1, 'async_trait>( &'life0 self, id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Option<JobProgress>, StoreError>> + Send + 'async_trait>>

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

A previous attempt’s last report may remain until the current holder replaces it; JobProgress::fence makes that provenance explicit.

impl ProgressStore for MemStore

fn write_job_progress<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, lease: &'life1 LeaseRef, update: &'life2 ProgressUpdate, ) -> Pin<Box<dyn Future<Output = Result<JobProgress, StoreError>> + Send + 'async_trait>>

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

impl ResultInspect for MemStore

fn get_job_result<'life0, 'life1, 'async_trait>( &'life0 self, id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Option<JobResult>, StoreError>> + Send + 'async_trait>>

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

Explicit result access. Implementations return None for a missing job or a job with no completed result; payload/list reads never include these bytes implicitly.

impl ResultStore for MemStore

fn ack_success_with_result<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, lease: &'life1 LeaseRef, logs: &'life2 [String], actual_weight: Option<u32>, result: &'life3 JobResult, ) -> 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,

impl Store for MemStore

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 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 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 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

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 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, Self: '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 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait, Self: '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_transactional(&self) -> Option<&(dyn Transactional + 'static)>

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 + 'static)>

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

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

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.