Akka.IO TCP: hold back WriteAck until the output pipe's flush completes

BREAKING_CHANGES_V1.6.md

@@ -40,6 +40,7 @@ of `Behavior`, `Wire`, `API` (combine with `+`).
 
 | Status | PR / Branch | Component | Type | Change | Migration |
 |--------|-------------|-----------|------|--------|-----------|
+| Merged | [#8646](https://github.com/akkadotnet/akka.net/pull/8646) | `Akka.IO` | Behavior | A TCP `WriteAck` now means the connection's output pipe is back below its resume mark (32 KB by default, or `Inet.SO.PipeBufferSize` when set) after this write, so acks can arrive later under load; a single large write may overshoot the pause mark. While a flush is pending, later writes (ack or `NoAck`) are still copied straight into the pipe -- `write-commands-queue-max-size` still caps a single write -- and their acks wait for the next flush, which batches whatever arrived in the meantime. Closes #8617. | None for ack-driven writers. `NoAck` writers that need a memory cap should use acks. |
 | Planned | [#8654](https://github.com/akkadotnet/akka.net/pull/8654) | `Akka` (core / `Akka.Event.LoggingBus`) | Behavior | Adds a first-party logger table to `LoggingBus.StartDefaultLoggers`, checked after the built-in table and before the `Type.GetType` fallback. Today it carries one row, Akka.Hosting's `LoggerFactoryLogger`, matched on type name and assembly with identity stripped (version, culture, public key token), the same as `TypeExtensions.ToBuiltInAkkaTypeName` does for built-in types (#8613). Two observable changes. (1) With `Akka.DynamicTypeLoading` on, a `LoggerFactoryLogger` name pinned to a different Akka.Hosting version -- what HOCON written against one Hosting build looks like once a newer build is loaded -- now resolves, where it previously fell through to `Type.GetType` and failed on the version mismatch. (2) With the switch off, the assembly-qualified name `AddLoggerFactory()` writes into `akka.loggers` now resolves instead of throwing, which is what the amended #8603 row below reflects. A name that matches neither table falls through exactly as before in both switch states. | None. |
 | Merged | [#8613](https://github.com/akkadotnet/akka.net/pull/8613) | `Akka` (core) | Behavior | Every `BuiltIn*` type-name table (loggers, schedulers, mailboxes, message-queue semantics, routers, supervisor-strategy configurators, serializers and serialization bindings, the DNS provider) now carries one key per type instead of two, and the lookup normalizes the configured value through a new `TypeExtensions.ToBuiltInAkkaTypeName` that accepts every spelling `Type.GetType` itself accepted for an Akka.NET type -- any assembly identity, an optional space after the comma, and any casing of the `Akka` assembly name. With `Akka.DynamicTypeLoading` off, a value like `"Akka.Dispatch.BoundedMailbox,akka"` now resolves where it previously threw a `ConfigurationException`; with the switch on, behavior is unchanged. | None. |
 | Merged | [#8608](https://github.com/akkadotnet/akka.net/pull/8608) | `Akka` (package MSBuild logic) | Behavior | The `Akka` package now ships `buildTransitive/Akka.targets`, which adds `<RuntimeHostConfigurationOption Include="Akka.DynamicTypeLoading" Value="false" Trim="true" />` whenever the consuming project sets `PublishAot=true` or `PublishTrimmed=true`. Consumers publishing for Native AOT or trimming therefore get the switch **off** automatically, without knowing it exists: core resolves the type names its own `akka.conf` uses from the built-in tables, the trimmer removes every reflection fallback behind the switch, and any other HOCON type name throws a `ConfigurationException` naming the setting instead of failing obscurely. Ordinary untrimmed builds are untouched -- the condition does not match, no item is added, and the switch keeps its default of on. The change is observable for **any** trimmed or Native AOT publish. The affected population is applications that hand-rooted a HOCON-named type -- with `TrimmerRootAssembly`, `TrimmerRootDescriptor` or `DynamicDependency` -- and relied on `Type.GetType` then finding it; that worked under ILC as well as under ILLink (the rooted type survives and is resolvable at runtime; all you got was an `IL2057` warning at build time), and it now throws. | None for an untrimmed application. A trimmed-but-not-AOT application that relied on hand-rooting a custom logger, mailbox, dispatcher, router, serializer, scheduler or provider must either move that type to the corresponding `Setup` API, or opt back out by declaring the option itself -- `<RuntimeHostConfigurationOption Include="Akka.DynamicTypeLoading" Value="true" />` -- which wins over the package's value. |

src/core/Akka.API.Tests/verify/CoreAPISpec.ApproveCore.DotNet.verified.txt

@@ -3767,8 +3767,7 @@ namespace Akka.IO
         System.Threading.Tasks.Task CloseAsync();
         System.Threading.Tasks.ValueTask<System.IO.Pipelines.FlushResult> FlushAsync(System.Threading.CancellationToken ct = default);
         System.Threading.Tasks.Task ShutdownAsync();
-        System.Threading.Tasks.ValueTask<System.IO.Pipelines.FlushResult> WriteAsync(System.Buffers.ReadOnlySequence<byte> data, System.Threading.CancellationToken ct = default);
-        System.Threading.Tasks.ValueTask<System.IO.Pipelines.FlushResult> WriteAsync(System.ReadOnlyMemory<byte> data, System.Threading.CancellationToken ct = default);
+        void Write(System.Buffers.ReadOnlySequence<byte> data);
     }
     public class Inet
     {
@@ -4163,8 +4162,7 @@ namespace Akka.IO
         public System.Threading.Tasks.ValueTask DisposeAsync() { }
         public System.Threading.Tasks.ValueTask<System.IO.Pipelines.FlushResult> FlushAsync(System.Threading.CancellationToken ct = default) { }
         public System.Threading.Tasks.Task ShutdownAsync() { }
-        public System.Threading.Tasks.ValueTask<System.IO.Pipelines.FlushResult> WriteAsync(System.Buffers.ReadOnlySequence<byte> data, System.Threading.CancellationToken ct = default) { }
-        public System.Threading.Tasks.ValueTask<System.IO.Pipelines.FlushResult> WriteAsync(System.ReadOnlyMemory<byte> data, System.Threading.CancellationToken ct = default) { }
+        public void Write(System.Buffers.ReadOnlySequence<byte> data) { }
     }
     public class Udp : Akka.Actor.ExtensionIdProvider<Akka.IO.UdpExt>
     {

src/core/Akka.API.Tests/verify/CoreAPISpec.ApproveCore.Net.verified.txt

@@ -3764,8 +3764,7 @@ namespace Akka.IO
         System.Threading.Tasks.Task CloseAsync();
         System.Threading.Tasks.ValueTask<System.IO.Pipelines.FlushResult> FlushAsync(System.Threading.CancellationToken ct = default);
         System.Threading.Tasks.Task ShutdownAsync();
-        System.Threading.Tasks.ValueTask<System.IO.Pipelines.FlushResult> WriteAsync(System.Buffers.ReadOnlySequence<byte> data, System.Threading.CancellationToken ct = default);
-        System.Threading.Tasks.ValueTask<System.IO.Pipelines.FlushResult> WriteAsync(System.ReadOnlyMemory<byte> data, System.Threading.CancellationToken ct = default);
+        void Write(System.Buffers.ReadOnlySequence<byte> data);
     }
     public class Inet
     {
@@ -4160,8 +4159,7 @@ namespace Akka.IO
         public System.Threading.Tasks.ValueTask DisposeAsync() { }
         public System.Threading.Tasks.ValueTask<System.IO.Pipelines.FlushResult> FlushAsync(System.Threading.CancellationToken ct = default) { }
         public System.Threading.Tasks.Task ShutdownAsync() { }
-        public System.Threading.Tasks.ValueTask<System.IO.Pipelines.FlushResult> WriteAsync(System.Buffers.ReadOnlySequence<byte> data, System.Threading.CancellationToken ct = default) { }
-        public System.Threading.Tasks.ValueTask<System.IO.Pipelines.FlushResult> WriteAsync(System.ReadOnlyMemory<byte> data, System.Threading.CancellationToken ct = default) { }
+        public void Write(System.Buffers.ReadOnlySequence<byte> data) { }
     }
     public class Udp : Akka.Actor.ExtensionIdProvider<Akka.IO.UdpExt>
     {

src/core/Akka.Tests/IO/BlockingWriteStream.cs

@@ -0,0 +1,183 @@
+//-----------------------------------------------------------------------
+// <copyright file="BlockingWriteStream.cs" company="Akka.NET Project">
+//     Copyright (C) 2009-2022 Lightbend Inc. <http://www.lightbend.com>
+//     Copyright (C) 2013-2025 .NET Foundation <https://github.com/akkadotnet/akka.net>
+// </copyright>
+//-----------------------------------------------------------------------
+
+using System;
+using System.Collections.Generic;
+using System.IO;
+using System.Linq;
+using System.Net;
+using System.Net.Sockets;
+using System.Threading;
+using System.Threading.Tasks;
+
+#nullable enable
+
+namespace Akka.Tests.IO
+{
+    /// <summary>
+    /// Test stream for a <c>TcpIncomingConnection</c>: the first write blocks until
+    /// <see cref="ReleaseFirstWrite"/>, which stalls the transport's write pump. Reads block
+    /// until <see cref="CompleteReads"/> signals EOF.
+    /// </summary>
+    internal sealed class BlockingWriteStream : Stream
+    {
+        private readonly object _sync = new();
+        private readonly List<int> _writeSizes = new();
+        private readonly TaskCompletionSource<bool> _firstWriteStarted =
+            new(TaskCreationOptions.RunContinuationsAsynchronously);
+        private readonly TaskCompletionSource<bool> _releaseFirstWrite =
+            new(TaskCreationOptions.RunContinuationsAsynchronously);
+        private readonly TaskCompletionSource<bool> _readsCompleted =
+            new(TaskCreationOptions.RunContinuationsAsynchronously);
+
+        private int _writeCount;
+
+        public Task FirstWriteStarted => _firstWriteStarted.Task;
+
+        public IReadOnlyList<int> WriteSizes
+        {
+            get
+            {
+                lock (_sync)
+                {
+                    return _writeSizes.ToArray();
+                }
+            }
+        }
+
+        public long BytesWritten => WriteSizes.Sum(x => (long)x);
+
+        public override bool CanRead => true;
+        public override bool CanSeek => false;
+        public override bool CanWrite => true;
+        public override long Length => throw new NotSupportedException();
+        public override long Position
+        {
+            get => throw new NotSupportedException();
+            set => throw new NotSupportedException();
+        }
+
+        public void ReleaseFirstWrite()
+        {
+            _releaseFirstWrite.TrySetResult(true);
+        }
+
+        /// <summary>Makes the stalled first write throw, which kills the write pump.</summary>
+        public void FailFirstWrite(Exception cause)
+        {
+            _releaseFirstWrite.TrySetException(cause);
+        }
+
+        /// <summary>Makes every pending and future read return 0 (peer EOF).</summary>
+        public void CompleteReads()
+        {
+            _readsCompleted.TrySetResult(true);
+        }
+
+        public override void Flush()
+        {
+        }
+
+        public override Task FlushAsync(CancellationToken cancellationToken)
+        {
+            return Task.CompletedTask;
+        }
+
+        public override int Read(byte[] buffer, int offset, int count)
+        {
+            throw new NotSupportedException();
+        }
+
+        public override Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
+        {
+            return ReadAsync(buffer.AsMemory(offset, count), cancellationToken).AsTask();
+        }
+
+        public override async ValueTask<int> ReadAsync(Memory<byte> buffer, CancellationToken cancellationToken = default)
+        {
+            await _readsCompleted.Task.WaitAsync(cancellationToken).ConfigureAwait(false);
+            return 0;
+        }
+
+        public override long Seek(long offset, SeekOrigin origin)
+        {
+            throw new NotSupportedException();
+        }
+
+        public override void SetLength(long value)
+        {
+            throw new NotSupportedException();
+        }
+
+        public override void Write(byte[] buffer, int offset, int count)
+        {
+            throw new NotSupportedException();
+        }
+
+        public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
+        {
+            return WriteAsync(buffer.AsMemory(offset, count), cancellationToken).AsTask();
+        }
+
+        public override async ValueTask WriteAsync(ReadOnlyMemory<byte> buffer,
+            CancellationToken cancellationToken = default)
+        {
+            var writeIndex = Interlocked.Increment(ref _writeCount);
+            lock (_sync)
+            {
+                _writeSizes.Add(buffer.Length);
+            }
+
+            if (writeIndex == 1)
+            {
+                _firstWriteStarted.TrySetResult(true);
+                await _releaseFirstWrite.Task.WaitAsync(cancellationToken).ConfigureAwait(false);
+            }
+        }
+    }
+
+    internal sealed class ConnectedSocketPair : IDisposable
+    {
+        private ConnectedSocketPair(Socket client, Socket server)
+        {
+            Client = client;
+            Server = server;
+        }
+
+        public Socket Client { get; }
+        public Socket Server { get; }
+
+        public static async Task<ConnectedSocketPair> CreateAsync()
+        {
+            using var listener = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
+            listener.Bind(new IPEndPoint(IPAddress.Loopback, 0));
+            listener.Listen(1);
+
+            var endpoint = (IPEndPoint)listener.LocalEndPoint!;
+            var client = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
+
+            try
+            {
+                var connectTask = client.ConnectAsync(endpoint);
+                var server = await listener.AcceptAsync();
+                await connectTask;
+                return new ConnectedSocketPair(client, server);
+            }
+            catch
+            {
+                client.Dispose();
+                throw;
+            }
+        }
+
+        public void Dispose()
+        {
+            Client.Dispose();
+            Server.Dispose();
+        }
+    }
+}