diff --git a/py/services/aria2_downloader.py b/py/services/aria2_downloader.py index 4de53e65..e5541a96 100644 --- a/py/services/aria2_downloader.py +++ b/py/services/aria2_downloader.py @@ -11,6 +11,7 @@ import os import secrets import shutil import socket +import time from dataclasses import dataclass from datetime import datetime from pathlib import Path @@ -28,6 +29,35 @@ logger = logging.getLogger(__name__) # is lost (daemon restart / RPC outage) before failing the download. MAX_TRANSFER_RECOVERY_ATTEMPTS = 2 +# stderr lines matching these markers indicate a disk write failure inside +# aria2 (piece cache flush or raw file write). They are promoted to INFO so +# the root cause (disk full, permission denied, file locked by another +# process, ...) is visible in the default logs; all other stderr output stays +# at DEBUG to avoid noise. +_DISK_WRITE_ERROR_MARKERS = ( + # aria2 wrapper messages (write disk cache flush path) + "write disk cache flush failure", + "error when trying to flush write cache", + "failed to write into the file", + "failed to open the file", + "failed to seek the file", + # underlying root-cause phrases reported via "cause: ..." (POSIX + Windows) + "no space left on device", + "not enough space on the disk", + "input/output error", + "permission denied", + "access is denied", + "disk quota exceeded", + "used by another process", + "sharing violation", +) + +# Minimum interval between INFO-level reports of the same stderr line so a +# repeated failure (e.g. aria2 retrying against a full disk) does not spam +# the log. +STDERR_ERROR_REPORT_INTERVAL = 60.0 + + def _try_certifi_ca_path() -> str | None: """Return the certifi CA bundle path if available, else None.""" try: @@ -94,6 +124,7 @@ class Aria2Downloader: self._poll_interval = 0.5 self._state_store = Aria2TransferStateStore() self._stderr_reader_task: Optional[asyncio.Task[Any]] = None + self._stderr_error_report: Dict[str, float] = {} @property def is_running(self) -> bool: @@ -447,16 +478,51 @@ class Aria2Downloader: blocks, which freezes the entire ``aria2c`` process — including its RPC handler. This background task reads lines from stderr as they arrive and forwards them to Python's logger. + + Lines that indicate a disk write failure (e.g. the "cause: No space + left on device" line that follows "Write disk cache flush failure") + are promoted to INFO so the root cause is visible without enabling + debug logging; every other line stays at DEBUG to avoid noise. """ try: assert self._process is not None and self._process.stderr is not None async for line in self._process.stderr: text = line.decode("utf-8", errors="replace").rstrip() if text: - logger.debug("aria2 stderr: %s", text) + if self._is_disk_write_error(text): + self._report_stderr_error(text) + else: + logger.debug("aria2 stderr: %s", text) except Exception: pass + @staticmethod + def _is_disk_write_error(text: str) -> bool: + lowered = text.lower() + return any(marker in lowered for marker in _DISK_WRITE_ERROR_MARKERS) + + def _report_stderr_error(self, text: str) -> None: + """INFO-log a disk write failure line, rate-limited per line text. + + aria2 re-emits the same error chain on every poll/retry while the + underlying condition persists; only the first occurrence within + ``STDERR_ERROR_REPORT_INTERVAL`` seconds is promoted to INFO. + """ + now = time.monotonic() + last = self._stderr_error_report.get(text) + if last is not None and now - last < STDERR_ERROR_REPORT_INTERVAL: + logger.debug("aria2 stderr (repeated disk write error): %s", text) + return + # Drop entries older than the window so the map stays bounded even + # during a long disk-full episode (piece indexes change per line). + self._stderr_error_report = { + line: timestamp + for line, timestamp in self._stderr_error_report.items() + if now - timestamp < STDERR_ERROR_REPORT_INTERVAL + } + self._stderr_error_report[text] = now + logger.info("aria2 disk write failure: %s", text) + async def _dispatch_progress(self, callback, snapshot: DownloadProgress) -> None: try: result = callback(snapshot, snapshot) diff --git a/tests/services/test_aria2_downloader.py b/tests/services/test_aria2_downloader.py index 58b9958b..3feead5f 100644 --- a/tests/services/test_aria2_downloader.py +++ b/tests/services/test_aria2_downloader.py @@ -1,6 +1,8 @@ from __future__ import annotations import asyncio +import logging +import time from pathlib import Path from unittest.mock import AsyncMock @@ -705,3 +707,104 @@ async def test_wait_until_ready_includes_stderr_in_error(): msg = str(exc_info.value) assert "code 28" in msg assert "ERROR: unknown option --fsync" in msg + + +def test_is_disk_write_error_matches_wrapper_and_cause_lines(): + downloader = Aria2Downloader() + + assert downloader._is_disk_write_error( + "[ERROR] [DownloadCommand.cc:127] errorCode=9 Write disk cache flush failure index=18798" + ) + assert downloader._is_disk_write_error( + "Exception: [AbstractDiskWriter.cc:454] errNum=28 errorCode=9 " + "Failed to write into the file D:\\models\\model.safetensors, " + "cause: No space left on device" + ) + # Windows: "used by another process" = the file is locked by antivirus etc. + assert downloader._is_disk_write_error( + "Exception: ... The process cannot access the file because it is " + "being used by another process." + ) + assert not downloader._is_disk_write_error( + "Download aborted. URI=https://example.com/model.safetensors" + ) + assert not downloader._is_disk_write_error("") + assert not downloader._is_disk_write_error("bad option: --fsync") + + +@pytest.mark.asyncio +async def test_drain_stderr_promotes_disk_write_failure_to_info(caplog): + downloader = Aria2Downloader() + + class FakeStderr: + def __init__(self, lines): + self._lines = list(lines) + + def __aiter__(self): + return self + + async def __anext__(self): + if not self._lines: + raise StopAsyncIteration + return self._lines.pop(0) + + proc = type( + "Proc", + (), + { + "stderr": FakeStderr( + [ + b"", + b"[ERROR] [WrDiskCacheEntry.cc:83] Error when trying to flush write cache", + b"Exception: [AbstractDiskWriter.cc:454] errNum=28 errorCode=9 " + b"Failed to write into the file D:\\models\\model.safetensors, " + b"cause: No space left on device", + b"Download aborted. URI=https://example.com/model.safetensors", + ] + ) + }, + )() + downloader._process = proc + + with caplog.at_level(logging.DEBUG, logger="py.services.aria2_downloader"): + await downloader._drain_stderr() + + info_records = [r for r in caplog.records if r.levelno == logging.INFO] + debug_records = [r for r in caplog.records if r.levelno == logging.DEBUG] + + assert any( + "Error when trying to flush write cache" in r.message for r in info_records + ) + assert any("No space left on device" in r.message for r in info_records) + assert any("Download aborted" in r.message for r in debug_records) + assert not any("Download aborted" in r.message for r in info_records) + + +def test_stderr_error_rate_limited_to_one_info_per_window(caplog, monkeypatch): + downloader = Aria2Downloader() + line = "Exception: ... cause: No space left on device" + + with caplog.at_level(logging.DEBUG, logger="py.services.aria2_downloader"): + downloader._report_stderr_error(line) + downloader._report_stderr_error(line) + monkeypatch.setattr(downloader, "_stderr_error_report", {}) + downloader._report_stderr_error(line) + + info_records = [r for r in caplog.records if r.levelno == logging.INFO] + debug_records = [r for r in caplog.records if r.levelno == logging.DEBUG] + + assert len(info_records) == 2 + assert len(debug_records) == 1 + assert "repeated disk write error" in debug_records[0].message + + +def test_stderr_error_report_prunes_expired_entries(): + downloader = Aria2Downloader() + old_line = "Exception: ... cause: No space left on device" + new_line = "Exception: ... cause: Input/output error" + downloader._stderr_error_report[old_line] = time.monotonic() - 120.0 + + downloader._report_stderr_error(new_line) + + assert old_line not in downloader._stderr_error_report + assert new_line in downloader._stderr_error_report