ironflow_engine/context/steps/
signal_send.rs

1//! Signal send step for [`WorkflowContext`].
2
3use std::sync::Arc;
4use std::time::Instant;
5
6use chrono::Utc;
7use rust_decimal::Decimal;
8use serde_json::{from_value, json, to_value};
9use tracing::info;
10use uuid::Uuid;
11
12use ironflow_store::models::{NewSignal, NewStep, StepKind, StepStatus, StepUpdate, step_trace_id};
13
14use crate::context::WorkflowContext;
15use crate::error::EngineError;
16use crate::plan::lock_plan;
17use crate::signal::{Signal, SignalDelivery, SignalOrigin, SignalSender};
18
19impl WorkflowContext {
20    /// Send a typed signal to the runs waiting for it, as a step of this run.
21    ///
22    /// The signal is named [`S::NAME`](Signal::NAME), carries `key` and has
23    /// `signal` as payload: every run waiting on that `(name, key)` with
24    /// [`wait_for_signal`](Self::wait_for_signal) and a matching schema
25    /// resumes, exactly as with `POST /api/v1/signals`, without any API key.
26    /// The step records the [`SignalDelivery`] and the audit log entry of the
27    /// signal names this run and step as its sender.
28    ///
29    /// The handler must declare `S` in
30    /// [`WorkflowHandler::sent_signals`](crate::handler::WorkflowHandler::sent_signals).
31    /// The declaration guards against a workflow sending a signal that is not
32    /// its own by mistake; it is not a security boundary. Anyone who can
33    /// start this workflow chooses the key and payload it sends, so derive
34    /// them from data you trust.
35    ///
36    /// A step that already completed is replayed without sending again. A
37    /// step interrupted mid-delivery (worker lost) and executed again in the
38    /// same attempt comes back with [`SignalDelivery::duplicate`] set instead
39    /// of sending twice. A retry of the run is a new attempt: it sends again,
40    /// like it runs a shell or HTTP step again.
41    ///
42    /// While planning, the step is recorded and nothing is sent.
43    ///
44    /// # Errors
45    ///
46    /// The step fails, and nothing is sent, with
47    /// [`EngineError::SignalNotAllowed`] when the handler does not declare
48    /// `S`, and [`EngineError::InvalidSignal`] when `key` is empty or longer
49    /// than [`MAX_SIGNAL_FIELD_LEN`](crate::signal::MAX_SIGNAL_FIELD_LEN).
50    /// Returns [`EngineError::ReplayDivergence`] when the step recorded at
51    /// this position has a different name or kind, and other [`EngineError`]
52    /// variants on store or delivery failures.
53    ///
54    /// # Examples
55    ///
56    /// ```no_run
57    /// use ironflow_engine::context::WorkflowContext;
58    /// use ironflow_engine::error::EngineError;
59    /// use ironflow_engine::signal::Signal;
60    /// use schemars::JsonSchema;
61    /// use serde::{Deserialize, Serialize};
62    ///
63    /// #[derive(Serialize, Deserialize, JsonSchema)]
64    /// struct HumanReplied {
65    ///     body: String,
66    /// }
67    ///
68    /// impl Signal for HumanReplied {
69    ///     const NAME: &'static str = "gitlab.human_replied";
70    /// }
71    ///
72    /// # async fn example(ctx: &mut WorkflowContext, discussion_id: &str) -> Result<(), EngineError> {
73    /// let reply = HumanReplied { body: "ship it".to_string() };
74    /// let delivery = ctx.send_signal("wake-waiting-run", &reply, discussion_id).await?;
75    /// println!("{} runs resumed", delivery.resumed.len());
76    /// # Ok(())
77    /// # }
78    /// ```
79    pub async fn send_signal<S: Signal>(
80        &mut self,
81        name: &str,
82        signal: &S,
83        key: &str,
84    ) -> Result<SignalDelivery, EngineError> {
85        let kind = StepKind::SignalSend;
86
87        // Plan mode: record the step, send nothing.
88        if let Some(plan) = self.plan().cloned() {
89            self.position += 1;
90            let mut recorder = lock_plan(&plan);
91            recorder.record_or_stop(name, kind, &self.workflow_name, None)?;
92            recorder.set_last(vec![name.to_string()]);
93            return Ok(SignalDelivery {
94                signal_id: Uuid::nil(),
95                duplicate: false,
96                resumed: Vec::new(),
97                rejected: Vec::new(),
98            });
99        }
100
101        let position = self.next_position();
102
103        if let Some(replayed) = self.try_replay_step(position, name, &kind, false)? {
104            return Ok(from_value(replayed.output)?);
105        }
106
107        let payload = to_value(signal)?;
108        let step = self
109            .store
110            .create_step(NewStep {
111                run_id: self.run_id,
112                trace_id: step_trace_id(self.run_id, name, position),
113                name: name.to_string(),
114                kind,
115                position,
116                input: Some(json!({ "name": S::NAME, "key": key, "payload": payload })),
117                is_error_handler: false,
118            })
119            .await?;
120        self.start_step(step.id, Utc::now()).await?;
121
122        let start = Instant::now();
123        let new_signal = NewSignal {
124            name: S::NAME.to_string(),
125            key: key.to_string(),
126            payload,
127            // Positions repeat across attempts, never within one: a step
128            // executed again after a lost worker reuses the key.
129            idempotency_id: Some(format!(
130                "run:{}:attempt:{}:step:{position}",
131                self.run_id, self.attempt
132            )),
133        };
134        let origin = SignalOrigin {
135            run_id: self.run_id,
136            step_id: step.id,
137        };
138        let delivered = match self.check_sent_signal(S::NAME) {
139            Ok(sender) => sender.send(new_signal, origin).await,
140            Err(err) => Err(err),
141        };
142
143        match delivered {
144            Ok(delivery) => {
145                let duration_ms = start.elapsed().as_millis() as u64;
146                self.total_duration_ms += duration_ms;
147                let step_id = self
148                    .complete_step(
149                        &step,
150                        StepUpdate {
151                            status: Some(StepStatus::Completed),
152                            output: Some(to_value(&delivery)?),
153                            duration_ms: Some(duration_ms),
154                            cost_usd: Some(Decimal::ZERO),
155                            completed_at: Some(Utc::now()),
156                            ..StepUpdate::default()
157                        },
158                    )
159                    .await?;
160                self.last_step_ids = vec![step_id];
161                info!(
162                    run_id = %self.run_id,
163                    step = %name,
164                    signal = %S::NAME,
165                    resumed = delivery.resumed.len(),
166                    duplicate = delivery.duplicate,
167                    "signal sent"
168                );
169                Ok(delivery)
170            }
171            Err(err) => {
172                self.fail_step(step.id, &err).await;
173                Err(err)
174            }
175        }
176    }
177
178    /// The sender to deliver `signal` with, once the handler declares it.
179    fn check_sent_signal(&self, signal: &str) -> Result<Arc<dyn SignalSender>, EngineError> {
180        let declared = self
181            .handler_resolver
182            .as_ref()
183            .and_then(|resolve| resolve(&self.workflow_name))
184            .is_some_and(|handler| handler.sent_signals().allows(signal));
185        if !declared {
186            return Err(EngineError::SignalNotAllowed {
187                workflow: self.workflow_name.clone(),
188                signal: signal.to_string(),
189            });
190        }
191        self.signal_sender.clone().ok_or_else(|| {
192            EngineError::StepConfig(
193                "no signal sender: run this workflow through an Engine".to_string(),
194            )
195        })
196    }
197}
ironflow_engine/context/steps/signal_send.rs

ironflow_engine/context/steps/
signal_send.rs
