EventTest

Libraries/Esiur/Protocol/EpConnection.cs

@@ -3755,11 +3755,16 @@ public partial class EpConnection : NetworkConnection, IStore
         // or resubscribes events.
         _attachedResources.Clear();
 
+        // Attachments requested by the peer belong to the socket session, even
+        // when this endpoint initiated the connection.  Keeping them on a
+        // bidirectional client connection makes the peer's first attachment
+        // after reconnect fail with AlreadyAttached.
+        UnsubscribeAll();
+
         if (Server != null)
         {
             _suspendedResources.Clear();
 
-            UnsubscribeAll();
             Instance?.Warehouse?.Remove(this);
 
             if (wasAuthenticated)

Tests/Unit/Integration/EventSubscriptionIntegrationTests.cs

@@ -1,5 +1,6 @@
 using Esiur.Protocol;
 using Esiur.Resource;
+using Esiur.Stores;
 
 namespace Esiur.Tests.Unit.Integration;
 
@@ -280,6 +281,44 @@ public class EventSubscriptionIntegrationTests
         Assert.Equal(new[] { "before", "after" }, received);
     }
 
+    [Fact]
+    public async Task BidirectionalPeerAttachment_CanBeRequestedAgainAfterAutomaticReconnect()
+    {
+        var firstAttachment = new TaskCompletionSource<EpResource>(
+            TaskCreationOptions.RunContinuationsAsynchronously);
+        var resumedAttachment = new TaskCompletionSource<EpResource>(
+            TaskCreationOptions.RunContinuationsAsynchronously);
+        var readyCount = 0;
+
+        await using var cluster = await IntegrationCluster.StartAsync(
+            _ => Task.CompletedTask,
+            serverCreated: server => server.ConnectionReady += connection =>
+            {
+                var attempt = Interlocked.Increment(ref readyCount);
+                _ = ResolveClientResourceAsync(
+                    connection,
+                    attempt == 1 ? firstAttachment : resumedAttachment);
+            },
+            populateClient: async warehouse =>
+            {
+                await warehouse.Put("public", new MemoryStore());
+                await warehouse.Put("public/beacon", new BeaconResource());
+            }).WaitAsync(TimeSpan.FromSeconds(10));
+
+        cluster.Connection.AutoReconnect = true;
+        cluster.Connection.ReconnectInterval = 1;
+        _ = await firstAttachment.Task.WaitAsync(TimeSpan.FromSeconds(5));
+
+        foreach (var serverConnection in cluster.Server.Connections.ToArray())
+            serverConnection.Destroy();
+
+        await WaitUntilAsync(() => !cluster.Connection.IsConnected, TimeSpan.FromSeconds(3));
+        var restored = await resumedAttachment.Task.WaitAsync(TimeSpan.FromSeconds(8));
+
+        Assert.Equal("public/beacon", restored.ResourceLink);
+        Assert.True(cluster.Connection.IsConnected);
+    }
+
     [Fact]
     public async Task On_PropertyPrefix_ListensWithNoWireSubscription()
     {
@@ -313,6 +352,20 @@ public class EventSubscriptionIntegrationTests
             await cluster.Connection.Get("sys/beacon"))
             .WaitAsync(TimeSpan.FromSeconds(10));
 
+    static async Task ResolveClientResourceAsync(
+        EpConnection connection,
+        TaskCompletionSource<EpResource> completion)
+    {
+        try
+        {
+            completion.TrySetResult((EpResource)await connection.Get("public/beacon"));
+        }
+        catch (Exception exception)
+        {
+            completion.TrySetException(exception);
+        }
+    }
+
     static async Task WaitUntilAsync(Func<bool> condition, TimeSpan timeout)
     {
         var deadline = DateTime.UtcNow + timeout;