diff --git a/backend/api/v1/endpoints/documents.py b/backend/api/v1/endpoints/documents.py index 54ee73b8d..d4fe0012e 100644 --- a/backend/api/v1/endpoints/documents.py +++ b/backend/api/v1/endpoints/documents.py @@ -23,6 +23,7 @@ from backend.processing.documents import SUPPORTED_EXTENSIONS from backend.services.recording_identity_service import get_recording_by_public_id from backend.utils.document_storage import ( + RecordingGoneError, delete_document_files, insert_document_or_remove_file, ) @@ -59,6 +60,14 @@ async def _get_owned_recording( return recording +async def _insert_document(db: AsyncSession, document: Document) -> None: + """Insert a new document's row, answering 404 if its recording went meanwhile.""" + try: + await insert_document_or_remove_file(db, document, logger=logger) + except RecordingGoneError as error: + raise HTTPException(status_code=404, detail="Recording not found") from error + + # Configuration for documents storage. Resolved through PathManager so it tracks # NOJOIN_DATA_DIR rather than the process working directory. DOCUMENTS_DIR = str(PathManager().documents_directory) @@ -96,7 +105,7 @@ async def create_text_document( status=DocumentStatus.PENDING, parse_mode=DocumentParseMode.STRUCTURAL, ) - await insert_document_or_remove_file(db, document, logger=logger) + await _insert_document(db, document) await db.commit() await db.refresh(document) @@ -294,7 +303,7 @@ async def upload_document( # noqa: PLR0913 - FastAPI dependencies are parameter DocumentParseMode.VISUAL if deep_parse else DocumentParseMode.STRUCTURAL ), ) - await insert_document_or_remove_file(db, document, logger=logger) + await _insert_document(db, document) await db.commit() await db.refresh(document) diff --git a/backend/tests/test_storage_hygiene_entrypoints.py b/backend/tests/test_storage_hygiene_entrypoints.py index 84dd5f4fc..399cced02 100644 --- a/backend/tests/test_storage_hygiene_entrypoints.py +++ b/backend/tests/test_storage_hygiene_entrypoints.py @@ -8,6 +8,7 @@ import logging import os +import sys import tempfile import uuid from collections.abc import Iterator @@ -446,3 +447,30 @@ def no_database(): ] assert len(failure) == 1 assert failure[0].exc_info is not None + + +def test_a_hygiene_module_that_fails_to_load_does_not_stop_the_daily_cleanup( + layout: Layout, + daily_mode: str, + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, +) -> None: + """The sweeps before hygiene must still run when its import fails. + + #660 put a FastAPI import on the hygiene path, which the worker images do + not install. The task imported hygiene on its first line, so it failed + before any sweep ran. + """ + monkeypatch.setitem(sys.modules, "backend.utils.storage_hygiene", None) + stale = write_file(layout.recordings / "failed" / "1_failed_x.wav", age_hours=72) + + with caplog.at_level(logging.ERROR): + system_tasks.cleanup_temp_recordings.run() + + assert not stale.exists(), daily_mode + failure = [ + r for r in caplog.records if r.getMessage().startswith("Storage hygiene failed") + ] + assert len(failure) == 1 + assert failure[0].exc_info is not None + assert isinstance(failure[0].exc_info[1], ImportError) diff --git a/backend/tests/test_worker_import_boundary.py b/backend/tests/test_worker_import_boundary.py new file mode 100644 index 000000000..1690a6eba --- /dev/null +++ b/backend/tests/test_worker_import_boundary.py @@ -0,0 +1,343 @@ +"""The workers must import without FastAPI or anything else only the API installs. + +The worker images install ``requirements/worker.txt``, which has no FastAPI, +while CI installs the backend and worker sets together, so a worker module that +imports FastAPI passes every test there and fails in production. #660 was that: +the restore reached ``backend.utils.document_storage``, which imported FastAPI +at module level, and every backup, restore and backup cleanup on the workers +failed with ``No module named 'fastapi'``. The chain started at an import +inside the task functions, so the workers still started cleanly. + +These tests work out every backend module a worker can reach, through imports +at module level and inside functions alike, then import them all in a +subprocess with the backend-only packages blocked, which is what the worker +images look like. ``test_api_torch_boundary.py`` holds the API to the same rule +the other way round. +""" + +from __future__ import annotations + +import ast +import importlib.metadata +import json +import re +import subprocess +import sys +from pathlib import Path + +import pytest + +from backend.celery_app import celery_app + +REPO_ROOT = Path(__file__).resolve().parents[2] +BACKEND_ROOT = REPO_ROOT / "backend" +REQUIREMENTS = REPO_ROOT / "requirements" + +# Top-level modules the API image has and the base worker image lacks: the pins +# only requirements/backend.txt makes, and what they pull in. worker-io carries +# a few of them through claude-agent-sdk, but worker-gpu and worker-cpu do not, +# and worker-cpu runs the backups. +BACKEND_ONLY_MODULES = frozenset( + { + "_re2", + "annotated_doc", + "docker", + "fastapi", + "httptools", + "jsonschema", + "jsonschema_specifications", + "mcp", + "mcp_types", + "re2", + "referencing", + "reportlab", + "rpds", + "sse_starlette", + "starlette", + "uvicorn", + "uvloop", + "watchfiles", + } +) + +# Pinned in backend.txt and not in worker.txt, yet in the worker images anyway, +# as a dependency of something worker.txt installs. +IN_WORKER_IMAGES_ANYWAY = frozenset({"markdown-it-py"}) # through rich + +# Installed in the worker images outside the requirements files, so absent from +# CI. A module that needs one is skipped by the import check; the static check +# below still reads it. +IMAGE_ONLY_PACKAGES = ("pyannote.audio",) + +# What a worker container runs besides the Celery app's include list: the entry +# module named on the command line, the GPU preflight the entrypoint runs, the +# script diarisation starts in a child process, and the storage hygiene command +# an administrator runs in worker-io. +WORKER_ENTRY_MODULES = ( + "backend.celery_app", + "backend.processing.gpu_preflight", + "backend.processing.diarize_subprocess_entry", + "backend.storage_hygiene", +) + +_BLOCKER = """ +import importlib.abc +import sys + +BLOCKED = frozenset({blocked!r}) + + +class _BackendOnly(importlib.abc.MetaPathFinder): + def find_spec(self, name, path=None, target=None): + if name.partition(".")[0] in BLOCKED: + raise ModuleNotFoundError(f"No module named {{name!r}}", name=name) + return None + + +sys.meta_path.insert(0, _BackendOnly()) +""".format(blocked=sorted(BACKEND_ONLY_MODULES)) + +_IMPORT_ALL = """ +import importlib +import json +import traceback + +image_only = {image_only!r} +modules = json.loads(sys.argv[1]) +failures = {{}} +skipped = {{}} + +# What `celery worker` imports as it starts: the app, then its include list. +from backend.celery_app import celery_app + +celery_app.loader.import_default_modules() + +for module in modules: + try: + importlib.import_module(module) + except ModuleNotFoundError as error: + missing = error.name or "" + if missing.startswith(image_only) and missing.partition(".")[0] not in BLOCKED: + skipped[module] = missing + else: + failures[module] = traceback.format_exc(limit=-6) + except Exception: + failures[module] = traceback.format_exc(limit=-6) + +print("REPORT " + json.dumps({{"failures": failures, "skipped": skipped}})) +""".format(image_only=IMAGE_ONLY_PACKAGES) + + +def _module_path(module: str) -> Path | None: + base = BACKEND_ROOT.parent.joinpath(*module.split(".")) + for candidate in (base.with_suffix(".py"), base / "__init__.py"): + if candidate.is_file(): + return candidate + return None + + +def _is_type_checking_block(node: ast.If) -> bool: + test = node.test + name = getattr(test, "id", None) or getattr(test, "attr", None) + return name == "TYPE_CHECKING" + + +def _guards_import_error(node: ast.Try) -> bool: + for handler in node.handlers: + caught = handler.type + names = ( + caught.elts if isinstance(caught, ast.Tuple) else [caught] if caught else [] + ) + for name in names: + if getattr(name, "id", None) in {"ImportError", "ModuleNotFoundError"}: + return True + return False + + +def _import_targets(node: ast.Import | ast.ImportFrom, package: str) -> list[str]: + if isinstance(node, ast.Import): + return [alias.name for alias in node.names] + if node.level: + parts = package.split(".") + base = parts[: len(parts) - node.level + 1] + target = ".".join(base + ([node.module] if node.module else [])) + else: + target = node.module or "" + # `from backend.pkg import submodule` imports the submodule too. + return [target] + [ + f"{target}.{alias.name}" for alias in node.names if alias.name != "*" + ] + + +def _children(node: ast.AST, guarded: bool) -> list[tuple[ast.AST, bool]]: + """The nodes under this one that can run, each with whether a guard covers it.""" + if isinstance(node, ast.If) and _is_type_checking_block(node): + return [(child, guarded) for child in node.orelse] + if isinstance(node, ast.Try) and _guards_import_error(node): + rest = [*node.handlers, *node.orelse, *node.finalbody] + return [(child, True) for child in node.body] + [ + (child, guarded) for child in rest + ] + return [(child, guarded) for child in ast.iter_child_nodes(node)] + + +def _imports(module: str, path: Path) -> list[tuple[str, int, bool]]: + """Every module this file imports: (name, line, guarded by an ImportError handler). + + Imports inside functions count, since a task imports most of its work + lazily. Imports under ``if TYPE_CHECKING:`` never run, so they do not. + """ + package = module if path.name == "__init__.py" else module.rpartition(".")[0] + found: list[tuple[str, int, bool]] = [] + pending: list[tuple[ast.AST, bool]] = [ + (ast.parse(path.read_text(encoding="utf-8")), False) + ] + while pending: + node, guarded = pending.pop() + if isinstance(node, ast.Import | ast.ImportFrom): + found.extend( + (name, node.lineno, guarded) for name in _import_targets(node, package) + ) + else: + pending.extend(_children(node, guarded)) + return found + + +def _worker_roots() -> list[str]: + roots = list(WORKER_ENTRY_MODULES) + list(celery_app.conf.include) + for path in sorted((BACKEND_ROOT / "worker").rglob("*.py")): + parts = path.relative_to(BACKEND_ROOT.parent).with_suffix("").parts + if parts[-1] == "__init__": + parts = parts[:-1] + roots.append(".".join(parts)) + return roots + + +def _worker_reachable_modules() -> dict[str, Path]: + """Every backend module a worker process can import, keyed by dotted name.""" + reachable: dict[str, Path] = {} + pending = _worker_roots() + while pending: + module = pending.pop() + if module in reachable: + continue + path = _module_path(module) + if path is None: + continue + reachable[module] = path + # Importing a.b.c runs a/__init__.py and a/b/__init__.py first. + parent = module.rpartition(".")[0] + if parent: + pending.append(parent) + pending.extend( + name + for name, _line, _guarded in _imports(module, path) + if name.partition(".")[0] == "backend" + ) + return reachable + + +def _run(body: str, *args: str) -> subprocess.CompletedProcess: + return subprocess.run( + [sys.executable, "-c", _BLOCKER + body, *args], + capture_output=True, + text=True, + timeout=600, + cwd=REPO_ROOT, + ) + + +def test_the_blocker_blocks_fastapi(): + """Guard the guard: a blocker that lets FastAPI through proves nothing.""" + result = _run("import fastapi\nprint('IMPORTED')") + assert "IMPORTED" not in result.stdout + assert "No module named 'fastapi'" in result.stderr + + +def test_the_walk_follows_imports_inside_task_functions(): + """The #660 chain began at an import inside the backup and restore tasks.""" + reachable = _worker_reachable_modules() + assert "backend.core.backup" in reachable + assert "backend.utils.document_storage" in reachable + assert not [name for name in reachable if name.startswith("backend.api.")] + + +def test_worker_modules_import_without_backend_only_packages(): + modules = sorted(_worker_reachable_modules()) + result = _run(_IMPORT_ALL, json.dumps(modules)) + reports = [ + line.removeprefix("REPORT ") + for line in result.stdout.splitlines() + if line.startswith("REPORT ") + ] + assert reports, ( + "The import check did not finish.\n" + f"stdout:\n{result.stdout[-4000:]}\nstderr:\n{result.stderr[-4000:]}" + ) + report = json.loads(reports[-1]) + assert not report["failures"], ( + "These modules fail to import in a worker image, which has no FastAPI " + "or other package only requirements/backend.txt installs. Raise a " + "domain error and convert it in the endpoint instead:\n\n" + + "\n".join( + f"{module}:\n{trace}" for module, trace in report["failures"].items() + ) + ) + + +def test_worker_modules_never_import_backend_only_packages(): + """Catch an import of one inside a function, which no import check runs. + + An import guarded by an ``except ImportError`` or ``ModuleNotFoundError`` + is allowed: the calendar service keeps its API-facing helpers that way. + """ + offenders = sorted( + { + f"{path.relative_to(REPO_ROOT)}:{line} imports {name.partition('.')[0]}" + for module, path in _worker_reachable_modules().items() + for name, line, guarded in _imports(module, path) + if name.partition(".")[0] in BACKEND_ONLY_MODULES and not guarded + } + ) + assert not offenders, ( + "Modules the workers import must not import a package only the API " + "image installs:\n" + "\n".join(offenders) + ) + + +def _requirement_names(filename: str) -> set[str]: + names = set() + for raw in (REQUIREMENTS / filename).read_text(encoding="utf-8").splitlines(): + line = raw.split("#", 1)[0].strip() + if not line or line.startswith("-"): + continue + name = re.split(r"[\[<>=!~;@\s]", line, maxsplit=1)[0] + names.add(re.sub(r"[-_.]+", "-", name).lower()) + return names + + +def test_every_backend_only_requirement_is_blocked(): + """A new pin in backend.txt alone must join the blocked modules above.""" + modules_of: dict[str, set[str]] = {} + for module, distributions in importlib.metadata.packages_distributions().items(): + for distribution in distributions: + key = re.sub(r"[-_.]+", "-", distribution).lower() + modules_of.setdefault(key, set()).add(module) + + backend_only = ( + _requirement_names("backend.txt") + - _requirement_names("worker.txt") + - IN_WORKER_IMAGES_ANYWAY + ) + not_installed = sorted(name for name in backend_only if name not in modules_of) + if not_installed: + pytest.fail(f"Install requirements/backend.txt first: {not_installed}") + unblocked = { + name: sorted(modules_of[name] - BACKEND_ONLY_MODULES) + for name in sorted(backend_only) + if modules_of[name] - BACKEND_ONLY_MODULES + } + assert not unblocked, ( + "Add these to BACKEND_ONLY_MODULES, or to IN_WORKER_IMAGES_ANYWAY if " + f"the worker images install them through another package: {unblocked}" + ) diff --git a/backend/utils/document_storage.py b/backend/utils/document_storage.py index cad89c5f1..9f194f277 100644 --- a/backend/utils/document_storage.py +++ b/backend/utils/document_storage.py @@ -6,6 +6,10 @@ documents' paths before the rows go and removes the files once the delete has committed. An upload writes its file before inserting its row, and removes the file again if that insert fails or is cancelled. + +The restore imports this module in the workers, whose images do not install +FastAPI, so nothing here may import it (#660). An upload whose recording has +gone raises RecordingGoneError, and the endpoint answers it with the 404. """ from __future__ import annotations @@ -16,8 +20,7 @@ from functools import partial from pathlib import Path -from fastapi import HTTPException -from fastapi.concurrency import run_in_threadpool +import anyio.to_thread from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession from sqlmodel import col, select @@ -34,6 +37,10 @@ _FOREIGN_KEY_VIOLATION = "23503" +class RecordingGoneError(LookupError): + """A new document's recording was deleted before its row went in.""" + + def documents_root_dir() -> Path: return PathManager().documents_directory @@ -134,10 +141,12 @@ async def delete_document_files( the job would then never run; so a cancelled call unlinks on the event loop before the cancellation propagates. If the thread had already started, the files are unlinked twice, which is harmless: a missing file is not an error. + The thread comes from anyio's pool, as FastAPI's ``run_in_threadpool`` would + take it, without importing FastAPI. """ unlink = partial(unlink_document_files, list(file_paths), logger=logger) try: - await run_in_threadpool(unlink) + await anyio.to_thread.run_sync(unlink) except asyncio.CancelledError: unlink() raise @@ -151,11 +160,12 @@ async def insert_document_or_remove_file( If the insert fails or is cancelled, nothing would ever reference the file, so it is removed before the error propagates. A recording deleted during the upload is the usual failure: the insert waits on the delete's lock, then - fails its foreign key, and that becomes the 404 a missing recording gets - everywhere else. Only the insert is guarded. A commit that raises may still - have landed, so its file stays. + fails its foreign key, and raises RecordingGoneError, which the endpoint + turns into the 404 a missing recording gets everywhere else. Only the insert + is guarded. A commit that raises may still have landed, so its file stays. """ file_path = document.file_path + recording_id = document.recording_id db.add(document) try: await db.flush() @@ -170,7 +180,5 @@ async def insert_document_or_remove_file( isinstance(error, IntegrityError) and getattr(error.orig, "sqlstate", None) == _FOREIGN_KEY_VIOLATION ): - raise HTTPException( - status_code=404, detail="Recording not found" - ) from error + raise RecordingGoneError(recording_id) from error raise diff --git a/backend/worker/tasks/system.py b/backend/worker/tasks/system.py index 69cf32dba..de3247e1b 100644 --- a/backend/worker/tasks/system.py +++ b/backend/worker/tasks/system.py @@ -15,12 +15,6 @@ def cleanup_temp_recordings(self): from backend.processing.audio_preprocessing import ( cleanup_stale_pipeline_temp_files, ) - from backend.utils.storage_hygiene import ( - STORAGE_HYGIENE_DELETE, - HygieneOptions, - run_storage_hygiene, - storage_hygiene_mode, - ) cleaned_count = cleanup_recording_audio_chunks(self.session, logger=logger) cleaned_count += cleanup_stale_recording_artifacts(max_age_hours=24, logger=logger) @@ -32,8 +26,16 @@ def cleanup_temp_recordings(self): logger.info(f"Cleanup complete. Removed {cleaned_count} items.") # Last, so the sweeps above have already run. It reads in a session of its - # own, opened after it lists the directories. + # own, opened after it lists the directories. Imported inside the guard, so + # a hygiene module that fails to load stops only this step (#660). try: + from backend.utils.storage_hygiene import ( + STORAGE_HYGIENE_DELETE, + HygieneOptions, + run_storage_hygiene, + storage_hygiene_mode, + ) + run_storage_hygiene( get_sync_session, HygieneOptions(delete=storage_hygiene_mode() == STORAGE_HYGIENE_DELETE), diff --git a/docs/DEVELOPMENT.md b/docs/DEVELOPMENT.md index a58e6ad41..8ad870b97 100644 --- a/docs/DEVELOPMENT.md +++ b/docs/DEVELOPMENT.md @@ -292,6 +292,7 @@ NOJOIN_TEST_POSTGRES_URL=postgresql://postgres:postgres@localhost:55432/nojoin_t - **Flaky tests.** Report a test that passes and fails with no code change as described in [CONTRIBUTING.md](../.github/CONTRIBUTING.md#reporting-issues); triage adds the `flaky` label. Fix it by removing the nondeterminism (timing, ordering, shared state, network), not by re-running it. A skip or `xfail` needs a linked issue and a stated condition for removing it. - **No Redis.** `backend/tests/conftest.py` fails any test that opens a Redis connection, printing `Test opened a real Redis connection` and the call site. Take the `stub_celery_dispatch` fixture for a Celery dispatch, `stub_meeting_edge_dispatch` for transcript mutations, or stub the client the code uses. Stub an attribute on a shared object (`celery_app.send_task`) or a function in its defining module; a stub on a package that re-exports a helper misses callers that imported the name earlier. - **Model registry.** `conftest.py` imports `backend.models.registry`, so a test module behaves the same alone as in the full suite. `test_api_mapper_configuration.py` checks that each real entry point registers every model. +- **Image boundaries.** CI installs the API and worker requirements together, so a test passes on an import that one of the images cannot make. `test_api_torch_boundary.py` imports API modules with torch blocked, and `test_worker_import_boundary.py` imports worker modules with FastAPI and the API's other packages blocked ([Worker imports](#audio-processing--ml-operations)). - **ffmpeg.** Tests that run real media through ffmpeg skip when `shutil.which("ffmpeg")` finds nothing. CI installs Ubuntu's ffmpeg 6.1, the series the worker image ships; the API image ships 7.1. A test that depends on one version's behaviour should say which. To simulate a host without ffmpeg, point `PATH` at an empty directory with `monkeypatch.setenv`. ## Browser Capture Development @@ -364,6 +365,7 @@ These apply to all tracked source, frontend included. ### Audio Processing & ML Operations - **Heavy imports**: import ML libraries such as `torch`, `whisper` and `pyannote` inside Celery task functions under [backend/worker/tasks/](../backend/worker/tasks/), never on API startup paths. +- **Worker imports**: the worker images install `requirements/worker.txt`, which has no FastAPI. A module a worker can import, by any import a task makes, must not import FastAPI or another package only `requirements/backend.txt` installs, inside a function or out. Raise a domain error and turn it into the `HTTPException` in the endpoint. CI installs both sets, so [test_worker_import_boundary.py](../backend/tests/test_worker_import_boundary.py) walks every module a worker can reach and imports them with those packages blocked. A new pin in `backend.txt` alone fails it until its modules join `BACKEND_ONLY_MODULES`. - **Task dispatch from async code**: never call `celery_app.send_task`, `.delay()` or `.apply_async()` inside an `async def`. Publishing is a blocking socket call, and an unreachable Redis would stall the event loop and every request on it. Use `dispatch_task` from [backend/core/task_dispatch.py](../backend/core/task_dispatch.py), or `dispatch_task_best_effort` where a dispatch failure must not fail the caller. A test fails on inline dispatch inside an `async def`, but it cannot see a synchronous helper that dispatches and is called from a coroutine, so check that case by reading. Worker code dispatches inline, which is safe only under the `solo` and `prefork` pools set in Compose ([Worker Concurrency Lanes](DEPLOYMENT.md#worker-concurrency-lanes)). - **ASR engines**: transcription dispatches to an engine under [backend/processing/engines/](../backend/processing/engines/) chosen by the `transcription_backend` setting (`whisper`, `parakeet` or `canary`). Every engine implements `TranscriptionEngine`; Parakeet and Canary share the `OnnxAsrEngine` base class. - **Window manifests**: read `RecordingAudioWindowManifest.asr_status` for live ASR coverage, not the legacy `status` projection, and never treat `live_processed` as diarisation-complete. Nothing writes the diarisation fields any more; [ARCHITECTURE.md](ARCHITECTURE.md) describes both lanes.