Torrents: stalled-torrent handling — abandon a dead magnet instead of holding a worker 6h (noldevin)

noldevin's first torrent was stuck "downloading metadata" — a dead magnet
with no peers. The poll loop would ride the full album deadline (6h default)
on it, holding the worker the whole time, with no built-in escape.

New stall handling, off the existing poll loop:
- core/download_plugins/torrent_stall.py — pure StallTracker (clock injected,
  no I/O): forward byte progress resets a stall clock; once a torrent spends
  the stall timeout in a working state (queued/downloading/stalled/error)
  with zero progress, it's stalled. seeding/completed/paused never count.
  Covers the metadata-stuck case (0 bytes, 0 progress) and a dead mid-download
  swarm with one rule.
- _handle_stalled: 'abandon' (default) removes the torrent + its partial data
  (a metadata stub is junk) and fails the download so the next source can try;
  'pause' parks it in the client for the user. Adapter errors are swallowed —
  the download still fails cleanly.
- two settings (download_source.torrent_stall_timeout_seconds = 600,
  torrent_stall_action = 'abandon'); timeout 0 disables, restoring the old
  ride-the-deadline behavior. Config-key driven, matching the existing
  album_bundle_* tuning knobs (no UI form, same as those).

Tests: 18 on the tracker + settings (timeout trip, progress reset, idle-state
exemption, pause→resume clock restart, disable, parse tolerance) + 3 on the
plugin action path (abandon removes w/ delete_files, pause pauses, adapter
error survived). 158 torrent-family tests pass.

config/settings.py

@@ -493,6 +493,15 @@ class ConfigManager:
                 # editing source.
                 "album_bundle_poll_interval_seconds": 2.0,
                 "album_bundle_timeout_seconds": 6 * 60 * 60,    # 6 hours
+                # Stalled-torrent handling (noldevin): abandon a torrent that
+                # makes zero download progress for this long (dead magnet
+                # stuck on "downloading metadata", no seeders) instead of
+                # holding the worker for the full album timeout. 0 disables.
+                "torrent_stall_timeout_seconds": 10 * 60,       # 10 minutes
+                # What to do when a torrent stalls: "abandon" (remove it +
+                # its partial data, fail the download so the next source can
+                # try) or "pause" (pause in the client, leave for the user).
+                "torrent_stall_action": "abandon",
             },
             "post_processing": {
                 # When a download is quarantined (AcoustID mismatch, integrity /

core/download_plugins/torrent.py

@@ -68,6 +68,11 @@ from core.download_plugins.album_bundle import (
     resolve_reported_save_path,
 )
 from core.download_plugins.base import DownloadSourcePlugin
+from core.download_plugins.torrent_stall import (
+    StallTracker,
+    get_stall_action,
+    get_stall_timeout,
+)
 from core.download_plugins.types import AlbumResult, DownloadStatus, TrackResult
 from core.prowlarr_client import (
     DEFAULT_MUSIC_CATEGORIES,
@@ -305,6 +310,11 @@ class TorrentDownloadPlugin(DownloadSourcePlugin):
         # but the same tolerance keeps a one-off connection failure
         # from killing an otherwise-healthy download.
         misses = TransientMissCounter()
+        # Stalled-torrent handling (noldevin): give up early on a torrent
+        # making zero progress (dead magnet stuck on metadata, no seeders)
+        # instead of holding this worker for the full album deadline. Read
+        # per-download so a settings change applies to in-flight torrents.
+        stall = StallTracker(get_stall_timeout())
         while time.monotonic() < deadline:
             if self.shutdown_check and self.shutdown_check():
                 return
@@ -345,10 +355,37 @@ class TorrentDownloadPlugin(DownloadSourcePlugin):
                 self._mark_error(download_id, status.error or "Torrent client reported error")
                 return
 
+            if stall.is_stalled(status.downloaded, status.state, time.monotonic()):
+                self._handle_stalled(download_id, torrent_hash, get_stall_action())
+                return
+
             time.sleep(_POLL_INTERVAL_SECONDS)
 
         self._mark_error(download_id, "Torrent download timed out")
 
+    def _handle_stalled(self, download_id: str, torrent_hash: str, action: str) -> None:
+        """A torrent made no progress past the stall timeout. Abandon it
+        (remove from client + delete its partial data) or pause it for the
+        user, then fail the download so the worker frees up."""
+        adapter = get_active_torrent_adapter()
+        timeout_min = round(get_stall_timeout() / 60, 1)
+        if adapter is not None:
+            try:
+                if action == "pause":
+                    run_async(adapter.pause(torrent_hash))
+                else:
+                    # delete_files: a stalled torrent's partial data is junk
+                    # (often just a metadata stub) — don't leave it on disk.
+                    run_async(adapter.remove(torrent_hash, delete_files=True))
+            except Exception as e:
+                logger.warning("Stalled-torrent %s on %s failed: %s",
+                               action, torrent_hash[:8] if torrent_hash else "?", e)
+        verb = "paused" if action == "pause" else "removed"
+        self._mark_error(
+            download_id,
+            f"Torrent stalled (no progress for {timeout_min} min) — {verb}",
+        )
+
     def _finalize_download(self, download_id: str, save_path: Optional[str]) -> None:
         """Adapter said complete. Walk the directory + pick the
         first audio file as the canonical ``file_path``."""

core/download_plugins/torrent_stall.py

@@ -0,0 +1,107 @@
+"""Stalled-torrent detection + policy (noldevin's request).
+
+A torrent can sit forever making zero progress — most commonly stuck
+"downloading metadata" on a magnet with no peers, but also a dead swarm
+mid-download. The torrent poll loop would just burn the full 6-hour album
+timeout on it. This module decides, from the live status stream, when a
+torrent has been stalled too long, and what to do about it.
+
+Design split, kept testable:
+- ``StallTracker`` is the pure decision core — feed it each poll's
+  ``(downloaded, state, now)`` and it answers "stalled too long?" using a
+  monotonic clock passed in (no time import, no I/O). Progress = bytes
+  moved since the last poll; any forward movement resets the stall clock.
+  Terminal/healthy-but-idle states (seeding, completed, paused) never count
+  as stalled — only states where the torrent is *supposed* to be working.
+- ``get_stall_timeout`` / ``get_stall_action`` read the two settings.
+
+A timeout of 0 disables stall handling entirely (back to the old behavior:
+ride the full poll deadline).
+"""
+
+from __future__ import annotations
+
+from config.settings import config_manager
+
+# 0 = disabled. 10 minutes is long enough to ride out a slow metadata fetch
+# or a brief peer drought, short enough to give up on a truly dead magnet
+# instead of holding a worker for 6 hours.
+DEFAULT_STALL_TIMEOUT_SECONDS = 10 * 60
+
+# What to do when a torrent stalls past the timeout:
+#   'abandon' — remove it from the client (and its partial data) + fail the
+#               download so the worker is freed and the next source can try.
+#   'pause'   — pause it in the client + fail the download, leaving the
+#               torrent for the user to inspect/resume manually.
+_VALID_ACTIONS = ("abandon", "pause")
+DEFAULT_STALL_ACTION = "abandon"
+
+# States where the torrent is meant to be making download progress, so a
+# lack of it counts toward the stall clock. Mirrors the adapter-uniform set
+# in core/torrent_clients/base.py. Notably EXCLUDES seeding/completed (done)
+# and paused (the user's own choice) — neither is a stall.
+STALLABLE_STATES = frozenset(("queued", "downloading", "stalled", "error"))
+
+
+def get_stall_timeout() -> float:
+    """Seconds of zero progress before a torrent is considered stalled.
+    0 (or invalid/negative) disables stall handling."""
+    raw = config_manager.get("download_source.torrent_stall_timeout_seconds",
+                             DEFAULT_STALL_TIMEOUT_SECONDS)
+    try:
+        value = float(raw)
+        if value >= 0:
+            return value
+    except (TypeError, ValueError):
+        pass
+    return DEFAULT_STALL_TIMEOUT_SECONDS
+
+
+def get_stall_action() -> str:
+    """What to do with a stalled torrent: 'abandon' (default) or 'pause'."""
+    raw = config_manager.get("download_source.torrent_stall_action",
+                             DEFAULT_STALL_ACTION)
+    action = str(raw or "").strip().lower()
+    return action if action in _VALID_ACTIONS else DEFAULT_STALL_ACTION
+
+
+class StallTracker:
+    """Tracks one torrent's forward progress across polls.
+
+    Pure + clock-injected so it tests without sleeping. ``timeout`` <= 0
+    disables it (``is_stalled`` always returns False)."""
+
+    def __init__(self, timeout_seconds: float):
+        self.timeout = float(timeout_seconds or 0)
+        self._last_downloaded = -1           # -1 = first observation
+        self._progress_since = None          # monotonic time of last forward movement
+
+    def is_stalled(self, downloaded: int, state: str, now: float) -> bool:
+        """Record this poll's observation; return True iff the torrent has
+        gone ``timeout`` seconds with no byte progress while in a state
+        that's supposed to be downloading.
+
+        ``downloaded`` is cumulative bytes; ``state`` is the adapter-uniform
+        state; ``now`` is a monotonic timestamp (seconds)."""
+        if self.timeout <= 0:
+            return False
+
+        downloaded = int(downloaded or 0)
+
+        # Forward progress (or first sighting) resets the stall clock.
+        if self._last_downloaded < 0 or downloaded > self._last_downloaded:
+            self._last_downloaded = downloaded
+            self._progress_since = now
+            return False
+        self._last_downloaded = downloaded
+
+        # Not in a working state → not a stall (seeding/paused/completed).
+        if state not in STALLABLE_STATES:
+            self._progress_since = now  # don't accrue stall time while idle-by-design
+            return False
+
+        if self._progress_since is None:
+            self._progress_since = now
+            return False
+
+        return (now - self._progress_since) >= self.timeout

tests/test_torrent_stall.py

@@ -0,0 +1,104 @@
+"""Stalled-torrent detection + policy (noldevin: 'stuck on downloading metadata').
+
+The pure StallTracker decides, from the per-poll status stream, when a
+torrent has gone too long with no byte progress while it's supposed to be
+downloading. Clock is injected so this tests without sleeping.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import patch
+
+import pytest
+
+from core.download_plugins.torrent_stall import (
+    StallTracker,
+    get_stall_action,
+    get_stall_timeout,
+)
+
+
+def test_no_progress_trips_after_timeout():
+    t = StallTracker(timeout_seconds=600)
+    # First sighting at t=0 (metadata fetch: 0 bytes, 'downloading').
+    assert t.is_stalled(0, "downloading", now=0) is False
+    assert t.is_stalled(0, "downloading", now=300) is False     # 5 min, under
+    assert t.is_stalled(0, "downloading", now=599) is False     # just under
+    assert t.is_stalled(0, "downloading", now=600) is True      # hit the timeout
+
+
+def test_forward_progress_resets_the_clock():
+    t = StallTracker(timeout_seconds=600)
+    t.is_stalled(0, "downloading", now=0)
+    t.is_stalled(0, "downloading", now=500)        # stalling...
+    assert t.is_stalled(1024, "downloading", now=550) is False  # bytes moved → reset
+    assert t.is_stalled(1024, "downloading", now=1000) is False  # 450s since reset
+    assert t.is_stalled(1024, "downloading", now=1150) is True   # 600s since reset
+
+
+def test_explicit_stalled_state_counts():
+    t = StallTracker(timeout_seconds=600)
+    t.is_stalled(2048, "stalled", now=0)
+    assert t.is_stalled(2048, "stalled", now=600) is True
+
+
+def test_idle_by_design_states_never_stall():
+    # Seeding / paused / completed aren't stalls even with zero progress.
+    for state in ("seeding", "completed", "paused"):
+        t = StallTracker(timeout_seconds=600)
+        t.is_stalled(5000, state, now=0)
+        assert t.is_stalled(5000, state, now=10_000) is False, state
+
+
+def test_state_flip_active_to_idle_to_active():
+    t = StallTracker(timeout_seconds=600)
+    t.is_stalled(0, "downloading", now=0)
+    t.is_stalled(0, "paused", now=500)             # user paused → clock parked
+    # Resumed; no bytes yet. Clock restarts from the un-pause, not from t=0.
+    assert t.is_stalled(0, "downloading", now=900) is False
+    assert t.is_stalled(0, "downloading", now=1100) is True   # 600s after un-pause
+
+
+def test_timeout_zero_disables():
+    t = StallTracker(timeout_seconds=0)
+    assert t.is_stalled(0, "downloading", now=0) is False
+    assert t.is_stalled(0, "downloading", now=10_000_000) is False
+
+
+# ── settings helpers ─────────────────────────────────────────────────────────
+
+def _cfg(values):
+    class _C:
+        def get(self, key, default=None):
+            return values.get(key, default)
+    return _C()
+
+
+@pytest.mark.parametrize("raw,expected", [
+    (300, 300.0),
+    ("450", 450.0),
+    (0, 0.0),                     # explicit disable honored
+    (-5, 10 * 60),               # negative → default
+    ("bad", 10 * 60),           # garbage → default
+    (None, 10 * 60),
+])
+def test_get_stall_timeout(raw, expected):
+    import core.download_plugins.torrent_stall as ts
+    with patch.object(ts, "config_manager",
+                      _cfg({"download_source.torrent_stall_timeout_seconds": raw})):
+        assert get_stall_timeout() == expected
+
+
+@pytest.mark.parametrize("raw,expected", [
+    ("abandon", "abandon"),
+    ("pause", "pause"),
+    ("PAUSE", "pause"),
+    ("nonsense", "abandon"),
+    ("", "abandon"),
+    (None, "abandon"),
+])
+def test_get_stall_action(raw, expected):
+    import core.download_plugins.torrent_stall as ts
+    with patch.object(ts, "config_manager",
+                      _cfg({"download_source.torrent_stall_action": raw})):
+        assert get_stall_action() == expected

tests/test_torrent_usenet_plugins.py

@@ -637,3 +637,66 @@ def test_registry_includes_torrent_and_usenet() -> None:
     names = registry.names()
     assert 'torrent' in names
     assert 'usenet' in names
+
+
+# ---------------------------------------------------------------------------
+# Stalled-torrent handling (noldevin) — the _handle_stalled action path
+# ---------------------------------------------------------------------------
+
+
+def test_handle_stalled_abandon_removes_and_fails():
+    plugin = TorrentDownloadPlugin()
+    with plugin._lock:
+        plugin.active_downloads['d1'] = {'state': 'InProgress, Downloading', 'progress': 0.0}
+
+    adapter = MagicMock()
+    adapter.remove = AsyncMock(return_value=True)
+    adapter.pause = AsyncMock(return_value=True)
+
+    with patch('core.download_plugins.torrent.get_active_torrent_adapter', return_value=adapter), \
+         patch('core.download_plugins.torrent.get_stall_timeout', return_value=600):
+        plugin._handle_stalled('d1', 'HASH123', 'abandon')
+
+    adapter.remove.assert_called_once()
+    assert adapter.remove.call_args.kwargs.get('delete_files') is True  # partial junk removed
+    adapter.pause.assert_not_called()
+    row = plugin.active_downloads['d1']
+    assert row['state'] == 'Completed, Errored'
+    assert 'stalled' in (row.get('error') or '').lower()
+    assert 'removed' in (row.get('error') or '').lower()
+
+
+def test_handle_stalled_pause_pauses_and_fails():
+    plugin = TorrentDownloadPlugin()
+    with plugin._lock:
+        plugin.active_downloads['d2'] = {'state': 'InProgress, Downloading', 'progress': 0.0}
+
+    adapter = MagicMock()
+    adapter.remove = AsyncMock(return_value=True)
+    adapter.pause = AsyncMock(return_value=True)
+
+    with patch('core.download_plugins.torrent.get_active_torrent_adapter', return_value=adapter), \
+         patch('core.download_plugins.torrent.get_stall_timeout', return_value=600):
+        plugin._handle_stalled('d2', 'HASH456', 'pause')
+
+    adapter.pause.assert_called_once()
+    adapter.remove.assert_not_called()                 # data left for the user
+    row = plugin.active_downloads['d2']
+    assert row['state'] == 'Completed, Errored'
+    assert 'paused' in (row.get('error') or '').lower()
+
+
+def test_handle_stalled_survives_adapter_error():
+    plugin = TorrentDownloadPlugin()
+    with plugin._lock:
+        plugin.active_downloads['d3'] = {'state': 'InProgress, Downloading'}
+
+    adapter = MagicMock()
+    adapter.remove = AsyncMock(side_effect=RuntimeError("client down"))
+
+    with patch('core.download_plugins.torrent.get_active_torrent_adapter', return_value=adapter), \
+         patch('core.download_plugins.torrent.get_stall_timeout', return_value=600):
+        plugin._handle_stalled('d3', 'HASH789', 'abandon')   # must not raise
+
+    # Download still fails cleanly even when the client call blew up.
+    assert plugin.active_downloads['d3']['state'] == 'Completed, Errored'