fix: kick() ignores routing label when _task_queues is narrowed to 1 by with_queues()

[NicolasFerec]

Problem

The multi-queue feature introduced in 0.6.0 encourages a pattern where one broker is configured with all queues, and worker entry-point factory functions use with_queues() to narrow which queue each worker process listens to:

broker = AioPikaBroker(
    "amqp://...",
    task_queues=[queue_default, queue_mail, queue_heavy],
)

@broker.task                                    # routes to queue_default
async def default_task(): ...

@broker.task(queue_name="queue_mail")           # routes to queue_mail
async def send_email(): ...

def get_worker_default():
    return broker.with_queues(queue_default)    # worker only listens to queue_default

def get_worker_mail():
    return broker.with_queues(queue_mail)       # worker only listens to queue_mail

Workers are started with:

taskiq worker ... -fsd myapp.broker:get_worker_default
taskiq worker ... -fsd myapp.broker:get_worker_mail

Bug

When default_task runs inside the get_worker_default worker and calls await send_email.kiq(), the message is silently routed to queue_default instead of queue_mail.

The root cause is in kick():

if len(self._task_queues) == 1:
    # ❌ Hardcodes the routing key to the single queue's name,
    #    completely ignoring any queue_name label on the message.
    routing_key_name = (
        self._task_queues[0].routing_key or self._task_queues[0].name
    )
else:
    routing_key_name = (
        parse_val(str, message.labels.get(self._label_for_routing)) or ""
    )

After with_queues(queue_default) narrows _task_queues to 1 item, every outgoing kick() call hits the shortcut and routes to queue_default — the queue_name="queue_mail" label is never read.

On a TOPIC exchange this produces no error; the message is simply delivered to the wrong queue.

Fix

Check the routing label first. Only fall back to the single-queue shortcut when no explicit label is present (preserving full backward compatibility for single-queue brokers that don't annotate their tasks):

routing_key_from_label = parse_val(str, message.labels.get(self._label_for_routing))
if routing_key_from_label:
    routing_key_name = routing_key_from_label
    if self._exchange.type == ExchangeType.DIRECT and routing_key_name not in {
        queue.routing_key or queue.name for queue in self._task_queues
    }:
        raise IncorrectRoutingKeyError(...)
elif len(self._task_queues) == 1:
    # Backward-compatible: single-queue brokers don't need a queue_name label.
    routing_key_name = (
        self._task_queues[0].routing_key or self._task_queues[0].name
    )
else:
    routing_key_name = ""

Also updated the with_queues() docstring to clarify its intended role (narrowing the listening scope only) vs. its unintended side-effect on kick routing that this PR removes.

Impact

• ✅ Fully backward compatible — brokers with a single queue and no queue_name labels are unaffected (they never set the label, so they always hit the elif len == 1 path)
• ✅ Fixes cross-queue kiq() calls from inside a narrowed worker
• ✅ DIRECT exchange validation still present for labelled messages

[Copilot]

Pull request overview

Fixes a routing bug in AioPikaBroker.kick() where an explicit queue_name label was ignored when with_queues() narrowed the broker to a single queue, causing messages to be published to the wrong routing key.

Changes:

• Prioritize the routing label in kick() and only fall back to the single-queue shortcut when no label is present.
• Expand with_queues() docstring to describe the worker-narrowing usage pattern and how routing is determined.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

The updated with_queues() docstring claims the "full broker (with all queues) is still used for publishing via kick()" and that this method "only affects which queues the worker listens to". In the current implementation with_queues() replaces self._task_queues, which is also used during startup() for queue declaration/binding and in kick() for the single-queue fallback and DIRECT routing validation. Please reword the docstring to match the actual behavior, or (if the intent is truly listen-only) split the internal state so narrowing only affects consumption while publishing/validation still sees the full queue set.

[Copilot]

With ExchangeType.DIRECT, the new label-first behavior will now raise IncorrectRoutingKeyError whenever with_queues() has narrowed _task_queues and a message is labelled for a routing key outside that narrowed set. That prevents cross-queue publishing from a narrowed worker even when the target queue exists/bound elsewhere (and is a behavior change from the previous silent misroute). Consider validating against the broker’s full configured queue set (separate from the listen set), or relaxing/adjusting this validation when _task_queues has been narrowed for listening.

[Copilot]

No test currently covers the regression scenario described in the PR: calling kick() (or task .kiq()) from a broker instance that has been narrowed to a single queue via with_queues(), while publishing a message with an explicit queue_name label for a different queue. Adding a routing test that reproduces this (and asserts the message lands in the labelled queue) would prevent this from regressing again.

[danfimov]

I think the correct approach here is to use two separate brokers: one (returned by get_worker_default) narrowed via with_queues() for listening a specific queue, and the broker that aware of all queues for publishing.

The current behavior is intentional — _task_queues is used to validate the routing key at kick() time. If the broker only knows about queue_default, it cannot guarantee that queue_name="queue_mail" is valid. This fix would silently allow publishing to an unknown routing key, which is a worse problem than the one being solved.

[NicolasFerec]

Thanks for the explanation and for taking the time to clarify this.
That makes sense. I’ll close the PR 👍