ledgence_orchestration_api/
observation.rs

1//! Compact, coherent observations of logical tasks, independent of worker reports.
2
3use crate::*;
4use ledgence_worker_api::{Error, Phase};
5use serde::Deserializer;
6use serde_json::Value;
7
8/// Maximum encoded compact status response, excluding HTTP headers.
9pub const TASK_STATUS_MAX_BYTES: usize = 16 * 1024;
10
11/// Scheduling metadata without application input, output, or package payloads.
12#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
13#[serde(deny_unknown_fields)]
14pub struct TaskStatus {
15    pub scope: Scope,
16    pub task_id: String,
17    pub run_id: String,
18    #[serde(default, skip_serializing_if = "Option::is_none")]
19    pub workflow_id: Option<String>,
20    #[serde(default, skip_serializing_if = "Option::is_none")]
21    pub workflow_activation_id: Option<String>,
22    pub queue: String,
23    #[serde(deserialize_with = "required_option")]
24    pub correlation_key: Option<String>,
25    pub state: TaskState,
26    pub attempt_count: u32,
27    #[serde(deserialize_with = "required_option")]
28    pub current_attempt_id: Option<String>,
29    /// Last allocated attempt, never inferred cancellation provenance.
30    #[serde(deserialize_with = "required_option")]
31    pub latest_attempt_id: Option<String>,
32    pub submitted_at: Timestamp,
33    pub available_at: Timestamp,
34    #[serde(deserialize_with = "required_option")]
35    pub terminal_at: Option<Timestamp>,
36    #[serde(deserialize_with = "required_option")]
37    pub cancel_requested_at: Option<Timestamp>,
38}
39
40impl TaskStatus {
41    /// Validate an adapter-produced observation before exposing it to a caller.
42    pub fn validate(&self) -> Result<()> {
43        let invalid = || observation_error("inconsistent task status");
44        self.scope.validate().map_err(|_| invalid())?;
45        for value in [&self.task_id, &self.run_id, &self.queue] {
46            validate_text(value, 128).map_err(|_| invalid())?;
47        }
48        for value in [&self.workflow_id, &self.workflow_activation_id]
49            .into_iter()
50            .flatten()
51        {
52            validate_text(value, 128).map_err(|_| invalid())?;
53        }
54        if let Some(activation) = &self.workflow_activation_id
55            && (self.workflow_id.is_none() || activation != &self.task_id)
56        {
57            return Err(invalid());
58        }
59        if let Some(value) = &self.correlation_key
60            && (value.len() > 512 || value.chars().any(char::is_control))
61        {
62            return Err(invalid());
63        }
64        for value in [&self.current_attempt_id, &self.latest_attempt_id]
65            .into_iter()
66            .flatten()
67        {
68            validate_text(value, 128).map_err(|_| invalid())?;
69        }
70        if self.attempt_count > 1000
71            || (self.attempt_count == 0) != self.latest_attempt_id.is_none()
72            || (self.state == TaskState::Active) != self.current_attempt_id.is_some()
73            || (self.state == TaskState::Active
74                && self.current_attempt_id != self.latest_attempt_id)
75            || self.state.is_terminal() != self.terminal_at.is_some()
76            || (self.state == TaskState::Cancelled && self.cancel_requested_at.is_none())
77            || (self.cancel_requested_at.is_some()
78                && !matches!(self.state, TaskState::Active | TaskState::Cancelled))
79            || (matches!(self.state, TaskState::Succeeded | TaskState::Failed)
80                && self.attempt_count == 0)
81        {
82            return Err(invalid());
83        }
84        // Matches the existing stored/core four-digit RFC3339 timestamp range.
85        if [
86            Some(self.submitted_at),
87            Some(self.available_at),
88            self.terminal_at,
89            self.cancel_requested_at,
90        ]
91        .into_iter()
92        .flatten()
93        .any(|at| at > 253_402_300_799_999)
94        {
95            return Err(invalid());
96        }
97        crate::submission::check_encoded_size(self, TASK_STATUS_MAX_BYTES, "task status")
98            .map_err(|_| invalid())
99    }
100}
101
102/// One coherent task observation. Pending is distinct from successful JSON null.
103#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
104#[serde(deny_unknown_fields)]
105pub struct TaskResult {
106    pub task: TaskStatus,
107    #[serde(deserialize_with = "required_option")]
108    pub outcome: Option<TaskOutcome>,
109}
110
111/// Terminal scheduling outcome. Cancellation never attributes an earlier attempt.
112#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
113#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
114pub enum TaskOutcome {
115    Succeeded {
116        attempt_id: String,
117        quiescence: Quiescence,
118        execution_may_have_started: bool,
119        #[serde(deserialize_with = "required_value")]
120        output: Value,
121    },
122    Failed {
123        attempt_id: String,
124        quiescence: Quiescence,
125        execution_may_have_started: bool,
126        failure: TaskFailure,
127    },
128    Cancelled {},
129}
130
131/// Application error identifiers are user-defined, unlike worker error kinds.
132#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
133#[serde(deny_unknown_fields)]
134pub struct ApplicationError {
135    pub kind: String,
136    pub message: String,
137}
138
139#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
140#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
141pub enum TaskFailure {
142    Application {
143        error: ApplicationError,
144    },
145    Execution {
146        #[serde(deserialize_with = "required_error")]
147        error: Error,
148        phase: Phase,
149        #[serde(deserialize_with = "required_cleanup_error")]
150        cleanup_error: Option<Error>,
151    },
152    AttemptLost {},
153}
154
155impl TaskResult {
156    /// Reject contradictory backend or transport results without inventing state.
157    pub fn validate(&self) -> Result<()> {
158        self.task.validate()?;
159        let invalid = || observation_error("inconsistent task outcome");
160        match (&self.outcome, self.task.state) {
161            (None, TaskState::Queued | TaskState::Active)
162            | (Some(TaskOutcome::Cancelled {}), TaskState::Cancelled) => Ok(()),
163            (
164                Some(TaskOutcome::Succeeded {
165                    attempt_id,
166                    output,
167                    execution_may_have_started,
168                    ..
169                }),
170                TaskState::Succeeded,
171            ) => {
172                if self.task.latest_attempt_id.as_ref() != Some(attempt_id)
173                    || !execution_may_have_started
174                {
175                    return Err(invalid());
176                }
177                validate_task_output(output, self.task.workflow_activation_id.is_some())
178                    .map_err(|_| invalid())
179            }
180            (
181                Some(TaskOutcome::Failed {
182                    attempt_id,
183                    quiescence,
184                    execution_may_have_started,
185                    failure,
186                }),
187                TaskState::Failed,
188            ) => {
189                if self.task.latest_attempt_id.as_ref() != Some(attempt_id)
190                    || (matches!(failure, TaskFailure::AttemptLost {})
191                        && *quiescence != Quiescence::Unconfirmed)
192                    || (matches!(failure, TaskFailure::Application { .. })
193                        && !execution_may_have_started)
194                {
195                    return Err(invalid());
196                }
197                Ok(())
198            }
199            _ => Err(invalid()),
200        }?;
201        if let Some(outcome) = &self.outcome {
202            // Projection removes report/owner context, so every valid compact
203            // outcome fits within its original 8 MiB settlement budget.
204            crate::submission::check_encoded_size(outcome, SETTLEMENT_MAX_BYTES, "task outcome")
205                .map_err(|_| invalid())?;
206        }
207        Ok(())
208    }
209}
210
211pub(crate) fn required_option<'de, D, T>(
212    deserializer: D,
213) -> std::result::Result<Option<T>, D::Error>
214where
215    D: Deserializer<'de>,
216    T: Deserialize<'de>,
217{
218    Option::<T>::deserialize(deserializer)
219}
220
221fn observation_error(message: &str) -> ContractError {
222    ContractError::Unavailable(message.into())
223}
224
225pub(crate) fn required_value<'de, D: Deserializer<'de>>(
226    deserializer: D,
227) -> std::result::Result<Value, D::Error> {
228    Value::deserialize(deserializer)
229}
230
231#[cfg(test)]
232#[path = "observation_tests.rs"]
233mod tests;
234
235// Keep the existing portable worker Error type without inheriting its older,
236// permissive unknown-field response decoding in this additive strict contract.
237#[derive(Deserialize)]
238#[serde(deny_unknown_fields)]
239struct StrictError {
240    kind: ledgence_worker_api::ErrorKind,
241    message: String,
242}
243impl From<StrictError> for Error {
244    fn from(value: StrictError) -> Self {
245        Self {
246            kind: value.kind,
247            message: value.message,
248        }
249    }
250}
251fn required_error<'de, D: Deserializer<'de>>(
252    deserializer: D,
253) -> std::result::Result<Error, D::Error> {
254    StrictError::deserialize(deserializer).map(Into::into)
255}
256fn required_cleanup_error<'de, D: Deserializer<'de>>(
257    deserializer: D,
258) -> std::result::Result<Option<Error>, D::Error> {
259    Option::<StrictError>::deserialize(deserializer).map(|value| value.map(Into::into))
260}
ledgence_orchestration_api/observation.rs

ledgence_orchestration_api/
observation.rs
