PikaTransport: Re-enable publisher confirms

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 = {}

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