diff --git a/README.md b/README.md index 9f9c7b6..2fa3091 100644 --- a/README.md +++ b/README.md @@ -454,10 +454,11 @@ collision suffixes. Each job publishes atomically into its own directory: ``` Work is staged under `exports/.work/`. A fsynced manifest makes the -publication boundary recoverable: after restart, Controller verifies and +publication boundary recoverable: after restart, a dedicated export worker verifies and finishes a fully prepared or already-published job, while a known partial job is marked interrupted and its private staging directory is removed. It does -not resume a partial yt-dlp process. Unknown directories are left untouched. +not delay web readiness, MQTT reconnect, or Player coordination and does not +resume a partial yt-dlp process. Unknown directories are left untouched. The web page lists live jobs before recent completed/failed/interrupted jobs and shows their output paths or error. It intentionally provides no file-serving, cancellation, or deletion endpoint. diff --git a/controller/config.py b/controller/config.py index f42cb85..93c8388 100644 --- a/controller/config.py +++ b/controller/config.py @@ -314,7 +314,10 @@ def load_config( max_queued_exports=_integer( downloads_raw.get("max_queued_exports", 20), "downloads.max_queued_exports", - maximum=1000, + # The web status endpoint deliberately returns at most 100 rows. + # Keep the queue bound within that view so every active job remains + # visible even when the configured queue is full. + maximum=100, ), ) if downloads.cookie_required and downloads.cookie_path is None: diff --git a/controller/downloads.py b/controller/downloads.py index 79170e6..cd20f4b 100644 --- a/controller/downloads.py +++ b/controller/downloads.py @@ -305,13 +305,29 @@ class DownloadManager: future.add_done_callback(self._forget_future) return reservation + def schedule_export_reconciliation(self) -> bool: + """Queue crash-artifact recovery without delaying core service startup.""" + + with self._lock: + if self._stopping: + return False + try: + future = self._export_executor.submit(self.reconcile_export_artifacts) + except BaseException: + LOG.exception("could not schedule utility export reconciliation") + return False + with self._lock: + self._futures.add(future) + future.add_done_callback(self._forget_future) + return True + def _forget_future(self, future: Future[Any]) -> None: with self._lock: self._futures.discard(future) try: future.result() except BaseException: - LOG.exception("download worker escaped its failure boundary") + LOG.exception("media worker escaped its failure boundary") def _emit(self, event: DownloadEvent) -> None: try: @@ -839,7 +855,14 @@ class DownloadManager: and source_info.pixel_format in {"yuv420p", "yuvj420p"} ) audio_compatible = source_info.audio_codec == "aac" - container_compatible = "mp4" in source_info.format_names + # ffprobe reports the same comma-separated demuxer aliases for MOV and + # MP4. A .mov source therefore needs at least a stream-copy remux; it + # must never be renamed to .mp4 solely because the alias list includes + # "mp4". + container_compatible = ( + source.suffix.lower() == ".mp4" + and "mp4" in source_info.format_names + ) if video_compatible and audio_compatible and container_compatible: os.replace(source, output) else: diff --git a/controller/service.py b/controller/service.py index 7ccd8ae..a633e59 100644 --- a/controller/service.py +++ b/controller/service.py @@ -148,12 +148,6 @@ class ControllerService: self.store, on_event=self._queue_download_event, ) - try: - self.downloads.reconcile_export_artifacts() - except BaseException: - LOG.exception( - "utility export reconciliation failed; Controller startup continues" - ) # Cleanup is deliberately deferred until fresh Player evidence # (or an explicit administrative reconciliation) makes it safe. self._reset_cleanup_evidence() @@ -171,6 +165,11 @@ class ControllerService: self._coordinator.start() self.mqtt.start() self._started = True + # A crash-published playlist may contain many files to probe. + # Queue that bounded work only after core readiness and keep it + # on the dedicated export executor, so MQTT reconnect and + # Player coordination never wait for recovery to finish. + self.downloads.schedule_export_reconciliation() except BaseException: self._stop_event.set() try: diff --git a/tests/test_config_contracts.py b/tests/test_config_contracts.py index f9999c7..b9966cf 100644 --- a/tests/test_config_contracts.py +++ b/tests/test_config_contracts.py @@ -209,7 +209,7 @@ class ConfigurationTests(unittest.TestCase): ("export_max_workers", 0), ("export_max_workers", 11), ("max_queued_exports", 0), - ("max_queued_exports", 1001), + ("max_queued_exports", 101), ) for field, value in cases: with self.subTest(field=field), tempfile.TemporaryDirectory() as directory: diff --git a/tests/test_exports.py b/tests/test_exports.py index 54974a0..8306f51 100644 --- a/tests/test_exports.py +++ b/tests/test_exports.py @@ -73,6 +73,7 @@ class FakeExportRunner: "mp4": ".mp4", "matroska": ".mkv", "webm": ".webm", + "mov": ".mov", }[self.container] for index in range(1, self.playlist_items + 1): (template.parent / f"{index:03d} - Tést Item {index} [id{index}]{extension}").write_bytes( @@ -112,7 +113,7 @@ class FakeExportRunner: ], "format": { "format_name": "mov,mp4,m4a,3gp,3g2,mj2" - if converted or self.container == "mp4" + if converted or self.container in {"mp4", "mov"} else self.container }, } @@ -240,6 +241,35 @@ class UtilityExportTests(unittest.TestCase): self.assertEqual(arguments[arguments.index("-c:a") + 1], "copy") self.assertNotIn("libx264", arguments) + def test_quicktime_alias_is_remuxed_instead_of_renamed_to_mp4(self): + runner = FakeExportRunner(container="mov") + result = self.manager(runner).enqueue_export( + "https://www.youtube.com/watch?v=mov", "mp4", None, None + ) + + self.assertEqual(self.store.get_export_job(result.job_id).status, "completed") + arguments = runner.ffmpeg_calls[0] + self.assertEqual(arguments[arguments.index("-c:v") + 1], "copy") + self.assertEqual(arguments[arguments.index("-c:a") + 1], "copy") + self.assertNotIn("libx264", arguments) + + def test_export_reconciliation_is_submitted_to_dedicated_executor(self): + export_executor = QueuedExecutor() + manager = DownloadManager( + self.config, + self.store, + runner=FakeExportRunner(), + executor=ImmediateExecutor(), + export_executor=export_executor, + ) + + self.assertTrue(manager.schedule_export_reconciliation()) + self.assertEqual(len(export_executor.calls), 1) + _, function, arguments, keyword_arguments = export_executor.calls[0] + self.assertEqual(function, manager.reconcile_export_artifacts) + self.assertEqual(arguments, ()) + self.assertEqual(keyword_arguments, {}) + def test_lower_resolution_mp4_is_preserved_without_scaling(self): runner = FakeExportRunner(height=720) result = self.manager(runner).enqueue_export( diff --git a/tests/test_service.py b/tests/test_service.py index 1f42862..1a77177 100644 --- a/tests/test_service.py +++ b/tests/test_service.py @@ -106,8 +106,9 @@ class FakeDownloadManager: max_active=20, ) - def reconcile_export_artifacts(self): + def schedule_export_reconciliation(self): self.export_reconciled = True + return True def recover_pending_deletions(self): self.recovered = True