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                max_cost_usd: None,
177            })
178            .await
179            .expect("create run")
180            .into_run();
181        store
182            .update_run_status(run.id, RunStatus::Running)
183            .await
184            .expect("to running");
185        store
186            .update_run_status(run.id, RunStatus::AwaitingApproval)
187            .await
188            .expect("to awaiting approval");
189
190        let step = store
191            .create_step(NewStep {
192                run_id: run.id,
193                trace_id: step_trace_id(run.id, "prod-gate", 0),
194                name: "prod-gate".to_string(),
195                kind: StepKind::Approval,
196                position: 0,
197                input: Some(serde_json::to_value(&config).expect("serialize config")),
198                is_error_handler: false,
199            })
200            .await
201            .expect("create step");
202        store
203            .update_step(
204                step.id,
205                StepUpdate {
206                    status: Some(StepStatus::Running),
207                    ..StepUpdate::default()
208                },
209            )
210            .await
211            .expect("to running");
212        store
213            .update_step(
214                step.id,
215                StepUpdate {
216                    status: Some(StepStatus::AwaitingApproval),
217                    approval_deadline_at: Some(Utc::now() - TimeDelta::seconds(1)),
218                    ..StepUpdate::default()
219                },
220            )
221            .await
222            .expect("arm an expired timer");
223
224        (store, Escalator::new(engine), run.id, step.id)
225    }
226
227    #[tokio::test]
228    async fn tick_auto_rejects_a_gate_past_its_deadline() {
229        let config = ApprovalConfig::new("Deploy?")
230            .with_deadline_secs(60)
231            .on_timeout(EscalationPolicy::AutoReject);
232        let (store, escalator, run_id, step_id) = expired_gate(config).await;
233
234        escalator.tick().await;
235
236        let run = store.get_run(run_id).await.unwrap().unwrap();
237        assert_eq!(run.status.state, RunStatus::Failed);
238        assert_eq!(run.error.as_deref(), Some("approval timeout"));
239
240        let step = store.get_step(step_id).await.unwrap().unwrap();
241        assert_eq!(step.status.state, StepStatus::Failed);
242        assert_eq!(step.error.as_deref(), Some("approval timeout"));
243        assert!(step.approval_deadline_at.is_none());
244    }
245
246    #[tokio::test]
247    async fn tick_leaves_a_gate_without_a_deadline_alone() {
248        let config = ApprovalConfig::new("Deploy?");
249        let (store, escalator, run_id, step_id) = expired_gate(config).await;
250
251        // Drop the timer the fixture armed: this gate carries no SLA.
252        store
253            .update_step(
254                step_id,
255                StepUpdate {
256                    clear_approval_deadline: true,
257                    ..StepUpdate::default()
258                },
259            )
260            .await
261            .expect("clear timer");
262
263        escalator.tick().await;
264
265        let run = store.get_run(run_id).await.unwrap().unwrap();
266        assert_eq!(run.status.state, RunStatus::AwaitingApproval);
267        let step = store.get_step(step_id).await.unwrap().unwrap();
268        assert_eq!(step.status.state, StepStatus::AwaitingApproval);
269    }
270
271    #[tokio::test]
272    async fn run_stops_on_shutdown() {
273        let config = ApprovalConfig::new("Deploy?");
274        let (_store, escalator, _run_id, _step_id) = expired_gate(config).await;
275        let shutdown = CancellationToken::new();
276        shutdown.cancel();
277
278        // Returns instead of looping forever.
279        tokio::time::timeout(Duration::from_secs(5), escalator.run(shutdown))
280            .await
281            .expect("escalator stopped");
282    }
283}
ironflow_api/escalator.rs

ironflow_api/
escalator.rs
