diff --git a/src/context/state.rs b/src/context/state.rs index 22aca311..768e4da6 100644 --- a/src/context/state.rs +++ b/src/context/state.rs @@ -48,6 +48,14 @@ impl JobState { pub fn can_transition_to(&self, target: JobState) -> bool { use JobState::*; + // Allow idempotent Completed -> Completed transition. + // Both the execution loop and the worker wrapper may race to mark a + // job complete; the second call should be a harmless no-op rather + // than an error that masks the successful completion. + if matches!((self, target), (Completed, Completed)) { + return true; + } + matches!( (self, target), // From Pending @@ -238,6 +246,18 @@ impl JobContext { )); } + // Idempotent: already in the target state, skip recording a duplicate + // transition. This handles the Completed -> Completed race between + // execution_loop and the worker wrapper. + if self.state == new_state { + tracing::debug!( + job_id = %self.job_id, + state = %self.state, + "idempotent state transition (already in target state), skipping" + ); + return Ok(()); + } + let transition = StateTransition { from: self.state, to: new_state, @@ -340,6 +360,45 @@ mod tests { assert!(!JobState::Accepted.can_transition_to(JobState::InProgress)); } + #[test] + fn test_completed_to_completed_is_idempotent() { + // Regression test for the race condition where both execution_loop + // and the worker wrapper call mark_completed(). The second call + // must succeed without error and must not record a duplicate + // transition. + let mut ctx = JobContext::new("Test", "Idempotent completion test"); + ctx.transition_to(JobState::InProgress, None).unwrap(); + ctx.transition_to(JobState::Completed, Some("first".into())) + .unwrap(); + assert_eq!(ctx.state, JobState::Completed); + let transitions_before = ctx.transitions.len(); + + // Second Completed -> Completed must be a no-op + let result = ctx.transition_to(JobState::Completed, Some("duplicate".into())); + assert!( + result.is_ok(), + "Completed -> Completed should be idempotent" + ); + assert_eq!(ctx.state, JobState::Completed); + assert_eq!( + ctx.transitions.len(), + transitions_before, + "idempotent transition should not record a new history entry" + ); + } + + #[test] + fn test_other_self_transitions_still_rejected() { + // Ensure we only allow Completed -> Completed, not arbitrary X -> X. + assert!(!JobState::Pending.can_transition_to(JobState::Pending)); + assert!(!JobState::InProgress.can_transition_to(JobState::InProgress)); + assert!(!JobState::Failed.can_transition_to(JobState::Failed)); + assert!(!JobState::Stuck.can_transition_to(JobState::Stuck)); + assert!(!JobState::Submitted.can_transition_to(JobState::Submitted)); + assert!(!JobState::Accepted.can_transition_to(JobState::Accepted)); + assert!(!JobState::Cancelled.can_transition_to(JobState::Cancelled)); + } + #[test] fn test_terminal_states() { assert!(JobState::Accepted.is_terminal()); diff --git a/src/worker/job.rs b/src/worker/job.rs index c6c555db..0f0e969e 100644 --- a/src/worker/job.rs +++ b/src/worker/job.rs @@ -1591,7 +1591,7 @@ mod tests { } #[tokio::test] - async fn test_mark_completed_twice_returns_error() { + async fn test_mark_completed_twice_is_idempotent() { let worker = make_worker(vec![]).await; worker @@ -1612,11 +1612,22 @@ mod tests { .unwrap(); assert_eq!(ctx.state, JobState::Completed); + // Second mark_completed should succeed (idempotent) rather than + // erroring, matching the fix for the execution_loop / worker wrapper + // race condition. let result = worker.mark_completed().await; assert!( - result.is_err(), - "Completed → Completed transition should be rejected by state machine" + result.is_ok(), + "Completed -> Completed transition should be idempotent" ); + + // State should still be Completed + let ctx = worker + .context_manager() + .get_context(worker.job_id) + .await + .unwrap(); + assert_eq!(ctx.state, JobState::Completed); } /// Build a Worker with the given approval context.