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(&self, task_name: &str, status: Option<JobStatus>) -> Result<u64> {
63        self.count_jobs_for_task_impl(task_name, status).await
64    }
65
66    async fn count_active_jobs_for_task(&self, task_name: &str) -> Result<u32> {
67        self.count_active_jobs_for_task_impl(task_name).await
68    }
69
70    async fn find_nonterminal_by_idempotency_key(&self, key: &str) -> Result<Option<String>> {
71        self.find_nonterminal_by_idempotency_key_impl(key).await
72    }
73
74    async fn upsert_run(&self, run: &Run) -> Result<()> {
75        self.upsert_run_impl(run).await
76    }
77
78    async fn get_run(&self, run_id: &str) -> Result<Option<Run>> {
79        self.get_run_impl(run_id).await
80    }
81
82    async fn list_runs(
83        &self,
84        job_id_filter: Option<&str>,
85        offset: usize,
86        limit: usize,
87    ) -> Result<Vec<Run>> {
88        self.list_runs_impl(job_id_filter, offset, limit).await
89    }
90
91    async fn finish_run(
92        &self,
93        run_id: &str,
94        status: RunStatus,
95        duration_ms: Option<i64>,
96        error_message: Option<String>,
97    ) -> Result<()> {
98        self.finish_run_impl(run_id, status, duration_ms, error_message)
99            .await
100    }
101
102    async fn count_runs(&self, job_id_filter: Option<&str>) -> Result<u64> {
103        self.count_runs_impl(job_id_filter).await
104    }
105
106    async fn count_runs_since(&self, since: DateTime<Utc>) -> Result<u64> {
107        self.count_runs_since_impl(since).await
108    }
109
110    async fn task_run_stats(&self, task_name: &str) -> Result<TaskRunStats> {
111        self.task_run_stats_impl(task_name).await
112    }
113
114    async fn get_task_config(&self, task_name: &str) -> Result<Option<TaskConfig>> {
115        self.get_task_config_impl(task_name).await
116    }
117
118    async fn upsert_task_config(&self, config: &TaskConfig) -> Result<()> {
119        self.upsert_task_config_impl(config).await
120    }
121
122    async fn try_claim_run_lease(
123        &self,
124        job_id: &str,
125        worker_id: &str,
126        ttl_secs: i64,
127    ) -> Result<Option<String>> {
128        self.try_claim_run_lease_impl(job_id, worker_id, ttl_secs)
129            .await
130    }
131
132    async fn extend_lease(&self, lease_id: &str, ttl_secs: i64) -> Result<()> {
133        self.extend_lease_impl(lease_id, ttl_secs).await
134    }
135
136    async fn release_lease(&self, lease_id: &str) -> Result<()> {
137        self.release_lease_impl(lease_id).await
138    }
139
140    async fn expired_lease_job_pairs(&self) -> Result<Vec<(String, String)>> {
141        self.expired_lease_job_pairs_impl().await
142    }
143}
boson_backend_sql_common/queue_impl.rs

boson_backend_sql_common/
queue_impl.rs
