fix(jobs): make completed->completed transition idempotent to prevent race errors (#1068)

* fix(jobs): make completed->completed transition idempotent to prevent race errors

Both execution_loop and the worker wrapper in execute() can race to call
mark_completed(). Previously the second call hit "Cannot transition from
completed to completed" and errored the job despite successful completion.

This narrowly allows only the Completed->Completed self-transition as
idempotent (early return with debug log, no duplicate history entry).
All other self-transitions remain rejected to preserve state machine
strictness.

Co-Authored-By: Claude Opus 4.6 <[email protected]>

* style: fix assert! formatting in idempotent completion test

Co-Authored-By: Claude Opus 4.6 <[email protected]>

---------

Co-authored-by: Claude Opus 4.6 <[email protected]>

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());

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.