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

ironflow_api/routes/
retry_run.rs
