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

ironflow_api/routes/
pause_workflow.rs
