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

ironflow_api/routes/
retry_run.rs
