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

ironflow_api/
escalator.rs
