diff --git a/adit/core/processors.py b/adit/core/processors.py index fac09960..7cbe03a5 100644 --- a/adit/core/processors.py +++ b/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 diff --git a/adit/core/tasks.py b/adit/core/tasks.py index f763ee87..e57254b1 100644 --- a/adit/core/tasks.py +++ b/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): diff --git a/adit/core/tests/test_tasks.py b/adit/core/tests/test_tasks.py index 0a4e8ded..7ea914ce 100644 --- a/adit/core/tests/test_tasks.py +++ b/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 diff --git a/adit/mass_transfer/migrations/0006_masstransfervolume_retriable.py b/adit/mass_transfer/migrations/0006_masstransfervolume_retriable.py new file mode 100644 index 00000000..24f321d1 --- /dev/null +++ b/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), + ), + ] diff --git a/adit/mass_transfer/migrations/0007_masstransfertask_anonymizer_seed.py b/adit/mass_transfer/migrations/0007_masstransfertask_anonymizer_seed.py new file mode 100644 index 00000000..d3adb7f8 --- /dev/null +++ b/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), + ), + ] diff --git a/adit/mass_transfer/models.py b/adit/mass_transfer/models.py index ba4d5d49..cb7b1c21 100644 --- a/adit/mass_transfer/models.py +++ b/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) diff --git a/adit/mass_transfer/processors.py b/adit/mass_transfer/processors.py index 9c0ebadf..1193252f 100644 --- a/adit/mass_transfer/processors.py +++ b/adit/mass_transfer/processors.py @@ -11,6 +11,7 @@ import pydicom from django.conf import settings +from django.db import transaction from django.utils import timezone from pydicom import Dataset from pydicom.errors import InvalidDicomError @@ -333,37 +334,64 @@ def process(self): "log": "Mass transfer requires at least one filter.", } - # Clean up on retry - if output_base: - partition_path = output_base / self.mass_task.partition_key - if partition_path.exists(): - shutil.rmtree(partition_path) - - MassTransferVolume.objects.filter( - job=job, - partition_key=self.mass_task.partition_key, - ).delete() + # A fresh queue cycle starts from a clean slate. Automatic + # Procrastinate retries (attempts >= 2) resume from the volumes + # persisted by the previous attempt. User-initiated Retry/Restart + # reset attempts to 0 via reset_tasks(), so they wipe again. + is_fresh_cycle = self.mass_task.attempts <= 1 + if is_fresh_cycle: + if output_base: + partition_path = output_base / self.mass_task.partition_key + if partition_path.exists(): + shutil.rmtree(partition_path) + + MassTransferVolume.objects.filter( + job=job, + partition_key=self.mass_task.partition_key, + ).delete() pseudonymizer: Pseudonymizer | None = None if job.pseudonymize and job.pseudonym_salt: pseudonymizer = Pseudonymizer(seed=job.pseudonym_salt) elif job.pseudonymize: - pseudonymizer = Pseudonymizer() + # Without a job-wide salt the anonymizer key must still be + # stable across automatic retries, or a resumed attempt would + # pseudonymize UIDs/dates differently than the volumes already + # transferred (splitting studies, duplicating instances on + # server destinations). + if is_fresh_cycle or not self.mass_task.anonymizer_seed: + self.mass_task.anonymizer_seed = secrets.token_hex(20) + self.mass_task.save(update_fields=["anonymizer_seed"]) + pseudonymizer = Pseudonymizer(seed=self.mass_task.anonymizer_seed) operator = DicomOperator(source_node.dicomserver, persistent=True) - # Discovery: query the source server for all matching series - discovered = self._discover_series(operator, filters) - operator.close() + volumes: list[MassTransferVolume] = [] + if not is_fresh_cycle: + volumes = list( + MassTransferVolume.objects.filter( + job=job, + partition_key=self.mass_task.partition_key, + ) + ) - # Create PENDING volumes so they appear in the UI immediately - volumes = self._create_pending_volumes(discovered, job, pseudonymizer) - grouped_volumes = self._group_volumes(volumes) + if volumes: + # Resumed run: re-queue only the volumes that failed retriably. + self._reset_retriable_volumes(volumes) + else: + # Discovery: query the source server for all matching series. + # Also reached on a resumed run that was interrupted before + # any volumes were created. + discovered = self._discover_series(operator, filters) + operator.close() + + # Create PENDING volumes so they appear in the UI immediately + volumes = self._create_pending_volumes(discovered, job, pseudonymizer) - # Transfer: fetch series grouped by study + # Transfer: fetch pending series grouped by study return self._transfer_grouped_series( operator, - grouped_volumes, + volumes, job, pseudonymizer, output_base, @@ -431,7 +459,11 @@ def _create_pending_volumes( ) ) - return MassTransferVolume.objects.bulk_create(volumes) + # Atomic so a worker kill mid-create rolls back to zero rows and the + # next attempt re-runs discovery, instead of resuming from a partial + # row set that silently drops the missing series. + with transaction.atomic(): + return MassTransferVolume.objects.bulk_create(volumes) @staticmethod def _group_volumes( @@ -445,26 +477,43 @@ def _group_volumes( ) return grouped + @staticmethod + def _reset_retriable_volumes(volumes: list[MassTransferVolume]) -> None: + """Reset volumes that failed retriably in a previous attempt to PENDING.""" + retriable_volumes = [ + volume + for volume in volumes + if volume.status == MassTransferVolume.Status.ERROR and volume.retriable + ] + for volume in retriable_volumes: + volume.status = MassTransferVolume.Status.PENDING + volume.retriable = False + volume.log = "" + volume.converted_file = "" + MassTransferVolume.objects.bulk_update( + retriable_volumes, ["status", "retriable", "log", "converted_file"] + ) + def _transfer_grouped_series( self, operator: DicomOperator, - grouped_volumes: dict[str, dict[str, list[MassTransferVolume]]], + volumes: list[MassTransferVolume], job: MassTransferJob, pseudonymizer: Pseudonymizer | None, output_base: Path | None, dest_operator: DicomOperator | None = None, ) -> dict: - """Transfer all grouped series. + """Transfer all pending volumes and summarize the whole partition. - Iterates patients -> studies -> volumes, updating each volume in place. + Iterates patients -> studies -> volumes, updating each volume in + place. Volumes already completed by a previous attempt are left + untouched. Raises RetriableDicomError at the end when retriable + failures remain and this is not the final task attempt. """ - total_processed = 0 - total_skipped = 0 - total_failed = 0 - total_volumes = 0 - study_count = 0 - failed_reasons: dict[str, int] = {} + pending = [v for v in volumes if v.status == MassTransferVolume.Status.PENDING] + grouped_volumes = self._group_volumes(pending) + study_count = 0 for patient_id, studies in grouped_volumes.items(): for study_uid, volumes_list in studies.items(): study_count += 1 @@ -481,8 +530,6 @@ def _transfer_grouped_series( # One fetch association per study try: for volume in volumes_list: - total_volumes += 1 - subject_id = volume.pseudonym or sanitize_filename(volume.patient_id) self._transfer_single_series( operator, @@ -493,26 +540,19 @@ def _transfer_grouped_series( output_base, dest_operator, ) - - if volume.status == MassTransferVolume.Status.ERROR: - total_failed += 1 - reason = _short_error_reason(volume.log) if volume.log else "Unknown" - failed_reasons[reason] = failed_reasons.get(reason, 0) + 1 - elif volume.status == MassTransferVolume.Status.SKIPPED: - total_skipped += 1 - else: - total_processed += 1 finally: operator.close() - return self._build_task_summary( - total_volumes, - study_count, - total_processed, - total_skipped, - total_failed, - failed_reasons, + retriable_failures = sum( + 1 for v in volumes if v.status == MassTransferVolume.Status.ERROR and v.retriable ) + if retriable_failures and not self.is_final_attempt: + raise RetriableDicomError( + f"{retriable_failures} of {len(volumes)} volumes failed retriably " + "and will be retried." + ) + + return self._build_task_summary(volumes) def _transfer_single_series( self, @@ -526,8 +566,10 @@ def _transfer_single_series( ) -> None: """Export (and optionally convert) a single series. - Updates volume fields in place and saves. Never raises except for - RetriableDicomError. + Updates volume fields in place and saves. Never raises: a retriable + failure is recorded on the volume (status=ERROR, retriable=True) so + that process() can schedule a task retry after the whole partition + was attempted. """ try: if dest_operator: @@ -557,6 +599,11 @@ def _transfer_single_series( / series_folder ) + if output_path.exists(): + # A previous attempt may have written a partial series + # (folder exports are not atomic) — start clean. + shutil.rmtree(output_path) + if job.convert_to_nifti: if volume.modality in settings.MODALITIES_EXCLUDED_FROM_NIFTI_CONVERSION: logger.debug( @@ -581,10 +628,13 @@ def _transfer_single_series( subject_id, output_path, ) - except RetriableDicomError: + except RetriableDicomError as err: + # Don't abort the partition for one dead series. process() checks + # for retriable volumes after the loop and raises once if a task + # retry should re-transfer them. volume.status = MassTransferVolume.Status.ERROR - volume.log = "Transfer interrupted by retriable error; task will be retried." - raise + volume.retriable = True + volume.log = str(err) except Exception as err: logger.exception( "Mass transfer failed for series %s", @@ -604,6 +654,7 @@ def _transfer_single_series( volume.save( update_fields=[ "status", + "retriable", "log", "study_instance_uid_pseudonymized", "series_instance_uid_pseudonymized", @@ -744,16 +795,27 @@ def _set_zero_image_status( f"Fetch returned 0 images (PACS reports {volume.number_of_images} instances)" ) - def _build_task_summary( - self, - total_volumes: int, - study_count: int, - total_processed: int, - total_skipped: int, - total_failed: int, - failed_reasons: dict[str, int], - ) -> dict: - """Build the final status dict returned to the task processor.""" + def _build_task_summary(self, volumes: list[MassTransferVolume]) -> dict: + """Build the final status dict from the state of all partition volumes. + + Counts cover the whole partition (including volumes completed by + earlier attempts), not just the volumes processed in this run. + """ + total_volumes = len(volumes) + study_count = len({v.study_instance_uid for v in volumes}) + total_processed = sum( + 1 + for v in volumes + if v.status in (MassTransferVolume.Status.EXPORTED, MassTransferVolume.Status.CONVERTED) + ) + total_skipped = sum(1 for v in volumes if v.status == MassTransferVolume.Status.SKIPPED) + failed_volumes = [v for v in volumes if v.status == MassTransferVolume.Status.ERROR] + total_failed = len(failed_volumes) + failed_reasons: dict[str, int] = {} + for volume in failed_volumes: + reason = _short_error_reason(volume.log) if volume.log else "Unknown" + failed_reasons[reason] = failed_reasons.get(reason, 0) + 1 + log_lines = [ f"Partition {self.mass_task.partition_key}", f"Studies found: {study_count}", diff --git a/adit/mass_transfer/tests/test_processor.py b/adit/mass_transfer/tests/test_processor.py index 2a8ca251..338952c7 100644 --- a/adit/mass_transfer/tests/test_processor.py +++ b/adit/mass_transfer/tests/test_processor.py @@ -687,6 +687,10 @@ def _make_process_env( processor.mass_task.pk = 42 processor.mass_task.partition_key = "20240101" + # Default to the first attempt of a queue cycle (fresh run: wipe + + # discovery). Resume behavior (attempts >= 2) is covered by the DB + # integration tests, which need real querysets. + processor.mass_task.attempts = 1 mocker.patch.object(processor, "is_suspended", return_value=False) mocker.patch("adit.mass_transfer.processors.DicomOperator") @@ -734,6 +738,10 @@ def _make_process_env_server_dest( processor.mass_task.pk = 42 processor.mass_task.partition_key = "20240101" + # Default to the first attempt of a queue cycle (fresh run: wipe + + # discovery). Resume behavior (attempts >= 2) is covered by the DB + # integration tests, which need real querysets. + processor.mass_task.attempts = 1 mocker.patch.object(processor, "is_suspended", return_value=False) @@ -762,21 +770,39 @@ def _make_process_env_server_dest( return processor, dest_operator -def test_process_reraises_retriable_dicom_error(mocker: MockerFixture, tmp_path: Path): +@pytest.mark.django_db +def test_process_transfers_all_volumes_then_raises_retriable(mocker: MockerFixture, tmp_path: Path): + """A retriable failure no longer aborts the partition: the remaining + volumes still transfer, and one RetriableDicomError is raised at the end + so Procrastinate retries only the failed volumes.""" processor = _make_process_env(mocker, tmp_path) - series = [_make_discovered(series_uid="s-1")] - + series = [ + _make_discovered(series_uid="s-1"), + _make_discovered(series_uid="s-2"), + ] mocker.patch.object(processor, "_discover_series", return_value=series) - mocker.patch.object( - processor, - "_export_series", - side_effect=RetriableDicomError("PACS connection lost"), - ) - with pytest.raises(RetriableDicomError, match="PACS connection lost"): + captured: dict[str, MassTransferVolume] = {} + + def fake_export(*args, **kwargs): + volume = args[1] + captured[volume.series_instance_uid] = volume + if volume.series_instance_uid == "s-1": + raise RetriableDicomError("PACS connection lost") + return (1, "", "") + + mocker.patch.object(processor, "_export_series", side_effect=fake_export) + + with pytest.raises(RetriableDicomError, match="1 of 2 volumes"): processor.process() + # The healthy volume was still transferred before the raise + assert captured["s-2"].status == MassTransferVolume.Status.EXPORTED + assert captured["s-1"].status == MassTransferVolume.Status.ERROR + assert captured["s-1"].retriable is True + +@pytest.mark.django_db def test_process_returns_warning_on_partial_failure(mocker: MockerFixture, tmp_path: Path): processor = _make_process_env(mocker, tmp_path) series = [ @@ -803,6 +829,7 @@ def fake_export(*args, **kwargs): assert "Failed: 1" in result["log"] +@pytest.mark.django_db def test_process_returns_failure_when_all_fail(mocker: MockerFixture, tmp_path: Path): processor = _make_process_env(mocker, tmp_path) series = [ @@ -848,6 +875,7 @@ def test_process_returns_failure_when_no_filters(mocker: MockerFixture, tmp_path assert "filter" in result["log"].lower() +@pytest.mark.django_db def test_process_returns_success_for_empty_partition(mocker: MockerFixture, tmp_path: Path): processor = _make_process_env(mocker, tmp_path) mocker.patch.object(processor, "_discover_series", return_value=[]) @@ -858,8 +886,10 @@ def test_process_returns_success_for_empty_partition(mocker: MockerFixture, tmp_ assert "No series found" in result["message"] -def test_process_cleans_partition_on_retry(mocker: MockerFixture, tmp_path: Path): - """On retry, ALL pre-existing volumes for the partition are deleted and rediscovered.""" +@pytest.mark.django_db +def test_process_cleans_partition_on_fresh_cycle(mocker: MockerFixture, tmp_path: Path): + """On the first attempt of a queue cycle, ALL pre-existing volumes for the + partition are deleted and rediscovered (user Retry/Restart semantics).""" processor = _make_process_env(mocker, tmp_path) series = [ _make_discovered(series_uid="s-1"), @@ -894,6 +924,7 @@ def fake_export(*args, **kwargs): # --------------------------------------------------------------------------- +@pytest.mark.django_db def test_process_server_destination_exports_and_uploads(mocker: MockerFixture): processor, mock_dest_operator = _make_process_env_server_dest(mocker) series = [_make_discovered(series_uid="s-1")] @@ -913,7 +944,8 @@ def fake_export(op, s, path, subject_id, pseudonymizer): assert result["status"] == MassTransferTask.Status.SUCCESS -def test_process_server_destination_cleans_volumes_on_retry(mocker: MockerFixture): +@pytest.mark.django_db +def test_process_server_destination_cleans_volumes_on_fresh_cycle(mocker: MockerFixture): """Server destination should still delete old DB volume records on retry.""" processor, _ = _make_process_env_server_dest(mocker) series = [_make_discovered(series_uid="s-1")] @@ -930,6 +962,7 @@ def test_process_server_destination_cleans_volumes_on_retry(mocker: MockerFixtur mock_filter_qs.delete.assert_called_once() +@pytest.mark.django_db def test_process_server_destination_closes_dest_operator(mocker: MockerFixture): """dest_operator.close() should be called even if transfer fails.""" processor, mock_dest_operator = _make_process_env_server_dest(mocker) @@ -985,6 +1018,7 @@ def test_export_series_to_server_skips_non_image_series(mocker: MockerFixture): assert volume.status == MassTransferVolume.Status.SKIPPED +@pytest.mark.django_db def test_server_destination_upload_dicom_error_marks_failure(mocker: MockerFixture): """When upload_images raises DicomError, the series should be marked as failed.""" processor, mock_dest_operator = _make_process_env_server_dest(mocker) @@ -1003,8 +1037,12 @@ def fake_export(op, s, path, subject_id, pseudonymizer): assert result["status"] == MassTransferTask.Status.FAILURE -def test_server_destination_upload_retriable_error_propagates(mocker: MockerFixture): - """When upload_images raises RetriableDicomError, it must propagate up.""" +@pytest.mark.django_db +def test_server_destination_upload_retriable_error_marks_volume_and_raises_at_end( + mocker: MockerFixture, +): + """When upload_images raises RetriableDicomError, the volume is marked + retriable and the aggregated end-of-loop error is raised.""" processor, mock_dest_operator = _make_process_env_server_dest(mocker) series = [_make_discovered(series_uid="s-1")] @@ -1016,10 +1054,21 @@ def fake_export(op, s, path, subject_id, pseudonymizer): mocker.patch.object(processor, "_export_series", side_effect=fake_export) mock_dest_operator.upload_images.side_effect = RetriableDicomError("Connection reset") - with pytest.raises(RetriableDicomError, match="Connection reset"): + with pytest.raises(RetriableDicomError, match="1 of 1 volumes"): processor.process() +def test_process_retriable_error_during_discovery_propagates(mocker: MockerFixture, tmp_path: Path): + """A RetriableDicomError raised before any volumes exist (discovery) + still aborts and retries the whole task unchanged.""" + processor = _make_process_env(mocker, tmp_path) + mocker.patch.object(processor, "_discover_series", side_effect=RetriableDicomError("PACS down")) + + with pytest.raises(RetriableDicomError, match="PACS down"): + processor.process() + + +@pytest.mark.django_db def test_process_none_mode_uses_patient_id_as_subject(mocker: MockerFixture, tmp_path: Path): """When pseudonymize=False, no pseudonymizer is used.""" processor = _make_process_env(mocker, tmp_path, pseudonymize=False, pseudonym_salt="") @@ -1044,6 +1093,7 @@ def fake_export(op, s, path, subject_id, pseudonymizer): assert result["status"] == MassTransferTask.Status.SUCCESS +@pytest.mark.django_db def test_process_pseudonymize_mode_same_study_same_pseudonym(mocker: MockerFixture, tmp_path: Path): """In non-linking mode, series in the same study share a pseudonym.""" processor = _make_process_env(mocker, tmp_path, pseudonym_salt="") @@ -1070,6 +1120,7 @@ def fake_export(op, s, path, subject_id, pseudonymizer): assert subject_ids[0] != "PAT1" +@pytest.mark.django_db def test_process_pseudonymize_mode_different_studies_different_pseudonyms( mocker: MockerFixture, tmp_path: Path ): @@ -1098,6 +1149,7 @@ def fake_export(op, s, path, subject_id, pseudonymizer): assert subject_ids[0] != "PAT1" +@pytest.mark.django_db def test_process_linking_mode_uses_deterministic_pseudonym(mocker: MockerFixture, tmp_path: Path): """In linking mode (pseudonymize with non-empty salt), pseudonyms are deterministic.""" processor = _make_process_env( @@ -1957,6 +2009,7 @@ def test_create_pending_volumes_deterministic_pseudonym(): assert "PAT2" in grouped +@pytest.mark.django_db def test_create_pending_volumes_no_anonymization(mocker: MockerFixture): """Without pseudonymizer, volumes have empty pseudonym.""" processor = _make_processor(mocker) @@ -1980,6 +2033,7 @@ def test_create_pending_volumes_no_anonymization(mocker: MockerFixture): assert volumes[1].pseudonym == "" +@pytest.mark.django_db def test_create_pending_volumes_random_assigns_per_study(mocker: MockerFixture): """With pseudonymizer but no salt, volumes get per-study random pseudonyms.""" from adit.core.utils.pseudonymizer import Pseudonymizer @@ -2059,6 +2113,7 @@ def test_group_volumes_multi_patient_multi_study(): # --------------------------------------------------------------------------- +@pytest.mark.django_db def test_retriable_error_saves_volume_as_error(mocker: MockerFixture, tmp_path: Path): """RetriableDicomError should save the current volume as ERROR before propagating.""" processor = _make_process_env(mocker, tmp_path) @@ -2313,3 +2368,358 @@ def fake_export(op, s, path, subject_id, pseudonymizer): # The path should contain the job-identifying folder expected_prefix = f"adit_mass_transfer_{job.pk}_{job.created.strftime('%Y%m%d')}_researcher" assert expected_prefix in str(export_paths[0]) + + +# --------------------------------------------------------------------------- +# resumable-retries tests (see +# docs/superpowers/specs/2026-07-16-mass-transfer-resumable-retries-design.md) +# --------------------------------------------------------------------------- + + +def test_transfer_single_series_marks_retriable_and_continues( + mocker: MockerFixture, tmp_path: Path +): + """A per-volume RetriableDicomError no longer propagates: the volume is + marked ERROR + retriable so a later attempt re-transfers it, and the + partition loop can continue.""" + processor = _make_processor(mocker) + processor.mass_task.partition_key = "20240101" + job = processor.mass_task.job + job.convert_to_nifti = False + mocker.patch.object(processor, "_export_series", side_effect=RetriableDicomError("boom")) + mocker.patch.object(MassTransferVolume, "save") + volume = MassTransferVolume( + series_instance_uid="s-1", study_datetime=timezone.now(), number_of_images=5 + ) + + # Must not raise + processor._transfer_single_series(mocker.MagicMock(), volume, job, None, "subj", tmp_path, None) + + assert volume.status == MassTransferVolume.Status.ERROR + assert volume.retriable is True + assert "boom" in volume.log + + +def test_transfer_single_series_permanent_error_not_retriable( + mocker: MockerFixture, tmp_path: Path +): + processor = _make_processor(mocker) + processor.mass_task.partition_key = "20240101" + job = processor.mass_task.job + job.convert_to_nifti = False + mocker.patch.object(processor, "_export_series", side_effect=DicomError("bad series")) + mocker.patch.object(MassTransferVolume, "save") + volume = MassTransferVolume( + series_instance_uid="s-1", study_datetime=timezone.now(), number_of_images=5 + ) + + processor._transfer_single_series(mocker.MagicMock(), volume, job, None, "subj", tmp_path, None) + + assert volume.status == MassTransferVolume.Status.ERROR + assert volume.retriable is False + + +def test_transfer_single_series_cleans_stale_series_folder(mocker: MockerFixture, tmp_path: Path): + """A partially written series folder from a previous attempt is removed + before re-export (folder exports are not atomic).""" + processor = _make_processor(mocker) + processor.mass_task.partition_key = "20240101" + job = processor.mass_task.job + job.convert_to_nifti = False + mocker.patch.object(MassTransferVolume, "save") + volume = MassTransferVolume( + series_instance_uid="s-1", + study_description="Brain CT", + series_description="Axial", + series_number=1, + study_datetime=timezone.now(), + number_of_images=5, + ) + output_path = ( + tmp_path + / "20240101" + / "subj" + / _study_folder_name(volume.study_description, volume.study_datetime) + / _series_folder_name( + volume.series_description, volume.series_number, volume.series_instance_uid + ) + ) + output_path.mkdir(parents=True) + stale_file = output_path / "stale.dcm" + stale_file.write_bytes(b"partial") + + mocker.patch.object(processor, "_export_series", side_effect=_fake_export_success) + + processor._transfer_single_series(mocker.MagicMock(), volume, job, None, "subj", tmp_path, None) + + assert not stale_file.exists() + assert volume.status == MassTransferVolume.Status.EXPORTED + + +@pytest.mark.django_db +def test_process_continues_past_dead_series_on_final_attempt(mocker: MockerFixture, tmp_path: Path): + """On the final attempt, one dead series among healthy ones yields WARNING: + the dead volume stays ERROR and no exception propagates.""" + processor = _make_process_env(mocker, tmp_path) + processor.is_final_attempt = True + series = [ + _make_discovered(series_uid="s-1"), + _make_discovered(series_uid="s-2"), + ] + mocker.patch.object(processor, "_discover_series", return_value=series) + + captured: dict[str, MassTransferVolume] = {} + + def fake_export(*args, **kwargs): + volume = args[1] + captured[volume.series_instance_uid] = volume + if volume.series_instance_uid == "s-1": + raise RetriableDicomError("PACS connection lost") + return (1, "", "") + + mocker.patch.object(processor, "_export_series", side_effect=fake_export) + + result = processor.process() # must not raise + + assert result["status"] == MassTransferTask.Status.WARNING + assert "Processed: 1" in result["log"] + assert "Failed: 1" in result["log"] + assert captured["s-1"].status == MassTransferVolume.Status.ERROR + assert "PACS connection lost" in captured["s-1"].log + assert captured["s-2"].status == MassTransferVolume.Status.EXPORTED + + +@pytest.mark.django_db +def test_process_final_attempt_all_dead_is_failure(mocker: MockerFixture, tmp_path: Path): + """On the final attempt where every series is dead, the task is FAILURE.""" + processor = _make_process_env(mocker, tmp_path) + processor.is_final_attempt = True + series = [ + _make_discovered(series_uid="s-1"), + _make_discovered(series_uid="s-2"), + ] + mocker.patch.object(processor, "_discover_series", return_value=series) + mocker.patch.object(processor, "_export_series", side_effect=RetriableDicomError("PACS down")) + + result = processor.process() # must not raise + + assert result["status"] == MassTransferTask.Status.FAILURE + assert "Failed: 2" in result["log"] + + +@pytest.mark.django_db +def test_process_resumes_without_wipe_or_rediscovery(mocker: MockerFixture, mass_transfer_env): + """An automatic retry (attempts >= 2) keeps completed volumes and their + files, skips discovery, and re-transfers only retriable volumes.""" + env = mass_transfer_env + env.task.attempts = 2 + now = timezone.now() + + exported = MassTransferVolume.objects.create( + job=env.job, + task=env.task, + partition_key=env.task.partition_key, + patient_id="PAT1", + study_instance_uid="study-1", + series_instance_uid="s-1", + study_datetime=now, + number_of_images=5, + status=MassTransferVolume.Status.EXPORTED, + ) + retriable = MassTransferVolume.objects.create( + job=env.job, + task=env.task, + partition_key=env.task.partition_key, + patient_id="PAT1", + study_instance_uid="study-1", + series_instance_uid="s-2", + study_datetime=now, + number_of_images=5, + status=MassTransferVolume.Status.ERROR, + retriable=True, + log="old retriable error", + ) + permanent = MassTransferVolume.objects.create( + job=env.job, + task=env.task, + partition_key=env.task.partition_key, + patient_id="PAT1", + study_instance_uid="study-1", + series_instance_uid="s-3", + study_datetime=now, + number_of_images=5, + status=MassTransferVolume.Status.ERROR, + retriable=False, + log="unreadable series", + ) + + # A file written by the previous attempt must survive the resumed run + base_dir = _destination_base_dir(env.destination, env.job) + prior_file = base_dir / env.task.partition_key / "done.dcm" + prior_file.parent.mkdir(parents=True, exist_ok=True) + prior_file.write_bytes(b"already transferred") + + processor = MassTransferTaskProcessor(env.task) + mocker.patch("adit.mass_transfer.processors.DicomOperator") + discover_mock = mocker.patch.object(processor, "_discover_series") + exported_uids: list[str] = [] + + def fake_export(op, volume, *args, **kwargs): + exported_uids.append(volume.series_instance_uid) + return (1, "", "") + + mocker.patch.object(processor, "_export_series", side_effect=fake_export) + + result = processor.process() + + discover_mock.assert_not_called() + assert exported_uids == ["s-2"] + assert prior_file.exists() + + exported.refresh_from_db() + retriable.refresh_from_db() + permanent.refresh_from_db() + assert exported.status == MassTransferVolume.Status.EXPORTED + assert retriable.status == MassTransferVolume.Status.EXPORTED + assert retriable.retriable is False + assert permanent.status == MassTransferVolume.Status.ERROR + + # Summary covers the whole partition, not just the resumed volumes + assert result["status"] == MassTransferTask.Status.WARNING + assert "Series found: 3" in result["log"] + assert "Processed: 2" in result["log"] + assert "Failed: 1" in result["log"] + + +@pytest.mark.django_db +def test_process_fresh_cycle_wipes_stale_volumes_and_rediscovers( + mocker: MockerFixture, mass_transfer_env +): + """attempts <= 1 (fresh job or user Retry/Restart) keeps today's + clean-slate semantics: rows and folder are wiped, discovery runs.""" + env = mass_transfer_env + env.task.attempts = 1 + + MassTransferVolume.objects.create( + job=env.job, + task=env.task, + partition_key=env.task.partition_key, + patient_id="PAT1", + study_instance_uid="study-old", + series_instance_uid="stale-1", + study_datetime=timezone.now(), + status=MassTransferVolume.Status.EXPORTED, + ) + + processor = MassTransferTaskProcessor(env.task) + mocker.patch("adit.mass_transfer.processors.DicomOperator") + mocker.patch.object( + processor, "_discover_series", return_value=[_make_discovered(series_uid="s-new")] + ) + mocker.patch.object(processor, "_export_series", side_effect=_fake_export_success) + + result = processor.process() + + assert not MassTransferVolume.objects.filter(series_instance_uid="stale-1").exists() + new_volume = MassTransferVolume.objects.get(job=env.job, series_instance_uid="s-new") + assert new_volume.status == MassTransferVolume.Status.EXPORTED + assert result["status"] == MassTransferTask.Status.SUCCESS + + +@pytest.mark.django_db +def test_process_persists_retriable_flag_for_next_attempt(mocker: MockerFixture, mass_transfer_env): + """The retriable flag round-trips through the DB so the NEXT attempt can + find and reset the volume.""" + env = mass_transfer_env + env.task.attempts = 1 + + processor = MassTransferTaskProcessor(env.task) + mocker.patch("adit.mass_transfer.processors.DicomOperator") + mocker.patch.object( + processor, + "_discover_series", + return_value=[ + _make_discovered(series_uid="s-1"), + _make_discovered(series_uid="s-2"), + ], + ) + + def fake_export(op, volume, *args, **kwargs): + if volume.series_instance_uid == "s-1": + raise RetriableDicomError("PACS connection lost") + return (1, "", "") + + mocker.patch.object(processor, "_export_series", side_effect=fake_export) + + with pytest.raises(RetriableDicomError, match="1 of 2 volumes"): + processor.process() + + dead = MassTransferVolume.objects.get(job=env.job, series_instance_uid="s-1") + healthy = MassTransferVolume.objects.get(job=env.job, series_instance_uid="s-2") + assert dead.status == MassTransferVolume.Status.ERROR + assert dead.retriable is True + assert "PACS connection lost" in dead.log + assert healthy.status == MassTransferVolume.Status.EXPORTED + assert healthy.retriable is False + + +@pytest.mark.django_db +def test_process_random_mode_generates_and_persists_anonymizer_seed( + mocker: MockerFixture, mass_transfer_env +): + """Fresh cycle in random mode: a seed is generated, saved on the task, and + used to construct the Pseudonymizer.""" + env = mass_transfer_env + env.job.pseudonymize = True + env.job.pseudonym_salt = "" + env.job.save(update_fields=["pseudonymize", "pseudonym_salt"]) + env.task.attempts = 1 + + processor = MassTransferTaskProcessor(env.task) + mocker.patch("adit.mass_transfer.processors.DicomOperator") + pseudonymizer_mock = mocker.patch("adit.mass_transfer.processors.Pseudonymizer") + mocker.patch.object(processor, "_discover_series", return_value=[]) + + processor.process() + + env.task.refresh_from_db() + assert env.task.anonymizer_seed != "" + pseudonymizer_mock.assert_called_once_with(seed=env.task.anonymizer_seed) + + +@pytest.mark.django_db +def test_process_random_mode_reuses_anonymizer_seed_on_resume( + mocker: MockerFixture, mass_transfer_env +): + """A resumed attempt must construct the Pseudonymizer with the seed + persisted by the previous attempt, not a new one.""" + env = mass_transfer_env + env.job.pseudonymize = True + env.job.pseudonym_salt = "" + env.job.save(update_fields=["pseudonymize", "pseudonym_salt"]) + env.task.attempts = 2 + env.task.anonymizer_seed = "seed-from-first-attempt" + env.task.save(update_fields=["anonymizer_seed"]) + + MassTransferVolume.objects.create( + job=env.job, + task=env.task, + partition_key=env.task.partition_key, + patient_id="PAT1", + study_instance_uid="study-1", + series_instance_uid="s-1", + study_datetime=timezone.now(), + status=MassTransferVolume.Status.EXPORTED, + ) + + processor = MassTransferTaskProcessor(env.task) + mocker.patch("adit.mass_transfer.processors.DicomOperator") + pseudonymizer_mock = mocker.patch("adit.mass_transfer.processors.Pseudonymizer") + discover_mock = mocker.patch.object(processor, "_discover_series") + + processor.process() + + discover_mock.assert_not_called() + env.task.refresh_from_db() + assert env.task.anonymizer_seed == "seed-from-first-attempt" + pseudonymizer_mock.assert_called_once_with(seed="seed-from-first-attempt") diff --git a/docs/superpowers/plans/2026-07-16-mass-transfer-resumable-retries.md b/docs/superpowers/plans/2026-07-16-mass-transfer-resumable-retries.md new file mode 100644 index 00000000..a857a1cb --- /dev/null +++ b/docs/superpowers/plans/2026-07-16-mass-transfer-resumable-retries.md @@ -0,0 +1,1157 @@ +# Mass Transfer Resumable Automatic Retries Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Automatic Procrastinate retries of a mass transfer task resume from persisted `MassTransferVolume` rows instead of wiping the partition and re-fetching everything, so one dead series costs only its own retries. + +**Architecture:** The partition wipe becomes conditional on a fresh queue cycle (`DicomTask.attempts <= 1`); resumed runs reload volume rows, reset retriably-failed ones to `PENDING`, and re-transfer only those. A per-volume `RetriableDicomError` never aborts the loop; one `RetriableDicomError` is raised at the end when retriable failures remain and attempts remain. The runner computes `is_final_attempt` from Procrastinate's per-cycle counter and passes it into the processor subprocess, replacing the stop-gap's `DicomTask.attempts`-based check (and its cancel→resume edge case). + +**Tech Stack:** Django 5.1, Procrastinate, pytest (pytest-django, pytest-mock), factory-boy. + +**Spec:** `docs/superpowers/specs/2026-07-16-mass-transfer-resumable-retries-design.md` + +## Global Constraints + +- Line length 100 (Ruff); Google Python Style Guide. +- Use `assert` for internal invariants (never run with `python -O`). +- Booleans: `default=False` is enough; no `blank`/`null` (non-string, but has a default). +- All test commands run through the dev containers: `uv run cli test -- `. Containers must be up (`uv run cli compose-up -- --watch`). +- Lint gate: `uv run cli lint` must pass at the end. +- Commit after every task. Current branch: `fix/mass-transfer-final-attempt-continue`. +- `settings.DICOM_TASK_MAX_ATTEMPTS` is 3 in `adit/settings/base.py:473` — never hardcode 3 in tests; reference the setting. + +--- + +### Task 1: Add `retriable` field to `MassTransferVolume` + +**Files:** +- Modify: `adit/mass_transfer/models.py` (class `MassTransferVolume`, around line 169) +- Create: `adit/mass_transfer/migrations/0006_masstransfervolume_retriable.py` (via makemigrations) + +**Interfaces:** +- Produces: `MassTransferVolume.retriable: bool` (model field, default `False`). Later tasks set it `True` alongside `status=Status.ERROR` when the failure was a `RetriableDicomError`, and reset it to `False` when the volume is re-queued. + +- [ ] **Step 1: Add the field** + +In `adit/mass_transfer/models.py`, `MassTransferVolume`, directly below the `status` field (line 169): + +```python + 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="") +``` + +- [ ] **Step 2: Generate the migration** + +Run: `uv run ./manage.py makemigrations mass_transfer` +Expected: creates `adit/mass_transfer/migrations/0006_masstransfervolume_retriable.py` containing exactly: + +```python +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), + ), + ] +``` + +(If the command must run in the container instead, use `uv run cli shell` conventions or run it via the web container; the generated file content is what matters.) + +- [ ] **Step 3: Sanity-check migrations apply in the test run** + +Run: `uv run cli test -- adit/mass_transfer/tests/test_processor.py -k "test_process_creates_volume_records_on_success" -v` +Expected: PASS (pytest-django builds the schema from migrations; a broken migration fails here). + +- [ ] **Step 4: Commit** + +```bash +git add adit/mass_transfer/models.py adit/mass_transfer/migrations/0006_masstransfervolume_retriable.py +git commit -m "feat(mass_transfer): add retriable flag to MassTransferVolume" +``` + +--- + +### Task 2: Runner passes `is_final_attempt` into the processor + +**Files:** +- Modify: `adit/core/processors.py` (class `DicomTaskProcessor`, around line 27-35) +- Modify: `adit/core/tasks.py:83-146` (`_run_dicom_task`) +- Test: `adit/core/tests/test_tasks.py` + +**Interfaces:** +- Consumes: nothing from other tasks. +- Produces: `DicomTaskProcessor.is_final_attempt: bool` (class attribute, default `False`), set by the runner before `process()` runs. `True` iff the current run is the last Procrastinate attempt of this queued job (`context.job.attempts + 1 >= settings.DICOM_TASK_MAX_ATTEMPTS`). Task 4 reads `self.is_final_attempt` in the mass transfer processor. + +- [ ] **Step 1: Write the failing test** + +Append to `adit/core/tests/test_tasks.py` (the file already defines `_FakeFuture`, `_make_context`, `ExampleProcessor`, and the factories; add `from django.conf import settings` to the imports at the top): + +```python +@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 +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `uv run cli test -- adit/core/tests/test_tasks.py -k "is_final_attempt" -v` +Expected: FAIL — the parametrized test with `captured["args"] == (..., expected_final)` fails because the subprocess is called with only 2 args; the default-attribute test fails with `AttributeError: ... has no attribute 'is_final_attempt'`. + +- [ ] **Step 3: Implement** + +In `adit/core/processors.py`, add to `DicomTaskProcessor` directly below the `logs` class attribute (line 30): + +```python +class DicomTaskProcessor(abc.ABC): + app_name: str + 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 +``` + +In `adit/core/tasks.py`, `_run_dicom_task`: after `dicom_task.save()` / the "Processing ... started" log (line 88), compute the flag and thread it through the subprocess; simplify the `except RetriableDicomError` handler to use it. The affected region (lines 83-146) becomes: + +```python + dicom_task.status = DicomTask.Status.IN_PROGRESS + dicom_task.start = timezone.now() + dicom_task.attempts += 1 + dicom_task.save() + + 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, 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() + + @concurrent.thread() + def _monitor_task(context: JobContext, future: ProcessFuture) -> None: + while not future.done(): + if context.should_abort(): + future.cancel() + sleep(settings.DICOM_TASK_CANCELED_MONITOR_INTERVAL) + db.close_old_connections() + + try: + 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"] + dicom_task.message = result["message"] + dicom_task.log = result["log"] + ensure_db_connection() +``` + +and the retriable handler (previously the `context.job.attempts + 1 < ...` check at line 134): + +```python + except RetriableDicomError as err: + logger.exception("Retriable error occurred during %s.", dicom_task) + + if not is_final_attempt: + dicom_task.status = DicomTask.Status.PENDING + dicom_task.message = "Task failed, but will be retried." + if dicom_task.log: + dicom_task.log += "\n" + dicom_task.log += str(err) + else: + dicom_task.status = DicomTask.Status.FAILURE + dicom_task.message = str(err) + + ensure_db_connection() + + raise err +``` + +Everything else in the function is unchanged. + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `uv run cli test -- adit/core/tests/test_tasks.py -v` +Expected: PASS — the two new tests plus all existing runner tests (especially `test_process_dicom_task_that_should_be_retried` and `test_process_dicom_task_transitions_to_failure_after_max_retries`, which exercise the refactored handler). + +- [ ] **Step 5: Commit** + +```bash +git add adit/core/processors.py adit/core/tasks.py adit/core/tests/test_tasks.py +git commit -m "feat(core): pass is_final_attempt from task runner into processors" +``` + +--- + +### Task 3: `_transfer_single_series` — mark retriable and continue; per-series cleanup + +**Files:** +- Modify: `adit/mass_transfer/processors.py:517-639` (`_is_final_attempt`, `_transfer_single_series`) +- Test: `adit/mass_transfer/tests/test_processor.py` (replaces `test_transfer_single_series_final_attempt_boundary`, lines 2331-2369) + +**Interfaces:** +- Consumes: `MassTransferVolume.retriable` (Task 1). +- Produces: `_transfer_single_series(operator, volume, job, pseudonymizer, subject_id, output_base, dest_operator)` never raises. On `RetriableDicomError` it sets `volume.status = ERROR`, `volume.retriable = True`, `volume.log = str(err)` and returns. It deletes a pre-existing series output folder before exporting (folder destinations). Task 4's loop and end-of-loop raise rely on exactly this contract. + +- [ ] **Step 1: Write the failing tests** + +In `adit/mass_transfer/tests/test_processor.py`, replace the whole "final-attempt-continue tests" section (the comment banner at lines 2325-2328 and `test_transfer_single_series_final_attempt_boundary` at lines 2331-2369; keep `test_process_continues_past_dead_series_on_final_attempt` and `test_process_final_attempt_all_dead_is_failure` for now — Task 4 updates them) with: + +```python +# --------------------------------------------------------------------------- +# resumable-retries tests (see +# docs/superpowers/specs/2026-07-16-mass-transfer-resumable-retries-design.md) +# --------------------------------------------------------------------------- + + +def test_transfer_single_series_marks_retriable_and_continues( + mocker: MockerFixture, tmp_path: Path +): + """A per-volume RetriableDicomError no longer propagates: the volume is + marked ERROR + retriable so a later attempt re-transfers it, and the + partition loop can continue.""" + processor = _make_processor(mocker) + processor.mass_task.partition_key = "20240101" + job = processor.mass_task.job + job.convert_to_nifti = False + mocker.patch.object(processor, "_export_series", side_effect=RetriableDicomError("boom")) + mocker.patch.object(MassTransferVolume, "save") + volume = MassTransferVolume( + series_instance_uid="s-1", study_datetime=timezone.now(), number_of_images=5 + ) + + # Must not raise + processor._transfer_single_series( + mocker.MagicMock(), volume, job, None, "subj", tmp_path, None + ) + + assert volume.status == MassTransferVolume.Status.ERROR + assert volume.retriable is True + assert "boom" in volume.log + + +def test_transfer_single_series_permanent_error_not_retriable( + mocker: MockerFixture, tmp_path: Path +): + processor = _make_processor(mocker) + processor.mass_task.partition_key = "20240101" + job = processor.mass_task.job + job.convert_to_nifti = False + mocker.patch.object(processor, "_export_series", side_effect=DicomError("bad series")) + mocker.patch.object(MassTransferVolume, "save") + volume = MassTransferVolume( + series_instance_uid="s-1", study_datetime=timezone.now(), number_of_images=5 + ) + + processor._transfer_single_series( + mocker.MagicMock(), volume, job, None, "subj", tmp_path, None + ) + + assert volume.status == MassTransferVolume.Status.ERROR + assert volume.retriable is False + + +def test_transfer_single_series_cleans_stale_series_folder( + mocker: MockerFixture, tmp_path: Path +): + """A partially written series folder from a previous attempt is removed + before re-export (folder exports are not atomic).""" + processor = _make_processor(mocker) + processor.mass_task.partition_key = "20240101" + job = processor.mass_task.job + job.convert_to_nifti = False + mocker.patch.object(MassTransferVolume, "save") + volume = MassTransferVolume( + series_instance_uid="s-1", + study_description="Brain CT", + series_description="Axial", + series_number=1, + study_datetime=timezone.now(), + number_of_images=5, + ) + output_path = ( + tmp_path + / "20240101" + / "subj" + / _study_folder_name(volume.study_description, volume.study_datetime) + / _series_folder_name( + volume.series_description, volume.series_number, volume.series_instance_uid + ) + ) + output_path.mkdir(parents=True) + stale_file = output_path / "stale.dcm" + stale_file.write_bytes(b"partial") + + mocker.patch.object(processor, "_export_series", side_effect=_fake_export_success) + + processor._transfer_single_series( + mocker.MagicMock(), volume, job, None, "subj", tmp_path, None + ) + + assert not stale_file.exists() + assert volume.status == MassTransferVolume.Status.EXPORTED +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `uv run cli test -- adit/mass_transfer/tests/test_processor.py -k "transfer_single_series" -v` +Expected: `test_transfer_single_series_marks_retriable_and_continues` FAILS (the RetriableDicomError propagates — the current code re-raises on non-final attempts, and `_make_processor`'s MagicMock task has a MagicMock `attempts` so `_is_final_attempt()` comparison is unreliable); `test_transfer_single_series_cleans_stale_series_folder` FAILS (stale file survives). The permanent-error test may already pass (`retriable` defaults to `False`). + +- [ ] **Step 3: Implement** + +In `adit/mass_transfer/processors.py`: + +a) Delete `_is_final_attempt()` entirely (lines 517-524). + +b) In `_transfer_single_series`, update the docstring and add the pre-export cleanup right after `output_path` is computed (currently lines 561-567), and replace the `except RetriableDicomError` branch (lines 593-608). The method becomes: + +```python + def _transfer_single_series( + self, + operator: DicomOperator, + volume: MassTransferVolume, + job: MassTransferJob, + pseudonymizer: Pseudonymizer | None, + subject_id: str, + output_base: Path | None, + dest_operator: DicomOperator | None = None, + ) -> None: + """Export (and optionally convert) a single series. + + Updates volume fields in place and saves. Never raises: a retriable + failure is recorded on the volume (status=ERROR, retriable=True) so + that process() can schedule a task retry after the whole partition + was attempted. + """ + try: + if dest_operator: + self._export_series_to_server( + operator, + volume, + pseudonymizer, + subject_id, + dest_operator, + ) + else: + assert output_base is not None + study_folder = _study_folder_name( + volume.study_description, + volume.study_datetime, + ) + series_folder = _series_folder_name( + volume.series_description, + volume.series_number, + volume.series_instance_uid, + ) + output_path = ( + output_base + / self.mass_task.partition_key + / subject_id + / study_folder + / series_folder + ) + + if output_path.exists(): + # A previous attempt may have written a partial series + # (folder exports are not atomic) — start clean. + shutil.rmtree(output_path) + + if job.convert_to_nifti: + if volume.modality in settings.MODALITIES_EXCLUDED_FROM_NIFTI_CONVERSION: + logger.debug( + f"Skipping series {volume.series_instance_uid} " + f"(modality {volume.modality} excluded from NIfTI conversion)" + ) + volume.status = MassTransferVolume.Status.SKIPPED + volume.log = f"Modality {volume.modality} excluded from NIfTI conversion" + else: + self._export_and_convert_series( + operator, + volume, + pseudonymizer, + subject_id, + output_path, + ) + else: + self._export_series_to_folder( + operator, + volume, + pseudonymizer, + subject_id, + output_path, + ) + except RetriableDicomError as err: + # Don't abort the partition for one dead series. process() checks + # for retriable volumes after the loop and raises once if a task + # retry should re-transfer them. + volume.status = MassTransferVolume.Status.ERROR + volume.retriable = True + volume.log = str(err) + except Exception as err: + logger.exception( + "Mass transfer failed for series %s", + volume.series_instance_uid, + ) + volume.status = MassTransferVolume.Status.ERROR + volume.log = str(err) + finally: + if volume.status == MassTransferVolume.Status.PENDING: + logger.error( + "Volume %s still PENDING after transfer — setting to ERROR.", + volume.series_instance_uid, + ) + volume.status = MassTransferVolume.Status.ERROR + volume.log = "Internal error: volume status was not updated after transfer." + try: + volume.save( + update_fields=[ + "status", + "retriable", + "log", + "study_instance_uid_pseudonymized", + "series_instance_uid_pseudonymized", + "converted_file", + "updated", + ] + ) + except Exception: + logger.exception( + "Failed to save volume %s status to database", + volume.series_instance_uid, + ) +``` + +Note: `"retriable"` was added to `update_fields`; the cleanup block is new; the retriable branch no longer raises and no longer consults the attempt number. + +- [ ] **Step 4: Run tests to verify the new ones pass** + +Run: `uv run cli test -- adit/mass_transfer/tests/test_processor.py -k "transfer_single_series" -v` +Expected: all three new tests PASS. Do NOT run the whole file yet — `process()`-level tests that expect the old abort behavior (`test_process_reraises_retriable_dicom_error`, the two remaining final-attempt tests, `test_server_destination_upload_retriable_error_propagates`) are now red; Task 4 fixes `process()` and those tests together. + +- [ ] **Step 5: Commit** + +```bash +git add adit/mass_transfer/processors.py adit/mass_transfer/tests/test_processor.py +git commit -m "feat(mass_transfer): mark retriable volume failures and clean stale series folders" +``` + +--- + +### Task 4: Resumable `process()` — conditional wipe, retriable reset, whole-partition summary, end-of-loop raise + +**Files:** +- Modify: `adit/mass_transfer/processors.py:301-374` (`process`), `448-515` (`_transfer_grouped_series`), `768-816` (`_build_task_summary`) +- Test: `adit/mass_transfer/tests/test_processor.py` (helpers `_make_process_env`/`_make_process_env_server_dest`, several existing `process()` tests, new DB-integration tests) + +**Interfaces:** +- Consumes: `MassTransferVolume.retriable` (Task 1), `self.is_final_attempt` (Task 2), the never-raises `_transfer_single_series` contract (Task 3). +- Produces: + - `process()` — wipes folder + volume rows only when `self.mass_task.attempts <= 1`; skips discovery when volume rows exist for the partition; raises `RetriableDicomError(f"{n} of {total} volumes failed retriably and will be retried.")` when retriable failures remain and `not self.is_final_attempt`. + - `_transfer_grouped_series(operator, volumes: list[MassTransferVolume], job, pseudonymizer, output_base, dest_operator=None) -> dict` — new signature: takes ALL partition volumes, transfers only the `PENDING` ones, raises at the end or returns the summary. + - `_build_task_summary(volumes: list[MassTransferVolume]) -> dict` — new signature: computes counts from the volume objects (whole partition), not loop counters. + - `_reset_retriable_volumes(volumes: list[MassTransferVolume]) -> None` — static method, resets ERROR+retriable volumes to PENDING in memory and in the DB. + +- [ ] **Step 1: Update the mocked-test helpers** + +In `adit/mass_transfer/tests/test_processor.py`, in BOTH `_make_process_env` (line 693) and `_make_process_env_server_dest` (line 743), replace: + +```python + # Default to a non-final attempt so per-volume retriable errors re-raise. + # Tests that exercise the final-attempt-continue path opt in explicitly. + processor.mass_task.attempts = settings.DICOM_TASK_MAX_ATTEMPTS - 1 +``` + +with: + +```python + # Default to the first attempt of a queue cycle (fresh run: wipe + + # discovery). Resume behavior (attempts >= 2) is covered by the DB + # integration tests, which need real querysets. + processor.mass_task.attempts = 1 +``` + +This matters: with a MagicMock `attempts >= 2` the new `process()` would take the resume path and call `list()` on a mocked queryset, which breaks. + +- [ ] **Step 2: Rewrite the stale mocked tests to the new contract (failing against current code)** + +a) Replace `test_process_reraises_retriable_dicom_error` (line 772) with: + +```python +def test_process_transfers_all_volumes_then_raises_retriable( + mocker: MockerFixture, tmp_path: Path +): + """A retriable failure no longer aborts the partition: the remaining + volumes still transfer, and one RetriableDicomError is raised at the end + so Procrastinate retries only the failed volumes.""" + processor = _make_process_env(mocker, tmp_path) + series = [ + _make_discovered(series_uid="s-1"), + _make_discovered(series_uid="s-2"), + ] + mocker.patch.object(processor, "_discover_series", return_value=series) + + captured: dict[str, MassTransferVolume] = {} + + def fake_export(*args, **kwargs): + volume = args[1] + captured[volume.series_instance_uid] = volume + if volume.series_instance_uid == "s-1": + raise RetriableDicomError("PACS connection lost") + return (1, "", "") + + mocker.patch.object(processor, "_export_series", side_effect=fake_export) + + with pytest.raises(RetriableDicomError, match="1 of 2 volumes"): + processor.process() + + # The healthy volume was still transferred before the raise + assert captured["s-2"].status == MassTransferVolume.Status.EXPORTED + assert captured["s-1"].status == MassTransferVolume.Status.ERROR + assert captured["s-1"].retriable is True +``` + +b) Replace `test_process_continues_past_dead_series_on_final_attempt` (line 2371) with (same body, but drive the behavior via `is_final_attempt` instead of `attempts`): + +```python +def test_process_continues_past_dead_series_on_final_attempt( + mocker: MockerFixture, tmp_path: Path +): + """On the final attempt, one dead series among healthy ones yields WARNING: + the dead volume stays ERROR and no exception propagates.""" + processor = _make_process_env(mocker, tmp_path) + processor.is_final_attempt = True + series = [ + _make_discovered(series_uid="s-1"), + _make_discovered(series_uid="s-2"), + ] + mocker.patch.object(processor, "_discover_series", return_value=series) + + captured: dict[str, MassTransferVolume] = {} + + def fake_export(*args, **kwargs): + volume = args[1] + captured[volume.series_instance_uid] = volume + if volume.series_instance_uid == "s-1": + raise RetriableDicomError("PACS connection lost") + return (1, "", "") + + mocker.patch.object(processor, "_export_series", side_effect=fake_export) + + result = processor.process() # must not raise + + assert result["status"] == MassTransferTask.Status.WARNING + assert "Processed: 1" in result["log"] + assert "Failed: 1" in result["log"] + assert captured["s-1"].status == MassTransferVolume.Status.ERROR + assert "PACS connection lost" in captured["s-1"].log + assert captured["s-2"].status == MassTransferVolume.Status.EXPORTED +``` + +c) Replace `test_process_final_attempt_all_dead_is_failure` (line 2404) with: + +```python +def test_process_final_attempt_all_dead_is_failure(mocker: MockerFixture, tmp_path: Path): + """On the final attempt where every series is dead, the task is FAILURE.""" + processor = _make_process_env(mocker, tmp_path) + processor.is_final_attempt = True + series = [ + _make_discovered(series_uid="s-1"), + _make_discovered(series_uid="s-2"), + ] + mocker.patch.object(processor, "_discover_series", return_value=series) + mocker.patch.object(processor, "_export_series", side_effect=RetriableDicomError("PACS down")) + + result = processor.process() # must not raise + + assert result["status"] == MassTransferTask.Status.FAILURE + assert "Failed: 2" in result["log"] +``` + +d) Rename `test_process_cleans_partition_on_retry` (line 868) to `test_process_cleans_partition_on_fresh_cycle` and update its docstring — behavior is unchanged for `attempts = 1` (the helper's new default): + +```python +def test_process_cleans_partition_on_fresh_cycle(mocker: MockerFixture, tmp_path: Path): + """On the first attempt of a queue cycle, ALL pre-existing volumes for the + partition are deleted and rediscovered (user Retry/Restart semantics).""" +``` + +(body unchanged). Apply the same rename/docstring treatment to `test_process_server_destination_cleans_volumes_on_retry` (line 923) → `test_process_server_destination_cleans_volumes_on_fresh_cycle`. + +e) Replace `test_server_destination_upload_retriable_error_propagates` (line 1013) with: + +```python +def test_server_destination_upload_retriable_error_marks_volume_and_raises_at_end( + mocker: MockerFixture, +): + """When upload_images raises RetriableDicomError, the volume is marked + retriable and the aggregated end-of-loop error is raised.""" + processor, mock_dest_operator = _make_process_env_server_dest(mocker) + series = [_make_discovered(series_uid="s-1")] + + mocker.patch.object(processor, "_discover_series", return_value=series) + + def fake_export(op, s, path, subject_id, pseudonymizer): + return (1, "pseudo-study-uid", "pseudo-series-uid") + + mocker.patch.object(processor, "_export_series", side_effect=fake_export) + mock_dest_operator.upload_images.side_effect = RetriableDicomError("Connection reset") + + with pytest.raises(RetriableDicomError, match="1 of 1 volumes"): + processor.process() +``` + +f) Add a new mocked test right after it, pinning the spec's "retriable error outside the per-volume loop still propagates as-is" clause: + +```python +def test_process_retriable_error_during_discovery_propagates( + mocker: MockerFixture, tmp_path: Path +): + """A RetriableDicomError raised before any volumes exist (discovery) + still aborts and retries the whole task unchanged.""" + processor = _make_process_env(mocker, tmp_path) + mocker.patch.object( + processor, "_discover_series", side_effect=RetriableDicomError("PACS down") + ) + + with pytest.raises(RetriableDicomError, match="PACS down"): + processor.process() +``` + +- [ ] **Step 3: Write the new DB-integration tests (failing)** + +Append after the tests from Task 3 in the resumable-retries section: + +```python +@pytest.mark.django_db +def test_process_resumes_without_wipe_or_rediscovery( + mocker: MockerFixture, mass_transfer_env +): + """An automatic retry (attempts >= 2) keeps completed volumes and their + files, skips discovery, and re-transfers only retriable volumes.""" + env = mass_transfer_env + env.task.attempts = 2 + now = timezone.now() + + exported = MassTransferVolume.objects.create( + job=env.job, + task=env.task, + partition_key=env.task.partition_key, + patient_id="PAT1", + study_instance_uid="study-1", + series_instance_uid="s-1", + study_datetime=now, + number_of_images=5, + status=MassTransferVolume.Status.EXPORTED, + ) + retriable = MassTransferVolume.objects.create( + job=env.job, + task=env.task, + partition_key=env.task.partition_key, + patient_id="PAT1", + study_instance_uid="study-1", + series_instance_uid="s-2", + study_datetime=now, + number_of_images=5, + status=MassTransferVolume.Status.ERROR, + retriable=True, + log="old retriable error", + ) + permanent = MassTransferVolume.objects.create( + job=env.job, + task=env.task, + partition_key=env.task.partition_key, + patient_id="PAT1", + study_instance_uid="study-1", + series_instance_uid="s-3", + study_datetime=now, + number_of_images=5, + status=MassTransferVolume.Status.ERROR, + retriable=False, + log="unreadable series", + ) + + # A file written by the previous attempt must survive the resumed run + base_dir = _destination_base_dir(env.destination, env.job) + prior_file = base_dir / env.task.partition_key / "done.dcm" + prior_file.parent.mkdir(parents=True, exist_ok=True) + prior_file.write_bytes(b"already transferred") + + processor = MassTransferTaskProcessor(env.task) + mocker.patch("adit.mass_transfer.processors.DicomOperator") + discover_mock = mocker.patch.object(processor, "_discover_series") + exported_uids: list[str] = [] + + def fake_export(op, volume, *args, **kwargs): + exported_uids.append(volume.series_instance_uid) + return (1, "", "") + + mocker.patch.object(processor, "_export_series", side_effect=fake_export) + + result = processor.process() + + discover_mock.assert_not_called() + assert exported_uids == ["s-2"] + assert prior_file.exists() + + exported.refresh_from_db() + retriable.refresh_from_db() + permanent.refresh_from_db() + assert exported.status == MassTransferVolume.Status.EXPORTED + assert retriable.status == MassTransferVolume.Status.EXPORTED + assert retriable.retriable is False + assert permanent.status == MassTransferVolume.Status.ERROR + + # Summary covers the whole partition, not just the resumed volumes + assert result["status"] == MassTransferTask.Status.WARNING + assert "Series found: 3" in result["log"] + assert "Processed: 2" in result["log"] + assert "Failed: 1" in result["log"] + + +@pytest.mark.django_db +def test_process_fresh_cycle_wipes_stale_volumes_and_rediscovers( + mocker: MockerFixture, mass_transfer_env +): + """attempts <= 1 (fresh job or user Retry/Restart) keeps today's + clean-slate semantics: rows and folder are wiped, discovery runs.""" + env = mass_transfer_env + env.task.attempts = 1 + + MassTransferVolume.objects.create( + job=env.job, + task=env.task, + partition_key=env.task.partition_key, + patient_id="PAT1", + study_instance_uid="study-old", + series_instance_uid="stale-1", + study_datetime=timezone.now(), + status=MassTransferVolume.Status.EXPORTED, + ) + + processor = MassTransferTaskProcessor(env.task) + mocker.patch("adit.mass_transfer.processors.DicomOperator") + mocker.patch.object( + processor, "_discover_series", return_value=[_make_discovered(series_uid="s-new")] + ) + mocker.patch.object(processor, "_export_series", side_effect=_fake_export_success) + + result = processor.process() + + assert not MassTransferVolume.objects.filter(series_instance_uid="stale-1").exists() + new_volume = MassTransferVolume.objects.get(job=env.job, series_instance_uid="s-new") + assert new_volume.status == MassTransferVolume.Status.EXPORTED + assert result["status"] == MassTransferTask.Status.SUCCESS + + +@pytest.mark.django_db +def test_process_persists_retriable_flag_for_next_attempt( + mocker: MockerFixture, mass_transfer_env +): + """The retriable flag round-trips through the DB so the NEXT attempt can + find and reset the volume.""" + env = mass_transfer_env + env.task.attempts = 1 + + processor = MassTransferTaskProcessor(env.task) + mocker.patch("adit.mass_transfer.processors.DicomOperator") + mocker.patch.object( + processor, + "_discover_series", + return_value=[ + _make_discovered(series_uid="s-1"), + _make_discovered(series_uid="s-2"), + ], + ) + + def fake_export(op, volume, *args, **kwargs): + if volume.series_instance_uid == "s-1": + raise RetriableDicomError("PACS connection lost") + return (1, "", "") + + mocker.patch.object(processor, "_export_series", side_effect=fake_export) + + with pytest.raises(RetriableDicomError, match="1 of 2 volumes"): + processor.process() + + dead = MassTransferVolume.objects.get(job=env.job, series_instance_uid="s-1") + healthy = MassTransferVolume.objects.get(job=env.job, series_instance_uid="s-2") + assert dead.status == MassTransferVolume.Status.ERROR + assert dead.retriable is True + assert "PACS connection lost" in dead.log + assert healthy.status == MassTransferVolume.Status.EXPORTED + assert healthy.retriable is False +``` + +Note for the implementer: `mass_transfer_env.destination` is a `DicomFolder` (a `DicomNode` subclass); existing tests in this file pass it directly to `_destination_base_dir(destination, job)` (see line 2118), which is already imported at the top of the file. + +- [ ] **Step 4: Run the new/updated tests to verify they fail** + +Run: `uv run cli test -- adit/mass_transfer/tests/test_processor.py -k "resumes or fresh_cycle or persists_retriable or transfers_all_volumes or final_attempt" -v` +Expected: FAIL — current `process()` wipes unconditionally, aborts on the first retriable error (via the old re-raise... which Task 3 already removed, so instead: no end-of-loop raise happens and dead volumes silently count as plain failures). + +- [ ] **Step 5: Implement the new `process()` and helpers** + +In `adit/mass_transfer/processors.py`: + +a) Replace the body of `process()` from the `try:` (line 326) down to the `_transfer_grouped_series` call (line 371) with: + +```python + try: + filters = job.get_filters() + + if not filters: + return { + "status": MassTransferTask.Status.FAILURE, + "message": "No filters configured for this job.", + "log": "Mass transfer requires at least one filter.", + } + + # A fresh queue cycle starts from a clean slate. Automatic + # Procrastinate retries (attempts >= 2) resume from the volumes + # persisted by the previous attempt. User-initiated Retry/Restart + # reset attempts to 0 via reset_tasks(), so they wipe again. + is_fresh_cycle = self.mass_task.attempts <= 1 + if is_fresh_cycle: + if output_base: + partition_path = output_base / self.mass_task.partition_key + if partition_path.exists(): + shutil.rmtree(partition_path) + + MassTransferVolume.objects.filter( + job=job, + partition_key=self.mass_task.partition_key, + ).delete() + + pseudonymizer: Pseudonymizer | None = None + if job.pseudonymize and job.pseudonym_salt: + pseudonymizer = Pseudonymizer(seed=job.pseudonym_salt) + elif job.pseudonymize: + pseudonymizer = Pseudonymizer() + + operator = DicomOperator(source_node.dicomserver, persistent=True) + + volumes: list[MassTransferVolume] = [] + if not is_fresh_cycle: + volumes = list( + MassTransferVolume.objects.filter( + job=job, + partition_key=self.mass_task.partition_key, + ) + ) + + if volumes: + # Resumed run: re-queue only the volumes that failed retriably. + self._reset_retriable_volumes(volumes) + else: + # Discovery: query the source server for all matching series. + # Also reached on a resumed run that was interrupted before + # any volumes were created. + discovered = self._discover_series(operator, filters) + operator.close() + + # Create PENDING volumes so they appear in the UI immediately + volumes = self._create_pending_volumes(discovered, job, pseudonymizer) + + # Transfer: fetch pending series grouped by study + return self._transfer_grouped_series( + operator, + volumes, + job, + pseudonymizer, + output_base, + dest_operator, + ) + finally: + if dest_operator: + dest_operator.close() +``` + +b) Add `_reset_retriable_volumes` below `_group_volumes`: + +```python + @staticmethod + def _reset_retriable_volumes(volumes: list[MassTransferVolume]) -> None: + """Reset volumes that failed retriably in a previous attempt to PENDING.""" + retriable_volumes = [ + volume + for volume in volumes + if volume.status == MassTransferVolume.Status.ERROR and volume.retriable + ] + for volume in retriable_volumes: + volume.status = MassTransferVolume.Status.PENDING + volume.retriable = False + volume.log = "" + volume.converted_file = "" + MassTransferVolume.objects.bulk_update( + retriable_volumes, ["status", "retriable", "log", "converted_file"] + ) +``` + +c) Replace `_transfer_grouped_series` (lines 448-515) with: + +```python + def _transfer_grouped_series( + self, + operator: DicomOperator, + volumes: list[MassTransferVolume], + job: MassTransferJob, + pseudonymizer: Pseudonymizer | None, + output_base: Path | None, + dest_operator: DicomOperator | None = None, + ) -> dict: + """Transfer all pending volumes and summarize the whole partition. + + Iterates patients -> studies -> volumes, updating each volume in + place. Volumes already completed by a previous attempt are left + untouched. Raises RetriableDicomError at the end when retriable + failures remain and this is not the final task attempt. + """ + pending = [v for v in volumes if v.status == MassTransferVolume.Status.PENDING] + grouped_volumes = self._group_volumes(pending) + + study_count = 0 + for patient_id, studies in grouped_volumes.items(): + for study_uid, volumes_list in studies.items(): + study_count += 1 + + if study_count > 1: + # Pacing delay between consecutive studies. Each study opens a + # fresh association and switches patient/study context, which is + # where a busy PACS is most likely to reject or drop requests. + # Series inside the same study fetch back-to-back over the already + # open association. + # TODO: Investigate if this is still necessary. + time.sleep(_DELAY_BETWEEN_STUDIES) + + # One fetch association per study + try: + for volume in volumes_list: + subject_id = volume.pseudonym or sanitize_filename(volume.patient_id) + self._transfer_single_series( + operator, + volume, + job, + pseudonymizer, + subject_id, + output_base, + dest_operator, + ) + finally: + operator.close() + + retriable_failures = sum( + 1 + for v in volumes + if v.status == MassTransferVolume.Status.ERROR and v.retriable + ) + if retriable_failures and not self.is_final_attempt: + raise RetriableDicomError( + f"{retriable_failures} of {len(volumes)} volumes failed retriably " + "and will be retried." + ) + + return self._build_task_summary(volumes) +``` + +d) Replace `_build_task_summary` (lines 768-816) with: + +```python + def _build_task_summary(self, volumes: list[MassTransferVolume]) -> dict: + """Build the final status dict from the state of all partition volumes. + + Counts cover the whole partition (including volumes completed by + earlier attempts), not just the volumes processed in this run. + """ + total_volumes = len(volumes) + study_count = len({v.study_instance_uid for v in volumes}) + total_processed = sum( + 1 + for v in volumes + if v.status + in (MassTransferVolume.Status.EXPORTED, MassTransferVolume.Status.CONVERTED) + ) + total_skipped = sum( + 1 for v in volumes if v.status == MassTransferVolume.Status.SKIPPED + ) + failed_volumes = [v for v in volumes if v.status == MassTransferVolume.Status.ERROR] + total_failed = len(failed_volumes) + failed_reasons: dict[str, int] = {} + for volume in failed_volumes: + reason = _short_error_reason(volume.log) if volume.log else "Unknown" + failed_reasons[reason] = failed_reasons.get(reason, 0) + 1 + + log_lines = [ + f"Partition {self.mass_task.partition_key}", + f"Studies found: {study_count}", + f"Series found: {total_volumes}", + f"Processed: {total_processed}", + ] + if total_skipped: + log_lines.append(f"Skipped: {total_skipped}") + if total_failed: + log_lines.append(f"Failed: {total_failed}") + if failed_reasons: + log_lines.append("Failure reasons:") + for reason, count in failed_reasons.items(): + log_lines.append(f" {count}x {reason}") + + if total_volumes == 0: + status = MassTransferTask.Status.SUCCESS + message = "No series found for this partition." + elif total_failed and not total_processed: + status = MassTransferTask.Status.FAILURE + message = f"All {total_failed} series failed during mass transfer." + else: + total_series = total_processed + total_failed + total_skipped + parts = [f"{total_processed} downloaded"] + if total_failed: + parts.append(f"{total_failed} failed") + if total_skipped: + parts.append(f"{total_skipped} skipped") + + status = ( + MassTransferTask.Status.WARNING if total_failed else MassTransferTask.Status.SUCCESS + ) + message = f"{study_count} studies, {total_series} series ({', '.join(parts)})." + + return { + "status": status, + "message": message, + "log": "\n".join(log_lines), + } +``` + +- [ ] **Step 6: Run the full processor test file** + +Run: `uv run cli test -- adit/mass_transfer/tests/test_processor.py -v` +Expected: ALL PASS. Failures to watch for: mocked tests where `MassTransferVolume.objects.filter` is patched — with `attempts = 1` the resume branch is never entered so the patched `filter(...).delete()` still covers the wipe; pseudonym-mode tests (lines 1030-1146) rely on `bulk_create` returning the in-memory objects, which the new flow still uses on the fresh path. + +- [ ] **Step 7: Run the whole mass_transfer + core suites** + +Run: `uv run cli test -- adit/mass_transfer/ adit/core/ -v` +Expected: ALL PASS (acceptance tests are excluded by default marker config; if any run and fail on unrelated infrastructure, note it, don't chase it). + +- [ ] **Step 8: Commit** + +```bash +git add adit/mass_transfer/processors.py adit/mass_transfer/tests/test_processor.py +git commit -m "feat(mass_transfer): resume automatic retries from persisted volume progress" +``` + +--- + +### Task 5: Final verification and spec cross-check + +**Files:** +- Modify: none expected (fixups only) + +**Interfaces:** none. + +- [ ] **Step 1: Lint** + +Run: `uv run cli lint` +Expected: clean. If Ruff flags the unused `patient_id`/`study_uid` loop variables in `_transfer_grouped_series` (they were already unused before this change), rename to `_patient_id`/`_study_uid` only if the linter complains — otherwise leave as-is. + +- [ ] **Step 2: Full test suite** + +Run: `uv run cli test` +Expected: PASS (same set of skips/exclusions as on the base branch — compare against `git stash`-free baseline only if something unrelated fails). + +- [ ] **Step 3: Spec conformance check** + +Re-read `docs/superpowers/specs/2026-07-16-mass-transfer-resumable-retries-design.md` section by section and confirm: +- Wipe keyed on `attempts <= 1` ✓ (Task 4) +- Discovery guarded by row existence, not attempt number ✓ (Task 4) +- Retriable reset clears `status`/`retriable`/`log`/`converted_file` ✓ (Task 4) +- Per-series cleanup before re-export, folder destinations only ✓ (Task 3) +- Continue past retriable errors on every attempt; `_is_final_attempt()` deleted ✓ (Task 3) +- Single end-of-loop raise gated on `is_final_attempt` from the runner ✓ (Tasks 2+4) +- Summary computed from whole-partition volume state ✓ (Task 4) + +- [ ] **Step 4: Commit any fixups** + +```bash +git add -A ':!.claude' +git commit -m "chore(mass_transfer): lint/test fixups for resumable retries" +``` + +(Skip the commit if there is nothing to fix.) diff --git a/docs/superpowers/specs/2026-06-12-mass-transfer-final-attempt-continue-design.md b/docs/superpowers/specs/2026-06-12-mass-transfer-final-attempt-continue-design.md new file mode 100644 index 00000000..e1cc70c7 --- /dev/null +++ b/docs/superpowers/specs/2026-06-12-mass-transfer-final-attempt-continue-design.md @@ -0,0 +1,99 @@ +# Mass Transfer: Continue Past Dead Volumes on the Final Task Attempt + +**Date:** 2026-06-12 +**Status:** Approved + +## Problem + +A mass transfer task (one partition) processes its volumes (series) sequentially. When a +single volume exhausts the stamina network retries, the resulting `RetriableDicomError` is +re-raised from `_transfer_single_series` (`adit/mass_transfer/processors.py:588-591`), +aborting the entire task. Procrastinate retries the whole task up to +`DICOM_TASK_MAX_ATTEMPTS` (default 3); each retry wipes and re-fetches the whole partition +and hits the same dead series again. After the final attempt the task is marked `FAILURE`, +even when only 1-2 of its volumes are actually unrecoverable (e.g. series on +archived/offline PACS storage). + +There is no way to exclude specific series: volumes are ephemeral — every task run deletes +all `MassTransferVolume` rows for the partition and re-discovers them from the PACS +(`processors.py:341-349, 360-364`). + +## Decision + +Keep the existing abort-and-retry behavior on non-final attempts (a transient PACS outage +can recover during the 2-4 min Procrastinate waits). On the **final attempt only**, mark +the dead volume `ERROR` and continue with the remaining volumes, so the partition completes +as `WARNING` instead of `FAILURE`. + +This was chosen over two alternatives: + +- **Always continue, retry at end:** every attempt would re-fetch the entire partition + (more PACS load) for little gain. +- **Never retry per-volume errors:** simplest, but a multi-minute PACS blip would + permanently fail volumes that a task-level retry would have recovered. + +A larger follow-up (durable volumes with per-volume retry granularity) is planned +separately; this change is forward-compatible with it and will be simplified by it. + +## Design + +### Behavior + +- On attempts 1 to N-1 (where N = `settings.DICOM_TASK_MAX_ATTEMPTS`): unchanged. A + per-volume `RetriableDicomError` marks the volume `ERROR` with log + `"Transfer interrupted by retriable error; task will be retried."` and re-raises, + triggering a task-level retry. +- On the final attempt (`mass_task.attempts >= settings.DICOM_TASK_MAX_ATTEMPTS`): the + volume is marked `ERROR` with log + `"Transfer failed after exhausting retries: "` and the transfer loop continues + with the remaining volumes. No exception propagates. + +`DicomTask.attempts` is incremented and saved by the task runner before `process()` runs +(`adit/core/tasks.py:85-86`), so the processor subprocess reads an accurate count from the +database. + +### Code changes + +Single file: `adit/mass_transfer/processors.py`. + +1. Add a small `_is_final_attempt()` helper on `MassTransferTaskProcessor`: + `self.mass_task.attempts >= settings.DICOM_TASK_MAX_ATTEMPTS`. +2. Branch in the `except RetriableDicomError` handler of `_transfer_single_series`: + re-raise only when not the final attempt; otherwise set the error log and fall through + to the existing `finally` block (which saves the volume). + +No model changes, no migration, no UI changes. + +### Status outcomes (existing machinery, untouched) + +- `_transfer_grouped_series` already counts `ERROR` volumes into `total_failed` and + aggregates `failed_reasons`. +- `_build_task_summary` already yields `WARNING` for partial failure and `FAILURE` when + all volumes failed. +- A `RetriableDicomError` raised outside the per-volume loop (e.g. during series + discovery, before any volumes exist) still propagates and retries/fails the whole task. + +### Known edge (accepted) + +After cancel -> resume, `DicomTask.attempts` is not reset, so a resumed task may treat its +first resumed run as final and continue past dead volumes instead of retrying. This is a +graceful degradation and is documented in a code comment. + +## Testing + +New tests in `adit/mass_transfer/tests/test_processor.py`: + +1. Non-final attempt: per-volume `RetriableDicomError` is re-raised; volume is `ERROR` + with the "will be retried" log (existing behavior preserved). +2. Final attempt, one dead volume among healthy ones: no exception, task result `WARNING`, + dead volume `ERROR` with the exhausted-retries log, other volumes `EXPORTED`. +3. Final attempt, all volumes dead: task result `FAILURE`. +4. Boundary: `attempts == DICOM_TASK_MAX_ATTEMPTS - 1` aborts (re-raises); + `attempts == DICOM_TASK_MAX_ATTEMPTS` continues. + +## Operational note: rescuing prod mass transfer job 1 + +After deploying, use **Retry** on the job. It resets the 3 failed tasks (`attempts=0`) and +re-queues them. Each task burns attempts 1-2 against the dead series (~6-8 min of retry +waits), then attempt 3 completes the partition as `WARNING` with only the 1-2 dead volumes +marked `ERROR`. No manual exclusion needed. diff --git a/docs/superpowers/specs/2026-07-16-mass-transfer-resumable-retries-design.md b/docs/superpowers/specs/2026-07-16-mass-transfer-resumable-retries-design.md new file mode 100644 index 00000000..a0054c6c --- /dev/null +++ b/docs/superpowers/specs/2026-07-16-mass-transfer-resumable-retries-design.md @@ -0,0 +1,192 @@ +# Mass Transfer: Resumable Automatic Retries (Durable Volume Progress) + +**Date:** 2026-07-16 +**Status:** Approved +**Supersedes:** `2026-06-12-mass-transfer-final-attempt-continue-design.md` + +## Problem + +A mass transfer task (one partition) wipes all of its state at the start of every run: +the destination partition folder is `rmtree`'d and all `MassTransferVolume` rows for the +partition are deleted (`adit/mass_transfer/processors.py:337-345`). Because of this, a +task retry is all-or-nothing: + +- A single volume that exhausts the stamina network retries raises `RetriableDicomError`, + which aborts the whole partition. Procrastinate retries the task, the wipe discards all + progress, every healthy series is re-fetched from the PACS, and the same dead series + aborts the run again — until attempts are exhausted and the task ends `FAILURE` with an + incomplete transfer. +- The stop-gap fix (2026-06-12 design) continues past dead volumes on the *final* attempt + only, turning the outcome into `WARNING`. It works, but it introduces an asymmetry + between attempts, still re-fetches the entire partition on every retry, and carries an + accepted edge case: after cancel → resume, `DicomTask.attempts` is not reset, so a + resumed run may wrongly treat its first attempt as final. + +The root cause is the unconditional wipe. The data model already supports durable +progress: `MassTransferVolume` rows persist per-series status, log, pseudonym, and +pseudonymized UIDs, with a `(job, series_instance_uid)` unique constraint. + +## Decision + +Make automatic Procrastinate retries **resume** instead of restart: + +- Wipe (folder + volume rows) only on the first attempt of a queue cycle + (`DicomTask.attempts <= 1`). User-initiated **Retry**/**Restart** reset `attempts` to 0 + via `reset_tasks()`, so they keep today's clean-slate semantics. +- On an automatic retry, skip discovery, reuse the existing volume rows, and re-process + only the volumes that have not completed. +- A per-volume `RetriableDicomError` no longer aborts the loop on *any* attempt: mark the + volume and continue with the rest of the partition. +- Retry scheduling happens once, at the end of the loop, by raising a single + `RetriableDicomError` — but only when retriable volumes remain *and* attempts remain. + +This supersedes the final-attempt special case, which is removed. + +### Rejected alternatives + +- **Keep the wipe, continue-and-raise-at-end on every attempt:** no model change, but + every retry re-transfers the whole partition — the PACS-load objection from the + 2026-06-12 design stands. +- **Per-volume attempt counters:** only matters for a volume that first fails late in the + attempt sequence (it gets fewer tries than max). Not worth an extra field and the + bookkeeping; the task-level attempt cap bounds every volume adequately. +- **Detect the final attempt from `DicomTask.attempts` in the processor (stop-gap + approach):** wrong after cancel → resume because `DicomTask.attempts` is cumulative + across queue cycles while Procrastinate's retry budget is per queued job. The runner + already computes the correct predicate (`adit/core/tasks.py:134`); pass it down instead. + +## Design + +### Model change + +Add to `MassTransferVolume`: + +```python +retriable = models.BooleanField(default=False) +``` + +Kept alongside `status=ERROR` (not a new status value), so the UI and the existing +summary logic are unaffected. One migration; no data migration needed (existing rows +default to `False`). + +Add to `MassTransferTask`: + +```python +anonymizer_seed = models.CharField(max_length=64, blank=True, default="") +``` + +Random-mode pseudonymization (no job-wide `pseudonym_salt`) otherwise constructs a new +`Pseudonymizer()` with a fresh random key on every attempt, so a resumed run would +pseudonymize UIDs/dates differently than the volumes already transferred by a previous +attempt. `anonymizer_seed` is generated on the first attempt of a queue cycle +(`is_fresh_cycle`, or when unset) and reused by every automatic retry within that cycle, +so the anonymizer key stays stable across attempts. It is regenerated by the next fresh +cycle (user-initiated Retry/Restart), matching the wipe-and-rediscover semantics of that +path. + +### Core change: pass `is_final_attempt` into the processor + +`_run_dicom_task` (`adit/core/tasks.py`) computes +`is_final_attempt = context.job.attempts + 1 >= settings.DICOM_TASK_MAX_ATTEMPTS` +(Procrastinate's `attempts` is 0-indexed — same arithmetic as the existing handler at +line 134) and passes it through `_process_dicom_task` into the subprocess, which sets it +as an attribute on the processor (`DicomTaskProcessor.is_final_attempt`, default +`False`). This uses the per-queue-cycle Procrastinate counter, so the cancel → resume +edge case of the stop-gap design disappears. + +### Processor flow (`adit/mass_transfer/processors.py`) + +`process()`: + +1. **Wipe only on a fresh cycle:** if `self.mass_task.attempts <= 1`, delete the + partition folder and the partition's volume rows (exactly today's cleanup). Otherwise + skip the wipe. +2. **Discovery only when needed:** if no volume rows exist for the partition, run + discovery and bulk-create `PENDING` volumes (today's behavior). Otherwise load the + existing rows. (The row-existence check, not the attempt number, guards discovery — + this also covers a task canceled before discovery finished and resumed later.) +3. **Reset retriable volumes:** bulk-update the partition's volumes with + `status=ERROR, retriable=True` back to `status=PENDING, retriable=False, log=""`, + clearing `converted_file`. The transfer loop then processes only `PENDING` volumes; + `EXPORTED` / `CONVERTED` / `SKIPPED` and permanent-`ERROR` volumes are never touched + again. Studies whose volumes are all complete are skipped entirely (no association, + no pacing delay). +4. **Per-series cleanup before re-export:** for folder destinations, if the volume's + series folder exists, delete it before exporting. The path is fully deterministic from + the volume row (`subject_id` from pseudonym/patient_id, study folder from + description + datetime, series folder from description + number + UID) — this replaces + the partition-wide `rmtree` as the defense against partially-written series + (`_export_series_to_folder` writes non-atomically). Server destinations need no + cleanup: re-running C-STORE is idempotent per SOP Instance UID — provided the + pseudonymized UIDs are stable across attempts, which the persisted per-cycle + `anonymizer_seed` guarantees for random mode (salted jobs are inherently stable since + the salt is fixed on the job, and non-pseudonymized transfers keep the original + source UIDs). +5. **Continue past retriable errors:** in `_transfer_single_series`, the + `except RetriableDicomError` branch sets `status=ERROR, retriable=True`, logs the + error, and returns normally (no re-raise, no final-attempt branch). + `_is_final_attempt()` and its special case are deleted. +6. **Raise once at the end:** after the loop, if any volume in the partition has + `status=ERROR, retriable=True` and `not self.is_final_attempt`, raise + `RetriableDicomError(f"{n} of {total} volumes failed retriably and will be retried.")`. + The existing runner handler marks the task `PENDING` ("Task failed, but will be + retried.") and Procrastinate reschedules with backoff. On the final attempt, return + the summary instead — retriable errors count as failures in it. + +### Summary computed from the database + +`_build_task_summary` currently aggregates loop-local counters, which would undercount on +a resumed run (the loop only sees the remaining volumes). Recompute the summary from the +partition's volume rows at the end of the run: processed = `EXPORTED` + `CONVERTED`, +skipped = `SKIPPED`, failed = `ERROR`, study count = distinct `study_instance_uid`, +failure reasons via `_short_error_reason(volume.log)`. Status mapping is unchanged +(`WARNING` for partial failure, `FAILURE` when all volumes failed, `SUCCESS` otherwise). + +### Behavior summary by entry path + +| Entry path | `DicomTask.attempts` on arrival | Behavior | +|---|---|---| +| Fresh task | 0 → runs as 1 | Wipe, discover, transfer (today's behavior) | +| Automatic retry (`RetriableDicomError`) | ≥ 1 → runs as ≥ 2 | Resume: no wipe, no discovery, only retriable/pending volumes re-processed | +| UI **Retry** / **Restart** (`reset_tasks()`) | reset to 0 → runs as 1 | Clean slate, as today | +| Cancel → **Resume** | not reset → runs as ≥ 2 | Resumes from existing volumes (improvement: previously redid the whole partition) | +| Worker killed mid-partition, Procrastinate retries | ≥ 1 → runs as ≥ 2 | Resume; the in-flight volume is still `PENDING`, its series folder is cleaned and re-exported | + +## Error handling + +- `RetriableDicomError` raised *outside* the per-volume loop (discovery, before volumes + exist) still propagates and retries the whole task — unchanged, and safe: with no rows + created, the retry re-runs discovery. +- Permanent per-volume exceptions keep today's behavior: `status=ERROR`, + `retriable=False`, loop continues. +- Zero-image fetches keep their current statuses (`SKIPPED` / permanent `ERROR`). +- The volume-save failure path and the PENDING-after-transfer guard in the `finally` + block are unchanged. + +## Testing + +Rework/extend `adit/mass_transfer/tests/test_processor.py`: + +1. Per-volume `RetriableDicomError` on a non-final attempt: volume `ERROR` + + `retriable=True`, remaining volumes still transferred, task raises + `RetriableDicomError` at the end. +2. Final attempt (`processor.is_final_attempt = True`): no raise, task result `WARNING`, + dead volume `ERROR`, healthy volumes `EXPORTED`. +3. Resume run (`attempts >= 2`, existing volume rows): no wipe, no discovery call, + completed volumes untouched (status, files on disk), only retriable/pending volumes + re-processed; retriable volumes reset to `PENDING` before the loop. +4. Per-series cleanup: a partially-written series folder from a prior attempt is deleted + before re-export; sibling completed series folders survive. +5. All volumes dead on final attempt: task result `FAILURE`. +6. Summary correctness on a resumed run: counts reflect the whole partition, not just the + volumes processed in the final run. +7. Core runner: `is_final_attempt` is `True` exactly when + `context.job.attempts + 1 >= DICOM_TASK_MAX_ATTEMPTS`. + +## Relationship to the stop-gap PR + +The 2026-06-12 stop-gap (`fix/mass-transfer-final-attempt-continue`) remains valid as an +immediate rescue for prod mass transfer job 1. This design deletes its special case +(`_is_final_attempt()` and the final-attempt branch) as part of implementation; whether +the stop-gap merges first or is replaced on the branch is a sequencing choice at +implementation time.