ironflow_api/routes/
retry_run.rs

1//! `POST /api/v1/runs/:id/retry` — Retry a failed run.
2
3use axum::extract::{Path, Query, State};
4use axum::http::StatusCode;
5use axum::response::IntoResponse;
6use chrono::Utc;
7use ironflow_auth::extractor::Authenticated;
8use ironflow_engine::error::HANDLER_VERSION_MISMATCH_CODE;
9use ironflow_engine::notify::{Event, RetryForcedEvent, RunCreatedEvent};
10use ironflow_store::entities::ApiKeyScope;
11use ironflow_store::models::{NewRun, RunStatus, TriggerKind};
12use serde::Deserialize;
13use uuid::Uuid;
14
15use crate::actor::run_actor_of;
16use crate::authorization::AuthorizeExt;
17use crate::entities::RunResponse;
18use crate::error::ApiError;
19use crate::response::ok;
20use crate::state::AppState;
21
22/// Query parameters for `POST /api/v1/runs/:id/retry`.
23#[derive(Debug, Deserialize, Default)]
24pub struct RetryQuery {
25    /// Force the retry even when the handler version has changed.
26    #[serde(default)]
27    pub force: bool,
28}
29
30/// Retry a failed run.
31///
32/// Creates a new `Pending` run with `TriggerKind::Retry` pointing to the
33/// original. Returns 400 if the run is not in a retryable state, 409 if an
34/// automatic retry is already armed, 409 if the handler version has
35/// changed since the original run (pass `?force=true` to override), and 409
36/// `CONCURRENCY_CONFLICT` if another active run took the original's
37/// concurrency key since.
38#[cfg_attr(
39    feature = "openapi",
40    utoipa::path(
41        post,
42        path = "/api/v1/runs/{id}/retry",
43        tags = ["runs"],
44        params(
45            ("id" = Uuid, Path, description = "Run ID"),
46            ("force" = Option<bool>, Query, description = "Force retry despite handler version mismatch"),
47        ),
48        responses(
49            (status = 201, description = "Run retry created successfully", body = RunResponse),
50            (status = 400, description = "Run cannot be retried"),
51            (status = 401, description = "Unauthorized"),
52            (status = 403, description = "Forbidden"),
53            (status = 404, description = "Run not found"),
54            (status = 409, description = "Version mismatch, automatic retry already armed, or concurrency key held by an active run (CONCURRENCY_CONFLICT)")
55        ),
56        security(("Bearer" = []))
57    )
58)]
59pub async fn retry_run(
60    auth: Authenticated,
61    State(state): State<AppState>,
62    Path(id): Path<Uuid>,
63    Query(query): Query<RetryQuery>,
64) -> Result<impl IntoResponse, ApiError> {
65    auth.require(ApiKeyScope::RunsManage, true)?;
66
67    let original = state.get_run_or_404(id).await?;
68
69    // Retrying means an automatic retry is already armed; creating a manual
70    // retry on top would produce a duplicate run once the timer fires.
71    if original.status.state == RunStatus::Retrying {
72        return Err(ApiError::Conflict(
73            "run is already waiting for an automatic retry; cancel it first to retry manually"
74                .to_string(),
75        ));
76    }
77
78    if !matches!(
79        original.status.state,
80        RunStatus::Failed | RunStatus::Cancelled | RunStatus::Warning
81    ) {
82        return Err(ApiError::BadRequest(format!(
83            "cannot retry run in {} state",
84            original.status.state
85        )));
86    }
87
88    let force = query.force;
89    let handler = state.engine.get_handler(&original.workflow_name);
90    let current_version = handler.and_then(|h| h.version().map(str::to_string));
91
92    if let Some(handler) = &handler
93        && !handler.is_version_compatible(original.handler_version.as_deref())
94        && !force
95    {
96        return Err(ApiError::Conflict(format!(
97            "{}: handler '{}' is now at version {}, but the run was created \
98             with version {}. Pass ?force=true to override.",
99            HANDLER_VERSION_MISMATCH_CODE,
100            original.workflow_name,
101            current_version.as_deref().unwrap_or("unknown"),
102            original.handler_version.as_deref().unwrap_or("unknown"),
103        )));
104    }
105
106    // When the handler is still registered, use its current version.
107    // When unregistered (e.g. removed between deploys), preserve the
108    // original version so version-tracking information is not lost.
109    let effective_version = current_version.clone().or(original.handler_version.clone());
110
111    let new_run = state
112        .store
113        .create_run(NewRun {
114            workflow_name: original.workflow_name.clone(),
115            trigger: TriggerKind::Retry { parent_run_id: id },
116            payload: original.payload,
117            max_retries: original.max_retries,
118            handler_version: effective_version,
119            labels: original.labels,
120            scheduled_at: None,
121            // The retry is attributed to the user who triggered it, not the
122            // original author, so the audit trail shows who actually acted.
123            created_by: Some(run_actor_of(&auth)),
124            // A retry must not inherit the parent's idempotency key: it is a
125            // new logical operation and must be eligible for its own dedup.
126            idempotency_key: None,
127            // The retry continues the same work, so it stays exclusive under the
128            // same key. The original is terminal and no longer holds it; a run
129            // that took the key since then makes the retry conflict.
130            concurrency_key: original.concurrency_key,
131            // The retry keeps the priority of the run it continues.
132            priority: original.priority,
133            // The retry stays in the same concurrency groups under the same limits.
134            concurrency_limits: original.concurrency_limits,
135            // Inherit the original cost cap so budget constraints survive retries.
136            max_cost_usd: original.max_cost_usd,
137            // The retry needs the same kind of worker as the original.
138            worker_tags: original.worker_tags,
139        })
140        .await?
141        .into_run();
142
143    // Only emit RetryForced when both the handler is registered (so
144    // current_version is meaningful) and the versions actually differ.
145    if force && handler.is_some() && original.handler_version != current_version {
146        state
147            .engine
148            .event_publisher()
149            .publish(Event::RetryForced(RetryForcedEvent {
150                run_id: new_run.id,
151                workflow_name: original.workflow_name.clone(),
152                original_version: original.handler_version.unwrap_or_default(),
153                current_version: current_version.unwrap_or_default(),
154                at: Utc::now(),
155            }));
156    }
157
158    state
159        .engine
160        .event_publisher()
161        .publish(Event::RunCreated(RunCreatedEvent {
162            run_id: new_run.id,
163            workflow_name: new_run.workflow_name.clone(),
164            at: Utc::now(),
165        }));
166
167    Ok((StatusCode::CREATED, ok(RunResponse::from(new_run))))
168}
169
170#[cfg(test)]
171mod tests {
172    use std::collections::HashMap;
173
174    use axum::Router;
175    use axum::body::Body;
176    use axum::http::{Request, StatusCode as HttpStatusCode};
177    use axum::response::Response;
178    use axum::routing::post;
179    use http_body_util::BodyExt;
180    use ironflow_auth::jwt::AccessToken;
181    use ironflow_core::providers::claude::ClaudeCodeProvider;
182    use ironflow_engine::context::WorkflowContext;
183    use ironflow_engine::engine::Engine;
184    use ironflow_engine::handler::{HandlerFuture, WorkflowHandler};
185    use ironflow_engine::notify::Event;
186    use ironflow_store::memory::InMemoryStore;
187    use ironflow_store::models::{NewRun, NewUser, RunActor, RunStatus, TriggerKind};
188    use ironflow_store::store::RunStore;
189    use ironflow_store::user_store::UserStore;
190    use rust_decimal::Decimal;
191    use serde_json::{Value as JsonValue, from_slice, from_value, json};
192    use std::sync::Arc;
193    use tokio::sync::broadcast;
194    use tower::ServiceExt;
195    use uuid::Uuid;
196
197    use super::*;
198    use crate::routes::test_helpers::create_user_auth_header;
199
200    fn test_state(store: Arc<InMemoryStore>) -> AppState {
201        Arc::new(InMemoryStore::new());
202        let provider = Arc::new(ClaudeCodeProvider::new());
203        let engine = Arc::new(Engine::new(store.clone(), provider));
204        let jwt_config = Arc::new(ironflow_auth::jwt::JwtConfig {
205            secret: "test-secret".to_string(),
206            access_token_ttl_secs: 900,
207            refresh_token_ttl_secs: 604800,
208            cookie_domain: None,
209            cookie_secure: false,
210        });
211        let (event_sender, _) = broadcast::channel::<Event>(1);
212        AppState::new(
213            store,
214            engine,
215            jwt_config,
216            "test-worker-token".to_string(),
217            event_sender,
218        )
219    }
220
221    #[tokio::test]
222    async fn retry_failed_run() {
223        let store = Arc::new(InMemoryStore::new());
224        let run = store
225            .create_run(NewRun {
226                created_by: None,
227                workflow_name: "test".to_string(),
228                trigger: TriggerKind::Manual,
229                payload: json!({"key": "value"}),
230                max_retries: 3,
231                handler_version: None,
232                labels: HashMap::new(),
233                scheduled_at: None,
234                idempotency_key: None,
235                concurrency_key: None,
236                priority: 0,
237                concurrency_limits: Vec::new(),
238                max_cost_usd: None,
239                worker_tags: Vec::new(),
240            })
241            .await
242            .unwrap()
243            .into_run();
244
245        store
246            .update_run_status(run.id, RunStatus::Running)
247            .await
248            .unwrap();
249        store
250            .update_run_status(run.id, RunStatus::Failed)
251            .await
252            .unwrap();
253
254        let state = test_state(store.clone());
255        let auth_header = create_user_auth_header(&state, "testuser", true).await;
256        let app = Router::new()
257            .route("/{id}/retry", post(retry_run))
258            .with_state(state);
259
260        let req = Request::builder()
261            .method("POST")
262            .uri(format!("/{}/retry", run.id))
263            .header("content-type", "application/json")
264            .header("authorization", auth_header)
265            .body(Body::from("{}"))
266            .unwrap();
267
268        let resp = app.oneshot(req).await.unwrap();
269        assert_eq!(resp.status(), HttpStatusCode::CREATED);
270
271        let body = resp.into_body().collect().await.unwrap().to_bytes();
272        let json_val: JsonValue = from_slice(&body).unwrap();
273        let new_id: Uuid = from_value(json_val["data"]["id"].clone()).unwrap();
274
275        let new_run = store.get_run(new_id).await.unwrap().unwrap();
276        assert_eq!(new_run.status.state, RunStatus::Pending);
277        assert!(matches!(new_run.trigger, TriggerKind::Retry { .. }));
278    }
279
280    #[tokio::test]
281    async fn retry_inherits_the_original_cost_cap() {
282        let store = Arc::new(InMemoryStore::new());
283        let cap = Decimal::new(250, 2);
284        let run = store
285            .create_run(NewRun {
286                workflow_name: "test".to_string(),
287                trigger: TriggerKind::Manual,
288                payload: json!({}),
289                max_retries: 1,
290                handler_version: None,
291                labels: HashMap::new(),
292                scheduled_at: None,
293                created_by: None,
294                idempotency_key: None,
295                concurrency_key: None,
296                priority: 0,
297                concurrency_limits: Vec::new(),
298                max_cost_usd: Some(cap),
299                worker_tags: Vec::new(),
300            })
301            .await
302            .unwrap()
303            .into_run();
304
305        // A run cancelled for reaching its cap is retryable.
306        store
307            .update_run_status(run.id, RunStatus::Running)
308            .await
309            .unwrap();
310        store
311            .update_run_status(run.id, RunStatus::Cancelled)
312            .await
313            .unwrap();
314
315        let state = test_state(store.clone());
316        let auth_header = create_user_auth_header(&state, "testuser", true).await;
317        let app = Router::new()
318            .route("/{id}/retry", post(retry_run))
319            .with_state(state);
320
321        let req = Request::builder()
322            .method("POST")
323            .uri(format!("/{}/retry", run.id))
324            .header("content-type", "application/json")
325            .header("authorization", auth_header)
326            .body(Body::from("{}"))
327            .unwrap();
328
329        let resp = app.oneshot(req).await.unwrap();
330        assert_eq!(resp.status(), HttpStatusCode::CREATED);
331
332        let body = resp.into_body().collect().await.unwrap().to_bytes();
333        let json_val: JsonValue = from_slice(&body).unwrap();
334        let new_id: Uuid = from_value(json_val["data"]["id"].clone()).unwrap();
335
336        let new_run = store.get_run(new_id).await.unwrap().unwrap();
337        assert_eq!(new_run.max_cost_usd, Some(cap));
338    }
339
340    #[tokio::test]
341    async fn retry_keeps_the_original_priority() {
342        let store = Arc::new(InMemoryStore::new());
343        let run = store
344            .create_run(NewRun {
345                workflow_name: "test".to_string(),
346                trigger: TriggerKind::Manual,
347                payload: json!({}),
348                max_retries: 1,
349                handler_version: None,
350                labels: HashMap::new(),
351                scheduled_at: None,
352                created_by: None,
353                idempotency_key: None,
354                concurrency_key: None,
355                priority: 42,
356                concurrency_limits: Vec::new(),
357                max_cost_usd: None,
358                worker_tags: Vec::new(),
359            })
360            .await
361            .unwrap()
362            .into_run();
363        store
364            .update_run_status(run.id, RunStatus::Running)
365            .await
366            .unwrap();
367        store
368            .update_run_status(run.id, RunStatus::Failed)
369            .await
370            .unwrap();
371
372        let state = test_state(store.clone());
373        let auth_header = create_user_auth_header(&state, "testuser", true).await;
374        let app = Router::new()
375            .route("/{id}/retry", post(retry_run))
376            .with_state(state);
377
378        let req = Request::builder()
379            .method("POST")
380            .uri(format!("/{}/retry", run.id))
381            .header("content-type", "application/json")
382            .header("authorization", auth_header)
383            .body(Body::from("{}"))
384            .unwrap();
385
386        let resp = app.oneshot(req).await.unwrap();
387        assert_eq!(resp.status(), HttpStatusCode::CREATED);
388
389        let body = resp.into_body().collect().await.unwrap().to_bytes();
390        let json_val: JsonValue = from_slice(&body).unwrap();
391        assert_eq!(json_val["data"]["priority"], 42);
392        let new_id: Uuid = from_value(json_val["data"]["id"].clone()).unwrap();
393
394        let new_run = store.get_run(new_id).await.unwrap().unwrap();
395        assert_eq!(new_run.priority, 42);
396    }
397
398    #[tokio::test]
399    async fn retry_keeps_the_original_worker_tags() {
400        let store = Arc::new(InMemoryStore::new());
401        let tags = vec!["gpu".to_string(), "region:eu".to_string()];
402        let run = store
403            .create_run(NewRun {
404                workflow_name: "test".to_string(),
405                trigger: TriggerKind::Manual,
406                payload: json!({}),
407                max_retries: 0,
408                handler_version: None,
409                labels: HashMap::new(),
410                scheduled_at: None,
411                created_by: None,
412                idempotency_key: None,
413                concurrency_key: None,
414                priority: 0,
415                concurrency_limits: Vec::new(),
416                max_cost_usd: None,
417                worker_tags: tags.clone(),
418            })
419            .await
420            .unwrap()
421            .into_run();
422        store
423            .update_run_status(run.id, RunStatus::Running)
424            .await
425            .unwrap();
426        store
427            .update_run_status(run.id, RunStatus::Failed)
428            .await
429            .unwrap();
430
431        let state = test_state(store.clone());
432        let auth_header = create_user_auth_header(&state, "testuser", true).await;
433        let app = Router::new()
434            .route("/{id}/retry", post(retry_run))
435            .with_state(state);
436
437        let req = Request::builder()
438            .method("POST")
439            .uri(format!("/{}/retry", run.id))
440            .header("content-type", "application/json")
441            .header("authorization", auth_header)
442            .body(Body::from("{}"))
443            .unwrap();
444
445        let resp = app.oneshot(req).await.unwrap();
446        assert_eq!(resp.status(), HttpStatusCode::CREATED);
447
448        let body = resp.into_body().collect().await.unwrap().to_bytes();
449        let json_val: JsonValue = from_slice(&body).unwrap();
450        assert_eq!(json_val["data"]["worker_tags"], json!(["gpu", "region:eu"]));
451        let new_id: Uuid = from_value(json_val["data"]["id"].clone()).unwrap();
452
453        let new_run = store.get_run(new_id).await.unwrap().unwrap();
454        assert_eq!(new_run.worker_tags, tags);
455    }
456
457    #[tokio::test]
458    async fn retry_pending_run_returns_400() {
459        let store = Arc::new(InMemoryStore::new());
460        let run = store
461            .create_run(NewRun {
462                created_by: None,
463                workflow_name: "test".to_string(),
464                trigger: TriggerKind::Manual,
465                payload: json!({}),
466                max_retries: 0,
467                handler_version: None,
468                labels: HashMap::new(),
469                scheduled_at: None,
470                idempotency_key: None,
471                concurrency_key: None,
472                priority: 0,
473                concurrency_limits: Vec::new(),
474                max_cost_usd: None,
475                worker_tags: Vec::new(),
476            })
477            .await
478            .unwrap()
479            .into_run();
480
481        let state = test_state(store);
482        let auth_header = create_user_auth_header(&state, "testuser", true).await;
483        let app = Router::new()
484            .route("/{id}/retry", post(retry_run))
485            .with_state(state);
486
487        let req = Request::builder()
488            .method("POST")
489            .uri(format!("/{}/retry", run.id))
490            .header("content-type", "application/json")
491            .header("authorization", auth_header)
492            .body(Body::from("{}"))
493            .unwrap();
494
495        let resp = app.oneshot(req).await.unwrap();
496        assert_eq!(resp.status(), HttpStatusCode::BAD_REQUEST);
497    }
498
499    #[tokio::test]
500    async fn retry_completed_run_returns_400() {
501        let store = Arc::new(InMemoryStore::new());
502        let run = store
503            .create_run(NewRun {
504                created_by: None,
505                workflow_name: "test".to_string(),
506                trigger: TriggerKind::Manual,
507                payload: json!({}),
508                max_retries: 0,
509                handler_version: None,
510                labels: HashMap::new(),
511                scheduled_at: None,
512                idempotency_key: None,
513                concurrency_key: None,
514                priority: 0,
515                concurrency_limits: Vec::new(),
516                max_cost_usd: None,
517                worker_tags: Vec::new(),
518            })
519            .await
520            .unwrap()
521            .into_run();
522
523        store
524            .update_run_status(run.id, RunStatus::Running)
525            .await
526            .unwrap();
527        store
528            .update_run_status(run.id, RunStatus::Completed)
529            .await
530            .unwrap();
531
532        let state = test_state(store);
533        let auth_header = create_user_auth_header(&state, "testuser", true).await;
534        let app = Router::new()
535            .route("/{id}/retry", post(retry_run))
536            .with_state(state);
537
538        let req = Request::builder()
539            .method("POST")
540            .uri(format!("/{}/retry", run.id))
541            .header("content-type", "application/json")
542            .header("authorization", auth_header)
543            .body(Body::from("{}"))
544            .unwrap();
545
546        let resp = app.oneshot(req).await.unwrap();
547        assert_eq!(resp.status(), HttpStatusCode::BAD_REQUEST);
548    }
549
550    #[tokio::test]
551    async fn retry_running_run_returns_400() {
552        let store = Arc::new(InMemoryStore::new());
553        let run = store
554            .create_run(NewRun {
555                created_by: None,
556                workflow_name: "test".to_string(),
557                trigger: TriggerKind::Manual,
558                payload: json!({}),
559                max_retries: 0,
560                handler_version: None,
561                labels: HashMap::new(),
562                scheduled_at: None,
563                idempotency_key: None,
564                concurrency_key: None,
565                priority: 0,
566                concurrency_limits: Vec::new(),
567                max_cost_usd: None,
568                worker_tags: Vec::new(),
569            })
570            .await
571            .unwrap()
572            .into_run();
573
574        store
575            .update_run_status(run.id, RunStatus::Running)
576            .await
577            .unwrap();
578
579        let state = test_state(store);
580        let auth_header = create_user_auth_header(&state, "testuser", true).await;
581        let app = Router::new()
582            .route("/{id}/retry", post(retry_run))
583            .with_state(state);
584
585        let req = Request::builder()
586            .method("POST")
587            .uri(format!("/{}/retry", run.id))
588            .header("content-type", "application/json")
589            .header("authorization", auth_header)
590            .body(Body::from("{}"))
591            .unwrap();
592
593        let resp = app.oneshot(req).await.unwrap();
594        assert_eq!(resp.status(), HttpStatusCode::BAD_REQUEST);
595    }
596
597    #[tokio::test]
598    async fn retry_run_awaiting_automatic_retry_returns_409() {
599        let store = Arc::new(InMemoryStore::new());
600        let run = store
601            .create_run(NewRun {
602                created_by: None,
603                workflow_name: "test".to_string(),
604                trigger: TriggerKind::Manual,
605                payload: json!({}),
606                max_retries: 3,
607                handler_version: None,
608                labels: HashMap::new(),
609                scheduled_at: None,
610                idempotency_key: None,
611                concurrency_key: None,
612                priority: 0,
613                concurrency_limits: Vec::new(),
614                max_cost_usd: None,
615                worker_tags: Vec::new(),
616            })
617            .await
618            .unwrap()
619            .into_run();
620
621        store
622            .update_run_status(run.id, RunStatus::Running)
623            .await
624            .unwrap();
625        store
626            .update_run_status(run.id, RunStatus::Retrying)
627            .await
628            .unwrap();
629
630        let state = test_state(store.clone());
631        let auth_header = create_user_auth_header(&state, "testuser", true).await;
632        let app = Router::new()
633            .route("/{id}/retry", post(retry_run))
634            .with_state(state);
635
636        let req = Request::builder()
637            .method("POST")
638            .uri(format!("/{}/retry", run.id))
639            .header("content-type", "application/json")
640            .header("authorization", auth_header)
641            .body(Body::from("{}"))
642            .unwrap();
643
644        let resp = app.oneshot(req).await.unwrap();
645        assert_eq!(resp.status(), HttpStatusCode::CONFLICT);
646
647        // No duplicate run was created.
648        let runs = store
649            .list_runs(ironflow_store::models::RunFilter::default(), 1, 50)
650            .await
651            .unwrap();
652        assert_eq!(runs.total, 1);
653    }
654
655    #[tokio::test]
656    async fn retry_cancelled_run_is_allowed() {
657        let store = Arc::new(InMemoryStore::new());
658        let run = store
659            .create_run(NewRun {
660                created_by: None,
661                workflow_name: "test".to_string(),
662                trigger: TriggerKind::Manual,
663                payload: json!({}),
664                max_retries: 0,
665                handler_version: None,
666                labels: HashMap::new(),
667                scheduled_at: None,
668                idempotency_key: None,
669                concurrency_key: None,
670                priority: 0,
671                concurrency_limits: Vec::new(),
672                max_cost_usd: None,
673                worker_tags: Vec::new(),
674            })
675            .await
676            .unwrap()
677            .into_run();
678
679        store
680            .update_run_status(run.id, RunStatus::Cancelled)
681            .await
682            .unwrap();
683
684        let state = test_state(store);
685        let auth_header = create_user_auth_header(&state, "testuser", true).await;
686        let app = Router::new()
687            .route("/{id}/retry", post(retry_run))
688            .with_state(state);
689
690        let req = Request::builder()
691            .method("POST")
692            .uri(format!("/{}/retry", run.id))
693            .header("content-type", "application/json")
694            .header("authorization", auth_header)
695            .body(Body::from("{}"))
696            .unwrap();
697
698        let resp = app.oneshot(req).await.unwrap();
699        assert_eq!(resp.status(), HttpStatusCode::CREATED);
700    }
701
702    #[tokio::test]
703    async fn retry_nonexistent_run_returns_404() {
704        let store = Arc::new(InMemoryStore::new());
705        let state = test_state(store);
706        let auth_header = create_user_auth_header(&state, "testuser", true).await;
707        let app = Router::new()
708            .route("/{id}/retry", post(retry_run))
709            .with_state(state);
710
711        let req = Request::builder()
712            .method("POST")
713            .uri(format!("/{}/retry", Uuid::now_v7()))
714            .header("content-type", "application/json")
715            .header("authorization", auth_header)
716            .body(Body::from("{}"))
717            .unwrap();
718
719        let resp = app.oneshot(req).await.unwrap();
720        assert_eq!(resp.status(), HttpStatusCode::NOT_FOUND);
721    }
722
723    // ---- created_by ----
724
725    #[tokio::test]
726    async fn retry_attributes_the_new_run_to_the_caller_not_the_original_author() {
727        let store = Arc::new(InMemoryStore::new());
728        let original_author = store
729            .create_user(NewUser {
730                email: "alice@example.com".to_string(),
731                username: "alice".to_string(),
732                password_hash: "hash".to_string(),
733                is_admin: Some(true),
734            })
735            .await
736            .unwrap();
737        let retrying_user = store
738            .create_user(NewUser {
739                email: "bob@example.com".to_string(),
740                username: "bob".to_string(),
741                password_hash: "hash".to_string(),
742                is_admin: Some(true),
743            })
744            .await
745            .unwrap();
746
747        let run = store
748            .create_run(NewRun {
749                workflow_name: "test".to_string(),
750                trigger: TriggerKind::Api,
751                payload: json!({}),
752                max_retries: 3,
753                handler_version: None,
754                labels: HashMap::new(),
755                scheduled_at: None,
756                created_by: Some(RunActor::User {
757                    user_id: original_author.id,
758                }),
759                idempotency_key: None,
760                concurrency_key: None,
761                priority: 0,
762                concurrency_limits: Vec::new(),
763                max_cost_usd: None,
764                worker_tags: Vec::new(),
765            })
766            .await
767            .unwrap()
768            .into_run();
769
770        store
771            .update_run_status(run.id, RunStatus::Running)
772            .await
773            .unwrap();
774        store
775            .update_run_status(run.id, RunStatus::Failed)
776            .await
777            .unwrap();
778
779        let state = test_state(store.clone());
780        let token =
781            AccessToken::for_user(retrying_user.id, "bob", true, &state.jwt_config).unwrap();
782        let app = Router::new()
783            .route("/{id}/retry", post(retry_run))
784            .with_state(state);
785
786        let req = Request::builder()
787            .method("POST")
788            .uri(format!("/{}/retry", run.id))
789            .header("content-type", "application/json")
790            .header("authorization", format!("Bearer {}", token.0))
791            .body(Body::from("{}"))
792            .unwrap();
793
794        let resp = app.oneshot(req).await.unwrap();
795        assert_eq!(resp.status(), HttpStatusCode::CREATED);
796
797        let body = resp.into_body().collect().await.unwrap().to_bytes();
798        let json_val: JsonValue = from_slice(&body).unwrap();
799        assert_eq!(json_val["data"]["created_by"]["kind"], "user");
800        assert_eq!(
801            json_val["data"]["created_by"]["id"],
802            retrying_user.id.to_string()
803        );
804        assert_eq!(json_val["data"]["created_by"]["label"], "bob");
805    }
806
807    // ---- Idempotency-Key ----
808
809    #[tokio::test]
810    async fn retry_does_not_inherit_the_idempotency_key() {
811        let store = Arc::new(InMemoryStore::new());
812        let run = store
813            .create_run(NewRun {
814                workflow_name: "test".to_string(),
815                trigger: TriggerKind::Api,
816                payload: json!({"key": "value"}),
817                max_retries: 3,
818                handler_version: None,
819                labels: HashMap::new(),
820                scheduled_at: None,
821                created_by: None,
822                idempotency_key: Some("github:abc-123".to_string()),
823                concurrency_key: None,
824                priority: 0,
825                concurrency_limits: Vec::new(),
826                max_cost_usd: None,
827                worker_tags: Vec::new(),
828            })
829            .await
830            .unwrap()
831            .into_run();
832
833        store
834            .update_run_status(run.id, RunStatus::Running)
835            .await
836            .unwrap();
837        store
838            .update_run_status(run.id, RunStatus::Failed)
839            .await
840            .unwrap();
841
842        let state = test_state(store.clone());
843        let auth_header = create_user_auth_header(&state, "testuser", true).await;
844        let app = Router::new()
845            .route("/{id}/retry", post(retry_run))
846            .with_state(state);
847
848        let req = Request::builder()
849            .method("POST")
850            .uri(format!("/{}/retry", run.id))
851            .header("content-type", "application/json")
852            .header("authorization", auth_header)
853            .body(Body::from("{}"))
854            .unwrap();
855
856        let resp = app.oneshot(req).await.unwrap();
857        assert_eq!(resp.status(), HttpStatusCode::CREATED);
858
859        let body = resp.into_body().collect().await.unwrap().to_bytes();
860        let json_val: JsonValue = from_slice(&body).unwrap();
861        assert!(
862            json_val["data"].get("idempotency_key").is_none(),
863            "the retry run must not carry the original key"
864        );
865
866        // The original key still resolves to the original run.
867        let bound = store
868            .find_run_by_idempotency_key("github:abc-123")
869            .await
870            .unwrap()
871            .expect("key still bound");
872        assert_eq!(bound.id, run.id);
873    }
874
875    // ---- Handler version compatibility ----
876
877    struct V2Handler;
878    impl WorkflowHandler for V2Handler {
879        fn name(&self) -> &str {
880            "versioned-wf"
881        }
882        fn version(&self) -> Option<&str> {
883            Some("2.0.0")
884        }
885        fn execute<'a>(&'a self, _ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
886            Box::pin(async { Ok(()) })
887        }
888    }
889
890    struct V2CompatHandler;
891    impl WorkflowHandler for V2CompatHandler {
892        fn name(&self) -> &str {
893            "compat-wf"
894        }
895        fn version(&self) -> Option<&str> {
896            Some("2.0.0")
897        }
898        fn compatible_versions(&self) -> &[&str] {
899            &["1.0.0"]
900        }
901        fn execute<'a>(&'a self, _ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
902            Box::pin(async { Ok(()) })
903        }
904    }
905
906    fn test_state_with_handlers(store: Arc<InMemoryStore>) -> AppState {
907        let provider = Arc::new(ClaudeCodeProvider::new());
908        let mut engine = Engine::new(store.clone(), provider);
909        engine.register(V2Handler).unwrap();
910        engine.register(V2CompatHandler).unwrap();
911        let jwt_config = Arc::new(ironflow_auth::jwt::JwtConfig {
912            secret: "test-secret".to_string(),
913            access_token_ttl_secs: 900,
914            refresh_token_ttl_secs: 604800,
915            cookie_domain: None,
916            cookie_secure: false,
917        });
918        let (event_sender, _) = broadcast::channel::<Event>(1);
919        AppState::new(
920            store,
921            Arc::new(engine),
922            jwt_config,
923            "test-worker-token".to_string(),
924            event_sender,
925        )
926    }
927
928    async fn create_failed_run(
929        store: &Arc<InMemoryStore>,
930        workflow_name: &str,
931        handler_version: Option<&str>,
932    ) -> ironflow_store::models::Run {
933        let run = store
934            .create_run(NewRun {
935                created_by: None,
936                workflow_name: workflow_name.to_string(),
937                trigger: TriggerKind::Manual,
938                payload: json!({}),
939                max_retries: 0,
940                handler_version: handler_version.map(str::to_string),
941                labels: HashMap::new(),
942                scheduled_at: None,
943                idempotency_key: None,
944                concurrency_key: None,
945                priority: 0,
946                concurrency_limits: Vec::new(),
947                max_cost_usd: None,
948                worker_tags: Vec::new(),
949            })
950            .await
951            .unwrap()
952            .into_run();
953
954        store
955            .update_run_status(run.id, RunStatus::Running)
956            .await
957            .unwrap();
958        store
959            .update_run_status(run.id, RunStatus::Failed)
960            .await
961            .unwrap();
962        run
963    }
964
965    #[tokio::test]
966    async fn retry_same_version_succeeds() {
967        let store = Arc::new(InMemoryStore::new());
968        let run = create_failed_run(&store, "versioned-wf", Some("2.0.0")).await;
969
970        let state = test_state_with_handlers(store);
971        let auth_header = create_user_auth_header(&state, "testuser", true).await;
972        let app = Router::new()
973            .route("/{id}/retry", post(retry_run))
974            .with_state(state);
975
976        let req = Request::builder()
977            .method("POST")
978            .uri(format!("/{}/retry", run.id))
979            .header("content-type", "application/json")
980            .header("authorization", auth_header)
981            .body(Body::from("{}"))
982            .unwrap();
983
984        let resp = app.oneshot(req).await.unwrap();
985        assert_eq!(resp.status(), HttpStatusCode::CREATED);
986    }
987
988    #[tokio::test]
989    async fn retry_version_mismatch_returns_409() {
990        let store = Arc::new(InMemoryStore::new());
991        let run = create_failed_run(&store, "versioned-wf", Some("1.0.0")).await;
992
993        let state = test_state_with_handlers(store);
994        let auth_header = create_user_auth_header(&state, "testuser", true).await;
995        let app = Router::new()
996            .route("/{id}/retry", post(retry_run))
997            .with_state(state);
998
999        let req = Request::builder()
1000            .method("POST")
1001            .uri(format!("/{}/retry", run.id))
1002            .header("content-type", "application/json")
1003            .header("authorization", auth_header)
1004            .body(Body::from("{}"))
1005            .unwrap();
1006
1007        let resp = app.oneshot(req).await.unwrap();
1008        assert_eq!(resp.status(), HttpStatusCode::CONFLICT);
1009
1010        let body = resp.into_body().collect().await.unwrap().to_bytes();
1011        let json_val: JsonValue = from_slice(&body).unwrap();
1012        let msg = json_val["error"]["message"].as_str().unwrap();
1013        assert!(msg.contains("HANDLER_VERSION_MISMATCH"));
1014        assert!(msg.contains("2.0.0"));
1015        assert!(msg.contains("1.0.0"));
1016    }
1017
1018    #[tokio::test]
1019    async fn retry_version_mismatch_with_force_succeeds() {
1020        let store = Arc::new(InMemoryStore::new());
1021        let run = create_failed_run(&store, "versioned-wf", Some("1.0.0")).await;
1022
1023        let state = test_state_with_handlers(store.clone());
1024        let auth_header = create_user_auth_header(&state, "testuser", true).await;
1025        let app = Router::new()
1026            .route("/{id}/retry", post(retry_run))
1027            .with_state(state);
1028
1029        let req = Request::builder()
1030            .method("POST")
1031            .uri(format!("/{}/retry?force=true", run.id))
1032            .header("content-type", "application/json")
1033            .header("authorization", auth_header)
1034            .body(Body::from("{}"))
1035            .unwrap();
1036
1037        let resp = app.oneshot(req).await.unwrap();
1038        assert_eq!(resp.status(), HttpStatusCode::CREATED);
1039
1040        let body = resp.into_body().collect().await.unwrap().to_bytes();
1041        let json_val: JsonValue = from_slice(&body).unwrap();
1042        let new_id: Uuid = from_value(json_val["data"]["id"].clone()).unwrap();
1043        let new_run = store.get_run(new_id).await.unwrap().unwrap();
1044        assert_eq!(new_run.handler_version, Some("2.0.0".to_string()));
1045    }
1046
1047    #[tokio::test]
1048    async fn retry_compatible_version_succeeds() {
1049        let store = Arc::new(InMemoryStore::new());
1050        let run = create_failed_run(&store, "compat-wf", Some("1.0.0")).await;
1051
1052        let state = test_state_with_handlers(store);
1053        let auth_header = create_user_auth_header(&state, "testuser", true).await;
1054        let app = Router::new()
1055            .route("/{id}/retry", post(retry_run))
1056            .with_state(state);
1057
1058        let req = Request::builder()
1059            .method("POST")
1060            .uri(format!("/{}/retry", run.id))
1061            .header("content-type", "application/json")
1062            .header("authorization", auth_header)
1063            .body(Body::from("{}"))
1064            .unwrap();
1065
1066        let resp = app.oneshot(req).await.unwrap();
1067        assert_eq!(resp.status(), HttpStatusCode::CREATED);
1068    }
1069
1070    #[tokio::test]
1071    async fn retry_null_version_is_compatible() {
1072        let store = Arc::new(InMemoryStore::new());
1073        let run = create_failed_run(&store, "versioned-wf", None).await;
1074
1075        let state = test_state_with_handlers(store);
1076        let auth_header = create_user_auth_header(&state, "testuser", true).await;
1077        let app = Router::new()
1078            .route("/{id}/retry", post(retry_run))
1079            .with_state(state);
1080
1081        let req = Request::builder()
1082            .method("POST")
1083            .uri(format!("/{}/retry", run.id))
1084            .header("content-type", "application/json")
1085            .header("authorization", auth_header)
1086            .body(Body::from("{}"))
1087            .unwrap();
1088
1089        let resp = app.oneshot(req).await.unwrap();
1090        assert_eq!(resp.status(), HttpStatusCode::CREATED);
1091    }
1092
1093    #[tokio::test]
1094    async fn retry_writes_current_handler_version() {
1095        let store = Arc::new(InMemoryStore::new());
1096        let run = create_failed_run(&store, "versioned-wf", Some("2.0.0")).await;
1097
1098        let state = test_state_with_handlers(store.clone());
1099        let auth_header = create_user_auth_header(&state, "testuser", true).await;
1100        let app = Router::new()
1101            .route("/{id}/retry", post(retry_run))
1102            .with_state(state);
1103
1104        let req = Request::builder()
1105            .method("POST")
1106            .uri(format!("/{}/retry", run.id))
1107            .header("content-type", "application/json")
1108            .header("authorization", auth_header)
1109            .body(Body::from("{}"))
1110            .unwrap();
1111
1112        let resp = app.oneshot(req).await.unwrap();
1113        assert_eq!(resp.status(), HttpStatusCode::CREATED);
1114
1115        let body = resp.into_body().collect().await.unwrap().to_bytes();
1116        let json_val: JsonValue = from_slice(&body).unwrap();
1117        let new_id: Uuid = from_value(json_val["data"]["id"].clone()).unwrap();
1118        let new_run = store.get_run(new_id).await.unwrap().unwrap();
1119        assert_eq!(new_run.handler_version, Some("2.0.0".to_string()));
1120    }
1121    /// A cancelled run holding `key`, ready to be retried.
1122    async fn cancelled_keyed_run(store: &InMemoryStore, key: &str) -> Uuid {
1123        let run = store
1124            .create_run(NewRun {
1125                workflow_name: "test".to_string(),
1126                trigger: TriggerKind::Manual,
1127                payload: json!({}),
1128                max_retries: 0,
1129                handler_version: None,
1130                labels: HashMap::new(),
1131                scheduled_at: None,
1132                created_by: None,
1133                idempotency_key: None,
1134                concurrency_key: Some(key.to_string()),
1135                priority: 0,
1136                concurrency_limits: Vec::new(),
1137                max_cost_usd: None,
1138                worker_tags: Vec::new(),
1139            })
1140            .await
1141            .unwrap()
1142            .into_run();
1143        store
1144            .update_run_status(run.id, RunStatus::Cancelled)
1145            .await
1146            .unwrap();
1147        run.id
1148    }
1149
1150    async fn post_retry(state: AppState, run_id: Uuid) -> Response {
1151        let auth_header = create_user_auth_header(&state, "testuser", true).await;
1152        let app = Router::new()
1153            .route("/{id}/retry", post(retry_run))
1154            .with_state(state);
1155        let req = Request::builder()
1156            .method("POST")
1157            .uri(format!("/{run_id}/retry"))
1158            .header("content-type", "application/json")
1159            .header("authorization", auth_header)
1160            .body(Body::from("{}"))
1161            .unwrap();
1162        app.oneshot(req).await.unwrap()
1163    }
1164
1165    #[tokio::test]
1166    async fn retry_inherits_the_original_concurrency_key() {
1167        let store = Arc::new(InMemoryStore::new());
1168        let original = cancelled_keyed_run(&store, "issue:12").await;
1169
1170        let resp = post_retry(test_state(store.clone()), original).await;
1171        assert_eq!(resp.status(), HttpStatusCode::CREATED);
1172
1173        let body = resp.into_body().collect().await.unwrap().to_bytes();
1174        let json_val: JsonValue = from_slice(&body).unwrap();
1175        assert_eq!(json_val["data"]["concurrency_key"], "issue:12");
1176        let new_id: Uuid = from_value(json_val["data"]["id"].clone()).unwrap();
1177        let new_run = store.get_run(new_id).await.unwrap().unwrap();
1178        assert_eq!(new_run.concurrency_key.as_deref(), Some("issue:12"));
1179    }
1180
1181    #[tokio::test]
1182    async fn retry_returns_409_when_the_concurrency_key_was_taken_since() {
1183        let store = Arc::new(InMemoryStore::new());
1184        let original = cancelled_keyed_run(&store, "issue:12").await;
1185        let holder = store
1186            .create_run(NewRun {
1187                workflow_name: "test".to_string(),
1188                trigger: TriggerKind::Manual,
1189                payload: json!({}),
1190                max_retries: 0,
1191                handler_version: None,
1192                labels: HashMap::new(),
1193                scheduled_at: None,
1194                created_by: None,
1195                idempotency_key: None,
1196                concurrency_key: Some("issue:12".to_string()),
1197                priority: 0,
1198                concurrency_limits: Vec::new(),
1199                max_cost_usd: None,
1200                worker_tags: Vec::new(),
1201            })
1202            .await
1203            .unwrap()
1204            .into_run();
1205
1206        let resp = post_retry(test_state(store.clone()), original).await;
1207        assert_eq!(resp.status(), HttpStatusCode::CONFLICT);
1208
1209        let body = resp.into_body().collect().await.unwrap().to_bytes();
1210        let json_val: JsonValue = from_slice(&body).unwrap();
1211        assert_eq!(json_val["error"]["code"], "CONCURRENCY_CONFLICT");
1212        assert_eq!(json_val["error"]["details"]["run_id"], json!(holder.id));
1213    }
1214}
ironflow_api/routes/retry_run.rs

ironflow_api/routes/
retry_run.rs
