diff --git a/deploy/CLOUDRUN.md b/deploy/CLOUDRUN.md index 859038b..279db72 100644 --- a/deploy/CLOUDRUN.md +++ b/deploy/CLOUDRUN.md @@ -14,6 +14,7 @@ when idle. | Port | `8000` | | Auth | public (`--allow-unauthenticated`) | | Request timeout | `600s` (`--timeout 600` — see note below) | +| CPU allocation | always-on (`--no-cpu-throttling` — see async-job note below) | > **Why 600s:** a real single-clip live generation measures **~330–350s** > end-to-end (Kling render ≈242s avg + input hosting, stitch, provenance, @@ -22,6 +23,15 @@ when idle. > script pins `--timeout 600` so synchronous `POST /reels*` requests outlive > the real generation path. +> **Why `--no-cpu-throttling`:** by default Cloud Run only allocates CPU to an +> instance while it has a request in flight — the moment a response is sent, +> CPU is throttled to near-zero. `POST /reels/jobs` (async submit + poll — +> see below) returns its `202` immediately and keeps generating in a +> **background thread** after that response has already gone out. Without +> `--no-cpu-throttling` that thread would barely progress between polls. This +> only affects billing while a job is actually in flight — Cloud Run still +> scales to zero (and bills nothing) when idle. + ## Prerequisites (one-time) ```bash @@ -110,6 +120,32 @@ then revoke the old keys. Cloud Run reads `:latest`, so the redeploy picks up the new version with no code change — and the old plaintext-env-var exposure is gone for good. +## Async job submission (submit + poll) + +`POST /reels/jobs` / `GET /reels/jobs/{job_id}` exist alongside the +synchronous `POST /reels*` routes for the same reason `--timeout 600` does: +edge proxies in front of Cloud Run — Firebase Hosting's rewrite proxy in +particular — cap a single request at **~60s**, well under a real generation's +~330–350s. The async pair splits that into a fast submit (`202` + a job id) +and a cheap poll (`GET /reels/jobs/{job_id}` → `queued` / `running` / `done` / +`failed`), so no single HTTP request needs to stay open for the full render. +See `src/cinemory/jobs.py` for the job-store design (status objects live in +the same B2/fake storage backend the rest of the app already uses, under +`jobs//status.json` — not a separate database). + +**Honest limitation:** the background worker runs **in-process, on the Cloud +Run instance that accepted the submit** — there is no external queue and no +separate worker fleet. A client that keeps polling keeps that instance warm +(a request in flight resets Cloud Run's idle-scale-down clock), so in +practice the job completes. But if that specific instance is scaled down +mid-job, the in-flight generation is lost with no automatic retry — the +stored status simply stops advancing. This is a deliberate, acceptable +tradeoff for a demo, not a production job queue; a production version would +hand the work to a durable queue (e.g. Cloud Tasks) consumed by a Cloud Run +**Job** (not a request-serving instance), so the work outlives any one +instance. `--no-cpu-throttling` (above) is required even for this demo +version — without it, the worker thread barely progresses between polls. + ## Domain mapping — cinemory.ai `cinemory.ai` is an **apex** domain, so it maps via **A + AAAA** records diff --git a/deploy/deploy-cloudrun.sh b/deploy/deploy-cloudrun.sh index 0ed7a1f..6fb8957 100644 --- a/deploy/deploy-cloudrun.sh +++ b/deploy/deploy-cloudrun.sh @@ -115,6 +115,11 @@ fi # (Kling render ~242s avg + hosting/stitch/provenance). Cloud Run's 300s default # 504'd those requests at the edge while the reel completed server-side # (proven live 2026-07-22), so the request deadline must sit above the real path. +# --no-cpu-throttling: POST /reels/jobs runs the actual generation in a +# background thread AFTER the request that submitted it has already returned +# (see src/cinemory/jobs.py) — by default Cloud Run only allocates CPU while a +# request is in flight, which would starve that thread between polls. See +# deploy/CLOUDRUN.md for the full async-job note. gcloud run deploy "${SERVICE}" \ --image "${IMAGE}" \ --region "${REGION}" \ @@ -123,6 +128,7 @@ gcloud run deploy "${SERVICE}" \ --port 8000 \ --timeout 600 \ --cpu 1 --memory 512Mi \ + --no-cpu-throttling \ --min-instances 0 --max-instances 4 \ --set-env-vars "${ENV_VARS}" \ ${SET_SECRETS_ARGS[@]+"${SET_SECRETS_ARGS[@]}"} \ diff --git a/src/cinemory/api.py b/src/cinemory/api.py index 21e9f64..db59157 100644 --- a/src/cinemory/api.py +++ b/src/cinemory/api.py @@ -7,6 +7,10 @@ POST /reels/upload generate a reel from real uploaded photos (base64 JSON) POST /reels/upload-multipart generate a reel from real uploaded photos (multipart) + POST /reels/jobs submit a reel generation as a background job (202 + + job_id) — does not block for the full generation + GET /reels/jobs/{job_id} + poll a submitted job's status (queued/running/done/failed) GET /reels/{name} fetch the stored provenance manifest for a reel GET /reels/{name}/video play back the stored reel (302 to a fresh presigned @@ -20,6 +24,15 @@ the reel is regenerated with the offline provider (storage unchanged) and the response says so honestly (``provider_degraded`` + the actual ``provider``, which the sealed manifest also records per step). + +Async submit + poll (``POST /reels/jobs`` / ``GET /reels/jobs/{job_id}``): the +synchronous ``POST /reels*`` routes above can run for minutes (a real render +averages ~242s — see ``deploy/CLOUDRUN.md``), longer than the Firebase Hosting +proxy cap (~60s), so a caller that cannot hold one request open that long +submits a job and polls it instead. The async routes run the exact same ingest +validation and pipeline as the synchronous ones (see ``_run_job`` below and the +``cinemory.jobs`` module for the job-store design and its honest limitation — +the worker runs in-process on the instance that accepted the submit). """ from __future__ import annotations @@ -28,6 +41,7 @@ import json import logging import os +import threading from collections.abc import Callable from pathlib import Path from typing import Annotated @@ -37,12 +51,13 @@ from fastapi.responses import RedirectResponse, Response from pydantic import BaseModel, Field -from . import config -from .adapters import FakeMediaProvider +from . import config, jobs +from .adapters import FakeMediaProvider, FakeStorage from .ingest import IngestError, build_spec_from_photos from .models import ReelResult, ReelSpec from .occasions import list_occasions from .pipeline import ReelPipeline +from .ports import StorageBackend from .provenance import verify_all from .stitch import FakeStitcher from .synthetic import synth_reel_spec @@ -93,7 +108,7 @@ def _reel_response(result: ReelResult) -> dict: } -def _run_reel(spec: ReelSpec) -> dict: +def _run_reel(spec: ReelSpec, pipeline: ReelPipeline | None = None) -> dict: """Run the pipeline; degrade THIS request honestly if the live provider fails. The core action must never 500 because a remote generation backend @@ -103,28 +118,35 @@ def _run_reel(spec: ReelSpec) -> dict: plus the provider that actually generated, and the sealed manifest records that provider on every step. A failure of the offline provider itself is a genuine bug and propagates (500) rather than being masked. + + ``pipeline`` defaults to the module-level synchronous pipeline. The async + job worker (``_run_job``) passes its own isolated pipeline instead (see + ``_job_pipeline``), so a background thread never shares mutable adapter + state with a request thread — the honest-degrade fallback below then also + writes to THAT pipeline's storage, not necessarily the module-level one. """ + pipeline = pipeline or _pipeline try: - body = _reel_response(_pipeline.run(spec)) - body["provider"] = _pipeline.provider.name + body = _reel_response(pipeline.run(spec)) + body["provider"] = pipeline.provider.name body["provider_degraded"] = False return body except HTTPException: raise except Exception as exc: - if isinstance(_pipeline.provider, FakeMediaProvider): + if isinstance(pipeline.provider, FakeMediaProvider): raise _log.exception( "live media provider %r failed for reel %r; regenerating this " "request with the offline provider (storage unchanged)", - _pipeline.provider.name, + pipeline.provider.name, spec.name, ) # The offline provider's deterministic clips are not decodable video, # so the regeneration pairs it with the offline stitcher (a real # ffmpeg stitcher would fail on them) — the exact offline generation # path, persisted to whatever storage the deployment actually uses. - fallback = ReelPipeline(FakeMediaProvider(), _storage, stitcher=FakeStitcher()) + fallback = ReelPipeline(FakeMediaProvider(), pipeline.storage, stitcher=FakeStitcher()) body = _reel_response(fallback.run(spec)) body["provider"] = fallback.provider.name body["provider_degraded"] = True @@ -134,16 +156,23 @@ def _run_reel(spec: ReelSpec) -> dict: return body -def _generate_from_photos( +def _build_spec( name: str, photos: list[tuple[str, bytes]], *, occasion: str, chapters: int, bridges: bool -) -> dict: - """Shared ingest path for both upload endpoints (base64 + multipart).""" +) -> ReelSpec: + """Ingest validation shared by every upload path — sync AND async.""" try: - spec = build_spec_from_photos( + return build_spec_from_photos( name, photos, occasion=occasion, chapters=chapters, bridges=bridges ) except IngestError as exc: raise HTTPException(400, str(exc)) from exc + + +def _generate_from_photos( + name: str, photos: list[tuple[str, bytes]], *, occasion: str, chapters: int, bridges: bool +) -> dict: + """Shared ingest path for both SYNCHRONOUS upload endpoints (base64 + multipart).""" + spec = _build_spec(name, photos, occasion=occasion, chapters=chapters, bridges=bridges) return _run_reel(spec) @@ -172,6 +201,20 @@ def create_reel(req: ReelRequest) -> dict: return _run_reel(spec) +def _decode_photos(photos: list[UploadedPhoto]) -> list[tuple[str, bytes]]: + """Decode base64 photo payloads — shared by the base64 upload path, sync + (``/reels/upload``) AND async (``/reels/jobs``): same validation, same 400 + message, same order.""" + decoded: list[tuple[str, bytes]] = [] + for i, p in enumerate(photos): + try: + data = base64.b64decode(p.content_base64, validate=True) + except (binascii.Error, ValueError) as exc: + raise HTTPException(400, f"photo {i} is not valid base64") from exc + decoded.append((p.filename or f"photo{i}.png", data)) + return decoded + + @app.post("/reels/upload") def upload_reel(req: UploadReelRequest) -> dict: """Generate a reel from real photo bytes sent as base64 JSON. @@ -180,13 +223,7 @@ def upload_reel(req: UploadReelRequest) -> dict: actual pixels; the decoded photos flow through the same storage + pipeline as the synthetic demo, sealing real SHA-256 provenance. """ - photos: list[tuple[str, bytes]] = [] - for i, p in enumerate(req.photos): - try: - data = base64.b64decode(p.content_base64, validate=True) - except (binascii.Error, ValueError) as exc: - raise HTTPException(400, f"photo {i} is not valid base64") from exc - photos.append((p.filename or f"photo{i}.png", data)) + photos = _decode_photos(req.photos) return _generate_from_photos( req.name, photos, occasion=req.occasion, chapters=req.chapters, bridges=req.bridges ) @@ -211,6 +248,111 @@ async def upload_reel_multipart( ) +# ── Async job submission (submit + poll) ─────────────────────────────────────── +# See the ``cinemory.jobs`` module for the job-store design and its honest +# limitation (the worker below runs in-process on the accepting instance). + + +def _job_storage() -> StorageBackend: + """Storage instance a background job thread may safely read/write. + + ``FakeStorage`` (offline/tests) is pure in-memory with no remote substrate + — a FRESH instance here would be invisible to request threads (including + this same process's own ``GET /reels/jobs/{job_id}`` reads), so the shared + module-level ``_storage`` is reused; its plain dict/list mutations are safe + to share across threads under the GIL (no read-modify-write cycle). + + ``B2Storage`` (live) is different: it keeps a mutable local ``index`` list + plus a remote read-modify-write merge on every ``put`` (a small race + already documented and accepted in ``b2_storage.py`` for independent + writers). Sharing the SAME Python object — and its local ``index`` list — + between this background thread and request-serving threads would add an + avoidable extra race on top of that one (a shared list can be reassigned by + one thread while another thread is mid-iteration over the old reference — + a lost update, distinct from the already-accepted remote-merge race). So + live mode gets its own fresh ``B2Storage`` per job; its constructor + re-reads the bucket's ``index.jsonl``, so nothing is lost — the bucket, not + the Python object, is the source of truth, and this is the exact + multi-writer pattern the adapter's own test suite already proves safe + (independent ``B2Storage`` instances against one bucket merge by key + rather than clobbering each other). + """ + if isinstance(_storage, FakeStorage): + return _storage + return config.build_storage() # pragma: no cover - exercised only in live mode + + +def _job_pipeline(storage: StorageBackend) -> ReelPipeline: + """The pipeline a background job thread runs its generation against. + + Unlike storage, the provider/stitcher have no cross-request visibility + requirement (nothing else ever reads them back by reference), so a fresh + instance is always built — this keeps ``GenblazeMediaProvider.last_manifest`` + and ``FakeMediaProvider.calls`` (both mutated per call, on the instance) + from ever being shared between this background thread and the module-level + synchronous pipeline's provider. + """ + return ReelPipeline(config.build_provider(), storage, stitcher=config.build_stitcher()) + + +def _run_job(job_id: str, spec: ReelSpec) -> None: + """Background worker thread: run the generation, updating the stored status. + + Runs the SAME honest-degrade path the synchronous endpoints use + (``_run_reel``), against an isolated pipeline/storage (see + ``_job_storage``/``_job_pipeline``) so this thread never mutates state a + request thread also touches. Never raises out of the thread and never + produces an HTTP 500 for a generation failure: caught here and recorded as + status ``failed`` with the exception CLASS NAME only (mirrors + ``_run_reel``'s ``degrade_reason``) — the full traceback goes to the log. + """ + storage = _job_storage() + try: + jobs.mark_running(storage, job_id) + result = _run_reel(spec, _job_pipeline(storage)) + jobs.mark_done(storage, job_id, result) + except Exception as exc: + _log.exception("background reel job %r failed", job_id) + jobs.mark_failed(storage, job_id, exc) + + +@app.post("/reels/jobs", status_code=202) +def create_reel_job(req: UploadReelRequest) -> dict: + """Submit a reel generation as a background job; poll ``GET /reels/jobs/{id}``. + + Accepts the IDENTICAL body as ``POST /reels/upload`` — same decoding, same + ingest validation, same 400s for a bad request — but returns immediately + (202) with a job id instead of blocking for the full generation. See the + module docstring and ``cinemory.jobs`` for why (edge proxy timeouts) and + for the honest limitation of the in-process worker. + """ + photos = _decode_photos(req.photos) + spec = _build_spec( + req.name, photos, occasion=req.occasion, chapters=req.chapters, bridges=req.bridges + ) + job_id = jobs.new_job_id() + jobs.create(_job_storage(), job_id) + threading.Thread( + target=_run_job, args=(job_id, spec), daemon=True, name=f"cinemory-job-{job_id}" + ).start() + return {"job_id": job_id, "status": "queued"} + + +@app.get("/reels/jobs/{job_id}") +def get_reel_job(job_id: str) -> dict: + """Poll a submitted job's status. + + 404 for an unknown job id. A stored-but-unreadable status object degrades + to the identical 404 (see ``cinemory.jobs.read``) rather than a 500 — from + the caller's side an unreadable job and a nonexistent one are the same + "nothing to show" answer. + """ + status = jobs.read(_storage, job_id) + if status is None: + raise HTTPException(404, f"no job {job_id!r}") + return status + + def _reel_index_match(name: str, *, kind: str, suffix: str) -> dict | None: """Resolve a reel's stored object row from the durable index. diff --git a/src/cinemory/jobs.py b/src/cinemory/jobs.py new file mode 100644 index 0000000..46465fd --- /dev/null +++ b/src/cinemory/jobs.py @@ -0,0 +1,150 @@ +"""B2-backed job store for async reel generation (the submit + poll pattern). + +A reel generation can run for minutes (a real Kling render averages ~242s; see +``deploy/CLOUDRUN.md``), which is longer than the Firebase Hosting proxy cap +(~60s) and comparable to Cloud Run's own request ceiling. ``POST /reels/jobs`` +(see ``cinemory.api``) avoids blocking on that: it validates the request, +writes a "queued" status object, starts the real generation in a background +thread, and returns immediately with a job id. ``GET /reels/jobs/{job_id}`` +polls that status back. + +A job's state is a small JSON object stored through the SAME +:class:`~cinemory.ports.StorageBackend` the rest of the app already uses (B2 in +live mode, the in-memory fake offline), under ``jobs//status.json`` — +NOT a separate in-process registry. That is what makes status pollable across +Cloud Run instances: the object store, not any one process's memory, is the +source of truth, so a scale-to-zero instance that never saw the submit can +still answer the poll by re-reading the key. The job id itself is a +:func:`secrets.token_urlsafe` token (unguessable, not a sequential id or a +``uuid1`` that would leak MAC address / timestamp bits). + +Honest limitation — read before assuming this is durable +---------------------------------------------------------- +The background worker runs **in-process, on the Cloud Run instance that +accepted the submit** (see ``cinemory.api._run_job``). There is no external +queue and no separate worker fleet. A client that keeps polling keeps that +instance warm (a request in flight resets Cloud Run's idle-scale-down clock), +so in practice the job completes. But if that specific instance is scaled down +mid-job — e.g. nobody polls for a while, or Cloud Run reclaims it for another +reason — the in-flight generation is lost with no automatic retry; the stored +status simply stops advancing past whatever it last reached. This is an +acceptable, deliberate tradeoff for a demo, not a production job queue. A +production version would hand the work to a durable queue (e.g. Cloud Tasks) +consumed by a Cloud Run *Job* (not a request-serving instance), so the work +outlives any one instance. See ``deploy/CLOUDRUN.md`` for the Cloud Run +``--no-cpu-throttling`` flag this needs even for the demo version (without it, +Cloud Run throttles CPU to near-zero between requests, and the background +thread would barely make progress between polls). +""" +from __future__ import annotations + +import json +import logging +import secrets +from datetime import datetime, timezone +from typing import Any + +from .keys import safe_component +from .ports import StorageBackend + +_log = logging.getLogger("cinemory.jobs") + +#: Every job status object lives under this prefix, in its own id-named folder +#: — a namespace that can never collide with a reel's own +#: ``////`` keys (different segment count; see +#: ``cinemory.keys.make_key``). +_KEY_PREFIX = "jobs" + +#: Terminal statuses a poller should stop on. +TERMINAL_STATUSES = ("done", "failed") + + +def new_job_id() -> str: + """A fresh, unguessable job id. + + ``secrets.token_urlsafe`` (a CSPRNG), never ``uuid1`` (leaks MAC address + + timestamp bits) and never a sequential counter (guessable/enumerable). + """ + return secrets.token_urlsafe(18) + + +def _now() -> str: + return datetime.now(timezone.utc).isoformat() + + +def _key(job_id: str) -> str: + # job_id is server-generated (url-safe alphabet, no '/'), but a lookup by + # job_id also flows in from the URL path on GET — sanitise defensively the + # same way every other user-reachable key segment in this codebase is + # (see cinemory.keys.safe_component), so a hostile id can never be shaped + # into a key outside this job's own prefix. A legitimate id is untouched + # by this (it is already within the safe charset). + return f"{_KEY_PREFIX}/{safe_component(job_id)}/status.json" + + +def _write(storage: StorageBackend, job_id: str, status: dict) -> dict: + storage.put( + _key(job_id), + json.dumps(status, indent=2).encode("utf-8"), + content_type="application/json", + ) + return status + + +def create(storage: StorageBackend, job_id: str) -> dict: + """Seed a new job as ``queued`` and persist it. The first write for this id.""" + now = _now() + return _write(storage, job_id, { + "job_id": job_id, + "status": "queued", + "created_at": now, + "updated_at": now, + }) + + +def read(storage: StorageBackend, job_id: str) -> dict | None: + """The stored status dict, or ``None`` if unknown/unreadable. + + Honest-degrade, mirroring ``cinemory.api.verify_reel``'s "unreadable bytes + never 500" contract: a job id that was never created, a storage read error, + corrupt JSON, or (defensively) JSON that isn't an object all read the same + way — ``None`` — so the caller (``GET /reels/jobs/{job_id}``) can turn any + of them into one honest 404 rather than leaking a stack trace. + """ + try: + parsed: Any = json.loads(storage.get(_key(job_id))) + except Exception: + return None + return parsed if isinstance(parsed, dict) else None + + +def _update(storage: StorageBackend, job_id: str, **fields: Any) -> dict: + """Read-modify-write the stored status: apply ``fields``, stamp ``updated_at``. + + Falls back to a fresh ``queued`` base if nothing was stored yet (defensive + only — the normal flow always calls :func:`create` first), so a transition + call can never itself raise for a missing predecessor object. + """ + current = read(storage, job_id) or { + "job_id": job_id, "status": "queued", "created_at": _now(), + } + current.update(fields) + current["updated_at"] = _now() + return _write(storage, job_id, current) + + +def mark_running(storage: StorageBackend, job_id: str) -> dict: + return _update(storage, job_id, status="running") + + +def mark_done(storage: StorageBackend, job_id: str, result: dict) -> dict: + return _update(storage, job_id, status="done", result=result) + + +def mark_failed(storage: StorageBackend, job_id: str, exc: BaseException) -> dict: + # Full detail goes to the log only; the stored/returned status carries the + # exception CLASS NAME alone — exception text may embed URLs/identifiers + # that don't belong in an API-readable object. Mirrors + # cinemory.api._run_reel's ``degrade_reason`` (same reasoning, same shape). + _log.warning("job %r failed: %s: %s", job_id, type(exc).__name__, exc) + return _update(storage, job_id, status="failed", error=type(exc).__name__) diff --git a/tests/integration/test_api_jobs.py b/tests/integration/test_api_jobs.py new file mode 100644 index 0000000..8199aa0 --- /dev/null +++ b/tests/integration/test_api_jobs.py @@ -0,0 +1,160 @@ +"""Integration tests for async reel generation (submit + poll) — the M2 fix. + +``POST /reels/jobs`` accepts the identical body as ``POST /reels/upload`` but +returns ``202`` + a job id immediately instead of blocking for the full +generation; ``GET /reels/jobs/{job_id}`` polls the stored status back. Driven +through the real FastAPI app, offline (FakeStorage + FakeMediaProvider + +FakeStitcher), so "background" generation is effectively instant — the tests +still poll with a short bounded timeout rather than assume one check +suffices, since the work genuinely runs on a separate thread. + +The existing synchronous endpoints (``tests/integration/test_api.py`` and +friends) are untouched by this file and by the refactor that backs it — see +``src/cinemory/api.py``'s ``_build_spec``/``_decode_photos``/``_run_reel``. +""" +from __future__ import annotations + +import base64 +import json +import time + +from fastapi.testclient import TestClient + +import cinemory.api as api +from cinemory import jobs +from cinemory.adapters import FakeMediaProvider +from cinemory.pipeline import ReelPipeline + +client = TestClient(api.app) + + +def _b64(data: bytes) -> str: + return base64.b64encode(data).decode() + + +def _poll_until_terminal(c: TestClient, job_id: str, *, timeout: float = 5.0) -> dict: + """Poll GET /reels/jobs/{job_id} until it reaches a terminal status. + + Offline generation is effectively instant (no real I/O), so 5s is a very + generous ceiling — this just avoids a hard assumption that the background + thread has already finished by the time the first poll lands. + """ + deadline = time.monotonic() + timeout + body: dict = {} + while time.monotonic() < deadline: + r = c.get(f"/reels/jobs/{job_id}") + assert r.status_code == 200 + body = r.json() + if body.get("status") in jobs.TERMINAL_STATUSES: + return body + time.sleep(0.02) + raise AssertionError( + f"job {job_id} did not reach a terminal status within {timeout}s: {body}" + ) + + +def test_submit_returns_202_with_job_id_and_queued_status(): + photos = [{"filename": f"p{i}.png", "content_base64": _b64(f"pixels-{i}".encode())} + for i in range(3)] + r = client.post("/reels/jobs", json={"name": "asyncjob-submit", "occasion": "wedding", + "chapters": 2, "photos": photos}) + assert r.status_code == 202 + body = r.json() + assert body["status"] == "queued" + assert isinstance(body["job_id"], str) and body["job_id"] + + +def test_poll_reaches_done_with_real_provenance_result(): + photos = [{"filename": f"p{i}.png", "content_base64": _b64(f"pixels-{i}".encode())} + for i in range(4)] + r = client.post("/reels/jobs", json={"name": "asyncjob-done", "occasion": "anniversary", + "chapters": 2, "photos": photos}) + assert r.status_code == 202 + job_id = r.json()["job_id"] + + status = _poll_until_terminal(client, job_id) + assert status["status"] == "done" + assert status["job_id"] == job_id + assert status["created_at"] and status["updated_at"] + + result = status["result"] + assert result["reel_name"] == "asyncjob-done" + assert result["occasion"] == "anniversary" + assert len(result["reel_sha256"]) == 64 + assert result["manifest_hash"] + assert result["steps"] == 4 # 4 photos -> 4 clips, no bridges by default + assert result["provider_degraded"] is False + assert "playback_url" in result and "manifest_uri" in result + + # The job's generation really landed in storage — fetchable the normal way. + manifest = client.get("/reels/asyncjob-done").json() + assert manifest["reel_name"] == "asyncjob-done" + + +def test_poll_with_bridges_matches_sync_step_count(): + photos = [{"filename": f"p{i}.png", "content_base64": _b64(f"x{i}".encode())} + for i in range(4)] + r = client.post("/reels/jobs", json={"name": "asyncjob-bridges", "chapters": 2, + "bridges": True, "photos": photos}) + assert r.status_code == 202 + status = _poll_until_terminal(client, r.json()["job_id"]) + assert status["status"] == "done" + assert status["result"]["steps"] == 5 # 4 clips + 1 bridge, same as the sync path + + +def test_submit_no_photos_is_400_synchronously(): + """Ingest validation happens BEFORE a job is created — same 400 as the + sync endpoint, no job id handed out for a request that will never run.""" + r = client.post("/reels/jobs", json={"name": "asyncjob-empty", "photos": []}) + assert r.status_code == 400 + + +def test_submit_bad_base64_is_400_synchronously(): + r = client.post("/reels/jobs", json={ + "name": "asyncjob-badb64", + "photos": [{"filename": "p.png", "content_base64": "not!base64!!"}]}) + assert r.status_code == 400 + + +def test_unknown_job_id_is_404(): + r = client.get("/reels/jobs/does-not-exist") + assert r.status_code == 404 + + +def test_corrupt_stored_status_degrades_to_404_not_500(): + """A status object that exists but doesn't parse degrades the same way an + unknown job id does (see cinemory.jobs.read) — never a 500.""" + api._storage.put("jobs/corrupt-job/status.json", b"not valid json") + r = client.get("/reels/jobs/corrupt-job") + assert r.status_code == 404 + + +def test_job_forced_failure_marks_status_failed_not_500(monkeypatch): + """A generation failure surfaces ONLY through the polled status — never as + an HTTP error anywhere in the submit/poll flow. Mirrors the sync path's + ``test_offline_provider_failure_is_a_real_500`` fixture (a FakeMediaProvider + subclass that raises), except here the offline failure has nowhere to + "surface" as a request 500, because the generation never happens on a + request thread at all.""" + + class _BrokenFake(FakeMediaProvider): + def generate(self, **_kwargs) -> bytes: + raise RuntimeError("offline provider bug (forced for this test)") + + monkeypatch.setattr( + api, "_job_pipeline", lambda storage: ReelPipeline(_BrokenFake(), storage) + ) + + photos = [{"filename": "p.png", "content_base64": _b64(b"px-bytes")}] + r = client.post("/reels/jobs", json={"name": "asyncjob-forced-fail", "photos": photos}) + # The submit itself never sees the failure — it happens later, in the + # background thread. + assert r.status_code == 202 + job_id = r.json()["job_id"] + + status = _poll_until_terminal(client, job_id) + assert status["status"] == "failed" + assert status["error"] == "RuntimeError" + assert "result" not in status + # Class name only — the raw exception text never reaches the response. + assert "forced for this test" not in json.dumps(status) diff --git a/tests/unit/test_jobs.py b/tests/unit/test_jobs.py new file mode 100644 index 0000000..fa2d9f3 --- /dev/null +++ b/tests/unit/test_jobs.py @@ -0,0 +1,111 @@ +"""Unit tests for the async job store (``cinemory.jobs``). + +Driven directly against a :class:`~cinemory.adapters.fake_storage.FakeStorage` +(the same ``StorageBackend`` port the real B2 adapter implements) — no HTTP, +no threading, so these pin the module's own contract in isolation. The full +submit -> background-thread -> poll flow is covered separately in +``tests/integration/test_api_jobs.py``. +""" +from __future__ import annotations + +import json + +from cinemory import jobs +from cinemory.adapters import FakeStorage + + +def test_new_job_id_is_url_safe_and_unique(): + a = jobs.new_job_id() + b = jobs.new_job_id() + assert a != b + assert len(a) >= 16 + allowed = set("ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789_-") + assert set(a) <= allowed + + +def test_create_writes_queued_status_and_is_readable_back(): + storage = FakeStorage() + status = jobs.create(storage, "job1") + assert status["job_id"] == "job1" + assert status["status"] == "queued" + assert status["created_at"] == status["updated_at"] + + read_back = jobs.read(storage, "job1") + assert read_back == status + # Stored through the same StorageBackend port the rest of the app uses, + # under the documented key — not a separate in-process registry. + assert storage.get("jobs/job1/status.json") + + +def test_read_unknown_job_is_none(): + storage = FakeStorage() + assert jobs.read(storage, "does-not-exist") is None + + +def test_read_corrupt_bytes_is_none(): + storage = FakeStorage() + storage.put("jobs/broken/status.json", b"not json at all") + assert jobs.read(storage, "broken") is None + + +def test_read_non_object_json_is_none(): + """Defensive: valid JSON that isn't an object (e.g. a bare number) still + degrades to None rather than being handed back as a "status".""" + storage = FakeStorage() + storage.put("jobs/weird/status.json", b"42") + assert jobs.read(storage, "weird") is None + + +def test_mark_running_then_done_preserves_created_at_and_advances_updated_at(): + storage = FakeStorage() + created = jobs.create(storage, "job2") + running = jobs.mark_running(storage, "job2") + assert running["status"] == "running" + assert running["created_at"] == created["created_at"] + + result = {"reel_name": "job2", "reel_sha256": "a" * 64, "steps": 2} + done = jobs.mark_done(storage, "job2", result) + assert done["status"] == "done" + assert done["result"] == result + assert done["created_at"] == created["created_at"] + + # The poll path (GET /reels/jobs/{id}) reads back exactly this. + assert jobs.read(storage, "job2") == done + + +def test_mark_failed_records_exception_class_name_only(): + storage = FakeStorage() + jobs.create(storage, "job3") + failed = jobs.mark_failed(storage, "job3", RuntimeError("some sensitive internal detail")) + assert failed["status"] == "failed" + assert failed["error"] == "RuntimeError" + assert "result" not in failed + assert "sensitive internal detail" not in json.dumps(failed) + + +def test_mark_running_without_prior_create_seeds_a_base_status(): + """Defensive fallback: a transition call never raises for a missing + predecessor object (the normal flow always calls create() first, via + cinemory.api.create_reel_job before the background thread starts).""" + storage = FakeStorage() + running = jobs.mark_running(storage, "orphan") + assert running["job_id"] == "orphan" + assert running["status"] == "running" + assert running["created_at"] + + +def test_hostile_job_id_key_stays_sanitised_and_namespaced(): + """A job id can never be shaped into a key outside jobs//... — the id + is sanitised the same way every other user-reachable key segment in this + codebase is (cinemory.keys.safe_component), even though in the real flow + job ids are always server-generated (secrets.token_urlsafe), never + attacker-supplied on write.""" + storage = FakeStorage() + jobs.create(storage, "../../evil") + keys = {row["key"] for row in storage.index} + assert keys # something was written + for key in keys: + segments = key.split("/") + assert segments[0] == "jobs" + assert ".." not in segments + assert not any(seg.startswith(".") for seg in segments)