1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
//! What a step node returns. Port of `platform/types.py` Pass / Drop / FanOut.
use chrono::{DateTime, Utc};
use serde_json::Value;
use crate::{event::Event, ResourceReservationRef};
/// Product-owned action data emitted by an `execute.*` step. The durable
/// driver adds tenant, run, lease, control and exact capability pins.
#[derive(Debug, Clone, PartialEq)]
pub struct PreparedAction {
/// Capability id the intent will pin.
pub capability_id: String,
/// Product idempotency key; the driver scopes it to the instance and state version.
pub idempotency_key: String,
/// Structured input handed to the provider.
pub input: Value,
/// Resource scope the action competes in (empty when unscoped).
pub resource_scope_id: String,
/// Fenced reservation for funds actions.
pub reservation: Option<ResourceReservationRef>,
/// Latest time by which the action must finish.
pub deadline: Option<DateTime<Utc>>,
}
impl PreparedAction {
/// Unscoped action with no reservation or deadline.
pub fn new(
capability_id: impl Into<String>,
idempotency_key: impl Into<String>,
input: Value,
) -> Self {
Self {
capability_id: capability_id.into(),
idempotency_key: idempotency_key.into(),
input,
resource_scope_id: String::new(),
reservation: None,
deadline: None,
}
}
/// Reject blank capability or idempotency identifiers.
pub fn validate(&self) -> Result<(), String> {
if self.capability_id.trim().is_empty() || self.idempotency_key.trim().is_empty() {
return Err("prepared action capability_id and idempotency_key are required".into());
}
Ok(())
}
}
/// The outcome of running one event through one [`crate::node::StepNode`].
#[derive(Debug, Clone)]
pub enum StepResult {
/// Forward the (possibly derived) event to the next step.
Pass(Event),
/// Stop propagation here. Filters and sink terminals return this.
Drop {
/// Why the event stopped.
reason: String,
/// Stable machine-readable code for material business outcomes. A
/// routine filter drop leaves this empty.
exit_reason: Option<String>,
},
/// Forward several events. Only nodes that declare `produces_fan_out`
/// may return this; the validator bans fan-out upstream of `execute.*`.
FanOut(Vec<Event>),
/// Persist an external action intent and continue with the supplied event.
/// The step never invokes the provider inline.
Action {
/// Event forwarded to the next step.
event: Event,
/// The effect to persist as an intent.
action: Box<PreparedAction>,
},
}
impl StepResult {
/// A routine drop that recorders do not persist as a step.
pub fn drop(reason: impl Into<String>) -> Self {
StepResult::Drop {
reason: reason.into(),
exit_reason: None,
}
}
/// A material gate/drop that recorders should persist as a run step.
pub fn gate(exit_reason: impl Into<String>) -> Self {
let code = exit_reason.into();
StepResult::Drop {
reason: code.clone(),
exit_reason: Some(code),
}
}
}