Fix disposed scope on TCP request timeout

src/DotNext.Tests/Net/Cluster/Consensus/Raft/NetworkTransport/ConnectionOriented/Tcp/TcpTransportTests.cs

@@ -96,6 +96,28 @@ public Task StressTest()
         return StressTestCore(CreateServer, CreateClient);
     }
 
+    [Fact]
+    public Task RequestTimeout()
+    {
+        static TcpServer CreateServer(ILocalMember member, EndPoint address, TimeSpan timeout) => new(address, 2, member, NullLoggerFactory.Instance)
+        {
+            MemoryAllocator = MemoryAllocator<byte>.Default,
+            ReceiveTimeout = timeout,
+            TransmissionBlockSize = 65535,
+            GracefulShutdownTimeout = 2000
+        };
+
+        static TcpClient CreateClient(EndPoint address, ILocalMember member, TimeSpan timeout) => new(member, address)
+        {
+            MemoryAllocator = MemoryAllocator<byte>.Default,
+            RequestTimeout = timeout,
+            ConnectTimeout = timeout,
+            TransmissionBlockSize = 65535,
+        };
+
+        return RequestTimeoutTest(CreateServer, CreateClient);
+    }
+
     [Theory]
     [InlineData(true)]
     [InlineData(false)]
@@ -393,4 +415,4 @@ private static RaftCluster.TcpConfiguration CreateConfiguration(int port, bool c
 
         return result;
     }
-}
+}

src/DotNext.Tests/Net/Cluster/Consensus/Raft/NetworkTransport/TransportTestSuite.cs

@@ -38,6 +38,7 @@ private sealed class LocalMember : Assert, ILocalMember
         internal ReceiveEntriesBehavior Behavior;
         internal byte[] ReceivedConfiguration = [];
         internal long ReceivedConfigurationVersion = -1L;
+        internal TimeSpan VoteDelay;
         private readonly ClusterMemberId localId = Random.Shared.Next<ClusterMemberId>();
 
         internal LocalMember(bool smallAmountOfMetadata = false)
@@ -125,13 +126,17 @@ async ValueTask<bool> ILocalMember.InstallConfigurationAsync<TConfiguration>(lon
             return true;
         }
 
-        ValueTask<Result<bool>> ILocalMember.VoteAsync(ClusterMemberId sender, long term, long lastLogIndex, long lastLogTerm, CancellationToken token)
+        async ValueTask<Result<bool>> ILocalMember.VoteAsync(ClusterMemberId sender, long term, long lastLogIndex, long lastLogTerm, CancellationToken token)
         {
             True(token.CanBeCanceled);
             Equal(42L, term);
             Equal(1L, lastLogIndex);
             Equal(56L, lastLogTerm);
-            return ValueTask.FromResult<Result<bool>>(new() { Term = 43L, Value = true });
+
+            if (VoteDelay > TimeSpan.Zero)
+                await Task.Delay(VoteDelay, token);
+
+            return new() { Term = 43L, Value = true };
         }
 
         ValueTask<Result<PreVoteResult>> ILocalMember.PreVoteAsync(ClusterMemberId sender, long term, long lastLogIndex, long lastLogTerm, CancellationToken token)
@@ -212,6 +217,27 @@ private protected async Task StressTestCore(ServerFactory serverFactory, ClientF
         });
     }
 
+    private protected async Task RequestTimeoutTest(ServerFactory serverFactory, ClientFactory clientFactory)
+    {
+        var serverAddr = new IPEndPoint(IPAddress.Loopback, 3789);
+        var serverTimeout = TimeSpan.FromMilliseconds(50D);
+        var member = new LocalMember { VoteDelay = DefaultTimeout };
+        await using var server = serverFactory(member, serverAddr, serverTimeout);
+        await server.StartAsync(TestToken);
+
+        using (var client = clientFactory(serverAddr, member, DefaultTimeout))
+        {
+            await ThrowsAsync<MemberUnavailableException>(
+                () => client.As<IRaftClusterMember>().VoteAsync(42L, 1L, 56L, TestToken));
+        }
+
+        member.VoteDelay = TimeSpan.Zero;
+        using var nextClient = clientFactory(serverAddr, member, DefaultTimeout);
+        var result = await nextClient.As<IRaftClusterMember>().VoteAsync(42L, 1L, 56L, TestToken);
+        True(result.Value);
+        Equal(43L, result.Term);
+    }
+
     private protected async Task MetadataRequestResponseTest(ServerFactory serverFactory, ClientFactory clientFactory, bool smallAmountOfMetadata)
     {
         var timeout = DefaultTimeout;
@@ -494,4 +520,4 @@ private protected async Task LeadershipCore(Func<int, bool, IPersistentState, Ra
 
     protected static WriteAheadLog CreateWal()
         => new(new() { Location = GetTempPath() }, IStateMachine.CreateNoOp());
-}
+}

src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/NetworkTransport/ConnectionOriented/Tcp/TcpServer.cs

@@ -127,15 +127,11 @@ private async void HandleConnection(Socket remoteClient)
                     break;
 
                 timeoutSource = multiplexer.Combine(receiveTimeout, lifecycleToken);
-                try
-                {
-                    await ProcessRequestAsync(messageType, protocol, timeoutSource.Token).ConfigureAwait(false);
-                }
-                finally
-                {
-                    // reset cancellation token
-                    await timeoutSource.DisposeAsync().ConfigureAwait(false);
-                }
+                await ProcessRequestAsync(messageType, protocol, timeoutSource.Token).ConfigureAwait(false);
+
+                // reset cancellation token
+                await timeoutSource.DisposeAsync().ConfigureAwait(false);
+                timeoutSource = default;
             }
         }
         catch (Exception e) when (e is SocketException { SocketErrorCode: SocketError.ConnectionReset } or { InnerException: SocketException { SocketErrorCode: SocketError.ConnectionReset } })
@@ -154,6 +150,7 @@ private async void HandleConnection(Socket remoteClient)
         }
         finally
         {
+            await timeoutSource.DisposeAsync().ConfigureAwait(false);
             await protocol.DisposeAsync().ConfigureAwait(false);
             if (protocol.BaseStream is SslStream ssl)
                 await ssl.DisposeAsync().ConfigureAwait(false);
@@ -267,4 +264,4 @@ protected override async ValueTask DisposeAsyncCore()
             logger.TcpGracefulShutdownFailed(GracefulShutdownTimeout);
         }
     }
-}
+}