From b934b69e5a42c26ad48830cfd260b8bffbb96409 Mon Sep 17 00:00:00 2001 From: Media Date: Wed, 12 Aug 2026 18:21:29 -0500 Subject: [PATCH] feat: persist isolated utility export jobs --- config.yaml | 5 + controller/config.py | 30 +++++ controller/models.py | 22 ++++ controller/store.py | 220 +++++++++++++++++++++++++++++++-- tests/test_config_contracts.py | 31 +++++ tests/test_store.py | 129 ++++++++++++++++++- 6 files changed, 423 insertions(+), 14 deletions(-) diff --git a/config.yaml b/config.yaml index e1e704f..db24f49 100644 --- a/config.yaml +++ b/config.yaml @@ -48,6 +48,11 @@ downloads: max_workers: 2 process_timeout_seconds: 7200 max_height: 1080 + # Utility exports live below media_base and never participate in slots. + export_directory: exports + export_max_height: 1080 + # Bounds queued + running utility work; Player slots retain their own bound. + max_queued_exports: 20 router: required_mode: video diff --git a/controller/config.py b/controller/config.py index c03bbc7..6c3257c 100644 --- a/controller/config.py +++ b/controller/config.py @@ -63,6 +63,9 @@ class DownloadConfig: max_workers: int = 2 process_timeout_seconds: float = 2 * 3600.0 max_height: int = 1080 + export_directory: Path = Path("exports") + export_max_height: int = 1080 + max_queued_exports: int = 20 @dataclass(frozen=True) @@ -172,6 +175,20 @@ def _optional_path(value: Any, name: str) -> Path | None: return Path(text) +def _relative_directory(value: Any, name: str) -> Path: + text = _text(value, name) + assert text is not None + path = Path(text) + if ( + path.is_absolute() + or not path.parts + or any(part in {".", ".."} for part in path.parts) + or "\\" in text + ): + raise ConfigError(f"{name} must be a safe relative directory") + return path + + def _env_value(env: Mapping[str, str], key: str, fallback: Any) -> Any: return env[key] if key in env else fallback @@ -277,6 +294,19 @@ def load_config( max_workers=_integer(downloads_raw.get("max_workers", 2), "downloads.max_workers", maximum=10), process_timeout_seconds=_number(downloads_raw.get("process_timeout_seconds", 2 * 3600), "downloads.process_timeout_seconds", minimum=1.0), max_height=_integer(downloads_raw.get("max_height", 1080), "downloads.max_height"), + export_directory=_relative_directory( + downloads_raw.get("export_directory", "exports"), + "downloads.export_directory", + ), + export_max_height=_integer( + downloads_raw.get("export_max_height", 1080), + "downloads.export_max_height", + ), + max_queued_exports=_integer( + downloads_raw.get("max_queued_exports", 20), + "downloads.max_queued_exports", + maximum=1000, + ), ) if downloads.cookie_required and downloads.cookie_path is None: raise ConfigError("downloads.cookie_path is required when cookie_required is true") diff --git a/controller/models.py b/controller/models.py index a174b07..7883128 100644 --- a/controller/models.py +++ b/controller/models.py @@ -10,6 +10,7 @@ MEDIA_STATES = frozenset( ) LIVE_JOB_STATES = frozenset({"reserved", "running"}) LIVE_GUARD_STATES = frozenset({"active", "review_required"}) +UTILITY_EXPORT_MODES = frozenset({"mp4", "m4a", "mp3"}) @dataclass(frozen=True) @@ -54,6 +55,27 @@ class DownloadJob: error: str | None +@dataclass(frozen=True) +class ExportReservation: + status: str + job_id: str | None = None + + +@dataclass(frozen=True) +class ExportJob: + job_id: str + mode: str + url: str + status: str + submitter: str | None + note: str | None + requested_at: float + started_at: float | None + finished_at: float | None + output_paths: tuple[str, ...] + error: str | None + + @dataclass(frozen=True) class DeleteToken: slot: int diff --git a/controller/store.py b/controller/store.py index 888ef6f..f3701af 100644 --- a/controller/store.py +++ b/controller/store.py @@ -25,6 +25,8 @@ from .models import ( DeleteToken, DownloadJob, DownloadReservation, + ExportJob, + ExportReservation, MigrationError, MigrationReport, PlayerObservation, @@ -33,10 +35,11 @@ from .models import ( SlotProtectedError, SlotRecord, StaleOperationError, + UTILITY_EXPORT_MODES, ) -SCHEMA_VERSION = 2 +SCHEMA_VERSION = 3 LEGACY_MIGRATION_NAME = "legacy_playlists_yaml_v1" MAX_LEGACY_YAML_BYTES = 8 * 1024 * 1024 @@ -104,6 +107,24 @@ CREATE UNIQUE INDEX IF NOT EXISTS one_live_download_per_slot ON download_jobs(slot_number) WHERE status IN ('reserved','running'); +CREATE TABLE IF NOT EXISTS export_jobs ( + job_id TEXT PRIMARY KEY, + mode TEXT NOT NULL CHECK (mode IN ('mp4','m4a','mp3')), + url TEXT NOT NULL, + status TEXT NOT NULL CHECK (status IN + ('queued','running','completed','failed','interrupted')), + submitter TEXT, + note TEXT, + requested_at REAL NOT NULL, + started_at REAL, + finished_at REAL, + output_paths_json TEXT NOT NULL DEFAULT '[]', + error TEXT +); + +CREATE INDEX IF NOT EXISTS export_jobs_recent +ON export_jobs(requested_at DESC, job_id DESC); + CREATE TABLE IF NOT EXISTS slot_guards ( guard_id TEXT PRIMARY KEY, slot_number INTEGER NOT NULL REFERENCES slots(slot_number), @@ -401,6 +422,13 @@ class ControllerStore: """, (now,), ) + connection.execute( + """ + INSERT OR IGNORE INTO schema_migrations(version, name, applied_at) + VALUES (3, 'utility_export_jobs', ?) + """, + (now,), + ) connection.execute(f"PRAGMA user_version = {SCHEMA_VERSION}") def _upgrade_database(self, connection: sqlite3.Connection) -> None: @@ -409,26 +437,58 @@ class ControllerStore: version = connection.execute("PRAGMA user_version").fetchone()[0] if version == SCHEMA_VERSION: return - if version != 1: + if version not in {1, 2}: raise MigrationError( - f"unsupported Controller database schema {version}; expected 1 or {SCHEMA_VERSION}" + f"unsupported Controller database schema {version}; expected 1, 2, or {SCHEMA_VERSION}" ) connection.execute("BEGIN IMMEDIATE") try: - columns = { - row["name"] - for row in connection.execute( - "PRAGMA table_info(player_observation)" - ).fetchall() - } - if "reconciliation_id" not in columns: + if version < 2: + columns = { + row["name"] + for row in connection.execute( + "PRAGMA table_info(player_observation)" + ).fetchall() + } + if "reconciliation_id" not in columns: + connection.execute( + "ALTER TABLE player_observation ADD COLUMN reconciliation_id TEXT" + ) connection.execute( - "ALTER TABLE player_observation ADD COLUMN reconciliation_id TEXT" + """ + INSERT OR IGNORE INTO schema_migrations(version, name, applied_at) + VALUES (2, 'external_playback_reconciliation', ?) + """, + (self._now(),), ) + connection.execute( + """ + CREATE TABLE IF NOT EXISTS export_jobs ( + job_id TEXT PRIMARY KEY, + mode TEXT NOT NULL CHECK (mode IN ('mp4','m4a','mp3')), + url TEXT NOT NULL, + status TEXT NOT NULL CHECK (status IN + ('queued','running','completed','failed','interrupted')), + submitter TEXT, + note TEXT, + requested_at REAL NOT NULL, + started_at REAL, + finished_at REAL, + output_paths_json TEXT NOT NULL DEFAULT '[]', + error TEXT + ) + """ + ) + connection.execute( + """ + CREATE INDEX IF NOT EXISTS export_jobs_recent + ON export_jobs(requested_at DESC, job_id DESC) + """ + ) connection.execute( """ INSERT OR IGNORE INTO schema_migrations(version, name, applied_at) - VALUES (2, 'external_playback_reconciliation', ?) + VALUES (3, 'utility_export_jobs', ?) """, (self._now(),), ) @@ -837,6 +897,15 @@ class ControllerStore: row["generation"], ), ) + connection.execute( + """ + UPDATE export_jobs + SET status = 'interrupted', finished_at = ?, + error = 'export interrupted by Controller restart' + WHERE status IN ('queued','running') + """, + (now,), + ) def list_slots(self) -> list[SlotRecord]: with self._lock: @@ -1136,6 +1205,133 @@ class ControllerStore: error=row["error"], ) + @staticmethod + def _row_to_export_job(row: sqlite3.Row) -> ExportJob: + try: + raw_paths = json.loads(row["output_paths_json"]) + except (TypeError, ValueError, json.JSONDecodeError): + raw_paths = [] + output_paths = tuple( + value for value in raw_paths if isinstance(value, str) + ) if isinstance(raw_paths, list) else () + return ExportJob( + job_id=row["job_id"], + mode=row["mode"], + url=row["url"], + status=row["status"], + submitter=row["submitter"], + note=row["note"], + requested_at=row["requested_at"], + started_at=row["started_at"], + finished_at=row["finished_at"], + output_paths=output_paths, + error=row["error"], + ) + + def reserve_export_job( + self, + url: str, + mode: str, + submitter: str | None, + note: str | None, + *, + max_active: int, + job_id: str | None = None, + ) -> ExportReservation: + if not isinstance(url, str) or not url.strip(): + raise ValueError("export URL must be non-empty") + if mode not in UTILITY_EXPORT_MODES: + raise ValueError("unsupported utility export mode") + if not isinstance(max_active, int) or isinstance(max_active, bool) or max_active < 1: + raise ValueError("max_active must be a positive integer") + job_id = job_id or uuid.uuid4().hex + with self._transaction() as connection: + active = connection.execute( + """ + SELECT COUNT(*) FROM export_jobs + WHERE status IN ('queued','running') + """ + ).fetchone()[0] + if active >= max_active: + return ExportReservation(status="queue_full") + connection.execute( + """ + INSERT INTO export_jobs + (job_id, mode, url, status, submitter, note, requested_at) + VALUES (?, ?, ?, 'queued', ?, ?, ?) + """, + (job_id, mode, url.strip(), submitter, note, self._now()), + ) + return ExportReservation(status="queued", job_id=job_id) + + def mark_export_running(self, job_id: str) -> bool: + with self._transaction() as connection: + return bool( + connection.execute( + """ + UPDATE export_jobs SET status = 'running', started_at = ? + WHERE job_id = ? AND status = 'queued' + """, + (self._now(), job_id), + ).rowcount + ) + + def complete_export_job( + self, + job_id: str, + output_paths: Sequence[str], + ) -> bool: + normalized = tuple(output_paths) + if not normalized or any( + not isinstance(path, str) or not path for path in normalized + ): + raise ValueError("completed export must contain output paths") + with self._transaction() as connection: + return bool( + connection.execute( + """ + UPDATE export_jobs + SET status = 'completed', finished_at = ?, + output_paths_json = ?, error = NULL + WHERE job_id = ? AND status = 'running' + """, + (self._now(), json.dumps(normalized), job_id), + ).rowcount + ) + + def fail_export_job(self, job_id: str, error: str) -> bool: + with self._transaction() as connection: + return bool( + connection.execute( + """ + UPDATE export_jobs + SET status = 'failed', finished_at = ?, error = ? + WHERE job_id = ? AND status IN ('queued','running') + """, + (self._now(), error, job_id), + ).rowcount + ) + + def get_export_job(self, job_id: str) -> ExportJob | None: + with self._lock: + row = self.connection.execute( + "SELECT * FROM export_jobs WHERE job_id = ?", (job_id,) + ).fetchone() + return None if row is None else self._row_to_export_job(row) + + def list_export_jobs(self, *, limit: int = 20) -> list[ExportJob]: + if not isinstance(limit, int) or isinstance(limit, bool) or not 1 <= limit <= 100: + raise ValueError("export job limit must be between 1 and 100") + with self._lock: + rows = self.connection.execute( + """ + SELECT * FROM export_jobs + ORDER BY requested_at DESC, job_id DESC LIMIT ? + """, + (limit,), + ).fetchall() + return [self._row_to_export_job(row) for row in rows] + def protect_slot( self, slot: int, diff --git a/tests/test_config_contracts.py b/tests/test_config_contracts.py index fcaf2ba..50d2100 100644 --- a/tests/test_config_contracts.py +++ b/tests/test_config_contracts.py @@ -82,6 +82,9 @@ class ConfigurationTests(unittest.TestCase): self.assertEqual(config.storage.uncertain_review_after_seconds, 900.0) self.assertEqual(config.mqtt.publish_ack_timeout_seconds, 2.0) self.assertEqual(config.mqtt.router_echo_grace_seconds, 2.0) + self.assertEqual(config.downloads.export_directory, Path("exports")) + self.assertEqual(config.downloads.export_max_height, 1080) + self.assertEqual(config.downloads.max_queued_exports, 20) def test_documented_environment_overrides_apply(self): env = { @@ -175,6 +178,34 @@ class ConfigurationTests(unittest.TestCase): with self.assertRaisesRegex(ConfigError, f"{field} must be an absolute path"): load_config(path, environ={}) + def test_export_directory_must_stay_relative_to_media_root(self): + invalid = ("/tmp/exports", "../exports", "nested/../exports", r"nested\\exports") + for value in invalid: + with self.subTest(value=value), tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "config.yaml" + path.write_text( + f"downloads:\n export_directory: {value!r}\n", + encoding="utf-8", + ) + with self.assertRaisesRegex(ConfigError, "safe relative directory"): + load_config(path, environ={}) + + def test_export_limits_are_bounded_positive_integers(self): + cases = ( + ("export_max_height", 0), + ("max_queued_exports", 0), + ("max_queued_exports", 1001), + ) + for field, value in cases: + with self.subTest(field=field), tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "config.yaml" + path.write_text( + f"downloads:\n {field}: {value}\n", + encoding="utf-8", + ) + with self.assertRaisesRegex(ConfigError, field): + load_config(path, environ={}) + class ImportLifecycleTests(unittest.TestCase): def test_controller_core_import_does_not_connect_or_create_runtime_paths(self): diff --git a/tests/test_store.py b/tests/test_store.py index 61a9c2b..bddb384 100644 --- a/tests/test_store.py +++ b/tests/test_store.py @@ -50,7 +50,7 @@ class FreshStoreTests(StoreTestCase): self.assertTrue(all(slot.media_state == "empty" for slot in slots)) self.assertEqual(store.player_observation().phase, "unknown") - def test_schema_one_database_upgrades_external_reconciliation_atomically(self): + def test_schema_one_database_upgrades_reconciliation_and_exports_atomically(self): with self.open_store() as store: database_path = store.path connection = sqlite3.connect(database_path) @@ -67,7 +67,7 @@ class FreshStoreTests(StoreTestCase): with self.open_store() as upgraded: self.assertEqual( upgraded.connection.execute("PRAGMA user_version").fetchone()[0], - 2, + 3, ) columns = { row["name"] @@ -80,6 +80,131 @@ class FreshStoreTests(StoreTestCase): "SELECT name FROM schema_migrations WHERE version = 2" ).fetchone() self.assertEqual(migration["name"], "external_playback_reconciliation") + export_migration = upgraded.connection.execute( + "SELECT name FROM schema_migrations WHERE version = 3" + ).fetchone() + self.assertEqual(export_migration["name"], "utility_export_jobs") + tables = { + row["name"] + for row in upgraded.connection.execute( + "SELECT name FROM sqlite_master WHERE type = 'table'" + ).fetchall() + } + self.assertIn("export_jobs", tables) + + def test_schema_two_database_adds_only_utility_export_jobs(self): + with self.open_store() as store: + database_path = store.path + connection = sqlite3.connect(database_path) + try: + connection.execute("DROP TABLE export_jobs") + connection.execute("DELETE FROM schema_migrations WHERE version = 3") + connection.execute("PRAGMA user_version = 2") + connection.commit() + finally: + connection.close() + + with self.open_store() as upgraded: + self.assertEqual( + upgraded.connection.execute("PRAGMA user_version").fetchone()[0], + 3, + ) + self.assertEqual(upgraded.list_export_jobs(), []) + + def test_export_jobs_are_slot_free_bounded_and_durable(self): + with self.open_store() as store: + first = store.reserve_export_job( + "https://www.youtube.com/watch?v=one", + "mp4", + "Alice", + "Sunday", + max_active=1, + job_id="export-one", + ) + self.assertEqual(first.status, "queued") + self.assertEqual(first.job_id, "export-one") + self.assertEqual( + store.reserve_export_job( + "https://www.youtube.com/watch?v=two", + "mp3", + None, + None, + max_active=1, + ).status, + "queue_full", + ) + self.assertTrue(store.mark_export_running("export-one")) + self.assertTrue( + store.complete_export_job( + "export-one", + ("exports/video/export-one/Service.mp4",), + ) + ) + job = store.get_export_job("export-one") + self.assertEqual(job.mode, "mp4") + self.assertEqual(job.status, "completed") + self.assertEqual(job.submitter, "Alice") + self.assertEqual( + job.output_paths, + ("exports/video/export-one/Service.mp4",), + ) + self.assertTrue(all(slot.media_state == "empty" for slot in store.list_slots())) + + def test_restart_marks_live_exports_interrupted_without_touching_slots(self): + store = self.open_store() + store.reserve_export_job( + "https://www.youtube.com/playlist?list=one", + "m4a", + None, + None, + max_active=2, + job_id="queued-export", + ) + store.reserve_export_job( + "https://www.youtube.com/watch?v=two", + "mp3", + None, + None, + max_active=2, + job_id="running-export", + ) + store.mark_export_running("running-export") + store.close() + + with self.open_store() as reopened: + jobs = {job.job_id: job for job in reopened.list_export_jobs()} + self.assertEqual(jobs["queued-export"].status, "interrupted") + self.assertEqual(jobs["running-export"].status, "interrupted") + self.assertTrue( + all("Controller restart" in job.error for job in jobs.values()) + ) + self.assertTrue( + all(slot.media_state == "empty" for slot in reopened.list_slots()) + ) + + def test_export_failures_do_not_corrupt_player_inventory(self): + with self.open_store() as store: + slot = store.reserve_download( + "https://www.youtube.com/watch?v=managed", + None, + None, + job_id="managed", + ) + store.fail_download( + "managed", slot.slot, slot.generation, "managed failure" + ) + before = store.get_slot(slot.slot) + export = store.reserve_export_job( + "https://www.youtube.com/watch?v=export", + "mp3", + None, + None, + max_active=5, + job_id="failed-export", + ) + self.assertEqual(export.status, "queued") + self.assertTrue(store.fail_export_job("failed-export", "ffmpeg failed")) + self.assertEqual(store.get_slot(slot.slot), before) def test_slot_reservation_is_committed_before_worker_transition(self): with self.open_store() as store: