Database: the #3515 fix also starts on older aiosqlite

- The stop() patch read Connection.stop, which aiosqlite only has from
  0.22.1; requirements.txt allows older versions and the in-app update
  keeps an installed one, so the backend failed to start. Older
  versions have no stop() for SQLAlchemy to call, so they are left
  alone
- A viewer leaving during a slow, shielded stream re-check is now
  noticed before the immediate retry, instead of being held for a
  second pool timeout
- Tests: aiosqlite without stop(), and one call for a viewer who left

backend/app/api/routes/camera.py

@@ -14,6 +14,7 @@ import zlib
 from collections.abc import AsyncGenerator
 from typing import Annotated
 
+import anyio.lowlevel
 from fastapi import APIRouter, Body, Depends, HTTPException, Query, Request
 from fastapi.responses import Response, StreamingResponse
 from pydantic import BaseModel
@@ -1098,6 +1099,10 @@ async def _while_allowed(frames, opener: StreamOpener, printer_id: int):
                     if time.monotonic() < next_check:
                         yield chunk
                         continue
+                    # The re-check is shielded, so a viewer who left while it
+                    # waited is only told now: let that land before a second
+                    # shielded wait, which would keep the stream (#3515).
+                    await anyio.lowlevel.checkpoint()
                     allowed = await _opener_allowed(opener, printer_id)
                 if not allowed:
                     logger.info("Camera stream for printer %s ended: its opener may no longer watch", printer_id)

backend/app/core/database.py

@@ -29,14 +29,18 @@ def _make_aiosqlite_stop_idempotent() -> None:
     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.
+
+    aiosqlite before 0.22.1 has no ``stop()``, and SQLAlchemy then never calls
+    one, so there is nothing to fix. requirements.txt still allows those
+    versions, and the in-app update keeps an installed one.
     """
     try:
         from aiosqlite.core import Connection
     except ImportError:  # PostgreSQL-only environment
         return
-    if getattr(Connection.stop, "_bambuddy_idempotent", False):
+    original = getattr(Connection, "stop", None)
+    if original is None or getattr(original, "_bambuddy_idempotent", False):
         return
-    original = Connection.stop
 
     @functools.wraps(original)
     def stop(self):

backend/tests/unit/test_cancelled_sqlite_close_3515.py

@@ -176,3 +176,54 @@ class TestTheStreamReadsRunToTheEnd:
 
         assert result is True
         assert session.closed
+
+
+class TestOlderAiosqlite:
+    def test_a_version_without_stop_is_left_alone(self, monkeypatch):
+        # Before 0.22.1 aiosqlite has no stop(), and SQLAlchemy calls none. The
+        # in-app update keeps such a version, so it must not fail the import.
+        class Connection:
+            pass
+
+        monkeypatch.setattr("aiosqlite.core.Connection", Connection)
+
+        database._make_aiosqlite_stop_idempotent()
+
+        assert not hasattr(Connection, "stop")
+
+
+class TestALeavingViewerDuringASlowReCheck:
+    async def test_the_retry_does_not_keep_the_stream(self, monkeypatch):
+        # Every database connection busy: the shielded re-check waits for one
+        # and times out. A viewer who left meanwhile must not be held through
+        # the immediate retry's own shielded wait as well.
+        from backend.app.api.routes import camera
+
+        monkeypatch.setattr(camera, "STREAM_RECHECK_SECONDS", 0.0)
+        monkeypatch.setattr(camera, "STREAM_RECHECK_RETRY_SECONDS", 0.0)
+        calls = 0
+
+        async def slow_failure(_opener, _printer_id):
+            nonlocal calls
+            calls += 1
+            with anyio.CancelScope(shield=True):
+                await asyncio.sleep(0.2)
+            raise TimeoutError("pool")
+
+        monkeypatch.setattr(camera, "stream_opener_still_allowed", slow_failure)
+
+        async def frames():
+            while True:
+                yield b"frame"
+                await asyncio.sleep(0)
+
+        async def watch():
+            async for _ in camera._while_allowed(frames(), auth.StreamOpener(auth_on=False), 1):
+                pass
+
+        async with anyio.create_task_group() as group:
+            group.start_soon(watch)
+            await asyncio.sleep(0.05)
+            group.cancel_scope.cancel()
+
+        assert calls == 1