Database: a closing camera view no longer spins a CPU core (#3515)

- aiosqlite 0.22.1 never finishes a close when stop() is called while
  it runs (omnilib/aiosqlite#403); SQLAlchemy does that when a
  cancelled close gives up, so a viewer leaving mid-read hung the
  connection and the stream's cancelled wait spun at 100% CPU
- Make aiosqlite's Connection.stop() idempotent at import, as the
  upstream fix does, for every SQLite connection
- Shield the camera light settings read and the stream re-check, so a
  leaving viewer no longer cuts them off halfway
- Tests: upstream's reproduction, a cancelled-read burst against the
  real engine in a subprocess, and both shielded reads

backend/app/core/auth.py

@@ -9,6 +9,7 @@ from dataclasses import dataclass, field
 from datetime import datetime, timedelta, timezone
 from typing import Annotated
 
+import anyio
 import jwt
 from fastapi import Depends, Header, HTTPException, status
 from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
@@ -1244,41 +1245,45 @@ async def stream_opener_still_allowed(opener: StreamOpener, printer_id: int) ->
     from backend.app.services.long_lived_tokens import _is_expired
 
     try:
-        async with async_session() as db:
-            auth_on = await is_auth_enabled(db)
-            if not opener.auth_on:
-                return not auth_on
-            if not auth_on:
-                return True
-            if opener.long_lived_id is not None:
-                record = await db.get(LongLivedToken, opener.long_lived_id)
-                if (
-                    record is None
-                    or record.revoked_at is not None
-                    or record.user_id != opener.long_lived_user_id
-                    or record.lookup_prefix != opener.long_lived_prefix
-                    or _is_expired(record, datetime.now(timezone.utc))
-                    or not await user_id_holds(db, record.user_id, Permission.CAMERA_VIEW, record.created_at)
-                ):
-                    return False
-                scope = await resolve_user_id_printer_scope(db, record.user_id)
-            elif opener.api_key_id is not None:
-                if await usable_api_key_prefix(db, opener.api_key_id) != opener.api_key_prefix:
-                    return False
-                if not await principal_holds(db, None, opener.api_key_id, Permission.CAMERA_VIEW):
-                    return False
-                scope = await principal_printer_scope(db, None, opener.api_key_id)
-            elif opener.username:
-                user = await get_user_by_username(db, opener.username)
-                if user is None or not user.is_active or not user.has_permission(Permission.CAMERA_VIEW.value):
-                    return False
-                changed = user.password_changed_at
-                if changed is not None and changed.replace(tzinfo=None) > opener.started_at:
+        # Shielded (#3515): the stream ends when its viewer leaves, and a read
+        # cancelled halfway costs its pooled connection. The short check runs
+        # to the end first; the video stream itself is never shielded.
+        with anyio.CancelScope(shield=True):
+            async with async_session() as db:
+                auth_on = await is_auth_enabled(db)
+                if not opener.auth_on:
+                    return not auth_on
+                if not auth_on:
+                    return True
+                if opener.long_lived_id is not None:
+                    record = await db.get(LongLivedToken, opener.long_lived_id)
+                    if (
+                        record is None
+                        or record.revoked_at is not None
+                        or record.user_id != opener.long_lived_user_id
+                        or record.lookup_prefix != opener.long_lived_prefix
+                        or _is_expired(record, datetime.now(timezone.utc))
+                        or not await user_id_holds(db, record.user_id, Permission.CAMERA_VIEW, record.created_at)
+                    ):
+                        return False
+                    scope = await resolve_user_id_printer_scope(db, record.user_id)
+                elif opener.api_key_id is not None:
+                    if await usable_api_key_prefix(db, opener.api_key_id) != opener.api_key_prefix:
+                        return False
+                    if not await principal_holds(db, None, opener.api_key_id, Permission.CAMERA_VIEW):
+                        return False
+                    scope = await principal_printer_scope(db, None, opener.api_key_id)
+                elif opener.username:
+                    user = await get_user_by_username(db, opener.username)
+                    if user is None or not user.is_active or not user.has_permission(Permission.CAMERA_VIEW.value):
+                        return False
+                    changed = user.password_changed_at
+                    if changed is not None and changed.replace(tzinfo=None) > opener.started_at:
+                        return False
+                    scope = await principal_printer_scope(db, opener.username, None)
+                else:
                     return False
-                scope = await principal_printer_scope(db, opener.username, None)
-            else:
-                return False
-            return scope.allows(printer_id)
+                return scope.allows(printer_id)
     except Exception:  # SEC-AUTH-EXC: logged and re-raised, never an answer
         logger.warning("Camera stream re-check failed", exc_info=True)
         raise

backend/app/core/database.py

@@ -14,6 +14,46 @@ from backend.app.core.db_dialect import is_sqlite
 logger = logging.getLogger(__name__)
 
 
+def _make_aiosqlite_stop_idempotent() -> None:
+    """Make aiosqlite's ``Connection.stop()`` safe to call twice (#3515).
+
+    aiosqlite 0.22.1 queues a stop request for the connection's worker thread
+    on every call, and the thread exits at the first. A second call's future
+    is then never resolved, and whoever awaits it waits forever. SQLAlchemy
+    reaches this on a cancelled read: it discards the connection, and when
+    that close is cancelled in turn it calls ``stop()`` while the close is
+    still running. The close never ends, its session's exit waits on it, a
+    pooled connection is lost, and in a camera stream the cancelled wait spun
+    one CPU core at 100%.
+
+    Upstream: omnilib/aiosqlite#403, fix in #404. Once a release has it, this
+    stays harmless (a second call returns the first call's future either way)
+    and can go.
+    """
+    try:
+        from aiosqlite.core import Connection
+    except ImportError:  # PostgreSQL-only environment
+        return
+    if getattr(Connection.stop, "_bambuddy_idempotent", False):
+        return
+    original = Connection.stop
+
+    @functools.wraps(original)
+    def stop(self):
+        try:
+            return self.__dict__["_bambuddy_stop_future"]
+        except KeyError:
+            future = original(self)
+            self.__dict__["_bambuddy_stop_future"] = future
+            return future
+
+    stop._bambuddy_idempotent = True  # type: ignore[attr-defined]
+    Connection.stop = stop
+
+
+_make_aiosqlite_stop_idempotent()
+
+
 def _set_sqlite_pragmas(dbapi_conn, connection_record):
     """Set SQLite pragmas on each new connection for concurrency and performance."""
     cursor = dbapi_conn.cursor()

backend/app/services/camera_light.py

@@ -24,6 +24,8 @@ import time
 from collections.abc import AsyncIterator
 from contextlib import asynccontextmanager
 
+import anyio
+
 logger = logging.getLogger(__name__)
 
 # Long enough to cover a viewer reconnecting and the camera wall polling each
@@ -56,9 +58,13 @@ async def _read_settings() -> tuple[str, float]:
     from backend.app.api.routes.settings import get_setting
     from backend.app.core import database
 
-    async with database.async_session() as db:
-        mode = await get_setting(db, "camera_light_mode")
-        delay = await get_setting(db, "camera_light_delay")
+    # Shielded (#3515): a viewer leaving cancels the stream, and a read
+    # cancelled halfway costs its pooled connection. The two short reads and
+    # the session's close run to the end first.
+    with anyio.CancelScope(shield=True):
+        async with database.async_session() as db:
+            mode = await get_setting(db, "camera_light_mode")
+            delay = await get_setting(db, "camera_light_delay")
     mode = mode if mode in _MODES else "off"
     try:
         seconds = float(delay) if delay is not None else 2.0

backend/tests/unit/test_cancelled_sqlite_close_3515.py

@@ -0,0 +1,178 @@
+"""A read cancelled by a leaving viewer must not hang its connection (#3515).
+
+A camera viewer leaving cancels its stream with AnyIO's level cancellation. A
+read cancelled halfway makes SQLAlchemy discard the connection; when that close
+is cancelled too, it calls aiosqlite's ``stop()`` while the close still runs.
+aiosqlite 0.22.1 then never finished the close (omnilib/aiosqlite#403), the
+session's exit waited on it forever, and the cancelled wait spun a CPU core.
+"""
+
+import asyncio
+import os
+import subprocess
+import sys
+import textwrap
+import threading
+from contextlib import asynccontextmanager
+from pathlib import Path
+from unittest.mock import AsyncMock, patch
+
+import aiosqlite
+import anyio
+import pytest
+
+from backend.app.core import auth, database  # noqa: F401 - database applies the aiosqlite fix
+from backend.app.services import camera_light
+
+pytestmark = pytest.mark.unit
+
+
+class TestAiosqliteStop:
+    async def test_a_stop_during_a_close_lets_the_close_finish(self):
+        # omnilib/aiosqlite#403's reproduction: the worker exits at the stop
+        # requested from outside, before the close's own stop is served.
+        gate = threading.Event()
+        conn = await aiosqlite.connect(":memory:")
+        await conn.create_function("hold", 0, lambda: gate.wait() and 1)
+        busy = asyncio.create_task(conn.execute("SELECT hold()"))
+        await asyncio.sleep(0.05)
+
+        closing = asyncio.create_task(conn.close())
+        await asyncio.sleep(0.05)
+        conn.stop()
+        gate.set()
+        await busy
+
+        await asyncio.wait_for(closing, timeout=5)
+
+    async def test_a_second_stop_returns_the_first_ones_future(self):
+        conn = await aiosqlite.connect(":memory:")
+        first = conn.stop()
+
+        assert conn.stop() is first
+        await asyncio.wait_for(first, timeout=5)
+
+
+# Many short reads, cancelled at once by an AnyIO task group as a leaving viewer
+# cancels its stream, against the engine Bambuddy creates. Run in a process of
+# its own: a hung close can't be cancelled, so the test would hang with it.
+_BURST = textwrap.dedent(
+    """
+    import asyncio, sys
+
+    import anyio
+    from sqlalchemy import text
+    from sqlalchemy.ext.asyncio import async_sessionmaker
+
+    from backend.app.core.database import _create_engine
+
+
+    async def main(url):
+        engine = _create_engine(url)
+        sessions = async_sessionmaker(engine, expire_on_commit=False)
+        async with engine.begin() as conn:
+            await conn.execute(text("create table settings (key text primary key, value text)"))
+            await conn.execute(text("insert into settings values ('camera_light_mode', 'off')"))
+
+        async def read():
+            try:
+                async with sessions() as session:
+                    await session.execute(text("select value from settings"))
+                    await session.execute(text("select value from settings"))
+            except Exception:
+                pass
+
+        for i in range(40):
+            async with anyio.create_task_group() as group:
+                for _ in range(8):
+                    group.start_soon(read)
+                await asyncio.sleep((i % 10) * 0.0002)
+                group.cancel_scope.cancel()
+        await engine.dispose()
+        await asyncio.sleep(1)
+        left = [t for t in asyncio.all_tasks() if t is not asyncio.current_task()]
+        print("pending", len(left))
+
+
+    anyio.run(main, sys.argv[1], backend="asyncio")
+    """
+)
+
+
+def test_cancelled_reads_leave_no_connection_behind(tmp_path):
+    url = f"sqlite+aiosqlite:///{tmp_path / 'burst.db'}"
+    # Its own DATABASE_URL: never the repo's .env.
+    env = {**os.environ, "DATABASE_URL": url}
+    try:
+        result = subprocess.run(
+            [sys.executable, "-c", _BURST, url],
+            cwd=Path(__file__).resolve().parents[3],
+            env=env,
+            capture_output=True,
+            text=True,
+            timeout=60,
+        )
+    except subprocess.TimeoutExpired:
+        pytest.fail("a cancelled read left its connection's close hanging")
+
+    assert result.returncode == 0, result.stderr[-2000:]
+    assert result.stdout.strip().splitlines()[-1] == "pending 0"
+
+
+class _Session:
+    """A session whose every step yields to the loop, so a cancel lands."""
+
+    def __init__(self):
+        self.closed = False
+
+    async def __aenter__(self):
+        await asyncio.sleep(0)
+        return self
+
+    async def __aexit__(self, *exc):
+        await asyncio.sleep(0)
+        self.closed = True
+
+
+class TestTheStreamReadsRunToTheEnd:
+    """Both run inside a camera stream; its viewer leaving must not cut them off."""
+
+    async def test_the_camera_light_settings(self):
+        session = _Session()
+        values = {"camera_light_mode": "all", "camera_light_delay": "3"}
+
+        async def get_setting(_db, key):
+            await asyncio.sleep(0)
+            return values[key]
+
+        result = None
+        with (
+            patch.object(database, "async_session", lambda: session),
+            patch("backend.app.api.routes.settings.get_setting", get_setting),
+            anyio.CancelScope() as viewer_left,
+        ):
+            viewer_left.cancel()
+            result = await camera_light._read_settings()
+
+        assert result == ("all", 3.0)
+        assert session.closed
+
+    async def test_the_stream_re_check(self):
+        session = _Session()
+
+        @asynccontextmanager
+        async def factory():
+            async with session:
+                yield session
+
+        result = None
+        with (
+            patch.object(auth, "async_session", factory),
+            patch.object(auth, "is_auth_enabled", AsyncMock(return_value=False)),
+            anyio.CancelScope() as viewer_left,
+        ):
+            viewer_left.cancel()
+            result = await auth.stream_opener_still_allowed(auth.StreamOpener(auth_on=False), printer_id=1)
+
+        assert result is True
+        assert session.closed