diff --git a/controller/service.py b/controller/service.py index 0e2768a..eccf1fd 100644 --- a/controller/service.py +++ b/controller/service.py @@ -404,13 +404,15 @@ class ControllerService: def _handle_call(self, action: str, payload: Any) -> Any: if action == "enqueue_download": assert self.downloads is not None - if not self._cleanup_allowed(): + url = payload["url"] + if not self._cleanup_allowed() and self._ingest_requires_evidence(url): raise ControllerUnavailableError( - "downloads are deferred until fresh Player evidence or " - "administrative playback reconciliation" + "refreshing or replacing existing playlist media is deferred " + "until fresh Player evidence or administrative playback " + "reconciliation" ) reservation = self.downloads.enqueue( - payload["url"], payload.get("submitter"), payload.get("note") + url, payload.get("submitter"), payload.get("note") ) result: dict[str, Any] = { "status": "queued" if reservation.status == "reserved" else reservation.status, @@ -1261,6 +1263,20 @@ class ControllerService: def _cleanup_allowed(self) -> bool: return self._player_evidence_ready or self._operator_cleanup_override + def _ingest_requires_evidence(self, url: Any) -> bool: + """True when ingesting ``url`` could reuse/refresh occupied media. + + A URL that does not already occupy a slot can only land in a slot + that ``reserve_download`` independently proves is empty and + unguarded, so it cannot mutate, delete, or replace media the Player + may currently be using. Only the reuse/refresh path needs the same + conservative evidence gate as deletion and expiry. + """ + + if not isinstance(url, str): + return False + return self.store.find_occupied_slot_for_url(url) is not None + def _schedule_deletion_recovery(self) -> None: if not self._cleanup_allowed(): return diff --git a/controller/store.py b/controller/store.py index 6d89bb1..888ef6f 100644 --- a/controller/store.py +++ b/controller/store.py @@ -962,6 +962,31 @@ class ControllerStore: (slot,), ).fetchone() is not None + def find_occupied_slot_for_url(self, url: str) -> int | None: + """Return the slot already holding ``url``, if any. + + A non-``None`` result means ingest for this exact URL would reuse or + refresh a slot that already holds media, rather than land in a slot + that is genuinely empty. Callers use this to decide whether an + ingest request needs fresh Player evidence before proceeding. + """ + + if not isinstance(url, str): + return None + url = url.strip() + if not url: + return None + with self._lock: + row = self.connection.execute( + """ + SELECT slot_number FROM slots + WHERE url = ? AND media_state <> 'empty' + ORDER BY slot_number LIMIT 1 + """, + (url,), + ).fetchone() + return None if row is None else int(row["slot_number"]) + def mark_download_running(self, job_id: str, slot: int, generation: int) -> bool: with self._transaction() as connection: now = self._now() diff --git a/tests/test_service.py b/tests/test_service.py index 462f7f3..c6ccd72 100644 --- a/tests/test_service.py +++ b/tests/test_service.py @@ -88,6 +88,11 @@ class FakeDownloadManager: self.expiry_calls = 0 self.stopped = False self.delete_requests = [] + self.enqueue_calls = [] + + def enqueue(self, url, submitter, note): + self.enqueue_calls.append((url, submitter, note)) + return self.store.reserve_download(url, submitter, note) def recover_pending_deletions(self): self.recovered = True @@ -492,22 +497,24 @@ class ControllerServiceTests(unittest.TestCase): self.assertIn("mqtt_disconnected", health["degraded_reasons"]) self.assertIn("awaiting_player_evidence", health["degraded_reasons"]) - def test_cleanup_and_downloads_wait_for_post_start_player_evidence(self): + def test_cleanup_waits_for_post_start_player_evidence_but_new_ingest_does_not(self): self.start_service() self.service._next_expiry = self.monotonic() time.sleep(0.2) self.assertEqual(self.downloads.expiry_calls, 0) self.assertFalse(self.downloads.recovered) - with self.assertRaisesRegex( - ControllerUnavailableError, - "deferred until fresh Player evidence", - ): - self.service.enqueue_download( - "https://example.test/deferred", - None, - None, - ) + + # A genuinely new playlist cannot mutate, delete, or replace any + # occupied slot, so it is not blocked by the missing evidence that + # still defers cleanup/expiry below. + result = self.service.enqueue_download( + "https://example.test/brand-new", + None, + None, + ) + self.assertEqual(result["status"], "queued") + self.assertFalse(result["reused_slot"]) self.send_message( self.config.topics.player_state, @@ -540,6 +547,106 @@ class ControllerServiceTests(unittest.TestCase): self.service.health()["degraded_reasons"], ) + def test_ingest_matching_an_occupied_slot_url_remains_gated_without_evidence(self): + self.start_service() + self.ready_slot(1) + + with self.assertRaisesRegex( + ControllerUnavailableError, + "fresh Player evidence", + ): + self.service.enqueue_download("https://example.test/1", None, None) + + def test_controller_only_restart_with_idle_connected_player_allows_new_ingest(self): + """Regression for the too-broad evidence gate deadlock. + + A Controller-only restart with the Player still connected and idle + never produces fresh detailed evidence, because the Player has no + reason to reconnect or emit unsolicited status. New ingest into an + empty slot must still work; refresh/reuse of already-occupied media, + expiry, cleanup recovery, and manual deletion must all stay gated + until fresh evidence or an explicit administrative override arrives. + """ + + self.start_service() + self.ready_slot(1) + self.send_message( + self.config.topics.player_status, + {"state": "idle", "message": "player online"}, + qos=1, + ) + self.wait_for( + lambda: "awaiting_player_evidence" + not in self.service.health()["degraded_reasons"], + "evidence did not clear before the simulated restart", + ) + self.service.stop() + + # Simulate a Controller-only restart against the same durable store: + # the Player process itself never restarted, stays connected and + # idle, and therefore never sends another fresh status. + self.events = queue.Queue() + restarted_store = ControllerStore(self.config.storage, now=self.wall) + restarted_mqtt = FakeMqttAdapter() + restarted_downloads = FakeDownloadManager(restarted_store) + restarted_service = ControllerService( + self.config, + store=restarted_store, + mqtt_adapter=restarted_mqtt, + event_queue=self.events, + download_manager=restarted_downloads, + monotonic=self.monotonic, + wall_clock=self.wall, + ) + restarted_service.start() + self.addCleanup( + lambda: restarted_service.stop() if restarted_service.started else None + ) + self.assertIn( + "awaiting_player_evidence", + restarted_service.health()["degraded_reasons"], + ) + + # A genuinely new playlist still ingests safely with no evidence. + result = restarted_service.enqueue_download( + "https://example.test/after-restart", None, None + ) + self.assertEqual(result["status"], "queued") + self.assertFalse(result["reused_slot"]) + self.assertNotEqual(result["slot"], 1) + + # Refreshing the URL already occupying slot 1 remains gated. + with self.assertRaisesRegex( + ControllerUnavailableError, + "fresh Player evidence", + ): + restarted_service.enqueue_download("https://example.test/1", None, None) + + # Expiry and cleanup recovery remain deferred. + restarted_service._next_expiry = self.monotonic() + time.sleep(0.2) + self.assertEqual(restarted_downloads.expiry_calls, 0) + self.assertFalse(restarted_downloads.recovered) + + # Manual deletion of occupied slot 1 still requires the explicit + # confirm_player_unknown override; it is not silently unlocked. + with self.assertRaisesRegex( + SlotProtectedError, + "confirm_player_unknown=true", + ): + restarted_service.submit_control({"action": "delete_slot", "slot": 1}) + + # Fresh Player evidence still unlocks cleanup normally afterward. + self.send_message( + self.config.topics.player_status, + {"state": "idle", "message": "player online"}, + qos=1, + ) + self.wait_for( + lambda: restarted_downloads.recovered, + "fresh evidence did not unlock deferred recovery after restart", + ) + def test_administrator_can_explicitly_unlock_cleanup_without_player_evidence(self): self.start_service() review = self.service.health()["cleanup_reconciliation"] diff --git a/tests/test_store.py b/tests/test_store.py index b491700..61a9c2b 100644 --- a/tests/test_store.py +++ b/tests/test_store.py @@ -98,6 +98,20 @@ class FreshStoreTests(StoreTestCase): self.assertTrue(store.mark_download_running("job-one", 1, 1)) self.assertEqual(store.get_slot(1).media_state, "downloading") + def test_find_occupied_slot_for_url_only_matches_non_empty_slots(self): + with self.open_store() as store: + url = "https://www.youtube.com/playlist?list=one" + self.assertIsNone(store.find_occupied_slot_for_url(url)) + self.assertIsNone(store.find_occupied_slot_for_url(" ")) + self.assertIsNone(store.find_occupied_slot_for_url(None)) + + store.reserve_download(url, None, None, job_id="job-one") + self.assertEqual(store.find_occupied_slot_for_url(url), 1) + self.assertEqual(store.find_occupied_slot_for_url(f" {url} "), 1) + self.assertIsNone( + store.find_occupied_slot_for_url("https://example.test/other") + ) + def test_concurrent_requests_get_only_one_remaining_slot(self): with self.open_store() as store: for index in range(1, 10):