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

ironflow_api/routes/
retry_run.rs
