feat: durable SQLite-backed job store (replaces in-memory _JOBS dict)

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)}")

saas_web.py

@@ -1,12 +1,14 @@
 """FastAPI upload UI for shrinking one media file through Codec Carver."""
 
+import os
 import tempfile
 import logging
 import shutil
 import uuid
 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,29 +445,37 @@ 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"]))
 
 
 @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