From 15770af7fa5eb367a510629a0c7d0c30b2daa442 Mon Sep 17 00:00:00 2001
From: Meher Khan <meher.khan@weareadaptive.com>
Date: Wed, 26 Aug 2026 09:54:13 +0100
Subject: [PATCH] Port ingress publication retry test for closing channel
 endpoint

---
 .../Client/AeronClusterAsyncConnectTest.cs    | 58 +++++++++++++++++++
 1 file changed, 58 insertions(+)

diff --git a/src/Adaptive.Cluster.Tests/Client/AeronClusterAsyncConnectTest.cs b/src/Adaptive.Cluster.Tests/Client/AeronClusterAsyncConnectTest.cs
index bdf70ec..166e161 100644
--- a/src/Adaptive.Cluster.Tests/Client/AeronClusterAsyncConnectTest.cs
+++ b/src/Adaptive.Cluster.Tests/Client/AeronClusterAsyncConnectTest.cs
@@ -15,6 +15,7 @@
  */
 
 using Adaptive.Aeron;
+using Adaptive.Aeron.Exceptions;
 using Adaptive.Aeron.LogBuffer;
 using Adaptive.Aeron.Protocol;
 using Adaptive.Aeron.Security;
@@ -237,6 +238,63 @@ public void ShouldCloseIngressPublicationsOnMembers()
             A.CallTo(() => _aeron.AsyncRemovePublication(publicationId2)).MustNotHaveHappened();
         }
 
+        [Test]
+        public void ShouldRetryMemberIngressPublicationWhenSendChannelEndpointIsClosing()
+        {
+            const long subscriptionId = 42L;
+            A.CallTo(() => _aeron.AsyncAddSubscription(_context.EgressChannel(), _context.EgressStreamId()))
+                .Returns(subscriptionId);
+            var subscription = A.Fake<Subscription>();
+            A.CallTo(() => subscription.TryResolveChannelEndpointPort())
+                .Returns("aeron:udp?endpoint=localhost:8888");
+            A.CallTo(() => _aeron.GetSubscription(subscriptionId)).Returns(subscription);
+
+            const int ingressStreamId = 878;
+            _context
+                .IsIngressExclusive(true)
+                .IngressEndpoints("0=localhost:20000")
+                .IngressStreamId(ingressStreamId);
+
+            const long publicationId = -6342756432L;
+            var publication = A.Fake<ExclusivePublication>();
+            A.CallTo(() => publication.IsConnected).Returns(true);
+            A.CallTo(() => _aeron.AsyncAddExclusivePublication("aeron:udp?endpoint=localhost:20000", ingressStreamId))
+                .Returns(publicationId);
+
+            A.CallTo(() => _aeron.GetExclusivePublication(publicationId))
+                .Throws(
+                    new RegistrationException(
+                        publicationId,
+                        (int)ErrorCode.RESOURCE_TEMPORARILY_UNAVAILABLE,
+                        ErrorCode.RESOURCE_TEMPORARILY_UNAVAILABLE,
+                        "send_channel_endpoint found in CLOSING state, please retry"
+                    )
+                )
+                .Once()
+                .Then.Returns(publication);
+
+            var asyncConnect = new AeronCluster.AsyncConnect(
+                _context,
+                _aeronContext.NanoClock().NanoTime() + OneHourInNanos
+            );
+
+            Assert.IsNull(asyncConnect.Poll());
+            Assert.AreEqual(AsyncConnectState.CREATE_INGRESS_PUBLICATIONS, asyncConnect.State());
+
+            Assert.IsNull(asyncConnect.Poll());
+            Assert.AreEqual(AsyncConnectState.AWAIT_PUBLICATION_CONNECTED, asyncConnect.State());
+
+            Assert.IsNull(asyncConnect.Poll());
+            Assert.AreEqual(AsyncConnectState.AWAIT_PUBLICATION_CONNECTED, asyncConnect.State());
+
+            Assert.IsNull(asyncConnect.Poll());
+            Assert.AreEqual(AsyncConnectState.SEND_MESSAGE, asyncConnect.State());
+
+            A.CallTo(() => _aeron.GetExclusivePublication(publicationId)).MustHaveHappened(2, Times.Exactly);
+
+            asyncConnect.Dispose();
+        }
+
         [Test]
         public void ShouldCloseIngressPublication()
         {