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

ironflow_api/routes/
retry_run.rs
