diff --git a/radis/core/models.py b/radis/core/models.py index 0ee55867..b7851036 100644 --- a/radis/core/models.py +++ b/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() diff --git a/radis/core/tests/test_models.py b/radis/core/tests/test_models.py index c0b702bb..bfae1758 100644 --- a/radis/core/tests/test_models.py +++ b/radis/core/tests/test_models.py @@ -636,3 +636,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