refactor: split endpoint reader and writer (#2392)

Author: rogeralsing

logs/exceptions.md

@@ -34,3 +34,6 @@
 
 ### Proto.Tests.ActorTests.StopActorWithLongRunningTask
 `Proto.TestKit.TestKitException: Expected user message of type System.Threading.Tasks.TaskCanceledException, but received system message of type Proto.Stopping`
+
+### Proto.Cluster.Tests.GossipCoreTests.Large_cluster_should_get_topology_consensus
+`Expected x.consensus to be True, but found False.`

logs/log1756126139.md

@@ -0,0 +1,5 @@
+## Log 1756126139
+- Split connection reader and writer loops into dedicated `ConnectionReader` and `ConnectionWriter` types under `Endpoints/Runner` to improve separation of concerns and testability.
+- Updated `ConnectionRunner` to orchestrate the new types and moved it into the new `Runner` directory.
+- Verified .NET SDK installation and rebuilt solution.
+- Ran core test suites; `Proto.Actor.Tests` and `Proto.Remote.Tests` passed, while `Proto.Cluster.Tests` had one failing test (`GossipCoreTests.Large_cluster_should_get_topology_consensus`).

src/Proto.Remote/Endpoints/Runner/ConnectionReader.cs

@@ -0,0 +1,64 @@
+namespace Proto.Remote;
+
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+using Grpc.Core;
+using Microsoft.Extensions.Logging;
+
+/// <summary>
+/// Handles incoming messages from the remote endpoint.
+/// </summary>
+internal sealed class ConnectionReader
+{
+    private readonly string _address;
+    private readonly ActorSystem _system;
+    private readonly IConnectionMode _mode;
+    private readonly ILogger _logger;
+
+    public ConnectionReader(string address, ActorSystem system, IConnectionMode mode, ILogger logger)
+    {
+        _address = address;
+        _system = system;
+        _mode = mode;
+        _logger = logger;
+    }
+
+    public async Task RunAsync(CancellationToken token, AsyncDuplexStreamingCall<RemoteMessage, RemoteMessage> call, string actorSystemId, CancellationTokenSource cts)
+    {
+        try
+        {
+            while (await call.ResponseStream.MoveNext(token).ConfigureAwait(false))
+            {
+                var currentMessage = call.ResponseStream.Current;
+                switch (currentMessage.MessageTypeCase)
+                {
+                    case RemoteMessage.MessageTypeOneofCase.DisconnectRequest:
+                        _logger.ReceivedDisconnectionRequest(_system.Address, _address);
+                        var terminated = new EndpointTerminatedEvent(false, _address, actorSystemId);
+                        _system.EventStream.Publish(terminated);
+                        break;
+                    default:
+                        _mode.HandleMessage(currentMessage, _address);
+                        break;
+                }
+            }
+
+            _logger.ReaderFinished(_system.Address, _address);
+        }
+        catch (OperationCanceledException)
+        {
+            _logger.ReaderCancelledDebug(_system.Address, _address);
+        }
+        catch (RpcException e) when (e.StatusCode == StatusCode.Cancelled)
+        {
+            _logger.ReaderCancelledWarning(_system.Address, _address);
+        }
+        catch (Exception e)
+        {
+            _logger.ReaderError(_system.Address, _address, e.GetType().Name);
+            cts.Cancel();
+            throw;
+        }
+    }
+}

…oto.Remote/Endpoints/ConnectionRunner.cs …ote/Endpoints/Runner/ConnectionRunner.cssrc/Proto.Remote/Endpoints/ConnectionRunner.cs renamed to src/Proto.Remote/Endpoints/Runner/ConnectionRunner.cs src/Proto.Remote/Endpoints/ConnectionRunner.cs renamed to src/Proto.Remote/Endpoints/Runner/ConnectionRunner.cs

@@ -1,8 +1,6 @@
 namespace Proto.Remote;
 
 using System;
-using System.Diagnostics;
-using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
 using Grpc.Core;
@@ -25,6 +23,8 @@ public sealed class ConnectionRunner
     private readonly Action _onDisconnected;
     private readonly Action<double> _recordWriteDuration;
     private readonly ILogger _logger;
+    private readonly ConnectionWriter _writer;
+    private readonly ConnectionReader _reader;
 
     public ConnectionRunner(string address, IEndpoint endpoint, ActorSystem system, RemoteConfig remoteConfig,
         IConnectionMode mode, TimeSpan backoff, int maxRetries, Random random, CancellationToken stopToken,
@@ -43,6 +43,9 @@ public ConnectionRunner(string address, IEndpoint endpoint, ActorSystem system,
         _onDisconnected = onDisconnected;
         _recordWriteDuration = recordWriteDuration;
         _logger = logger;
+
+        _writer = new ConnectionWriter(address, endpoint, system, remoteConfig, recordWriteDuration, logger);
+        _reader = new ConnectionReader(address, system, mode, logger);
     }
 
     public async Task RunAsync()
@@ -96,8 +99,8 @@ public async Task RunAsync()
 
                 var combinedToken = CancellationTokenSource.CreateLinkedTokenSource(_stopToken, cts.Token).Token;
 
-                var writer = RunWriterAsync(combinedToken, call, cts);
-                var reader = RunReaderAsync(combinedToken, call, actorSystemId, cts);
+                var writer = _writer.RunAsync(combinedToken, call, cts);
+                var reader = _reader.RunAsync(combinedToken, call, actorSystemId, cts);
 
                 _logger.Connected(_system.Address, _address);
 
@@ -150,92 +153,6 @@ public async Task RunAsync()
         }
     }
 
-    private async Task RunWriterAsync(CancellationToken token, AsyncDuplexStreamingCall<RemoteMessage, RemoteMessage> call, CancellationTokenSource cts)
-    {
-        while (!token.IsCancellationRequested)
-        {
-            while (_endpoint.OutgoingStash.TryPop(out var messages))
-            {
-                var batch = MessageBatchFactory.CreateBatch(_system, _remoteConfig, messages);
-                try
-                {
-                    var sw = Stopwatch.StartNew();
-                    await call.RequestStream.WriteAsync(new RemoteMessage { MessageBatch = batch }, token).ConfigureAwait(false);
-                    sw.Stop();
-                    _recordWriteDuration(sw.Elapsed.TotalSeconds);
-                }
-                catch (Exception)
-                {
-                    _ = _endpoint.OutgoingStash.Append(messages);
-                    cts.Cancel();
-                    throw;
-                }
-            }
-
-            try
-            {
-                await foreach (var messages in _endpoint.Outgoing.Reader.ReadAllAsync(token).ConfigureAwait(false))
-                {
-                    var batch = MessageBatchFactory.CreateBatch(_system, _remoteConfig, messages);
-                    try
-                    {
-                        var sw = Stopwatch.StartNew();
-                        await call.RequestStream.WriteAsync(new RemoteMessage { MessageBatch = batch }, token).ConfigureAwait(false);
-                        sw.Stop();
-                        _recordWriteDuration(sw.Elapsed.TotalSeconds);
-                    }
-                    catch (Exception)
-                    {
-                        _ = _endpoint.OutgoingStash.Append(messages);
-                        cts.Cancel();
-                        throw;
-                    }
-                }
-            }
-            catch (OperationCanceledException)
-            {
-                _logger.WriterCancelled(_system.Address, _address);
-            }
-        }
-    }
-
-    private async Task RunReaderAsync(CancellationToken token, AsyncDuplexStreamingCall<RemoteMessage, RemoteMessage> call, string actorSystemId, CancellationTokenSource cts)
-    {
-        try
-        {
-            while (await call.ResponseStream.MoveNext(token).ConfigureAwait(false))
-            {
-                var currentMessage = call.ResponseStream.Current;
-                switch (currentMessage.MessageTypeCase)
-                {
-                    case RemoteMessage.MessageTypeOneofCase.DisconnectRequest:
-                        _logger.ReceivedDisconnectionRequest(_system.Address, _address);
-                        var terminated = new EndpointTerminatedEvent(false, _address, actorSystemId);
-                        _system.EventStream.Publish(terminated);
-                        break;
-                    default:
-                        _mode.HandleMessage(currentMessage, _address);
-                        break;
-                }
-            }
-
-            _logger.ReaderFinished(_system.Address, _address);
-        }
-        catch (OperationCanceledException)
-        {
-            _logger.ReaderCancelledDebug(_system.Address, _address);
-        }
-        catch (RpcException e) when (e.StatusCode == StatusCode.Cancelled)
-        {
-            _logger.ReaderCancelledWarning(_system.Address, _address);
-        }
-        catch (Exception e)
-        {
-            _logger.ReaderError(_system.Address, _address, e.GetType().Name);
-            cts.Cancel();
-            throw;
-        }
-    }
 
     private bool ShouldStop(RestartStatistics rs)
     {

src/Proto.Remote/Endpoints/Runner/ConnectionWriter.cs

@@ -0,0 +1,82 @@
+namespace Proto.Remote;
+
+using System;
+using System.Diagnostics;
+using System.Threading;
+using System.Threading.Tasks;
+using Grpc.Core;
+using Microsoft.Extensions.Logging;
+using Proto.Extensions;
+using System.Linq;
+
+/// <summary>
+/// Handles sending message batches to the remote endpoint.
+/// </summary>
+internal sealed class ConnectionWriter
+{
+    private readonly string _address;
+    private readonly IEndpoint _endpoint;
+    private readonly RemoteConfig _remoteConfig;
+    private readonly ActorSystem _system;
+    private readonly Action<double> _recordWriteDuration;
+    private readonly ILogger _logger;
+
+    public ConnectionWriter(string address, IEndpoint endpoint, ActorSystem system, RemoteConfig remoteConfig, Action<double> recordWriteDuration, ILogger logger)
+    {
+        _address = address;
+        _endpoint = endpoint;
+        _system = system;
+        _remoteConfig = remoteConfig;
+        _recordWriteDuration = recordWriteDuration;
+        _logger = logger;
+    }
+
+    public async Task RunAsync(CancellationToken token, AsyncDuplexStreamingCall<RemoteMessage, RemoteMessage> call, CancellationTokenSource cts)
+    {
+        while (!token.IsCancellationRequested)
+        {
+            while (_endpoint.OutgoingStash.TryPop(out var messages))
+            {
+                var batch = MessageBatchFactory.CreateBatch(_system, _remoteConfig, messages);
+                try
+                {
+                    var sw = Stopwatch.StartNew();
+                    await call.RequestStream.WriteAsync(new RemoteMessage { MessageBatch = batch }, token).ConfigureAwait(false);
+                    sw.Stop();
+                    _recordWriteDuration(sw.Elapsed.TotalSeconds);
+                }
+                catch (Exception)
+                {
+                    _ = _endpoint.OutgoingStash.Append(messages);
+                    cts.Cancel();
+                    throw;
+                }
+            }
+
+            try
+            {
+                await foreach (var messages in _endpoint.Outgoing.Reader.ReadAllAsync(token).ConfigureAwait(false))
+                {
+                    var batch = MessageBatchFactory.CreateBatch(_system, _remoteConfig, messages);
+                    try
+                    {
+                        var sw = Stopwatch.StartNew();
+                        await call.RequestStream.WriteAsync(new RemoteMessage { MessageBatch = batch }, token).ConfigureAwait(false);
+                        sw.Stop();
+                        _recordWriteDuration(sw.Elapsed.TotalSeconds);
+                    }
+                    catch (Exception)
+                    {
+                        _ = _endpoint.OutgoingStash.Append(messages);
+                        cts.Cancel();
+                        throw;
+                    }
+                }
+            }
+            catch (OperationCanceledException)
+            {
+                _logger.WriterCancelled(_system.Address, _address);
+            }
+        }
+    }
+}