fix: a Postgres connection was held across every celery round trip (4295)

Operator, 2026-09-23: *"something about changing the cap number is blocking to
the website... it shouldn't be"*.

Nothing here was slow in itself. A database connection was held across work
that is slow, and that is why it surfaced as the whole site stalling rather
than as one slow page.

`lane_view` took the session and kept it open through a celery inspect whose
budget is 11s. The System tab polls that endpoint every 15s — and with a lane
not answering, every inspect runs to nearly its full budget, so each poll
pinned a connection for most of the interval. SQLAlchemy's default pool is 5
plus 10 overflow. Two browser tabs, `/api/system/health` doing the same thing,
and a cap change adding two more inspects exhausts it, and every OTHER request
then waits for a connection.

Split so the database work finishes before the broker work starts:

- `lane_settings(session)` reads the caps and the oldest running task, then
  the session closes. `lane_view(settings)` does the inspect with none held.
- `store_lane_cap(session, …)` validates and commits, then the session closes.
  `push_lane_cap(lane, …)` does the live push with none held.

And a second finding while measuring it: **raising a cap now costs no broker
round trip at all.** The first cut only knew on/off, so it inspected on every
raise to find out whether the pool needed lowering — the control meant to be
instant still waited out an inspect. `store_lane_cap` returns the PREVIOUS cap
so the push knows the direction; only a lowering needs to say anything.

The guard is structural, not timed: `lane_view` and `push_lane_cap` must not
ACCEPT a session. A timing test would be flaky, and a call-order test would
pass against a version that took the session and merely used it early.

`/api/system/health` has the same shape and is NOT fixed here — it is
rate-limited by `refresh_if_stale` so it does not inspect on every request.
Worth doing, separately.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01LVjrnpQjRgHdvq95rASoiR

backend/app/api/workers.py

@@ -23,7 +23,13 @@ from datetime import UTC, datetime
 from quart import Blueprint, jsonify, request
 
 from ..extensions import get_session
-from ..services.worker_control import LaneUpdateRefused, lane_view, set_lane
+from ..services.worker_control import (
+    LaneUpdateRefused,
+    lane_settings,
+    lane_view,
+    push_lane_cap,
+    store_lane_cap,
+)
 from ..services.worker_lanes import LANES_BY_NAME
 from ._responses import error_response as _bad
 
@@ -41,8 +47,14 @@ async def list_lanes():
     reply would show them the value from before their own change and read as
     the control having failed.
     """
+    # The session closes BEFORE the broker work. Holding a Postgres connection
+    # across a celery inspect is what made this page block the whole site —
+    # see `worker_control.LaneSettings`. This endpoint polls every 15s and the
+    # inspect budget is 11s, so each poll was pinning a connection for most of
+    # the interval.
     async with get_session() as session:
-        lanes = await lane_view(session)
+        settings = await lane_settings(session)
+    lanes = await lane_view(settings)
     return jsonify({
         "lanes": lanes,
         "fetched_at": datetime.now(UTC).isoformat(),
@@ -85,9 +97,13 @@ async def update_lane(name: str):
     if not isinstance(value, int) or isinstance(value, bool):
         return _bad("invalid_body", detail="slots_cap must be an integer")
 
+    # Store, close the session, THEN push. Same reason as the GET above, and
+    # more sharply here: a cap change could do three broker round trips, all
+    # of them previously with a connection held.
     async with get_session() as session:
         try:
-            result = await set_lane(session, lane, slots_cap=value)
+            was_cap = await store_lane_cap(session, lane, value)
         except LaneUpdateRefused as exc:
             return _bad("refused", detail=str(exc))
+    result = await push_lane_cap(lane, value, was_cap=was_cap)
     return jsonify(result)

backend/app/services/worker_control.py

@@ -334,14 +334,53 @@ async def _rows_by_name(session: AsyncSession) -> dict[str, WorkerLane]:
     return rows
 
 
-async def lane_view(session: AsyncSession) -> list[dict]:
+@dataclass
+class LaneSettings:
+    """What the DATABASE knows about the lanes — read and finished with before
+    anything touches the broker.
+
+    This exists because holding a Postgres connection across a celery round
+    trip is what made the System tab block the whole site (operator,
+    2026-09-23: *"something about changing the cap number is blocking to the
+    website"*).
+
+    `lane_view` used to take the session and keep it open through an inspect
+    whose budget is eleven seconds — and that page polls every fifteen. With a
+    lane not answering, every inspect ran to nearly its full budget, so each
+    poll pinned a connection for ten seconds. SQLAlchemy's default pool is
+    five connections plus ten overflow; a couple of browser tabs, the health
+    endpoint doing the same thing, and a cap change adding two more inspects
+    exhausts that, and every OTHER request then waits on a connection.
+
+    Nothing was slow in itself. The slowness was a scarce resource held across
+    it, which is why it surfaced as the whole site stalling rather than as one
+    slow page.
+    """
+
+    caps: dict[str, int]
+    oldest_by_queue: dict[str, datetime]
+
+
+async def lane_settings(session: AsyncSession) -> LaneSettings:
+    """Every DB read the lane view needs, in one short-lived session."""
+    rows = await _rows_by_name(session)
+    return LaneSettings(
+        caps={name: row.slots_cap for name, row in rows.items()},
+        oldest_by_queue=await _oldest_running_by_queue(session),
+    )
+
+
+async def lane_view(settings: LaneSettings) -> list[dict]:
     """Every lane: what is configured, what is live, what it may grow to.
 
-    One call rather than making the UI join three sources. `pending` is the
+    Takes the settings rather than a session ON PURPOSE — see `LaneSettings`.
-    honest backlog — Redis depth PLUS reserved — because celery prefetches and
+    Everything below this line is broker work, and no database connection is
-    LLEN alone reads 0 while a worker holds tasks in memory.
+    held while it happens.
+
+    `pending` is the honest backlog — Redis depth PLUS reserved — because
+    celery prefetches and LLEN alone reads 0 while a worker holds tasks in
+    memory.
     """
-    rows = await _rows_by_name(session)
     # A deadline, because this is a request path and `to_thread` on its own is
     # an await with no bound (rule 156). `inspect_lanes_sync` never raises and
     # every inner call has its own timeout, so the only way past the budget is
@@ -359,12 +398,12 @@ async def lane_view(session: AsyncSession) -> list[dict]:
         )
         live = {lane.name: LaneLiveState() for lane in LANES}
     depths = await asyncio.to_thread(_queue_depths_sync)
-    oldest = await _oldest_running_by_queue(session)
+    oldest = settings.oldest_by_queue
 
     now = datetime.now(UTC)
     out = []
     for lane in LANES:
-        row = rows[lane.name]
+        cap = settings.caps[lane.name]
         state = live[lane.name]
         # None for a queue the broker did not answer for, which must not be
         # silently summed as zero — an unknown depth is not an empty one.
@@ -376,11 +415,11 @@ async def lane_view(session: AsyncSession) -> list[dict]:
             "name": lane.name,
             "display_name": lane.display_name,
             "queues": list(lane.queues),
-            "slots_cap": row.slots_cap,
+            "slots_cap": cap,
             "ceiling": derived_ceiling(lane),
             # DERIVED, never stored. A cap of zero means no consumers, so
             # "off" and "may use no workers" cannot disagree.
-            "enabled": row.slots_cap > 0,
+            "enabled": cap > 0,
             "memory_bound": lane.memory_bound,
             "optional": lane.optional,
             # What raising this lane's cap will download, so the UI can say
@@ -485,30 +524,14 @@ class LaneUpdateRefused(ValueError):
     the UI shows — a greyed control with no explanation reads as a bug."""
 
 
-async def set_lane(session: AsyncSession, lane: Lane, *, slots_cap: int) -> dict:
+async def store_lane_cap(
-    """Store the operator's cap for `lane`, then make the live lane obey it.
+    session: AsyncSession, lane: Lane, slots_cap: int,
+) -> int:
+    """Validate and store the cap. Returns the PREVIOUS cap. DB only.
 
-    ONE value, since 2026-09-23. It used to take `slots`, `slots_cap`,
+    Split from the live push for the reason `LaneSettings` gives at length: a
-    `enabled` and `autoscale`, which was four ways of saying two things — and
+    Postgres connection must not be held across a celery round trip. Everything
-    two of them were the caller's job to keep in agreement with each other.
+    here is fast and finished with before `push_lane_cap` starts.
-
-    ## What happens live, and what does not
-
-    A cap CHANGE is pushed immediately in one direction only: lowering it
-    shrinks the pool now, because a cap the operator just lowered should not
-    be exceeded for up to a minute. Raising it does NOT grow the pool here —
-    a cap is permission, not a request, and growing on permission would put
-    slots on a lane with nothing to do. The sizing pass adds them on its next
-    tick if there is work, which is the whole point of it being always on.
-
-    Consumers follow the cap in both directions and immediately: zero means
-    off, and off must take effect when it is asked for.
-
-    A failed PUSH is not a failed setting. The value is saved either way and
-    the sizing pass carries it within a minute; the result says `applied:
-    false` with a reason so the UI can say "saved, not yet live" rather than
-    "that didn't work" (lesson #4202 — a live change that does not survive,
-    with nothing saying so).
     """
     rows = await _rows_by_name(session)
     row = rows[lane.name]
@@ -522,11 +545,40 @@ async def set_lane(session: AsyncSession, lane: Lane, *, slots_cap: int) -> dict
             f"({ceiling} for {lane.display_name})"
         )
 
-    was_on = row.slots_cap > 0
+    was_cap = row.slots_cap
     row.slots_cap = slots_cap
     await session.commit()
+    # The previous value, because the push needs the DIRECTION: lowering a cap
+    # has to reach the running lane now, and raising one has nothing to say.
+    return was_cap
 
-    now_on = slots_cap > 0
+
+async def push_lane_cap(lane: Lane, slots_cap: int, *, was_cap: int) -> dict:
+    """Make the running lane obey a cap that is already stored. NO database.
+
+    ## What is pushed, and what is not
+
+    Consumers follow the cap immediately in BOTH directions: zero means off,
+    and off must take effect when it is asked for rather than up to a minute
+    later.
+
+    The pool is only ever pushed DOWNWARD. Raising a cap is permission, not a
+    request — growing on permission would put workers on a lane with nothing
+    to do — so the sizing pass spends it on its next tick if there is work.
+    That also makes the common case (raising a cap) free: no broker round trip
+    AT ALL, which is the difference between a control that answers instantly
+    and one that takes ten seconds. Keyed on the previous cap rather than on
+    "is it on" — the first cut only knew on/off, so it inspected on every
+    raise to find out whether the pool needed lowering, and the control it was
+    meant to make instant still waited out an inspect.
+
+    A failed push is not a failed setting. The value is already stored and the
+    sizing pass carries it within a minute; `applied: false` with a reason
+    lets the UI say "saved, not yet live" rather than "that didn't work"
+    (lesson #4202 — a live change that does not survive, with nothing saying
+    so).
+    """
+    was_on, now_on = was_cap > 0, slots_cap > 0
     applied, error = True, None
     if now_on != was_on:
         applied, error = await asyncio.to_thread(set_lane_enabled_sync, lane, now_on)
@@ -536,9 +588,10 @@ async def set_lane(session: AsyncSession, lane: Lane, *, slots_cap: int) -> dict
         applied, error = await asyncio.to_thread(
             set_lane_slots_sync, lane, MIN_POOL_SLOTS,
         )
-    elif applied and now_on:
+    elif applied and now_on and slots_cap < was_cap:
-        # Only DOWNWARD. See the docstring: raising a cap is permission, and
+        # LOWERED on a running lane. Only this direction needs a message, and
-        # the sizing pass decides whether there is work to spend it on.
+        # only when the pool is actually above the new cap — so it reads the
+        # live pool rather than resizing blind. A raise never reaches here.
         live = await asyncio.to_thread(inspect_lanes_sync)
         current = live[lane.name].pool
         if current is not None and current > slots_cap:
@@ -561,8 +614,8 @@ async def set_lane(session: AsyncSession, lane: Lane, *, slots_cap: int) -> dict
 
     return {
         "name": lane.name,
-        "slots_cap": row.slots_cap,
+        "slots_cap": slots_cap,
-        "ceiling": ceiling,
+        "ceiling": derived_ceiling(lane),
         "enabled": now_on,
         "applied": applied,
         "apply_error": error,

tests/test_api_workers.py

@@ -255,3 +255,81 @@ async def _refreshed(db, name: str, expected: int) -> None:
     row = await _lane_row(db, name)
     await db.refresh(row)
     assert row.slots_cap == expected, "a refused write must store nothing"
+
+
+# --- no database connection is held across a broker round trip ---------------
+
+
+@pytest.mark.asyncio
+async def test_the_lane_read_holds_no_session_while_it_inspects(monkeypatch):
+    """Operator, 2026-09-23: *"something about changing the cap number is
+    blocking to the website... it shouldn't be"*.
+
+    Nothing here was slow in itself. A Postgres connection was held across a
+    celery inspect whose budget is eleven seconds, on a page that polls every
+    fifteen — so with a lane not answering, each poll pinned a connection for
+    most of the interval. SQLAlchemy's default pool is five plus ten overflow;
+    two browser tabs, the health endpoint doing the same, and a cap change
+    adding more inspects exhausts it, and every OTHER request then waits on a
+    connection. It surfaced as the whole site stalling rather than as one slow
+    page, which is why it took a screenshot to find.
+
+    Asserted STRUCTURALLY rather than by timing: `lane_view` must not accept a
+    session at all. A timing test would be flaky, and a mock-call-order test
+    would pass against a version that took the session and merely used it
+    early — the property that matters is that it CANNOT.
+    """
+    import inspect as _inspect
+
+    from backend.app.services.worker_control import lane_settings, lane_view
+
+    assert "session" not in _inspect.signature(lane_view).parameters, (
+        "lane_view takes a session again; the broker work must run with none held"
+    )
+    # And the DB half still exists, so the split did not simply lose the reads.
+    assert "session" in _inspect.signature(lane_settings).parameters
+
+
+@pytest.mark.asyncio
+async def test_the_cap_write_holds_no_session_while_it_pushes():
+    """The same property on the write path, where it was worse: a cap change
+    could make three broker round trips, each with a connection held."""
+    import inspect as _inspect
+
+    from backend.app.services.worker_control import push_lane_cap, store_lane_cap
+
+    assert "session" in _inspect.signature(store_lane_cap).parameters
+    assert "session" not in _inspect.signature(push_lane_cap).parameters, (
+        "push_lane_cap takes a session again; the push must run with none held"
+    )
+
+
+@pytest.mark.asyncio
+async def test_raising_a_cap_costs_no_broker_round_trip_at_all(
+    client, db, no_live_workers, monkeypatch,
+):
+    """The common case must be instant. Raising a cap is permission, not a
+    request — the sizing pass spends it — so there is nothing to tell the
+    broker, and the operator's `+` should answer immediately rather than
+    waiting out an inspect."""
+    from backend.app.services import worker_control as wc
+
+    calls = []
+    monkeypatch.setattr(
+        wc, "inspect_lanes_sync", lambda: calls.append("inspect") or {},
+    )
+    monkeypatch.setattr(
+        wc, "set_lane_slots_sync",
+        lambda *a, **k: calls.append("resize") or (True, None),
+    )
+    monkeypatch.setattr(
+        wc, "set_lane_enabled_sync",
+        lambda *a, **k: calls.append("consumers") or (True, None),
+    )
+
+    await client.post("/api/system/workers/worker", json={"slots_cap": 1})
+    calls.clear()
+    resp = await client.post("/api/system/workers/worker", json={"slots_cap": 6})
+
+    assert resp.status_code == 200
+    assert calls == [], f"raising a cap talked to the broker: {calls}"