ironflow_api/routes/
pause_workflow.rs

1//! `POST /api/v1/workflows/:name/pause` and `POST /api/v1/workflows/:name/resume`
2//! — Hold back every queued run of a workflow, and release them.
3
4use axum::extract::{Path, State};
5use axum::response::IntoResponse;
6use ironflow_auth::extractor::Authenticated;
7use ironflow_engine::error::EngineError;
8use ironflow_store::entities::ApiKeyScope;
9
10use crate::authorization::AuthorizeExt;
11use crate::entities::WorkflowPauseResponse;
12use crate::error::ApiError;
13use crate::response::ok;
14use crate::state::AppState;
15
16/// Pause a workflow: workers stop picking its queued runs.
17///
18/// Runs already executing are left alone (pause them one by one with
19/// `POST /api/v1/runs/{id}/pause`), and new runs are still created: they
20/// wait in the queue until the workflow is resumed. Pausing a workflow
21/// already paused succeeds and keeps its first `paused_at`.
22#[cfg_attr(
23    feature = "openapi",
24    utoipa::path(
25        post,
26        path = "/api/v1/workflows/{name}/pause",
27        tags = ["workflows"],
28        params(("name" = String, Path, description = "Workflow name")),
29        responses(
30            (status = 200, description = "Workflow paused", body = WorkflowPauseResponse),
31            (status = 401, description = "Unauthorized"),
32            (status = 403, description = "Forbidden"),
33            (status = 404, description = "Workflow not found")
34        ),
35        security(("Bearer" = []))
36    )
37)]
38pub async fn pause_workflow(
39    auth: Authenticated,
40    State(state): State<AppState>,
41    Path(name): Path<String>,
42) -> Result<impl IntoResponse, ApiError> {
43    auth.require(ApiKeyScope::Admin, true)?;
44
45    let pause = state
46        .engine
47        .pause_workflow(&name, Some(auth.user_id))
48        .await
49        .map_err(|err| map_engine_error(err, &name))?;
50
51    Ok(ok(WorkflowPauseResponse::from(pause)))
52}
53
54/// Resume a paused workflow: workers pick its queued runs again.
55///
56/// Resuming a workflow that is not paused succeeds and changes nothing.
57#[cfg_attr(
58    feature = "openapi",
59    utoipa::path(
60        post,
61        path = "/api/v1/workflows/{name}/resume",
62        tags = ["workflows"],
63        params(("name" = String, Path, description = "Workflow name")),
64        responses(
65            (status = 200, description = "Workflow resumed", body = WorkflowPauseResponse),
66            (status = 401, description = "Unauthorized"),
67            (status = 403, description = "Forbidden"),
68            (status = 404, description = "Workflow not found")
69        ),
70        security(("Bearer" = []))
71    )
72)]
73pub async fn resume_workflow(
74    auth: Authenticated,
75    State(state): State<AppState>,
76    Path(name): Path<String>,
77) -> Result<impl IntoResponse, ApiError> {
78    auth.require(ApiKeyScope::Admin, true)?;
79
80    state
81        .engine
82        .resume_workflow(&name)
83        .await
84        .map_err(|err| map_engine_error(err, &name))?;
85
86    Ok(ok(WorkflowPauseResponse::resumed(&name)))
87}
88
89/// Map an error of a workflow pause or resume to its HTTP status.
90fn map_engine_error(err: EngineError, name: &str) -> ApiError {
91    match err {
92        EngineError::InvalidWorkflow(_) => ApiError::WorkflowNotFound(name.to_string()),
93        EngineError::Store(err) => ApiError::from(err),
94        other => ApiError::Internal(other.to_string()),
95    }
96}
97
98#[cfg(test)]
99mod tests {
100    use std::sync::Arc;
101
102    use axum::Router;
103    use axum::body::Body;
104    use axum::http::{Request, StatusCode};
105    use axum::routing::post;
106    use http_body_util::BodyExt;
107    use ironflow_auth::jwt::JwtConfig;
108    use ironflow_core::providers::claude::ClaudeCodeProvider;
109    use ironflow_engine::context::WorkflowContext;
110    use ironflow_engine::engine::{Engine, ExecutionMode};
111    use ironflow_engine::handler::{HandlerFuture, WorkflowHandler};
112    use ironflow_engine::notify::Event;
113    use ironflow_store::memory::InMemoryStore;
114    use ironflow_store::store::RunStore;
115    use serde_json::{Value as JsonValue, from_slice};
116    use tokio::sync::broadcast;
117    use tower::ServiceExt;
118
119    use super::*;
120    use crate::routes::test_helpers::create_user_auth_header;
121
122    struct Deploy;
123
124    impl WorkflowHandler for Deploy {
125        fn name(&self) -> &str {
126            "deploy"
127        }
128
129        fn execute<'a>(&'a self, _ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
130            Box::pin(async move { Ok(()) })
131        }
132    }
133
134    fn test_state(store: Arc<InMemoryStore>) -> AppState {
135        let provider = Arc::new(ClaudeCodeProvider::new());
136        let mut engine =
137            Engine::new(store.clone(), provider).with_execution_mode(ExecutionMode::Workers);
138        engine.register(Deploy).unwrap();
139        let jwt_config = Arc::new(JwtConfig {
140            secret: "test-secret".to_string(),
141            access_token_ttl_secs: 900,
142            refresh_token_ttl_secs: 604800,
143            cookie_domain: None,
144            cookie_secure: false,
145        });
146        let (event_sender, _) = broadcast::channel::<Event>(1);
147        AppState::new(
148            store,
149            Arc::new(engine),
150            jwt_config,
151            "test-worker-token".to_string(),
152            event_sender,
153        )
154    }
155
156    async fn send(
157        state: AppState,
158        action: &str,
159        name: &str,
160        admin: bool,
161    ) -> (StatusCode, JsonValue) {
162        let auth_header = create_user_auth_header(&state, "testuser", admin).await;
163        let app = Router::new()
164            .route("/{name}/pause", post(pause_workflow))
165            .route("/{name}/resume", post(resume_workflow))
166            .with_state(state);
167
168        let req = Request::builder()
169            .method("POST")
170            .uri(format!("/{name}/{action}"))
171            .header("authorization", auth_header)
172            .body(Body::empty())
173            .unwrap();
174
175        let resp = app.oneshot(req).await.unwrap();
176        let status = resp.status();
177        let body = resp.into_body().collect().await.unwrap().to_bytes();
178        (status, from_slice(&body).unwrap())
179    }
180
181    #[tokio::test]
182    async fn pause_workflow_returns_paused() {
183        let store = Arc::new(InMemoryStore::new());
184
185        let (status, body) = send(test_state(store.clone()), "pause", "deploy", true).await;
186
187        assert_eq!(status, StatusCode::OK);
188        assert_eq!(body["data"]["workflow_name"], "deploy");
189        assert!(body["data"]["paused_at"].is_string());
190        assert!(body["data"]["paused_by"].is_string());
191        let pauses = store.list_workflow_pauses().await.unwrap();
192        assert_eq!(pauses.len(), 1);
193        assert_eq!(pauses[0].workflow_name, "deploy");
194    }
195
196    #[tokio::test]
197    async fn pause_unknown_workflow_returns_404() {
198        let store = Arc::new(InMemoryStore::new());
199
200        let (status, _) = send(test_state(store.clone()), "pause", "unknown", true).await;
201
202        assert_eq!(status, StatusCode::NOT_FOUND);
203        assert!(store.list_workflow_pauses().await.unwrap().is_empty());
204    }
205
206    #[tokio::test]
207    async fn pause_workflow_non_admin_returns_403() {
208        let store = Arc::new(InMemoryStore::new());
209
210        let (status, _) = send(test_state(store.clone()), "pause", "deploy", false).await;
211
212        assert_eq!(status, StatusCode::FORBIDDEN);
213        assert!(store.list_workflow_pauses().await.unwrap().is_empty());
214    }
215
216    #[tokio::test]
217    async fn resume_workflow_returns_not_paused() {
218        let store = Arc::new(InMemoryStore::new());
219        store.pause_workflow("deploy", None).await.unwrap();
220
221        let (status, body) = send(test_state(store.clone()), "resume", "deploy", true).await;
222
223        assert_eq!(status, StatusCode::OK);
224        assert_eq!(body["data"]["workflow_name"], "deploy");
225        assert!(body["data"].get("paused_at").is_none());
226        assert!(store.list_workflow_pauses().await.unwrap().is_empty());
227    }
228
229    #[tokio::test]
230    async fn resume_workflow_not_paused_is_accepted() {
231        let store = Arc::new(InMemoryStore::new());
232
233        let (status, body) = send(test_state(store), "resume", "deploy", true).await;
234
235        assert_eq!(status, StatusCode::OK);
236        assert!(body["data"].get("paused_at").is_none());
237    }
238
239    #[tokio::test]
240    async fn resume_unknown_workflow_returns_404() {
241        let store = Arc::new(InMemoryStore::new());
242
243        let (status, _) = send(test_state(store), "resume", "unknown", true).await;
244
245        assert_eq!(status, StatusCode::NOT_FOUND);
246    }
247
248    #[tokio::test]
249    async fn resume_workflow_non_admin_returns_403() {
250        let store = Arc::new(InMemoryStore::new());
251        store.pause_workflow("deploy", None).await.unwrap();
252
253        let (status, _) = send(test_state(store.clone()), "resume", "deploy", false).await;
254
255        assert_eq!(status, StatusCode::FORBIDDEN);
256        assert_eq!(store.list_workflow_pauses().await.unwrap().len(), 1);
257    }
258}
ironflow_api/routes/pause_workflow.rs

ironflow_api/routes/
pause_workflow.rs
