a3s_flow/model/
signal.rs

1use chrono::{DateTime, Utc};
2use serde::de::DeserializeOwned;
3use serde::{Deserialize, Serialize};
4
5use crate::error::{FlowError, Result};
6
7use super::JsonValue;
8
9/// One durable asynchronous message delivered to a workflow execution.
10///
11/// `signal_id` is the caller-owned idempotency identity. A caller must reuse
12/// both the target run ID and this value when retrying an uncertain delivery.
13#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
14#[non_exhaustive]
15pub struct WorkflowSignal {
16    /// Caller-owned idempotency identity.
17    pub signal_id: String,
18    /// Declared signal contract name.
19    pub name: String,
20    /// Application-defined JSON payload.
21    pub payload: JsonValue,
22}
23
24impl WorkflowSignal {
25    /// Creates a signal delivery with a stable caller-owned identity.
26    pub fn new(signal_id: impl Into<String>, name: impl Into<String>, payload: JsonValue) -> Self {
27        Self {
28            signal_id: signal_id.into(),
29            name: name.into(),
30            payload,
31        }
32    }
33
34    pub(crate) fn validate(&self) -> Result<()> {
35        if self.signal_id.trim().is_empty() {
36            return Err(FlowError::InvalidTransition(
37                "workflow signal id must not be empty".to_string(),
38            ));
39        }
40        validate_signal_name(&self.name)
41    }
42}
43
44/// A received signal together with its durable history position.
45#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
46#[non_exhaustive]
47pub struct WorkflowSignalSnapshot {
48    /// Caller-owned idempotency identity.
49    pub signal_id: String,
50    /// Declared signal contract name.
51    pub name: String,
52    /// Application-defined JSON payload.
53    pub payload: JsonValue,
54    /// UTC time at which the signal was persisted.
55    pub received_at: DateTime<Utc>,
56    /// Event sequence that recorded the delivery.
57    pub received_sequence: u64,
58    /// Stable wait identity that consumed the signal.
59    #[serde(default, skip_serializing_if = "Option::is_none")]
60    pub consumed_by: Option<String>,
61}
62
63impl WorkflowSignalSnapshot {
64    /// Decode the persisted signal payload into a host-defined serde type.
65    pub fn payload_as<T>(&self) -> Result<T>
66    where
67        T: DeserializeOwned,
68    {
69        serde_json::from_value(self.payload.clone()).map_err(FlowError::from)
70    }
71}
72
73/// Materialized lifecycle state of a deterministic signal wait.
74#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
75#[non_exhaustive]
76#[serde(rename_all = "snake_case")]
77pub enum SignalWaitStatus {
78    /// The wait has not consumed a signal.
79    Waiting,
80    /// The wait was paired with one durable signal.
81    Completed,
82    /// The owning run cancelled the wait.
83    Cancelled,
84}
85
86/// A deterministic workflow wait for the next unconsumed signal of one name.
87#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
88#[non_exhaustive]
89pub struct SignalWaitSnapshot {
90    /// Replay-stable identity of the wait.
91    pub wait_id: String,
92    /// Signal contract accepted by the wait.
93    pub signal_name: String,
94    /// Current lifecycle state.
95    pub status: SignalWaitStatus,
96    /// UTC time at which the wait was created.
97    pub created_at: DateTime<Utc>,
98    /// Event sequence that created the wait.
99    pub created_sequence: u64,
100    /// Signal paired with the wait, when completed.
101    #[serde(default, skip_serializing_if = "Option::is_none")]
102    pub signal_id: Option<String>,
103    /// UTC time at which the signal was paired.
104    #[serde(default, skip_serializing_if = "Option::is_none")]
105    pub completed_at: Option<DateTime<Utc>>,
106    /// Event sequence that completed the wait.
107    #[serde(default, skip_serializing_if = "Option::is_none")]
108    pub completed_sequence: Option<u64>,
109}
110
111pub(crate) fn validate_signal_name(name: &str) -> Result<()> {
112    if name.trim().is_empty() {
113        return Err(FlowError::InvalidTransition(
114            "workflow signal name must not be empty".to_string(),
115        ));
116    }
117    Ok(())
118}
119
120pub(crate) fn validate_signal_wait(wait_id: &str, signal_name: &str) -> Result<()> {
121    if wait_id.trim().is_empty() {
122        return Err(FlowError::InvalidTransition(
123            "workflow signal wait id must not be empty".to_string(),
124        ));
125    }
126    validate_signal_name(signal_name)
127}
a3s_flow/model/signal.rs

a3s_flow/model/
signal.rs
