feat(mass_transfer): resumable automatic retries (durable volume progress)

adit/core/processors.py

@@ -30,6 +30,10 @@ class DicomTaskProcessor(abc.ABC):
     dicom_task_class: type[DicomTask]
     app_settings_class: type[DicomAppSettings]
     logs: list[DicomLogEntry] = []
+    # Set by the task runner before process() runs: True when the current run
+    # is the last Procrastinate attempt of this queued job, i.e. no automatic
+    # retry will follow a RetriableDicomError.
+    is_final_attempt: bool = False
 
     def __init__(self, dicom_task: DicomTask) -> None:
         self.dicom_task = dicom_task

adit/core/tasks.py

@@ -87,10 +87,21 @@ def _run_dicom_task(
 
     logger.info(f"Processing of {dicom_task} started.")
 
+    # Cave, the attempts of the Procrastinate job must not be the same number
+    # as the attempts of the DicomTask. The DicomTask could be started by multiple
+    # Procrastinate jobs (e.g. if the user canceled and resumed the same task).
+    # Procrastinate's attempts is 0-indexed (counts previous attempts).
+    # On attempt N, attempts = N-1, so the final attempt is when
+    # attempts + 1 >= max_attempts.
+    is_final_attempt = context.job.attempts + 1 >= settings.DICOM_TASK_MAX_ATTEMPTS
+
     @concurrent.process(timeout=process_timeout, daemon=True)
-    def _process_dicom_task(model_label: str, task_id: int) -> ProcessingResult:
+    def _process_dicom_task(
+        model_label: str, task_id: int, is_final_attempt: bool
+    ) -> ProcessingResult:
         dicom_task = get_dicom_task(model_label, task_id)
         processor = get_dicom_processor(dicom_task)
+        processor.is_final_attempt = is_final_attempt
 
         logger.info(f"Start processing of {dicom_task}.")
         return processor.process()
@@ -104,7 +115,9 @@ def _monitor_task(context: JobContext, future: ProcessFuture) -> None:
         db.close_old_connections()
 
     try:
-        future = cast(ProcessFuture, _process_dicom_task(model_label, task_id))
+        future = cast(
+            ProcessFuture, _process_dicom_task(model_label, task_id, is_final_attempt)
+        )
         _monitor_task(context, future)
         result: ProcessingResult = future.result()
         dicom_task.status = result["status"]
@@ -125,13 +138,7 @@ def _monitor_task(context: JobContext, future: ProcessFuture) -> None:
     except RetriableDicomError as err:
         logger.exception("Retriable error occurred during %s.", dicom_task)
 
-        # Cave, the the attempts of the Procrastinate job must not be the same number
-        # as the attempts of the DicomTask. The DicomTask could be started by multiple
-        # Procrastinate jobs (e.g. if the user canceled and resumed the same task).
-        # Procrastinate's attempts is 0-indexed (counts previous attempts).
-        # On attempt N, attempts = N-1. We want FAILURE on the final attempt,
-        # which is when attempts + 1 >= max_attempts.
-        if context.job.attempts + 1 < settings.DICOM_TASK_MAX_ATTEMPTS:
+        if not is_final_attempt:
             dicom_task.status = DicomTask.Status.PENDING
             dicom_task.message = "Task failed, but will be retried."
             if dicom_task.log:
@@ -161,7 +168,11 @@ def _monitor_task(context: JobContext, future: ProcessFuture) -> None:
 
     finally:
         dicom_task.end = timezone.now()
-        dicom_task.save()
+        # Only the fields this runner owns: a full-field save here would
+        # clobber task fields the processor subprocess persisted mid-run
+        # (e.g. MassTransferTask.anonymizer_seed) with this stale parent
+        # instance.
+        dicom_task.save(update_fields=["status", "message", "log", "end"])
         logger.info(f"Processing of {dicom_task} ended.")
 
         with pglock.advisory(DISTRIBUTED_LOCK):

adit/core/tests/test_tasks.py

@@ -4,6 +4,7 @@
 
 import pytest
 from adit_radis_shared.common.utils.testing_helpers import run_worker_once
+from django.conf import settings
 from procrastinate import JobContext
 from pytest_mock import MockerFixture
 
@@ -467,6 +468,54 @@ def test_run_dicom_task_accepts_in_progress_task_on_retry(mocker: MockerFixture)
     assert dicom_task.message == "recovered"
 
 
+@pytest.mark.django_db(transaction=True)
+def test_run_dicom_task_final_save_does_not_clobber_subprocess_writes(mocker: MockerFixture):
+    """The processor subprocess may persist task fields mid-run (e.g.
+    MassTransferTask.anonymizer_seed). The runner's finally-save uses a stale
+    parent-process instance and must therefore only write the fields it owns."""
+    dicom_job = ExampleTransferJobFactory.create(status=DicomJob.Status.PENDING)
+    dicom_task = ExampleTransferTaskFactory.create(status=DicomTask.Status.PENDING, job=dicom_job)
+    model_label = get_model_label(ExampleTransferTask)
+
+    result: ProcessingResult = {
+        "status": DicomTask.Status.SUCCESS,
+        "message": "ok",
+        "log": "",
+    }
+
+    def fake_process(*p_args, **p_kwargs):
+        def decorator(func):
+            def wrapper(*args, **kwargs):
+                # Simulate the subprocess persisting a field mid-run, after the
+                # parent already loaded its stale copy of the task.
+                ExampleTransferTask.objects.filter(pk=dicom_task.pk).update(attempts=99)
+                return _FakeFuture(result=result)
+
+            return wrapper
+
+        return decorator
+
+    def fake_thread(*t_args, **t_kwargs):
+        def decorator(func):
+            def wrapper(*args, **kwargs):
+                return None
+
+            return wrapper
+
+        return decorator
+
+    mocker.patch.object(tasks_module.concurrent, "process", side_effect=fake_process)
+    mocker.patch.object(tasks_module.concurrent, "thread", side_effect=fake_thread)
+
+    tasks_module._run_dicom_task(_make_context(), model_label, dicom_task.pk)
+
+    dicom_task.refresh_from_db()
+    assert dicom_task.status == DicomTask.Status.SUCCESS
+    assert dicom_task.attempts == 99, (
+        "runner's finally-save clobbered a field written from the subprocess"
+    )
+
+
 @pytest.mark.django_db
 def test_check_disk_space_warns_when_over_limit(mocker: MockerFixture):
     from adit.core.factories import DicomFolderFactory
@@ -506,3 +555,60 @@ def test_check_disk_space_no_warning_when_under_limit(mocker: MockerFixture):
     tasks_module.check_disk_space()
 
     mail_mock.assert_not_called()
+
+
+@pytest.mark.django_db(transaction=True)
+@pytest.mark.parametrize(
+    "procrastinate_attempts,expected_final",
+    [
+        (0, False),
+        (settings.DICOM_TASK_MAX_ATTEMPTS - 1, True),
+    ],
+)
+def test_run_dicom_task_passes_is_final_attempt_to_subprocess(
+    mocker: MockerFixture, procrastinate_attempts: int, expected_final: bool
+):
+    """The runner computes is_final_attempt from Procrastinate's 0-indexed
+    per-job attempt counter and passes it into the processor subprocess."""
+    dicom_job = ExampleTransferJobFactory.create(status=DicomJob.Status.PENDING)
+    dicom_task = ExampleTransferTaskFactory.create(status=DicomTask.Status.PENDING, job=dicom_job)
+    model_label = get_model_label(ExampleTransferTask)
+
+    result: ProcessingResult = {
+        "status": DicomTask.Status.SUCCESS,
+        "message": "ok",
+        "log": "",
+    }
+    captured: dict[str, tuple] = {}
+
+    def fake_process(*p_args, **p_kwargs):
+        def decorator(func):
+            def wrapper(*args, **kwargs):
+                captured["args"] = args
+                return _FakeFuture(result=result)
+
+            return wrapper
+
+        return decorator
+
+    def fake_thread(*t_args, **t_kwargs):
+        def decorator(func):
+            def wrapper(*args, **kwargs):
+                return None
+
+            return wrapper
+
+        return decorator
+
+    mocker.patch.object(tasks_module.concurrent, "process", side_effect=fake_process)
+    mocker.patch.object(tasks_module.concurrent, "thread", side_effect=fake_thread)
+
+    tasks_module._run_dicom_task(
+        _make_context(attempts=procrastinate_attempts), model_label, dicom_task.pk
+    )
+
+    assert captured["args"] == (model_label, dicom_task.pk, expected_final)
+
+
+def test_dicom_task_processor_is_final_attempt_defaults_to_false():
+    assert DicomTaskProcessor.is_final_attempt is False

adit/mass_transfer/migrations/0006_masstransfervolume_retriable.py

@@ -0,0 +1,18 @@
+# Generated by Django 6.0.7 on 2026-07-16 14:15
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("mass_transfer", "0005_add_partition_constraint"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="masstransfervolume",
+            name="retriable",
+            field=models.BooleanField(default=False),
+        ),
+    ]

adit/mass_transfer/migrations/0007_masstransfertask_anonymizer_seed.py

@@ -0,0 +1,18 @@
+# Generated by Django 6.0.7 on 2026-07-16 14:56
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("mass_transfer", "0006_masstransfervolume_retriable"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="masstransfertask",
+            name="anonymizer_seed",
+            field=models.CharField(blank=True, default="", max_length=64),
+        ),
+    ]

adit/mass_transfer/models.py

@@ -98,6 +98,11 @@ class MassTransferTask(TransferTask):
     partition_end = models.DateTimeField()
     partition_key = models.CharField(max_length=64)
 
+    # Anonymizer seed for jobs without a fixed pseudonym_salt. Generated on the
+    # first attempt of a queue cycle and reused by automatic retries so that a
+    # resumed run pseudonymizes UIDs/dates consistently with earlier attempts.
+    anonymizer_seed = models.CharField(max_length=64, blank=True, default="")
+
     volumes: models.QuerySet["MassTransferVolume"]
 
     class Meta:
@@ -167,6 +172,9 @@ class Status(models.TextChoices):
     converted_file = models.TextField(blank=True, default="")
 
     status = models.CharField(max_length=16, choices=Status.choices, default=Status.PENDING)
+    # Only meaningful with status=ERROR: the failure was a RetriableDicomError,
+    # so a later task attempt resets this volume to PENDING and re-transfers it.
+    retriable = models.BooleanField(default=False)
     log = models.TextField(blank=True, default="")
 
     created = models.DateTimeField(auto_now_add=True)