Merge pull request #1454 from rabbitmq/rabbitmq-dotnet-client-1043-6.x

Back-port https://github.com/rabbitmq/rabbitmq-dotnet-client/pull/145…

Author: lukebakken

projects/RabbitMQ.Client/client/impl/ModelBase.cs

@@ -1194,11 +1194,28 @@ public void BasicPublish(string exchange,
                 }
             }
 
-            _Private_BasicPublish(exchange,
-                routingKey,
-                mandatory,
-                basicProperties,
-                body);
+            try
+            {
+                _Private_BasicPublish(exchange,
+                    routingKey,
+                    mandatory,
+                    basicProperties,
+                    body);
+            }
+            catch
+            {
+                if (NextPublishSeqNo > 0)
+                {
+                    lock (_confirmLock)
+                    {
+                        NextPublishSeqNo--;
+                        _pendingDeliveryTags.RemoveLast();
+                        _deliveryTagsCountdown.Reset(_pendingDeliveryTags.Count);
+                    }
+                }
+
+                throw;
+            }
         }
 
         public void UpdateSecret(string newSecret, string reason)

projects/Unit/TestConfirmSelect.cs

@@ -29,6 +29,7 @@
 //  Copyright (c) 2007-2020 VMware, Inc.  All rights reserved.
 //---------------------------------------------------------------------------
 
+using System;
 using NUnit.Framework;
 
 namespace RabbitMQ.Client.Unit
@@ -63,5 +64,61 @@ protected void Publish()
         {
             Model.BasicPublish("", "amq.fanout", null, encoding.GetBytes("message"));
         }
+
+        [Test]
+        [TestCase(255)]
+        [TestCase(256)]
+        public void TestDeliveryTagDiverged_GH1043(int correlationIdLength)
+        {
+            bool.TryParse(Environment.GetEnvironmentVariable("RABBITMQ_VERBOSE"), out bool verbose);
+
+            byte[] body = RandomMessageBody();
+
+            Model.ExchangeDeclare("sample", "fanout", autoDelete: true);
+            if (verbose)
+            {
+                Model.BasicAcks += (s, e) => Console.WriteLine("Acked {0}", e.DeliveryTag);
+            }
+            Model.ConfirmSelect();
+
+            IBasicProperties properties = Model.CreateBasicProperties();
+            if (verbose)
+            {
+                Console.WriteLine("Client delivery tag {0}", Model.NextPublishSeqNo);
+            }
+            Model.BasicPublish(exchange: "sample", routingKey: string.Empty, properties, body);
+            Model.WaitForConfirmsOrDie();
+
+            try
+            {
+                properties = Model.CreateBasicProperties();
+                properties.CorrelationId = new string('o', correlationIdLength);
+                if (verbose)
+                {
+                    Console.WriteLine("Client delivery tag {0}", Model.NextPublishSeqNo);
+                }
+                Model.BasicPublish("sample", string.Empty, properties, body);
+                Model.WaitForConfirmsOrDie();
+            }
+            catch (Exception e)
+            {
+                if (verbose)
+                {
+                    Console.WriteLine("Error when trying to publish with long string: {0}", e.Message);
+                }
+            }
+
+            properties = Model.CreateBasicProperties();
+            if (verbose)
+            {
+                Console.WriteLine("Client delivery tag {0}", Model.NextPublishSeqNo);
+            }
+            Model.BasicPublish("sample", string.Empty, properties, body);
+            Model.WaitForConfirmsOrDie();
+            if (verbose)
+            {
+                Console.WriteLine("I'm done...");
+            }
+        }
     }
 }