ironflow_api/
escalator.rs

1//! Resolution of approval gates that missed their SLA deadline.
2//!
3//! An approval gate can carry a deadline. The deadline lives on the step row, so
4//! it survives an API or worker restart. The [`Escalator`] periodically hands
5//! every expired gate to the engine's
6//! [`ApprovalEscalator`](ironflow_engine::escalation::ApprovalEscalator), which
7//! applies the gate's configured escalation policy.
8//!
9//! Gates without a deadline are never touched, and a deadline fires at most once
10//! even when several API instances run this loop.
11
12use std::sync::Arc;
13use std::time::Duration;
14
15use ironflow_engine::engine::Engine;
16use ironflow_engine::escalation::{ApprovalEscalator, EscalationAction};
17use tokio::time::interval;
18use tokio_util::sync::CancellationToken;
19use tracing::{error, info};
20
21/// How often expired approval deadlines are collected.
22pub const DEFAULT_ESCALATOR_INTERVAL: Duration = Duration::from_secs(30);
23
24/// How many gates a single tick resolves.
25///
26/// Bounded so that a burst of expiries (a long outage with many open gates)
27/// resorbs progressively instead of holding a long transaction on the steps
28/// table.
29pub const DEFAULT_ESCALATOR_BATCH_SIZE: u32 = 50;
30
31/// Periodic task that escalates approval gates past their deadline.
32///
33/// # Examples
34///
35/// ```no_run
36/// use std::sync::Arc;
37/// use std::time::Duration;
38/// use ironflow_api::escalator::Escalator;
39/// use ironflow_core::providers::claude::ClaudeCodeProvider;
40/// use ironflow_engine::engine::Engine;
41/// use ironflow_store::memory::InMemoryStore;
42/// use ironflow_store::store::Store;
43/// use tokio_util::sync::CancellationToken;
44///
45/// # async fn example() {
46/// let store: Arc<dyn Store> = Arc::new(InMemoryStore::new());
47/// let engine = Arc::new(Engine::new(store, Arc::new(ClaudeCodeProvider::new())));
48///
49/// let escalator = Escalator::new(engine).interval(Duration::from_secs(15));
50/// tokio::spawn(escalator.run(CancellationToken::new()));
51/// # }
52/// ```
53pub struct Escalator {
54    escalator: ApprovalEscalator,
55    interval: Duration,
56}
57
58impl Escalator {
59    /// Create an escalator with the default interval and batch size.
60    pub fn new(engine: Arc<Engine>) -> Self {
61        Self {
62            escalator: ApprovalEscalator::new(engine).batch_size(DEFAULT_ESCALATOR_BATCH_SIZE),
63            interval: DEFAULT_ESCALATOR_INTERVAL,
64        }
65    }
66
67    /// Set how often expired deadlines are collected.
68    ///
69    /// Keep it well below the shortest SLA in use, otherwise a gate overshoots
70    /// its deadline by up to one interval.
71    pub fn interval(mut self, interval: Duration) -> Self {
72        self.interval = interval;
73        self
74    }
75
76    /// Set how many gates a single tick resolves.
77    pub fn batch_size(self, batch_size: u32) -> Self {
78        Self {
79            escalator: self.escalator.batch_size(batch_size),
80            interval: self.interval,
81        }
82    }
83
84    /// Run the escalation loop until `shutdown` is cancelled.
85    ///
86    /// Store errors are logged and the loop keeps going: a transient database
87    /// failure must not silently stop escalation.
88    pub async fn run(self, shutdown: CancellationToken) {
89        let mut ticker = interval(self.interval);
90        // The first tick fires immediately; skip it so startup is not a burst.
91        ticker.tick().await;
92
93        info!(
94            interval_secs = self.interval.as_secs(),
95            "approval escalator started"
96        );
97
98        loop {
99            tokio::select! {
100                _ = shutdown.cancelled() => {
101                    info!("approval escalator stopped");
102                    return;
103                }
104                _ = ticker.tick() => {
105                    self.tick().await;
106                }
107            }
108        }
109    }
110
111    /// Escalate one batch of expired gates.
112    ///
113    /// Exposed for tests and for callers that drive the schedule themselves.
114    pub async fn tick(&self) {
115        let records = match self.escalator.tick().await {
116            Ok(records) => records,
117            Err(err) => {
118                error!(error = %err, "failed to collect expired approval deadlines");
119                return;
120            }
121        };
122
123        for record in &records {
124            // A stale gate resolved on its own between the claim and the
125            // escalation; that is routine, not news.
126            if record.action == EscalationAction::Stale {
127                continue;
128            }
129
130            info!(
131                run_id = %record.run_id,
132                step_id = %record.step_id,
133                stage = record.stage,
134                action = ?record.action,
135                reason = %record.reason,
136                "approval gate escalated"
137            );
138        }
139    }
140}
141
142#[cfg(test)]
143mod tests {
144    use chrono::{TimeDelta, Utc};
145    use ironflow_core::providers::claude::ClaudeCodeProvider;
146    use ironflow_engine::config::{ApprovalConfig, EscalationPolicy};
147    use ironflow_store::entities::{
148        NewRun, NewStep, RunStatus, StepKind, StepStatus, StepUpdate, TriggerKind, step_trace_id,
149    };
150    use ironflow_store::memory::InMemoryStore;
151    use ironflow_store::store::{RunStore, Store};
152    use serde_json::json;
153    use std::collections::HashMap;
154    use uuid::Uuid;
155
156    use super::*;
157
158    /// Build an escalator over a store holding one run stuck on an expired gate.
159    async fn expired_gate(config: ApprovalConfig) -> (Arc<InMemoryStore>, Escalator, Uuid, Uuid) {
160        let store = Arc::new(InMemoryStore::new());
161        let store_dyn: Arc<dyn Store> = store.clone();
162        let engine = Arc::new(Engine::new(store_dyn, Arc::new(ClaudeCodeProvider::new())));
163
164        let run = store
165            .create_run(NewRun {
166                created_by: None,
167                workflow_name: "deploy".to_string(),
168                trigger: TriggerKind::Manual,
169                payload: json!({}),
170                max_retries: 0,
171                handler_version: None,
172                labels: HashMap::new(),
173                scheduled_at: None,
174                idempotency_key: None,
175                concurrency_key: None,
176                concurrency_limits: Vec::new(),
177                max_cost_usd: None,
178            })
179            .await
180            .expect("create run")
181            .into_run();
182        store
183            .update_run_status(run.id, RunStatus::Running)
184            .await
185            .expect("to running");
186        store
187            .update_run_status(run.id, RunStatus::AwaitingApproval)
188            .await
189            .expect("to awaiting approval");
190
191        let step = store
192            .create_step(NewStep {
193                run_id: run.id,
194                trace_id: step_trace_id(run.id, "prod-gate", 0),
195                name: "prod-gate".to_string(),
196                kind: StepKind::Approval,
197                position: 0,
198                input: Some(serde_json::to_value(&config).expect("serialize config")),
199                is_error_handler: false,
200            })
201            .await
202            .expect("create step");
203        store
204            .update_step(
205                step.id,
206                StepUpdate {
207                    status: Some(StepStatus::Running),
208                    ..StepUpdate::default()
209                },
210            )
211            .await
212            .expect("to running");
213        store
214            .update_step(
215                step.id,
216                StepUpdate {
217                    status: Some(StepStatus::AwaitingApproval),
218                    approval_deadline_at: Some(Utc::now() - TimeDelta::seconds(1)),
219                    ..StepUpdate::default()
220                },
221            )
222            .await
223            .expect("arm an expired timer");
224
225        (store, Escalator::new(engine), run.id, step.id)
226    }
227
228    #[tokio::test]
229    async fn tick_auto_rejects_a_gate_past_its_deadline() {
230        let config = ApprovalConfig::new("Deploy?")
231            .with_deadline_secs(60)
232            .on_timeout(EscalationPolicy::AutoReject);
233        let (store, escalator, run_id, step_id) = expired_gate(config).await;
234
235        escalator.tick().await;
236
237        let run = store.get_run(run_id).await.unwrap().unwrap();
238        assert_eq!(run.status.state, RunStatus::Failed);
239        assert_eq!(run.error.as_deref(), Some("approval timeout"));
240
241        let step = store.get_step(step_id).await.unwrap().unwrap();
242        assert_eq!(step.status.state, StepStatus::Failed);
243        assert_eq!(step.error.as_deref(), Some("approval timeout"));
244        assert!(step.approval_deadline_at.is_none());
245    }
246
247    #[tokio::test]
248    async fn tick_leaves_a_gate_without_a_deadline_alone() {
249        let config = ApprovalConfig::new("Deploy?");
250        let (store, escalator, run_id, step_id) = expired_gate(config).await;
251
252        // Drop the timer the fixture armed: this gate carries no SLA.
253        store
254            .update_step(
255                step_id,
256                StepUpdate {
257                    clear_approval_deadline: true,
258                    ..StepUpdate::default()
259                },
260            )
261            .await
262            .expect("clear timer");
263
264        escalator.tick().await;
265
266        let run = store.get_run(run_id).await.unwrap().unwrap();
267        assert_eq!(run.status.state, RunStatus::AwaitingApproval);
268        let step = store.get_step(step_id).await.unwrap().unwrap();
269        assert_eq!(step.status.state, StepStatus::AwaitingApproval);
270    }
271
272    #[tokio::test]
273    async fn run_stops_on_shutdown() {
274        let config = ApprovalConfig::new("Deploy?");
275        let (_store, escalator, _run_id, _step_id) = expired_gate(config).await;
276        let shutdown = CancellationToken::new();
277        shutdown.cancel();
278
279        // Returns instead of looping forever.
280        tokio::time::timeout(Duration::from_secs(5), escalator.run(shutdown))
281            .await
282            .expect("escalator stopped");
283    }
284}
ironflow_api/escalator.rs

ironflow_api/
escalator.rs
