Skip to main content

QueueBackend

boson_core::backend

Trait QueueBackend 

Source 
pub trait QueueBackend:
    Send
    + Sync
    + Debug {

Show 27 methods    // Required methods
    fn upsert_job<'life0, 'life1, 'async_trait>(
        &'life0 self,
        job: &'life1 Job,
    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn enqueue_with_policies<'life0, 'life1, 'async_trait>(
        &'life0 self,
        job: Job,
        task_config: &'life1 TaskConfig,
    ) -> Pin<Box<dyn Future<Output = Result<(String, JobEnqueueDisposition)>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn get_job<'life0, 'life1, 'async_trait>(
        &'life0 self,
        job_id: &'life1 str,
    ) -> Pin<Box<dyn Future<Output = Result<Option<Job>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn list_jobs<'life0, 'async_trait>(
        &'life0 self,
        status_filter: Option<JobStatus>,
        offset: usize,
        limit: usize,
    ) -> Pin<Box<dyn Future<Output = Result<Vec<Job>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;
    fn cancel_job_if_active<'life0, 'life1, 'async_trait>(
        &'life0 self,
        job_id: &'life1 str,
    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn try_claim_job<'life0, 'life1, 'async_trait>(
        &'life0 self,
        job_id: &'life1 str,
    ) -> Pin<Box<dyn Future<Output = Result<Option<Job>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn revert_job_to_queued<'life0, 'life1, 'async_trait>(
        &'life0 self,
        job_id: &'life1 str,
    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn distinct_pools_queued<'life0, 'async_trait>(
        &'life0 self,
    ) -> Pin<Box<dyn Future<Output = Result<Vec<String>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;
    fn list_queued_for_pool_sorted<'life0, 'life1, 'async_trait>(
        &'life0 self,
        pool: &'life1 str,
        limit: usize,
    ) -> Pin<Box<dyn Future<Output = Result<Vec<Job>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn count_jobs<'life0, 'async_trait>(
        &'life0 self,
        status_filter: Option<JobStatus>,
    ) -> Pin<Box<dyn Future<Output = Result<u64>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;
    fn count_jobs_for_task<'life0, 'life1, 'async_trait>(
        &'life0 self,
        task_name: &'life1 str,
        status: Option<JobStatus>,
    ) -> Pin<Box<dyn Future<Output = Result<u64>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn count_active_jobs_for_task<'life0, 'life1, 'async_trait>(
        &'life0 self,
        task_name: &'life1 str,
    ) -> Pin<Box<dyn Future<Output = Result<u32>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn find_nonterminal_by_idempotency_key<'life0, 'life1, 'async_trait>(
        &'life0 self,
        key: &'life1 str,
    ) -> Pin<Box<dyn Future<Output = Result<Option<String>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn upsert_run<'life0, 'life1, 'async_trait>(
        &'life0 self,
        run: &'life1 Run,
    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn get_run<'life0, 'life1, 'async_trait>(
        &'life0 self,
        run_id: &'life1 str,
    ) -> Pin<Box<dyn Future<Output = Result<Option<Run>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn list_runs<'life0, 'life1, 'async_trait>(
        &'life0 self,
        job_id_filter: Option<&'life1 str>,
        offset: usize,
        limit: usize,
    ) -> Pin<Box<dyn Future<Output = Result<Vec<Run>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn finish_run<'life0, 'life1, 'async_trait>(
        &'life0 self,
        run_id: &'life1 str,
        status: RunStatus,
        duration_ms: Option<i64>,
        error_message: Option<String>,
    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn count_runs<'life0, 'life1, 'async_trait>(
        &'life0 self,
        job_id_filter: Option<&'life1 str>,
    ) -> Pin<Box<dyn Future<Output = Result<u64>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn count_runs_since<'life0, 'async_trait>(
        &'life0 self,
        since: DateTime<Utc>,
    ) -> Pin<Box<dyn Future<Output = Result<u64>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;
    fn task_run_stats<'life0, 'life1, 'async_trait>(
        &'life0 self,
        task_name: &'life1 str,
    ) -> Pin<Box<dyn Future<Output = Result<TaskRunStats>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn get_task_config<'life0, 'life1, 'async_trait>(
        &'life0 self,
        task_name: &'life1 str,
    ) -> Pin<Box<dyn Future<Output = Result<Option<TaskConfig>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn upsert_task_config<'life0, 'life1, 'async_trait>(
        &'life0 self,
        config: &'life1 TaskConfig,
    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn try_claim_run_lease<'life0, 'life1, 'life2, 'async_trait>(
        &'life0 self,
        job_id: &'life1 str,
        worker_id: &'life2 str,
        ttl_secs: i64,
    ) -> Pin<Box<dyn Future<Output = Result<Option<String>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait,
             'life2: 'async_trait;
    fn extend_lease<'life0, 'life1, 'async_trait>(
        &'life0 self,
        lease_id: &'life1 str,
        ttl_secs: i64,
    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn release_lease<'life0, 'life1, 'async_trait>(
        &'life0 self,
        lease_id: &'life1 str,
    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn expired_lease_job_pairs<'life0, 'async_trait>(
        &'life0 self,
    ) -> Pin<Box<dyn Future<Output = Result<Vec<(String, String)>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;

    // Provided method
    fn pop_claim_from_pool<'life0, 'life1, 'async_trait>(
        &'life0 self,
        _pool: &'life1 str,
    ) -> Pin<Box<dyn Future<Output = Result<Option<Job>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait { ... }

}
Expand description

Stable async trait for queue persistence (jobs, runs, config, leases).

Implement this trait in a separate adapter crate (for example myorg-boson-backend-redis) using only DTOs from boson-core. The runtime holds Arc<dyn QueueBackend> and calls these methods from the worker loop, enqueue path, and HTTP admin handlers.

§How to implement

  1. Create an adapter crate — depend on boson-core, async-trait, and your storage client.
  2. Map DTOs to storage — persist Job, Run, TaskConfig, and lease records using the types in crate::models; do not invent parallel schemas.
  3. Implement every method — group by jobs, runs, task config, and leases (see below).
  4. Honor the contract — especially idempotent enqueue, atomic claim, and lease exclusivity.
  5. Wire at boot — pass Arc::new(YourBackend::new(...)) to boson_runtime::BosonBuilder::queue_backend.
  6. Validate — copy integration tests from MemQueueBackend or SqlQueueBackend.

§Method groups

GroupKey methodsNotes
Jobsenqueue_with_policies, try_claim_job, list_queued_for_pool_sortedApply TaskConfig rate limits on enqueue; claim must be atomic
Runsupsert_run, finish_run, task_run_statsOne run row per execution attempt
Task configget_task_config, upsert_task_configOptional per-task overrides stored by the runtime
Leasestry_claim_run_lease, extend_lease, expired_lease_job_pairsRequired for multi-worker deploys; may no-op for single-node dev

§Reference implementations

§Example

Skeleton only — every trait method must be implemented (see reference adapters above):

ⓘ
use std::sync::Arc;

use async_trait::async_trait;
use boson_core::{Job, JobEnqueueDisposition, QueueBackend, Result, TaskConfig};
use boson_runtime::Boson;

#[derive(Debug)]
pub struct MyQueueBackend {
    // pool, client, or in-memory store
}

#[async_trait]
impl QueueBackend for MyQueueBackend {
    async fn enqueue_with_policies(
        &self,
        job: Job,
        task_config: &TaskConfig,
    ) -> Result<(String, JobEnqueueDisposition)> {
        // insert job; honor idempotency_key and rate limits from task_config
        todo!()
    }

    async fn try_claim_job(&self, job_id: &str) -> Result<Option<Job>> {
        // atomically queued -> running, or return None
        todo!()
    }

    // ... implement all remaining QueueBackend methods
}

// Integrator wiring (see boson crate Getting started — Mode 1 / Mode 2):
let boson = Boson::builder()
    .queue_backend(Arc::new(MyQueueBackend { /* ... */ }))
    .execution_context_factory(boson_core::JsonExecutionContextFactory)
    .auto_registry()
    .build()?;

§Contract

  • Implementations must be Send + Sync; hosts hold Arc<dyn QueueBackend>.
  • Idempotency: when job.idempotency_key is set, enqueue_with_policies returns ReusedIdempotent with the existing job id if a non-terminal job already exists for that key.
  • Claim: try_claim_job must atomically transition queued → running or return None.
  • Leases: try_claim_run_lease returns None when another worker holds an active lease.

Required Methods§

Source

fn upsert_job<'life0, 'life1, 'async_trait>( &'life0 self, job: &'life1 Job, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Persist or update a job row.

Source

fn enqueue_with_policies<'life0, 'life1, 'async_trait>( &'life0 self, job: Job, task_config: &'life1 TaskConfig, ) -> Pin<Box<dyn Future<Output = Result<(String, JobEnqueueDisposition)>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Insert job with idempotency semantics (see trait-level contract).

Source

fn get_job<'life0, 'life1, 'async_trait>( &'life0 self, job_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Option<Job>>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Load one job by id.

Source

fn list_jobs<'life0, 'async_trait>( &'life0 self, status_filter: Option<JobStatus>, offset: usize, limit: usize, ) -> Pin<Box<dyn Future<Output = Result<Vec<Job>>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait,

List jobs with optional status filter and pagination.

Source

fn cancel_job_if_active<'life0, 'life1, 'async_trait>( &'life0 self, job_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Cancel a job if it is still active (queued or running).

Source

fn try_claim_job<'life0, 'life1, 'async_trait>( &'life0 self, job_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Option<Job>>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Atomically claim a queued job for execution.

Source

fn revert_job_to_queued<'life0, 'life1, 'async_trait>( &'life0 self, job_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Revert a job to queued (retry path).

Source

fn distinct_pools_queued<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<Vec<String>>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait,

Distinct pool names with queued jobs.

Source

fn list_queued_for_pool_sorted<'life0, 'life1, 'async_trait>( &'life0 self, pool: &'life1 str, limit: usize, ) -> Pin<Box<dyn Future<Output = Result<Vec<Job>>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Queued jobs for one pool, sorted by priority then created time.

Source

fn count_jobs<'life0, 'async_trait>( &'life0 self, status_filter: Option<JobStatus>, ) -> Pin<Box<dyn Future<Output = Result<u64>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait,

Count jobs, optionally filtered by status.

Source

fn count_jobs_for_task<'life0, 'life1, 'async_trait>( &'life0 self, task_name: &'life1 str, status: Option<JobStatus>, ) -> Pin<Box<dyn Future<Output = Result<u64>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Count jobs for one task, optionally filtered by status.

Source

fn count_active_jobs_for_task<'life0, 'life1, 'async_trait>( &'life0 self, task_name: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<u32>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Count active (queued + running) jobs for rate-limit checks.

Source

fn find_nonterminal_by_idempotency_key<'life0, 'life1, 'async_trait>( &'life0 self, key: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Option<String>>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Find non-terminal job by idempotency key.

Used by Self::enqueue_with_policies to return JobEnqueueDisposition::ReusedIdempotent when a job with the same key is still queued or running. Returns None when no matching non-terminal job exists.

Source

fn upsert_run<'life0, 'life1, 'async_trait>( &'life0 self, run: &'life1 Run, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Persist or update a run row.

Source

fn get_run<'life0, 'life1, 'async_trait>( &'life0 self, run_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Option<Run>>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Load one run by id.

Source

fn list_runs<'life0, 'life1, 'async_trait>( &'life0 self, job_id_filter: Option<&'life1 str>, offset: usize, limit: usize, ) -> Pin<Box<dyn Future<Output = Result<Vec<Run>>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

List runs with optional job filter and pagination.

Source

fn finish_run<'life0, 'life1, 'async_trait>( &'life0 self, run_id: &'life1 str, status: RunStatus, duration_ms: Option<i64>, error_message: Option<String>, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Mark a run terminal with outcome fields.

Source

fn count_runs<'life0, 'life1, 'async_trait>( &'life0 self, job_id_filter: Option<&'life1 str>, ) -> Pin<Box<dyn Future<Output = Result<u64>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Count runs, optionally filtered by job id.

Source

fn count_runs_since<'life0, 'async_trait>( &'life0 self, since: DateTime<Utc>, ) -> Pin<Box<dyn Future<Output = Result<u64>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait,

Count runs with started_at >= since.

Source

fn task_run_stats<'life0, 'life1, 'async_trait>( &'life0 self, task_name: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<TaskRunStats>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Aggregate run totals for one task.

Source

fn get_task_config<'life0, 'life1, 'async_trait>( &'life0 self, task_name: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Option<TaskConfig>>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Load task config by name.

Source

fn upsert_task_config<'life0, 'life1, 'async_trait>( &'life0 self, config: &'life1 TaskConfig, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Persist task config.

Source

fn try_claim_run_lease<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, job_id: &'life1 str, worker_id: &'life2 str, ttl_secs: i64, ) -> Pin<Box<dyn Future<Output = Result<Option<String>>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Attempt to claim a run lease for job_id (stored as lease task id).

Returns Some(lease_id) when this worker holds the lease, or None when another worker holds an active lease for the same job.

Source

fn extend_lease<'life0, 'life1, 'async_trait>( &'life0 self, lease_id: &'life1 str, ttl_secs: i64, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Extend lease TTL for a held lease.

Source

fn release_lease<'life0, 'life1, 'async_trait>( &'life0 self, lease_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Release a held lease.

Source

fn expired_lease_job_pairs<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<Vec<(String, String)>>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait,

Expired leases as (lease_record_id, job_id).

Provided Methods§

Source

fn pop_claim_from_pool<'life0, 'life1, 'async_trait>( &'life0 self, _pool: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Option<Job>>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Atomically claim the highest-priority queued job from pool when supported.

Backends that support single-round-trip claim (Redis, NATS WorkQueue) override this to pop and claim in one operation. The default returns None so callers fall back to Self::list_queued_for_pool_sorted + Self::try_claim_job.

Dyn Compatibility§

This trait is dyn compatible.

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

Implementors§