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 487fd4c70..c1c4b205a 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 288df0237..fb2120cac 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; 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..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,32 +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; - 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 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)