Add Mass Transfer app with partitioned DICOM export and conversion

.github/workflows/ci.yml

@@ -35,6 +35,7 @@ jobs:
             adit_dev-web:latest
             adit_dev-default_worker:latest
             adit_dev-dicom_worker:latest
+            adit_dev-mass_transfer_worker:latest
             adit_dev-receiver:latest
           cache-from: type=gha
           cache-to: type=gha,mode=max

AGENTS.md

@@ -116,6 +116,10 @@ Key variables in `.env` (see `example.env`):
 - **Type Checking**: pyright in basic mode (migrations excluded)
 - **Linting**: Ruff with E, F, I, DJ rules
 
+### Assertions
+
+- Use `assert` for internal programming error checks (preconditions, invariants). Do not replace with `ValueError` or similar — this app is never run with `python -O`.
+
 ### Django Field Conventions
 
 - Text/char fields: use `blank=True` alone (not `null=True`)

adit/core/management/commands/cleanup_jobs_and_tasks.py

@@ -4,21 +4,24 @@
 from adit.batch_query.models import BatchQueryJob, BatchQueryTask
 from adit.batch_transfer.models import BatchTransferJob, BatchTransferTask
 from adit.core.models import DicomJob, DicomTask
+from adit.mass_transfer.models import MassTransferJob, MassTransferTask
 from adit.selective_transfer.models import SelectiveTransferJob, SelectiveTransferTask
 
 
 class Command(BaseCommand):
     help = "Cleanup all DICOM jobs and tasks that are stuck."
 
     def cleanup_tasks(self, model: type[DicomTask]):
+        job_model: type[DicomJob] | None = model._meta.get_field("job").related_model  # type: ignore[assignment]
+        assert job_model is not None
         job_ids = set()
 
         message = "Unexpected crash while processing this task."
         task_log = "The worker crashed unexpectedly and this task was manually set to failed."
 
         tasks_in_progress = model.objects.filter(status=model.Status.IN_PROGRESS).all()
         for task in tasks_in_progress:
-            task.status = SelectiveTransferTask.Status.FAILURE
+            task.status = model.Status.FAILURE
             task.message = message
             task.log = task_log
             task.save()
@@ -27,14 +30,14 @@ def cleanup_tasks(self, model: type[DicomTask]):
         tasks_pending = model.objects.filter(Q(status=model.Status.PENDING)).all()
         for task in tasks_pending:
             if task.queued_job_id is None:
-                task.status = SelectiveTransferTask.Status.FAILURE
+                task.status = model.Status.FAILURE
                 task.message = message
                 task.log = task_log
                 task.save()
                 job_ids.add(task.job_id)
 
         for job_id in job_ids:
-            job = SelectiveTransferJob.objects.get(id=job_id)
+            job = job_model.objects.get(id=job_id)
             job.post_process(suppress_email=True)
 
     def cleanup_jobs(self, model: type[DicomJob]):
@@ -45,7 +48,7 @@ def cleanup_jobs(self, model: type[DicomJob]):
         ).all()
 
         for job in jobs:
-            job.status = SelectiveTransferJob.Status.FAILURE
+            job.status = model.Status.FAILURE
             job.message = message
             job.save()
 
@@ -65,5 +68,7 @@ def handle(self, *args, **options):
         self.cleanup_jobs(BatchQueryJob)
         self.cleanup_tasks(BatchTransferTask)
         self.cleanup_jobs(BatchTransferJob)
+        self.cleanup_tasks(MassTransferTask)
+        self.cleanup_jobs(MassTransferJob)
 
         self.stdout.write("Done")

adit/core/migrations/0016_add_max_search_results_to_dicomserver.py

@@ -0,0 +1,18 @@
+# Generated by Django 5.2.8 on 2026-03-09 12:00
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('core', '0015_delete_queuedtask'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='dicomserver',
+            name='max_search_results',
+            field=models.PositiveIntegerField(default=200),
+        ),
+    ]

adit/core/migrations/0017_review_fixes.py

@@ -0,0 +1,21 @@
+# Generated by Django 6.0.3 on 2026-03-29 13:52
+
+import django.core.validators
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("core", "0016_add_max_search_results_to_dicomserver"),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name="dicomserver",
+            name="max_search_results",
+            field=models.PositiveIntegerField(
+                default=200, validators=[django.core.validators.MinValueValidator(1)]
+            ),
+        ),
+    ]

adit/core/models.py

@@ -155,6 +155,11 @@ class DicomServer(DicomNode):
     dicomweb_stow_prefix = models.CharField(blank=True, max_length=2000)
     dicomweb_authorization_header = models.CharField(blank=True, max_length=2000)
 
+    # C-FIND result limit before recursive time-window splitting
+    max_search_results = models.PositiveIntegerField(
+        default=200, validators=[MinValueValidator(1)]
+    )
+
     objects: DicomNodeManager["DicomServer"] = DicomNodeManager["DicomServer"]()
 
 
@@ -391,7 +396,7 @@ class Status(models.TextChoices):
     job = models.ForeignKey(DicomJob, on_delete=models.CASCADE, related_name="tasks")
     source_id: int
     source = models.ForeignKey(DicomNode, related_name="+", on_delete=models.PROTECT)
-    queued_job_id: int
+    queued_job_id: int | None
     queued_job = models.OneToOneField(
         ProcrastinateJob, null=True, on_delete=models.SET_NULL, related_name="+"
     )
@@ -417,6 +422,7 @@ def __str__(self) -> str:
 
     def get_absolute_url(self) -> str: ...
 
+
     def queue_pending_task(self) -> None:
         """Queues a dicom task."""
         assert self.status == DicomTask.Status.PENDING

adit/core/tasks.py

@@ -53,32 +53,30 @@ def backup_db(*args, **kwargs):
     call_command("dbbackup", "--clean", "-v 2")
 
 
-@app.task(
-    queue="dicom",
-    pass_context=True,
-    # TODO: Increase the priority slightly when it will be retried
-    # See https://github.com/procrastinate-org/procrastinate/issues/1096
-    #
-    # Two-level retry strategy:
-    # 1. Network layer (Stamina): Fast retries for transient failures (5-10 attempts)
-    #    - Applied at DIMSE/DICOMweb connector level
-    #    - Handles: connection timeouts, HTTP 503, temporary server unavailability
-    # 2. Task layer (Procrastinate): Slow retries for complete operation failures
-    #    - Applied here (max_attempts below)
-    #    - Only triggers after network-level retries are exhausted
-    #    - Retries the entire task
-    retry=RetryStrategy(
-        max_attempts=settings.DICOM_TASK_MAX_ATTEMPTS,
-        wait=settings.DICOM_TASK_RETRY_WAIT,
-        linear_wait=settings.DICOM_TASK_LINEAR_WAIT,
-        retry_exceptions={RetriableDicomError},
-    ),
+DICOM_TASK_RETRY_STRATEGY = RetryStrategy(
+    max_attempts=settings.DICOM_TASK_MAX_ATTEMPTS,
+    wait=settings.DICOM_TASK_RETRY_WAIT,
+    linear_wait=settings.DICOM_TASK_LINEAR_WAIT,
+    retry_exceptions={RetriableDicomError},
 )
-def process_dicom_task(context: JobContext, model_label: str, task_id: int):
+
+
+def _run_dicom_task(
+    context: JobContext,
+    model_label: str,
+    task_id: int,
+    *,
+    process_timeout: int | None = None,
+):
     assert context.job
 
     dicom_task = get_dicom_task(model_label, task_id)
-    assert dicom_task.status == DicomTask.Status.PENDING
+    # The assertion status == PENDING assumed that tasks always arrive fresh,
+    # but in reality a retried task can arrive in a half-finished state.
+    # A task may still be IN_PROGRESS if the worker was killed before the
+    # finally block could update its status.  Accept both PENDING and
+    # IN_PROGRESS so the retry can proceed.
+    assert dicom_task.status in (DicomTask.Status.PENDING, DicomTask.Status.IN_PROGRESS)
 
     # When the first DICOM task of a job is processed then the status of the
     # job switches from PENDING to IN_PROGRESS
@@ -96,7 +94,7 @@ def process_dicom_task(context: JobContext, model_label: str, task_id: int):
 
     logger.info(f"Processing of {dicom_task} started.")
 
-    @concurrent.process(timeout=settings.DICOM_TASK_PROCESS_TIMEOUT, daemon=True)
+    @concurrent.process(timeout=process_timeout, daemon=True)
     def _process_dicom_task(model_label: str, task_id: int) -> ProcessingResult:
         dicom_task = get_dicom_task(model_label, task_id)
         processor = get_dicom_processor(dicom_task)
@@ -121,11 +119,18 @@ def _monitor_task(context: JobContext, future: ProcessFuture) -> None:
         dicom_task.log = result["log"]
         ensure_db_connection()
 
+    except futures.CancelledError:
+        dicom_task.status = DicomTask.Status.CANCELED
+        dicom_task.message = "Task was canceled."
+        ensure_db_connection()
+
+
     except futures.TimeoutError:
         dicom_task.message = "Task was aborted due to timeout."
         dicom_task.status = DicomTask.Status.FAILURE
         ensure_db_connection()
 
+
     except RetriableDicomError as err:
         logger.exception("Retriable error occurred during %s.", dicom_task)
 
@@ -146,6 +151,7 @@ def _monitor_task(context: JobContext, future: ProcessFuture) -> None:
             dicom_task.message = str(err)
 
         ensure_db_connection()
+
         raise err
 
     except Exception as err:
@@ -162,6 +168,7 @@ def _monitor_task(context: JobContext, future: ProcessFuture) -> None:
 
         ensure_db_connection()
 
+
     finally:
         dicom_task.end = timezone.now()
         dicom_task.save()
@@ -176,3 +183,25 @@ def _monitor_task(context: JobContext, future: ProcessFuture) -> None:
 
         # TODO: https://github.com/procrastinate-org/procrastinate/issues/1106
         db.close_old_connections()
+
+
+@app.task(
+    queue="dicom",
+    pass_context=True,
+    # TODO: Increase the priority slightly when it will be retried
+    # See https://github.com/procrastinate-org/procrastinate/issues/1096
+    #
+    # Two-level retry strategy:
+    # 1. Network layer (Stamina): Fast retries for transient failures (5-10 attempts)
+    #    - Applied at DIMSE/DICOMweb connector level
+    #    - Handles: connection timeouts, HTTP 503, temporary server unavailability
+    # 2. Task layer (Procrastinate): Slow retries for complete operation failures
+    #    - Applied here (max_attempts below)
+    #    - Only triggers after network-level retries are exhausted
+    #    - Retries the entire task
+    retry=DICOM_TASK_RETRY_STRATEGY,
+)
+def process_dicom_task(context: JobContext, model_label: str, task_id: int):
+    _run_dicom_task(
+        context, model_label, task_id, process_timeout=settings.DICOM_TASK_PROCESS_TIMEOUT
+    )

adit/core/templates/core/home.html

@@ -46,6 +46,12 @@ <h1 class="display-4 d-flex align-items-center">
         <dd>
             Transfer or download multiple studies specified in a batch file.
         </dd>
+        <dt>
+            <a href="{% url 'mass_transfer_job_create' %}">Mass Transfer</a>
+        </dt>
+        <dd>
+            Transfer large volumes of imaging data over a time range using reusable filters.
+        </dd>
         <dt>
             <a href="{% url 'dicom_explorer_form' %}">DICOM Explorer</a>
         </dt>

adit/core/tests/utils/test_dimse_connector.py

@@ -158,3 +158,68 @@ def test_abort_connection_with_no_connection(self):
 
         # Assert
         assert connector.assoc is None
+
+    def test_service_switch_closes_and_reopens_connection(self, mocker):
+        """Test that switching services (e.g. C-FIND -> C-GET) closes the old connection
+        and opens a new one with the correct presentation contexts."""
+        server = DicomServerFactory.create()
+        connector = DimseConnector(server, auto_connect=True)
+
+        associate_mock = mocker.patch("adit.core.utils.dimse_connector.AE.associate")
+
+        # First association for C-FIND
+        find_assoc = create_association_mock()
+        find_assoc.is_alive.return_value = True
+        find_assoc.send_c_find.return_value = DicomTestHelper.create_successful_c_find_responses(
+            [{"PatientID": "12345", "QueryRetrieveLevel": "STUDY"}]
+        )
+
+        # Second association for C-GET
+        get_assoc = create_association_mock()
+        get_assoc.is_alive.return_value = True
+        get_assoc.send_c_get.return_value = DicomTestHelper.create_successful_c_get_response()
+
+        associate_mock.side_effect = [find_assoc, get_assoc]
+
+        # Act: perform a C-FIND
+        query = QueryDataset.create(
+            PatientID="12345",
+            StudyInstanceUID="1.2.3.4.5",
+            QueryRetrieveLevel="STUDY",
+        )
+        list(connector.send_c_find(query))
+
+        # After C-FIND with auto_close, connection is closed
+        assert connector.assoc is None
+
+        # Now open a persistent connection for C-FIND, then switch to C-GET
+        connector.auto_close = False
+
+        # Reset the mock for the persistent connections
+        find_assoc2 = create_association_mock()
+        find_assoc2.is_alive.return_value = True
+        find_assoc2.send_c_find.return_value = DicomTestHelper.create_successful_c_find_responses(
+            [{"PatientID": "12345", "QueryRetrieveLevel": "STUDY"}]
+        )
+
+        get_assoc2 = create_association_mock()
+        get_assoc2.is_alive.return_value = True
+        get_assoc2.send_c_get.return_value = DicomTestHelper.create_successful_c_get_response()
+
+        associate_mock.side_effect = [find_assoc2, get_assoc2]
+
+        # C-FIND with auto_close=False keeps connection open
+        list(connector.send_c_find(query))
+        assert connector.assoc is find_assoc2
+        assert connector._current_service == "C-FIND"
+
+        # C-GET should close the C-FIND connection and open a new one
+        store_handler = MagicMock()
+        store_errors = []
+        connector.send_c_get(query, store_handler, store_errors)
+
+        # The C-FIND association should have been released (closed)
+        assert find_assoc2.release.called
+        # The connector should now be on the C-GET association
+        assert connector.assoc is get_assoc2
+        assert connector._current_service == "C-GET"