From 19e1efb05763dfcd8bdc9f2859ce6055062c43e1 Mon Sep 17 00:00:00 2001 From: Nicholas Devenish Date: Fri, 6 Nov 2026 14:55:25 +0000 Subject: [PATCH] PikaTransport: Re-enable publisher confirms Non-transactional sends now go through a dedicated channel in publisher-confirms mode, so basic_publish blocks until the broker acknowledges the message, and raises UnroutableError/NackError if it is rejected. Confirm and transaction modes are mutually exclusive on a channel, so the shared channel is kept in plain mode for transactions not tied to a subscription. Broker rejections are now delivered to the caller's future only, rather than also being raised into the connection runloop, which would otherwise terminate the thread without reconnecting. Since PikaTransport.send uses mandatory=True, sending to a queue that does not exist now raises instead of silently dropping the message. Fixes #96 --- src/workflows/transport/pika_transport.py | 32 +++++++++++-- tests/transport/test_pika.py | 56 ++++++++++++++++++++++- 2 files changed, 83 insertions(+), 5 deletions(-) diff --git a/src/workflows/transport/pika_transport.py b/src/workflows/transport/pika_transport.py index 4cbd706..7c86915 100644 --- a/src/workflows/transport/pika_transport.py +++ b/src/workflows/transport/pika_transport.py @@ -801,8 +801,12 @@ def __init__( self._channel_has_active_tx: dict[BlockingChannel, bool] = {} # Information on whether a channel is transactional self._channel_is_transactional: dict[BlockingChannel, bool] = {} - # A common, shared channel, used for sending messages outside of transactions. - self._pika_shared_channel: BlockingChannel | None + # A common, shared channel, used for transactions not tied to a subscription. + self._pika_shared_channel: BlockingChannel | None = None + # A channel in publisher-confirms mode, used for sending messages outside + # of transactions. Confirm and transaction modes are mutually exclusive + # on a channel, so this has to be kept separate from the shared channel. + self._pika_confirm_channel: BlockingChannel | None = None # Are we allowed to reconnect. Can only be turned off, never on self._reconnection_allowed: bool = True # Our list of connection parameters, so we know where to connect to @@ -1117,7 +1121,9 @@ def _send() -> None: channel = self._transaction_on_channel.inverse[transaction_id] self._channel_has_active_tx[channel] = True else: - channel = self._get_shared_channel() + channel = self._get_confirm_channel() + # On a confirm-mode channel this blocks until the broker + # has acknowledged the message channel.basic_publish( exchange=exchange, routing_key=routing_key, @@ -1126,6 +1132,14 @@ def _send() -> None: mandatory=mandatory, ) future.set_result(None) + except ( + pika.exceptions.UnroutableError, + pika.exceptions.NackError, + ) as e: + # The broker rejected this message. This is a problem for + # the sender, not for the connection, so don't propagate + # it into (and terminate) the connection runloop. + future.set_exception(e) except BaseException as e: future.set_exception(e) raise @@ -1367,9 +1381,18 @@ def _get_shared_channel(self) -> BlockingChannel: if not self._pika_shared_channel: self._pika_shared_channel = self._connection.channel() - ##### self._pika_shared_channel.confirm_delivery() return self._pika_shared_channel + def _get_confirm_channel(self) -> BlockingChannel: + """Get the publisher-confirms channel. Create if necessary.""" + + assert self._connection is not None + + if not self._pika_confirm_channel: + self._pika_confirm_channel = self._connection.channel() + self._pika_confirm_channel.confirm_delivery() + return self._pika_confirm_channel + def _recreate_subscriptions(self) -> None: """Resubscribe all existing subscriptions""" old_subscriptions = self._subscriptions @@ -1481,6 +1504,7 @@ def _run(self) -> None: # Clear the channels because this might be a reconnect self._pika_channels = bidict() self._pika_shared_channel = None + self._pika_confirm_channel = None self._transaction_on_channel = bidict() self._channel_has_active_tx = {} self._channel_is_transactional = {} diff --git a/tests/transport/test_pika.py b/tests/transport/test_pika.py index 516c21e..ea998cf 100644 --- a/tests/transport/test_pika.py +++ b/tests/transport/test_pika.py @@ -1068,16 +1068,70 @@ def test_pikathread_send(connection_params, test_channel): ).result() # But should fail with it declared - pytest.xfail("UnroutableError is not raised without publisher confirms, #96") with pytest.raises(pika.exceptions.UnroutableError): thread.send( "", "unroutable-missing-queue", "Another Message", mandatory=True ).result() + # A rejected message must not take down the connection + assert thread.connection_alive + thread.send("", queue, "After unroutable").result() + assert test_channel.basic_get(queue, auto_ack=True)[2] == b"After unroutable" + finally: thread.join(stop=True) +def test_pikathread_send_confirmed_during_shared_transaction( + connection_params, test_channel +): + """Confirmed sends must coexist with a transaction on the shared channel.""" + queue = test_channel.temporary_queue_declare(exclusive=True) + thread = _PikaThread(connection_params) + try: + thread.start() + thread.wait_for_connection() + + # Open a transaction on the shared (non-subscription) channel + thread.tx_select(transaction_id=1, subscription_id=None).result() + thread.send("", queue, "in transaction", transaction_id=1).result() + + # A non-transactional send is confirmed and delivered immediately + thread.send("", queue, "outside transaction").result() + assert test_channel.basic_get(queue, auto_ack=True)[2] == b"outside transaction" + # ...while the transactional message is still held back + assert test_channel.basic_get(queue, auto_ack=True)[0] is None + + thread.tx_commit(transaction_id=1).result() + assert test_channel.basic_get(queue, auto_ack=True)[2] == b"in transaction" + + # Unroutable messages are still caught while the transaction is open + thread.tx_select(transaction_id=2, subscription_id=None).result() + with pytest.raises(pika.exceptions.UnroutableError): + thread.send( + "", "unroutable-missing-queue", "Another Message", mandatory=True + ).result() + thread.tx_rollback(transaction_id=2).result() + assert thread.connection_alive + finally: + thread.join(re_raise=True, stop=True) + + +def test_pikatransport_send_unroutable_raises(connection_params): + """PikaTransport.send uses mandatory=True, so a missing queue now raises.""" + transport = PikaTransport() + with mock.patch.object( + transport, "_generate_connection_parameters", return_value=connection_params + ): + transport.connect() + try: + with pytest.raises(pika.exceptions.UnroutableError): + transport.send("unroutable-missing-queue", "message") + assert transport.is_connected() + finally: + transport.disconnect() + + def test_pikathread_send_named_exchange(connection_params, test_channel): exchange = test_channel.temporary_exchange_declare( exchange_type="topic", auto_delete=True