Suspend destination folder on systemic errors (disk full, stale NFS) - #384
Suspend destination folder on systemic errors (disk full, stale NFS)#384NumericalAdvantage wants to merge 2 commits into
Conversation
When a systemic infrastructure error occurs during mass transfer (e.g. ENOSPC, stale NFS handle, read-only filesystem), the processor now: 1. Stops processing remaining series immediately instead of grinding through them all 2. Sets DicomFolder.suspended = True on the destination 3. Sends an alert email to admins 4. Returns FAILURE for the current task Subsequent tasks for the same destination check the suspended flag at the start of process() and return WARNING without touching the PACS, preventing thousands of wasted queries. Non-systemic errors (bad DICOM data, conversion failures) continue to be handled per-series as before.
📝 WalkthroughWalkthroughAdds a persistent ChangesDICOM destination suspension
Estimated code review effort: 3 (Moderate) | ~20 minutes Suggested reviewers: Sequence Diagram(s)sequenceDiagram
participant process
participant DicomFolder
participant SeriesTransfer
participant Administrators
process->>DicomFolder: Check suspended state
process->>SeriesTransfer: Transfer series
SeriesTransfer-->>process: Return systemic error
process->>DicomFolder: Set suspended=True
process->>Administrators: Send failure alert
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 5
🧹 Nitpick comments (1)
adit/mass_transfer/processors.py (1)
78-92: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winClassify systemic errors by exception chain, not by message text.
The
DicomErrorbranch matches the substring"Out of disk space". This couples the suspension behavior to one exact message string produced elsewhere. If that message is reworded or localized,_is_systemic_errorreturnsFalse. The destination is then never suspended, and the task falls back to per-series continuation on a full disk. The failure is silent.The check also misses a systemic
OSErrorwrapped in any other exception type, for example a conversion error raised from_convert_series.Walk
__cause__and__context__instead. The errno survives rewording. Keep the message test as a fallback for the case where noOSErroris chained.♻️ Proposed refactor
def _is_systemic_error(err: Exception) -> bool: """Return True if the error indicates a systemic infrastructure problem. Systemic errors (disk full, stale NFS handle, read-only filesystem, I/O error) affect all remaining series equally, so continuing is pointless. + + The exception chain is inspected so that a systemic OSError is still detected + when it is wrapped in a DicomError or a conversion error. """ - if isinstance(err, OSError) and err.errno in _SYSTEMIC_ERRNOS: - return True + seen: set[int] = set() + current: BaseException | None = err + while current is not None and id(current) not in seen: + seen.add(id(current)) + if isinstance(current, OSError) and current.errno in _SYSTEMIC_ERRNOS: + return True + current = current.__cause__ or current.__context__ if isinstance(err, DicomError) and "Out of disk space" in str(err): return True return FalseConfirm where the disk-space
DicomErrormessage is raised, and whether it chains the originalOSError:#!/bin/bash # Locate the raise site of the disk-space message and check for exception chaining. rg -n -C6 'Out of disk space' --type=py # Check whether DicomError raises in the core write path preserve the cause. rg -n -C4 'raise DicomError\(' --type=py | rg -n -C4 'from '🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@adit/mass_transfer/processors.py` around lines 78 - 92, Update _is_systemic_error to traverse the exception’s __cause__ and __context__ chain, classifying any chained OSError with an errno in _SYSTEMIC_ERRNOS as systemic, including OSErrors wrapped by other exception types. Preserve the “Out of disk space” DicomError message check only as a fallback when no chained OSError matches, and avoid relying on that message for errors with a non-systemic chain.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@adit/core/models.py`:
- Around line 177-180: Add blank=True to the DicomFolder.suspended BooleanField
while keeping default=False and excluding null=True, so the Django admin accepts
an unchecked value and the field remains a two-state boolean.
In `@adit/mass_transfer/processors.py`:
- Around line 699-700: Update the systemic-error handling in
_transfer_grouped_series to reconcile every unattempted volume still in PENDING,
marking it ERROR or SKIPPED and emitting an explanatory log before propagating
the systemic failure. Ensure this reconciliation runs when the systemic handler
at line 413 is reached, including volumes not yet processed. Revise the
_transfer_grouped_series docstring to state its actual exception contract rather
than claiming it only raises RetriableDicomError.
- Around line 424-426: The source operator created at line 392 with
persistent=True is not being closed in the new finally block, leaving the DIMSE
association open if _discover_series or _create_pending_volumes raises an
exception. Assign the source operator at line 392 to a variable name that the
finally block can access (the same variable name used elsewhere in the function
for clarity), then add a close call for this source operator in the finally
block alongside the existing dest_operator.close() call. Since
DicomOperator.close() safely handles being called multiple times, the source
operator can be safely closed in the finally block even if it was already closed
on the success path.
- Around line 439-457: Wrap the send_mail_to_admins call in error handling to
prevent SMTP exceptions from propagating out of _suspend_destination and
altering the task result, since the suspension is already persisted.
Additionally, use a conditional update when saving folder.suspended so the mail
alert is sent only when this call transitions the folder from suspended=False to
suspended=True, preventing duplicate emails when multiple concurrent tasks hit
the same error.
In `@adit/mass_transfer/tests/test_processor.py`:
- Around line 2499-2520: Correct test_process_stops_on_systemic_error’s
docstring to describe the DicomError message branch, bind the
send_mail_to_admins patch and assert it is called once, and add a separate
end-to-end test for OSError(errno.ENOSPC, ...) that verifies failure,
destination suspension, and a single administrator alert.
---
Nitpick comments:
In `@adit/mass_transfer/processors.py`:
- Around line 78-92: Update _is_systemic_error to traverse the exception’s
__cause__ and __context__ chain, classifying any chained OSError with an errno
in _SYSTEMIC_ERRNOS as systemic, including OSErrors wrapped by other exception
types. Preserve the “Out of disk space” DicomError message check only as a
fallback when no chained OSError matches, and avoid relying on that message for
errors with a non-systemic chain.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: 095128f7-b412-46cf-84ad-252fff27e2e3
📒 Files selected for processing (4)
adit/core/migrations/0018_dicomfolder_suspended.pyadit/core/models.pyadit/mass_transfer/processors.pyadit/mass_transfer/tests/test_processor.py
| suspended = models.BooleanField( | ||
| default=False, | ||
| help_text="Suspended destinations skip processing (e.g. disk full).", | ||
| ) |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Find admin registrations and ModelForms bound to DicomFolder.
fd -e py | xargs rg -n -C4 'DicomFolder'Repository: openradx/adit
Length of output: 45720
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "=== DicomFolder model and admin ==="
sed -n '163,182p' adit/core/models.py
sed -n '66,78p' adit/core/admin.py
echo "=== suspended/help_text references ==="
rg -n -C3 'suspended|unsuspend|Suspended' adit/core adit -g '*.py'
echo "=== BooleanField without blank in migrations ==="
python3 - <<'PY'
from pathlib import Path
import re
text = Path('adit/core/migrations/0001_initial.py').read_text()
checks = re.findall(r"models.BooleanField\([^)]*?\)", text, re.S)
for c in checks:
print(c.replace('\n', ' ').replace(' ', ' ').strip())
PYRepository: openradx/adit
Length of output: 23694
Add blank=True to the DicomFolder.suspended field.
DicomFolderAdmin exposes DicomFolder in Django admin. A BooleanField without blank=True is rendered as required, so an administrator cannot save the folder while suspended is unchecked. Keep this as a two-state boolean; do not add null=True, because processors.py reads it as a plain suspension gate.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@adit/core/models.py` around lines 177 - 180, Add blank=True to the
DicomFolder.suspended BooleanField while keeping default=False and excluding
null=True, so the Django admin accepts an unchecked value and the field remains
a two-state boolean.
Source: Coding guidelines
| finally: | ||
| if dest_operator: | ||
| dest_operator.close() |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Close the source operator in the new finally block.
The new finally closes only dest_operator. The source operator is created at line 392 with persistent=True and is closed at line 396 on the success path, or inside _transfer_grouped_series per study. If _discover_series at line 395 or _create_pending_volumes at line 399 raises, neither close runs. The persistent DIMSE association to the source PACS stays open until the worker process exits.
_discover_series raises DicomError on its own paths, for example the window-too-small case at line 1064. So this path is reachable.
Bind the operator to a variable that the finally block can see, then close it there. DicomOperator.close() already swallows its own exceptions, so a double close is safe.
🔧 Proposed fix
dest_operator: DicomOperator | None = None
output_base: Path | None = None
+ operator: DicomOperator | None = None
if destination_node.node_type == DicomNode.NodeType.SERVER:- if dest_operator:
+ if operator:
+ operator.close()
+ if dest_operator:
dest_operator.close()Assign through the same name at line 392 so the finally block observes it:
operator = DicomOperator(source_node.dicomserver, persistent=True)🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@adit/mass_transfer/processors.py` around lines 424 - 426, The source operator
created at line 392 with persistent=True is not being closed in the new finally
block, leaving the DIMSE association open if _discover_series or
_create_pending_volumes raises an exception. Assign the source operator at line
392 to a variable name that the finally block can access (the same variable name
used elsewhere in the function for clarity), then add a close call for this
source operator in the finally block alongside the existing
dest_operator.close() call. Since DicomOperator.close() safely handles being
called multiple times, the source operator can be safely closed in the finally
block even if it was already closed on the success path.
| folder.suspended = True | ||
| folder.save(update_fields=["suspended"]) | ||
|
|
||
| logger.critical( | ||
| "Destination folder '%s' suspended due to systemic error (job %d): %s", | ||
| destination_node.name, | ||
| job.pk, | ||
| err, | ||
| ) | ||
|
|
||
| from adit.core.utils.mail import send_mail_to_admins | ||
|
|
||
| send_mail_to_admins( | ||
| f"Mass transfer destination '{destination_node.name}' suspended", | ||
| f"Destination folder '{destination_node.name}' was automatically suspended " | ||
| f"due to a systemic error during mass transfer job {job.pk}.\n\n" | ||
| f"Error: {err}\n\n" | ||
| f"Please fix the underlying issue and unsuspend the folder in the admin panel.", | ||
| ) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Contain mail failures and make the suspension idempotent.
Two problems affect this alerting path:
-
send_mail_to_adminsperforms SMTP I/O with no error handling. If the mail backend is unavailable or raises, the exception propagates out of_suspend_destination, then out of theexcept Exceptionhandler at line 412. The caller loses the intendedFAILUREresult dict and the task fails with an unhandled exception instead. The suspension is already persisted at that point, so the two outcomes diverge. Wrap the mail call so an alert failure cannot change the task result. -
Mass transfer runs many partition tasks in parallel against one destination. Every in-flight task that hits the same full disk calls
_suspend_destination. Each call saves the row again and sends another administrator email. A single disk-full event then produces one email per in-flight partition. Use a conditional update and alert only when this call performed the transition.
🔧 Proposed fix
folder = destination_node.dicomfolder
- folder.suspended = True
- folder.save(update_fields=["suspended"])
+ # Atomic transition so concurrent partition tasks alert only once.
+ transitioned = (
+ type(folder).objects.filter(pk=folder.pk, suspended=False).update(suspended=True) == 1
+ )
+ folder.suspended = True
+ if not transitioned:
+ return
logger.critical(
"Destination folder '%s' suspended due to systemic error (job %d): %s",
destination_node.name,
job.pk,
err,
)
from adit.core.utils.mail import send_mail_to_admins
- send_mail_to_admins(
- f"Mass transfer destination '{destination_node.name}' suspended",
- f"Destination folder '{destination_node.name}' was automatically suspended "
- f"due to a systemic error during mass transfer job {job.pk}.\n\n"
- f"Error: {err}\n\n"
- f"Please fix the underlying issue and unsuspend the folder in the admin panel.",
- )
+ try:
+ send_mail_to_admins(
+ f"Mass transfer destination '{destination_node.name}' suspended",
+ f"Destination folder '{destination_node.name}' was automatically suspended "
+ f"due to a systemic error during mass transfer job {job.pk}.\n\n"
+ f"Error: {err}\n\n"
+ f"Please fix the underlying issue and unsuspend the folder in the admin panel.",
+ )
+ except Exception:
+ logger.exception(
+ "Failed to send suspension alert for destination '%s'", destination_node.name
+ )🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@adit/mass_transfer/processors.py` around lines 439 - 457, Wrap the
send_mail_to_admins call in error handling to prevent SMTP exceptions from
propagating out of _suspend_destination and altering the task result, since the
suspension is already persisted. Additionally, use a conditional update when
saving folder.suspended so the mail alert is sent only when this call
transitions the folder from suspended=False to suspended=True, preventing
duplicate emails when multiple concurrent tasks hit the same error.
| if _is_systemic_error(err): | ||
| raise |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift
Reconcile the remaining PENDING volumes after a systemic abort, and update the stale contract.
The re-raise aborts the loop in _transfer_grouped_series. Volumes that were never attempted keep Status.PENDING. _build_task_summary is never reached, so nothing reconciles them.
A retry normally repairs this, because lines 381-384 delete and recreate the partition volumes. That repair no longer happens here. The destination is suspended by the same failure, so every later task for it returns early at line 339 with WARNING. The orphaned PENDING volumes then stay in the UI indefinitely and misreport the partition as still in progress.
Mark the unattempted volumes as ERROR or SKIPPED with an explanatory log when the systemic handler at line 413 runs.
The docstring at lines 621-622 states "Never raises except for RetriableDicomError." That is no longer accurate. Update it, because callers depend on the stated contract.
Do you want me to draft the reconciliation step in the systemic handler?
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@adit/mass_transfer/processors.py` around lines 699 - 700, Update the
systemic-error handling in _transfer_grouped_series to reconcile every
unattempted volume still in PENDING, marking it ERROR or SKIPPED and emitting an
explanatory log before propagating the systemic failure. Ensure this
reconciliation runs when the systemic handler at line 413 is reached, including
volumes not yet processed. Revise the _transfer_grouped_series docstring to
state its actual exception contract rather than claiming it only raises
RetriableDicomError.
| def test_process_stops_on_systemic_error(mocker: MockerFixture, mass_transfer_env): | ||
| """When export raises ENOSPC, the task stops immediately and the destination is suspended.""" | ||
| env = mass_transfer_env | ||
| series = [ | ||
| _make_discovered(patient_id="PAT1", series_uid="series-1"), | ||
| _make_discovered(patient_id="PAT1", series_uid="series-2"), | ||
| ] | ||
|
|
||
| processor = MassTransferTaskProcessor(env.task) | ||
| mocker.patch.object(processor, "_discover_series", return_value=series) | ||
| mocker.patch("adit.mass_transfer.processors.DicomOperator") | ||
| mocker.patch.object( | ||
| processor, | ||
| "_export_series", | ||
| side_effect=DicomError("Out of disk space while trying to save 'test.dcm'."), | ||
| ) | ||
| mocker.patch("adit.core.utils.mail.send_mail_to_admins") | ||
|
|
||
| result = processor.process() | ||
|
|
||
| assert result["status"] == MassTransferTask.Status.FAILURE | ||
| assert "suspended" in result["message"].lower() or "suspended" in result["log"].lower() |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Correct the docstring and assert that the alert is sent.
Two gaps in this test:
-
The docstring says "When export raises ENOSPC". The side effect raises
DicomError("Out of disk space while trying to save 'test.dcm'."), which exercises the message branch of_is_systemic_error, not the errno branch. No integration test drives a realOSError(errno.ENOSPC, ...)throughprocess(). Fix the docstring, and add a case that raisesOSError(errno.ENOSPC, "No space left on device")so the errno branch is covered end to end. -
Line 2515 patches
send_mail_to_adminsbut never asserts on the mock. The PR states that a systemic error sends an alert to administrators. That requirement is currently untested. Bind the patch and assert the call.
The patch target also depends on the function-local import at processors.py line 449. If that import moves to module scope, this patch stops intercepting and the test still passes. Asserting the call makes that regression visible.
🧪 Proposed fix
`@pytest.mark.django_db`
def test_process_stops_on_systemic_error(mocker: MockerFixture, mass_transfer_env):
- """When export raises ENOSPC, the task stops immediately and the destination is suspended."""
+ """A disk-space DicomError stops the task immediately and suspends the destination."""
env = mass_transfer_env
series = [
_make_discovered(patient_id="PAT1", series_uid="series-1"),
_make_discovered(patient_id="PAT1", series_uid="series-2"),
]
processor = MassTransferTaskProcessor(env.task)
mocker.patch.object(processor, "_discover_series", return_value=series)
mocker.patch("adit.mass_transfer.processors.DicomOperator")
mocker.patch.object(
processor,
"_export_series",
side_effect=DicomError("Out of disk space while trying to save 'test.dcm'."),
)
- mocker.patch("adit.core.utils.mail.send_mail_to_admins")
+ send_mail_mock = mocker.patch("adit.core.utils.mail.send_mail_to_admins")
result = processor.process()
assert result["status"] == MassTransferTask.Status.FAILURE
assert "suspended" in result["message"].lower() or "suspended" in result["log"].lower()
+ send_mail_mock.assert_called_once()Add an errno-branch integration test:
`@pytest.mark.django_db`
def test_process_stops_on_enospc_oserror(mocker: MockerFixture, mass_transfer_env):
"""A raw ENOSPC OSError also suspends the destination and stops the task."""
env = mass_transfer_env
processor = MassTransferTaskProcessor(env.task)
mocker.patch.object(
processor,
"_discover_series",
return_value=[_make_discovered(patient_id="PAT1", series_uid="series-1")],
)
mocker.patch("adit.mass_transfer.processors.DicomOperator")
mocker.patch.object(
processor,
"_export_series",
side_effect=OSError(errno.ENOSPC, "No space left on device"),
)
send_mail_mock = mocker.patch("adit.core.utils.mail.send_mail_to_admins")
result = processor.process()
assert result["status"] == MassTransferTask.Status.FAILURE
env.destination.refresh_from_db()
assert env.destination.suspended is True
send_mail_mock.assert_called_once()🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@adit/mass_transfer/tests/test_processor.py` around lines 2499 - 2520, Correct
test_process_stops_on_systemic_error’s docstring to describe the DicomError
message branch, bind the send_mail_to_admins patch and assert it is called once,
and add a separate end-to-end test for OSError(errno.ENOSPC, ...) that verifies
failure, destination suspension, and a single administrator alert.
1 participant
Replaces #331, which was opened in April, had drifted 49 commits behind
main, and no longer merged. Same change, rebased onto currentmain; its CI failure was purely staleness and is gone (866 tests pass locally).The problem
When the NFS destination filled up, mass transfer failed silently and expensively.
_handle_fetched_imageraises aDicomErroronENOSPC, but the genericexcept Exceptionin the per-series loop marked that volume ERROR and moved on to the next series — re-downloading from the PACS and failing on write again, for every remaining series in the task and every subsequent task for that destination.The fix
DicomFolder.suspendedfield (migration0018).ENOSPC,ESTALE,EROFS,EIO, or aDicomErrorreporting out-of-disk — the processor suspends the destination folder, stops immediately, and returns FAILURE.Verification
pytest adit/mass_transfer/tests/test_processor.py→ 122 passed-m "not acceptance"→ 866 passed, 3 xfailedmakemigrations --check→ no changes detectedRebase notes
Only two trivial conflicts, both resolved without behaviour change: the import block (dropped the unnecessary
from __future__ import annotations; adit is Python 3.12+), and the end of the test file wheremainand this branch had each appended a different test section — both are kept.Summary by CodeRabbit
Release Notes
New Features
Tests