From 48108a356923ba4795b37c35cd4906a46abb8745 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Wed, 23 Sep 2026 15:21:21 -0400 Subject: [PATCH] fix: a lane that is OFF was not attributable to itself (4295) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Operator, 2026-09-23: *"clean up the stale service_seen rows"*. **They were not stale.** They were phantoms, written on purpose, and they will come back on every install that turns a lane off — so the rows are the smaller half of this. A celery worker was attributed to its lane by the queues it was CONSUMING. A lane at cap 0 has its consumers cancelled, so it answers `active_queues()` with an empty list, matches no lane, and is dropped. Three consequences, all on the operator's screen at once: 1. The lanes table reported the lane **not answering** — the signal for a crashed worker, not for one the operator turned off. 2. The roster grew a phantom row named **`Worker ()`** — the empty queue set rendered as a display name — shown "running" beside the real lane's row going stale, because nothing updated it any more. 3. **The container went unhealthy.** `healthcheck._lanes_ok` requires every lane present. ML ships at cap 0, so a fresh install was permanently unhealthy and Swarm restarts an unhealthy task forever. That third one is the severe one, and its docstring asserted the opposite of what the code did — *"a disabled lane still runs its process with its consumers cancelled, so it answers inspect and is healthy"*. It answers. It was not attributed. A comment can be right about the intent and wrong about the program, and this one had been wrong since the consolidated container shipped. `worker_lanes.lane_for_node` attributes by NODE NAME instead: identity travels with the process rather than with what it happens to be doing. `gen_supervisord` already sets `CELERY_NODENAME={lane.name}` per program — the information was there and nothing read it. Falls back to the queue set for a deployment that names no node, and `docker-compose.yml` now sets one per service so the multi-service stack gets it too. The roster keys on the LANE's queue set when the node resolves, which is the same string the row already had while it was consuming — so an existing row keeps updating rather than a second one appearing. Migration 0106 deletes the one key the bug produced, `celery:`. Deliberately NOT a retention sweep: the roster never forgets on purpose, so a quiet row is what it is FOR, and only a row that cannot correspond to anything real is safe to remove. An `agent:agent` row, if one exists, is left alone — nothing here can tell an abandoned agent id from a second agent that is genuinely down, and hiding a dead GPU agent is the one thing the roster must not do. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01LVjrnpQjRgHdvq95rASoiR --- .../0106_prune_phantom_roster_rows.py | 70 ++++++++++++ backend/app/scripts/healthcheck.py | 16 ++- backend/app/services/service_roster.py | 17 ++- backend/app/services/worker_control.py | 9 +- backend/app/services/worker_lanes.py | 38 +++++++ docker-compose.yml | 15 +++ tests/test_service_roster.py | 77 +++++++++++++ tests/test_worker_control.py | 101 ++++++++++++++++++ 8 files changed, 336 insertions(+), 7 deletions(-) create mode 100644 alembic/versions/0106_prune_phantom_roster_rows.py diff --git a/alembic/versions/0106_prune_phantom_roster_rows.py b/alembic/versions/0106_prune_phantom_roster_rows.py new file mode 100644 index 0000000..14d24d2 --- /dev/null +++ b/alembic/versions/0106_prune_phantom_roster_rows.py @@ -0,0 +1,70 @@ +"""service_seen — delete the roster rows the fixed code can no longer write. + +Operator, 2026-09-23: *"clean up the stale service_seen rows"*. They were not +stale. They were PHANTOMS, written on purpose by code that identified a celery +worker from the queues it was consuming. + +A lane at cap 0 has its consumers cancelled, so it answers `active_queues()` +with an empty list. The roster grouped on that empty set, wrote it under the +key `celery:` and rendered `role_display_name(())` as the display name — a row +called **`Worker ()`**, reported as running, beside the real lane's row going +stale because nothing updated it any more. + +`worker_lanes.lane_for_node` fixes the cause: a worker is attributed by its +NODE NAME, which survives having no consumers. Nothing will write `celery:` +again. + +## Why a migration and not a retention sweep + +Lesson #4202: a guard that refuses to produce a bad value does not undo the +bad value already stored. The row is the thing that has to change. + +And it must be deleted rather than aged out, because the roster deliberately +NEVER forgets — *"anything that has run at least once stays listed, that is +what lets a stopped one be noticed rather than simply vanishing"*. A row that +merely goes quiet is exactly what the roster is for. Only a row that cannot +correspond to anything real is safe to remove, and `celery:` is precisely +that: the empty queue set, which no correctly-attributed worker can produce. + +## What is deliberately NOT deleted + +**Celery rows with a real but unmatched queue set.** A deployment slicing +`CELERY_QUEUES` differently is supported and its rows are true. It is not this +migration's business to decide that somebody else's worker is obsolete. + +**Agent rows, including a possible `agent:agent` from a build that omitted +`agent_id`.** Nothing here can tell an abandoned agent id from a second agent +that is currently down, and deleting a real one would hide a genuinely dead +GPU agent — the one thing the roster exists to show. If such a row is present +it needs a person to look at it, not a migration guessing. + +Revision ID: 0106 +Revises: 0105 +Create Date: 2026-09-23 + +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "0106" +down_revision: Union[str, None] = "0105" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + # Exactly the one key the empty queue set produced. Matched literally + # rather than by a LIKE or a prefix: `celery:` with nothing after it is + # the phantom, and `celery:ml` is a real lane. + op.execute( + sa.text("DELETE FROM service_seen WHERE key = :key").bindparams(key="celery:") + ) + + +def downgrade() -> None: + # Nothing. The row carried no information — an empty queue set and a + # timestamp — and the roster re-learns anything real on its next refresh. + # Re-creating it would put a phantom back. + pass diff --git a/backend/app/scripts/healthcheck.py b/backend/app/scripts/healthcheck.py index 78b07e2..fcbaf52 100644 --- a/backend/app/scripts/healthcheck.py +++ b/backend/app/scripts/healthcheck.py @@ -121,11 +121,19 @@ def _this_node_ok() -> tuple[bool, str]: def _lanes_ok() -> tuple[bool, str]: """Every lane in the table is answering. - Deliberately ignores whether a lane is ENABLED: a disabled lane still runs - its process with its consumers cancelled, so it answers `inspect` and is + Deliberately ignores whether a lane is ON: a lane at cap 0 still runs its + process with its consumers cancelled, so it answers `inspect` and is healthy. Health is "is the process alive"; whether it should be consuming - is a settings question the reconcile owns, and conflating them would make - turning a lane off in the UI mark the container unhealthy. + is a settings question the sizing pass owns, and conflating them would + make turning a lane off mark the container unhealthy. + + That was not merely a risk — it was happening. Until 2026-09-23 a worker + was attributed to its lane by the queues it was CONSUMING, and a lane with + its consumers cancelled reports none, so it read as absent and this check + failed. ML ships at cap 0, so a fresh install was permanently unhealthy + and Swarm restarts an unhealthy task forever. The docstring above said the + right thing while the code did the opposite; `worker_lanes.lane_for_node` + is what makes it true. """ from ..services.worker_control import inspect_lanes_sync from ..services.worker_lanes import LANES diff --git a/backend/app/services/service_roster.py b/backend/app/services/service_roster.py index 83de7db..7a77363 100644 --- a/backend/app/services/service_roster.py +++ b/backend/app/services/service_roster.py @@ -36,7 +36,7 @@ from sqlalchemy.dialects.postgresql import insert as pg_insert from sqlalchemy.ext.asyncio import AsyncSession from ..models import ServiceSeen -from .worker_lanes import LANES +from .worker_lanes import LANES, lane_for_node log = logging.getLogger(__name__) @@ -118,7 +118,20 @@ def _inspect_celery_sync() -> dict[tuple[str, ...], dict]: grouped: dict[tuple[str, ...], dict] = {} for hostname, queues in active_queues.items(): - key = tuple(sorted({q["name"] for q in queues})) + # Keyed on the LANE's queue set when the node name identifies one, so + # a lane keeps the same roster row whether or not it is consuming. + # + # Grouping on the ACTIVE queues alone meant a lane at cap 0 — which + # cancels its consumers — reported an empty set, landed under the key + # `celery:`, and rendered as a phantom row named `Worker ()` while its + # real row went stale beside it. Both symptoms on the operator's + # screen, 2026-09-23, from this one line. + # + # Deriving the key from `lane.queue_key` rather than inventing a new + # one keeps every existing row: it is the same string the lane already + # had while it was running. + lane = lane_for_node(hostname) + key = lane.queue_key if lane else tuple(sorted({q["name"] for q in queues})) entry = grouped.setdefault(key, {"hostnames": [], "active": 0}) entry["hostnames"].append(hostname) entry["active"] += len(active_tasks.get(hostname, [])) diff --git a/backend/app/services/worker_control.py b/backend/app/services/worker_control.py index 88d599f..c4c1d7d 100644 --- a/backend/app/services/worker_control.py +++ b/backend/app/services/worker_control.py @@ -62,6 +62,7 @@ from .worker_lanes import ( MIN_POOL_SLOTS, Lane, derived_ceiling, + lane_for_node, ) log = logging.getLogger(__name__) @@ -189,7 +190,13 @@ def inspect_lanes_sync() -> dict[str, LaneLiveState]: return out for hostname, queues in active_queues.items(): - lane = _lane_for_queues(tuple(q["name"] for q in queues)) + # The NODE NAME first — see `lane_for_node`. A lane at cap 0 has its + # consumers cancelled and answers with an empty queue list, which + # matches no lane, so attributing by queues alone dropped every lane + # the operator had turned off and reported it as "not answering". + lane = lane_for_node(hostname) or _lane_for_queues( + tuple(q["name"] for q in queues) + ) if lane is None: # A deployment slicing CELERY_QUEUES differently. Reported by the # roster under its raw queue list; it simply has no lane row to diff --git a/backend/app/services/worker_lanes.py b/backend/app/services/worker_lanes.py index 7694696..8a23513 100644 --- a/backend/app/services/worker_lanes.py +++ b/backend/app/services/worker_lanes.py @@ -332,6 +332,44 @@ def container_cpu_count() -> int | None: return os.cpu_count() +def lane_for_node(hostname: str) -> Lane | None: + """`ml@7f3c9a1b` -> the ml lane. None for a node this build did not name. + + ## Why the node name, and not the queues it is consuming + + Because a lane that is OFF is consuming nothing, and "nothing" identifies + no lane at all. + + Both the roster and `inspect_lanes_sync` used to map a worker to its lane + through `active_queues()`. That is exact while the lane is running and + useless the moment it is not: a lane at cap 0 has its consumers cancelled, + so it answers the broadcast with an EMPTY queue list, matches no lane, and + is dropped. Three things followed, and the operator saw all three at once + on 2026-09-23: + + 1. The lanes table showed the lane as **not answering** — which is the + signal for a crashed worker, not for one the operator turned off. + 2. The roster grew a phantom row called **`Worker ()`**, the empty queue + set rendered as a display name, "running" beside the real lane's row + going stale. + 3. **The container went unhealthy.** `healthcheck._lanes_ok` requires + every lane in the table to be present, and its docstring asserted the + opposite of what the code did — *"a disabled lane still runs its + process with its consumers cancelled, so it answers inspect and is + healthy"*. It answers; it is not attributed. ML ships at cap 0, so a + fresh install would have been permanently unhealthy, and Swarm + restarts an unhealthy task forever. + + The node name survives all of that: `gen_supervisord` sets + `CELERY_NODENAME={lane.name}` per program and the entrypoint passes it to + `celery -n`, so the identity travels with the PROCESS rather than with + what it happens to be doing. Falls back to the queue set for a deployment + that sets no node name — the multi-service compose stack, where every node + is `celery@`. + """ + return LANES_BY_NAME.get(hostname.split("@", 1)[0]) + + def derived_ceiling(lane: Lane) -> int: """The most slots `lane` may be given on this container. diff --git a/docker-compose.yml b/docker-compose.yml index 2248734..671dca0 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -180,6 +180,10 @@ services: environment: <<: *app_env CELERY_QUEUES: default,import,thumbnail,download + # Names the celery node for the roster. A lane whose consumers are + # cancelled reports no queues, so the NODE is the only thing left that + # identifies it — see services/worker_lanes.lane_for_node. + CELERY_NODENAME: worker CELERY_CONCURRENCY: "2" # /downloads dropped — nothing in the app references it (operator-flagged # 2026-06-07: it wasn't mapped in prod and everything worked). @@ -200,6 +204,10 @@ services: environment: <<: *app_env CELERY_QUEUES: maintenance,scan + # Names the celery node for the roster. A lane whose consumers are + # cancelled reports no queues, so the NODE is the only thing left that + # identifies it — see services/worker_lanes.lane_for_node. + CELERY_NODENAME: scheduler volumes: - ./images:/images - ./import:/import @@ -223,6 +231,10 @@ services: environment: <<: *app_env CELERY_QUEUES: maintenance_long + # Names the celery node for the roster. A lane whose consumers are + # cancelled reports no queues, so the NODE is the only thing left that + # identifies it — see services/worker_lanes.lane_for_node. + CELERY_NODENAME: maintenance_long CELERY_CONCURRENCY: "1" # Only /images: backups write to /images/_backups, audits read /images, and # the admin tasks (re-extract/cascade-delete/normalize) operate on /images. @@ -241,6 +253,9 @@ services: deploy: *deploy_policy environment: <<: *app_env + # See the worker service — the node name is what identifies a lane + # whose consumers are cancelled. + CELERY_NODENAME: ml volumes: - ./images:/images:ro - ./models:/models diff --git a/tests/test_service_roster.py b/tests/test_service_roster.py index 372bee4..c837297 100644 --- a/tests/test_service_roster.py +++ b/tests/test_service_roster.py @@ -52,3 +52,80 @@ def test_the_round_trip_count_matches_the_calls_actually_made(): f"_inspect_celery_sync makes {calls} inspect calls but " f"INSPECT_ROUND_TRIPS says {sr.INSPECT_ROUND_TRIPS}" ) + + +# --- a lane that is off keeps its own row ------------------------------------ + + +def _stub_inspect(monkeypatch, active_queues): + class _Insp: + def __init__(self, **_): + pass + + def active_queues(self): + return active_queues + + def active(self): + return {} + + class _Control: + inspect = _Insp + + import sys + import types + mod = types.ModuleType("backend.app.celery_app") + + class _C: + pass + + c = _C() + c.control = _Control() + mod.celery = c + monkeypatch.setitem(sys.modules, "backend.app.celery_app", mod) + + +def test_a_lane_with_no_consumers_keeps_the_row_it_had_while_running(monkeypatch): + """The phantom, and the reason the operator had two wrong rows at once. + + A lane at cap 0 cancels its consumers, so it answers with an EMPTY queue + list. Grouped on that, it landed under the key `celery:` and rendered as a + row called `Worker ()` — reported running — while the real lane's row went + stale beside it because nothing updated it any more. + + Keyed on the LANE's queue set now, which is the same string the row + already had while the lane was consuming. So turning a lane off updates + its row instead of minting a second one. + """ + from backend.app.services.worker_lanes import LANES_BY_NAME + + _stub_inspect(monkeypatch, {"ml@abc123": []}) + + grouped = sr._inspect_celery_sync() + + assert list(grouped) == [LANES_BY_NAME["ml"].queue_key] + assert () not in grouped, "the empty queue set is the phantom `Worker ()`" + + +def test_the_row_is_the_same_one_whether_the_lane_is_consuming_or_not(monkeypatch): + """Stated as an identity rather than as two separate assertions: if these + keys ever differ, turning a lane off silently starts a second roster row + and the first goes stale — which is exactly what happened.""" + ml_queues = [{"name": "ml"}] + + _stub_inspect(monkeypatch, {"ml@abc123": ml_queues}) + on = set(sr._inspect_celery_sync()) + _stub_inspect(monkeypatch, {"ml@abc123": []}) + off = set(sr._inspect_celery_sync()) + + assert on == off + + +def test_a_worker_this_build_did_not_name_is_still_grouped_by_its_queues( + monkeypatch, +): + """The fallback, and the case the roster exists to report honestly: a + deployment slicing CELERY_QUEUES differently gets its raw queue list + rather than a name this code invented for it.""" + _stub_inspect(monkeypatch, {"celery@xyz": [{"name": "odd"}]}) + + assert list(sr._inspect_celery_sync()) == [("odd",)] diff --git a/tests/test_worker_control.py b/tests/test_worker_control.py index c98133d..8febdb9 100644 --- a/tests/test_worker_control.py +++ b/tests/test_worker_control.py @@ -570,3 +570,104 @@ def test_the_sizing_task_is_registered_and_scheduled(): tasks = {e["task"] for e in entries.values()} assert "backend.app.tasks.maintenance.reconcile_worker_lanes" not in tasks assert "backend.app.tasks.maintenance.autoscale_worker_lanes" not in tasks + + +# --- a lane that is OFF is still its own lane --------------------------------- +# +# The operator's screen, 2026-09-23: "ML tagging has not checked in for 135 +# min" beside a phantom row called "Worker ()" reported as running. One cause: +# a worker was attributed to its lane by the queues it was CONSUMING, and a +# lane at cap 0 has its consumers cancelled, so it consumes none and matched +# nothing. + + +def _stub_active_queues(monkeypatch, by_host): + """Only `active_queues` — the read that decides which lane a node IS.""" + class _Insp: + def __init__(self, **_): + pass + + def active_queues(self): + return by_host + + def stats(self): + return {h: {"pool": {"max-concurrency": 1}} for h in by_host} + + def active(self): + return {} + + def reserved(self): + return {} + + control = _stub_control(monkeypatch) + control.inspect = _Insp + import sys + import types + mod = types.ModuleType("backend.app.celery_app") + + class _C: + pass + + c = _C() + c.control = control + mod.celery = c + monkeypatch.setitem(sys.modules, "backend.app.celery_app", mod) + + +def test_a_lane_with_its_consumers_cancelled_is_still_found(monkeypatch): + """THE bug. `ml@host` consuming nothing must read as the ml lane, present, + rather than as no lane at all. + + Attributed by the NODE NAME, which survives having no consumers — + `gen_supervisord` sets `CELERY_NODENAME={lane.name}` per program for + exactly this.""" + _stub_active_queues(monkeypatch, {"ml@abc123": []}) + + live = wc.inspect_lanes_sync() + + assert live["ml"].present is True + assert live["ml"].consuming == set(), "it is present AND consuming nothing" + + +def test_that_is_what_keeps_the_container_healthy(monkeypatch): + """Why it mattered more than a cosmetic row. + + `healthcheck._lanes_ok` requires every lane to be present, and ML ships at + cap 0 — so a fresh install reported a lane not answering, the container + went permanently unhealthy, and Swarm restarts an unhealthy task forever. + """ + _stub_active_queues(monkeypatch, { + "worker@a": [{"name": q} for q in LANES_BY_NAME["worker"].queues], + "scheduler@a": [{"name": q} for q in LANES_BY_NAME["scheduler"].queues], + "maintenance_long@a": [ + {"name": q} for q in LANES_BY_NAME["maintenance_long"].queues + ], + "ml@a": [], # off, as it ships + }) + + live = wc.inspect_lanes_sync() + + missing = sorted(name for name, s in live.items() if not s.present) + assert missing == [], f"the healthcheck would fail the container: {missing}" + + +def test_a_node_this_build_did_not_name_still_matches_on_its_queues(monkeypatch): + """The fallback. The multi-service compose stack ran every worker as + `celery@`, and a deployment that sets no node name must keep + working — the node name is an improvement, not a requirement.""" + _stub_active_queues(monkeypatch, { + "celery@xyz": [{"name": q} for q in LANES_BY_NAME["worker"].queues], + }) + + assert wc.inspect_lanes_sync()["worker"].present is True + + +def test_an_unknown_worker_is_still_ignored_rather_than_guessed_at(monkeypatch): + """A deployment slicing CELERY_QUEUES differently has no lane row to + control. It belongs to the roster, not here — and must not be attributed + to whichever lane happens to be first.""" + _stub_active_queues(monkeypatch, {"celery@xyz": [{"name": "something-else"}]}) + + live = wc.inspect_lanes_sync() + + assert all(not s.present for s in live.values())