From 9dce63cf5e01f9a21e56bb6b15c4fbaaa6208ad5 Mon Sep 17 00:00:00 2001 From: Gutemberg Ribeiro Date: Sat, 31 Oct 2026 10:08:51 -0300 Subject: [PATCH 1/3] A follower reads an entry ending where the leader's buffer ended through to its final frame --- .../ConnectionOriented/ProtocolStream.ReadOperations.cs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/NetworkTransport/ConnectionOriented/ProtocolStream.ReadOperations.cs b/src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/NetworkTransport/ConnectionOriented/ProtocolStream.ReadOperations.cs index e514127b7..0d60547a7 100644 --- a/src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/NetworkTransport/ConnectionOriented/ProtocolStream.ReadOperations.cs +++ b/src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/NetworkTransport/ConnectionOriented/ProtocolStream.ReadOperations.cs @@ -146,7 +146,8 @@ public bool TryReadFrameData(int length, out ReadOnlyMemory frame) // the header is in the buffer EndReadFrameHeader(); var writtenBuffer = WrittenBuffer; - if (frameSize >= length && writtenBuffer.Length >= length) + // only a final frame: a payload ending where the leader's buffer ended is followed by an empty final frame, which must be read too + if (readState is ReadState.EndOfStreamReached && frameSize >= length && writtenBuffer.Length >= length) { AdvanceReadCursor(length); frame = writtenBuffer.Slice(0, length); From 6d54f8a91f0e1b8670ee233f931ea217a47e93fc Mon Sep 17 00:00:00 2001 From: Gutemberg Ribeiro Date: Sat, 31 Oct 2026 10:09:48 -0300 Subject: [PATCH 2/3] Test: entries whose payload ends anywhere around the end of the client's buffer, each followed by another --- .../Tcp/TcpTransportTests.cs | 22 +++++++++++++++ .../NetworkTransport/TransportTestSuite.cs | 28 +++++++++++++++++++ 2 files changed, 50 insertions(+) diff --git a/src/DotNext.Tests/Net/Cluster/Consensus/Raft/NetworkTransport/ConnectionOriented/Tcp/TcpTransportTests.cs b/src/DotNext.Tests/Net/Cluster/Consensus/Raft/NetworkTransport/ConnectionOriented/Tcp/TcpTransportTests.cs index 1b2c9663e..2fc1a9e03 100644 --- a/src/DotNext.Tests/Net/Cluster/Consensus/Raft/NetworkTransport/ConnectionOriented/Tcp/TcpTransportTests.cs +++ b/src/DotNext.Tests/Net/Cluster/Consensus/Raft/NetworkTransport/ConnectionOriented/Tcp/TcpTransportTests.cs @@ -229,6 +229,28 @@ public Task SendingLogEntries(int payloadSize, ReceiveEntriesBehavior behavior, return SendingLogEntriesTest(CreateServer, CreateClient, payloadSize, behavior, useEmptyEntry); } + [Fact] + public Task SendingLogEntriesAcrossFrames() + { + static TcpServer CreateServer(ILocalMember member, EndPoint address, TimeSpan timeout) => new(address, 100, member, NullLoggerFactory.Instance) + { + MemoryAllocator = MemoryAllocator.Default, + TransmissionBlockSize = 400, + ReceiveTimeout = timeout, + GracefulShutdownTimeout = 2000 + }; + + static TcpClient CreateClient(EndPoint address, ILocalMember member, TimeSpan timeout) => new(member, address) + { + MemoryAllocator = MemoryAllocator.Default, + TransmissionBlockSize = 400, + RequestTimeout = timeout, + ConnectTimeout = timeout, + }; + + return SendingLogEntriesAcrossFramesTest(CreateServer, CreateClient, 1000); + } + [Theory] [InlineData(0, ReceiveEntriesBehavior.ReceiveAll)] [InlineData(0, ReceiveEntriesBehavior.ReceiveFirst)] diff --git a/src/DotNext.Tests/Net/Cluster/Consensus/Raft/NetworkTransport/TransportTestSuite.cs b/src/DotNext.Tests/Net/Cluster/Consensus/Raft/NetworkTransport/TransportTestSuite.cs index 275a1ec8c..561f97952 100644 --- a/src/DotNext.Tests/Net/Cluster/Consensus/Raft/NetworkTransport/TransportTestSuite.cs +++ b/src/DotNext.Tests/Net/Cluster/Consensus/Raft/NetworkTransport/TransportTestSuite.cs @@ -324,6 +324,34 @@ private protected async Task SendingLogEntriesTest(ServerFactory serverFactory, } } + private protected async Task SendingLogEntriesAcrossFramesTest(ServerFactory serverFactory, ClientFactory clientFactory, int maxPayloadSize) + { + var timeout = DefaultTimeout; + var member = new LocalMember(false); + + //prepare server + var serverAddr = new IPEndPoint(IPAddress.Loopback, 3789); + await using var server = serverFactory(member, serverAddr, timeout); + await server.StartAsync(TestToken); + + //prepare client + using var client = clientFactory(serverAddr, member, timeout); + + // One of the sizes ends the first entry's payload exactly where the client's buffer ends: the frame goes out + // full and not final, followed by an empty final frame, which the server must read before the next entry. + var entry2 = new BufferedEntry(11L, false, RandomBytes(100)); + for (var payloadSize = 1; payloadSize <= maxPayloadSize; payloadSize++) + { + member.ReceivedEntries.Clear(); + var entry1 = new BufferedEntry(10L, false, RandomBytes(payloadSize)); + var result = await client.As().AppendEntriesAsync(42L, [entry1, entry2], 1, 56, 10, TestToken); + Equal(HeartbeatResult.ReplicatedWithLeaderTerm, result.Value.Result); + Equal(2, member.ReceivedEntries.Count); + Equal(entry1, member.ReceivedEntries[0]); + Equal(entry2, member.ReceivedEntries[1]); + } + } + private protected async Task SendingSnapshotTest(ServerFactory serverFactory, ClientFactory clientFactory, int payloadSize) { var timeout = DefaultTimeout; From 29e1ee866f67247271d41e8dd19ce48681466b19 Mon Sep 17 00:00:00 2001 From: Gutemberg Ribeiro Date: Mon, 2 Nov 2026 12:24:03 -0300 Subject: [PATCH 3/3] Remove ProtocolStream.TryReadFrameData: every received entry is read through the stream Its fast path saved a copy only when an entry's payload came whole in a final frame that fit in the buffer, and with the write-ahead log copying into its own pages it saves nothing. It was also where an entry ending at the end of the sender's buffer went wrong, its empty final frame left to be read as the next entry's metadata. --- .../ProtocolStream.ReadOperations.cs | 27 ------------------- .../Server.LogEntryProducer.cs | 4 +-- 2 files changed, 1 insertion(+), 30 deletions(-) diff --git a/src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/NetworkTransport/ConnectionOriented/ProtocolStream.ReadOperations.cs b/src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/NetworkTransport/ConnectionOriented/ProtocolStream.ReadOperations.cs index 0d60547a7..9cf2268d8 100644 --- a/src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/NetworkTransport/ConnectionOriented/ProtocolStream.ReadOperations.cs +++ b/src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/NetworkTransport/ConnectionOriented/ProtocolStream.ReadOperations.cs @@ -134,33 +134,6 @@ private void SkipFrame() frameSize -= count; } - public bool TryReadFrameData(int length, out ReadOnlyMemory frame) - { - if (readState is ReadState.FrameNotStarted) - { - BeginReadFrameHeader(); - - // how many bytes we should read from the stream to parse the frame headers - if (FrameHeadersSize - bufferEnd <= 0) - { - // the header is in the buffer - EndReadFrameHeader(); - var writtenBuffer = WrittenBuffer; - // only a final frame: a payload ending where the leader's buffer ended is followed by an empty final frame, which must be read too - if (readState is ReadState.EndOfStreamReached && frameSize >= length && writtenBuffer.Length >= length) - { - AdvanceReadCursor(length); - frame = writtenBuffer.Slice(0, length); - frameSize -= length; - return true; - } - } - } - - frame = default; - return false; - } - public sealed override int Read(Span output) { check_state: diff --git a/src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/NetworkTransport/ConnectionOriented/Server.LogEntryProducer.cs b/src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/NetworkTransport/ConnectionOriented/Server.LogEntryProducer.cs index ecca65105..489c91b2c 100644 --- a/src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/NetworkTransport/ConnectionOriented/Server.LogEntryProducer.cs +++ b/src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/NetworkTransport/ConnectionOriented/Server.LogEntryProducer.cs @@ -71,9 +71,7 @@ ValueTask IDataTransferObject.WriteToAsync(TWriter writer, Cancellation Debug.Assert(!consumed); consumed = true; - return metadata.Length is { } length and <= int.MaxValue && stream.TryReadFrameData((int)length, out var payload) - ? writer.Invoke(payload, token) - : writer.CopyFromAsync(stream, count: null, token); + return writer.CopyFromAsync(stream, count: null, token); } ValueTask IDataTransferObject.TransformAsync(TTransformation transformation, CancellationToken token)