boson_backend_mem/
mem_queue_backend.rs

1//! In-memory [`QueueBackend`](boson_core::QueueBackend) implementation.
2
3use std::sync::RwLock;
4
5use async_trait::async_trait;
6use boson_core::{
7    Job, JobEnqueueDisposition, JobStatus, QueueBackend, Result, Run, RunStatus, TaskConfig,
8    TaskRunStats,
9};
10use chrono::{DateTime, Utc};
11
12use crate::enqueue_rate::EnqueueRateLimiter;
13use crate::store::{read, write, write_fallible, Inner};
14
15/// Process-local queue backend (not durable).
16///
17/// **When to use:** Mode 1 embedded apps, CI, and unit tests. **Not** for Mode 2 remote workers —
18/// another process cannot see this memory. Prefer
19/// [`SqliteQueueBackend`](https://docs.rs/boson-backend-sqlite) or
20/// [`PostgresQueueBackend`](https://docs.rs/boson-backend-postgres) when processes share a queue.
21///
22/// Getting started:
23/// [Mode 1](https://docs.rs/uf-boson/latest/boson/index.html#mode-1--embedded-one-binary).
24///
25/// Thread-safe via `RwLock`.
26#[derive(Debug)]
27pub struct MemQueueBackend {
28    inner: RwLock<Inner>,
29    enqueue_rate: EnqueueRateLimiter,
30}
31
32impl Default for MemQueueBackend {
33    fn default() -> Self {
34        Self::new()
35    }
36}
37
38impl MemQueueBackend {
39    /// New empty backend.
40    ///
41    /// # Examples
42    ///
43    /// Wire into [`Boson::builder`](https://docs.rs/boson-runtime/latest/boson_runtime/struct.Boson.html#method.builder)
44    /// (feature `mem` on the `boson` facade):
45    ///
46    /// ```rust
47    /// use std::sync::Arc;
48    ///
49    /// use boson_backend_mem::MemQueueBackend;
50    /// use boson_core::QueueBackend;
51    ///
52    /// let backend: Arc<dyn QueueBackend> = Arc::new(MemQueueBackend::new());
53    /// let _ = backend;
54    /// ```
55    ///
56    /// Full boot with [`Boson`](https://docs.rs/boson-runtime):
57    /// `Boson::builder().queue_backend(Arc::new(MemQueueBackend::new()))…`.
58    #[must_use]
59    pub fn new() -> Self {
60        Self {
61            inner: RwLock::new(Inner::new()),
62            enqueue_rate: EnqueueRateLimiter::new(),
63        }
64    }
65}
66
67#[async_trait]
68impl QueueBackend for MemQueueBackend {
69    async fn upsert_job(&self, job: &Job) -> Result<()> {
70        write(&self.inner, |inner| crate::jobs::upsert_job(inner, job))
71    }
72
73    async fn enqueue_with_policies(
74        &self,
75        job: Job,
76        task_config: &TaskConfig,
77    ) -> Result<(String, JobEnqueueDisposition)> {
78        write_fallible(&self.inner, |inner| {
79            crate::jobs::enqueue_with_policies(inner, &self.enqueue_rate, &job, task_config)
80        })
81    }
82
83    async fn get_job(&self, job_id: &str) -> Result<Option<Job>> {
84        read(&self.inner, |inner| crate::jobs::get_job(inner, job_id))
85    }
86
87    async fn list_jobs(
88        &self,
89        status_filter: Option<JobStatus>,
90        offset: usize,
91        limit: usize,
92    ) -> Result<Vec<Job>> {
93        read(&self.inner, |inner| {
94            crate::jobs::list_jobs(inner, status_filter, offset, limit)
95        })
96    }
97
98    async fn cancel_job_if_active(&self, job_id: &str) -> Result<()> {
99        write_fallible(&self.inner, |inner| crate::jobs::cancel_job_if_active(inner, job_id))
100    }
101
102    async fn try_claim_job(&self, job_id: &str) -> Result<Option<Job>> {
103        write(&self.inner, |inner| crate::jobs::try_claim_job(inner, job_id))
104    }
105
106    async fn revert_job_to_queued(&self, job_id: &str) -> Result<()> {
107        write(&self.inner, |inner| crate::jobs::revert_job_to_queued(inner, job_id))
108    }
109
110    async fn distinct_pools_queued(&self) -> Result<Vec<String>> {
111        read(&self.inner, crate::jobs::distinct_pools_queued)
112    }
113
114    async fn list_queued_for_pool_sorted(
115        &self,
116        pool: &str,
117        limit: usize,
118    ) -> Result<Vec<Job>> {
119        read(&self.inner, |inner| {
120            crate::jobs::list_queued_for_pool_sorted(inner, pool, limit)
121        })
122    }
123
124    async fn pop_claim_from_pool(&self, pool: &str) -> Result<Option<Job>> {
125        write(&self.inner, |inner| crate::jobs::pop_claim_from_pool(inner, pool))
126    }
127
128    async fn count_jobs(&self, status_filter: Option<JobStatus>) -> Result<u64> {
129        read(&self.inner, |inner| crate::jobs::count_jobs(inner, status_filter))
130    }
131
132    async fn count_jobs_for_task(
133        &self,
134        task_name: &str,
135        status: Option<JobStatus>,
136    ) -> Result<u64> {
137        read(&self.inner, |inner| {
138            crate::jobs::count_jobs_for_task(inner, task_name, status)
139        })
140    }
141
142    async fn count_active_jobs_for_task(&self, task_name: &str) -> Result<u32> {
143        read(&self.inner, |inner| {
144            crate::jobs::count_active_jobs_for_task(inner, task_name)
145        })
146    }
147
148    async fn find_nonterminal_by_idempotency_key(&self, key: &str) -> Result<Option<String>> {
149        read(&self.inner, |inner| {
150            crate::jobs::find_nonterminal_by_idempotency_key(inner, key)
151        })
152    }
153
154    async fn upsert_run(&self, run: &Run) -> Result<()> {
155        write(&self.inner, |inner| crate::runs::upsert_run(inner, run))
156    }
157
158    async fn get_run(&self, run_id: &str) -> Result<Option<Run>> {
159        read(&self.inner, |inner| crate::runs::get_run(inner, run_id))
160    }
161
162    async fn list_runs(
163        &self,
164        job_id_filter: Option<&str>,
165        offset: usize,
166        limit: usize,
167    ) -> Result<Vec<Run>> {
168        read(&self.inner, |inner| {
169            crate::runs::list_runs(inner, job_id_filter, offset, limit)
170        })
171    }
172
173    async fn finish_run(
174        &self,
175        run_id: &str,
176        status: RunStatus,
177        duration_ms: Option<i64>,
178        error_message: Option<String>,
179    ) -> Result<()> {
180        write(&self.inner, |inner| {
181            crate::runs::finish_run(inner, run_id, status, duration_ms, error_message);
182        })
183    }
184
185    async fn count_runs(&self, job_id_filter: Option<&str>) -> Result<u64> {
186        read(&self.inner, |inner| crate::runs::count_runs(inner, job_id_filter))
187    }
188
189    async fn count_runs_since(&self, since: DateTime<Utc>) -> Result<u64> {
190        read(&self.inner, |inner| crate::runs::count_runs_since(inner, since))
191    }
192
193    async fn task_run_stats(&self, task_name: &str) -> Result<TaskRunStats> {
194        read(&self.inner, |inner| crate::runs::task_run_stats(inner, task_name))
195    }
196
197    async fn get_task_config(&self, task_name: &str) -> Result<Option<TaskConfig>> {
198        read(&self.inner, |inner| crate::task_config::get_task_config(inner, task_name))
199    }
200
201    async fn upsert_task_config(&self, config: &TaskConfig) -> Result<()> {
202        write(&self.inner, |inner| crate::task_config::upsert_task_config(inner, config))
203    }
204
205    async fn try_claim_run_lease(
206        &self,
207        job_id: &str,
208        worker_id: &str,
209        ttl_secs: i64,
210    ) -> Result<Option<String>> {
211        write(&self.inner, |inner| {
212            crate::leases::try_claim_run_lease(inner, job_id, worker_id, ttl_secs)
213        })
214    }
215
216    async fn extend_lease(&self, lease_id: &str, ttl_secs: i64) -> Result<()> {
217        write(&self.inner, |inner| crate::leases::extend_lease(inner, lease_id, ttl_secs))
218    }
219
220    async fn release_lease(&self, lease_id: &str) -> Result<()> {
221        write(&self.inner, |inner| crate::leases::release_lease(inner, lease_id))
222    }
223
224    async fn expired_lease_job_pairs(&self) -> Result<Vec<(String, String)>> {
225        read(&self.inner, crate::leases::expired_lease_job_pairs)
226    }
227}
boson_backend_mem/mem_queue_backend.rs

boson_backend_mem/
mem_queue_backend.rs
