ironflow-api 2.52.3

//! `POST /api/v1/runs/:id/cancel` — Cancel a run and the sub-workflow runs below it.

use axum::extract::{Path, State};
use axum::response::IntoResponse;
use ironflow_auth::extractor::Authenticated;
use ironflow_engine::error::EngineError;
use ironflow_store::entities::ApiKeyScope;
use ironflow_store::error::StoreError;
use uuid::Uuid;

use crate::authorization::AuthorizeExt;
use crate::entities::{CancelRunResponse, RunResponse};
use crate::error::ApiError;
use crate::response::ok;
use crate::state::AppState;

/// Cancel a run that has not finished, with every sub-workflow run below it.
///
/// The run and its active descendants move to `Cancelled`, their open steps
/// are closed and their concurrency keys released (see
/// [`Engine::cancel_run`](ironflow_engine::engine::Engine::cancel_run)). The
/// response lists the descendants cancelled. Cancelling a child whose chain
/// is suspended wakes its root, whose `Workflow` step then fails (or, with
/// `allow_failure`, completes with the `Cancelled` child).
///
/// Cancelling a run already `Cancelled` succeeds and cancels only what was
/// still active below it. Returns 400 for a run that finished otherwise.
#[cfg_attr(
    feature = "openapi",
    utoipa::path(
        post,
        path = "/api/v1/runs/{id}/cancel",
        tags = ["runs"],
        params(("id" = Uuid, Path, description = "Run ID")),
        responses(
            (status = 200, description = "Run cancelled, with the sub-runs cancelled along", body = CancelRunResponse),
            (status = 400, description = "Run cannot be cancelled"),
            (status = 401, description = "Unauthorized"),
            (status = 403, description = "Forbidden"),
            (status = 404, description = "Run not found")
        ),
        security(("Bearer" = []))
    )
)]
pub async fn cancel_run(
    auth: Authenticated,
    State(state): State<AppState>,
    Path(id): Path<Uuid>,
) -> Result<impl IntoResponse, ApiError> {
    auth.require(ApiKeyScope::RunsManage, true)?;

    let cancellation = state.engine.cancel_run(id).await.map_err(|err| match err {
        EngineError::Store(StoreError::RunNotFound(id)) => ApiError::RunNotFound(id),
        EngineError::Store(StoreError::InvalidTransition { from, .. }) => {
            ApiError::BadRequest(format!("cannot cancel run in {from} state"))
        }
        EngineError::Store(err) => ApiError::from(err),
        other => ApiError::Internal(other.to_string()),
    })?;

    Ok(ok(CancelRunResponse {
        run: RunResponse::from(cancellation.run),
        cancelled_descendants: cancellation.cancelled_descendants,
    }))
}

#[cfg(test)]
mod tests {
    use std::collections::HashMap;

    use axum::Router;
    use axum::body::Body;
    use axum::http::{Request, StatusCode as HttpStatusCode};
    use axum::routing::post;
    use chrono::Utc;
    use http_body_util::BodyExt;
    use ironflow_core::providers::claude::ClaudeCodeProvider;
    use ironflow_engine::engine::Engine;
    use ironflow_engine::notify::Event;
    use ironflow_store::memory::InMemoryStore;
    use ironflow_store::models::{NewRun, RunStatus, TriggerKind};
    use ironflow_store::store::RunStore;
    use serde_json::{Value as JsonValue, from_slice, json};
    use std::sync::Arc;
    use tokio::sync::broadcast;
    use tower::ServiceExt;
    use uuid::Uuid;

    use super::*;
    use crate::routes::test_helpers::create_user_auth_header;

    fn test_state(store: Arc<InMemoryStore>) -> AppState {
        Arc::new(InMemoryStore::new());
        let provider = Arc::new(ClaudeCodeProvider::new());
        let engine = Arc::new(Engine::new(store.clone(), provider));
        let jwt_config = Arc::new(ironflow_auth::jwt::JwtConfig {
            secret: "test-secret".to_string(),
            access_token_ttl_secs: 900,
            refresh_token_ttl_secs: 604800,
            cookie_domain: None,
            cookie_secure: false,
        });
        let (event_sender, _) = broadcast::channel::<Event>(1);
        AppState::new(
            store,
            engine,
            jwt_config,
            "test-worker-token".to_string(),
            event_sender,
        )
    }

    #[tokio::test]
    async fn cancel_pending_run() {
        let store = Arc::new(InMemoryStore::new());
        let run = store
            .create_run(NewRun {
                created_by: None,
                workflow_name: "test".to_string(),
                trigger: TriggerKind::Manual,
                payload: json!({}),
                max_retries: 0,
                handler_version: None,
                labels: HashMap::new(),
                scheduled_at: None,
                idempotency_key: None,
                concurrency_key: None,
                priority: 0,
                concurrency_limits: Vec::new(),
                max_cost_usd: None,
                worker_tags: Vec::new(),
            })
            .await
            .unwrap()
            .into_run();

        let state = test_state(store.clone());
        let auth_header = create_user_auth_header(&state, "testuser", true).await;
        let app = Router::new()
            .route("/{id}/cancel", post(cancel_run))
            .with_state(state);

        let req = Request::builder()
            .method("POST")
            .uri(format!("/{}/cancel", run.id))
            .header("content-type", "application/json")
            .header("authorization", auth_header)
            .body(Body::from("{}"))
            .unwrap();

        let resp = app.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), HttpStatusCode::OK);

        let body = resp.into_body().collect().await.unwrap().to_bytes();
        let json_val: JsonValue = from_slice(&body).unwrap();
        assert_eq!(json_val["data"]["status"], "cancelled");

        let cancelled = store.get_run(run.id).await.unwrap().unwrap();
        assert_eq!(cancelled.status.state, RunStatus::Cancelled);
    }

    #[tokio::test]
    async fn cancel_completed_run_returns_400() {
        let store = Arc::new(InMemoryStore::new());
        let run = store
            .create_run(NewRun {
                created_by: None,
                workflow_name: "test".to_string(),
                trigger: TriggerKind::Manual,
                payload: json!({}),
                max_retries: 0,
                handler_version: None,
                labels: HashMap::new(),
                scheduled_at: None,
                idempotency_key: None,
                concurrency_key: None,
                priority: 0,
                concurrency_limits: Vec::new(),
                max_cost_usd: None,
                worker_tags: Vec::new(),
            })
            .await
            .unwrap()
            .into_run();

        store
            .update_run_status(run.id, RunStatus::Running)
            .await
            .unwrap();
        store
            .update_run_status(run.id, RunStatus::Completed)
            .await
            .unwrap();

        let state = test_state(store);
        let auth_header = create_user_auth_header(&state, "testuser", true).await;
        let app = Router::new()
            .route("/{id}/cancel", post(cancel_run))
            .with_state(state);

        let req = Request::builder()
            .method("POST")
            .uri(format!("/{}/cancel", run.id))
            .header("content-type", "application/json")
            .header("authorization", auth_header)
            .body(Body::from("{}"))
            .unwrap();

        let resp = app.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), HttpStatusCode::BAD_REQUEST);
    }

    #[tokio::test]
    async fn cancel_running_run() {
        let store = Arc::new(InMemoryStore::new());
        let run = store
            .create_run(NewRun {
                created_by: None,
                workflow_name: "test".to_string(),
                trigger: TriggerKind::Manual,
                payload: json!({}),
                max_retries: 0,
                handler_version: None,
                labels: HashMap::new(),
                scheduled_at: None,
                idempotency_key: None,
                concurrency_key: None,
                priority: 0,
                concurrency_limits: Vec::new(),
                max_cost_usd: None,
                worker_tags: Vec::new(),
            })
            .await
            .unwrap()
            .into_run();

        store
            .update_run_status(run.id, RunStatus::Running)
            .await
            .unwrap();

        let state = test_state(store);
        let auth_header = create_user_auth_header(&state, "testuser", true).await;
        let app = Router::new()
            .route("/{id}/cancel", post(cancel_run))
            .with_state(state);

        let req = Request::builder()
            .method("POST")
            .uri(format!("/{}/cancel", run.id))
            .header("content-type", "application/json")
            .header("authorization", auth_header)
            .body(Body::from("{}"))
            .unwrap();

        let resp = app.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), HttpStatusCode::OK);

        let body = resp.into_body().collect().await.unwrap().to_bytes();
        let json_val: JsonValue = from_slice(&body).unwrap();
        assert_eq!(json_val["data"]["status"], "cancelled");
    }

    #[tokio::test]
    async fn cancel_run_waiting_for_a_retry_stops_it_for_good() {
        let store = Arc::new(InMemoryStore::new());
        let run = store
            .create_run(NewRun {
                created_by: None,
                workflow_name: "test".to_string(),
                trigger: TriggerKind::Manual,
                payload: json!({}),
                max_retries: 3,
                handler_version: None,
                labels: HashMap::new(),
                scheduled_at: None,
                idempotency_key: None,
                concurrency_key: None,
                priority: 0,
                concurrency_limits: Vec::new(),
                max_cost_usd: None,
                worker_tags: Vec::new(),
            })
            .await
            .unwrap()
            .into_run();

        store
            .update_run_status(run.id, RunStatus::Running)
            .await
            .unwrap();
        store
            .update_run(
                run.id,
                ironflow_store::models::RunUpdate {
                    status: Some(RunStatus::Retrying),
                    increment_retry: true,
                    scheduled_at: Some(Utc::now() - chrono::TimeDelta::seconds(1)),
                    ..Default::default()
                },
            )
            .await
            .unwrap();

        let state = test_state(store.clone());
        let auth_header = create_user_auth_header(&state, "testuser", true).await;
        let app = Router::new()
            .route("/{id}/cancel", post(cancel_run))
            .with_state(state);

        let req = Request::builder()
            .method("POST")
            .uri(format!("/{}/cancel", run.id))
            .header("content-type", "application/json")
            .header("authorization", auth_header)
            .body(Body::from("{}"))
            .unwrap();

        let resp = app.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), HttpStatusCode::OK);

        // Its backoff had already elapsed, yet it must never be picked up again.
        assert!(store.pick_next_pending(None).await.unwrap().is_none());
    }

    #[tokio::test]
    async fn cancel_failed_run_returns_400() {
        let store = Arc::new(InMemoryStore::new());
        let run = store
            .create_run(NewRun {
                created_by: None,
                workflow_name: "test".to_string(),
                trigger: TriggerKind::Manual,
                payload: json!({}),
                max_retries: 0,
                handler_version: None,
                labels: HashMap::new(),
                scheduled_at: None,
                idempotency_key: None,
                concurrency_key: None,
                priority: 0,
                concurrency_limits: Vec::new(),
                max_cost_usd: None,
                worker_tags: Vec::new(),
            })
            .await
            .unwrap()
            .into_run();

        store
            .update_run_status(run.id, RunStatus::Running)
            .await
            .unwrap();
        store
            .update_run_status(run.id, RunStatus::Failed)
            .await
            .unwrap();

        let state = test_state(store);
        let auth_header = create_user_auth_header(&state, "testuser", true).await;
        let app = Router::new()
            .route("/{id}/cancel", post(cancel_run))
            .with_state(state);

        let req = Request::builder()
            .method("POST")
            .uri(format!("/{}/cancel", run.id))
            .header("content-type", "application/json")
            .header("authorization", auth_header)
            .body(Body::from("{}"))
            .unwrap();

        let resp = app.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), HttpStatusCode::BAD_REQUEST);
    }

    #[tokio::test]
    async fn cancel_nonexistent_run_returns_404() {
        let store = Arc::new(InMemoryStore::new());
        let state = test_state(store);
        let auth_header = create_user_auth_header(&state, "testuser", true).await;
        let app = Router::new()
            .route("/{id}/cancel", post(cancel_run))
            .with_state(state);

        let req = Request::builder()
            .method("POST")
            .uri(format!("/{}/cancel", Uuid::now_v7()))
            .header("content-type", "application/json")
            .header("authorization", auth_header)
            .body(Body::from("{}"))
            .unwrap();

        let resp = app.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), HttpStatusCode::NOT_FOUND);
    }
}