Self-review fixes before opening PR

Three findings from a final review pass:

1. **Worker clobbered Cancelled with Errored when impl returned
   None / raised mid-cancel.** The legacy per-client thread workers
   each had a guard (``if state != 'Cancelled': state = 'Errored'``);
   the shared worker dropped it. Fix: new ``_mark_terminal`` helper
   in BackgroundDownloadWorker reads current state before writing
   the terminal one and leaves Cancelled alone. SoundCloud test
   updated back to the strict Cancelled-only assertion (had been
   loosened to accept Errored as a workaround). Two new pinning
   tests catch the regression.

2. **Dead code in engine.py.** ``find_record`` and
   ``iter_all_records`` had no production callers — only tests.
   Removed them. Concurrent-add stress test rewritten to use the
   per-source iterator that's actually in use.

3. **Silent ``except Exception: pass`` in cross-source query
   methods.** Faithful to legacy behavior (one source failing
   shouldn't take down aggregation) but Cin's standard is "log
   even when you swallow." Each silent-swallow site now logs at
   debug level so the source name + exception are inspectable
   without adding warning-level noise.

Suite still green (2049 passed).

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:
+            except Exception as exc:
-                pass
+                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:
+            except Exception as exc:
-                pass
+                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:
+            except Exception as exc:
-                pass
+                logger.debug("%s cancel_download failed: %s", source_name, exc)
         return False
 
     async def clear_all_completed_downloads(self) -> bool:

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, {
+                    self._mark_terminal(
-                        'state': 'Errored',
+                        source_name, download_id,
-                        'error': str(exc),
+                        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, {
+                    self._mark_terminal(source_name, download_id, success=False)
-                        'state': 'Errored',
-                    })
                     logger.error(
                         "%s download %s failed (impl returned None)",
                         source_name, download_id,
                     )
 
         except Exception as exc:
-            # Defensive — anything in the worker_loop itself
+            # Defensive — semaphore / sleep shouldn't blow up the
-            # (semaphore, sleep) shouldn't blow up the thread, but
+            # thread, but if they do the record needs SOME terminal
-            # if it does the record gets marked Errored so the
+            # state or it sits at 'Initializing' forever.
-            # download doesn't sit forever in 'Initializing'.
             logger.exception(
                 "%s worker_loop crashed for download %s: %s",
                 source_name, download_id, exc,
             )
-            self._engine.update_record(source_name, download_id, {
+            self._mark_terminal(
-                'state': 'Errored',
+                source_name, download_id,
-                'error': f'worker crash: {exc}',
+                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)

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()
 

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
 
 

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
+    None, the worker must NOT overwrite the explicit Cancelled state
-    state with a generic Errored state.
+    with a generic Errored state. The legacy per-client thread had
-
+    this guard; engine.worker._mark_terminal preserves it for every
-    NOTE: Phase C7 lifted state into engine.worker. The worker DOES
+    source via a single check."""
-    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."""
     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']
+    assert engine.get_record('soundcloud', download_id)['state'] == 'Cancelled'
-    # 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')
 
 
 # ---------------------------------------------------------------------------