boson_core/backend/queue_backend.rs
1//! [`QueueBackend`] — queue persistence trait for custom storage adapters.
2//!
3//! See **How to implement** on [`QueueBackend`] for a step-by-step guide and reference adapters.
4
5use std::fmt::Debug;
6use std::sync::Arc;
7
8use async_trait::async_trait;
9use chrono::{DateTime, Utc};
10
11use crate::error::Result;
12use crate::models::{Job, JobStatus, Run, RunStatus, TaskConfig, TaskRunStats};
13
14/// Default router key for single-backend hosts.
15pub const DEFAULT_BACKEND_NAME: &str = "default";
16
17/// Whether [`QueueBackend::enqueue_with_policies`] inserted or reused a job.
18#[derive(Debug, Clone, Copy, PartialEq, Eq)]
19pub enum JobEnqueueDisposition {
20 /// A new job row was inserted.
21 InsertedNew,
22 /// An existing non-terminal job matched the idempotency key.
23 ReusedIdempotent,
24}
25
26/// Stable async trait for queue persistence (jobs, runs, config, leases).
27///
28/// Implement this trait in a separate adapter crate (for example `myorg-boson-backend-redis`) using
29/// only DTOs from `boson-core`. The runtime holds `Arc<dyn QueueBackend>` and calls these methods
30/// from the worker loop, enqueue path, and HTTP admin handlers.
31///
32/// # How to implement
33///
34/// 1. **Create an adapter crate** — depend on `boson-core`, `async-trait`, and your storage client.
35/// 2. **Map DTOs to storage** — persist [`Job`], [`Run`], [`TaskConfig`], and lease records using
36/// the types in [`crate::models`]; do not invent parallel schemas.
37/// 3. **Implement every method** — group by jobs, runs, task config, and leases (see below).
38/// 4. **Honor the contract** — especially idempotent enqueue, atomic claim, and lease exclusivity.
39/// 5. **Wire at boot** — pass `Arc::new(YourBackend::new(...))` to
40/// [`boson_runtime::BosonBuilder::queue_backend`](https://docs.rs/boson-runtime/latest/boson_runtime/struct.BosonBuilder.html#method.queue_backend).
41/// 6. **Validate** — copy integration tests from [`MemQueueBackend`](https://docs.rs/boson-backend-mem/latest/boson_backend_mem/struct.MemQueueBackend.html)
42/// or [`SqlQueueBackend`](https://docs.rs/boson-backend-sql-common/latest/boson_backend_sql_common/struct.SqlQueueBackend.html).
43///
44/// # Method groups
45///
46/// | Group | Key methods | Notes |
47/// |-------|-------------|-------|
48/// | **Jobs** | [`enqueue_with_policies`](Self::enqueue_with_policies), [`try_claim_job`](Self::try_claim_job), [`list_queued_for_pool_sorted`](Self::list_queued_for_pool_sorted) | Apply [`TaskConfig`] rate limits on enqueue; claim must be atomic |
49/// | **Runs** | [`upsert_run`](Self::upsert_run), [`finish_run`](Self::finish_run), [`task_run_stats`](Self::task_run_stats) | One run row per execution attempt |
50/// | **Task config** | [`get_task_config`](Self::get_task_config), [`upsert_task_config`](Self::upsert_task_config) | Optional per-task overrides stored by the runtime |
51/// | **Leases** | [`try_claim_run_lease`](Self::try_claim_run_lease), [`extend_lease`](Self::extend_lease), [`expired_lease_job_pairs`](Self::expired_lease_job_pairs) | Required for multi-worker deploys; may no-op for single-node dev |
52///
53/// # Reference implementations
54///
55/// - [`MemQueueBackend`](https://docs.rs/boson-backend-mem/latest/boson_backend_mem/struct.MemQueueBackend.html) — in-memory, ideal starting point (~200 lines per module)
56/// - [`SqlQueueBackend`](https://docs.rs/boson-backend-sql-common/latest/boson_backend_sql_common/struct.SqlQueueBackend.html) — shared SQL for SQLite/PostgreSQL
57///
58/// # Example
59///
60/// Skeleton only — every trait method must be implemented (see reference adapters above):
61///
62/// ```ignore
63/// use std::sync::Arc;
64///
65/// use async_trait::async_trait;
66/// use boson_core::{Job, JobEnqueueDisposition, QueueBackend, Result, TaskConfig};
67/// use boson_runtime::Boson;
68///
69/// #[derive(Debug)]
70/// pub struct MyQueueBackend {
71/// // pool, client, or in-memory store
72/// }
73///
74/// #[async_trait]
75/// impl QueueBackend for MyQueueBackend {
76/// async fn enqueue_with_policies(
77/// &self,
78/// job: Job,
79/// task_config: &TaskConfig,
80/// ) -> Result<(String, JobEnqueueDisposition)> {
81/// // insert job; honor idempotency_key and rate limits from task_config
82/// todo!()
83/// }
84///
85/// async fn try_claim_job(&self, job_id: &str) -> Result<Option<Job>> {
86/// // atomically queued -> running, or return None
87/// todo!()
88/// }
89///
90/// // ... implement all remaining QueueBackend methods
91/// }
92///
93/// // Integrator wiring (Integrating the server):
94/// let boson = Boson::builder()
95/// .queue_backend(Arc::new(MyQueueBackend { /* ... */ }))
96/// .execution_context_factory(boson_core::JsonExecutionContextFactory)
97/// .auto_registry()
98/// .build()?;
99/// ```
100///
101/// # Contract
102///
103/// - Implementations must be `Send + Sync`; hosts hold `Arc<dyn QueueBackend>`.
104/// - Idempotency: when `job.idempotency_key` is set, `enqueue_with_policies` returns
105/// [`ReusedIdempotent`](JobEnqueueDisposition::ReusedIdempotent) with the existing job id
106/// if a non-terminal job already exists for that key.
107/// - Claim: `try_claim_job` must atomically transition `queued` → `running` or return `None`.
108/// - Leases: `try_claim_run_lease` returns `None` when another worker holds an active lease.
109#[async_trait]
110pub trait QueueBackend: Send + Sync + Debug {
111 // --- Jobs ---
112
113 /// Persist or update a job row.
114 async fn upsert_job(&self, job: &Job) -> Result<()>;
115
116 /// Insert job with idempotency semantics (see trait-level contract).
117 async fn enqueue_with_policies(
118 &self,
119 job: Job,
120 task_config: &TaskConfig,
121 ) -> Result<(String, JobEnqueueDisposition)>;
122
123 /// Load one job by id.
124 async fn get_job(&self, job_id: &str) -> Result<Option<Job>>;
125
126 /// List jobs with optional status filter and pagination.
127 async fn list_jobs(
128 &self,
129 status_filter: Option<JobStatus>,
130 offset: usize,
131 limit: usize,
132 ) -> Result<Vec<Job>>;
133
134 /// Cancel a job if it is still active (`queued` or `running`).
135 async fn cancel_job_if_active(&self, job_id: &str) -> Result<()>;
136
137 /// Atomically claim a queued job for execution.
138 async fn try_claim_job(&self, job_id: &str) -> Result<Option<Job>>;
139
140 /// Revert a job to `queued` (retry path).
141 async fn revert_job_to_queued(&self, job_id: &str) -> Result<()>;
142
143 /// Distinct pool names with queued jobs.
144 async fn distinct_pools_queued(&self) -> Result<Vec<String>>;
145
146 /// Queued jobs for one pool, sorted by priority then created time.
147 async fn list_queued_for_pool_sorted(&self, pool: &str, limit: usize) -> Result<Vec<Job>>;
148
149 /// Atomically claim the highest-priority queued job from `pool` when supported.
150 ///
151 /// Backends that support single-round-trip claim (Redis, NATS `WorkQueue`) override this to
152 /// pop and claim in one operation. The default returns `None` so callers fall back to
153 /// [`Self::list_queued_for_pool_sorted`] + [`Self::try_claim_job`].
154 async fn pop_claim_from_pool(&self, _pool: &str) -> Result<Option<Job>> {
155 Ok(None)
156 }
157
158 /// Count jobs, optionally filtered by status.
159 async fn count_jobs(&self, status_filter: Option<JobStatus>) -> Result<u64>;
160
161 /// Count jobs for one task, optionally filtered by status.
162 async fn count_jobs_for_task(
163 &self,
164 task_name: &str,
165 status: Option<JobStatus>,
166 ) -> Result<u64>;
167
168 /// Count active (`queued` + `running`) jobs for rate-limit checks.
169 async fn count_active_jobs_for_task(&self, task_name: &str) -> Result<u32>;
170
171 /// Find non-terminal job by idempotency key.
172 ///
173 /// Used by [`Self::enqueue_with_policies`] to return
174 /// [`JobEnqueueDisposition::ReusedIdempotent`](JobEnqueueDisposition::ReusedIdempotent)
175 /// when a job with the same key is still `queued` or `running`. Returns `None` when no
176 /// matching non-terminal job exists.
177 async fn find_nonterminal_by_idempotency_key(
178 &self,
179 key: &str,
180 ) -> Result<Option<String>>;
181
182 // --- Runs ---
183
184 /// Persist or update a run row.
185 async fn upsert_run(&self, run: &Run) -> Result<()>;
186
187 /// Load one run by id.
188 async fn get_run(&self, run_id: &str) -> Result<Option<Run>>;
189
190 /// List runs with optional job filter and pagination.
191 async fn list_runs(
192 &self,
193 job_id_filter: Option<&str>,
194 offset: usize,
195 limit: usize,
196 ) -> Result<Vec<Run>>;
197
198 /// Mark a run terminal with outcome fields.
199 async fn finish_run(
200 &self,
201 run_id: &str,
202 status: RunStatus,
203 duration_ms: Option<i64>,
204 error_message: Option<String>,
205 ) -> Result<()>;
206
207 /// Count runs, optionally filtered by job id.
208 async fn count_runs(&self, job_id_filter: Option<&str>) -> Result<u64>;
209
210 /// Count runs with `started_at >= since`.
211 async fn count_runs_since(&self, since: DateTime<Utc>) -> Result<u64>;
212
213 /// Aggregate run totals for one task.
214 async fn task_run_stats(&self, task_name: &str) -> Result<TaskRunStats>;
215
216 // --- Task config ---
217
218 /// Load task config by name.
219 async fn get_task_config(&self, task_name: &str) -> Result<Option<TaskConfig>>;
220
221 /// Persist task config.
222 async fn upsert_task_config(&self, config: &TaskConfig) -> Result<()>;
223
224 // --- Leases (distributed workers) ---
225 //
226 // Required when multiple worker processes share one backend (`BOSON_LEASE_TTL_SECS` > 0).
227 // Single-node dev backends may no-op these methods.
228
229 /// Attempt to claim a run lease for `job_id` (stored as lease task id).
230 ///
231 /// Returns `Some(lease_id)` when this worker holds the lease, or `None` when another worker
232 /// holds an active lease for the same job.
233 async fn try_claim_run_lease(
234 &self,
235 job_id: &str,
236 worker_id: &str,
237 ttl_secs: i64,
238 ) -> Result<Option<String>>;
239
240 /// Extend lease TTL for a held lease.
241 async fn extend_lease(&self, lease_id: &str, ttl_secs: i64) -> Result<()>;
242
243 /// Release a held lease.
244 async fn release_lease(&self, lease_id: &str) -> Result<()>;
245
246 /// Expired leases as `(lease_record_id, job_id)`.
247 async fn expired_lease_job_pairs(&self) -> Result<Vec<(String, String)>>;
248}
249
250/// Resolve the default backend from the process-global [`QueueRouter`](crate::QueueRouter).
251///
252/// # Errors
253///
254/// Returns [`BosonError::Internal`](crate::BosonError::Internal) when the global router
255/// was not installed via [`QueueRouter::set_global`](crate::QueueRouter::set_global).
256/// Propagates [`QueueRouter::resolve`](crate::QueueRouter::resolve) errors for the default name.
257pub fn default_backend_from_global() -> Result<Arc<dyn QueueBackend>> {
258 let router = crate::QueueRouter::try_global().ok_or_else(|| {
259 crate::BosonError::Internal("QueueRouter::set_global was not called".into())
260 })?;
261 router.resolve(DEFAULT_BACKEND_NAME)
262}