From 3265761361309b71403f5bac34be39f28418fcd6 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 7 Oct 2026 17:29:34 +0000 Subject: [PATCH] feat: torrents download in the background, so MAM grabs aren't capped at the worker count A worker stayed with each torrent until the client finished downloading it, so only MAX_CONCURRENT_DOWNLOADS torrents (3 by default) were ever in flight, however many MAM unsatisfied slots were free. - Once a hash-keyed torrent client (Deluge, qBittorrent, Transmission, rTorrent) has had a torrent for 30 seconds (or it stalls), its worker is freed. The task stays downloading, with the client's progress (checked every 30 seconds), and goes back to the workers to import when the client has finished it; it fails if the torrent leaves the client. - BookQueue.detach frees the worker and re-saves the retry data (with the hash), so a retry after a restart finds the torrent in the client. - Coming back to import isn't a new MAM snatch: the handoff is recorded once per task in the client, and such tasks don't count as pending snatches. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_0184DMrDNkAEmGzPntj3ziqB --- CHANGELOG.md | 15 +- docs/troubleshooting.md | 20 +- shelfmark/core/models.py | 1 + shelfmark/core/queue.py | 29 +++ shelfmark/download/orchestrator.py | 183 +++++++++++------- .../release_sources/prowlarr/mam_account.py | 5 +- tests/download/test_wait_in_client.py | 139 +++++++++---- tests/prowlarr/test_mam_unsat_safeguards.py | 22 +++ 8 files changed, 296 insertions(+), 118 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 36af3e918..2bb0e50c6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,6 +16,14 @@ Releases of this fork. Images: `ghcr.io/mikaeltarquin/shelfmark:` and ` ### Changed +- **Torrents download in the background, so more go at once.** A worker used to stay + with each torrent until the client had downloaded it, so only **Max Concurrent + Downloads** torrents (3 by default) were ever in flight, however many MyAnonamouse + slots were free. Now, once Deluge, qBittorrent, Transmission or rTorrent has had a + torrent for 30 seconds, its worker moves on to the next download; the torrent stays in + **Downloads** with the client's progress, and a worker comes back to pick up the files + when it's finished. MAM grabs are now limited by the free unsatisfied slots (and the + slots kept free), not by the workers. - **The MyAnonamouse panel's Buffer is now Room to your ratio**: how much more can be downloaded before the ratio drops below **Keep Ratio At Least**, counting Shelfmark's active MAM downloads. The plain buffer is in its tooltip. @@ -32,10 +40,9 @@ Releases of this fork. Images: `ghcr.io/mikaeltarquin/shelfmark:` and ` lifts, with a countdown in the top bar. - qBittorrent's MyAnonamouse torrents with no working tracker are no longer left out of the slot timing. -- **A torrent with no progress for 5 minutes waits in the torrent client instead of being - cancelled.** It was left downloading there with nothing to pick up the files, and shown - as Cancelled. Now it waits on **Activity → Queued** ("Waiting in Deluge: no progress…") - and carries on once the client shows it moving or finished. +- **A torrent with no progress for 5 minutes is no longer cancelled.** It was left + downloading in the torrent client with nothing to pick up the files, and shown as + Cancelled. Now it stays in Downloads, saying how long it has made no progress. - **A retry finds its torrent in the client by its hash**, without downloading the .torrent again, so it works while MyAnonamouse has frozen downloads. The hash is kept from when the torrent was handed to the client. diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md index 6c15a0321..063a26d45 100644 --- a/docs/troubleshooting.md +++ b/docs/troubleshooting.md @@ -37,15 +37,17 @@ IP registration (mousehole, VPN) and the downloads go once a re-announce succeed until the time shown. See [Unsatisfied limit](myanonamouse-enrichment.md#unsatisfied-limit). -## "Waiting in Deluge: no progress for 5 minutes" - -Not an error: the torrent is in your torrent client but hasn't moved for 5 minutes (no -peers yet, queued in the client, or the tracker refusing it for now). It goes on -downloading there, and Shelfmark picks it up again once the client shows it moving or -finished (checked every 2 minutes), so it doesn't hold up other downloads meanwhile. If -the torrent is removed from the client, the download fails saying so. Cancel it from -Activity to stop waiting. Like other waiting downloads it doesn't survive a restart: use -Retry, which finds the torrent in the client. +## "Downloading in Deluge" or "No progress in Deluge for … minutes" + +Not an error. Once the torrent client has had a torrent for 30 seconds, Shelfmark leaves +it downloading there and moves on to the next download, so more torrents download at +once than there are download workers. The row in **Downloads** shows the client's +progress (checked every 30 seconds), and the files are picked up when it's finished. +"No progress" means it hasn't moved for 5 minutes or more: no peers yet, queued in the +client, or the tracker refusing it for now. If the torrent is removed from the client, +the download fails saying so. Cancel it from Activity to stop watching it (the torrent +stays in the client). After a restart it shows as interrupted: Retry finds the torrent +in the client by its hash. ## A queued book doesn't download diff --git a/shelfmark/core/models.py b/shelfmark/core/models.py index d4fc77ba7..d5ad9ff25 100644 --- a/shelfmark/core/models.py +++ b/shelfmark/core/models.py @@ -110,6 +110,7 @@ class DownloadTask: # The info hash the torrent client holds the download by, once handed to one that keeps # it (Deluge, qBittorrent, Transmission, rTorrent): a stalled one waits there for peers. torrent_client_hash: str | None = None + handoff_recorded: bool = False # Counted as a snatch already (see BookQueue) retry_ratio_limit: float | None = None # Optional post-download seeding ratio retry_seeding_time_limit_minutes: int | None = None # Optional post-download seeding time limit retry_source_context: dict[str, Any] = field( diff --git a/shelfmark/core/queue.py b/shelfmark/core/queue.py index 89ffdd13b..322772ee3 100644 --- a/shelfmark/core/queue.py +++ b/shelfmark/core/queue.py @@ -165,7 +165,10 @@ def _apply_status_locked( and previous_status in _NOT_STARTED and status not in _NOT_STARTED and status != QueueStatus.CANCELLED + # Back only to pick up files its torrent client finished: no new snatch. + and not (task.torrent_client_hash and task.handoff_recorded) ): + task.handoff_recorded = True now = time.time() self._handoffs[book_id] = (now, task) cutoff = now - _HANDOFF_MEMORY_SECONDS @@ -207,6 +210,9 @@ def forget_handoff(self, task_id: str) -> None: """Drop a task's handoff: it never reached the indexer (refused at grab).""" with self._lock: self._handoffs.pop(task_id, None) + task = self._task_data.get(task_id) + if task is not None: + task.handoff_recorded = False def hold(self, task_id: str, message: str) -> bool: """Put a picked-up task back as queued without queueing it, to wait for room. @@ -225,6 +231,29 @@ def hold(self, task_id: str, message: str) -> bool: self._update_status(task_id, QueueStatus.QUEUED) return True + def detach(self, task_id: str, message: str) -> bool: + """Free a downloading task's worker while its download goes on elsewhere. + + For a torrent the client is downloading: it stays downloading, with `message`, + until `enqueue_existing` brings it back to pick up the files (or it is + cancelled). Its retry data is saved again, so a retry after a restart finds + the torrent in the client. + """ + with self._lock: + task = self._task_data.get(task_id) + if task is None or self._status.get(task_id) != QueueStatus.DOWNLOADING: + return False + self._active_downloads.pop(task_id, None) + self._cancel_flags.pop(task_id, None) + task.status_message = message + hook = self._queue_hook + if hook is not None: + try: + hook(task_id, task) + except _QUEUE_HOOK_ERRORS as exc: + logger.warning("Queue hook failed while detaching task %s: %s", task_id, exc) + return True + def update_status(self, book_id: str, status: QueueStatus) -> None: """Update status of a book in the queue.""" with self._lock: diff --git a/shelfmark/download/orchestrator.py b/shelfmark/download/orchestrator.py index 89be5311a..13e6ff22f 100644 --- a/shelfmark/download/orchestrator.py +++ b/shelfmark/download/orchestrator.py @@ -1291,28 +1291,26 @@ def _ensure_slot_watcher() -> None: _slot_watcher.start() -# --- Waiting in the torrent client ---------------------------------------------------- +# --- Torrents downloading in the background ------------------------------------------- # -# A torrent that makes no progress for the stall timeout is usually waiting on the -# torrent client (no peers yet, the client's own queue, a tracker refusing it for now), -# not stuck in Shelfmark, and it goes on downloading there. Cancelling it would leave -# it downloading with nothing to pick up the files, so instead it gives up its worker -# and waits in the queue, and goes back to the workers once the client shows it -# progressing (or finished). Cancel it from Activity to drop it. - -_CLIENT_RECHECK_SECONDS = 2 * 60 -CLIENT_WAIT_MESSAGE = ( - f"Waiting in {{client}}: no progress for {STALL_TIMEOUT // 60} minutes (no peers yet, " - f"or queued there). Picks up when it moves (checked every " - f"{_CLIENT_RECHECK_SECONDS // 60} minutes)" -) -_parking: set[str] = set() # Stalled torrents handing their worker back -_waiting_in_client: dict[str, float] = {} # task id -> its progress when it stopped +# Once a torrent client that keeps torrents by hash (Deluge, qBittorrent, Transmission, +# rTorrent) has a torrent, nothing needs a worker until its files are ready: holding one +# would let only as many torrents download as there are workers (3 by default), however +# many MyAnonamouse slots are free. So after a short while (or at once, when it stalls) +# the worker hands it over and is free for the next download. It shows as downloading, +# with the client's progress, and goes back to the workers when the client has finished +# it, to pick up the files. Cancel it from Activity to drop it. + +# Small downloads usually finish within this: no need to hand them over. +WATCH_AFTER_SECONDS = 30 +_WATCH_INTERVAL_SECONDS = 30 +_watching: set[str] = set() # Handing their worker back; see `_watch_in_background` +_watched: dict[str, tuple[float, float]] = {} # task id -> (progress, when it last moved) _client_watcher: threading.Thread | None = None -def _park_stalled_torrent(task_id: str, cancel_flag: Event) -> bool: - """Send a stalled download that's in the torrent client back to wait there.""" +def _watch_in_background(task_id: str, cancel_flag: Event) -> bool: + """Hand a torrent the client is downloading to the background watcher.""" task = book_queue.get_task(task_id) if ( task is None @@ -1321,43 +1319,77 @@ def _park_stalled_torrent(task_id: str, cancel_flag: Event) -> bool: ): return False with _held_lock: - _parking.add(task_id) - logger.info("Download %s stalled in the torrent client; it waits there", task_id) + _watching.add(task_id) + logger.debug("Download %s goes on in the torrent client; its worker is free", task_id) cancel_flag.set() # The worker stops polling and leaves the torrent in the client return True -def _settle_parked(task_id: str) -> bool: - """Once its worker has stopped, put a parked download in the queue to wait.""" +def _due_for_watching( + task_ids: Iterable[str], in_client_since: dict[str, float], now: float +) -> list[str]: + """The workers' torrents downloading in the client for `WATCH_AFTER_SECONDS`. + + `in_client_since` remembers when each was first seen there. + """ + due: list[str] = [] + for task_id in task_ids: + task = book_queue.get_task(task_id) + if ( + task is None + or not task.torrent_client_hash + or book_queue.get_task_status(task_id) != QueueStatus.DOWNLOADING + ): + continue + if now - in_client_since.setdefault(task_id, now) >= WATCH_AFTER_SECONDS: + due.append(task_id) + return due + + +def _settle_watched(task_id: str) -> bool: + """Once its worker has stopped, leave a handed-over torrent downloading in the client.""" with _held_lock: - if task_id not in _parking: + if task_id not in _watching: return False - _parking.discard(task_id) + _watching.discard(task_id) task = book_queue.get_task(task_id) if task is None: return False progress = task.progress or 0.0 - if not book_queue.hold(task_id, CLIENT_WAIT_MESSAGE.format(client=_client_name())): + message = f"Downloading in {_client_name()} ({progress:.0f}%)" + if not book_queue.detach(task_id, message): return False with _held_lock: - _waiting_in_client[task_id] = progress + _watched[task_id] = (progress, time.time()) _ensure_client_watcher() if ws_manager: ws_manager.broadcast_status_update(queue_status()) return True -def waiting_in_client() -> list[str]: - """Downloads waiting for the torrent client to move them.""" +def watched_downloads() -> list[str]: + """Torrents downloading in the client with no worker, watched in the background.""" with _held_lock: - return list(_waiting_in_client) + return list(_watched) -def release_moving_torrents() -> list[str]: - """Queue the waiting downloads whose torrents have moved; fail ones that are gone.""" +def _watch_message(client_name: str, progress: float, idle_seconds: float) -> str: + if idle_seconds >= STALL_TIMEOUT: + return ( + f"No progress in {client_name} for {int(idle_seconds // 60)} minutes at " + f"{progress:.0f}% (no peers yet, or queued there)" + ) + return f"Downloading in {client_name} ({progress:.0f}%)" + + +def check_watched_downloads() -> list[str]: + """Update the watched torrents' progress; send finished ones back to the workers. + + Returns the ones sent back. One no longer in the client fails, saying so. + """ with _held_lock: - waiting = dict(_waiting_in_client) - if not waiting: + watched = dict(_watched) + if not watched: return [] from shelfmark.download.clients import DownloadState, client_label, get_client @@ -1367,13 +1399,15 @@ def release_moving_torrents() -> list[str]: except Exception as exc: # noqa: BLE001 - each client library raises its own errors logger.debug("Could not reach the torrent client: %s", exc) return [] - released: list[str] = [] - failed = False - for task_id, progress in waiting.items(): + name = client_label(client) or "the torrent client" + now = time.time() + finished: list[str] = [] + changed = False + for task_id, (last_progress, last_moved) in watched.items(): task = book_queue.get_task(task_id) - if task is None or book_queue.get_task_status(task_id) != QueueStatus.QUEUED: + if task is None or book_queue.get_task_status(task_id) != QueueStatus.DOWNLOADING: with _held_lock: # Cancelled, or retried by hand - _waiting_in_client.pop(task_id, None) + _watched.pop(task_id, None) continue if client is None or not task.torrent_client_hash: continue @@ -1382,29 +1416,38 @@ def release_moving_torrents() -> list[str]: except Exception as exc: # noqa: BLE001 - try again next round logger.debug("Could not read %s from the torrent client: %s", task_id, exc) continue - if status.state == DownloadState.ERROR and "not found" in (status.message or "").lower(): + if status.state == DownloadState.ERROR: + if "not found" not in (status.message or "").lower(): + continue # Can't tell; try again next round with _held_lock: - _waiting_in_client.pop(task_id, None) - book_queue.update_status_message( - task_id, f"The torrent is no longer in {client_label(client) or 'the client'}" - ) + _watched.pop(task_id, None) + book_queue.update_status_message(task_id, f"The torrent is no longer in {name}") book_queue.update_status(task_id, QueueStatus.ERROR) - failed = True + changed = True continue - if status.state == DownloadState.ERROR: - continue # Can't tell; try again next round - if not status.complete and (status.progress or 0.0) <= progress: + if status.complete: + with _held_lock: + _watched.pop(task_id, None) + if book_queue.enqueue_existing(task_id, priority=-10): + book_queue.update_status_message(task_id, f"Finished in {name}; importing") + finished.append(task_id) continue - with _held_lock: - _waiting_in_client.pop(task_id, None) - if book_queue.enqueue_existing(task_id): - book_queue.update_status_message(task_id, "Moving again in the torrent client") - released.append(task_id) - if released: - logger.info("Torrents moving again in the client: %s", released) - if (released or failed) and ws_manager: + progress = status.progress or 0.0 + moved_at = now if progress > last_progress else last_moved + if moved_at != last_moved: + with _held_lock: + if task_id in _watched: + _watched[task_id] = (progress, moved_at) + message = _watch_message(name, progress, now - moved_at) + if task.progress != progress or task.status_message != message: + book_queue.update_progress(task_id, progress) + book_queue.update_status_message(task_id, message) + changed = True + if finished: + logger.info("Torrents finished in the client, importing: %s", finished) + if (finished or changed) and ws_manager: ws_manager.broadcast_status_update(queue_status()) - return released + return finished def _client_name() -> str: @@ -1419,13 +1462,13 @@ def _client_name() -> str: def _client_watch_loop() -> None: while True: - time.sleep(_CLIENT_RECHECK_SECONDS) + time.sleep(_WATCH_INTERVAL_SECONDS) try: - release_moving_torrents() + check_watched_downloads() except Exception: # the watcher must outlive any one check - logger.exception("Checking downloads waiting in the torrent client failed") + logger.exception("Checking torrents downloading in the client failed") with _held_lock: - if not _waiting_in_client: + if not _watched: return @@ -1446,7 +1489,7 @@ def _process_single_download(task_id: str, cancel_flag: Event) -> None: is_mam = bool(task and task.mam_torrent_id) with _held_lock: _held_for_slot.pop(task_id, None) # Picked up: released, or retried by hand - _waiting_in_client.pop(task_id, None) + _watched.pop(task_id, None) if is_mam and not cancel_flag.is_set(): wait = _slot_wait(task_id) if wait is not None and _hold_for_slot(task_id, wait): @@ -1460,7 +1503,7 @@ def _process_single_download(task_id: str, cancel_flag: Event) -> None: _cleanup_progress_tracking(task_id) if cancel_flag.is_set(): - if _settle_parked(task_id): + if _settle_watched(task_id): return book_queue.update_status(task_id, QueueStatus.CANCELLED) # Broadcast cancellation @@ -1553,6 +1596,8 @@ def concurrent_download_loop() -> None: with ThreadPoolExecutor(max_workers=max_workers, thread_name_prefix="Download") as executor: active_futures: dict[Future, tuple[str, Event]] = {} # Track active download futures stalled_tasks: set[str] = set() # Track tasks already cancelled due to stall + # When each worker's torrent was first seen downloading in the client + in_client_since: dict[str, float] = {} consecutive_errors = 0 while True: @@ -1562,9 +1607,10 @@ def concurrent_download_loop() -> None: for future in completed_futures: task_id, cancel_flag = active_futures.pop(future) stalled_tasks.discard(task_id) + in_client_since.pop(task_id, None) if future.cancelled(): _cleanup_progress_tracking(task_id) - if cancel_flag.is_set() and _settle_parked(task_id): + if cancel_flag.is_set() and _settle_watched(task_id): pass elif cancel_flag.is_set(): logger.info("Download cancelled: %s", task_id) @@ -1588,7 +1634,7 @@ def concurrent_download_loop() -> None: continue _cleanup_progress_tracking(task_id) - if cancel_flag.is_set() and _settle_parked(task_id): + if cancel_flag.is_set() and _settle_watched(task_id): pass elif cancel_flag.is_set(): logger.info("Download cancelled: %s", task_id) @@ -1615,10 +1661,17 @@ def concurrent_download_loop() -> None: ] flags = dict(active_futures.values()) for task_id in _find_stalled_tasks(candidates, current_time): - if not _park_stalled_torrent(task_id, flags[task_id]): + if not _watch_in_background(task_id, flags[task_id]): _cancel_stalled_task(task_id) stalled_tasks.add(task_id) + # Torrents the client has been downloading a while: free their workers + for task_id in _due_for_watching(candidates, in_client_since, current_time): + if task_id not in stalled_tasks and _watch_in_background( + task_id, flags[task_id] + ): + stalled_tasks.add(task_id) + # Start new downloads if we have capacity while len(active_futures) < max_workers: next_download = book_queue.get_next() diff --git a/shelfmark/release_sources/prowlarr/mam_account.py b/shelfmark/release_sources/prowlarr/mam_account.py index 9487b79db..f9dba439e 100644 --- a/shelfmark/release_sources/prowlarr/mam_account.py +++ b/shelfmark/release_sources/prowlarr/mam_account.py @@ -648,7 +648,10 @@ def pending_unsat_count( counted = { task.task_id for status, task in _active_mam_tasks() - if task.mam_torrent_id and status in not_started + if task.mam_torrent_id + and status in not_started + # Back only to pick up the files of a torrent the client has: MAM counts it + and not (task.torrent_client_hash and task.handoff_recorded) } since = (stats_fetched_at if stats_fetched_at is not None else time.time()) - _UNSAT_LAG_SECONDS snatched = [ diff --git a/tests/download/test_wait_in_client.py b/tests/download/test_wait_in_client.py index 6f5fffa75..42c8edba0 100644 --- a/tests/download/test_wait_in_client.py +++ b/tests/download/test_wait_in_client.py @@ -1,4 +1,4 @@ -"""A stalled torrent waits in the torrent client instead of being cancelled, and a +"""A torrent the client is downloading goes on there without holding a worker, and a download in the client is found by its hash, without downloading the .torrent again. """ @@ -23,8 +23,8 @@ def queue(monkeypatch): monkeypatch.setattr(orchestrator, "book_queue", fresh) monkeypatch.setattr(orchestrator, "ws_manager", None) monkeypatch.setattr(orchestrator, "_held_for_slot", {}) - monkeypatch.setattr(orchestrator, "_parking", set()) - monkeypatch.setattr(orchestrator, "_waiting_in_client", {}) + monkeypatch.setattr(orchestrator, "_watching", set()) + monkeypatch.setattr(orchestrator, "_watched", {}) monkeypatch.setattr(orchestrator, "_ensure_client_watcher", lambda: None) monkeypatch.setattr(orchestrator, "_client_name", lambda: "Deluge") monkeypatch.setattr(mam_account, "grab_allowance", lambda **_kw: mam_account.GrabAllowance(9)) @@ -55,15 +55,16 @@ def find_existing(self, url, category=None): return (HASH, self.status) if url in self.present else None -def _stall_in_client(queue, *, progress: float = 0.0, client_hash: str | None = HASH): - """A download handed to the client, stalled there: what the stall check sees.""" +def _in_client(queue, *, progress: float = 0.0, client_hash: str | None = HASH): + """A download handed to the client, then handed over to the watcher (after + `WATCH_AFTER_SECONDS`, or a stall), or cancelled when it can't be.""" def download(task_id, flag): task = queue.get_task(task_id) task.torrent_client_hash = client_hash task.progress = progress queue.update_status(task_id, QueueStatus.DOWNLOADING) - if not orchestrator._park_stalled_torrent(task_id, flag): + if not orchestrator._watch_in_background(task_id, flag): orchestrator._cancel_stalled_task(task_id) return None @@ -75,70 +76,130 @@ def _status(progress: float, *, complete: bool = False) -> DownloadStatus: return DownloadStatus(progress, state, None, complete, None) -def test_a_stalled_torrent_waits_in_the_client_instead_of_being_cancelled(queue, monkeypatch): - monkeypatch.setattr(orchestrator, "_download_task", _stall_in_client(queue, progress=12.0)) +def _use(monkeypatch, client) -> None: + monkeypatch.setattr("shelfmark.download.clients.get_client", lambda _p: client) + + +def test_a_torrent_in_the_client_frees_its_worker_and_stays_downloading(queue, monkeypatch): + monkeypatch.setattr(orchestrator, "_download_task", _in_client(queue, progress=12.0)) orchestrator._process_single_download("t1", _add(queue, "t1")) - assert queue.get_task_status("t1") == QueueStatus.QUEUED - assert queue.get_task("t1").status_message.startswith("Waiting in Deluge") - assert orchestrator.waiting_in_client() == ["t1"] - assert orchestrator._waiting_in_client["t1"] == 12.0 - assert queue.get_next() is None # Not back with the workers until it moves + assert queue.get_task_status("t1") == QueueStatus.DOWNLOADING + assert queue.get_task("t1").status_message == "Downloading in Deluge (12%)" + assert orchestrator.watched_downloads() == ["t1"] + assert queue.get_next() is None # Not back with the workers until it's finished + + +def test_its_retry_data_is_saved_with_the_hash(queue, monkeypatch): + saved: list[str | None] = [] + queue.set_queue_hook(lambda _id, task: saved.append(task.torrent_client_hash)) + monkeypatch.setattr(orchestrator, "_download_task", _in_client(queue)) + orchestrator._process_single_download("t1", _add(queue, "t1")) + assert saved[-1] == HASH -def test_a_stall_outside_a_torrent_client_still_cancels(queue, monkeypatch): - monkeypatch.setattr(orchestrator, "_download_task", _stall_in_client(queue, client_hash=None)) +def test_a_download_outside_a_torrent_client_still_cancels_on_a_stall(queue, monkeypatch): + monkeypatch.setattr(orchestrator, "_download_task", _in_client(queue, client_hash=None)) orchestrator._process_single_download("t1", _add(queue, "t1")) assert queue.get_task_status("t1") == QueueStatus.CANCELLED - assert orchestrator.waiting_in_client() == [] + assert orchestrator.watched_downloads() == [] -def test_goes_back_to_the_workers_once_it_moves(queue, monkeypatch): - monkeypatch.setattr(orchestrator, "_download_task", _stall_in_client(queue, progress=12.0)) +def test_progress_is_shown_and_a_finished_torrent_goes_back_to_import(queue, monkeypatch): + monkeypatch.setattr(orchestrator, "_download_task", _in_client(queue, progress=12.0)) orchestrator._process_single_download("t1", _add(queue, "t1")) - client = FakeClient(_status(12.0)) - monkeypatch.setattr("shelfmark.download.clients.get_client", lambda _p: client) + client = FakeClient(_status(40.0)) + _use(monkeypatch, client) + + assert orchestrator.check_watched_downloads() == [] + assert queue.get_task("t1").progress == 40.0 + assert queue.get_task("t1").status_message == "Downloading in Deluge (40%)" - assert orchestrator.release_moving_torrents() == [] - client.status = _status(30.0) - assert orchestrator.release_moving_torrents() == ["t1"] - assert orchestrator.waiting_in_client() == [] + client.status = _status(100.0, complete=True) + assert orchestrator.check_watched_downloads() == ["t1"] + assert orchestrator.watched_downloads() == [] picked = queue.get_next() assert picked is not None assert picked[0] == "t1" -def test_a_finished_torrent_goes_back_too(queue, monkeypatch): - monkeypatch.setattr(orchestrator, "_download_task", _stall_in_client(queue)) +def test_a_torrent_with_no_progress_says_so(queue, monkeypatch): + monkeypatch.setattr(orchestrator, "_download_task", _in_client(queue, progress=5.0)) orchestrator._process_single_download("t1", _add(queue, "t1")) - client = FakeClient(_status(0.0, complete=True)) - monkeypatch.setattr("shelfmark.download.clients.get_client", lambda _p: client) - assert orchestrator.release_moving_torrents() == ["t1"] + _use(monkeypatch, FakeClient(_status(5.0))) + later = orchestrator.time.time() + orchestrator.STALL_TIMEOUT + 60 + monkeypatch.setattr(orchestrator.time, "time", lambda: later) + + orchestrator.check_watched_downloads() + message = queue.get_task("t1").status_message + assert message.startswith("No progress in Deluge for 6 minutes at 5%") + assert queue.get_task_status("t1") == QueueStatus.DOWNLOADING def test_a_torrent_removed_from_the_client_fails(queue, monkeypatch): - monkeypatch.setattr(orchestrator, "_download_task", _stall_in_client(queue)) + monkeypatch.setattr(orchestrator, "_download_task", _in_client(queue)) orchestrator._process_single_download("t1", _add(queue, "t1")) - client = FakeClient(DownloadStatus.error("Torrent not found")) - monkeypatch.setattr("shelfmark.download.clients.get_client", lambda _p: client) + _use(monkeypatch, FakeClient(DownloadStatus.error("Torrent not found"))) - assert orchestrator.release_moving_torrents() == [] + assert orchestrator.check_watched_downloads() == [] assert queue.get_task_status("t1") == QueueStatus.ERROR assert queue.get_task("t1").status_message == "The torrent is no longer in Deluge" -def test_a_cancelled_wait_is_dropped(queue, monkeypatch): - monkeypatch.setattr(orchestrator, "_download_task", _stall_in_client(queue)) +def test_a_cancelled_one_is_dropped(queue, monkeypatch): + monkeypatch.setattr(orchestrator, "_download_task", _in_client(queue)) orchestrator._process_single_download("t1", _add(queue, "t1")) - queue.cancel_download("t1") - monkeypatch.setattr("shelfmark.download.clients.get_client", lambda _p: FakeClient()) + assert queue.cancel_download("t1") is True + _use(monkeypatch, FakeClient()) - assert orchestrator.release_moving_torrents() == [] - assert orchestrator.waiting_in_client() == [] + assert orchestrator.check_watched_downloads() == [] + assert orchestrator.watched_downloads() == [] assert queue.get_task_status("t1") == QueueStatus.CANCELLED +def test_due_for_watching_after_a_while_in_the_client(queue): + _add(queue, "t1") + task = queue.get_task("t1") + task.torrent_client_hash = HASH + queue.update_status("t1", QueueStatus.DOWNLOADING) + since: dict[str, float] = {} + + assert orchestrator._due_for_watching(["t1"], since, 1000.0) == [] + assert orchestrator._due_for_watching(["t1"], since, 1010.0) == [] + later = 1000.0 + orchestrator.WATCH_AFTER_SECONDS + assert orchestrator._due_for_watching(["t1"], since, later) == ["t1"] + + +def test_not_due_before_the_client_has_it(queue): + _add(queue, "t1") + queue.update_status("t1", QueueStatus.DOWNLOADING) # No hash: not in a client + assert orchestrator._due_for_watching(["t1"], {"t1": 0.0}, 10_000.0) == [] + + +class TestOneSnatch: + """Coming back to import doesn't count as a new MAM snatch.""" + + def test_the_handoff_is_recorded_once(self, queue): + _add(queue, "t1", mam_torrent_id=7) + queue.get_task("t1").torrent_client_hash = HASH + queue.update_status("t1", QueueStatus.DOWNLOADING) + first = queue.handoff_times_since(0) + assert [task.task_id for _at, task in first] == ["t1"] + + assert queue.enqueue_existing("t1") + queue.get_next() + queue.update_status("t1", QueueStatus.RESOLVING) + queue.update_status("t1", QueueStatus.DOWNLOADING) + assert queue.handoff_times_since(0) == first + + def test_a_refused_grab_counts_on_the_next_try(self, queue): + _add(queue, "t1", mam_torrent_id=7) + queue.update_status("t1", QueueStatus.DOWNLOADING) + queue.forget_handoff("t1") + assert queue.get_task("t1").handoff_recorded is False + + class TestFoundByHash: URL = "https://prowlarr/download?id=1" diff --git a/tests/prowlarr/test_mam_unsat_safeguards.py b/tests/prowlarr/test_mam_unsat_safeguards.py index 6e8f3a69f..39bfeb64b 100644 --- a/tests/prowlarr/test_mam_unsat_safeguards.py +++ b/tests/prowlarr/test_mam_unsat_safeguards.py @@ -330,3 +330,25 @@ def get(_url, params, timeout): for _ in range(3): assert [t.name for t in qbit.list_tracker_torrents("myanonamouse")] == ["P"] assert sorted(asked) == ["o", "p"] + + +def test_a_torrent_back_only_to_import_is_not_a_new_snatch(monkeypatch): + from shelfmark.core.models import DownloadTask, QueueStatus + from shelfmark.core.queue import book_queue + + new = DownloadTask(task_id="new", source="prowlarr", title="t", mam_torrent_id=1) + back = DownloadTask( + task_id="back", + source="prowlarr", + title="t", + mam_torrent_id=2, + torrent_client_hash="a" * 40, + handoff_recorded=True, + ) + monkeypatch.setattr( + mam_account, + "_active_mam_tasks", + lambda: [(QueueStatus.QUEUED, new), (QueueStatus.QUEUED, back)], + ) + monkeypatch.setattr(book_queue, "handoff_times_since", lambda _since: []) + assert mam_account.pending_unsat_count() == 1