Read every received entry through the stream: an entry ending at the end of the sender's buffer is read through to its final frame

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<byte>.Default,
+            TransmissionBlockSize = 400,
+            ReceiveTimeout = timeout,
+            GracefulShutdownTimeout = 2000
+        };
+
+        static TcpClient CreateClient(EndPoint address, ILocalMember member, TimeSpan timeout) => new(member, address)
+        {
+            MemoryAllocator = MemoryAllocator<byte>.Default,
+            TransmissionBlockSize = 400,
+            RequestTimeout = timeout,
+            ConnectTimeout = timeout,
+        };
+
+        return SendingLogEntriesAcrossFramesTest(CreateServer, CreateClient, 1000);
+    }
+
     [Theory]
     [InlineData(0, ReceiveEntriesBehavior.ReceiveAll)]
     [InlineData(0, ReceiveEntriesBehavior.ReceiveFirst)]

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<IRaftClusterMember>().AppendEntriesAsync<BufferedEntry, BufferedEntry[]>(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;

src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/NetworkTransport/ConnectionOriented/ProtocolStream.ReadOperations.cs

@@ -134,32 +134,6 @@ private void SkipFrame()
         frameSize -= count;
     }
 
-    public bool TryReadFrameData(int length, out ReadOnlyMemory<byte> 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;
-                if (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<byte> output)
     {
     check_state:

src/cluster/DotNext.Net.Cluster/Net/Cluster/Consensus/Raft/NetworkTransport/ConnectionOriented/Server.LogEntryProducer.cs

@@ -71,9 +71,7 @@ ValueTask IDataTransferObject.WriteToAsync<TWriter>(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<TResult> IDataTransferObject.TransformAsync<TResult, TTransformation>(TTransformation transformation, CancellationToken token)