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.
71pub fn serialize_payload<S, State, Task, Config, Context, Error>(
72    payload: &S::Payload,
73) -> Result<String>
74where
75    S: BackgroundTaskSpec<State, Task, Config, Context, Error>,
76    Task: TaskRecord<S::Kind>,
77{
78    serde_json::to_string(payload).map_err(|error| {
79        TaskCoreError::codec(format!("serialize {} task payload: {error}", S::KIND))
80    })
81}
82
83/// Serializes a typed task result.
84pub fn serialize_result<S, State, Task, Config, Context, Error>(
85    result: &S::Result,
86) -> Result<String>
87where
88    S: BackgroundTaskSpec<State, Task, Config, Context, Error>,
89    Task: TaskRecord<S::Kind>,
90{
91    serde_json::to_string(result).map_err(|error| {
92        TaskCoreError::codec(format!("serialize {} task result: {error}", S::KIND))
93    })
94}
95
96/// Decodes a task payload as the typed payload for `S`.
97pub fn decode_payload_as<S, State, Task, Config, Context, Error>(task: &Task) -> Result<S::Payload>
98where
99    S: BackgroundTaskSpec<State, Task, Config, Context, Error>,
100    Task: TaskRecord<S::Kind>,
101{
102    if task.kind() != S::KIND {
103        return Err(TaskCoreError::invalid_value(format!(
104            "task #{} kind mismatch: expected {}, got {}",
105            task.id(),
106            S::KIND,
107            task.kind()
108        )));
109    }
110
111    serde_json::from_str(task.payload_json()).map_err(|error| {
112        TaskCoreError::codec(format!(
113            "parse payload for task #{} ({}): {error}",
114            task.id(),
115            task.kind()
116        ))
117    })
118}
119
120/// Decodes a task result as the typed result for `S`.
121pub fn decode_result_as<S, State, Task, Config, Context, Error>(
122    task: &Task,
123) -> Result<Option<S::Result>>
124where
125    S: BackgroundTaskSpec<State, Task, Config, Context, Error>,
126    Task: TaskRecord<S::Kind>,
127{
128    if task.kind() != S::KIND {
129        return Err(TaskCoreError::invalid_value(format!(
130            "task #{} kind mismatch: expected {}, got {}",
131            task.id(),
132            S::KIND,
133            task.kind()
134        )));
135    }
136
137    let Some(raw) = task.result_json() else {
138        return Ok(None);
139    };
140
141    serde_json::from_str(raw).map(Some).map_err(|error| {
142        TaskCoreError::codec(format!(
143            "parse result for task #{} ({}): {error}",
144            task.id(),
145            task.kind()
146        ))
147    })
148}
149
150/// Object-safe task spec used by registries and dispatchers.
151pub trait ErasedBackgroundTaskSpec<
152    State,
153    Task,
154    Config,
155    Context,
156    Kind,
157    Lane,
158    PayloadEnvelope,
159    ResultEnvelope,
160    Error,
161>: Sync where
162    Task: TaskRecord<Kind>,
163    Kind: Copy + Eq + std::fmt::Debug + std::fmt::Display + Send + Sync + 'static,
164    Lane: Copy + Eq + std::fmt::Debug + Send + Sync + 'static,
165{
166    /// Initial step specs for this task kind.
167    fn step_specs(&self) -> &'static [TaskStepSpec];
168
169    /// Dispatch lane used by this task kind.
170    fn lane(&self) -> Lane;
171
172    /// Maximum attempts for new tasks of this kind.
173    fn max_attempts(&self, runtime_config: &Config) -> i32;
174
175    /// Decodes the product task payload envelope.
176    fn decode_payload(&self, task: &Task) -> Result<PayloadEnvelope>;
177
178    /// Decodes the product task result envelope.
179    fn decode_result(&self, task: &Task) -> Result<Option<ResultEnvelope>>;
180
181    /// Classifies a task failure for retry behavior.
182    fn retry_class(&self, error: &Error) -> TaskRetryClass;
183
184    /// Processes the task.
185    fn process<'a>(
186        &self,
187        state: &'a State,
188        task: &'a Task,
189        context: Context,
190    ) -> TaskProcessFuture<'a, Error>;
191}
192
193/// Zero-sized adapter from typed task specs to object-safe task specs.
194pub struct TaskSpecAdapter<S>(std::marker::PhantomData<S>);
195
196impl<S> TaskSpecAdapter<S> {
197    /// Creates a task spec adapter.
198    pub const fn new() -> Self {
199        Self(std::marker::PhantomData)
200    }
201}
202
203impl<S> Default for TaskSpecAdapter<S> {
204    fn default() -> Self {
205        Self::new()
206    }
207}
208
209impl<S, State, Task, Config, Context, Kind, Lane, PayloadEnvelope, ResultEnvelope, Error>
210    ErasedBackgroundTaskSpec<
211        State,
212        Task,
213        Config,
214        Context,
215        Kind,
216        Lane,
217        PayloadEnvelope,
218        ResultEnvelope,
219        Error,
220    > for TaskSpecAdapter<S>
221where
222    S: BackgroundTaskSpec<
223            State,
224            Task,
225            Config,
226            Context,
227            Error,
228            Kind = Kind,
229            Lane = Lane,
230            PayloadEnvelope = PayloadEnvelope,
231            ResultEnvelope = ResultEnvelope,
232        > + Sync,
233    Task: TaskRecord<Kind>,
234    Kind: Copy + Eq + std::fmt::Debug + std::fmt::Display + Send + Sync + 'static,
235    Lane: Copy + Eq + std::fmt::Debug + Send + Sync + 'static,
236{
237    fn step_specs(&self) -> &'static [TaskStepSpec] {
238        S::step_specs()
239    }
240
241    fn lane(&self) -> Lane {
242        S::lane()
243    }
244
245    fn max_attempts(&self, runtime_config: &Config) -> i32 {
246        S::max_attempts(runtime_config)
247    }
248
249    fn decode_payload(&self, task: &Task) -> Result<PayloadEnvelope> {
250        Ok(S::wrap_payload(decode_payload_as::<
251            S,
252            State,
253            Task,
254            Config,
255            Context,
256            Error,
257        >(task)?))
258    }
259
260    fn decode_result(&self, task: &Task) -> Result<Option<ResultEnvelope>> {
261        Ok(decode_result_as::<S, State, Task, Config, Context, Error>(task)?.map(S::wrap_result))
262    }
263
264    fn retry_class(&self, error: &Error) -> TaskRetryClass {
265        S::retry_class(error)
266    }
267
268    fn process<'a>(
269        &self,
270        state: &'a State,
271        task: &'a Task,
272        context: Context,
273    ) -> TaskProcessFuture<'a, Error> {
274        S::process(state, task, context)
275    }
276}
aster_forge_tasks/spec.rs

aster_forge_tasks/
spec.rs
