diff --git a/core/download_engine/engine.py b/core/download_engine/engine.py index 2226ccb1..358ef07e 100644 --- a/core/download_engine/engine.py +++ b/core/download_engine/engine.py @@ -174,34 +174,6 @@ class DownloadEngine: for record in snapshot: yield record - def iter_all_records(self) -> Iterator[Tuple[str, DownloadRecord]]: - """Yield ``(source_name, record_copy)`` for every active - download across every source. Used by Phase B3's unified - ``get_all_downloads`` query.""" - with self.state_lock: - snapshot = [ - (source, dict(record)) - for (source, _), record in self._records.items() - ] - for source, record in snapshot: - yield source, record - - def find_record(self, download_id: str) -> Optional[Tuple[str, DownloadRecord]]: - """Look up a record by download_id alone (no source hint). - Used by ``cancel_download`` / ``get_download_status`` API - endpoints that don't pass the source name. Returns - ``(source_name, record_copy)`` or None. - - O(N) over total downloads — fine for the tens-to-hundreds - of in-flight transfers SoulSync sees, would need an index - if downloads scaled to thousands. - """ - with self.state_lock: - for (source, dl_id), record in self._records.items(): - if dl_id == download_id: - return source, dict(record) - return None - # ------------------------------------------------------------------ # Cross-source query dispatch — Phase B2 surface # ------------------------------------------------------------------ @@ -222,29 +194,32 @@ class DownloadEngine: async def get_all_downloads(self): """Aggregated view across every registered plugin's active - downloads. Returns a flat list of DownloadStatus objects.""" + downloads. Per-plugin exceptions are swallowed (one source + failing shouldn't take down cross-source aggregation) but + logged at debug level — same defensive shape the legacy + orchestrator had.""" all_downloads = [] - for plugin in self._plugins.values(): + for source_name, plugin in self._plugins.items(): if plugin is None: continue try: all_downloads.extend(await plugin.get_all_downloads()) - except Exception: - pass + except Exception as exc: + logger.debug("%s get_all_downloads failed: %s", source_name, exc) return all_downloads async def get_download_status(self, download_id: str): """Find a download_id across every plugin. Returns the first plugin's response or None if no plugin owns it.""" - for plugin in self._plugins.values(): + for source_name, plugin in self._plugins.items(): if plugin is None: continue try: status = await plugin.get_download_status(download_id) if status: return status - except Exception: - pass + except Exception as exc: + logger.debug("%s get_download_status failed: %s", source_name, exc) return None async def cancel_download(self, download_id: str, @@ -265,24 +240,26 @@ class DownloadEngine: return await target_plugin.cancel_download( download_id, source_hint, remove, ) - except Exception: + except Exception as exc: + logger.debug("%s cancel_download failed: %s", source_hint, exc) return False soulseek = self._plugins.get('soulseek') if soulseek is not None: try: return await soulseek.cancel_download(download_id, source_hint, remove) - except Exception: + except Exception as exc: + logger.debug("soulseek cancel_download failed: %s", exc) return False # No hint → ask every plugin until one cancels successfully. - for plugin in self._plugins.values(): + for source_name, plugin in self._plugins.items(): if plugin is None: continue try: if await plugin.cancel_download(download_id, source_hint, remove): return True - except Exception: - pass + except Exception as exc: + logger.debug("%s cancel_download failed: %s", source_name, exc) return False async def clear_all_completed_downloads(self) -> bool: diff --git a/core/download_engine/worker.py b/core/download_engine/worker.py index 93ab17e1..8ba48a2a 100644 --- a/core/download_engine/worker.py +++ b/core/download_engine/worker.py @@ -247,10 +247,10 @@ class BackgroundDownloadWorker: "%s download %s failed (impl raised): %s", source_name, download_id, exc, ) - self._engine.update_record(source_name, download_id, { - 'state': 'Errored', - 'error': str(exc), - }) + self._mark_terminal( + source_name, download_id, + success=False, error=str(exc), + ) return self._last_download_at[source_name] = time.time() @@ -266,24 +266,40 @@ class BackgroundDownloadWorker: source_name, download_id, file_path, ) else: - self._engine.update_record(source_name, download_id, { - 'state': 'Errored', - }) + self._mark_terminal(source_name, download_id, success=False) logger.error( "%s download %s failed (impl returned None)", source_name, download_id, ) except Exception as exc: - # Defensive — anything in the worker_loop itself - # (semaphore, sleep) shouldn't blow up the thread, but - # if it does the record gets marked Errored so the - # download doesn't sit forever in 'Initializing'. + # Defensive — semaphore / sleep shouldn't blow up the + # thread, but if they do the record needs SOME terminal + # state or it sits at 'Initializing' forever. logger.exception( "%s worker_loop crashed for download %s: %s", source_name, download_id, exc, ) - self._engine.update_record(source_name, download_id, { - 'state': 'Errored', - 'error': f'worker crash: {exc}', - }) + self._mark_terminal( + source_name, download_id, + success=False, error=f'worker crash: {exc}', + ) + + def _mark_terminal(self, source_name: str, download_id: str, + success: bool, error: Optional[str] = None) -> None: + """Write a terminal state, but DON'T clobber an explicit + 'Cancelled' state set by the user via cancel_download. + Mirrors the legacy per-client guard + (``if state != 'Cancelled': state = 'Errored'``) every + client used to hand-roll inside its thread worker.""" + record = self._engine.get_record(source_name, download_id) + if record is None: + return # already removed (e.g. cancel + remove) + if record.get('state') == 'Cancelled': + return # user explicitly cancelled — leave it + patch: Dict[str, Any] = { + 'state': 'Completed, Succeeded' if success else 'Errored', + } + if error is not None: + patch['error'] = error + self._engine.update_record(source_name, download_id, patch) diff --git a/tests/downloads/test_background_download_worker.py b/tests/downloads/test_background_download_worker.py index 14c45126..70335d79 100644 --- a/tests/downloads/test_background_download_worker.py +++ b/tests/downloads/test_background_download_worker.py @@ -155,6 +155,67 @@ def test_worker_marks_completed_on_successful_impl(): assert record['file_path'] == '/tmp/done.flac' +def test_worker_preserves_cancelled_when_impl_returns_none(): + """Pinning: if the user cancels mid-download (state flips to + Cancelled via engine.update_record from cancel_download), the + worker must NOT clobber it back to Errored when impl returns + None. The legacy per-client thread workers had this guard + (``if state != 'Cancelled': state = 'Errored'``); the shared + worker preserves that contract.""" + engine = DownloadEngine() + + def impl(download_id, target_id, display_name): + # Simulate user cancelling mid-impl by writing Cancelled. + engine.update_record('youtube', download_id, {'state': 'Cancelled'}) + return None # impl returns None because download was interrupted + + download_id = engine.worker.dispatch( + source_name='youtube', + target_id='vid', + display_name='X', + original_filename='vid||X', + impl_callable=impl, + ) + + deadline = time.time() + 2.0 + while time.time() < deadline: + record = engine.get_record('youtube', download_id) + if record and record['state'] in ('Cancelled', 'Errored'): + break + time.sleep(0.01) + + record = engine.get_record('youtube', download_id) + assert record['state'] == 'Cancelled', ( + f"Worker clobbered user's Cancelled with {record['state']}" + ) + + +def test_worker_preserves_cancelled_when_impl_raises(): + """Same Cancelled-preserve guard, but for the impl-raises path.""" + engine = DownloadEngine() + + def impl(download_id, target_id, display_name): + engine.update_record('youtube', download_id, {'state': 'Cancelled'}) + raise RuntimeError("simulated mid-cancel exception") + + download_id = engine.worker.dispatch( + source_name='youtube', + target_id='vid', + display_name='X', + original_filename='vid||X', + impl_callable=impl, + ) + + deadline = time.time() + 2.0 + while time.time() < deadline: + record = engine.get_record('youtube', download_id) + if record and record['state'] in ('Cancelled', 'Errored'): + break + time.sleep(0.01) + + assert engine.get_record('youtube', download_id)['state'] == 'Cancelled' + + def test_worker_marks_errored_when_impl_returns_none(): engine = DownloadEngine() diff --git a/tests/downloads/test_download_engine.py b/tests/downloads/test_download_engine.py index d0c519b5..1845f085 100644 --- a/tests/downloads/test_download_engine.py +++ b/tests/downloads/test_download_engine.py @@ -140,19 +140,6 @@ def test_iter_records_for_source_filters_correctly(): assert td_records[0]['title'] == 'C' -def test_iter_all_records_yields_source_paired_with_record(): - engine = DownloadEngine() - engine.add_record('youtube', 'yt-1', {'title': 'A'}) - engine.add_record('tidal', 'td-1', {'title': 'B'}) - - pairs = list(engine.iter_all_records()) - assert len(pairs) == 2 - sources = {source for source, _ in pairs} - titles = {record['title'] for _, record in pairs} - assert sources == {'youtube', 'tidal'} - assert titles == {'A', 'B'} - - def test_iter_yields_shallow_copies(): """Iteration returns COPIES — caller can hold the list and mutate each record without affecting engine state. Important: future @@ -168,28 +155,6 @@ def test_iter_yields_shallow_copies(): assert fresh['title'] == 'A' -# --------------------------------------------------------------------------- -# find_record — id-only lookup -# --------------------------------------------------------------------------- - - -def test_find_record_returns_source_and_copy(): - engine = DownloadEngine() - engine.add_record('youtube', 'shared-id-shape', {'title': 'A'}) - - result = engine.find_record('shared-id-shape') - assert result is not None - source, record = result - assert source == 'youtube' - assert record['title'] == 'A' - - -def test_find_record_returns_none_for_unknown_id(): - engine = DownloadEngine() - engine.add_record('youtube', 'yt-1', {}) - assert engine.find_record('nonexistent') is None - - # --------------------------------------------------------------------------- # Thread safety — basic concurrent-mutation smoke # --------------------------------------------------------------------------- @@ -215,7 +180,11 @@ def test_concurrent_adds_dont_lose_records(): for t in threads: t.join() - total = sum(1 for _ in engine.iter_all_records()) + total = sum( + 1 + for n in range(4) + for _ in engine.iter_records_for_source(f'src-{n}') + ) assert total == 4 * 50 # 200 records, none lost diff --git a/tests/test_soundcloud_client.py b/tests/test_soundcloud_client.py index 7eee88ed..1c904724 100644 --- a/tests/test_soundcloud_client.py +++ b/tests/test_soundcloud_client.py @@ -322,18 +322,10 @@ def test_download_thread_marks_failed_when_sync_returns_none(tmp_dl: Path) -> No def test_download_thread_does_not_clobber_cancelled_state(tmp_dl: Path) -> None: """If a user cancels mid-download and the sync function then returns - None, the engine state should NOT overwrite the explicit Cancelled - state with a generic Errored state. - - NOTE: Phase C7 lifted state into engine.worker. The worker DOES - overwrite Cancelled with Errored when impl returns None — the - Cancelled-preserve guard the legacy per-client thread had isn't - in the shared worker. This test now pins the new behavior. - Future work: if Cancelled-preserve becomes important again, add - that guard to BackgroundDownloadWorker uniformly. For now the - cancellation-mid-download path is the user's explicit cancel - button, which calls cancel_download (engine.update_record) AFTER - the worker has already written its terminal state.""" + None, the worker must NOT overwrite the explicit Cancelled state + with a generic Errored state. The legacy per-client thread had + this guard; engine.worker._mark_terminal preserves it for every + source via a single check.""" client = SoundcloudClient(download_path=str(tmp_dl)) engine = _wire_engine(client) @@ -345,18 +337,13 @@ def test_download_thread_does_not_clobber_cancelled_state(tmp_dl: Path) -> None: with patch.object(client, '_download_sync', side_effect=_slow_sync): download_id = _run(client.download('soundcloud', '1||u||n')) - # Wait for the worker to finish (state will be terminal). deadline = time.time() + 2 while time.time() < deadline: record = engine.get_record('soundcloud', download_id) if record and record['state'] in ('Cancelled', 'Errored'): break time.sleep(0.05) - final_state = engine.get_record('soundcloud', download_id)['state'] - # Worker may have overwritten Cancelled with Errored — accept - # either since the overwrite-prevention isn't in the shared - # worker yet. - assert final_state in ('Cancelled', 'Errored') + assert engine.get_record('soundcloud', download_id)['state'] == 'Cancelled' # ---------------------------------------------------------------------------