graphile_worker_utils/
executor.rs

1use std::time::Duration;
2
3use graphile_worker_database::DbExecutorArg;
4use graphile_worker_job::{DbJob, Job};
5use graphile_worker_job_spec::JobSpec;
6use graphile_worker_queries::add_job::types::RawJobSpec;
7use graphile_worker_queries::errors::GraphileWorkerError;
8use graphile_worker_recovery::ActiveWorkerRow;
9use graphile_worker_task_handler::{BatchTaskHandler, TaskHandler};
10use serde::Serialize;
11
12use super::client::WorkerUtils;
13use super::types::RescheduleJobOptions;
14use super::{actions, add, maintenance};
15
16/// A scoped `WorkerUtils` facade that routes operations through a caller-provided executor.
17///
18/// Use [`WorkerUtils::with_executor`] when job operations must participate in an existing
19/// transaction. The caller owns the executor lifecycle and remains responsible for committing or
20/// rolling back transactions.
21///
22/// Maintenance operations that manage their own transaction or shared state (`cleanup`,
23/// `migrate`, and stale-worker sweeps) remain available only on [`WorkerUtils`].
24pub struct WorkerUtilsWithExecutor<E> {
25    utils: WorkerUtils,
26    executor: E,
27}
28
29impl<E> WorkerUtilsWithExecutor<E>
30where
31    E: DbExecutorArg,
32{
33    pub(super) fn new(utils: WorkerUtils, executor: E) -> Self {
34        Self { utils, executor }
35    }
36
37    /// Returns the configured `WorkerUtils`, releasing the injected executor.
38    pub fn into_inner(self) -> WorkerUtils {
39        self.utils
40    }
41
42    /// Adds a job to the queue with type safety.
43    #[tracing::instrument(
44        "add_job",
45        skip_all,
46        fields(
47            messaging.system = "graphile-worker",
48            messaging.operation.name = "add_job"
49        )
50    )]
51    pub async fn add_job<T: TaskHandler>(
52        &mut self,
53        payload: T,
54        spec: JobSpec,
55    ) -> Result<Job, GraphileWorkerError> {
56        add::add_job(&self.utils, &mut self.executor, payload, spec).await
57    }
58
59    /// Adds a job to the queue with a raw identifier and payload.
60    #[tracing::instrument(
61        "add_raw_job",
62        skip_all,
63        fields(
64            messaging.system = "graphile-worker",
65            messaging.operation.name = "add_job"
66        )
67    )]
68    pub async fn add_raw_job<P>(
69        &mut self,
70        identifier: &str,
71        payload: P,
72        spec: JobSpec,
73    ) -> Result<Job, GraphileWorkerError>
74    where
75        P: Serialize,
76    {
77        add::add_raw_job(&self.utils, &mut self.executor, identifier, payload, spec).await
78    }
79
80    /// Adds multiple jobs of the same type in one database operation.
81    #[tracing::instrument(
82        "add_jobs",
83        skip_all,
84        fields(
85            messaging.system = "graphile-worker",
86            messaging.operation.name = "add_jobs",
87            messaging.batch.message_count = jobs.len()
88        )
89    )]
90    pub async fn add_jobs<T: TaskHandler + Clone>(
91        &mut self,
92        jobs: &[(T, &JobSpec)],
93    ) -> Result<Vec<Job>, GraphileWorkerError> {
94        add::add_jobs(&self.utils, &mut self.executor, jobs).await
95    }
96
97    /// Adds multiple jobs with raw identifiers and payloads in one database operation.
98    #[tracing::instrument(
99        "add_raw_jobs",
100        skip_all,
101        fields(
102            messaging.system = "graphile-worker",
103            messaging.operation.name = "add_jobs",
104            messaging.batch.message_count = jobs.len()
105        )
106    )]
107    pub async fn add_raw_jobs(
108        &mut self,
109        jobs: &[RawJobSpec],
110    ) -> Result<Vec<Job>, GraphileWorkerError> {
111        add::add_raw_jobs(&self.utils, &mut self.executor, jobs).await
112    }
113
114    /// Adds a batch job to the queue with type safety.
115    #[tracing::instrument(
116        "add_batch_job",
117        skip_all,
118        fields(
119            messaging.system = "graphile-worker",
120            messaging.operation.name = "add_batch_job",
121            messaging.batch.message_count = payloads.len()
122        )
123    )]
124    pub async fn add_batch_job<T: BatchTaskHandler>(
125        &mut self,
126        payloads: Vec<T>,
127        spec: JobSpec,
128    ) -> Result<Job, GraphileWorkerError> {
129        add::add_batch_job(&self.utils, &mut self.executor, payloads, spec).await
130    }
131
132    /// Removes a job from the queue by its job key.
133    pub async fn remove_job(&mut self, job_key: &str) -> Result<(), GraphileWorkerError> {
134        actions::remove_job(&self.utils, &mut self.executor, job_key).await
135    }
136
137    /// Marks multiple jobs as completed.
138    pub async fn complete_jobs(&mut self, ids: &[i64]) -> Result<Vec<DbJob>, GraphileWorkerError> {
139        actions::complete_jobs(&self.utils, &mut self.executor, ids).await
140    }
141
142    /// Marks multiple jobs as permanently failed with a reason.
143    pub async fn permanently_fail_jobs(
144        &mut self,
145        ids: &[i64],
146        reason: &str,
147    ) -> Result<Vec<DbJob>, GraphileWorkerError> {
148        actions::permanently_fail_jobs(&self.utils, &mut self.executor, ids, reason).await
149    }
150
151    /// Reschedules multiple jobs with new parameters.
152    pub async fn reschedule_jobs(
153        &mut self,
154        ids: &[i64],
155        options: RescheduleJobOptions,
156    ) -> Result<Vec<DbJob>, GraphileWorkerError> {
157        actions::reschedule_jobs(&self.utils, &mut self.executor, ids, options).await
158    }
159
160    /// Lists workers registered in the heartbeat table.
161    pub async fn list_active_workers(
162        &mut self,
163        sweep_threshold: Duration,
164    ) -> Result<Vec<ActiveWorkerRow>, GraphileWorkerError> {
165        maintenance::list_active_workers(&self.utils, &mut self.executor, sweep_threshold).await
166    }
167
168    /// Force unlocks worker records in the database.
169    pub async fn force_unlock_workers(
170        &mut self,
171        worker_ids: &[&str],
172    ) -> Result<(), GraphileWorkerError> {
173        actions::force_unlock_workers(&self.utils, &mut self.executor, worker_ids).await
174    }
175}
graphile_worker_utils/executor.rs

graphile_worker_utils/
executor.rs
