spiral/
task_handle.rs

1//! Handles to the durable Tasks that resource methods submit.
2
3use std::fmt::{Debug, Formatter};
4
5use uuid::Uuid;
6
7use crate::api::tasks::{
8    CreateTaskResponse, SubmitTaskRequest, TaskDefinition, TaskOptions, TaskProgress, TaskResult,
9};
10use crate::ids::ProjectId;
11use crate::resources::task::{
12    HttpTaskClient, Task, TaskDetails, TaskStatus, TaskWaitError, TaskWaitOptions,
13};
14use crate::{Error, Project};
15
16/// Who executes a submitted Task. It never changes what the Task does.
17#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
18pub enum Run {
19    /// Any compatible WorkerPool or standalone Worker claims the Task.
20    #[default]
21    Remote,
22}
23
24/// A submitted Task, which resolves to the [`TaskResult`] of versions it published.
25///
26/// Dropping a handle or abandoning [`Self::wait`] never cancels the Task; only
27/// [`Self::cancel`] does.
28#[derive(Clone)]
29pub struct TaskHandle {
30    tasks: HttpTaskClient,
31    task_id: Uuid,
32}
33
34impl Debug for TaskHandle {
35    fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result {
36        formatter
37            .debug_struct("TaskHandle")
38            .field("task_id", &self.task_id)
39            .finish()
40    }
41}
42
43impl Project {
44    /// Submit any Task definition, including kinds no resource method wraps.
45    ///
46    /// Returns `None` when the overlap policy skipped the submission because an equivalent
47    /// Task already waits to run.
48    pub async fn submit_task(
49        &self,
50        definition: TaskDefinition,
51        options: Option<TaskOptions>,
52        run: Run,
53    ) -> Result<Option<TaskHandle>, Error> {
54        submit(
55            self.client.tasks(),
56            ProjectId::from(self.id.as_str()),
57            definition,
58            options,
59            run,
60        )
61        .await
62    }
63
64    /// A handle to a Task this Project already holds. It makes no request.
65    pub fn task(&self, task_id: Uuid) -> TaskHandle {
66        TaskHandle::new(self.client.tasks().clone(), task_id)
67    }
68}
69
70/// Submit `definition` to run as `run` selects.
71pub(crate) async fn submit(
72    tasks: &HttpTaskClient,
73    project_id: ProjectId,
74    definition: TaskDefinition,
75    options: Option<TaskOptions>,
76    run: Run,
77) -> Result<Option<TaskHandle>, Error> {
78    let Run::Remote = run;
79    let response = tasks
80        .submit_task(
81            project_id.to_string(),
82            SubmitTaskRequest {
83                definition,
84                parent_task_id: None,
85                options,
86                claim: None,
87                idempotency_key: Uuid::now_v7().to_string(),
88            },
89        )
90        .await?;
91    match response {
92        CreateTaskResponse::Accepted(accepted) => {
93            Ok(Some(TaskHandle::new(tasks.clone(), accepted.task.task_id)))
94        }
95        CreateTaskResponse::Skipped(_) => Ok(None),
96        CreateTaskResponse::Unknown { tag, .. } => Err(Error::Protocol(format!(
97            "unknown Task submission outcome `{tag}`"
98        ))),
99    }
100}
101
102impl TaskHandle {
103    pub(crate) fn new(tasks: HttpTaskClient, task_id: Uuid) -> Self {
104        Self { tasks, task_id }
105    }
106
107    /// The Task's stable identifier.
108    pub fn id(&self) -> Uuid {
109        self.task_id
110    }
111
112    /// Read the Task's current status, progress and Attempts.
113    pub async fn details(&self) -> Result<TaskDetails, Error> {
114        Ok(self.tasks.read_task(self.task_id).await?)
115    }
116
117    /// The Task's current status.
118    pub async fn status(&self) -> Result<TaskStatus, Error> {
119        Ok(self.details().await?.task.status)
120    }
121
122    /// The progress its current or latest Attempt last reported, if any.
123    pub async fn progress(&self) -> Result<Option<TaskProgress>, Error> {
124        let details = self.details().await?;
125        let current = details.task.current_attempt_id;
126        let attempt = details
127            .attempts
128            .into_iter()
129            .filter(|attempt| current.is_none_or(|current| attempt.attempt_id == current))
130            .max_by_key(|attempt| attempt.ordinal);
131        Ok(attempt.and_then(|attempt| attempt.progress))
132    }
133
134    /// Request cancellation; the running Attempt stops at its supervisor's next check.
135    pub async fn cancel(&self) -> Result<Task, Error> {
136        Ok(self.tasks.cancel_task(self.task_id).await?)
137    }
138
139    /// Wait for the Task to finish and return the versions it published.
140    pub async fn wait(&self) -> Result<TaskResult, Error> {
141        self.wait_with(TaskWaitOptions::default()).await
142    }
143
144    /// Wait with an explicit polling interval and client deadline.
145    pub async fn wait_with(&self, options: TaskWaitOptions) -> Result<TaskResult, Error> {
146        let task = self
147            .tasks
148            .wait(self.task_id, options)
149            .await
150            .map_err(|error| match error {
151                TaskWaitError::Timeout { task_id } => Error::TaskWaitTimeout { task_id },
152                TaskWaitError::Service(error) => error.into(),
153            })?;
154        match task.status {
155            TaskStatus::Succeeded => task.published.ok_or_else(|| {
156                Error::Protocol("the succeeded Task returned no published versions".to_owned())
157            }),
158            TaskStatus::Cancelled => Err(Error::TaskCancelled {
159                task_id: self.task_id,
160            }),
161            _ => {
162                let failure = task.failure;
163                Err(Error::TaskFailed {
164                    task_id: self.task_id,
165                    code: failure.as_ref().map(|failure| failure.code.clone()),
166                    message: failure.map(|failure| failure.message).unwrap_or_default(),
167                })
168            }
169        }
170    }
171}
spiral/task_handle.rs

spiral/
task_handle.rs
