boson_backend_sql_common/
queue_impl.rs

1use async_trait::async_trait;
2use boson_core::{
3    Job, JobEnqueueDisposition, JobStatus, QueueBackend, Result, Run, RunStatus, TaskConfig,
4    TaskRunStats,
5};
6use chrono::{DateTime, Utc};
7
8use crate::SqlQueueBackend;
9
10#[async_trait]
11impl QueueBackend for SqlQueueBackend {
12    async fn upsert_job(&self, job: &Job) -> Result<()> {
13        self.upsert_job_impl(job).await
14    }
15
16    async fn enqueue_with_policies(
17        &self,
18        job: Job,
19        task_config: &TaskConfig,
20    ) -> Result<(String, JobEnqueueDisposition)> {
21        self.enqueue_with_policies_impl(&self.enqueue_rate, job, task_config)
22            .await
23    }
24
25    async fn get_job(&self, job_id: &str) -> Result<Option<Job>> {
26        self.get_job_impl(job_id).await
27    }
28
29    async fn list_jobs(
30        &self,
31        status_filter: Option<JobStatus>,
32        offset: usize,
33        limit: usize,
34    ) -> Result<Vec<Job>> {
35        self.list_jobs_impl(status_filter, offset, limit).await
36    }
37
38    async fn cancel_job_if_active(&self, job_id: &str) -> Result<()> {
39        self.cancel_job_if_active_impl(job_id).await
40    }
41
42    async fn try_claim_job(&self, job_id: &str) -> Result<Option<Job>> {
43        self.try_claim_job_impl(job_id).await
44    }
45
46    async fn revert_job_to_queued(&self, job_id: &str) -> Result<()> {
47        self.revert_job_to_queued_impl(job_id).await
48    }
49
50    async fn distinct_pools_queued(&self) -> Result<Vec<String>> {
51        self.distinct_pools_queued_impl().await
52    }
53
54    async fn list_queued_for_pool_sorted(&self, pool: &str, limit: usize) -> Result<Vec<Job>> {
55        self.list_queued_for_pool_sorted_impl(pool, limit).await
56    }
57
58    async fn count_jobs(&self, status_filter: Option<JobStatus>) -> Result<u64> {
59        self.count_jobs_impl(status_filter).await
60    }
61
62    async fn count_jobs_for_task(
63        &self,
64        task_name: &str,
65        status: Option<JobStatus>,
66    ) -> Result<u64> {
67        self.count_jobs_for_task_impl(task_name, status).await
68    }
69
70    async fn count_active_jobs_for_task(&self, task_name: &str) -> Result<u32> {
71        self.count_active_jobs_for_task_impl(task_name).await
72    }
73
74    async fn find_nonterminal_by_idempotency_key(&self, key: &str) -> Result<Option<String>> {
75        self.find_nonterminal_by_idempotency_key_impl(key).await
76    }
77
78    async fn upsert_run(&self, run: &Run) -> Result<()> {
79        self.upsert_run_impl(run).await
80    }
81
82    async fn get_run(&self, run_id: &str) -> Result<Option<Run>> {
83        self.get_run_impl(run_id).await
84    }
85
86    async fn list_runs(
87        &self,
88        job_id_filter: Option<&str>,
89        offset: usize,
90        limit: usize,
91    ) -> Result<Vec<Run>> {
92        self.list_runs_impl(job_id_filter, offset, limit).await
93    }
94
95    async fn finish_run(
96        &self,
97        run_id: &str,
98        status: RunStatus,
99        duration_ms: Option<i64>,
100        error_message: Option<String>,
101    ) -> Result<()> {
102        self.finish_run_impl(run_id, status, duration_ms, error_message)
103            .await
104    }
105
106    async fn count_runs(&self, job_id_filter: Option<&str>) -> Result<u64> {
107        self.count_runs_impl(job_id_filter).await
108    }
109
110    async fn count_runs_since(&self, since: DateTime<Utc>) -> Result<u64> {
111        self.count_runs_since_impl(since).await
112    }
113
114    async fn task_run_stats(&self, task_name: &str) -> Result<TaskRunStats> {
115        self.task_run_stats_impl(task_name).await
116    }
117
118    async fn get_task_config(&self, task_name: &str) -> Result<Option<TaskConfig>> {
119        self.get_task_config_impl(task_name).await
120    }
121
122    async fn upsert_task_config(&self, config: &TaskConfig) -> Result<()> {
123        self.upsert_task_config_impl(config).await
124    }
125
126    async fn try_claim_run_lease(
127        &self,
128        job_id: &str,
129        worker_id: &str,
130        ttl_secs: i64,
131    ) -> Result<Option<String>> {
132        self.try_claim_run_lease_impl(job_id, worker_id, ttl_secs)
133            .await
134    }
135
136    async fn extend_lease(&self, lease_id: &str, ttl_secs: i64) -> Result<()> {
137        self.extend_lease_impl(lease_id, ttl_secs).await
138    }
139
140    async fn release_lease(&self, lease_id: &str) -> Result<()> {
141        self.release_lease_impl(lease_id).await
142    }
143
144    async fn expired_lease_job_pairs(&self) -> Result<Vec<(String, String)>> {
145        self.expired_lease_job_pairs_impl().await
146    }
147}
boson_backend_sql_common/queue_impl.rs

boson_backend_sql_common/
queue_impl.rs
