Skip to content

Topic subscription — what topics=['*'] means, and how it once meant nothing

Start here for "the workers look healthy and the queue never moves". (The other shape of that symptom — a topic wedged by an unreaped in-flight message — is in orphan-recovery.md.)

The failure it documents (#26)

On sandbox/home, fastpluggy-tasks-worker 0.3.301, RabbitMQ broker:

GET /plugin-pkg/tasks_worker/api/broker/topics   -> {"topics":[]}
GET /plugin-pkg/tasks_worker/api/broker/workers  -> 2 workers, topics=['*'], capacity=8

Broker side: track_page held 3035 messages with 0 consumers, default 13 with 0. Only the coordination queues (fp.node.<id>) had a consumer. This had been true since 2026-07-27 — 18 days — and nothing anywhere was red.

Three mechanisms had to line up, and each was individually "reasonable":

  1. RabbitMQBroker.register_worker skipped the wildcard. The loop read for t in w.topics: if t != "*": ensure_topic(t); _start_topic_consumer(t). With topics=['*'] the body never ran, so the registration started no consumer at all.
  2. TaskRunner.resolve_topics expanded '*' to [] and accepted it. It asked broker.get_topics(), got an empty list, and returned it — under the comment "to avoid spinning threads with no topic". An empty result was indistinguishable from a correct one.
  3. get_topics() could not answer. It reports only what the current process knows: its topic configs, its per-topic counters, and the topics of workers it has seen register — excluding '*'. On a cold start all three are empty, and the third stays empty because the only workers present advertise '*'.

That last point is the lock, and it is circular: the executor asks which topics exist; the answer can only contain what a non-wildcard worker already declared; there are none. In a warm cluster the configs propagate over the fanout exchange via config_request, but when every process boots together, nobody can answer.

The claim("*") path was never involved — its "Wildcard topic claiming not yet implemented" warning has zero occurrences across the whole window, because claim was never called with '*' either.

Why nothing detected it

  • Two workers, registered, alive, heartbeating, capacity 8, topics=['*']. Any check that counts workers sees a healthy cluster.
  • Worse, get_topics() counts any worker with '*' in its topics as a subscriber to every topic. Had a topic appeared, it would have reported 2 subscribers for 0 consumers.
  • The only symptom available anywhere was a queue depth that does not go down.

How it is fixed

A worker advertises exactly the topics it consumes. start_executor registers the resolved list, never the raw ['*']. The wildcard intent survives in the worker meta (wildcard_topics, requested_topics), so the UI can still say "takes any topic" without the topic list lying. ExecutorManager re-registers through TaskRunner.advertise_topics() whenever the consumed set changes at runtime, so the registration cannot freeze at its boot value.

The wildcard expands from several sources and is never empty (core/topic_discovery.py): the broker's known topics, topics declared by locally registered tasks, topics on the beat schedule table, and — always — the effective default topic as a floor. default is where every un-topiced submit lands, so a worker claiming to handle everything that skips it is simply wrong.

register_worker expands '*' instead of skipping it, and an expansion that comes out empty is logged at ERROR rather than being a silent no-op.

A zero-topic executor is loud, at boot and repeatedly. TaskExecutor.start_executor logs ERROR when it starts with no topic; ExecutorManager re-raises that alarm every 60 s for as long as the condition holds. A single line at boot is a line nobody scrolls back to after 18 days — the repetition is what makes it reachable from a log query.

What is deliberately NOT done: the management HTTP API

Asking RabbitMQ which queues exist is the obvious idea, and it is the wrong one here. AMQP 0-9-1 has no "list queues" operation — only the management HTTP API can enumerate them, and that API is not available in production: the home user is declared with --tags "", and the app's own logs already show Could not ensure vhost via management API: HTTP Error 401.

A discovery path built on it passes every test (a test container's guest user is an administrator) and fails on the only host that matters. Worse, it would look like it works. tests/runner/test_wildcard_topics.py pins this by failing the test if discovery attempts any HTTP call.

The honest consequence: a queue created by a process that has since died, named by no topic config, no counter, no registered worker, no task in the registry and no schedule row, cannot be discovered by this plugin. It has to be named — by the task that declares it, by a schedule, or by an explicit topics=[...] on the worker.

Operating notes

  • To pin a worker to specific queues, pass them explicitly; that path is unchanged and bypasses discovery entirely.
  • force_task_topic still overrides everything, on both publish and consume, and it is what the floor resolves to when set.
  • If you see executor is consuming ZERO topics in the logs, the worker is registered and heartbeating and will never pick up a single task. That is the alarm this whole document exists for.