ledgence_orchestration_api/
workflow_events.rs

1//! Directly addressed, one-shot workflow events and persisted wait deadlines.
2use crate::*;
3use ledgence_worker_api::{validate_json_cloudevent, validate_wire_value};
4use serde_json::Value;
5
6pub const WORKFLOW_EVENT_MAX_BYTES: usize = 64 * 1024;
7pub const WORKFLOW_EVENT_COMMAND_MAX_BYTES: usize = 70 * 1024;
8pub const WORKFLOW_MAX_PENDING_EVENTS: usize = 128;
9pub const WORKFLOW_PENDING_EVENTS_MAX_BYTES: usize = 256 * 1024;
10/// Relative waits are bounded to 365 days. Zero means immediately eligible.
11pub const WORKFLOW_MAX_DELAY_MS: u64 = 365 * 24 * 60 * 60 * 1_000;
12const MAX_TIMESTAMP: u64 = 253_402_300_799_999;
13
14/// The sender's original JSON CloudEvent. This is an external event profile,
15/// without required Ledgence execution identifiers. Routing authority comes
16/// from the command's scope/workflow/key, never from event extension attributes.
17#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
18#[serde(transparent)]
19pub struct WorkflowEvent(Value);
20impl WorkflowEvent {
21    pub fn new(value: Value) -> Result<Self> {
22        let event = Self(value);
23        event.validate()?;
24        Ok(event)
25    }
26    pub fn value(&self) -> &Value {
27        &self.0
28    }
29    pub fn id(&self) -> &str {
30        self.0.get("id").and_then(Value::as_str).unwrap_or_default()
31    }
32    pub fn source(&self) -> &str {
33        self.0
34            .get("source")
35            .and_then(Value::as_str)
36            .unwrap_or_default()
37    }
38    pub fn validate(&self) -> Result<()> {
39        bounded(&self.0, WORKFLOW_EVENT_MAX_BYTES, "workflow event")?;
40        validate_json_cloudevent(&self.0)?;
41        validate_text(self.id(), 128)?;
42        validate_text(self.source(), 2048)?;
43        validate_wire_value(&self.0["data"])?;
44        Ok(())
45    }
46    pub fn trace_context(&self) -> Option<TraceContext> {
47        self.0
48            .get("traceparent")
49            .and_then(Value::as_str)
50            .map(|traceparent| TraceContext {
51                traceparent: traceparent.to_owned(),
52                tracestate: self
53                    .0
54                    .get("tracestate")
55                    .and_then(Value::as_str)
56                    .map(str::to_owned),
57            })
58    }
59}
60
61#[derive(Debug, Clone, Serialize, Deserialize)]
62#[serde(deny_unknown_fields)]
63pub struct WorkflowEventCommand {
64    pub scope: Scope,
65    pub workflow_id: String,
66    pub key: String,
67    pub event: WorkflowEvent,
68}
69impl WorkflowEventCommand {
70    pub fn validate(&self) -> Result<()> {
71        self.scope.validate()?;
72        validate_text(&self.workflow_id, 128)?;
73        validate_text(&self.key, 128)?;
74        self.event.validate()?;
75        bounded(
76            self,
77            WORKFLOW_EVENT_COMMAND_MAX_BYTES,
78            "workflow event command",
79        )
80    }
81}
82
83/// Durable acceptance, not a promise that a controller has processed the event.
84/// `accepted_at` is immutable across reconciliation and comes from store time.
85#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
86#[serde(deny_unknown_fields)]
87pub struct WorkflowEventReceipt {
88    pub scope: Scope,
89    pub workflow_id: String,
90    pub key: String,
91    pub event_id: String,
92    pub event_source: String,
93    pub accepted_at: Timestamp,
94    pub already_accepted: bool,
95}
96impl WorkflowEventReceipt {
97    pub fn validate(&self) -> Result<()> {
98        self.scope.validate()?;
99        validate_text(&self.workflow_id, 128)?;
100        validate_text(&self.key, 128)?;
101        validate_text(&self.event_id, 128)?;
102        validate_text(&self.event_source, 2048)?;
103        timestamp(self.accepted_at)
104    }
105    pub fn matches(&self, command: &WorkflowEventCommand) -> bool {
106        self.scope == command.scope
107            && self.workflow_id == command.workflow_id
108            && self.key == command.key
109            && self.event_id == command.event.id()
110            && self.event_source == command.event.source()
111    }
112}
113
114/// A single named rendezvous. Keys are one-shot across a workflow, so callbacks
115/// from an earlier iteration cannot accidentally satisfy a later wait.
116#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
117#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
118pub enum WorkflowWait {
119    Event {
120        key: String,
121        #[serde(deserialize_with = "crate::observation::required_option")]
122        timeout_ms: Option<u64>,
123    },
124    Timer {
125        key: String,
126        delay_ms: u64,
127    },
128}
129impl WorkflowWait {
130    pub fn key(&self) -> &str {
131        match self {
132            Self::Event { key, .. } | Self::Timer { key, .. } => key,
133        }
134    }
135    pub fn validate(&self) -> Result<()> {
136        validate_text(self.key(), 128)?;
137        let duration = match self {
138            Self::Event { timeout_ms, .. } => *timeout_ms,
139            Self::Timer { delay_ms, .. } => Some(*delay_ms),
140        };
141        if duration.is_some_and(|duration| duration > WORKFLOW_MAX_DELAY_MS) {
142            return Err(invalid("workflow wait duration exceeds 365 days"));
143        }
144        Ok(())
145    }
146    /// Anchor exactly once when accepting the decision. Scheduler retries must
147    /// read the persisted deadline rather than call this with a newer time.
148    pub fn deadline(&self, accepted_at: Timestamp) -> Result<Option<Timestamp>> {
149        self.validate()?;
150        timestamp(accepted_at)?;
151        let duration = match self {
152            Self::Event { timeout_ms, .. } => *timeout_ms,
153            Self::Timer { delay_ms, .. } => Some(*delay_ms),
154        };
155        duration
156            .map(|duration| {
157                let deadline = accepted_at
158                    .checked_add(duration)
159                    .ok_or_else(|| invalid("workflow deadline overflow"))?;
160                timestamp(deadline)?;
161                Ok(deadline)
162            })
163            .transpose()
164    }
165}
166
167/// Immutable next-activation input selected under workflow authority. At an
168/// event deadline, only store acceptance strictly before the deadline wins.
169#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
170#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
171pub enum WorkflowWake {
172    Event {
173        key: String,
174        event: WorkflowEvent,
175        accepted_at: Timestamp,
176    },
177    Timeout {
178        key: String,
179        deadline: Timestamp,
180    },
181    Timer {
182        key: String,
183        deadline: Timestamp,
184    },
185}
186impl WorkflowWake {
187    pub fn key(&self) -> &str {
188        match self {
189            Self::Event { key, .. } | Self::Timeout { key, .. } | Self::Timer { key, .. } => key,
190        }
191    }
192    pub fn validate(&self) -> Result<()> {
193        validate_text(self.key(), 128)?;
194        match self {
195            Self::Event {
196                event, accepted_at, ..
197            } => {
198                event.validate()?;
199                timestamp(*accepted_at)
200            }
201            Self::Timeout { deadline, .. } | Self::Timer { deadline, .. } => timestamp(*deadline),
202        }
203    }
204}
205
206fn timestamp(at: Timestamp) -> Result<()> {
207    if at > MAX_TIMESTAMP {
208        Err(invalid("workflow timestamp exceeds supported range"))
209    } else {
210        Ok(())
211    }
212}
213fn invalid(message: &str) -> ContractError {
214    ContractError::InvalidInput(message.into())
215}
216fn bounded(value: &impl Serialize, bytes: usize, label: &str) -> Result<()> {
217    crate::submission::check_encoded_size(value, bytes, label).map_err(Into::into)
218}
219
220#[cfg(test)]
221mod tests {
222    use super::*;
223    use serde_json::json;
224    fn event() -> Value {
225        json!({"specversion":"1.0","id":"evt_1","source":"urn:billing","type":"invoice.paid","datacontenttype":"application/json","data":{"invoice_id":"INV-1042"}})
226    }
227    #[test]
228    fn external_event_preserves_json_without_fabricated_invocation_ids() {
229        let mut value = event();
230        value["traceparent"] = json!("00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01");
231        value["businessid"] = json!("INV-1042");
232        let event = WorkflowEvent::new(value.clone()).unwrap();
233        assert_eq!(event.value(), &value);
234        assert!(event.trace_context().is_some());
235        assert!(ledgence_worker_api::CloudEvent::new(value).is_err());
236    }
237    #[test]
238    fn event_profile_rejects_bad_context_and_invalid_payload() {
239        for (key, bad) in [
240            ("specversion", json!("2.0")),
241            ("source", json!("has a space")),
242            ("id", json!("")),
243            ("datacontenttype", json!("text/plain")),
244            ("traceparent", json!("invalid")),
245            ("time", json!("yesterday")),
246            ("subject", Value::Null),
247            ("InvalidName", json!("x")),
248            ("dataschema", json!("relative")),
249            ("tracestate", json!("a=b")),
250        ] {
251            let mut value = event();
252            value[key] = bad;
253            assert!(WorkflowEvent::new(value).is_err(), "{key}");
254        }
255        let mut missing = event();
256        missing.as_object_mut().unwrap().remove("data");
257        assert!(WorkflowEvent::new(missing).is_err());
258        let mut large = event();
259        large["data"] = json!("x".repeat(WORKFLOW_EVENT_MAX_BYTES));
260        assert!(WorkflowEvent::new(large).is_err());
261        let mut deep = Value::Null;
262        for _ in 0..65 {
263            deep = json!([deep]);
264        }
265        let mut value = event();
266        value["data"] = deep;
267        assert!(WorkflowEvent::new(value).is_err());
268    }
269    #[test]
270    fn wait_shape_duration_and_timestamp_boundaries_are_strict() {
271        let zero = WorkflowWait::Timer {
272            key: "sleep".into(),
273            delay_ms: 0,
274        };
275        assert_eq!(zero.deadline(42).unwrap(), Some(42));
276        let maximum = WorkflowWait::Event {
277            key: "approval".into(),
278            timeout_ms: Some(WORKFLOW_MAX_DELAY_MS),
279        };
280        assert_eq!(
281            maximum.deadline(42).unwrap(),
282            Some(42 + WORKFLOW_MAX_DELAY_MS)
283        );
284        assert!(maximum.deadline(MAX_TIMESTAMP).is_err());
285        assert!(serde_json::from_value::<WorkflowWait>(json!({"kind":"event","key":"a"})).is_err());
286        for value in [
287            json!({"kind":"timer","key":"a","delay_ms":true}),
288            json!({"kind":"timer","key":"a","delay_ms":-1}),
289            json!({"kind":"event","key":"a","timeout_ms":null,"extra":1}),
290        ] {
291            assert!(serde_json::from_value::<WorkflowWait>(value).is_err());
292        }
293        assert!(
294            WorkflowWait::Timer {
295                key: "a".into(),
296                delay_ms: WORKFLOW_MAX_DELAY_MS + 1
297            }
298            .validate()
299            .is_err()
300        );
301        assert_eq!(
302            WorkflowWait::Event {
303                key: "a".into(),
304                timeout_ms: None
305            }
306            .deadline(42)
307            .unwrap(),
308            None
309        );
310    }
311    #[test]
312    fn receipt_identity_binds_scope_workflow_key_and_source_id() {
313        let command = WorkflowEventCommand {
314            scope: Scope {
315                tenant_id: "t".into(),
316                namespace: "n".into(),
317            },
318            workflow_id: "wf_1".into(),
319            key: "approval".into(),
320            event: WorkflowEvent::new(event()).unwrap(),
321        };
322        command.validate().unwrap();
323        let receipt = WorkflowEventReceipt {
324            scope: command.scope.clone(),
325            workflow_id: command.workflow_id.clone(),
326            key: command.key.clone(),
327            event_id: command.event.id().into(),
328            event_source: command.event.source().into(),
329            accepted_at: 42,
330            already_accepted: false,
331        };
332        receipt.validate().unwrap();
333        assert!(receipt.matches(&command));
334        for variant in 0..6 {
335            let mut changed = receipt.clone();
336            match variant {
337                0 => changed.scope.tenant_id.push('x'),
338                1 => changed.scope.namespace.push('x'),
339                2 => changed.workflow_id.push('x'),
340                3 => changed.key.push('x'),
341                4 => changed.event_id.push('x'),
342                _ => changed.event_source.push('x'),
343            };
344            assert!(!changed.matches(&command));
345        }
346    }
347    #[test]
348    fn frozen_wake_shares_input_budget_and_does_not_change_legacy_contexts() {
349        let base = json!({"v":1,"workflow_id":"wf_1","activation_id":"task_1","revision":0,
350            "continuation":"after","state":null,"inputs":{},"local_steps":[]});
351        let mut context: WorkflowActivationContext = serde_json::from_value(base.clone()).unwrap();
352        context.validate().unwrap();
353        assert_eq!(serde_json::to_value(&context).unwrap(), base);
354        context.wake = Some(WorkflowWake::Event {
355            key: "approval".into(),
356            event: WorkflowEvent::new(event()).unwrap(),
357            accepted_at: 42,
358        });
359        context.validate().unwrap();
360        let frozen = serde_json::to_value(&context).unwrap();
361        assert_eq!(
362            frozen["wake"]["event"]["data"]["invoice_id"],
363            json!("INV-1042")
364        );
365        for bad in [
366            json!({"kind":"timeout","key":"a"}),
367            json!({"kind":"timer","key":"a","deadline":42,"extra":true}),
368            json!({"kind":"event","key":"a","event":null,"accepted_at":42}),
369        ] {
370            let mut value = base.clone();
371            value["wake"] = bad;
372            assert!(
373                serde_json::from_value::<WorkflowActivationContext>(value.clone()).is_err()
374                    || serde_json::from_value::<WorkflowActivationContext>(value)
375                        .unwrap()
376                        .validate()
377                        .is_err()
378            );
379        }
380        context.inputs.insert(
381            "large".into(),
382            WorkflowChildResult::Task(WorkflowTaskResult {
383                task_id: "child_1".into(),
384                state: TaskState::Succeeded,
385                outcome: TaskOutcome::Succeeded {
386                    output: json!("x".repeat(WORKFLOW_INPUTS_MAX_BYTES - 512)),
387                    attempt_id: "att_1".into(),
388                    execution_may_have_started: true,
389                    quiescence: Quiescence::Confirmed,
390                },
391            }),
392        );
393        context.wake = None;
394        context.validate().unwrap();
395        let mut value = event();
396        value["data"] = json!("x".repeat(1024));
397        context.wake = Some(WorkflowWake::Event {
398            key: "approval".into(),
399            event: WorkflowEvent::new(value).unwrap(),
400            accepted_at: 42,
401        });
402        assert!(context.validate().is_err());
403    }
404    #[test]
405    fn external_wait_decision_is_explicit_and_preserves_child_wait_semantics() {
406        let value = json!({"v":1,"activation_id":"task_1","revision":0,"kind":"wait",
407            "continuation":"after","state":null,"commands":[],
408            "wait":{"kind":"event","key":"approval","timeout_ms":null}});
409        assert!(WorkflowDecision::decode(&value).is_ok());
410        let mut missing = value.clone();
411        missing.as_object_mut().unwrap().remove("wait");
412        assert!(WorkflowDecision::decode(&missing).is_err());
413        let mut extra = value;
414        extra["until"] = json!([]);
415        assert!(WorkflowDecision::decode(&extra).is_err());
416    }
417    #[test]
418    fn event_timestamps_use_the_same_strict_profile_as_python_wakes() {
419        for at in [
420            "2024-02-29T12:00:00Z",
421            "2024-02-29t12:00:00z",
422            "2024-01-31T23:59:60Z",
423            "2024-02-01T00:59:60+01:00",
424            "0000-01-01T00:00:00Z",
425        ] {
426            let mut value = event();
427            value["time"] = json!(at);
428            assert!(WorkflowEvent::new(value).is_ok(), "{at}");
429        }
430        for at in [
431            "2024-02-29 12:00:00Z",
432            "2024-02-29x12:00:00Z",
433            "2024-01-15T12:00:60Z",
434            "2024-01-31T23:59:60+01:00",
435        ] {
436            let mut value = event();
437            value["time"] = json!(at);
438            assert!(WorkflowEvent::new(value).is_err(), "{at}");
439        }
440    }
441}
ledgence_orchestration_api/workflow_events.rs

ledgence_orchestration_api/
workflow_events.rs
