fix: RMQ.Sync producer selects confirms and subscribes handlers once per channel

bugfixes/0054-rmq-sync-confirm-resubscribe/.issue-number

@@ -0,0 +1 @@
+4567

src/Paramore.Brighter.MessagingGateway.RMQ.Sync/RmqMessageProducer.cs

@@ -37,6 +37,7 @@ THE SOFTWARE. */
 using Paramore.Brighter.Logging;
 using Paramore.Brighter.Observability;
 using Paramore.Brighter.Tasks;
+using RabbitMQ.Client;
 using RabbitMQ.Client.Events;
 
 namespace Paramore.Brighter.MessagingGateway.RMQ.Sync
@@ -58,7 +59,8 @@ public partial class RmqMessageProducer : RmqMessageGateway, IAmAMessageProducer
         static readonly object s_lock = new();
         private RmqPublication _publication;
         private readonly ConcurrentDictionary<ulong, PendingConfirmation> _pendingConfirmations = new ConcurrentDictionary<ulong, PendingConfirmation>();
-        private bool _confirmsSelected;
+        // The channel in confirm mode with our handlers on it; the gateway replaces a closed channel with a new instance
+        private IModel? _confirmChannel;
         private readonly int _waitForConfirmsTimeOutInMilliseconds;
         private event Func<PublishConfirmationResult, Task>? _onMessagePublishedAsync;
         // The ack/nack handlers run on the client's connection loop, which must never block on a
@@ -167,17 +169,14 @@ private void SendWithDelay(Message message, TimeSpan? delay, bool useSchedulerAs
                     var rmqMessagePublisher = new RmqMessagePublisher(Channel!, Connection);
 
                     message.Persist = Connection.PersistMessages;
-                    Channel!.BasicAcks += OnPublishSucceeded;
-                    Channel.BasicNacks += OnPublishFailed;
-                    Channel.ConfirmSelect();
-                    _confirmsSelected = true;
+                    SelectConfirmsOnChannel();
 
                     BrighterTracer.WriteProducerEvent(Span, MessagingSystem.RabbitMQ, message, _instrumentationOptions);
 
                     Log.PublishingMessage(s_logger, Connection.Exchange.Name, Connection.AmpqUri!.GetSanitizedUri(), delay.Value.TotalMilliseconds,
                         message.Header.Topic.Value, message.Persist, message.Id.Value, message.Body.Value);
 
-                    _pendingConfirmations.TryAdd(Channel.NextPublishSeqNo, new PendingConfirmation(message.Id, message.Header.Topic, publishContext));
+                    _pendingConfirmations.TryAdd(Channel!.NextPublishSeqNo, new PendingConfirmation(message.Id, message.Header.Topic, publishContext));
 
                      if (delay == TimeSpan.Zero || DelaySupported || Scheduler == null)
                      {
@@ -202,6 +201,7 @@ private void SendWithDelay(Message message, TimeSpan? delay, bool useSchedulerAs
             catch (IOException io)
             {
                 Log.ErrorTalkingToSocket(s_logger, io, Connection.AmpqUri!.GetSanitizedUri());
+                DetachConfirms();
                 // The new channel numbers its delivery tags from 1 again, so its acks must not settle the old channel's
                 _pendingConfirmations.Clear();
                 ResetConnectionToBroker();
@@ -248,7 +248,7 @@ protected override void Dispose(bool disposing)
             {
                 if (disposing)
                 {
-                    if (Channel != null && Channel.IsOpen && _confirmsSelected)
+                    if (Channel != null && Channel.IsOpen && ReferenceEquals(Channel, _confirmChannel))
                     {
                         //In the event this fails, then consequence is not marked as sent in outbox
                         //As we are disposing, just let that happen
@@ -265,10 +265,39 @@ protected override void Dispose(bool disposing)
             }
             finally
             {
-                base.Dispose(disposing);
+                try
+                {
+                    // Detach while the channel is still alive, even if waiting failed; base.Dispose disposes it
+                    if (disposing) DetachConfirms();
+                }
+                finally
+                {
+                    base.Dispose(disposing);
+                }
             }
         }
 
+        // Confirm mode, and the handlers, are set up once per channel, not once per Send
+        private void SelectConfirmsOnChannel()
+        {
+            if (ReferenceEquals(Channel, _confirmChannel)) return;
+
+            DetachConfirms();
+            Channel!.ConfirmSelect();
+            Channel.BasicAcks += OnPublishSucceeded;
+            Channel.BasicNacks += OnPublishFailed;
+            _confirmChannel = Channel;
+        }
+
+        private void DetachConfirms()
+        {
+            if (_confirmChannel is null) return;
+
+            _confirmChannel.BasicAcks -= OnPublishSucceeded;
+            _confirmChannel.BasicNacks -= OnPublishFailed;
+            _confirmChannel = null;
+        }
+
         private void OnPublishFailed(object? sender, BasicNackEventArgs e) => SettleConfirmations(e.DeliveryTag, e.Multiple, success: false);
 
         private void OnPublishSucceeded(object? sender, BasicAckEventArgs e) => SettleConfirmations(e.DeliveryTag, e.Multiple, success: true);
@@ -277,7 +306,7 @@ private void SettleConfirmations(ulong deliveryTag, bool multiple, bool success)
         {
             foreach (var confirmedTag in ConfirmedDeliveryTags(deliveryTag, multiple))
             {
-                // Claim before raising: the handlers are subscribed once per Send, so only one invocation may win each tag
+                // Claim before raising, so a tag is settled at most once even if its ack and nack handlers race
                 if (_pendingConfirmations.TryRemove(confirmedTag, out PendingConfirmation confirmation))
                 {
                     RaisePublishConfirmation(new PublishConfirmationResult(success, confirmation.MessageId, confirmation.Topic, confirmation.Context));

tests/Paramore.Brighter.RMQ.Sync.Tests/MessagingGateway/Reactor/When_sending_several_messages_on_one_channel_should_select_confirms_once.cs

@@ -0,0 +1,51 @@
+using System;
+using Paramore.Brighter.MessagingGateway.RMQ.Sync;
+using Paramore.Brighter.RMQ.Sync.Tests.TestDoubles;
+using Xunit;
+
+namespace Paramore.Brighter.RMQ.Sync.Tests.MessagingGateway.Reactor;
+
+[Trait("Category", "RMQ")]
+[Collection("RMQ")]
+public class RmqConfirmSelectOncePerChannelTests : IDisposable
+{
+    private readonly RoutingKey _routingKey = new(Guid.NewGuid().ToString());
+    private readonly CoalescingConfirmsRmqProducer _messageProducer;
+
+    public RmqConfirmSelectOncePerChannelTests()
+    {
+        var rmqConnection = new RmqMessagingGatewayConnection
+        {
+            AmpqUri = new AmqpUriSpecification(new Uri("amqp://guest:guest@localhost:5672/%2f")),
+            Exchange = new Exchange("paramore.brighter.exchange")
+        };
+
+        _messageProducer = new CoalescingConfirmsRmqProducer(rmqConnection);
+
+        //we need a queue to avoid a discard
+        new QueueFactory(rmqConnection, new ChannelName(Guid.NewGuid().ToString()), new RoutingKeys(_routingKey))
+            .Create(TimeSpan.FromMilliseconds(1000));
+    }
+
+    [Fact]
+    public void When_sending_several_messages_on_one_channel_should_select_confirms_once()
+    {
+        //Arrange — the first send opens the channel
+        _messageProducer.Send(NewMessage());
+
+        //Act — further sends reuse the same channel
+        _messageProducer.Send(NewMessage());
+        _messageProducer.Send(NewMessage());
+
+        //Assert — the channel was put into confirm mode, and given its confirm handlers, only once
+        Assert.Equal(1, _messageProducer.Confirms.ConfirmSelectCount);
+        Assert.Equal(1, _messageProducer.Confirms.AckSubscriberCount);
+        Assert.Equal(1, _messageProducer.Confirms.NackSubscriberCount);
+    }
+
+    private Message NewMessage() => new(
+        new MessageHeader(Id.Random(), _routingKey, MessageType.MT_COMMAND),
+        new MessageBody("confirm me"));
+
+    public void Dispose() => _messageProducer.Dispose();
+}

tests/Paramore.Brighter.RMQ.Sync.Tests/MessagingGateway/Reactor/When_the_channel_resets_should_move_confirm_handlers_to_the_new_channel.cs

@@ -0,0 +1,59 @@
+using System;
+using Paramore.Brighter.MessagingGateway.RMQ.Sync;
+using Paramore.Brighter.RMQ.Sync.Tests.TestDoubles;
+using Xunit;
+
+namespace Paramore.Brighter.RMQ.Sync.Tests.MessagingGateway.Reactor;
+
+[Trait("Category", "RMQ")]
+[Collection("RMQ")]
+public class RmqChannelResetConfirmHandlerTests : IDisposable
+{
+    private readonly RoutingKey _routingKey = new(Guid.NewGuid().ToString());
+    private readonly CoalescingConfirmsRmqProducer _messageProducer;
+
+    public RmqChannelResetConfirmHandlerTests()
+    {
+        var rmqConnection = new RmqMessagingGatewayConnection
+        {
+            AmpqUri = new AmqpUriSpecification(new Uri("amqp://guest:guest@localhost:5672/%2f")),
+            Exchange = new Exchange("paramore.brighter.exchange")
+        };
+
+        _messageProducer = new CoalescingConfirmsRmqProducer(rmqConnection);
+
+        //we need a queue to avoid a discard
+        new QueueFactory(rmqConnection, new ChannelName(Guid.NewGuid().ToString()), new RoutingKeys(_routingKey))
+            .Create(TimeSpan.FromMilliseconds(1000));
+    }
+
+    [Fact]
+    public void When_the_channel_resets_should_move_confirm_handlers_to_the_new_channel()
+    {
+        //Arrange — a send on the old channel, then its socket fails on the next send
+        _messageProducer.Send(NewMessage());
+
+        _messageProducer.Confirms.FailNextPublish();
+        Assert.Throws<ChannelFailureException>(() => _messageProducer.Send(NewMessage()));
+
+        //Act — the next send opens a new channel
+        _messageProducer.Send(NewMessage());
+
+        //Assert — the failed channel no longer calls the producer, and the new one is set up once
+        var oldChannel = _messageProducer.Channels[0];
+        var newChannel = _messageProducer.Channels[1];
+
+        Assert.Equal(0, oldChannel.AckSubscriberCount);
+        Assert.Equal(0, oldChannel.NackSubscriberCount);
+
+        Assert.Equal(1, newChannel.ConfirmSelectCount);
+        Assert.Equal(1, newChannel.AckSubscriberCount);
+        Assert.Equal(1, newChannel.NackSubscriberCount);
+    }
+
+    private Message NewMessage() => new(
+        new MessageHeader(Id.Random(), _routingKey, MessageType.MT_COMMAND),
+        new MessageBody("confirm me"));
+
+    public void Dispose() => _messageProducer.Dispose();
+}

tests/Paramore.Brighter.RMQ.Sync.Tests/MessagingGateway/Reactor/When_the_shared_connection_is_dropped_should_select_confirms_on_the_new_channel.cs

@@ -0,0 +1,58 @@
+using System;
+using Paramore.Brighter.MessagingGateway.RMQ.Sync;
+using Paramore.Brighter.RMQ.Sync.Tests.TestDoubles;
+using RabbitMQ.Client;
+using Xunit;
+
+namespace Paramore.Brighter.RMQ.Sync.Tests.MessagingGateway.Reactor;
+
+[Trait("Category", "RMQ")]
+[Collection("RMQ")]
+public class RmqSharedConnectionDroppedConfirmTests : IDisposable
+{
+    private readonly RoutingKey _routingKey = new(Guid.NewGuid().ToString());
+    private readonly RmqMessagingGatewayConnection _rmqConnection;
+    private readonly CoalescingConfirmsRmqProducer _messageProducer;
+
+    public RmqSharedConnectionDroppedConfirmTests()
+    {
+        _rmqConnection = new RmqMessagingGatewayConnection
+        {
+            AmpqUri = new AmqpUriSpecification(new Uri("amqp://guest:guest@localhost:5672/%2f")),
+            Exchange = new Exchange("paramore.brighter.exchange")
+        };
+
+        _messageProducer = new CoalescingConfirmsRmqProducer(_rmqConnection);
+
+        //we need a queue to avoid a discard
+        new QueueFactory(_rmqConnection, new ChannelName(Guid.NewGuid().ToString()), new RoutingKeys(_routingKey))
+            .Create(TimeSpan.FromMilliseconds(1000));
+    }
+
+    [Fact]
+    public void When_the_shared_connection_is_dropped_should_select_confirms_on_the_new_channel()
+    {
+        //Arrange — after a send, another gateway drops the pooled connection this producer's channel is on
+        _messageProducer.Send(NewMessage());
+
+        new RmqMessageGatewayConnectionPool(_rmqConnection.Name!, _rmqConnection.Heartbeat)
+            .RemoveConnection(new ConnectionFactory { Uri = _rmqConnection.AmpqUri!.Uri });
+
+        //Act — the next send finds its channel closed, with no IOException, and opens a new one
+        var error = Record.Exception(() => _messageProducer.Send(NewMessage()));
+
+        //Assert — the send succeeds and the new channel is set up once
+        Assert.Null(error);
+
+        var newChannel = _messageProducer.Channels[1];
+        Assert.Equal(1, newChannel.ConfirmSelectCount);
+        Assert.Equal(1, newChannel.AckSubscriberCount);
+        Assert.Equal(1, newChannel.NackSubscriberCount);
+    }
+
+    private Message NewMessage() => new(
+        new MessageHeader(Id.Random(), _routingKey, MessageType.MT_COMMAND),
+        new MessageBody("confirm me"));
+
+    public void Dispose() => _messageProducer.Dispose();
+}

tests/Paramore.Brighter.RMQ.Sync.Tests/TestDoubles/CoalescingConfirmsRmqChannel.cs

@@ -43,6 +43,10 @@ public class CoalescingConfirmsRmqChannel : DispatchProxy
     private EventHandler<BasicNackEventArgs>? _nacks;
     private bool _failNextPublish;
 
+    public int AckSubscriberCount => _acks?.GetInvocationList().Length ?? 0;
+    public int NackSubscriberCount => _nacks?.GetInvocationList().Length ?? 0;
+    public int ConfirmSelectCount { get; private set; }
+
     public static CoalescingConfirmsRmqChannel Wrap(IModel channel)
     {
         var proxy = Create<IModel, CoalescingConfirmsRmqChannel>();
@@ -90,6 +94,16 @@ private void Raise<TArgs>(EventHandler<TArgs>? handlers, TArgs args)
             case "add_BasicNacks":
                 _nacks += (EventHandler<BasicNackEventArgs>)args![0]!;
                 return null;
+            // Removals still reach the real channel, which rejects them once disposed, as it would in production
+            case "remove_BasicAcks":
+                _acks -= (EventHandler<BasicAckEventArgs>)args![0]!;
+                break;
+            case "remove_BasicNacks":
+                _nacks -= (EventHandler<BasicNackEventArgs>)args![0]!;
+                break;
+            case "ConfirmSelect":
+                ConfirmSelectCount++;
+                break;
             case "BasicPublish" when _failNextPublish:
                 _failNextPublish = false;
                 throw new IOException("Injected socket failure.");

tests/Paramore.Brighter.RMQ.Sync.Tests/TestDoubles/CoalescingConfirmsRmqProducer.cs

@@ -25,6 +25,7 @@ THE SOFTWARE. */
 
 #nullable enable
 
+using System.Collections.Generic;
 using Paramore.Brighter.MessagingGateway.RMQ.Sync;
 using RabbitMQ.Client;
 
@@ -33,12 +34,20 @@ namespace Paramore.Brighter.RMQ.Sync.Tests.TestDoubles;
 public class CoalescingConfirmsRmqProducer(RmqMessagingGatewayConnection connection)
     : RmqMessageProducer(connection)
 {
+    private readonly List<CoalescingConfirmsRmqChannel> _channels = [];
+
     public CoalescingConfirmsRmqChannel Confirms => (CoalescingConfirmsRmqChannel)Channel!;
 
+    // Every channel the producer has opened, oldest first, so a test can inspect one it has replaced
+    public IReadOnlyList<CoalescingConfirmsRmqChannel> Channels => _channels;
+
     protected override void ConnectToBroker(OnMissingChannel makeExchange)
     {
         base.ConnectToBroker(makeExchange);
-        if (Channel is not CoalescingConfirmsRmqChannel)
-            Channel = (IModel)(object)CoalescingConfirmsRmqChannel.Wrap(Channel!);
+        if (Channel is CoalescingConfirmsRmqChannel) return;
+
+        var confirms = CoalescingConfirmsRmqChannel.Wrap(Channel!);
+        _channels.Add(confirms);
+        Channel = (IModel)(object)confirms;
     }
 }