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

ironflow_api/
escalator.rs
