Torrents: fix stall handling on "downloading metadata" + stop orphaning in qbit

noldevin: a magnet stuck "downloading metadata" ran 11h despite a 15-min stall
timeout, got cleared from SoulSync but left active in qbit, then re-grabbed as a
duplicate. Two bugs:

1. Stall never fired on metaDL. StallTracker reset its clock on any `downloaded`
   byte increase, but a metaDL torrent's byte counter still ticks up from DHT/peer
   protocol overhead while making no real progress — so the clock reset forever.
   Fix: the byte counter only counts once metadata is in (size>0). During the
   metadata phase (size==0) the only thing that counts as progress is *obtaining*
   metadata, so a magnet that can't even do that within the timeout is correctly
   flagged stalled. size=None preserves the old byte-only behavior (back-compat).

2. Orphaned in qbit. The monitor's stall exit removed the torrent, but the `error`
   exit and the 6h deadline exit only marked the download failed — leaving the
   torrent active in qbit, untracked here, so SoulSync re-grabbed the same dead
   torrent (qbit logs the duplicate-add). Fix: both terminal exits now run
   _cleanup_torrent (shared with the stall path), which removes+deletes (abandon)
   or pauses per the stall action — nothing is left orphaned.

Tests (10 new): metaDL byte-noise no longer resets the clock (stalls at timeout);
obtaining metadata resets it; real byte-progress still tracked after metadata;
_cleanup_torrent removes+delete_files on abandon / pauses on pause / no-ops on
empty hash or no adapter / swallows a client error. 151 torrent tests green.

core/download_plugins/torrent.py

@@ -352,34 +352,60 @@ class TorrentDownloadPlugin(DownloadSourcePlugin):
                 self._finalize_download(download_id, last_save_path)
                 return
             if status.state == 'error':
+                # Clean the dead torrent out of the client, or it's left orphaned
+                # (active in qbit, untracked here) and re-grabbed as a duplicate.
+                self._cleanup_torrent(torrent_hash, get_stall_action())
                 self._mark_error(download_id, status.error or "Torrent client reported error")
                 return
 
-            if stall.is_stalled(status.downloaded, status.state, time.monotonic()):
+            if stall.is_stalled(status.downloaded, status.state, time.monotonic(),
+                                size=status.size):
                 self._handle_stalled(download_id, torrent_hash, get_stall_action())
                 return
 
             time.sleep(_POLL_INTERVAL_SECONDS)
 
+        # Deadline reached. One last status check closes the race where the
+        # torrent completed during the final poll interval — finalize it instead
+        # of deleting a just-finished download's files. Otherwise clean it out of
+        # the client, or it sits orphaned in qbit (e.g. a metadata-stuck magnet
+        # that escaped the stall timer) and gets re-grabbed as a duplicate.
+        try:
+            final = run_async(adapter.get_status(torrent_hash))
+        except Exception:
+            final = None
+        if final is not None and final.state in _COMPLETE_STATES:
+            self._finalize_download(download_id, final.save_path or last_save_path)
+            return
+        self._cleanup_torrent(torrent_hash, get_stall_action())
         self._mark_error(download_id, "Torrent download timed out")
 
+    def _cleanup_torrent(self, torrent_hash: str, action: str) -> None:
+        """Remove (abandon) or pause a dead/stalled/timed-out torrent in the
+        client so it isn't left ORPHANED — active in qbit but no longer tracked
+        here, which makes SoulSync re-grab the same dead torrent as a duplicate
+        on the next attempt (noldevin). Best-effort: a client error is logged,
+        not raised, so the download still fails cleanly."""
+        adapter = get_active_torrent_adapter()
+        if adapter is None or not torrent_hash:
+            return
+        try:
+            if action == "pause":
+                run_async(adapter.pause(torrent_hash))
+            else:
+                # delete_files: a stalled/failed 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("Torrent cleanup (%s) on %s failed: %s",
+                           action, torrent_hash[:8] if torrent_hash else "?", e)
+
     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)
+        self._cleanup_torrent(torrent_hash, action)
         verb = "paused" if action == "pause" else "removed"
         self._mark_error(
             download_id,

core/download_plugins/torrent_stall.py

@@ -74,26 +74,50 @@ class StallTracker:
     def __init__(self, timeout_seconds: float):
         self.timeout = float(timeout_seconds or 0)
         self._last_downloaded = -1           # -1 = first observation
+        self._had_metadata = None            # None = first observation; else size>0?
         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.
+    def is_stalled(self, downloaded: int, state: str, now: float,
+                   size: int = None) -> bool:
+        """Record this poll's observation; return True iff the torrent has gone
+        ``timeout`` seconds with no real forward progress while in a working state.
 
-        ``downloaded`` is cumulative bytes; ``state`` is the adapter-uniform
-        state; ``now`` is a monotonic timestamp (seconds)."""
+        ``downloaded`` is cumulative payload bytes; ``state`` is the adapter-uniform
+        state; ``now`` is a monotonic timestamp; ``size`` is the torrent's total
+        size in bytes (0/None while still fetching metadata).
+
+        Metadata-phase fix (#852-adjacent torrent report): a magnet stuck
+        "downloading metadata" reports ``size==0`` and a ``downloaded`` byte
+        counter that still ticks up from DHT/peer-protocol overhead even though it
+        makes no actual progress. Treating those bumps as progress reset the stall
+        clock forever, so a dead magnet never timed out. Now the byte counter only
+        counts once metadata is in (``size>0``); during the metadata phase the only
+        thing that counts as progress is *obtaining* the metadata, so a torrent
+        that can't even do that within the timeout is correctly flagged stalled.
+        """
         if self.timeout <= 0:
             return False
 
         downloaded = int(downloaded or 0)
+        # size is None when the caller doesn't track it (assume metadata present —
+        # the old byte-progress behavior); an explicit size==0 is the metadata
+        # phase (metaDL), where the byte counter is unreliable noise.
+        has_metadata = size is None or int(size) > 0
 
-        # Forward progress (or first sighting) resets the stall clock.
-        if self._last_downloaded < 0 or downloaded > self._last_downloaded:
+        # Real forward progress: first sighting, metadata just arrived, or (only
+        # once we have metadata) more payload bytes. Byte bumps during the
+        # metadata phase are protocol noise and do NOT count.
+        progressed = (
+            self._had_metadata is None                                  # first poll
+            or (has_metadata and not self._had_metadata)                # got metadata
+            or (has_metadata and downloaded > self._last_downloaded)    # more payload
+        )
+        self._had_metadata = has_metadata
         self._last_downloaded = downloaded
+
+        if progressed:
             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:

tests/test_torrent_cleanup_orphan.py

@@ -0,0 +1,67 @@
+"""noldevin: a dead torrent (metaDL stuck / errored / timed out) was left ORPHANED
+in qbit — cleared from SoulSync but still active in the client, then re-grabbed as
+a duplicate. The monitor's terminal exits now call _cleanup_torrent, which removes
+(abandon) or pauses it in the client."""
+
+from __future__ import annotations
+
+from unittest.mock import patch
+
+import pytest
+
+from core.download_plugins.torrent import TorrentDownloadPlugin
+
+
+class _FakeAdapter:
+    def __init__(self):
+        self.removed = []
+        self.paused = []
+
+    async def remove(self, h, delete_files=False):
+        self.removed.append((h, delete_files))
+
+    async def pause(self, h):
+        self.paused.append(h)
+
+
+@pytest.fixture
+def plugin():
+    with patch('core.download_plugins.torrent.ProwlarrClient'):
+        yield TorrentDownloadPlugin()
+
+
+def test_abandon_removes_torrent_and_deletes_files(plugin):
+    fake = _FakeAdapter()
+    with patch('core.download_plugins.torrent.get_active_torrent_adapter', return_value=fake):
+        plugin._cleanup_torrent('abc123', 'abandon')
+    assert fake.removed == [('abc123', True)]   # removed + partial data deleted
+    assert fake.paused == []
+
+
+def test_pause_action_pauses_not_removes(plugin):
+    fake = _FakeAdapter()
+    with patch('core.download_plugins.torrent.get_active_torrent_adapter', return_value=fake):
+        plugin._cleanup_torrent('abc123', 'pause')
+    assert fake.paused == ['abc123']
+    assert fake.removed == []
+
+
+def test_no_hash_is_a_noop(plugin):
+    fake = _FakeAdapter()
+    with patch('core.download_plugins.torrent.get_active_torrent_adapter', return_value=fake):
+        plugin._cleanup_torrent('', 'abandon')
+        plugin._cleanup_torrent(None, 'abandon')
+    assert fake.removed == [] and fake.paused == []
+
+
+def test_no_adapter_is_a_noop(plugin):
+    with patch('core.download_plugins.torrent.get_active_torrent_adapter', return_value=None):
+        plugin._cleanup_torrent('abc123', 'abandon')   # must not raise
+
+
+def test_client_error_is_swallowed(plugin):
+    class _Boom:
+        async def remove(self, h, delete_files=False):
+            raise RuntimeError("qbit down")
+    with patch('core.download_plugins.torrent.get_active_torrent_adapter', return_value=_Boom()):
+        plugin._cleanup_torrent('abc123', 'abandon')   # best-effort: logged, not raised

tests/test_torrent_stall.py

@@ -102,3 +102,38 @@ def test_get_stall_action(raw, expected):
     with patch.object(ts, "config_manager",
                       _cfg({"download_source.torrent_stall_action": raw})):
         assert get_stall_action() == expected
+
+
+# ── metadata-phase noise (noldevin #2: metaDL stuck 11h, stall never fired) ──
+def test_metadata_phase_byte_noise_does_not_reset_clock():
+    """A magnet stuck 'downloading metadata' reports size==0 and a downloaded
+    counter that still ticks up from DHT/peer overhead. Those bumps must NOT
+    reset the stall clock, or the dead magnet never times out (the bug)."""
+    t = StallTracker(timeout_seconds=600)
+    assert t.is_stalled(0,     "downloading", now=0,   size=0) is False  # first
+    assert t.is_stalled(16384, "downloading", now=120, size=0) is False  # noise bump
+    assert t.is_stalled(32768, "downloading", now=300, size=0) is False  # more noise
+    assert t.is_stalled(40000, "downloading", now=480, size=0) is False  # still under
+    # Despite the byte counter climbing the whole time, no metadata was obtained
+    # → stalled at the timeout.
+    assert t.is_stalled(50000, "downloading", now=600, size=0) is True
+
+
+def test_obtaining_metadata_resets_the_clock():
+    """size 0 -> >0 means metadata arrived — real progress, reset the clock."""
+    t = StallTracker(timeout_seconds=600)
+    assert t.is_stalled(0, "downloading", now=0,   size=0) is False
+    assert t.is_stalled(0, "downloading", now=500, size=0) is False        # accruing
+    # metadata arrives at t=550 (size now known) → progress → clock resets
+    assert t.is_stalled(0, "downloading", now=550, size=10_000_000) is False
+    assert t.is_stalled(0, "downloading", now=900, size=10_000_000) is False  # <600 since reset
+    assert t.is_stalled(0, "downloading", now=1150, size=10_000_000) is True  # 600 later, no bytes
+
+
+def test_real_download_progress_tracked_after_metadata():
+    """Once metadata is in, byte progress resets the clock as normal."""
+    t = StallTracker(timeout_seconds=600)
+    assert t.is_stalled(0,      "downloading", now=0,   size=10_000_000) is False
+    assert t.is_stalled(500000, "downloading", now=400, size=10_000_000) is False  # progress
+    assert t.is_stalled(500000, "downloading", now=900, size=10_000_000) is False  # <600 since
+    assert t.is_stalled(500000, "downloading", now=1001, size=10_000_000) is True  # stalled