Нет возможности выявить некорректное состояние канала #722

Activity

1. mosquito commented on Oct 6, 2026

   

   mosquito

   on Oct 6, 2026

   Owner

   More actions

   Please double check the latest aiormq and aio-pika, this behavior should be resolved at all.
2. lutermaks commented on Oct 7, 2026

   

   lutermaks

   on Oct 7, 2026

   Author

   More actions

   The logs show an error like: "WARNING: 2026-10-07 06:27:31 aiormq.channel: Consumer 'ctag1.f7fae0eedb4149f18875ed493572036d' cancelled by the broker on <Channel: "1" 0x7fab77db9e50 at>".
   However, close_callbacks are not triggered. Querying RabbitMQ during a node restart doesn't help either; for instance, when running:

   declare_ok = await self._channel.declare_queue(
   name=queue_name,
   passive=True,
   timeout=2
   )

   the old tag still remains in _consumers.
   Versions: aio_pika = 10.1.0, aiormq = 7.2.1
3. lutermaks commented on Oct 7, 2026

   

   lutermaks

   on Oct 7, 2026

   Author

   More actions

   Using on_consumer_cancel_callbacks at the aiormq level fixed the issue
4. mosquito commented on Oct 7, 2026

   

   mosquito

   on Oct 7, 2026

   Owner

   More actions

   Good to know, I will think about options how can it might been fixed
5. lutermaks commented on Oct 7, 2026

   

   lutermaks

   on Oct 7, 2026

   Author

   More actions

   OK, thank you
6. lutermaks

   closed this as completedon Oct 7, 2026
7. SaInekK commented on Oct 7, 2026

   

   SaInekK

   on Oct 7, 2026

   Last edited by SaInekK

   More actions

   @mosquito since you're thinking about options: I reproduced this on a local three-node cluster and tried a couple of fixes. Same root cause as #334 and #390.

   When it happens. The queue is a classic queue whose home node is not the node the client is connected to. When that node goes down (rolling restart, Kubernetes node maintenance), the broker sends basic.cancel. The connection and the channel stay open, so RobustConnection / RobustChannel have nothing to recover, and the consumer is gone until the process restarts. Quorum queues are not affected: the leader moves and the consumer stays.

   RabbitMQ 4.0, client on node 1, classic queue declared via node 2, docker stop node 2, then start it again. aio-pika 10.1.0 / aiormq 7.2.1 (9.6.2 / 6.9.4 behaves the same, just without the warning):

   classic: before        consumers=1
   aiormq.channel: Consumer 'ctag1.c9ea17bd1fbc4992880c83e8b35934ad' cancelled by the broker on <Channel: "1" at 0x10951e410>
   classic: node 2 down   consumers=0 connection_closed=False channel_closed=False
   classic: node 2 back   consumers=0 delivered=[]
   

   Calling restore() straight from the cancel callback makes it worse. While node 2 is down, the re-declare gets 404 NOT_FOUND ... process is stopped by supervisor and the broker closes the channel. RobustChannel._on_close tries restore() once, it fails with the same 404, and nothing tries again. After node 2 is back the channel is still closed: publish raises ChannelInvalidStateError: ... closed and queue.restore() raises Channel was not opened, while the connection looks healthy. @lutermaks, if your callback re-declares right away, you will probably hit this once the node stays down for more than a moment.

   What worked. On cancel, retry with a pause: declare the queue on a short-lived separate channel, close it, and only after that succeeds call restore() on the real channel. While the node is down only the throwaway channels fail. With 20 classic queues on the stopped node, all 20 consumers came back, messages were delivered, and the main channel stayed open.

   Proposal. I'd be glad to send a PR along these lines if the direction works for you: RobustChannel subscribes to the underlay channel's on_consumer_cancel_callbacks (re-registered on every reopen), and one background loop per connection probes and re-consumes only the cancelled tags, stopping when the channel or the connection closes, in line with #717. One loop rather than one per queue, because a node failure cancels every classic queue on it at once, and per-queue loops would be the channel open/close churn from mosquito/aiormq#263 (I couldn't trigger it with 20 queues, but it is cheap to avoid). The hook is aiormq's on_consumer_cancel_callbacks (mosquito/aiormq#238), which the current aiormq>=7.1.3 floor already includes.

   One question before I start: if an operator deleted the queue (no node failure), should it come back? A regular declare in the probe would recreate it; a passive one would wait until someone declares it again. For the node-down case a regular declare is harmless: in my test (durable classic queues, RabbitMQ 4.0.9, node stopped with docker stop) it got 404 NOT_FOUND and did not create the queue on another node. I haven't tried a hard kill or a partition.

   For the test suite: deleting the queue from a second channel produces the same basic.cancel, so recovery can be covered on a single broker. The cluster case stays a manual repro.

   cluster.sh, repro.py, naive_restore.py and its output

   #!/usr/bin/env bash
   # Three-node RabbitMQ 4.0 cluster: node N listens on localhost:568N.
   set -euo pipefail
   IMAGE=rabbitmq:4.0-management
   COOKIE=basic-cancel-repro
   
   case "${1:-up}" in
   up)
     docker network create cancel-repro >/dev/null
     for i in 1 2 3; do
       docker run -d --name "cr-r$i" --hostname "cr-r$i" --network cancel-repro \
         -e RABBITMQ_ERLANG_COOKIE="$COOKIE" -p "568$i:5672" "$IMAGE" >/dev/null
     done
     # -u rabbitmq: a root exec creates .erlang.cookie before the server does, and the node then crashes with eacces
     for i in 1 2 3; do
       until docker exec -u rabbitmq "cr-r$i" rabbitmq-diagnostics -q check_running >/dev/null 2>&1; do sleep 1; done
     done
     for i in 2 3; do
       docker exec -u rabbitmq "cr-r$i" rabbitmqctl -q stop_app
       docker exec -u rabbitmq "cr-r$i" rabbitmqctl -q join_cluster rabbit@cr-r1
       docker exec -u rabbitmq "cr-r$i" rabbitmqctl -q start_app
     done
     docker exec -u rabbitmq cr-r1 rabbitmqctl -q eval 'rabbit_nodes:list_running().'
     ;;
   down)
     docker rm -f cr-r1 cr-r2 cr-r3 >/dev/null 2>&1 || true
     docker network rm cancel-repro >/dev/null 2>&1 || true
     ;;
   esac

   """A consumer on a classic queue is silently lost when the queue's home node restarts.
   
   The queue is declared through node 2 (its home), the consumer connects to node 1.
   Needs cluster.sh up. Usage: python repro.py [classic|quorum]
   """
   import asyncio
   import logging
   import subprocess
   import sys
   
   import aio_pika
   
   logging.basicConfig(level=logging.WARNING, format="%(name)s: %(message)s")
   
   QUEUE = "basic-cancel-repro"
   QTYPE = sys.argv[1] if len(sys.argv) > 1 else "classic"
   
   
   def sh(*cmd: str) -> str:
       return subprocess.run(cmd, capture_output=True, text=True).stdout
   
   
   def consumers_on_broker() -> int:
       out = sh("docker", "exec", "cr-r1", "rabbitmqctl", "-q", "list_consumers", "queue_name")
       return sum(1 for line in out.splitlines() if line.strip() == QUEUE)
   
   
   async def wait_node2_running() -> None:
       for _ in range(90):
           if "rabbit@cr-r2" in sh("docker", "exec", "cr-r1", "rabbitmqctl", "-q", "eval", "rabbit_nodes:list_running()."):
               return
           await asyncio.sleep(1)
       raise RuntimeError("node 2 did not come back")
   
   
   async def main() -> None:
       args = {"x-queue-type": QTYPE}
       home = await aio_pika.connect("amqp://guest:guest@localhost:5682/")
       async with home:
           ch = await home.channel()
           await ch.queue_delete(QUEUE)
           await ch.declare_queue(QUEUE, durable=True, arguments=args)
   
       conn = await aio_pika.connect_robust("amqp://guest:guest@localhost:5681/?heartbeat=5")
       channel = await conn.channel()
       queue = await channel.declare_queue(QUEUE, durable=True, arguments=args)
       got: list[bytes] = []
   
       async def on_message(message: aio_pika.abc.AbstractIncomingMessage) -> None:
           async with message.process():
               got.append(message.body)
   
       await queue.consume(on_message)
       await asyncio.sleep(1)
       print(f"{QTYPE}: before        consumers={consumers_on_broker()}", flush=True)
   
       subprocess.run(["docker", "stop", "cr-r2"], capture_output=True)
       await asyncio.sleep(5)
       print(f"{QTYPE}: node 2 down   consumers={consumers_on_broker()} connection_closed={conn.is_closed} channel_closed={channel.is_closed}", flush=True)
   
       subprocess.run(["docker", "start", "cr-r2"], capture_output=True)
       await wait_node2_running()
       await asyncio.sleep(5)
       await channel.default_exchange.publish(aio_pika.Message(b"after"), routing_key=QUEUE)
       await asyncio.sleep(3)
       print(f"{QTYPE}: node 2 back   consumers={consumers_on_broker()} delivered={got}", flush=True)
       await conn.close()
   
   
   asyncio.run(main())

   """Naive fix: call queue.restore() straight from the basic.cancel callback (classic queue).
   
   The queue is declared through node 2 (its home), the consumer connects to node 1.
   While node 2 is down the immediate restore() hits queue.declare() -> 404; see what is left.
   Needs cluster.sh up. Usage: python naive_restore.py
   """
   import asyncio
   import logging
   import subprocess
   
   import aio_pika
   
   logging.basicConfig(level=logging.WARNING, format="%(name)s: %(message)s")
   
   QUEUE = "basic-cancel-naive-restore"
   ARGS = {"x-queue-type": "classic"}
   
   
   def sh(*cmd: str) -> str:
       return subprocess.run(cmd, capture_output=True, text=True).stdout
   
   
   def consumers_on_broker() -> int:
       out = sh("docker", "exec", "cr-r1", "rabbitmqctl", "-q", "list_consumers", "queue_name")
       return sum(1 for line in out.splitlines() if line.strip() == QUEUE)
   
   
   async def wait_node2_running() -> None:
       for _ in range(90):
           if "rabbit@cr-r2" in sh("docker", "exec", "cr-r1", "rabbitmqctl", "-q", "eval", "rabbit_nodes:list_running()."):
               return
           await asyncio.sleep(1)
       raise RuntimeError("node 2 did not come back")
   
   
   async def attempt(label: str, coro) -> None:
       try:
           await asyncio.wait_for(coro, 15)
           print(f"  {label}: ok", flush=True)
       except BaseException as e:
           print(f"  {label}: {type(e).__name__}: {e}", flush=True)
   
   
   async def main() -> None:
       home = await aio_pika.connect("amqp://guest:guest@localhost:5682/")
       async with home:
           ch = await home.channel()
           await ch.queue_delete(QUEUE)
           await ch.declare_queue(QUEUE, durable=True, arguments=ARGS)
   
       conn = await aio_pika.connect_robust("amqp://guest:guest@localhost:5681/?heartbeat=5")
       channel = await conn.channel()
       queue = await channel.declare_queue(QUEUE, durable=True, arguments=ARGS)
       got: list[bytes] = []
   
       async def on_message(message: aio_pika.abc.AbstractIncomingMessage) -> None:
           async with message.process():
               got.append(message.body)
   
       await queue.consume(on_message)
   
       restore_results: list[str] = []
   
       async def restore_now() -> None:
           try:
               await queue.restore()
               restore_results.append("ok")
           except BaseException as e:
               restore_results.append(f"{type(e).__name__}: {e}")
   
       def on_cancel(frame) -> None:
           print("  basic.cancel received -> scheduling queue.restore()", flush=True)
           asyncio.get_running_loop().create_task(restore_now())
   
       (await channel.get_underlay_channel()).on_consumer_cancel_callbacks.add(on_cancel)
       await asyncio.sleep(1)
       print(f"before        consumers={consumers_on_broker()}", flush=True)
   
       subprocess.run(["docker", "stop", "cr-r2"], capture_output=True)
       await asyncio.sleep(5)
       print(f"node 2 down   consumers={consumers_on_broker()} connection_closed={conn.is_closed} channel_closed={channel.is_closed}", flush=True)
       print(f"  immediate restore: {restore_results or 'not finished'}", flush=True)
       await attempt("publish while down", channel.default_exchange.publish(aio_pika.Message(b"down"), routing_key=QUEUE))
   
       subprocess.run(["docker", "start", "cr-r2"], capture_output=True)
       await wait_node2_running()
       await asyncio.sleep(8)
       print(f"node 2 back   consumers={consumers_on_broker()} connection_closed={conn.is_closed} channel_closed={channel.is_closed}", flush=True)
       await attempt("publish after back", channel.default_exchange.publish(aio_pika.Message(b"after"), routing_key=QUEUE))
       await attempt("explicit queue.restore()", queue.restore())
       await asyncio.sleep(3)
       print(f"  delivered={got}", flush=True)
       await conn.close()
   
   
   asyncio.run(main())

   before        consumers=1
   aiormq.channel: Consumer 'ctag1.1d56c9d693c048fd8d72c7d084ab4c12' cancelled by the broker on <Channel: "1" at 0x10a86e410>
     basic.cancel received -> scheduling queue.restore()
   aio_pika.tools: Callback <OneShotCallback: cb=<bound method RobustChannel._on_close of <aio_pika.robust_channel.RobustChannel object at 0x10a8d52b0>>> error: ChannelNotFoundEntity: NOT_FOUND - queue 'basic-cancel-naive-restore' in vhost '/' process is stopped by supervisor
   node 2 down   consumers=0 connection_closed=False channel_closed=True
     immediate restore: ["ChannelNotFoundEntity: NOT_FOUND - queue 'basic-cancel-naive-restore' in vhost '/' process is stopped by supervisor"]
     publish while down: ChannelInvalidStateError: <aio_pika.robust_channel.RobustChannel object at 0x10a8d52b0> closed
   node 2 back   consumers=0 connection_closed=False channel_closed=True
     publish after back: ChannelInvalidStateError: <aio_pika.robust_channel.RobustChannel object at 0x10a8d52b0> closed
     explicit queue.restore(): ChannelInvalidStateError: Channel was not opened
     delivered=[]
   
8. lutermaks commented on Oct 8, 2026

   

   lutermaks

   on Oct 8, 2026

   Author

   More actions

   Yes, I'm facing the exact same issue. In the end, I subscribed to on_consumer_cancel_callbacks and implemented a retry loop to re-register the consumer while the node is restarting. If exceptions like ChannelNotFoundEntity or ChannelInvalidStateError occur, I add a delay and try to register again
9. mosquito

   added 2 commits that reference this issue on Oct 10, 2026

   Restore robust consumers after broker cancellation

   ...

   This issue will close once commit 9895dc9 is merged into the 'master' branch.

   9895dc9

   Restore robust consumers after broker cancellation

   ...

   This issue will close once commit 7e5adfb is merged into the 'master' branch.

   7e5adfb
10. SaInekK commented on Oct 12, 2026

    

    SaInekK

    on Oct 12, 2026

    More actions

    @mosquito thanks for #723! Re-ran my cluster repro on aio-pika 10.1.1 / aiormq 7.2.2 (RabbitMQ 4.0, classic queue homed on node 2, client on node 1, docker stop / docker start node 2):

    classic: before        consumers=1
    aiormq.channel: Consumer 'ctag1.25097247554047719f59ba15465d6b42' cancelled by the broker on <Channel: "1" at 0x107037330>
    classic: node 2 down   consumers=0 connection_closed=False channel_closed=False
    classic: node 2 back   consumers=1 delivered=[b'after']
    

    With 20 classic queues on the stopped node and all consumers on one channel, all 20 came back about 3 s after the node was up, each got its message, and the channel stayed open. Quorum queues behave as before.