azums_core/backend/
mod.rs

1pub mod memory;
2pub mod mock;
3pub mod observability;
4pub mod stream;
5
6pub use memory::{MemoryAttempt, MemoryBackend};
7pub use mock::{CallRecord, MockBackend};
8pub use observability::{JobExplanation, JobObservationEvent, ObservabilityBackend, QueueMetrics};
9pub use stream::StreamBackend;
10
11use crate::model::{BackendCapabilities, BackendSemanticCapabilities, Job, JobListItem, NewJob};
12use async_trait::async_trait;
13use chrono::{DateTime, Utc};
14use std::pin::Pin;
15use uuid::Uuid;
16
17/// Type alias for asynchronous notification event streams produced by [`StorageBackend::subscribe`].
18pub type NotificationStream = Pin<Box<dyn futures_core::Stream<Item = ()> + Send>>;
19
20/// Async, backend-agnostic storage interface for job queue operations.
21///
22/// Implementations of `StorageBackend` manage job persistence, leasing, retry scheduling,
23/// Dead-Letter Queue (DLQ) routing, maintenance archiving, and health probes.
24#[async_trait]
25pub trait StorageBackend: Send + Sync {
26    /// Declares this backend's storage and coordination capabilities.
27    fn capabilities(&self) -> BackendCapabilities;
28
29    /// Declares detailed semantic strength for this backend.
30    ///
31    /// Built-in profiles are derived from [`Self::capabilities`]. Custom backends with a novel
32    /// capability combination should override this method instead of leaving the result unknown.
33    fn semantic_capabilities(&self) -> Option<BackendSemanticCapabilities> {
34        self.capabilities().semantics()
35    }
36
37    /// Returns reference to StreamBackend if supported by this storage implementation.
38    fn as_stream(&self) -> Option<&dyn StreamBackend> {
39        None
40    }
41
42    /// Returns reference to backend-native observability data if supported.
43    fn as_observability(&self) -> Option<&dyn ObservabilityBackend> {
44        None
45    }
46
47    /// Executes backend schema migrations or setup steps.
48    async fn run_migrations(&self) -> anyhow::Result<()>;
49
50    /// Performs a health check to verify backend connectivity and readiness.
51    async fn health_check(&self) -> anyhow::Result<()>;
52
53    /// Enqueues a new job into the backend queue.
54    async fn enqueue(&self, job: NewJob) -> anyhow::Result<Uuid>;
55
56    /// Subscribes to job enqueue notification events for a specific queue.
57    async fn subscribe(&self, queue: &str) -> anyhow::Result<NotificationStream>;
58
59    /// Leases up to `batch_size` runnable jobs for a specified worker ID.
60    async fn lease_jobs_batch(
61        &self,
62        queue: &str,
63        worker_id: &str,
64        lease_seconds: i64,
65        batch_size: i64,
66    ) -> anyhow::Result<Vec<Job>>;
67
68    /// Leases up to `batch_size` runnable jobs with specified queue ordering preference (`QueueOrdering`).
69    async fn lease_jobs_batch_with_ordering(
70        &self,
71        queue: &str,
72        worker_id: &str,
73        lease_seconds: i64,
74        batch_size: i64,
75        ordering: crate::model::QueueOrdering,
76    ) -> anyhow::Result<Vec<Job>> {
77        let _ = ordering;
78        self.lease_jobs_batch(queue, worker_id, lease_seconds, batch_size)
79            .await
80    }
81
82    /// Reaps expired locks from inactive workers, resetting their status back to queued.
83    async fn reap_expired_locks(&self) -> anyhow::Result<u64>;
84
85    /// Starts job execution attempt records, returning `(job_id, attempt_id, attempt_number)` tuples.
86    async fn start_attempts_batch(
87        &self,
88        dataset_ids: &[String],
89        job_ids: &[Uuid],
90        worker_id: &str,
91    ) -> anyhow::Result<Vec<(Uuid, Uuid, i32)>>;
92
93    /// Marks a single job execution attempt as succeeded.
94    async fn mark_succeeded(
95        &self,
96        job_id: Uuid,
97        attempt_id: Uuid,
98        worker_id: &str,
99        latency_ms: i32,
100    ) -> anyhow::Result<()>;
101
102    /// Marks a batch of job execution attempts as succeeded.
103    async fn mark_succeeded_batch(
104        &self,
105        dataset_id: &str,
106        updates: &[(Uuid, Uuid, i32)],
107        worker_id: &str,
108    ) -> anyhow::Result<()>;
109
110    /// Records a failed attempt and reschedules the job for a future retry attempt.
111    #[allow(clippy::too_many_arguments)]
112    async fn reschedule_for_retry(
113        &self,
114        job_id: Uuid,
115        attempt_id: Uuid,
116        worker_id: &str,
117        latency_ms: i32,
118        next_run_at: DateTime<Utc>,
119        error_code: &str,
120        error_message: &str,
121        attempt_no: i32,
122    ) -> anyhow::Result<()>;
123
124    /// Records a failed attempt and transitions the job to the Dead-Letter Queue (DLQ).
125    #[allow(clippy::too_many_arguments)]
126    async fn mark_dlq(
127        &self,
128        job_id: Uuid,
129        attempt_id: Uuid,
130        worker_id: &str,
131        latency_ms: i32,
132        reason_code: &str,
133        error_code: &str,
134        error_message: &str,
135        attempt_no: i32,
136    ) -> anyhow::Result<()>;
137
138    /// Moves succeeded jobs older than `cutoff` into an archive table or storage location.
139    async fn archive_succeeded_older_than(
140        &self,
141        cutoff: DateTime<Utc>,
142        limit: i64,
143    ) -> anyhow::Result<u64>;
144
145    /// Prunes attempt audit logs and decision records for succeeded jobs older than `cutoff`.
146    async fn delete_history_for_succeeded_older_than(
147        &self,
148        cutoff: DateTime<Utc>,
149        limit: i64,
150    ) -> anyhow::Result<(u64, u64)>;
151
152    /// Performs database maintenance (e.g., VACUUM ANALYZE in Postgres, PRAGMA incremental_vacuum in SQLite).
153    async fn perform_maintenance(&self) -> anyhow::Result<()> {
154        Ok(())
155    }
156
157    /// Extends the lease lock expiration for an in-flight running job.
158    /// Returns `true` if the lease was extended, `false` if the job lock was lost or reaped.
159    async fn extend_lease(
160        &self,
161        _job_id: Uuid,
162        _worker_id: &str,
163        _lease_seconds: i64,
164    ) -> anyhow::Result<bool> {
165        Ok(true)
166    }
167
168    /// Cancels a queued/scheduled job or a running job owned by `worker_id`.
169    ///
170    /// Queued and scheduled jobs may be cancelled without a worker ID. Running jobs require the
171    /// worker lease owner so terminality cannot be bypassed by another worker. Terminal jobs reject
172    /// cancellation.
173    async fn cancel_job(&self, job_id: Uuid, worker_id: Option<&str>) -> anyhow::Result<()>;
174
175    /// Fetches a single job record by ID.
176    async fn get_job(&self, job_id: Uuid) -> anyhow::Result<Option<Job>>;
177
178    /// Fetches a list of jobs matching filters with cursor pagination.
179    async fn list_jobs(
180        &self,
181        queue: Option<&str>,
182        status: Option<&str>,
183        limit: i64,
184        cursor_created_at: Option<DateTime<Utc>>,
185        cursor_id: Option<Uuid>,
186    ) -> anyhow::Result<Vec<JobListItem>>;
187
188    /// Atomically replays a job by ID into the queue.
189    async fn replay_job(
190        &self,
191        job_id: Uuid,
192        override_queue: Option<&str>,
193        override_run_at: Option<DateTime<Utc>>,
194    ) -> anyhow::Result<Uuid>;
195
196    /// Dequeues and leases up to `batch_size` runnable jobs (alias for `lease_jobs_batch`).
197    async fn dequeue_and_lease(
198        &self,
199        queue: &str,
200        worker_id: &str,
201        lease_seconds: i64,
202        batch_size: i64,
203    ) -> anyhow::Result<Vec<Job>> {
204        self.lease_jobs_batch(queue, worker_id, lease_seconds, batch_size)
205            .await
206    }
207
208    /// Marks a job attempt as completed (alias for `mark_succeeded`).
209    async fn complete_job(
210        &self,
211        job_id: Uuid,
212        attempt_id: Uuid,
213        worker_id: &str,
214        latency_ms: i32,
215    ) -> anyhow::Result<()> {
216        self.mark_succeeded(job_id, attempt_id, worker_id, latency_ms)
217            .await
218    }
219
220    /// Reschedules a job for retry (alias for `reschedule_for_retry`).
221    #[allow(clippy::too_many_arguments)]
222    async fn retry_job(
223        &self,
224        job_id: Uuid,
225        attempt_id: Uuid,
226        worker_id: &str,
227        latency_ms: i32,
228        next_run_at: DateTime<Utc>,
229        error_code: &str,
230        error_message: &str,
231        attempt_no: i32,
232    ) -> anyhow::Result<()> {
233        self.reschedule_for_retry(
234            job_id,
235            attempt_id,
236            worker_id,
237            latency_ms,
238            next_run_at,
239            error_code,
240            error_message,
241            attempt_no,
242        )
243        .await
244    }
245
246    /// Moves a job to DLQ on failure (alias for `mark_dlq`).
247    #[allow(clippy::too_many_arguments)]
248    async fn fail_job(
249        &self,
250        job_id: Uuid,
251        attempt_id: Uuid,
252        worker_id: &str,
253        latency_ms: i32,
254        reason_code: &str,
255        error_code: &str,
256        error_message: &str,
257        attempt_no: i32,
258    ) -> anyhow::Result<()> {
259        self.mark_dlq(
260            job_id,
261            attempt_id,
262            worker_id,
263            latency_ms,
264            reason_code,
265            error_code,
266            error_message,
267            attempt_no,
268        )
269        .await
270    }
271}
azums_core/backend/mod.rs

azums_core/backend/
mod.rs
