fix(core): job canceled during a running task no longer ends as SUCCESS

radis/core/models.py

@@ -83,6 +83,11 @@ def update_job_state(self) -> bool:
             is a continuous job there could be added new tasks later on.
         """
 
+        # The caller may hold an instance loaded long before this call (a worker
+        # keeps the job in memory across a whole task run, during which the user
+        # may cancel). Decide on the current DB status, not the stale one.
+        self.refresh_from_db(fields=["status"])
+
         if not self.tasks.exists():
             self.status = AnalysisJob.Status.CANCELED
             self.message = "No tasks remaining."
@@ -130,7 +135,21 @@ def update_job_state(self) -> bool:
             raise AssertionError(f"Invalid task status of {self}.")
 
         self.ended_at = timezone.now()
-        self.save()
+        # Guarded write: the refresh above and this save are not atomic, so a
+        # cancel can still land in between. A final status must never replace
+        # a cancel; zero updated rows means a concurrent cancel won.
+        updated = (
+            type(self)
+            .objects.filter(pk=self.pk)
+            .exclude(status__in=[AnalysisJob.Status.CANCELING, AnalysisJob.Status.CANCELED])
+            .update(status=self.status, message=self.message, ended_at=self.ended_at)
+        )
+        if updated == 0:
+            type(self).objects.filter(pk=self.pk, status=AnalysisJob.Status.CANCELING).update(
+                status=AnalysisJob.Status.CANCELED, ended_at=self.ended_at
+            )
+            self.status = AnalysisJob.Status.CANCELED
+            return False
 
         if self.send_finished_mail:
             self._send_job_finished_mail()

radis/core/tests/test_models.py

@@ -599,3 +599,48 @@ def test_task_timestamps_behavior(self):
         task.created_at = None
         with pytest.raises(Exception):
             task.save()
+
+
+@pytest.mark.django_db
+def test_update_job_state_on_stale_instance_respects_concurrent_cancel():
+    """A worker holds a job instance loaded before the user canceled (it kept it
+    in memory across a long task run). When the last task finishes and the worker
+    evaluates the final state, the CANCELING recorded in the DB must win: the job
+    ends CANCELED, never SUCCESS."""
+    user = UserFactory.create()
+    job = ExtractionJobFactory.create(owner=user, status=AnalysisJob.Status.IN_PROGRESS)
+    ExtractionTaskFactory.create(job=job, status=AnalysisTask.Status.SUCCESS)
+    ExtractionTaskFactory.create(job=job, status=AnalysisTask.Status.CANCELED)
+
+    # The cancel view writes CANCELING directly to the DB; the worker's `job`
+    # instance still holds IN_PROGRESS.
+    type(job).objects.filter(pk=job.pk).update(status=AnalysisJob.Status.CANCELING)
+
+    job.update_job_state()
+
+    job.refresh_from_db()
+    assert job.status == AnalysisJob.Status.CANCELED
+
+
+@pytest.mark.django_db
+def test_final_state_write_guarded_against_cancel_landing_after_refresh(monkeypatch):
+    """Even a cancel that lands in the sliver between the status refresh and the
+    final-state write must win over SUCCESS — the write has to be guarded, not
+    just preceded by a refresh."""
+    user = UserFactory.create()
+    job = ExtractionJobFactory.create(owner=user, status=AnalysisJob.Status.IN_PROGRESS)
+    ExtractionTaskFactory.create(job=job, status=AnalysisTask.Status.SUCCESS)
+
+    real_refresh = job.refresh_from_db
+
+    def refresh_then_cancel(*args, **kwargs):
+        real_refresh(*args, **kwargs)
+        # The cancel view writes CANCELING right after the worker refreshed.
+        type(job).objects.filter(pk=job.pk).update(status=AnalysisJob.Status.CANCELING)
+
+    monkeypatch.setattr(job, "refresh_from_db", refresh_then_cancel)
+
+    job.update_job_state()
+
+    db_status = type(job).objects.values_list("status", flat=True).get(pk=job.pk)
+    assert db_status == AnalysisJob.Status.CANCELED