diff --git a/src/core/Akka.Remote.Tests/Transport/DotNettyGracefulCloseSpec.cs b/src/core/Akka.Remote.Tests/Transport/DotNettyGracefulCloseSpec.cs new file mode 100644 index 00000000000..537c8f2bfcb --- /dev/null +++ b/src/core/Akka.Remote.Tests/Transport/DotNettyGracefulCloseSpec.cs @@ -0,0 +1,304 @@ +//----------------------------------------------------------------------- +// +// Copyright (C) 2009-2022 Lightbend Inc. +// Copyright (C) 2013-2025 .NET Foundation +// +//----------------------------------------------------------------------- + +#nullable enable +using System; +using System.Diagnostics; +using System.IO; +using System.Linq; +using System.Net; +using System.Net.Security; +using System.Net.Sockets; +using System.Threading; +using System.Threading.Tasks; +using Akka.Actor; +using Akka.Configuration; +using Akka.Remote.Transport; +using Akka.Remote.Transport.DotNetty; +using Akka.TestKit; +using DotNetty.Transport.Channels; +using FluentAssertions; +using Google.Protobuf; +using Xunit; + +namespace Akka.Remote.Tests.Transport +{ + /// + /// A graceful disassociate must not reset the connection, or a Windows peer drops our last frames (#8589). + /// + public class DotNettyGracefulCloseSpec : AkkaSpec + { + private static readonly Config Config = ConfigurationFactory.ParseString(@" + akka { + loglevel = INFO + actor.provider = remote + remote.dot-netty.tcp { + port = 0 + hostname = ""127.0.0.1"" + } + }"); + + private static readonly Config ShortFlushWait = + ConfigurationFactory.ParseString("akka.remote.flush-wait-on-shutdown = 300ms").WithFallback(Config); + + private static readonly Config TlsOverrides = ConfigurationFactory.ParseString(@" + enable-ssl = true + ssl { + suppress-validation = true + require-mutual-authentication = false + certificate { + path = ""Resources/akka-validcert.pfx"" + password = ""password"" + } + }"); + + private static readonly ByteString Payload = ByteString.CopyFrom(Enumerable.Repeat((byte)7, 512).ToArray()); + + public DotNettyGracefulCloseSpec(ITestOutputHelper output) : base(Config, output) + { + } + + [Fact(DisplayName = "Should_deliver_last_frame_and_clean_EOF_When_disassociating_with_unread_inbound_data")] + public async Task Should_deliver_last_frame_and_clean_EOF_When_disassociating_with_unread_inbound_data() + { + var c = await ConnectRawPeer(Sys); + try + { + // we never read these bytes; zeros decode as empty frames if we do + c.Peer.Send(new byte[1024]); + + c.Handle.Write(Payload).Should().BeTrue(); + c.Handle.Disassociate("test", Log); + + (await ReadToEnd(new NetworkStream(c.Peer))).Should().Be(4 + Payload.Length); + await Task.Delay(200); + SocketError(c.Peer).Should().Be(0, "a reset makes a Windows peer drop the frames it has not read yet"); + + // our side closes once it reads the peer's FIN, well before the 2 s flush-wait deadline + c.Peer.Close(); + await c.Channel.CloseCompletion.WaitAsync(TimeSpan.FromSeconds(1)); + } + finally + { + c.Peer.Dispose(); + await c.Transport.Shutdown(); + } + } + + [Fact(DisplayName = "Should_deliver_last_frame_without_reset_When_TLS_is_enabled")] + public async Task Should_deliver_last_frame_without_reset_When_TLS_is_enabled() + { + var c = await ConnectRawPeer(Sys, TlsOverrides.WithFallback(Sys.Settings.Config.GetConfig("akka.remote.dot-netty.tcp"))); + try + { + // the server side needs reads on to finish the handshake + c.Handle.ReadHandlerSource.SetResult(new ActorHandleEventListener(CreateTestProbe())); + await using var tls = new SslStream(new NetworkStream(c.Peer), false, (_, _, _, _) => true); + await tls.AuthenticateAsClientAsync("localhost").WaitAsync(TimeSpan.FromSeconds(5)); + + c.Handle.Write(Payload).Should().BeTrue(); + c.Handle.Disassociate("test", Log); + + // traffic that races our FIN must be drained through TlsHandler, not answered with a reset + await tls.WriteAsync(new byte[1024]); + + // no close_notify precedes our FIN; SslStream reports the plain EOF as end of stream + (await ReadToEnd(tls)).Should().Be(4 + Payload.Length); + await Task.Delay(200); + SocketError(c.Peer).Should().Be(0); + + c.Peer.Close(); + await c.Channel.CloseCompletion.WaitAsync(TimeSpan.FromSeconds(1)); + } + finally + { + c.Peer.Dispose(); + await c.Transport.Shutdown(); + } + } + + [Fact(DisplayName = "Should_terminate_well_under_flush_wait_When_an_association_is_open")] + public async Task Should_terminate_well_under_flush_wait_When_an_association_is_open() + { + var config = ConfigurationFactory.ParseString("akka.remote.flush-wait-on-shutdown = 5s").WithFallback(Config); + var a = ActorSystem.Create("GracefulA", config); + var b = ActorSystem.Create("GracefulB", config); + try + { + var probe = CreateTestProbe(a); + var bAddress = ((ExtendedActorSystem)b).Provider.DefaultAddress; + a.ActorSelection(new RootActorPath(bAddress) / "user" / "missing").Tell(new Identify(1), probe.Ref); + await probe.ExpectMsgAsync(); + + // the peer answers our FIN at once, so no graceful close should sit out the 5 s flush-wait + var sw = Stopwatch.StartNew(); + await a.Terminate().WaitAsync(TimeSpan.FromSeconds(15)); + sw.Elapsed.Should().BeLessThan(TimeSpan.FromSeconds(3)); + } + finally + { + await ShutdownAsync(a); + await ShutdownAsync(b); + } + } + + [Fact(DisplayName = "Should_find_the_DotNetty_socket_field_the_half_close_depends_on")] + public void Should_find_the_DotNetty_socket_field_the_half_close_depends_on() + { + // a DotNetty upgrade that renames this field silently turns the graceful close back into a reset + DotNettyTransport.SocketField.Should().NotBeNull(); + DotNettyTransport.SocketField!.FieldType.Should().Be(typeof(Socket)); + } + + [Fact(DisplayName = "Should_force_close_after_flush_wait_When_peer_never_closes")] + public async Task Should_force_close_after_flush_wait_When_peer_never_closes() + { + var sys = ActorSystem.Create("ShortFlushWait", ShortFlushWait); + try + { + var c = await ConnectRawPeer(sys); + try + { + c.Handle.Disassociate("test", Log); + await c.Channel.CloseCompletion.WaitAsync(TimeSpan.FromSeconds(1.5)); + } + finally + { + c.Peer.Dispose(); + await c.Transport.Shutdown(); + } + } + finally + { + await ShutdownAsync(sys); + } + } + + [Fact(DisplayName = "Should_refuse_writes_When_disassociated")] + public async Task Should_refuse_writes_When_disassociated() + { + var c = await ConnectRawPeer(Sys); + try + { + c.Handle.Disassociate("test", Log); + c.Handle.Write(Payload).Should().BeFalse(); + } + finally + { + c.Peer.Dispose(); + await c.Transport.Shutdown(); + } + } + + [Fact(DisplayName = "Should_finish_transport_shutdown_within_flush_wait_When_a_drain_is_stuck")] + public async Task Should_finish_transport_shutdown_within_flush_wait_When_a_drain_is_stuck() + { + var sys = ActorSystem.Create("ShortFlushWait", ShortFlushWait); + try + { + var c = await ConnectRawPeer(sys); + try + { + c.Handle.Disassociate("test", Log); + + var sw = Stopwatch.StartNew(); + await c.Transport.Shutdown().WaitAsync(TimeSpan.FromSeconds(3)); + sw.Elapsed.Should().BeLessThan(TimeSpan.FromSeconds(1.5)); + c.Channel.Open.Should().BeFalse(); + } + finally + { + c.Peer.Dispose(); + await c.Transport.Shutdown(); + } + } + finally + { + await ShutdownAsync(sys); + } + } + + [Fact(DisplayName = "Should_close_both_sides_without_reset_When_both_disassociate_at_once")] + public async Task Should_close_both_sides_without_reset_When_both_disassociate_at_once() + { + var config = Sys.Settings.Config.GetConfig("akka.remote.dot-netty.tcp"); + var t1 = new TcpTransport(Sys, config); + var t2 = new TcpTransport(Sys, config); + try + { + var p1 = CreateTestProbe(); + var p2 = CreateTestProbe(); + var (_, l1) = await t1.Listen(); + l1.SetResult(new ActorAssociationEventListener(p1)); + var (a2, l2) = await t2.Listen(); + l2.SetResult(new ActorAssociationEventListener(p2)); + + var outbound = await t1.Associate(a2); + outbound.ReadHandlerSource.SetResult(new ActorHandleEventListener(p1)); + var inbound = (await p2.ExpectMsgAsync()).Association; + inbound.ReadHandlerSource.SetResult(new ActorHandleEventListener(p2)); + + await AwaitConditionAsync(() => Task.FromResult(t1.ConnectionGroup.Count == 2 && t2.ConnectionGroup.Count == 2)); + var chan1 = t1.ConnectionGroup.Single(x => !ReferenceEquals(x, t1.ServerChannel)); + var chan2 = t2.ConnectionGroup.Single(x => !ReferenceEquals(x, t2.ServerChannel)); + + await EventFilter.Info(contains: "reset").ExpectAsync(0, async () => + { + for (var i = 0; i < 10; i++) + { + outbound.Write(Payload); + inbound.Write(Payload); + } + + await Task.WhenAll( + Task.Run(() => outbound.Disassociate("test", Log)), + Task.Run(() => inbound.Disassociate("test", Log))); + + // both sides see the other's FIN; neither waits for the 2 s flush-wait deadline + await Task.WhenAll(chan1.CloseCompletion, chan2.CloseCompletion).WaitAsync(TimeSpan.FromSeconds(1)); + }); + } + finally + { + await t1.Shutdown(); + await t2.Shutdown(); + } + } + + private sealed record Connection(TcpTransport Transport, AssociationHandle Handle, IChannel Channel, Socket Peer); + + // A raw socket connects in. No ReadHandlerSource is set, so our side never turns AutoRead on. + private async Task ConnectRawPeer(ActorSystem system, Config? transportConfig = null) + { + var transport = new TcpTransport(system, transportConfig ?? system.Settings.Config.GetConfig("akka.remote.dot-netty.tcp")); + var probe = CreateTestProbe(system); + var (address, listener) = await transport.Listen(); + listener.SetResult(new ActorAssociationEventListener(probe)); + + var peer = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp); + await peer.ConnectAsync(IPAddress.Loopback, address.Port!.Value).WaitAsync(TimeSpan.FromSeconds(3)); + var handle = (await probe.ExpectMsgAsync()).Association; + await AwaitConditionAsync(() => Task.FromResult(transport.ConnectionGroup.Count == 2)); + var channel = transport.ConnectionGroup.Single(x => !ReferenceEquals(x, transport.ServerChannel)); + return new Connection(transport, handle, channel, peer); + } + + private static async Task ReadToEnd(Stream peer) + { + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(3)); + var buffer = new byte[4096]; + var total = 0; + int n; + while ((n = await peer.ReadAsync(buffer, cts.Token)) > 0) + total += n; + return total; + } + + private static int SocketError(Socket peer) => + (int)peer.GetSocketOption(SocketOptionLevel.Socket, SocketOptionName.Error)!; + } +} diff --git a/src/core/Akka.Remote/Transport/DotNetty/DotNettyTransport.cs b/src/core/Akka.Remote/Transport/DotNetty/DotNettyTransport.cs index 4baf4499431..d0b1472a2fa 100644 --- a/src/core/Akka.Remote/Transport/DotNetty/DotNettyTransport.cs +++ b/src/core/Akka.Remote/Transport/DotNetty/DotNettyTransport.cs @@ -7,11 +7,13 @@ #nullable enable using System; +using System.Collections.Concurrent; using System.Collections.Generic; using System.Linq; using System.Net; using System.Net.Security; using System.Net.Sockets; +using System.Reflection; using System.Runtime.Serialization; using System.Security.Cryptography.X509Certificates; using System.Threading.Tasks; @@ -22,6 +24,7 @@ using DotNetty.Buffers; using DotNetty.Codecs; using ByteOrder = DotNetty.Buffers.ByteOrder; +using DotNetty.Common.Concurrency; using DotNetty.Common.Utilities; using DotNetty.Handlers.Tls; using DotNetty.Transport.Bootstrapping; @@ -127,6 +130,11 @@ internal abstract class DotNettyTransport : Transport private readonly IEventLoopGroup _serverEventLoopGroup; private readonly IEventLoopGroup _clientEventLoopGroup; + private readonly TimeSpan _flushWait; + private readonly ConcurrentDictionary _draining = new(); + + internal static readonly FieldInfo? SocketField = + typeof(AbstractSocketChannel).GetField("Socket", BindingFlags.Instance | BindingFlags.NonPublic); protected DotNettyTransport(ActorSystem system, Config config) { @@ -149,6 +157,9 @@ protected DotNettyTransport(ActorSystem system, Config config) _clientEventLoopGroup = new MultithreadEventLoopGroup(Settings.ClientSocketWorkerPoolSize); ConnectionGroup = new ConcurrentSet(); AssociationListenerPromise = new TaskCompletionSource(); + _flushWait = system.Settings.Config.GetTimeSpan("akka.remote.flush-wait-on-shutdown", TimeSpan.FromSeconds(2)); + if (SocketField == null) + Log.Warning("DotNetty's AbstractSocketChannel.Socket field was not found; disassociate falls back to a full close, which can reset the connection."); SchemeIdentifier = (Settings.EnableSsl ? "ssl." : string.Empty) + Settings.TransportMode.ToString().ToLowerInvariant(); } @@ -250,10 +261,76 @@ public override Task Associate(Address remoteAddress) protected abstract Task AssociateInternal(Address remoteAddress); + // Half-close once lastWrite is in the kernel, then read until the peer's FIN. Closing with unread + // inbound data sends an RST, and a Windows peer then drops our last frames (#8589). + internal void BeginGracefulClose(TcpSocketChannel channel, Task lastWrite) + { + // register before any async hop, so a Shutdown that follows the disassociate waits for this close + _draining[channel] = channel.CloseCompletion; + channel.CloseCompletion.ContinueWith(_ => _draining.TryRemove(channel, out Task? _), TaskContinuationOptions.ExecuteSynchronously); + + try + { + // not the ActorSystem scheduler: it may already be shutting down + channel.EventLoop.Schedule(() => channel.CloseAsync(), _flushWait); + } + catch (RejectedExecutionException) + { + channel.CloseAsync(); + return; + } + + lastWrite.ContinueWith(_ => + { + try + { + channel.EventLoop.Execute(() => HalfClose(channel)); + } + catch (RejectedExecutionException) + { + channel.CloseAsync(); // event loop already shut down + } + }, TaskContinuationOptions.ExecuteSynchronously); + } + + private void HalfClose(TcpSocketChannel channel) + { + if (!channel.Active) + return; + + channel.Pipeline.FireUserEventTriggered(DrainInbound.Instance); + channel.Configuration.AutoRead = true; + try + { + // Not channel.ShutdownOutputAsync(): Socket.Shutdown clears Socket.Connected, which DotNetty reads as + // EOF, so it would close over unread data. Shut down through a second Socket on the same handle. + if (SocketField?.GetValue(channel) is not Socket socket) + { + channel.CloseAsync(); + return; + } + + using var alias = new Socket(new SafeSocketHandle(socket.Handle, ownsHandle: false)); + // Shutdown re-applies the alias's blocking mode to the shared handle, and on Windows the + // handle constructor assumes blocking; DotNetty's socket must stay non-blocking. + alias.Blocking = false; + alias.Shutdown(SocketShutdown.Send); + } + catch (Exception ex) + { + Log.Warning(ex, "Half-close failed on channel [{0}->{1}]; closing it outright", channel.LocalAddress, channel.RemoteAddress); + channel.CloseAsync(); + } + } + public override async Task Shutdown() { try { + // give graceful closes up to flush-wait to finish before the force-close below + if (!_draining.IsEmpty) + await Task.WhenAny(Task.WhenAll(_draining.Values), Task.Delay(_flushWait)).ConfigureAwait(false); + var tasks = new List(); foreach (var channel in ConnectionGroup) { diff --git a/src/core/Akka.Remote/Transport/DotNetty/TcpTransport.cs b/src/core/Akka.Remote/Transport/DotNetty/TcpTransport.cs index 11568123f58..dfdb29fbe2f 100644 --- a/src/core/Akka.Remote/Transport/DotNetty/TcpTransport.cs +++ b/src/core/Akka.Remote/Transport/DotNetty/TcpTransport.cs @@ -17,6 +17,7 @@ using DotNetty.Common.Utilities; using DotNetty.Handlers.Tls; using DotNetty.Transport.Channels; +using DotNetty.Transport.Channels.Sockets; using Google.Protobuf; namespace Akka.Remote.Transport.DotNetty @@ -35,9 +36,17 @@ public TlsHandshakeFailureReason(string message) public override string ToString() => Message; } + // Fired on a channel that is closing gracefully: discard whatever the peer still sends. + internal sealed class DrainInbound + { + public static readonly DrainInbound Instance = new(); + private DrainInbound() { } + } + internal abstract class TcpHandlers : CommonHandlers { private IHandleEventListener _listener; + private bool _draining; // only touched on the event loop protected void NotifyListener(IHandleEvent msg) { @@ -67,7 +76,7 @@ public override void ChannelInactive(IChannelHandlerContext context) public override void ChannelRead(IChannelHandlerContext context, object message) { var buf = ((IByteBuffer)message); - if (buf.ReadableBytes > 0) + if (!_draining && buf.ReadableBytes > 0) { // no need to copy the byte buffer contents; ByteString does that automatically var bytes = ByteString.CopyFrom(buf.Array, buf.ArrayOffset + buf.ReaderIndex, buf.ReadableBytes); @@ -80,6 +89,12 @@ public override void ChannelRead(IChannelHandlerContext context, object message) public override void UserEventTriggered(IChannelHandlerContext context, object evt) { + if (evt is DrainInbound) + { + _draining = true; + return; + } + if (evt is TlsHandshakeCompletionEvent { IsSuccessful: false } tlsEvent) { var ex = tlsEvent.Exception ?? new Exception("TLS handshake failed."); @@ -244,22 +259,29 @@ private void InitOutbound(IChannel channel, IPEndPoint socketAddress, object msg internal sealed class TcpAssociationHandle : AssociationHandle { private readonly IChannel _channel; + private readonly DotNettyTransport _transport; + private readonly object _gate = new(); + private Task _lastWrite = Task.CompletedTask; + private bool _closing; public TcpAssociationHandle(Address localAddress, Address remoteAddress, DotNettyTransport transport, IChannel channel) : base(localAddress, remoteAddress) { _channel = channel; + _transport = transport; } public override bool Write(ByteString payload) { - if (_channel.Open) + lock (_gate) { - var data = ToByteBuffer(_channel, payload); - _channel.WriteAndFlushAsync(data); + // a write after the output shutdown fails in DoWrite and hard-closes the socket + if (_closing || !_channel.Open) + return false; + + _lastWrite = _channel.WriteAndFlushAsync(ToByteBuffer(_channel, payload)); return true; } - return false; } [MethodImpl(MethodImplOptions.AggressiveInlining)] @@ -272,7 +294,19 @@ private static IByteBuffer ToByteBuffer(IChannel channel, ByteString payload) public override void Disassociate() #pragma warning restore CS0672 { - _channel.CloseAsync(); + Task lastWrite; + lock (_gate) + { + if (_closing) + return; + _closing = true; + lastWrite = _lastWrite; + } + + if (_channel is TcpSocketChannel { Active: true } tcp) + _transport.BeginGracefulClose(tcp, lastWrite); + else + _channel.CloseAsync(); } }