Release: dev → main (first public release) #258
@@ -51,6 +51,29 @@ REFRESH_TTL_SECONDS = 20.0
|
|||||||
# page that exists to explain it.
|
# page that exists to explain it.
|
||||||
INSPECT_TIMEOUT_SECONDS = 2.0
|
INSPECT_TIMEOUT_SECONDS = 2.0
|
||||||
|
|
||||||
|
# How many broadcast round trips `_inspect_celery_sync` makes. Named, because
|
||||||
|
# the wrapper's budget is derived from it and the two must not drift.
|
||||||
|
#
|
||||||
|
# `active_queues()` and `active()` are separate broadcasts, and a broadcast
|
||||||
|
# with no `destination` cannot know how many replies to expect — so each one
|
||||||
|
# waits out its full timeout rather than returning on the last reply. The sync
|
||||||
|
# call therefore costs ~2 x INSPECT_TIMEOUT_SECONDS in the ordinary case, not
|
||||||
|
# once.
|
||||||
|
INSPECT_ROUND_TRIPS = 2
|
||||||
|
|
||||||
|
# Slack for the thread handoff. `asyncio.to_thread` hands work to the default
|
||||||
|
# executor, and on a loaded web process — the operator's showcase page pulling
|
||||||
|
# ninety thumbnails a second — the thread may not even be scheduled inside the
|
||||||
|
# budget, let alone finish.
|
||||||
|
#
|
||||||
|
# This exists because the wrapper used to allow `INSPECT_TIMEOUT_SECONDS * 2`,
|
||||||
|
# which LOOKS like a safety factor and is exactly the worst case with nothing
|
||||||
|
# left over. Observed on the operator's first consolidated deploy, 2026-09-23:
|
||||||
|
# a TimeoutError traceback per refresh while the two inspect calls were
|
||||||
|
# working perfectly. A budget equal to the work is a budget that fails under
|
||||||
|
# any load at all.
|
||||||
|
INSPECT_SLACK_SECONDS = 3.0
|
||||||
|
|
||||||
# Queue set -> the name an operator recognises. Sorted-tuple keys, because the
|
# Queue set -> the name an operator recognises. Sorted-tuple keys, because the
|
||||||
# order celery reports them in is not guaranteed.
|
# order celery reports them in is not guaranteed.
|
||||||
#
|
#
|
||||||
@@ -86,6 +109,10 @@ def _inspect_celery_sync() -> dict[tuple[str, ...], dict]:
|
|||||||
from ..celery_app import celery as celery_app
|
from ..celery_app import celery as celery_app
|
||||||
|
|
||||||
insp = celery_app.control.inspect(timeout=INSPECT_TIMEOUT_SECONDS)
|
insp = celery_app.control.inspect(timeout=INSPECT_TIMEOUT_SECONDS)
|
||||||
|
# TWO broadcasts, each waiting out its own timeout — see
|
||||||
|
# INSPECT_ROUND_TRIPS, which the caller's budget is derived from. Adding a
|
||||||
|
# third call here without updating that constant puts the wrapper back
|
||||||
|
# under the work it is waiting for.
|
||||||
active_queues = insp.active_queues() or {}
|
active_queues = insp.active_queues() or {}
|
||||||
active_tasks = insp.active() or {}
|
active_tasks = insp.active() or {}
|
||||||
|
|
||||||
@@ -137,7 +164,10 @@ async def refresh_celery_roster(session: AsyncSession) -> None:
|
|||||||
try:
|
try:
|
||||||
grouped = await asyncio.wait_for(
|
grouped = await asyncio.wait_for(
|
||||||
asyncio.to_thread(_inspect_celery_sync),
|
asyncio.to_thread(_inspect_celery_sync),
|
||||||
timeout=INSPECT_TIMEOUT_SECONDS * 2,
|
timeout=(
|
||||||
|
INSPECT_TIMEOUT_SECONDS * INSPECT_ROUND_TRIPS
|
||||||
|
+ INSPECT_SLACK_SECONDS
|
||||||
|
),
|
||||||
)
|
)
|
||||||
except Exception:
|
except Exception:
|
||||||
log.warning("service roster: celery inspect failed; roster not refreshed", exc_info=True)
|
log.warning("service roster: celery inspect failed; roster not refreshed", exc_info=True)
|
||||||
|
|||||||
@@ -0,0 +1,54 @@
|
|||||||
|
"""The roster's inspect budget must exceed the inspect work.
|
||||||
|
|
||||||
|
Both tests here assert a RELATION between constants rather than their values.
|
||||||
|
The failure they exist for is not a wrong number — it is the budget and the
|
||||||
|
work drifting apart, which is invisible in each one read on its own.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from backend.app.services import service_roster as sr
|
||||||
|
|
||||||
|
|
||||||
|
def test_the_wrapper_budget_exceeds_the_work_it_waits_for():
|
||||||
|
"""The bug the operator's first consolidated deploy surfaced (2026-09-23).
|
||||||
|
|
||||||
|
`_inspect_celery_sync` makes INSPECT_ROUND_TRIPS broadcasts, and a
|
||||||
|
broadcast with no `destination` cannot know how many replies to expect, so
|
||||||
|
each waits out its full timeout rather than returning on the last reply.
|
||||||
|
The sync call therefore costs round_trips x INSPECT_TIMEOUT_SECONDS.
|
||||||
|
|
||||||
|
The wrapper allowed `INSPECT_TIMEOUT_SECONDS * 2`, which reads like a
|
||||||
|
safety factor and is exactly that worst case with nothing left over. Live
|
||||||
|
result: a TimeoutError traceback on every refresh while the two inspect
|
||||||
|
calls were working perfectly, and a roster that stopped advancing.
|
||||||
|
|
||||||
|
Asserted as a RELATION, not as a number. A third inspect call added to the
|
||||||
|
sync function is the way this silently comes back, and the only thing that
|
||||||
|
keeps the two honest is deriving one from the other.
|
||||||
|
"""
|
||||||
|
budget = (
|
||||||
|
sr.INSPECT_TIMEOUT_SECONDS * sr.INSPECT_ROUND_TRIPS
|
||||||
|
+ sr.INSPECT_SLACK_SECONDS
|
||||||
|
)
|
||||||
|
work = sr.INSPECT_TIMEOUT_SECONDS * sr.INSPECT_ROUND_TRIPS
|
||||||
|
assert budget > work, (
|
||||||
|
f"the wrapper allows {budget}s for {work}s of broadcasts — a budget "
|
||||||
|
f"equal to the work fails under any load at all"
|
||||||
|
)
|
||||||
|
assert sr.INSPECT_SLACK_SECONDS > 0
|
||||||
|
|
||||||
|
|
||||||
|
def test_the_round_trip_count_matches_the_calls_actually_made():
|
||||||
|
"""INSPECT_ROUND_TRIPS is only true if someone keeps it true, so read the
|
||||||
|
source rather than trusting the constant: the budget above is derived from
|
||||||
|
it, and a call added without updating it puts the wrapper back under the
|
||||||
|
work."""
|
||||||
|
import inspect as _inspect
|
||||||
|
|
||||||
|
src = _inspect.getsource(sr._inspect_celery_sync)
|
||||||
|
calls = src.count("insp.")
|
||||||
|
assert calls == sr.INSPECT_ROUND_TRIPS, (
|
||||||
|
f"_inspect_celery_sync makes {calls} inspect calls but "
|
||||||
|
f"INSPECT_ROUND_TRIPS says {sr.INSPECT_ROUND_TRIPS}"
|
||||||
|
)
|
||||||
Reference in New Issue
Block a user