haste_repository/
async_job.rs

1//! Storage for asynchronous jobs: work a request kicks off and a worker
2//! finishes later, followed by the client through the FHIR Asynchronous
3//! Request Pattern.
4//!
5//! A job is one row in `async_jobs`. The server creates it (`accepted`), a
6//! worker claims it (`in-progress`) and holds it under a lease it renews while
7//! it works, then records the result (`completed` / `failed`). The client can
8//! cancel it at any point (`cancelled`), and every job expires.
9//!
10//! See the `20261014120000_async_jobs` migration for why this is its own table
11//! rather than a kind on the indexing lock table.
12
13use haste_fhir_operation_error::OperationOutcomeError;
14use haste_jwt::{ProjectId, TenantId, claims::UserTokenClaims};
15use serde::{Deserialize, Serialize};
16use sqlx::types::{Json, time::OffsetDateTime};
17use std::time::Duration;
18
19/// Where a job is in its life. Mirrors the `async_job_status` enum.
20#[derive(Clone, Copy, Debug, PartialEq, Eq, sqlx::Type, Serialize, Deserialize)]
21#[sqlx(type_name = "async_job_status", rename_all = "kebab-case")]
22#[serde(rename_all = "kebab-case")]
23pub enum AsyncJobStatus {
24    Accepted,
25    InProgress,
26    Completed,
27    Failed,
28    Cancelled,
29}
30
31impl AsyncJobStatus {
32    #[must_use]
33    pub fn as_str(&self) -> &'static str {
34        match self {
35            AsyncJobStatus::Accepted => "accepted",
36            AsyncJobStatus::InProgress => "in-progress",
37            AsyncJobStatus::Completed => "completed",
38            AsyncJobStatus::Failed => "failed",
39            AsyncJobStatus::Cancelled => "cancelled",
40        }
41    }
42
43    /// Whether the job may still run: queued, or claimed and running.
44    #[must_use]
45    pub fn is_live(&self) -> bool {
46        matches!(self, AsyncJobStatus::Accepted | AsyncJobStatus::InProgress)
47    }
48}
49
50impl std::fmt::Display for AsyncJobStatus {
51    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
52        f.write_str(self.as_str())
53    }
54}
55
56/// A job as stored.
57#[derive(sqlx::FromRow, Clone, Debug)]
58pub struct AsyncJob {
59    pub id: String,
60    pub tenant: TenantId,
61    pub project: ProjectId,
62    pub kind: String,
63    pub status: AsyncJobStatus,
64    pub request: String,
65    pub parameters: Json<serde_json::Value>,
66    /// The submitter's verified token claims; the worker runs the job as them.
67    pub claims: Json<UserTokenClaims>,
68    pub submitted_by: String,
69    pub progress: Option<String>,
70    pub output: Option<Json<serde_json::Value>>,
71    pub output_content_type: Option<String>,
72    /// An `OperationOutcome`, once failed.
73    pub error: Option<Json<serde_json::Value>>,
74    pub attempts: i32,
75    pub max_attempts: i32,
76    pub locked_by: Option<String>,
77    pub lease_expires_at: Option<OffsetDateTime>,
78    pub created_at: OffsetDateTime,
79    pub updated_at: OffsetDateTime,
80    pub started_at: Option<OffsetDateTime>,
81    pub finished_at: Option<OffsetDateTime>,
82    pub expires_at: OffsetDateTime,
83}
84
85/// What a kick-off records about a new job.
86#[derive(Debug)]
87pub struct CreateAsyncJob {
88    pub kind: String,
89    pub request: String,
90    pub parameters: serde_json::Value,
91    pub claims: UserTokenClaims,
92    /// How long the job and its result are kept, from now and again from
93    /// completion.
94    pub retention: Duration,
95    pub max_attempts: i32,
96}
97
98/// Identifies the worker holding a job, and for how long one renewal lasts.
99#[derive(Clone, Debug)]
100pub struct JobLease {
101    pub worker: String,
102    pub duration: Duration,
103}
104
105/// What a lease renewal found.
106#[derive(Clone, Copy, Debug, PartialEq, Eq)]
107pub enum LeaseRenewal {
108    /// Still ours; keep going.
109    Renewed,
110    /// The job is no longer ours to run: it was cancelled, or its lease lapsed
111    /// and another worker took it. Stop working on it.
112    Lost,
113}
114
115/// How a finished job ended.
116#[derive(Clone, Debug)]
117pub enum JobResult {
118    Completed {
119        output: serde_json::Value,
120        content_type: String,
121    },
122    Failed {
123        /// An `OperationOutcome` resource as JSON.
124        outcome: serde_json::Value,
125    },
126}
127
128pub trait AsyncJobProvider {
129    /// Queues a job. Its id is generated here and is what the status URL
130    /// carries.
131    fn create_job(
132        &self,
133        tenant: &TenantId,
134        project: &ProjectId,
135        job: CreateAsyncJob,
136    ) -> impl Future<Output = Result<AsyncJob, OperationOutcomeError>> + Send;
137
138    /// A job of this project, whatever its state, until it expires.
139    fn read_job(
140        &self,
141        tenant: &TenantId,
142        project: &ProjectId,
143        id: &str,
144    ) -> impl Future<Output = Result<Option<AsyncJob>, OperationOutcomeError>> + Send;
145
146    /// Jobs of this project that are queued or running. Kick-offs cap this.
147    fn count_live_jobs(
148        &self,
149        tenant: &TenantId,
150        project: &ProjectId,
151    ) -> impl Future<Output = Result<i64, OperationOutcomeError>> + Send;
152
153    /// Cancels a job of this project. A queued job never runs; a running one
154    /// finds out at its next lease renewal; a finished one gives up its
155    /// result. Returns the job as it was before, or `None` if there is no
156    /// such job or it was already cancelled.
157    fn cancel_job(
158        &self,
159        tenant: &TenantId,
160        project: &ProjectId,
161        id: &str,
162    ) -> impl Future<Output = Result<Option<AsyncJob>, OperationOutcomeError>> + Send;
163
164    /// Takes the oldest claimable job of one of `kinds` for `lease.worker`:
165    /// queued, or in progress with a lapsed lease, and with attempts left.
166    /// Claiming is one statement, so two workers never take the same job.
167    fn claim_job(
168        &self,
169        kinds: &[&str],
170        lease: &JobLease,
171    ) -> impl Future<Output = Result<Option<AsyncJob>, OperationOutcomeError>> + Send;
172
173    /// Extends the lease on a job this worker holds and records `progress`
174    /// if given. `Lost` means stop: the job was cancelled or taken over.
175    fn renew_lease(
176        &self,
177        id: &str,
178        lease: &JobLease,
179        progress: Option<&str>,
180    ) -> impl Future<Output = Result<LeaseRenewal, OperationOutcomeError>> + Send;
181
182    /// Records the result of a job this worker holds and releases the lease.
183    /// Returns false if the job was no longer this worker's to finish (it
184    /// was cancelled or taken over meanwhile), in which case the result is
185    /// dropped.
186    fn finish_job(
187        &self,
188        id: &str,
189        worker: &str,
190        result: JobResult,
191        retention: Duration,
192    ) -> impl Future<Output = Result<bool, OperationOutcomeError>> + Send;
193
194    /// Fails every in-progress job whose lease has lapsed and that has no
195    /// attempts left, recording `outcome` on each. Returns how many.
196    fn fail_abandoned_jobs(
197        &self,
198        outcome: &serde_json::Value,
199        retention: Duration,
200    ) -> impl Future<Output = Result<u64, OperationOutcomeError>> + Send;
201
202    /// Removes every job past its `expires_at`, returning them so their
203    /// handlers can remove what they produced.
204    fn delete_expired_jobs(
205        &self,
206        limit: i64,
207    ) -> impl Future<Output = Result<Vec<AsyncJob>, OperationOutcomeError>> + Send;
208}
haste_repository/async_job.rs

haste_repository/
async_job.rs
