Fix TerminateAsync to fire Closed event so producer is unregistered from recovery set

Agent-Logs-Url: https://github.com/Havret/dotnet-activemq-artemis-client/sessions/8bc2f1b9-864e-4a9f-a756-be8df181d7e6

Co-authored-by: Havret <9103861+Havret@users.noreply.github.com>

Author: Copilot

src/ArtemisNetClient/AutoRecovering/AutoRecoveringProducerBase.cs

@@ -80,6 +80,7 @@ public async Task TerminateAsync(Exception exception)
             _manualResetEvent.Set();
             Log.ProducerTerminated(Logger, exception);
             await DisposeResourceSafe(UnderlyingResource).ConfigureAwait(false);
+            Closed?.Invoke(this);
         }
 
         public async ValueTask DisposeAsync()

test/ArtemisNetClient.UnitTests/AutoRecovering/AutoRecoveringAnonymousProducerSpec.cs

@@ -1,6 +1,8 @@
 using System;
+using System.Threading;
 using System.Threading.Tasks;
 using ActiveMQ.Artemis.Client.Exceptions;
+using ActiveMQ.Artemis.Client.InternalUtilities;
 using ActiveMQ.Artemis.Client.UnitTests.Utils;
 using Amqp;
 using Amqp.Framing;
@@ -82,5 +84,57 @@ public async Task Should_terminate_anonymous_producer_on_Send_when_link_closed_w
             Assert.Throws<ProducerClosedException>(() =>
                 producer.Send("a2", null, new Message("bar"), CancellationToken));
         }
+        [Fact]
+        public async Task Should_not_recreate_anonymous_producer_on_connection_recovery_after_terminal_unauthorized_access_error()
+        {
+            var endpoint = GetUniqueEndpoint();
+            var producerAttached = new ManualResetEvent(false);
+            var testHandler = new TestHandler(@event =>
+            {
+                switch (@event.Id)
+                {
+                    case EventId.LinkRemoteOpen when @event.Context is Attach attach && attach.Role:
+                        producerAttached.Set();
+                        break;
+                }
+            });
+
+            var host1 = CreateOpenedContainerHost(endpoint, testHandler);
+            var linkProcessor = host1.CreateTestLinkProcessor();
+
+            ListenerLink producerLink = null;
+            linkProcessor.SetHandler(context =>
+            {
+                producerLink = context.Link;
+                return false;
+            });
+
+            var connection = await CreateConnection(endpoint);
+            var producer = await connection.CreateAnonymousProducerAsync();
+
+            Assert.True(producerAttached.WaitOne(Timeout));
+            producerAttached.Reset();
+
+            try
+            {
+                await producerLink.CloseAsync(Timeout, new Error(ErrorCode.UnauthorizedAccess) { Description = "Unauthorized" });
+            }
+            catch (Exception)
+            {
+                // ignored
+            }
+
+            // SendAsync triggers TerminateAsync which should remove the producer from the recovery set.
+            await Assert.ThrowsAsync<ProducerClosedException>(() =>
+                producer.SendAsync("a1", null, new Message("foo"), null, CancellationToken));
+
+            host1.Dispose();
+            using var host2 = CreateOpenedContainerHost(endpoint, testHandler);
+
+            // Producer should NOT be recreated after connection recovery.
+            Assert.False(producerAttached.WaitOne(ShortTimeout));
+
+            await DisposeUtil.DisposeAll(connection, host2);
+        }
     }
 }

test/ArtemisNetClient.UnitTests/AutoRecovering/AutoRecoveringProducerSpec.cs

@@ -276,6 +276,58 @@ public async Task Should_not_retry_Send_when_producer_link_closed_with_not_found
             Assert.Throws<ProducerClosedException>(() => producer.Send(new Message("bar"), CancellationToken));
         }
 
+        [Fact]
+        public async Task Should_not_recreate_producer_on_connection_recovery_after_terminal_not_found_error()
+        {
+            var endpoint = GetUniqueEndpoint();
+            var producerAttached = new ManualResetEvent(false);
+            var testHandler = new TestHandler(@event =>
+            {
+                switch (@event.Id)
+                {
+                    case EventId.LinkRemoteOpen when @event.Context is Attach attach && attach.Role:
+                        producerAttached.Set();
+                        break;
+                }
+            });
+
+            var host1 = CreateOpenedContainerHost(endpoint, testHandler);
+            var linkProcessor = host1.CreateTestLinkProcessor();
+
+            ListenerLink producerLink = null;
+            linkProcessor.SetHandler(context =>
+            {
+                producerLink = context.Link;
+                return false;
+            });
+
+            var connection = await CreateConnection(endpoint);
+            var producer = await connection.CreateProducerAsync("a1", RoutingType.Anycast);
+
+            Assert.True(producerAttached.WaitOne(Timeout));
+            producerAttached.Reset();
+
+            try
+            {
+                await producerLink.CloseAsync(Timeout, new Error(Amqp.ErrorCode.NotFound) { Description = "AMQ119002: target address a1 does not exist" });
+            }
+            catch (Exception)
+            {
+                // ignored
+            }
+
+            // SendAsync triggers TerminateAsync which should remove the producer from the recovery set.
+            await Assert.ThrowsAsync<ProducerClosedException>(() => producer.SendAsync(new Message("foo"), CancellationToken));
+
+            host1.Dispose();
+            using var host2 = CreateOpenedContainerHost(endpoint, testHandler);
+
+            // Producer should NOT be recreated after connection recovery.
+            Assert.False(producerAttached.WaitOne(ShortTimeout));
+
+            await DisposeUtil.DisposeAll(connection, host2);
+        }
+
         private async Task<(IProducer producer, MessageProcessor messageProcessor, TestContainerHost host, IConnection connection)> CreateReattachedProducer()
         {
             var endpoint = GetUniqueEndpoint();