aster_forge_tasks/
spec.rs

1//! Typed task specification adapters.
2
3use std::future::Future;
4use std::pin::Pin;
5
6use serde::{Serialize, de::DeserializeOwned};
7
8use crate::{Result, TaskCoreError, TaskRecord, TaskRetryClass, TaskStepSpec};
9
10/// Boxed future returned by task processors.
11pub type TaskProcessFuture<'a, Error> =
12    Pin<Box<dyn Future<Output = std::result::Result<(), Error>> + Send + 'a>>;
13
14/// Product-owned typed task specification.
15///
16/// The generic parameters keep Forge independent from product state, persisted task model, runtime
17/// config, execution context, error type, task kind enum, lane enum, and payload/result wrapper
18/// enums. Product crates implement this trait for each task kind and register those specs with
19/// [`crate::task_registry!`].
20pub trait BackgroundTaskSpec<State, Task, Config, Context, Error>: Sync
21where
22    Task: TaskRecord<Self::Kind>,
23{
24    /// Product-owned task kind enum.
25    type Kind: Copy + Eq + std::fmt::Debug + std::fmt::Display + Send + Sync + 'static;
26    /// Product-owned task lane enum.
27    type Lane: Copy + Eq + std::fmt::Debug + Send + Sync + 'static;
28    /// Typed task payload.
29    type Payload: Serialize + DeserializeOwned + Clone + Send + Sync + 'static;
30    /// Typed task result.
31    type Result: Serialize + DeserializeOwned + Clone + Send + Sync + 'static;
32    /// Product task payload envelope enum.
33    type PayloadEnvelope;
34    /// Product task result envelope enum.
35    type ResultEnvelope;
36
37    /// Task kind handled by this spec.
38    const KIND: Self::Kind;
39
40    /// Initial step specs for this task kind.
41    fn step_specs() -> &'static [TaskStepSpec];
42
43    /// Dispatch lane used by this task kind.
44    fn lane() -> Self::Lane;
45
46    /// Maximum attempts for new tasks of this kind.
47    fn max_attempts(_runtime_config: &Config) -> i32 {
48        1
49    }
50
51    /// Wraps the typed payload into the product payload envelope.
52    fn wrap_payload(payload: Self::Payload) -> Self::PayloadEnvelope;
53
54    /// Wraps the typed result into the product result envelope.
55    fn wrap_result(result: Self::Result) -> Self::ResultEnvelope;
56
57    /// Processes the task.
58    fn process<'a>(
59        state: &'a State,
60        task: &'a Task,
61        context: Context,
62    ) -> TaskProcessFuture<'a, Error>;
63
64    /// Classifies a task failure for retry behavior.
65    fn retry_class(_error: &Error) -> TaskRetryClass {
66        TaskRetryClass::Manual
67    }
68}
69
70/// Serializes a typed task payload.
71///
72/// # Errors
73///
74/// Returns [`TaskCoreError`] when the typed payload cannot be serialized as JSON.
75pub fn serialize_payload<S, State, Task, Config, Context, Error>(
76    payload: &S::Payload,
77) -> Result<String>
78where
79    S: BackgroundTaskSpec<State, Task, Config, Context, Error>,
80    Task: TaskRecord<S::Kind>,
81{
82    serde_json::to_string(payload).map_err(|error| {
83        TaskCoreError::codec(format!("serialize {} task payload: {error}", S::KIND))
84    })
85}
86
87/// Serializes a typed task result.
88///
89/// # Errors
90///
91/// Returns [`TaskCoreError`] when the typed result cannot be serialized as JSON.
92pub fn serialize_result<S, State, Task, Config, Context, Error>(
93    result: &S::Result,
94) -> Result<String>
95where
96    S: BackgroundTaskSpec<State, Task, Config, Context, Error>,
97    Task: TaskRecord<S::Kind>,
98{
99    serde_json::to_string(result).map_err(|error| {
100        TaskCoreError::codec(format!("serialize {} task result: {error}", S::KIND))
101    })
102}
103
104/// Decodes a task payload as the typed payload for `S`.
105///
106/// # Errors
107///
108/// Returns [`TaskCoreError`] when the stored payload is absent or cannot be decoded.
109pub fn decode_payload_as<S, State, Task, Config, Context, Error>(task: &Task) -> Result<S::Payload>
110where
111    S: BackgroundTaskSpec<State, Task, Config, Context, Error>,
112    Task: TaskRecord<S::Kind>,
113{
114    if task.kind() != S::KIND {
115        return Err(TaskCoreError::invalid_value(format!(
116            "task #{} kind mismatch: expected {}, got {}",
117            task.id(),
118            S::KIND,
119            task.kind()
120        )));
121    }
122
123    serde_json::from_str(task.payload_json()).map_err(|error| {
124        TaskCoreError::codec(format!(
125            "parse payload for task #{} ({}): {error}",
126            task.id(),
127            task.kind()
128        ))
129    })
130}
131
132/// Decodes a task result as the typed result for `S`.
133///
134/// # Errors
135///
136/// Returns [`TaskCoreError`] when a stored result cannot be decoded.
137pub fn decode_result_as<S, State, Task, Config, Context, Error>(
138    task: &Task,
139) -> Result<Option<S::Result>>
140where
141    S: BackgroundTaskSpec<State, Task, Config, Context, Error>,
142    Task: TaskRecord<S::Kind>,
143{
144    if task.kind() != S::KIND {
145        return Err(TaskCoreError::invalid_value(format!(
146            "task #{} kind mismatch: expected {}, got {}",
147            task.id(),
148            S::KIND,
149            task.kind()
150        )));
151    }
152
153    let Some(raw) = task.result_json() else {
154        return Ok(None);
155    };
156
157    serde_json::from_str(raw).map(Some).map_err(|error| {
158        TaskCoreError::codec(format!(
159            "parse result for task #{} ({}): {error}",
160            task.id(),
161            task.kind()
162        ))
163    })
164}
165
166/// Object-safe task spec used by registries and dispatchers.
167pub trait ErasedBackgroundTaskSpec<
168    State,
169    Task,
170    Config,
171    Context,
172    Kind,
173    Lane,
174    PayloadEnvelope,
175    ResultEnvelope,
176    Error,
177>: Sync where
178    Task: TaskRecord<Kind>,
179    Kind: Copy + Eq + std::fmt::Debug + std::fmt::Display + Send + Sync + 'static,
180    Lane: Copy + Eq + std::fmt::Debug + Send + Sync + 'static,
181{
182    /// Initial step specs for this task kind.
183    fn step_specs(&self) -> &'static [TaskStepSpec];
184
185    /// Dispatch lane used by this task kind.
186    fn lane(&self) -> Lane;
187
188    /// Maximum attempts for new tasks of this kind.
189    fn max_attempts(&self, runtime_config: &Config) -> i32;
190
191    /// Decodes the product task payload envelope.
192    ///
193    /// # Errors
194    ///
195    /// Returns [`TaskCoreError`] when the erased task payload cannot be decoded.
196    fn decode_payload(&self, task: &Task) -> Result<PayloadEnvelope>;
197
198    /// Decodes the product task result envelope.
199    ///
200    /// # Errors
201    ///
202    /// Returns [`TaskCoreError`] when the erased task result cannot be decoded.
203    fn decode_result(&self, task: &Task) -> Result<Option<ResultEnvelope>>;
204
205    /// Classifies a task failure for retry behavior.
206    fn retry_class(&self, error: &Error) -> TaskRetryClass;
207
208    /// Processes the task.
209    fn process<'a>(
210        &self,
211        state: &'a State,
212        task: &'a Task,
213        context: Context,
214    ) -> TaskProcessFuture<'a, Error>;
215}
216
217/// Zero-sized adapter from typed task specs to object-safe task specs.
218pub struct TaskSpecAdapter<S>(std::marker::PhantomData<S>);
219
220impl<S> TaskSpecAdapter<S> {
221    /// Creates a task spec adapter.
222    #[must_use]
223    pub const fn new() -> Self {
224        Self(std::marker::PhantomData)
225    }
226}
227
228impl<S> Default for TaskSpecAdapter<S> {
229    fn default() -> Self {
230        Self::new()
231    }
232}
233
234impl<S, State, Task, Config, Context, Kind, Lane, PayloadEnvelope, ResultEnvelope, Error>
235    ErasedBackgroundTaskSpec<
236        State,
237        Task,
238        Config,
239        Context,
240        Kind,
241        Lane,
242        PayloadEnvelope,
243        ResultEnvelope,
244        Error,
245    > for TaskSpecAdapter<S>
246where
247    S: BackgroundTaskSpec<
248            State,
249            Task,
250            Config,
251            Context,
252            Error,
253            Kind = Kind,
254            Lane = Lane,
255            PayloadEnvelope = PayloadEnvelope,
256            ResultEnvelope = ResultEnvelope,
257        > + Sync,
258    Task: TaskRecord<Kind>,
259    Kind: Copy + Eq + std::fmt::Debug + std::fmt::Display + Send + Sync + 'static,
260    Lane: Copy + Eq + std::fmt::Debug + Send + Sync + 'static,
261{
262    fn step_specs(&self) -> &'static [TaskStepSpec] {
263        S::step_specs()
264    }
265
266    fn lane(&self) -> Lane {
267        S::lane()
268    }
269
270    fn max_attempts(&self, runtime_config: &Config) -> i32 {
271        S::max_attempts(runtime_config)
272    }
273
274    fn decode_payload(&self, task: &Task) -> Result<PayloadEnvelope> {
275        Ok(S::wrap_payload(decode_payload_as::<
276            S,
277            State,
278            Task,
279            Config,
280            Context,
281            Error,
282        >(task)?))
283    }
284
285    fn decode_result(&self, task: &Task) -> Result<Option<ResultEnvelope>> {
286        Ok(decode_result_as::<S, State, Task, Config, Context, Error>(task)?.map(S::wrap_result))
287    }
288
289    fn retry_class(&self, error: &Error) -> TaskRetryClass {
290        S::retry_class(error)
291    }
292
293    fn process<'a>(
294        &self,
295        state: &'a State,
296        task: &'a Task,
297        context: Context,
298    ) -> TaskProcessFuture<'a, Error> {
299        S::process(state, task, context)
300    }
301}
aster_forge_tasks/spec.rs

aster_forge_tasks/
spec.rs
