From 5d88d4d0ff6b43e4dc59e2ba2f5d444a679cf38d Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Mon, 6 Jul 2026 13:27:05 +0900 Subject: [PATCH] feat: durable SQLite-backed job store for the async web job API Replace the per-process in-memory _JOBS dict with job_store.JobStore, a SQLite (WAL) store shared by every worker process on the host: - job state survives service restarts and crashes; polling clients of interrupted jobs get an honest 'failed: Interrupted by service restart' instead of a permanent 404/hang - status/result requests work under multi-worker uvicorn/gunicorn, since all workers now read the same store - startup maintenance reconciles the store with reality: orphaned queued/processing jobs are failed and their temp workspaces reclaimed; terminal jobs past the retention window are purged together with their workspaces (no more disk leaks from results nobody downloaded) - CODEC_CARVER_JOB_DB points the store at a persistent volume; CODEC_CARVER_JOB_RETENTION_SECONDS tunes the sweep window - field-whitelisted SQL, one short-lived connection per operation, delete-returning semantics so workspace cleanup happens exactly once under concurrent deletes Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01CJRVbDrp1vGYkJgNHMGPpG --- job_store.py | 208 ++++++++++++++++++++++++++++++++++++++++ saas_web.py | 72 +++++++++++--- tests/test_job_store.py | 175 +++++++++++++++++++++++++++++++++ tests/test_saas_web.py | 118 +++++++++++++++++------ 4 files changed, 533 insertions(+), 40 deletions(-) create mode 100644 job_store.py create mode 100644 tests/test_job_store.py diff --git a/job_store.py b/job_store.py new file mode 100644 index 0000000..94796a0 --- /dev/null +++ b/job_store.py @@ -0,0 +1,208 @@ +"""SQLite-backed durable job store for the async web job API. + +The web service's job endpoints (``POST /jobs`` -> ``GET /jobs/{id}`` -> +``GET /jobs/{id}/result``) need to remember job state between requests. +Keeping that state in a per-process dict loses every job on restart and +breaks status polling as soon as the app runs with more than one worker +process. This module persists jobs in a small SQLite database instead: + +* Job state survives process restarts and crashes. +* Multiple worker processes on the same host share one consistent view + (SQLite WAL mode allows concurrent readers alongside a single writer). +* Jobs orphaned by a dead process can be detected and failed explicitly + (`recover_interrupted`), and old terminal jobs can be swept together + with their on-disk workspaces (`purge_stale`). + +The database location defaults to a per-host path under the system temp +directory; production deployments should point ``CODEC_CARVER_JOB_DB`` at +a persistent volume. +""" + +import contextlib +import os +import sqlite3 +import tempfile +import time +from pathlib import Path + +#: Environment variable that overrides the job database location. +DB_PATH_ENV = "CODEC_CARVER_JOB_DB" + +#: Job states considered "in flight". Jobs found in these states at process +#: startup were orphaned by a previous process, because background workers do +#: not survive a restart. +ACTIVE_STATUSES = ("queued", "processing") + +#: Job states that will never change again. +TERMINAL_STATUSES = ("done", "failed") + +#: Default retention for terminal jobs before `purge_stale` reaps them. +DEFAULT_RETENTION_SECONDS = 24 * 60 * 60 + +#: Columns callers may set through `create` / `update`. Everything else is +#: managed by the store itself. +_MUTABLE_FIELDS = frozenset( + {"status", "temp_dir", "output_path", "output_name", "error"} +) + +_SCHEMA = """ +CREATE TABLE IF NOT EXISTS jobs ( + job_id TEXT PRIMARY KEY, + status TEXT NOT NULL, + temp_dir TEXT, + output_path TEXT, + output_name TEXT, + error TEXT, + created_at REAL NOT NULL, + updated_at REAL NOT NULL +) +""" + + +def default_db_path() -> Path: + """Return the job database path, honouring the ``CODEC_CARVER_JOB_DB`` env var. + + The fallback lives under the system temp directory so the service works + with zero configuration and all worker processes on a host agree on the + location. Deployments that must survive host reboots should set the env + var to a path on a persistent volume. + """ + configured = os.environ.get(DB_PATH_ENV) + if configured: + return Path(configured) + return Path(tempfile.gettempdir()) / "codec-carver" / "jobs.db" + + +class JobStore: + """Durable job-state store backed by a single SQLite database file. + + Every operation opens a short-lived connection, so one instance may be + shared freely across threads, and separate processes pointed at the same + path see a single consistent store. + """ + + def __init__(self, db_path: Path | str): + """Create the store, its parent directory, and the schema if missing.""" + self.db_path = Path(db_path) + self.db_path.parent.mkdir(parents=True, exist_ok=True) + with self._connect() as conn: + conn.execute(_SCHEMA) + + @contextlib.contextmanager + def _connect(self): + """Yield a short-lived WAL connection wrapped in one transaction. + + The connection is committed on success, rolled back on error, and + always closed, so no file handles outlive the operation. + """ + conn = sqlite3.connect(self.db_path, timeout=30) + try: + conn.row_factory = sqlite3.Row + conn.execute("PRAGMA journal_mode=WAL") + conn.execute("PRAGMA synchronous=NORMAL") + with conn: + yield conn + finally: + conn.close() + + def create(self, job_id: str, **fields) -> None: + """Insert a new job row. Unknown fields raise ``ValueError``.""" + self._check_fields(fields) + fields.setdefault("status", "queued") + now = time.time() + columns = ["job_id", "created_at", "updated_at", *fields] + placeholders = ", ".join("?" for _ in columns) + values = [job_id, now, now, *fields.values()] + with self._connect() as conn: + conn.execute( + f"INSERT INTO jobs ({', '.join(columns)}) VALUES ({placeholders})", + values, + ) + + def get(self, job_id: str) -> dict | None: + """Return one job as a dict, or ``None`` if the id is unknown.""" + with self._connect() as conn: + row = conn.execute( + "SELECT * FROM jobs WHERE job_id = ?", (job_id,) + ).fetchone() + return dict(row) if row is not None else None + + def all_jobs(self) -> list[dict]: + """Return every job row, oldest first. Intended for ops and tests.""" + with self._connect() as conn: + rows = conn.execute( + "SELECT * FROM jobs ORDER BY created_at" + ).fetchall() + return [dict(row) for row in rows] + + def update(self, job_id: str, **fields) -> bool: + """Update mutable fields on a job; return ``False`` for unknown ids.""" + self._check_fields(fields) + if not fields: + return self.get(job_id) is not None + assignments = ", ".join(f"{name} = ?" for name in fields) + values = [*fields.values(), time.time(), job_id] + with self._connect() as conn: + cursor = conn.execute( + f"UPDATE jobs SET {assignments}, updated_at = ? WHERE job_id = ?", + values, + ) + return cursor.rowcount > 0 + + def delete(self, job_id: str) -> dict | None: + """Remove a job and return its final row, or ``None`` if unknown. + + Returning the row lets callers release resources the job owned + (its temp workspace) exactly once, even under concurrent deletes: + only one caller observes the row. + """ + with self._connect() as conn: + row = conn.execute( + "DELETE FROM jobs WHERE job_id = ? RETURNING *", (job_id,) + ).fetchone() + return dict(row) if row is not None else None + + def recover_interrupted(self) -> list[dict]: + """Fail every queued/processing job left behind by a dead process. + + Background workers die with the process that spawned them, so any job + still marked active when a fresh process starts will never progress. + Marking them ``failed`` turns silent hangs into an honest error for + polling clients. Returns the affected rows (post-update) so the caller + can clean up their temp workspaces. + """ + placeholders = ", ".join("?" for _ in ACTIVE_STATUSES) + with self._connect() as conn: + rows = conn.execute( + f"UPDATE jobs SET status = 'failed', " + f"error = 'Interrupted by service restart', updated_at = ? " + f"WHERE status IN ({placeholders}) RETURNING *", + (time.time(), *ACTIVE_STATUSES), + ).fetchall() + return [dict(row) for row in rows] + + def purge_stale( + self, max_age_seconds: float = DEFAULT_RETENTION_SECONDS + ) -> list[dict]: + """Delete terminal jobs untouched for ``max_age_seconds``. + + Finished jobs whose results are never downloaded would otherwise keep + their rows (and, for ``done`` jobs, their output workspaces) forever. + Returns the deleted rows so the caller can remove those workspaces. + """ + cutoff = time.time() - max_age_seconds + placeholders = ", ".join("?" for _ in TERMINAL_STATUSES) + with self._connect() as conn: + rows = conn.execute( + f"DELETE FROM jobs WHERE status IN ({placeholders}) " + f"AND updated_at < ? RETURNING *", + (*TERMINAL_STATUSES, cutoff), + ).fetchall() + return [dict(row) for row in rows] + + @staticmethod + def _check_fields(fields: dict) -> None: + """Reject field names outside the mutable-column whitelist.""" + unknown = set(fields) - _MUTABLE_FIELDS + if unknown: + raise ValueError(f"Unknown job fields: {sorted(unknown)}") diff --git a/saas_web.py b/saas_web.py index d1a8e44..0e4c6a7 100644 --- a/saas_web.py +++ b/saas_web.py @@ -1,5 +1,6 @@ """FastAPI upload UI for shrinking one media file through Codec Carver.""" +import os import tempfile import logging import shutil @@ -7,6 +8,7 @@ from pathlib import Path from fastapi import FastAPI, UploadFile, File, BackgroundTasks, Form, Request from fastapi.responses import HTMLResponse, FileResponse, JSONResponse +import job_store import media_shrinker app = FastAPI(title="Codec Carver SaaS") @@ -334,9 +336,44 @@ def shrink_media( # The synchronous /shrink endpoint blocks for the whole conversion, which is # impractical for long recordings. These endpoints let a client submit a job, # poll its status, and download the result when ready (Upload -> Processing -> -# Result). The store is in-memory and per-process; swap for a shared store to -# scale horizontally. -_JOBS: dict[str, dict] = {} +# Result). Job state lives in a SQLite-backed store (see job_store.py), so it +# survives process restarts and is shared by every worker process on the host. +JOB_STORE = job_store.JobStore(job_store.default_db_path()) + +#: How long finished jobs (and their workspaces) are kept before the startup +#: sweep reclaims them. Override with CODEC_CARVER_JOB_RETENTION_SECONDS. +JOB_RETENTION_SECONDS = float( + os.environ.get( + "CODEC_CARVER_JOB_RETENTION_SECONDS", job_store.DEFAULT_RETENTION_SECONDS + ) +) + + +def run_job_store_maintenance() -> None: + """Reconcile the durable store with reality at process startup. + + Two classes of jobs need attention after a restart: + + * queued/processing jobs — their background workers died with the old + process, so they would stay "in flight" forever. Mark them failed so + polling clients get an honest answer, and reclaim their workspaces. + * old terminal jobs — results nobody downloaded within the retention + window. Drop the rows and reclaim their workspaces. + """ + interrupted = JOB_STORE.recover_interrupted() + for job in interrupted: + if job.get("temp_dir"): + cleanup_temp_dir(Path(job["temp_dir"])) + if interrupted: + logger.warning( + "Failed %d job(s) interrupted by a previous shutdown", len(interrupted) + ) + for job in JOB_STORE.purge_stale(JOB_RETENTION_SECONDS): + if job.get("temp_dir"): + cleanup_temp_dir(Path(job["temp_dir"])) + + +app.router.on_startup.append(run_job_store_maintenance) def _run_job( @@ -348,7 +385,7 @@ def _run_job( temp_dir_path: Path, ) -> None: """Background worker: shrink one uploaded file and record the outcome.""" - _JOBS[job_id]["status"] = "processing" + JOB_STORE.update(job_id, status="processing") try: results = media_shrinker.convert_file( source=source_path, @@ -358,21 +395,22 @@ def _run_job( ) except Exception: logger.exception("Job processing failed") - _JOBS[job_id].update(status="failed", error="Processing failed") + JOB_STORE.update(job_id, status="failed", error="Processing failed") cleanup_temp_dir(temp_dir_path) return if results and results[0].output_path and results[0].output_path.exists(): output_path = results[0].output_path - _JOBS[job_id].update( + JOB_STORE.update( + job_id, status="done", output_path=str(output_path), output_name=output_path.name, ) else: logger.error("Job produced no output: %r", results) - _JOBS[job_id].update( - status="failed", error="Processing failed or no output generated" + JOB_STORE.update( + job_id, status="failed", error="Processing failed or no output generated" ) cleanup_temp_dir(temp_dir_path) @@ -397,7 +435,7 @@ def submit_job( ) job_id = uuid.uuid4().hex - _JOBS[job_id] = {"status": "queued", "temp_dir": str(temp_dir_path)} + JOB_STORE.create(job_id, status="queued", temp_dir=str(temp_dir_path)) background_tasks.add_task( _run_job, job_id, source_path, input_dir, output_dir, target_bytes, temp_dir_path ) @@ -407,15 +445,19 @@ def submit_job( @app.get("/jobs/{job_id}") def job_status(job_id: str): """Return the current status of a previously submitted job.""" - job = _JOBS.get(job_id) + job = JOB_STORE.get(job_id) if job is None: return JSONResponse(status_code=404, content={"error": "Unknown job"}) return {"job_id": job_id, "status": job["status"], "error": job.get("error")} def _cleanup_job(job_id: str) -> None: - """Forget a job and remove its temporary workspace.""" - job = _JOBS.pop(job_id, None) + """Forget a job and remove its temporary workspace. + + `JobStore.delete` returns the row to exactly one caller, so the workspace + is removed once even if cleanup is triggered concurrently. + """ + job = JOB_STORE.delete(job_id) if job is not None and job.get("temp_dir"): cleanup_temp_dir(Path(job["temp_dir"])) @@ -423,13 +465,17 @@ def _cleanup_job(job_id: str) -> None: @app.get("/jobs/{job_id}/result") def job_result(job_id: str, background_tasks: BackgroundTasks): """Download a finished job's output, then clean up its workspace.""" - job = _JOBS.get(job_id) + job = JOB_STORE.get(job_id) if job is None: return JSONResponse(status_code=404, content={"error": "Unknown job"}) if job["status"] != "done": return JSONResponse( status_code=409, content={"error": f"Job is {job['status']}"} ) + if not job.get("output_path") or job.get("temp_dir") is None: + return JSONResponse( + status_code=410, content={"error": "Result no longer available"} + ) # Defense in depth: only ever serve a regular file that lives inside this # job's own temp workspace. `job_id` is an opaque store key and is never # used to build a path, but confining the served path makes traversal diff --git a/tests/test_job_store.py b/tests/test_job_store.py new file mode 100644 index 0000000..b70c81c --- /dev/null +++ b/tests/test_job_store.py @@ -0,0 +1,175 @@ +"""Unit tests for the SQLite-backed durable job store.""" + +import shutil +import sqlite3 +import tempfile +import threading +import unittest +from pathlib import Path +from unittest.mock import patch + +import job_store +from job_store import JobStore + + +class JobStoreTests(unittest.TestCase): + """CRUD, durability, recovery, and concurrency behaviour of JobStore.""" + + def setUp(self) -> None: + self.db_dir = Path(tempfile.mkdtemp(prefix="codec_carver_jobdb_")) + self.store = JobStore(self.db_dir / "jobs.db") + + def tearDown(self) -> None: + shutil.rmtree(self.db_dir, ignore_errors=True) + + def test_create_and_get_roundtrip(self): + self.store.create("a", temp_dir="/tmp/ws-a") + job = self.store.get("a") + self.assertEqual(job["job_id"], "a") + self.assertEqual(job["status"], "queued") + self.assertEqual(job["temp_dir"], "/tmp/ws-a") + self.assertIsNone(job["output_path"]) + self.assertGreater(job["created_at"], 0) + self.assertEqual(job["created_at"], job["updated_at"]) + + def test_get_unknown_returns_none(self): + self.assertIsNone(self.store.get("nope")) + + def test_create_duplicate_id_raises(self): + self.store.create("dup") + with self.assertRaises(sqlite3.IntegrityError): + self.store.create("dup") + + def test_update_changes_fields_and_timestamp(self): + self.store.create("b") + created = self.store.get("b")["created_at"] + with patch("job_store.time.time", return_value=created + 5): + updated = self.store.update( + "b", status="done", output_path="/tmp/out.flac", output_name="out.flac" + ) + self.assertTrue(updated) + job = self.store.get("b") + self.assertEqual(job["status"], "done") + self.assertEqual(job["output_path"], "/tmp/out.flac") + self.assertEqual(job["output_name"], "out.flac") + self.assertGreater(job["updated_at"], job["created_at"]) + + def test_update_unknown_id_returns_false(self): + self.assertFalse(self.store.update("ghost", status="failed")) + + def test_update_without_fields_reports_existence(self): + self.store.create("exists") + self.assertTrue(self.store.update("exists")) + self.assertFalse(self.store.update("missing")) + + def test_unknown_field_rejected(self): + self.store.create("c") + with self.assertRaises(ValueError): + self.store.update("c", is_admin=True) + with self.assertRaises(ValueError): + self.store.create("d", **{"status; DROP TABLE jobs": "x"}) + + def test_delete_returns_row_exactly_once(self): + self.store.create("e", temp_dir="/tmp/ws-e") + first = self.store.delete("e") + second = self.store.delete("e") + self.assertEqual(first["temp_dir"], "/tmp/ws-e") + self.assertIsNone(second) + self.assertIsNone(self.store.get("e")) + + def test_state_survives_reopen(self): + # The core durability property: a brand-new store handle on the same + # path (fresh process, different worker) sees identical state. + self.store.create("persist", temp_dir="/tmp/ws-p") + self.store.update("persist", status="done", output_name="out.flac") + + reopened = JobStore(self.store.db_path) + job = reopened.get("persist") + self.assertEqual(job["status"], "done") + self.assertEqual(job["output_name"], "out.flac") + + def test_all_jobs_ordered_by_creation(self): + with patch("job_store.time.time", side_effect=[100.0, 200.0]): + self.store.create("first") + self.store.create("second") + self.assertEqual([j["job_id"] for j in self.store.all_jobs()], ["first", "second"]) + + def test_recover_interrupted_fails_active_jobs_only(self): + self.store.create("q", status="queued") + self.store.create("p", status="processing") + self.store.create("d", status="done") + self.store.create("f", status="failed", error="original error") + + recovered = self.store.recover_interrupted() + + self.assertEqual(sorted(j["job_id"] for j in recovered), ["p", "q"]) + for job_id in ("q", "p"): + job = self.store.get(job_id) + self.assertEqual(job["status"], "failed") + self.assertEqual(job["error"], "Interrupted by service restart") + self.assertEqual(self.store.get("d")["status"], "done") + self.assertEqual(self.store.get("f")["error"], "original error") + + def test_purge_stale_removes_only_old_terminal_jobs(self): + with patch("job_store.time.time", return_value=1000.0): + self.store.create("old-done", status="done") + self.store.create("old-failed", status="failed") + self.store.create("old-active", status="processing") + with patch("job_store.time.time", return_value=2000.0): + self.store.create("new-done", status="done") + purged = self.store.purge_stale(max_age_seconds=500) + + self.assertEqual( + sorted(j["job_id"] for j in purged), ["old-done", "old-failed"] + ) + self.assertIsNone(self.store.get("old-done")) + self.assertIsNone(self.store.get("old-failed")) + # In-flight jobs are never purged regardless of age; fresh terminal + # jobs stay within the retention window. + self.assertIsNotNone(self.store.get("old-active")) + self.assertIsNotNone(self.store.get("new-done")) + + def test_concurrent_writers_do_not_corrupt_the_store(self): + errors: list[Exception] = [] + + def worker(index: int) -> None: + try: + store = JobStore(self.store.db_path) + for step in range(10): + job_id = f"job-{index}-{step}" + store.create(job_id, temp_dir=f"/tmp/{job_id}") + store.update(job_id, status="done") + except Exception as exc: # pragma: no cover - failure path + errors.append(exc) + + threads = [threading.Thread(target=worker, args=(i,)) for i in range(8)] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + + self.assertEqual(errors, []) + jobs = self.store.all_jobs() + self.assertEqual(len(jobs), 80) + self.assertTrue(all(job["status"] == "done" for job in jobs)) + + def test_default_db_path_honours_env_override(self): + with patch.dict( + "os.environ", {job_store.DB_PATH_ENV: "/data/jobs/custom.db"} + ): + self.assertEqual( + job_store.default_db_path(), Path("/data/jobs/custom.db") + ) + + def test_default_db_path_falls_back_to_tempdir(self): + with patch.dict("os.environ", {}, clear=False): + import os + + os.environ.pop(job_store.DB_PATH_ENV, None) + path = job_store.default_db_path() + self.assertEqual(path.name, "jobs.db") + self.assertEqual(path.parent.name, "codec-carver") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_saas_web.py b/tests/test_saas_web.py index 35563dd..f522b02 100644 --- a/tests/test_saas_web.py +++ b/tests/test_saas_web.py @@ -1,5 +1,7 @@ import asyncio import io +import shutil +import tempfile import unittest from unittest.mock import patch, MagicMock from pathlib import Path @@ -9,6 +11,7 @@ from fastapi.responses import Response import saas_web +from job_store import JobStore from saas_web import app from media_shrinker import ConversionResult @@ -311,18 +314,24 @@ def test_get_ui_includes_preset_buttons(self): unittest.main() -class JobModelTests(unittest.TestCase): - """Async job API: submit -> status -> result, plus all error paths.""" +class JobStoreTestCase(unittest.TestCase): + """Base class that swaps in an isolated on-disk job store per test.""" def setUp(self) -> None: - saas_web._JOBS.clear() + self._db_dir = Path(tempfile.mkdtemp(prefix="codec_carver_jobdb_")) + self._original_store = saas_web.JOB_STORE + saas_web.JOB_STORE = JobStore(self._db_dir / "jobs.db") def tearDown(self) -> None: - for job in list(saas_web._JOBS.values()): - temp = job.get("temp_dir") - if temp: - saas_web.cleanup_temp_dir(Path(temp)) - saas_web._JOBS.clear() + for job in saas_web.JOB_STORE.all_jobs(): + if job.get("temp_dir"): + saas_web.cleanup_temp_dir(Path(job["temp_dir"])) + saas_web.JOB_STORE = self._original_store + shutil.rmtree(self._db_dir, ignore_errors=True) + + +class JobModelTests(JobStoreTestCase): + """Async job API: submit -> status -> result, plus all error paths.""" @patch("saas_web.media_shrinker.convert_file") def test_job_lifecycle_submit_status_result(self, mock_convert_file): @@ -356,19 +365,18 @@ def fake_convert(**kwargs): def test_result_outside_workspace_rejected(self): # A "done" job whose output escaped its workspace must not be served. - import tempfile - workspace = Path(tempfile.mkdtemp(prefix="codec_carver_")) outside_dir = Path(tempfile.mkdtemp()) escaped = outside_dir / "escaped.flac" escaped.write_bytes(b"secret") try: - saas_web._JOBS["escape"] = { - "status": "done", - "output_path": str(escaped), - "output_name": "escaped.flac", - "temp_dir": str(workspace), - } + saas_web.JOB_STORE.create( + "escape", + status="done", + output_path=str(escaped), + output_name="escaped.flac", + temp_dir=str(workspace), + ) response = client.get("/jobs/escape/result") self.assertEqual(response.status_code, 410) finally: @@ -434,29 +442,85 @@ def test_result_unknown_job_returns_404(self): self.assertEqual(response.status_code, 404) def test_result_not_ready_returns_409(self): - saas_web._JOBS["pending"] = {"status": "processing", "temp_dir": ""} + saas_web.JOB_STORE.create("pending", status="processing", temp_dir="") response = client.get("/jobs/pending/result") self.assertEqual(response.status_code, 409) self.assertIn("processing", response.json()["error"]) def test_result_missing_file_returns_410(self): - saas_web._JOBS["gone"] = { - "status": "done", - "output_path": "/nonexistent/output.flac", - "output_name": "output.flac", - "temp_dir": "", - } + saas_web.JOB_STORE.create( + "gone", + status="done", + output_path="/nonexistent/output.flac", + output_name="output.flac", + temp_dir="", + ) response = client.get("/jobs/gone/result") self.assertEqual(response.status_code, 410) - def test_cleanup_job_removes_workspace(self): - import tempfile + def test_result_without_recorded_output_returns_410(self): + # A "done" row with no output_path (e.g. hand-edited or corrupted + # store) must degrade to 410, not crash on Path(None). + saas_web.JOB_STORE.create("noout", status="done", temp_dir="") + response = client.get("/jobs/noout/result") + self.assertEqual(response.status_code, 410) + def test_cleanup_job_removes_workspace(self): temp_dir = Path(tempfile.mkdtemp(prefix="codec_carver_")) - saas_web._JOBS["c"] = {"status": "done", "temp_dir": str(temp_dir)} + saas_web.JOB_STORE.create("c", status="done", temp_dir=str(temp_dir)) saas_web._cleanup_job("c") self.assertFalse(temp_dir.exists()) - self.assertNotIn("c", saas_web._JOBS) + self.assertIsNone(saas_web.JOB_STORE.get("c")) + + +class JobDurabilityTests(JobStoreTestCase): + """The properties the in-memory dict could not provide.""" + + @patch("saas_web.media_shrinker.convert_file", side_effect=RuntimeError("boom")) + def test_job_status_survives_store_reopen(self, _mock_convert): + # Submitting through one store handle and polling through a fresh + # handle on the same database simulates a process restart (or a + # status request served by a different worker process). + submit = client.post( + "/jobs", + files={"file": ("in.wav", io.BytesIO(b"wav data"), "audio/wav")}, + data={"target_bytes": 10000}, + ) + job_id = submit.json()["job_id"] + + saas_web.JOB_STORE = JobStore(saas_web.JOB_STORE.db_path) + status = client.get(f"/jobs/{job_id}") + self.assertEqual(status.status_code, 200) + self.assertEqual(status.json()["status"], "failed") + + def test_startup_maintenance_fails_interrupted_jobs(self): + workspace = Path(tempfile.mkdtemp(prefix="codec_carver_")) + saas_web.JOB_STORE.create("orphan", status="processing", temp_dir=str(workspace)) + + saas_web.run_job_store_maintenance() + + job = saas_web.JOB_STORE.get("orphan") + self.assertEqual(job["status"], "failed") + self.assertIn("restart", job["error"]) + self.assertFalse(workspace.exists()) + + status = client.get("/jobs/orphan") + self.assertEqual(status.json()["status"], "failed") + + def test_startup_maintenance_keeps_recent_terminal_jobs(self): + saas_web.JOB_STORE.create("fresh-done", status="done", temp_dir="") + saas_web.run_job_store_maintenance() + self.assertIsNotNone(saas_web.JOB_STORE.get("fresh-done")) + + def test_startup_maintenance_purges_expired_terminal_jobs(self): + workspace = Path(tempfile.mkdtemp(prefix="codec_carver_")) + saas_web.JOB_STORE.create("expired", status="done", temp_dir=str(workspace)) + + with patch.object(saas_web, "JOB_RETENTION_SECONDS", -1.0): + saas_web.run_job_store_maintenance() + + self.assertIsNone(saas_web.JOB_STORE.get("expired")) + self.assertFalse(workspace.exists()) class UploadValidationTests(unittest.TestCase):