diff --git a/foreign/csharp/Iggy_SDK/ConnectionStream/TcpConnectionStream.cs b/foreign/csharp/Iggy_SDK/ConnectionStream/TcpConnectionStream.cs deleted file mode 100644 index ce016a70f4..0000000000 --- a/foreign/csharp/Iggy_SDK/ConnectionStream/TcpConnectionStream.cs +++ /dev/null @@ -1,53 +0,0 @@ -// Licensed to the Apache Software Foundation (ASF) under one -// or more contributor license agreements. See the NOTICE file -// distributed with this work for additional information -// regarding copyright ownership. The ASF licenses this file -// to you under the Apache License, Version 2.0 (the -// "License"); you may not use this file except in compliance -// with the License. You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, -// software distributed under the License is distributed on an -// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY -// KIND, either express or implied. See the License for the -// specific language governing permissions and limitations -// under the License. - -namespace Apache.Iggy.ConnectionStream; - -internal sealed class TcpConnectionStream : IConnectionStream -{ - private readonly Stream _stream; - - public TcpConnectionStream(Stream stream) - { - _stream = stream; - } - - public ValueTask SendAsync(ReadOnlyMemory payload, CancellationToken cancellationToken = default) - { - return _stream.WriteAsync(payload, cancellationToken); - } - - public ValueTask ReadAsync(Memory buffer, CancellationToken cancellationToken = default) - { - return _stream.ReadAsync(buffer, cancellationToken); - } - - public Task FlushAsync(CancellationToken cancellationToken = default) - { - return _stream.FlushAsync(cancellationToken); - } - - public void Close() - { - _stream.Close(); - } - - public void Dispose() - { - _stream.Dispose(); - } -} diff --git a/foreign/csharp/Iggy_SDK/Consumers/IggyConsumer.Rented.cs b/foreign/csharp/Iggy_SDK/Consumers/IggyConsumer.Rented.cs index 9b4bbfaeb8..a545df5fbb 100644 --- a/foreign/csharp/Iggy_SDK/Consumers/IggyConsumer.Rented.cs +++ b/foreign/csharp/Iggy_SDK/Consumers/IggyConsumer.Rented.cs @@ -102,9 +102,10 @@ await _rentedChannel.Writer.WriteAsync(new ReceivedRentedMessage /// protected async Task PollRentedMessagesAsync(CancellationToken ct) { - if (!_joinedConsumerGroup) + if (!_joinedConsumerGroup || !IsGroupMembershipCurrent()) { LogConsumerGroupNotJoinedYetSkippingPolling(); + await TryRecoverGroupMembershipAsync(ct); return; } diff --git a/foreign/csharp/Iggy_SDK/Consumers/IggyConsumer.cs b/foreign/csharp/Iggy_SDK/Consumers/IggyConsumer.cs index b5317cfdef..91c48a1d57 100644 --- a/foreign/csharp/Iggy_SDK/Consumers/IggyConsumer.cs +++ b/foreign/csharp/Iggy_SDK/Consumers/IggyConsumer.cs @@ -23,6 +23,7 @@ using Apache.Iggy.IggyClient; using Apache.Iggy.Kinds; using Apache.Iggy.Utils; +using Apache.Iggy.Vsr; using Microsoft.Extensions.Logging; namespace Apache.Iggy.Consumers; @@ -33,6 +34,13 @@ namespace Apache.Iggy.Consumers; /// public partial class IggyConsumer : IAsyncDisposable { + /// + /// Backoff between poll iterations that found the group membership missing. It bounds the retry rate of + /// the poll-side rejoin below and keeps the receive loop from spinning while the membership is + /// unrecoverable (client disconnected, group deleted). + /// + private const int GroupRejoinRetryDelayMs = 1_000; + private readonly Channel _channel; private readonly IIggyClient _client; private readonly IggyConsumerConfig _config; @@ -46,6 +54,13 @@ public partial class IggyConsumer : IAsyncDisposable private int _disposeState; private volatile bool _isInitialized; private volatile bool _joinedConsumerGroup; + + /// + /// Consensus session generation the group membership was established under. A later generation means the + /// server session that held the membership is gone and the group must be rejoined. + /// + private ulong _joinedSessionGeneration; + private long _lastPolledAtMs; /// Whether this consumer has been initialized via . @@ -263,8 +278,13 @@ private async Task InitializeConsumerGroupAsync(CancellationToken ct = default) return; } + // Captured before the join: a session reset racing the join lands the membership on a later generation, + // and the mismatch triggers one redundant (idempotent) rejoin instead of a missed one. + var sessionGeneration = (_client as ISessionGenerationProvider)?.SessionGeneration ?? 0; + if (_config.Consumer.Type == ConsumerType.Consumer) { + Interlocked.Exchange(ref _joinedSessionGeneration, sessionGeneration); _joinedConsumerGroup = true; return; } @@ -306,6 +326,7 @@ private async Task InitializeConsumerGroupAsync(CancellationToken ct = default) await _client.JoinConsumerGroupAsync(_config.StreamId, _config.TopicId, Identifier.String(_consumerGroupName), ct); + Interlocked.Exchange(ref _joinedSessionGeneration, sessionGeneration); _joinedConsumerGroup = true; LogConsumerGroupJoined(_consumerGroupName); } @@ -364,9 +385,10 @@ private void ThrowIfAutoCommitWithEncryptor() /// private async Task PollMessagesAsync(CancellationToken ct) { - if (!_joinedConsumerGroup) + if (!_joinedConsumerGroup || !IsGroupMembershipCurrent()) { LogConsumerGroupNotJoinedYetSkippingPolling(); + await TryRecoverGroupMembershipAsync(ct); return; } @@ -452,6 +474,57 @@ private async Task PollMessagesAsync(CancellationToken ct) } } + /// + /// Whether the session generation the group membership was stamped under is still the transport's + /// current one. Gating the poll here instead of clearing the joined flag on a Disconnected event survives + /// a late event landing after a rejoin already re-stamped the generation: state events are published + /// outside the state lock, so their order is not guaranteed. Runs outside + /// , hence the interlocked read. + /// + private bool IsGroupMembershipCurrent() + { + if (_config.Consumer.Type != ConsumerType.ConsumerGroup + || _client is not ISessionGenerationProvider generationProvider) + { + return true; + } + + return generationProvider.SessionGeneration == Interlocked.Read(ref _joinedSessionGeneration); + } + + /// + /// Poll-side rejoin for a membership found missing or stamped under a dead session. The state-event + /// rejoin swallows its failures so the event loop survives them, and the transport suppresses a repeat + /// event for an unchanged state, so without this backstop one failed rejoin would park the consumer for + /// good. The trailing delay keeps the receive loop from spinning while the membership stays gone. + /// + private async Task TryRecoverGroupMembershipAsync(CancellationToken ct) + { + if (_config.Consumer.Type == ConsumerType.ConsumerGroup && _config.JoinConsumerGroup) + { + await _connectionStateSemaphore.WaitAsync(ct); + try + { + if (!_joinedConsumerGroup || !IsGroupMembershipCurrent()) + { + _joinedConsumerGroup = false; + await RejoinConsumerGroupOnReconnectionAsync(); + } + } + finally + { + _connectionStateSemaphore.Release(); + } + + if (_joinedConsumerGroup && IsGroupMembershipCurrent()) + { + return; + } + } + + await Task.Delay(GroupRejoinRetryDelayMs, ct); + } + /// /// Implements polling interval throttling to avoid excessive server requests. /// Uses monotonic time tracking to ensure proper intervals even with clock adjustments. @@ -502,9 +575,33 @@ private async Task OnClientConnectionStateChangedAsync(ConnectionStateChangedEve { LogConnectionStateChanged(e.PreviousState, e.CurrentState); + if (_config.Consumer.Type == ConsumerType.Consumer) + { + return; + } + await _connectionStateSemaphore.WaitAsync(); try { + if (_client is ISessionGenerationProvider generationProvider) + { + if (e.CurrentState != ConnectionState.Authenticated) + { + return; + } + + if (_joinedConsumerGroup + && generationProvider.SessionGeneration == Interlocked.Read(ref _joinedSessionGeneration)) + { + return; + } + + _joinedConsumerGroup = false; + await RejoinConsumerGroupOnReconnectionAsync(); + + return; + } + if (e.CurrentState == ConnectionState.Disconnected) { _joinedConsumerGroup = false; diff --git a/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.Vsr.cs b/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.Vsr.cs index f70a3b23b2..2985703fa6 100644 --- a/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.Vsr.cs +++ b/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.Vsr.cs @@ -16,16 +16,15 @@ // under the License. using System.Buffers; -using System.Buffers.Binary; using System.IO.Hashing; using System.Runtime.ExceptionServices; -using Apache.Iggy.ConnectionStream; using Apache.Iggy.Contracts; using Apache.Iggy.Contracts.Auth; using Apache.Iggy.Contracts.Tcp; using Apache.Iggy.Enums; using Apache.Iggy.Exceptions; using Apache.Iggy.Kinds; +using Apache.Iggy.Mappers; using Apache.Iggy.Messages; using Apache.Iggy.Utils; using Apache.Iggy.Vsr; @@ -39,7 +38,7 @@ namespace Apache.Iggy.IggyClient.Implementations; /// register handshake, and the client-side partitioning and consumer-group assignment the broker does not /// resolve server-side. The command surface lives in . /// -public sealed partial class TcpMessageStream +public sealed partial class TcpMessageStream : ISessionGenerationProvider { /// /// Upper bound for a whole VSR request: the transient replays and the leader failovers share it, and so @@ -48,15 +47,6 @@ public sealed partial class TcpMessageStream /// private const int VsrRequestTimeoutMs = 30_000; - /// Backoff between replays of a transiently refused request. - private const int VsrTransientRetryIntervalMs = 50; - - /// - /// Largest body still sent as one contiguous frame with its header. Beyond this the copy outweighs the - /// syscall and the extra segment it saves, so header and body go out as two writes. - /// - private const int VsrContiguousFrameLimit = 4 * 1024; - /// /// How long a request replays on the same connection /// before the leader roster is re-checked. A node that stopped being primary refuses forever, so @@ -75,9 +65,8 @@ public sealed partial class TcpMessageStream /// /// Cap on consecutive leader redirects, so a flapping roster cannot spin the connect loop or the - /// transient failover path. The budget is client-wide and resets on a roster check that finds the - /// current node is the leader, and on every request that completes, so a client that outlives more - /// leader changes than the cap does not latch onto a follower for good. + /// transient failover path. Each operation - a request, a connect, a register - spends its own local + /// budget, so a long-lived client never latches onto a follower for good. /// private const int VsrMaxLeaderRedirects = 3; @@ -108,12 +97,9 @@ public sealed partial class TcpMessageStream private readonly ConsensusSession _consensusSession = new(); private readonly ConsumerGroupClientState _groupState = new(); - private readonly byte[] _vsrReplyHeaderBuffer = new byte[VsrHeader.HEADER_SIZE]; - // The redirect budget is refunded by a completed request, and the roster check a redirect runs is itself a - // request. Without this the refund lands between the check and the increment that reads the budget, and the - // counter never leaves zero. Nonzero for the duration of a roster read, so that refund is skipped. - private int _leaderProbeDepth; + /// + ulong ISessionGenerationProvider.SessionGeneration => _consensusSession.Generation; /// /// Runs the consensus register handshake and binds the session it commits. Everything before the bind @@ -133,21 +119,18 @@ public sealed partial class TcpMessageStream { await LogoutUserAsync(token); } - else if (_state == ConnectionState.Authenticated) + else if (State == ConnectionState.Authenticated) { - SetConnectionStateAsync(ConnectionState.Connected); + SetConnectionState(ConnectionState.Connected); } - var payload = new byte[4 + BufferSizes.INITIAL_BYTES_LENGTH + message.Length]; - TcpMessageStreamHelpers.CreatePayload(payload, message, code); - - SetConnectionStateAsync(ConnectionState.Authenticating); + SetConnectionState(ConnectionState.Authenticating); LoginRegisterResponse response; try { - Interlocked.Exchange(ref _skipAutoLoginOnce, 1); - using IMemoryOwner responseBuffer = await SendWithResponseAsync(payload, token); + using IMemoryOwner responseBuffer = + await SendWithResponseAsync(code, message, autoLoginOnReconnect: false, token: token); response = LoginRegister.Deserialize(responseBuffer.Memory.Span); _consensusSession.Bind(response.Session); @@ -155,22 +138,18 @@ public sealed partial class TcpMessageStream catch { await ResetConsensusSessionAsync(); - if (_state == ConnectionState.Authenticating) + if (State == ConnectionState.Authenticating) { - SetConnectionStateAsync(ConnectionState.Connected); + SetConnectionState(ConnectionState.Connected); } throw; } - finally - { - Interlocked.Exchange(ref _skipAutoLoginOnce, 0); - } _logger.LogInformation( "Authenticated against the server, version {ServerVersion}, protocol version {ServerProtocolVersion}", response.ServerVersion, response.ServerProtocolVersion); - SetConnectionStateAsync(ConnectionState.Authenticated); + SetConnectionState(ConnectionState.Authenticated); var authResponse = new AuthResponse((int)response.UserId, null); if (IsConnecting) @@ -182,16 +161,21 @@ public sealed partial class TcpMessageStream { _logger.LogWarning("Maximum leader redirections reached while registering, staying on {Address}", _currentAddress); - - return authResponse; + } + else if (await RedirectAsync(token)) + { + await ConnectAsync(false, token); + continue; } - if (!await RedirectAsync(token)) + // The redirect probe can tear the connection down without throwing, and success on a client that is + // no longer bound would leave the caller unauthenticated with nothing left to re-authenticate it. + if (State != ConnectionState.Authenticated) { - return authResponse; + throw new NotConnectedException(); } - await ConnectAsync(false, token); + return authResponse; } } @@ -328,10 +312,8 @@ private async Task SyncGroupAssignmentAsync(Identifier streamId, Identifier topi CancellationToken token) { var message = TcpContracts.GetGroup(streamId, topicId, groupId); - var payload = new byte[4 + BufferSizes.INITIAL_BYTES_LENGTH + message.Length]; - TcpMessageStreamHelpers.CreatePayload(payload, message, CommandCodes.SYNC_CONSUMER_GROUP_CODE); - - using IMemoryOwner responseBuffer = await SendWithResponseAsync(payload, token); + using IMemoryOwner responseBuffer = + await SendWithResponseAsync(CommandCodes.SYNC_CONSUMER_GROUP_CODE, message, token: token); var key = new GroupKey(streamId, topicId, groupId); if (responseBuffer.Memory.Length == 0) @@ -373,9 +355,9 @@ private async Task RefreshGroupAssignmentsAsync(CancellationToken token) /// /// Points the client at the current leader when it is not the node this connection is on, leaving the - /// stream closed for the caller to reconnect. The redirect budget is client-wide: the connect loop and - /// the transient failover path spend the same counter, and it is refunded as soon as a roster check - /// lands on the leader. + /// stream closed for the caller to reconnect. The caller owns the redirect budget: every operation that + /// follows redirects - a request, a connect, a register - counts them locally against + /// . /// private async Task RedirectAsync(CancellationToken token) { @@ -384,24 +366,19 @@ private async Task RedirectAsync(CancellationToken token) var currentLeaderNode = await GetCurrentLeaderNodeAsync(token); if (currentLeaderNode == null) { - Interlocked.Exchange(ref _leaderRedirectCount, 0); return false; } var leaderAddress = ServerAddress.HostPort(currentLeaderNode.Ip, currentLeaderNode.Endpoints.Tcp); - // Compare against the endpoint the socket resolved, not the configured string: a client - // configured with a hostname is otherwise never "on" the leader the roster names by IP, - // and every login would reconnect it to the node it is already talking to. - var connectedAddress = _currentRemoteAddress.Length > 0 ? _currentRemoteAddress : _currentAddress; - if (ServerAddress.IsSame(leaderAddress, connectedAddress)) - { - Interlocked.Exchange(ref _leaderRedirectCount, 0); - return false; - } - - if (Interlocked.Increment(ref _leaderRedirectCount) > VsrMaxLeaderRedirects) + // The roster may name the leader by either side of this connection: the address the client dialed + // (a hostname, an advertised_address) or the endpoint the socket resolved to. IsSame does no DNS + // resolution, so a match on either means the client is already on the leader; checking only one + // side redirect-loops the other kind of roster. Both are evidence of where the socket landed, so + // both come from the connect loop: _currentAddress may already name a leader the client never + // reached, and matching on it would silently refuse the redirect that gets it there. + if ((_currentRemoteAddress.Length > 0 && ServerAddress.IsSame(leaderAddress, _currentRemoteAddress)) + || (_connectedAddress.Length > 0 && ServerAddress.IsSame(leaderAddress, _connectedAddress))) { - _logger.LogWarning("Maximum leader redirections reached, continuing on {Address}", _currentAddress); return false; } @@ -414,7 +391,7 @@ private async Task RedirectAsync(CancellationToken token) try { _currentAddress = leaderAddress; - DropVsrConnectionLocked(_stream); + DropVsrConnectionLocked(_connection); } finally { @@ -427,12 +404,11 @@ private async Task RedirectAsync(CancellationToken token) private async Task GetCurrentLeaderNodeAsync(CancellationToken token) { var leaderlessDeadline = Environment.TickCount64 + VsrLeaderlessWaitMs; - Interlocked.Increment(ref _leaderProbeDepth); try { while (true) { - var clusterMetadata = await GetClusterMetadataAsync(token); + var clusterMetadata = await ReadClusterMetadataNoRedirectAsync(token); if (clusterMetadata == null) { return null; @@ -469,38 +445,58 @@ private async Task RedirectAsync(CancellationToken token) { return null; } - catch (Exception e) when (e is not OperationCanceledException) + catch (Exception e) when (e is not OperationCanceledException && !VsrConnection.IsConnectionException(e)) { _logger.LogWarning(e, "Failed to read the cluster metadata, continuing on {Address}", _currentAddress); return null; } - finally + } + + /// + /// Reads the roster without following redirects: the probe is what redirects are decided from, so a + /// probe that redirected or reconnected would reenter the very loop that called it. A probe the current + /// node keeps refusing simply fails, and the caller stays where it is. + /// + private async Task ReadClusterMetadataNoRedirectAsync(CancellationToken token) + { + using IMemoryOwner responseBuffer = await SendRawAsync(CommandCodes.GET_CLUSTER_METADATA_CODE, + ReadOnlyMemory.Empty, token, allowRedirect: false); + + if (responseBuffer.Memory.Length == 0) { - Interlocked.Decrement(ref _leaderProbeDepth); + return null; } + + return BinaryMapper.MapClusterMetadata(responseBuffer.Memory.Span); } /// - /// Sends a consensus-framed request. The call sites still build the classic - /// [size u32][code u32][body] buffer, so the code is read back from it here and the body is written - /// right after the 256-byte consensus header - two writes, no concatenation. + /// Sends a consensus-framed request: small frames go out as a single coalesced write, larger bodies + /// as a second write straight from the caller's buffer. /// /// /// One deadline bounds the whole request across transient replays AND leader failovers. Login and /// register replay on this connection for the whole budget instead: the connect flow owns leader /// redirection for the handshake, and reconnecting from underneath it would recurse. /// - private async Task> SendRawVsrAsync(ReadOnlyMemory payload, CancellationToken token) + private async Task> SendRawAsync(int code, ReadOnlyMemory body, + CancellationToken token, bool allowRedirect = true) { - var code = (int)BinaryPrimitives.ReadUInt32LittleEndian(payload.Span.Slice(4, 4)); - ReadOnlyMemory body = payload[8..]; + ObjectDisposedException.ThrowIf(_disposed, this); + + if (State is ConnectionState.Disconnected or ConnectionState.Connecting) + { + throw new NotConnectedException(); + } + var isLoginRegister = code is CommandCodes.LOGIN_REGISTER_CODE or CommandCodes.LOGIN_REGISTER_WITH_PAT_CODE; + var clearSensitiveReply = HasSensitiveReply(code); var overallDeadline = Environment.TickCount64 + VsrRequestTimeoutMs; - var headerBuffer = ArrayPool.Shared.Rent(VsrHeader.HEADER_SIZE); - Memory header = headerBuffer.AsMemory(0, VsrHeader.HEADER_SIZE); var requestEncoded = false; - TcpConnectionStream? lastStream = null; + var redirects = 0; + var redirectBudgetLogged = false; + VsrConnection? lastConnection = null; try { @@ -510,20 +506,13 @@ private async Task> SendRawVsrAsync(ReadOnlyMemory payl ? overallDeadline : Math.Min(overallDeadline, Environment.TickCount64 + VsrTransientFailoverCheckMs); - var attempt = await SendVsrAttemptAsync(code, body, header, transientDeadline, overallDeadline, - token); + var attempt = await SendVsrAttemptAsync(code, body, transientDeadline, overallDeadline, + clearSensitiveReply, token); requestEncoded |= attempt.Encoded; - lastStream = attempt.Stream; + lastConnection = attempt.Connection; if (attempt.Error is null) { - // A roster read taken by RedirectAsync must not refund the budget it is about to be charged - // against, or the cap can never be reached. - if (Volatile.Read(ref _leaderRedirectCount) != 0 && Volatile.Read(ref _leaderProbeDepth) == 0) - { - Interlocked.Exchange(ref _leaderRedirectCount, 0); - } - return attempt.Response!; } @@ -532,12 +521,20 @@ private async Task> SendRawVsrAsync(ReadOnlyMemory payl StatusCode: VsrError.TRANSIENT_NOT_ACCEPTED, FromServer: true } && !isLoginRegister + && allowRedirect && Environment.TickCount64 < overallDeadline) { - if (await RedirectAsync(token)) + if (redirects < VsrMaxLeaderRedirects && await RedirectAsync(token)) { + redirects++; await ConnectAsync(token); } + else if (redirects >= VsrMaxLeaderRedirects && !redirectBudgetLogged) + { + redirectBudgetLogged = true; + _logger.LogWarning("Maximum leader redirections reached, continuing on {Address}", + _currentAddress); + } continue; } @@ -566,15 +563,11 @@ private async Task> SendRawVsrAsync(ReadOnlyMemory payl { if (requestEncoded) { - await DropVsrConnectionAsync(lastStream); + await DropVsrConnectionAsync(lastConnection); } throw; } - finally - { - ArrayPool.Shared.Return(headerBuffer); - } } /// @@ -582,6 +575,19 @@ private async Task> SendRawVsrAsync(ReadOnlyMemory payl /// the client refused or discarded, and a NOT_COMMITTED that outlived its replay deadline all leave the /// outcome of a request the server may still commit unknowable. /// + /// + /// Whether the reply body may carry a credential - a raw personal access token, a session secret - and + /// therefore must be zeroed before its pooled buffer is handed back for reuse. + /// + private static bool HasSensitiveReply(int code) + { + return code is CommandCodes.LOGIN_USER_CODE + or CommandCodes.LOGIN_REGISTER_CODE + or CommandCodes.LOGIN_WITH_PERSONAL_ACCESS_TOKEN_CODE + or CommandCodes.LOGIN_REGISTER_WITH_PAT_CODE + or CommandCodes.CREATE_PERSONAL_ACCESS_TOKEN_CODE; + } + private static bool IsDefinitiveVerdict(Exception error) { return error is IggyInvalidStatusCodeException @@ -592,236 +598,31 @@ private static bool IsDefinitiveVerdict(Exception error) } /// - /// One attempt on the current connection: encode the header into , write the - /// frame, and replay it - same session, same request id - while the server answers transiently. + /// One attempt on the current connection, resolved under the sending lock so the frame and its reply + /// cannot be split across two sockets, and so a teardown after the lock is gone can tell this + /// connection from a replacement a reconnect installed since. /// - private async ValueTask SendVsrAttemptAsync(int code, ReadOnlyMemory body, Memory header, - long transientDeadline, long readDeadline, CancellationToken token) + private async ValueTask SendVsrAttemptAsync(int code, ReadOnlyMemory body, + long transientDeadline, long readDeadline, bool clearSensitiveReply, CancellationToken token) { await _sendingSemaphore.WaitAsync(token); - - var encoded = false; - var requestStarted = false; - byte[]? frameBuffer = null; - - // Read the stream once for the whole attempt. Nothing may swap the field without the sending lock this - // call holds, so the frame and its reply cannot be split across two sockets, and a teardown after the - // lock is gone can tell this connection from a replacement a reconnect installed since. - var stream = _stream; try { - // A small request goes out as one write. Two writes cost two syscalls and, with Nagle disabled, two - // TCP segments (two TLS records when encrypted) for what the reference encoder sends as a single - // contiguous frame. Above the threshold the copy costs more than the extra write saves. Renting - // inside the try keeps the semaphore paired with its release even if the pool throws. - if (body.Length <= VsrContiguousFrameLimit) - { - frameBuffer = ArrayPool.Shared.Rent(VsrHeader.HEADER_SIZE + body.Length); - } - - VsrHeader.EncodeRequestHeader(header.Span, _consensusSession, code, body.Span); - - encoded = true; - - var frame = Memory.Empty; - if (frameBuffer is not null) - { - frame = frameBuffer.AsMemory(0, VsrHeader.HEADER_SIZE + body.Length); - header.CopyTo(frame); - body.CopyTo(frame[VsrHeader.HEADER_SIZE..]); - } - - while (true) - { - try - { - // Everything that fails without reaching the socket has to fail before this point: past it a - // failure is reported as an outcome the server alone knows, which for a replicated write - // tells the caller its request may have committed twice. - token.ThrowIfCancellationRequested(); - requestStarted = true; - - if (frameBuffer is not null) - { - await stream.SendAsync(frame, token); - } - else - { - await stream.SendAsync(header, token); - await stream.SendAsync(body, token); - } - - await stream.FlushAsync(token); - - IMemoryOwner response = await ReadVsrReplyAsync(stream, readDeadline, token); - - return VsrAttempt.Ok(response, stream); - } - catch (IggyInvalidStatusCodeException e) when (IsReplayableTransient(e, transientDeadline, - readDeadline)) - { - var governingDeadline = e.StatusCode == VsrError.TRANSIENT_NOT_COMMITTED - ? readDeadline - : transientDeadline; - var remaining = governingDeadline - Environment.TickCount64; - await Task.Delay((int)Math.Clamp(remaining, 0, VsrTransientRetryIntervalMs), token); - } - catch (Exception e) when (IsConnectionException(e)) - { - DropVsrConnectionLocked(stream); - - return VsrAttempt.Failed(encoded, e, requestStarted, stream); - } - catch (OperationCanceledException e) - { - DropVsrConnectionLocked(stream); - - return VsrAttempt.Failed(encoded, e, requestStarted, stream); - } - catch (Exception e) - { - return VsrAttempt.Failed(encoded, e, requestStarted, stream); - } - } - } - catch (OperationCanceledException e) - { - if (encoded) + var connection = _connection; + if (connection is null) { - DropVsrConnectionLocked(stream); + return VsrAttempt.Failed(false, new NotConnectedException(), false, null); } - return VsrAttempt.Failed(encoded, e, requestStarted, stream); - } - catch (Exception e) - { - return VsrAttempt.Failed(encoded, e, requestStarted, stream); + return await connection.SendAttemptAsync(code, body, transientDeadline, readDeadline, + clearSensitiveReply, token); } finally { - if (frameBuffer is not null) - { - ArrayPool.Shared.Return(frameBuffer); - } - _sendingSemaphore.Release(); } } - private async Task> ReadVsrReplyAsync(TcpConnectionStream stream, long readDeadline, - CancellationToken token) - { - var remaining = readDeadline - Environment.TickCount64; - if (remaining <= 0) - { - throw new IOException($"Timed out after {VsrRequestTimeoutMs} ms waiting for a consensus reply."); - } - - // One timer for the whole reply: the deadline covers the frame, not each partial read, so a per-read - // source would both re-arm the budget and allocate a timer per socket read. - using var readCancellation = CancellationTokenSource.CreateLinkedTokenSource(token); - readCancellation.CancelAfter((int)Math.Min(remaining, VsrRequestTimeoutMs)); - - await ReadExactVsrAsync(stream, _vsrReplyHeaderBuffer, readCancellation.Token, token); - - var command = VsrHeader.PeekCommand(_vsrReplyHeaderBuffer); - if (command == Command2.Eviction) - { - var eviction = VsrHeader.ReadEviction(_vsrReplyHeaderBuffer); - _logger.LogWarning("Consensus session evicted by the server: {Reason}", eviction.Reason); - DropVsrConnectionLocked(stream); - - throw new VsrSessionEvictedException(VsrReplyDecoder.ToException(eviction)); - } - - if (command != Command2.Reply) - { - // Neither a reply nor an eviction: this frame was never an answer to the outstanding request, so - // whatever the peer does send for it would be read as the next request's reply and handed to the - // wrong caller. The size field of a frame the client cannot model is no basis for resynchronising. - DropVsrConnectionLocked(stream); - - throw VsrError.Exception(VsrError.INVALID_COMMAND, - $"Unexpected consensus frame {command} on a client connection."); - } - - int bodySize; - try - { - bodySize = VsrReplyDecoder.ReadBodySize(_vsrReplyHeaderBuffer); - if (VsrHeader.HEADER_SIZE + (long)bodySize > _configuration.MaxResponseFrameSize) - { - throw VsrError.Exception(VsrError.INVALID_COMMAND, - $"Reply frame of {VsrHeader.HEADER_SIZE + bodySize} bytes exceeds the configured maximum of " + - $"{_configuration.MaxResponseFrameSize} bytes."); - } - } - catch - { - // An announced size the client refuses to read - undersized, oversized - leaves the body on the - // wire, so the stream no longer sits on a frame boundary and the next reply would decode body bytes - // as a header. - DropVsrConnectionLocked(stream); - - throw; - } - - if (bodySize == 0) - { - VsrReplyDecoder.Decode(_vsrReplyHeaderBuffer, ReadOnlyMemory.Empty); - - return EmptyMemoryOwner.Instance; - } - - var buffer = ArrayPool.Shared.Rent(bodySize); - try - { - await ReadExactVsrAsync(stream, buffer.AsMemory(0, bodySize), readCancellation.Token, token); - ReadOnlyMemory decoded = VsrReplyDecoder.Decode(_vsrReplyHeaderBuffer, buffer.AsMemory(0, bodySize)); - if (decoded.IsEmpty) - { - ArrayPool.Shared.Return(buffer); - - return EmptyMemoryOwner.Instance; - } - - // The decoded payload is always a suffix of the body - the funnel only strips the leading - // committed result section. - return new PooledMemoryOwner(buffer, bodySize - decoded.Length, decoded.Length); - } - catch - { - ArrayPool.Shared.Return(buffer); - throw; - } - } - - private async ValueTask ReadExactVsrAsync(TcpConnectionStream stream, Memory buffer, - CancellationToken readToken, - CancellationToken token) - { - var totalRead = 0; - while (totalRead < buffer.Length) - { - int readBytes; - try - { - readBytes = await stream.ReadAsync(buffer[totalRead..], readToken); - } - catch (OperationCanceledException) when (!token.IsCancellationRequested) - { - throw new IOException($"Timed out after {VsrRequestTimeoutMs} ms waiting for a consensus reply."); - } - - if (readBytes == 0) - { - throw new IggyZeroBytesException(); - } - - totalRead += readBytes; - } - } - /// /// Drops the consensus session and the group state scoped to it. Consumer-group assignments are fenced by /// a generation the coordinator tracks per session, so carrying them into a new session would fence every @@ -877,27 +678,28 @@ private async ValueTask TryEnterSendingSemaphoreAsync() /// /// Drops the connection along with the session. A late or half-read reply would desync the framing of the - /// next request, so the stream cannot be reused. The caller must hold , - /// which owns every write to . + /// next request, so the socket cannot be reused. The caller must hold , + /// which owns every write to . /// - /// + /// /// The connection the caller was using. A reconnect that completed in the meantime already closed it and /// re-armed the session, so dropping anything but the live one would tear down a healthy replacement. /// - private void DropVsrConnectionLocked(TcpConnectionStream? stream) + private void DropVsrConnectionLocked(VsrConnection? connection) { - if (!ReferenceEquals(_stream, stream)) + if (connection is null || !ReferenceEquals(_connection, connection)) { return; } ResetConsensusSession(); - _stream?.Close(); - SetConnectionStateAsync(ConnectionState.Disconnected); + _connection = null; + SetConnectionState(ConnectionState.Disconnected); + connection.Dispose(); } /// Drops the connection on behalf of a caller that no longer holds the sending lock. - private async ValueTask DropVsrConnectionAsync(TcpConnectionStream? stream) + private async ValueTask DropVsrConnectionAsync(VsrConnection? connection) { // Dispose already closed the stream, and taking a disposed semaphore here would replace the // cancellation the caller is about to rethrow with an ObjectDisposedException. Dispose can still land @@ -910,73 +712,11 @@ private async ValueTask DropVsrConnectionAsync(TcpConnectionStream? stream) try { - DropVsrConnectionLocked(stream); + DropVsrConnectionLocked(connection); } finally { _sendingSemaphore.Release(); } } - - private static bool IsReplayableTransient(IggyInvalidStatusCodeException error, long transientDeadline, - long readDeadline) - { - if (!error.FromServer) - { - return false; - } - - return error.StatusCode switch - { - VsrError.TRANSIENT_NOT_COMMITTED => Environment.TickCount64 < readDeadline, - VsrError.TRANSIENT_NOT_ACCEPTED => Environment.TickCount64 < transientDeadline, - _ => false - }; - } - - /// Outcome of one call on the current connection. - /// Whether the header was encoded, i.e. whether a request id may have been consumed. - /// The decoded reply payload, non-null exactly when is null. - /// The failure that ended the attempt, or null on success. - /// - /// Whether any byte of the frame was written, which makes the server-side outcome unknowable on failure. - /// - /// - /// The connection the attempt ran on, so a caller that drops it after releasing the sending lock can tell - /// its own connection from a replacement a reconnect installed since. - /// - private readonly record struct VsrAttempt( - bool Encoded, - IMemoryOwner? Response, - Exception? Error, - bool RequestStarted, - TcpConnectionStream? Stream) - { - public static VsrAttempt Ok(IMemoryOwner response, TcpConnectionStream stream) - { - return new VsrAttempt(true, response, null, true, stream); - } - - public static VsrAttempt Failed(bool encoded, Exception error, bool requestStarted, - TcpConnectionStream? stream) - { - return new VsrAttempt(encoded, null, error, requestStarted, stream); - } - } - - /// Owns a pooled buffer while exposing only the decoded payload slice inside it. - internal sealed class PooledMemoryOwner(byte[] buffer, int start, int length) : IMemoryOwner - { - private int _disposed; - - public Memory Memory => buffer.AsMemory(start, length); - - public void Dispose() - { - if (Interlocked.Exchange(ref _disposed, 1) == 0) - { - ArrayPool.Shared.Return(buffer); - } - } - } } diff --git a/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.cs b/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.cs index c68ad62e69..42b3dd7e50 100644 --- a/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.cs +++ b/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.cs @@ -20,11 +20,9 @@ using System.Net; using System.Net.Security; using System.Net.Sockets; -using System.Runtime.CompilerServices; using System.Runtime.InteropServices; using System.Security.Cryptography.X509Certificates; using Apache.Iggy.Configuration; -using Apache.Iggy.ConnectionStream; using Apache.Iggy.Contracts; using Apache.Iggy.Contracts.Auth; using Apache.Iggy.Contracts.Tcp; @@ -47,8 +45,6 @@ namespace Apache.Iggy.IggyClient.Implementations; /// public sealed partial class TcpMessageStream : IIggyClient { - private const int InvalidCommandStatus = 3; - private static readonly HashSet SessionControlCodes = [ CommandCodes.LOGIN_USER_CODE, @@ -64,6 +60,7 @@ public sealed partial class TcpMessageStream : IIggyClient private readonly SemaphoreSlim _connectionSemaphore; private readonly ILogger _logger; private readonly SemaphoreSlim _sendingSemaphore; + private VsrConnection? _connection; private string _currentAddress = string.Empty; // The address the socket actually connected to, as an IP the roster can be compared against. @@ -71,22 +68,21 @@ public sealed partial class TcpMessageStream : IIggyClient // mentions), so leader comparisons made against it would move a client that is already on the // leader. Written only by the connect loop. private string _currentRemoteAddress = string.Empty; + + // The dialed address whose connect actually landed. _currentAddress moves ahead of the socket - + // a redirect rewrites it before the reconnect - so a leader comparison against it would treat a + // node the client never reached as "already there". Written only by the connect loop. + private string _connectedAddress = string.Empty; private X509Certificate2Collection _customCaStore = []; private volatile bool _disposed; private int _isConnecting; private DateTimeOffset _lastConnectionTime; - private int _leaderRedirectCount; - - // Both are written by the connect and redirect paths, which do not hold the sending semaphore the request - // paths read them under, so they are accessed through Interlocked rather than as plain fields. Losing an - // update to the skip flag leaves a connection reporting Connected that never authenticated; losing one to - // the redirect counter over- or under-spends the redirect budget. - private int _skipAutoLoginOnce; - private volatile ConnectionState _state = ConnectionState.Disconnected; - private TcpConnectionStream _stream = null!; + private int _stateValue = (int)ConnectionState.Disconnected; private bool IsConnecting => Volatile.Read(ref _isConnecting) != 0; + private ConnectionState State => (ConnectionState)Volatile.Read(ref _stateValue); + internal TcpMessageStream(IggyClientConfigurator configuration, ILoggerFactory loggerFactory) { _configuration = configuration; @@ -103,10 +99,10 @@ internal TcpMessageStream(IggyClientConfigurator configuration, ILoggerFactory l public void Dispose() { _disposed = true; - _stream?.Close(); - _stream?.Dispose(); + _connection?.Dispose(); + _connection = null; - SetConnectionStateAsync(ConnectionState.Disconnected); + SetConnectionState(ConnectionState.Disconnected); _sendingSemaphore.Dispose(); _connectionSemaphore.Dispose(); _connectGate.Dispose(); @@ -138,10 +134,8 @@ public string GetCurrentAddress() public async Task CreateStreamAsync(string name, CancellationToken token = default) { var message = TcpContracts.CreateStream(name); - var payload = new byte[4 + BufferSizes.INITIAL_BYTES_LENGTH + message.Length]; - TcpMessageStreamHelpers.CreatePayload(payload, message, CommandCodes.CREATE_STREAM_CODE); - - using IMemoryOwner responseBuffer = await SendWithResponseAsync(payload, token); + using IMemoryOwner responseBuffer + = await SendWithResponseAsync(CommandCodes.CREATE_STREAM_CODE, message, token: token); if (responseBuffer.Memory.Length == 0) { @@ -155,10 +149,8 @@ public string GetCurrentAddress() public async Task GetStreamByIdAsync(Identifier streamId, CancellationToken token = default) { var message = TcpMessageStreamHelpers.GetBytesFromIdentifier(streamId); - var payload = new byte[4 + BufferSizes.INITIAL_BYTES_LENGTH + message.Length]; - TcpMessageStreamHelpers.CreatePayload(payload, message, CommandCodes.GET_STREAM_CODE); - - using IMemoryOwner responseBuffer = await SendWithResponseAsync(payload, token); + using IMemoryOwner responseBuffer + = await SendWithResponseAsync(CommandCodes.GET_STREAM_CODE, message, token: token); if (responseBuffer.Memory.Length == 0) { @@ -172,10 +164,8 @@ public string GetCurrentAddress() public async Task> GetStreamsAsync(CancellationToken token = default) { var message = Array.Empty(); - var payload = new byte[4 + BufferSizes.INITIAL_BYTES_LENGTH + message.Length]; - TcpMessageStreamHelpers.CreatePayload(payload, message, CommandCodes.GET_STREAMS_CODE); - - using IMemoryOwner responseBuffer = await SendWithResponseAsync(payload, token); + using IMemoryOwner responseBuffer + = await SendWithResponseAsync(CommandCodes.GET_STREAMS_CODE, message, token: token); if (responseBuffer.Memory.Length == 0) { @@ -189,30 +179,21 @@ public async Task> GetStreamsAsync(CancellationTok public async Task UpdateStreamAsync(Identifier streamId, string name, CancellationToken token = default) { var message = TcpContracts.UpdateStream(streamId, name); - var payload = new byte[4 + BufferSizes.INITIAL_BYTES_LENGTH + message.Length]; - TcpMessageStreamHelpers.CreatePayload(payload, message, CommandCodes.UPDATE_STREAM_CODE); - - await SendAckAsync(payload, token); + await SendAckAsync(CommandCodes.UPDATE_STREAM_CODE, message, token); } /// public async Task PurgeStreamAsync(Identifier streamId, CancellationToken token = default) { var message = TcpMessageStreamHelpers.GetBytesFromIdentifier(streamId); - var payload = new byte[4 + BufferSizes.INITIAL_BYTES_LENGTH + message.Length]; - TcpMessageStreamHelpers.CreatePayload(payload, message, CommandCodes.PURGE_STREAM_CODE); - - await SendAckAsync(payload, token); + await SendAckAsync(CommandCodes.PURGE_STREAM_CODE, message, token); } /// public async Task DeleteStreamAsync(Identifier streamId, CancellationToken token = default) { var message = TcpMessageStreamHelpers.GetBytesFromIdentifier(streamId); - var payload = new byte[4 + BufferSizes.INITIAL_BYTES_LENGTH + message.Length]; - TcpMessageStreamHelpers.CreatePayload(payload, message, CommandCodes.DELETE_STREAM_CODE); - - await SendAckAsync(payload, token); + await SendAckAsync(CommandCodes.DELETE_STREAM_CODE, message, token); } /// @@ -220,10 +201,8 @@ public async Task> GetTopicsAsync(Identifier stream CancellationToken token = default) { var message = TcpMessageStreamHelpers.GetBytesFromIdentifier(streamId); - var payload = new byte[4 + BufferSizes.INITIAL_BYTES_LENGTH + message.Length]; - TcpMessageStreamHelpers.CreatePayload(payload, message, CommandCodes.GET_TOPICS_CODE); - - using IMemoryOwner responseBuffer = await SendWithResponseAsync(payload, token); + using IMemoryOwner responseBuffer + = await SendWithResponseAsync(CommandCodes.GET_TOPICS_CODE, message, token: token); if (responseBuffer.Memory.Length == 0) { @@ -238,10 +217,8 @@ public async Task> GetTopicsAsync(Identifier stream CancellationToken token = default) { var message = TcpContracts.GetTopicById(streamId, topicId); - var payload = new byte[4 + BufferSizes.INITIAL_BYTES_LENGTH + message.Length]; - TcpMessageStreamHelpers.CreatePayload(payload, message, CommandCodes.GET_TOPIC_CODE); - - using IMemoryOwner responseBuffer = await SendWithResponseAsync(payload, token); + using IMemoryOwner responseBuffer + = await SendWithResponseAsync(CommandCodes.GET_TOPIC_CODE, message, token: token); if (responseBuffer.Memory.Length == 0) { @@ -259,10 +236,8 @@ public async Task> GetTopicsAsync(Identifier stream var messageExpiryValue = DurationHelpers.ToDuration(messageExpiry); var message = TcpContracts.CreateTopic(streamId, name, partitionsCount, compressionAlgorithm, replicationFactor, messageExpiryValue, maxTopicSize); - var payload = new byte[4 + BufferSizes.INITIAL_BYTES_LENGTH + message.Length]; - TcpMessageStreamHelpers.CreatePayload(payload, message, CommandCodes.CREATE_TOPIC_CODE); - - using IMemoryOwner responseBuffer = await SendWithResponseAsync(payload, token); + using IMemoryOwner responseBuffer + = await SendWithResponseAsync(CommandCodes.CREATE_TOPIC_CODE, message, token: token); if (responseBuffer.Memory.Length == 0) { @@ -281,20 +256,14 @@ public async Task UpdateTopicAsync(Identifier streamId, Identifier topicId, stri var messageExpiryValue = DurationHelpers.ToDuration(messageExpiry); var message = TcpContracts.UpdateTopic(streamId, topicId, name, compressionAlgorithm, maxTopicSize, messageExpiryValue, replicationFactor); - var payload = new byte[4 + BufferSizes.INITIAL_BYTES_LENGTH + message.Length]; - TcpMessageStreamHelpers.CreatePayload(payload, message, CommandCodes.UPDATE_TOPIC_CODE); - - await SendAckAsync(payload, token); + await SendAckAsync(CommandCodes.UPDATE_TOPIC_CODE, message, token); } /// public async Task DeleteTopicAsync(Identifier streamId, Identifier topicId, CancellationToken token = default) { var message = TcpContracts.DeleteTopic(streamId, topicId); - var payload = new byte[4 + BufferSizes.INITIAL_BYTES_LENGTH + message.Length]; - TcpMessageStreamHelpers.CreatePayload(payload, message, CommandCodes.DELETE_TOPIC_CODE); - - await SendAckAsync(payload, token); + await SendAckAsync(CommandCodes.DELETE_TOPIC_CODE, message, token); _groupState.InvalidatePartitionCount(new TopicKey(streamId, topicId)); } @@ -302,10 +271,7 @@ public async Task DeleteTopicAsync(Identifier streamId, Identifier topicId, Canc public async Task PurgeTopicAsync(Identifier streamId, Identifier topicId, CancellationToken token = default) { var message = TcpContracts.PurgeTopic(streamId, topicId); - var payload = new byte[4 + BufferSizes.INITIAL_BYTES_LENGTH + message.Length]; - TcpMessageStreamHelpers.CreatePayload(payload, message, CommandCodes.PURGE_TOPIC_CODE); - - await SendAckAsync(payload, token); + await SendAckAsync(CommandCodes.PURGE_TOPIC_CODE, message, token); } @@ -379,10 +345,7 @@ public async Task StoreOffsetAsync(Consumer consumer, Identifier streamId, Ident uint? partitionId, CancellationToken token = default) { var message = TcpContracts.UpdateOffset(streamId, topicId, consumer, offset, partitionId); - var payload = new byte[4 + BufferSizes.INITIAL_BYTES_LENGTH + message.Length]; - TcpMessageStreamHelpers.CreatePayload(payload, message, CommandCodes.STORE_CONSUMER_OFFSET_CODE); - - await SendAckAsync(payload, token); + await SendAckAsync(CommandCodes.STORE_CONSUMER_OFFSET_CODE, message, token); } /// @@ -390,10 +353,8 @@ public async Task StoreOffsetAsync(Consumer consumer, Identifier streamId, Ident uint? partitionId, CancellationToken token = default) { var message = TcpContracts.GetOffset(streamId, topicId, consumer, partitionId); - var payload = new byte[4 + BufferSizes.INITIAL_BYTES_LENGTH + message.Length]; - TcpMessageStreamHelpers.CreatePayload(payload, message, CommandCodes.GET_CONSUMER_OFFSET_CODE); - - using IMemoryOwner responseBuffer = await SendWithResponseAsync(payload, token); + using IMemoryOwner responseBuffer + = await SendWithResponseAsync(CommandCodes.GET_CONSUMER_OFFSET_CODE, message, token: token); if (responseBuffer.Memory.Length == 0) { @@ -408,10 +369,7 @@ public async Task DeleteOffsetAsync(Consumer consumer, Identifier streamId, Iden CancellationToken token = default) { var message = TcpContracts.DeleteOffset(streamId, topicId, consumer, partitionId); - var payload = new byte[4 + BufferSizes.INITIAL_BYTES_LENGTH + message.Length]; - TcpMessageStreamHelpers.CreatePayload(payload, message, CommandCodes.DELETE_CONSUMER_OFFSET_CODE); - - await SendAckAsync(payload, token); + await SendAckAsync(CommandCodes.DELETE_CONSUMER_OFFSET_CODE, message, token); } /// @@ -420,10 +378,8 @@ public async Task> GetConsumerGroupsAsync(I CancellationToken token = default) { var message = TcpContracts.GetGroups(streamId, topicId); - var payload = new byte[4 + BufferSizes.INITIAL_BYTES_LENGTH + message.Length]; - TcpMessageStreamHelpers.CreatePayload(payload, message, CommandCodes.GET_CONSUMER_GROUPS_CODE); - - using IMemoryOwner responseBuffer = await SendWithResponseAsync(payload, token); + using IMemoryOwner responseBuffer + = await SendWithResponseAsync(CommandCodes.GET_CONSUMER_GROUPS_CODE, message, token: token); if (responseBuffer.Memory.Length == 0) { @@ -438,10 +394,8 @@ public async Task> GetConsumerGroupsAsync(I Identifier groupId, CancellationToken token = default) { var message = TcpContracts.GetGroup(streamId, topicId, groupId); - var payload = new byte[4 + BufferSizes.INITIAL_BYTES_LENGTH + message.Length]; - TcpMessageStreamHelpers.CreatePayload(payload, message, CommandCodes.GET_CONSUMER_GROUP_CODE); - - using IMemoryOwner responseBuffer = await SendWithResponseAsync(payload, token); + using IMemoryOwner responseBuffer + = await SendWithResponseAsync(CommandCodes.GET_CONSUMER_GROUP_CODE, message, token: token); if (responseBuffer.Memory.Length == 0) { @@ -456,10 +410,8 @@ public async Task> GetConsumerGroupsAsync(I string name, CancellationToken token = default) { var message = TcpContracts.CreateGroup(streamId, topicId, name); - var payload = new byte[4 + BufferSizes.INITIAL_BYTES_LENGTH + message.Length]; - TcpMessageStreamHelpers.CreatePayload(payload, message, CommandCodes.CREATE_CONSUMER_GROUP_CODE); - - using IMemoryOwner responseBuffer = await SendWithResponseAsync(payload, token); + using IMemoryOwner responseBuffer + = await SendWithResponseAsync(CommandCodes.CREATE_CONSUMER_GROUP_CODE, message, token: token); if (responseBuffer.Memory.Length == 0) { @@ -474,10 +426,7 @@ public async Task DeleteConsumerGroupAsync(Identifier streamId, Identifier topic CancellationToken token = default) { var message = TcpContracts.DeleteGroup(streamId, topicId, groupId); - var payload = new byte[4 + BufferSizes.INITIAL_BYTES_LENGTH + message.Length]; - TcpMessageStreamHelpers.CreatePayload(payload, message, CommandCodes.DELETE_CONSUMER_GROUP_CODE); - - await SendAckAsync(payload, token); + await SendAckAsync(CommandCodes.DELETE_CONSUMER_GROUP_CODE, message, token); _groupState.DeregisterGroup(new GroupKey(streamId, topicId, groupId)); } @@ -486,10 +435,7 @@ public async Task JoinConsumerGroupAsync(Identifier streamId, Identifier topicId CancellationToken token = default) { var message = TcpContracts.JoinGroup(streamId, topicId, groupId); - var payload = new byte[4 + BufferSizes.INITIAL_BYTES_LENGTH + message.Length]; - TcpMessageStreamHelpers.CreatePayload(payload, message, CommandCodes.JOIN_CONSUMER_GROUP_CODE); - - await SendAckAsync(payload, token); + await SendAckAsync(CommandCodes.JOIN_CONSUMER_GROUP_CODE, message, token); // A join rebalances the group, so whatever this client holds for it is a generation behind and every // poll under it would be fenced until the first re-sync. @@ -501,10 +447,7 @@ public async Task LeaveConsumerGroupAsync(Identifier streamId, Identifier topicI CancellationToken token = default) { var message = TcpContracts.LeaveGroup(streamId, topicId, groupId); - var payload = new byte[4 + BufferSizes.INITIAL_BYTES_LENGTH + message.Length]; - TcpMessageStreamHelpers.CreatePayload(payload, message, CommandCodes.LEAVE_CONSUMER_GROUP_CODE); - - await SendAckAsync(payload, token); + await SendAckAsync(CommandCodes.LEAVE_CONSUMER_GROUP_CODE, message, token); _groupState.DeregisterGroup(new GroupKey(streamId, topicId, groupId)); } @@ -513,10 +456,7 @@ public async Task DeletePartitionsAsync(Identifier streamId, Identifier topicId, CancellationToken token = default) { var message = TcpContracts.DeletePartitions(streamId, topicId, partitionsCount); - var payload = new byte[4 + BufferSizes.INITIAL_BYTES_LENGTH + message.Length]; - TcpMessageStreamHelpers.CreatePayload(payload, message, CommandCodes.DELETE_PARTITIONS_CODE); - - await SendAckAsync(payload, token); + await SendAckAsync(CommandCodes.DELETE_PARTITIONS_CODE, message, token); _groupState.InvalidatePartitionCount(new TopicKey(streamId, topicId)); } @@ -525,10 +465,7 @@ public async Task CreatePartitionsAsync(Identifier streamId, Identifier topicId, CancellationToken token = default) { var message = TcpContracts.CreatePartitions(streamId, topicId, partitionsCount); - var payload = new byte[4 + BufferSizes.INITIAL_BYTES_LENGTH + message.Length]; - TcpMessageStreamHelpers.CreatePayload(payload, message, CommandCodes.CREATE_PARTITIONS_CODE); - - await SendAckAsync(payload, token); + await SendAckAsync(CommandCodes.CREATE_PARTITIONS_CODE, message, token); _groupState.InvalidatePartitionCount(new TopicKey(streamId, topicId)); } @@ -537,20 +474,14 @@ public async Task DeleteSegmentsAsync(Identifier streamId, Identifier topicId, u uint segmentsCount, CancellationToken token = default) { var message = TcpContracts.DeleteSegments(streamId, topicId, partitionId, segmentsCount); - var payload = new byte[4 + BufferSizes.INITIAL_BYTES_LENGTH + message.Length]; - TcpMessageStreamHelpers.CreatePayload(payload, message, CommandCodes.DELETE_SEGMENTS_CODE); - - await SendAckAsync(payload, token); + await SendAckAsync(CommandCodes.DELETE_SEGMENTS_CODE, message, token); } /// public async Task GetMeAsync(CancellationToken token = default) { var message = Array.Empty(); - var payload = new byte[4 + BufferSizes.INITIAL_BYTES_LENGTH + message.Length]; - TcpMessageStreamHelpers.CreatePayload(payload, message, CommandCodes.GET_ME_CODE); - - using IMemoryOwner responseBuffer = await SendWithResponseAsync(payload, token); + using IMemoryOwner responseBuffer = await SendWithResponseAsync(CommandCodes.GET_ME_CODE, message, token: token); if (responseBuffer.Memory.Length == 0) { @@ -564,10 +495,8 @@ public async Task DeleteSegmentsAsync(Identifier streamId, Identifier topicId, u public async Task GetStatsAsync(CancellationToken token = default) { var message = Array.Empty(); - var payload = new byte[4 + BufferSizes.INITIAL_BYTES_LENGTH + message.Length]; - TcpMessageStreamHelpers.CreatePayload(payload, message, CommandCodes.GET_STATS_CODE); - - using IMemoryOwner responseBuffer = await SendWithResponseAsync(payload, token); + using IMemoryOwner responseBuffer + = await SendWithResponseAsync(CommandCodes.GET_STATS_CODE, message, token: token); if (responseBuffer.Memory.Length == 0) { @@ -581,10 +510,8 @@ public async Task DeleteSegmentsAsync(Identifier streamId, Identifier topicId, u public async Task GetClusterMetadataAsync(CancellationToken token = default) { var message = Array.Empty(); - var payload = new byte[4 + BufferSizes.INITIAL_BYTES_LENGTH + message.Length]; - TcpMessageStreamHelpers.CreatePayload(payload, message, CommandCodes.GET_CLUSTER_METADATA_CODE); - - using IMemoryOwner responseBuffer = await SendWithResponseAsync(payload, token); + using IMemoryOwner responseBuffer + = await SendWithResponseAsync(CommandCodes.GET_CLUSTER_METADATA_CODE, message, token: token); if (responseBuffer.Memory.Length == 0) { @@ -598,10 +525,7 @@ public async Task DeleteSegmentsAsync(Identifier streamId, Identifier topicId, u public async Task PingAsync(CancellationToken token = default) { var message = Array.Empty(); - var payload = new byte[4 + BufferSizes.INITIAL_BYTES_LENGTH + message.Length]; - TcpMessageStreamHelpers.CreatePayload(payload, message, CommandCodes.PING_CODE); - - await SendAckAsync(payload, token); + await SendAckAsync(CommandCodes.PING_CODE, message, token); await RefreshGroupAssignmentsAsync(token); } @@ -611,10 +535,7 @@ public async Task GetSnapshotAsync(SnapshotCompression compression, IList snapshotTypes, CancellationToken token = default) { var message = TcpContracts.GetSnapshot(compression, snapshotTypes); - var payload = new byte[4 + BufferSizes.INITIAL_BYTES_LENGTH + message.Length]; - TcpMessageStreamHelpers.CreatePayload(payload, message, CommandCodes.GET_SNAPSHOT_CODE); - - using IMemoryOwner result = await SendWithResponseAsync(payload, token); + using IMemoryOwner result = await SendWithResponseAsync(CommandCodes.GET_SNAPSHOT_CODE, message, token: token); return result.Memory.Span.ToArray(); } @@ -624,16 +545,13 @@ public async Task SendBinaryRequestAsync(uint code, byte[] payload, Canc { if (SessionControlCodes.Contains(code)) { - throw new IggyInvalidStatusCodeException(InvalidCommandStatus, - $"Invalid response status code: {InvalidCommandStatus}"); + throw VsrError.Exception(VsrError.INVALID_COMMAND, + $"Command {code} cannot be sent as a raw binary request."); } - var buffer = new byte[4 + BufferSizes.INITIAL_BYTES_LENGTH + payload.Length]; - TcpMessageStreamHelpers.CreatePayload(buffer, payload, (int)code); - - using IMemoryOwner result = await SendWithResponseAsync(buffer, token); + using IMemoryOwner result = await SendWithResponseAsync((int)code, payload, token: token); - return result.Memory.Length <= 1 ? [] : result.Memory.Span.ToArray(); + return result.Memory.Length == 0 ? [] : result.Memory.Span.ToArray(); } /// @@ -646,10 +564,8 @@ public Task ConnectAsync(CancellationToken token = default) public async Task> GetClientsAsync(CancellationToken token = default) { var message = Array.Empty(); - var payload = new byte[4 + BufferSizes.INITIAL_BYTES_LENGTH + message.Length]; - TcpMessageStreamHelpers.CreatePayload(payload, message, CommandCodes.GET_CLIENTS_CODE); - - using IMemoryOwner responseBuffer = await SendWithResponseAsync(payload, token); + using IMemoryOwner responseBuffer + = await SendWithResponseAsync(CommandCodes.GET_CLIENTS_CODE, message, token: token); if (responseBuffer.Memory.Length == 0) { @@ -663,10 +579,8 @@ public async Task> GetClientsAsync(CancellationTok public async Task GetClientByIdAsync(uint clientId, CancellationToken token = default) { var message = TcpContracts.GetClient(clientId); - var payload = new byte[4 + BufferSizes.INITIAL_BYTES_LENGTH + message.Length]; - TcpMessageStreamHelpers.CreatePayload(payload, message, CommandCodes.GET_CLIENT_CODE); - - using IMemoryOwner responseBuffer = await SendWithResponseAsync(payload, token); + using IMemoryOwner responseBuffer + = await SendWithResponseAsync(CommandCodes.GET_CLIENT_CODE, message, token: token); if (responseBuffer.Memory.Length == 0) { @@ -680,10 +594,8 @@ public async Task> GetClientsAsync(CancellationTok public async Task GetUserAsync(Identifier userId, CancellationToken token = default) { var message = TcpContracts.GetUser(userId); - var payload = new byte[4 + BufferSizes.INITIAL_BYTES_LENGTH + message.Length]; - TcpMessageStreamHelpers.CreatePayload(payload, message, CommandCodes.GET_USER_CODE); - - using IMemoryOwner responseBuffer = await SendWithResponseAsync(payload, token); + using IMemoryOwner responseBuffer + = await SendWithResponseAsync(CommandCodes.GET_USER_CODE, message, token: token); if (responseBuffer.Memory.Length == 0) { @@ -697,10 +609,8 @@ public async Task> GetClientsAsync(CancellationTok public async Task> GetUsersAsync(CancellationToken token = default) { var message = Array.Empty(); - var payload = new byte[4 + BufferSizes.INITIAL_BYTES_LENGTH + message.Length]; - TcpMessageStreamHelpers.CreatePayload(payload, message, CommandCodes.GET_USERS_CODE); - - using IMemoryOwner responseBuffer = await SendWithResponseAsync(payload, token); + using IMemoryOwner responseBuffer + = await SendWithResponseAsync(CommandCodes.GET_USERS_CODE, message, token: token); if (responseBuffer.Memory.Length == 0) { @@ -715,10 +625,8 @@ public async Task> GetUsersAsync(CancellationToken t Permissions? permissions = null, CancellationToken token = default) { var message = TcpContracts.CreateUser(userName, password, status, permissions); - var payload = new byte[4 + BufferSizes.INITIAL_BYTES_LENGTH + message.Length]; - TcpMessageStreamHelpers.CreatePayload(payload, message, CommandCodes.CREATE_USER_CODE); - - using IMemoryOwner responseBuffer = await SendWithResponseAsync(payload, token); + using IMemoryOwner responseBuffer + = await SendWithResponseAsync(CommandCodes.CREATE_USER_CODE, message, token: token); if (responseBuffer.Memory.Length == 0) { @@ -732,10 +640,7 @@ public async Task> GetUsersAsync(CancellationToken t public async Task DeleteUserAsync(Identifier userId, CancellationToken token = default) { var message = TcpContracts.DeleteUser(userId); - var payload = new byte[4 + BufferSizes.INITIAL_BYTES_LENGTH + message.Length]; - TcpMessageStreamHelpers.CreatePayload(payload, message, CommandCodes.DELETE_USER_CODE); - - await SendAckAsync(payload, token); + await SendAckAsync(CommandCodes.DELETE_USER_CODE, message, token); } /// @@ -743,10 +648,7 @@ public async Task UpdateUserAsync(Identifier userId, string? userName = null, Us CancellationToken token = default) { var message = TcpContracts.UpdateUser(userId, userName, status); - var payload = new byte[4 + BufferSizes.INITIAL_BYTES_LENGTH + message.Length]; - TcpMessageStreamHelpers.CreatePayload(payload, message, CommandCodes.UPDATE_USER_CODE); - - await SendAckAsync(payload, token); + await SendAckAsync(CommandCodes.UPDATE_USER_CODE, message, token); } /// @@ -754,10 +656,7 @@ public async Task UpdatePermissionsAsync(Identifier userId, Permissions? permiss CancellationToken token = default) { var message = TcpContracts.UpdatePermissions(userId, permissions); - var payload = new byte[4 + BufferSizes.INITIAL_BYTES_LENGTH + message.Length]; - TcpMessageStreamHelpers.CreatePayload(payload, message, CommandCodes.UPDATE_PERMISSIONS_CODE); - - await SendAckAsync(payload, token); + await SendAckAsync(CommandCodes.UPDATE_PERMISSIONS_CODE, message, token); } /// @@ -765,16 +664,13 @@ public async Task ChangePasswordAsync(Identifier userId, string currentPassword, CancellationToken token = default) { var message = TcpContracts.ChangePassword(userId, currentPassword, newPassword); - var payload = new byte[4 + BufferSizes.INITIAL_BYTES_LENGTH + message.Length]; - TcpMessageStreamHelpers.CreatePayload(payload, message, CommandCodes.CHANGE_PASSWORD_CODE); - - await SendAckAsync(payload, token); + await SendAckAsync(CommandCodes.CHANGE_PASSWORD_CODE, message, token); } /// public async Task LoginUserAsync(string userName, string password, CancellationToken token = default) { - if (_state == ConnectionState.Disconnected) + if (State == ConnectionState.Disconnected) { throw new NotConnectedException(); } @@ -786,21 +682,17 @@ public async Task ChangePasswordAsync(Identifier userId, string currentPassword, /// public async Task LogoutUserAsync(CancellationToken token = default) { - var message = Array.Empty(); - var payload = new byte[4 + BufferSizes.INITIAL_BYTES_LENGTH + message.Length]; - TcpMessageStreamHelpers.CreatePayload(payload, message, CommandCodes.LOGOUT_USER_CODE); - try { - await SendAckAsync(payload, token); + await SendAckAsync(CommandCodes.LOGOUT_USER_CODE, ReadOnlyMemory.Empty, token); } finally { await ResetConsensusSessionAsync(); - if (_state == ConnectionState.Authenticated) + if (State == ConnectionState.Authenticated) { - SetConnectionStateAsync(ConnectionState.Connected); + SetConnectionState(ConnectionState.Connected); } } } @@ -810,10 +702,8 @@ public async Task> GetPersonalAccessT CancellationToken token = default) { var message = Array.Empty(); - var payload = new byte[4 + BufferSizes.INITIAL_BYTES_LENGTH + message.Length]; - TcpMessageStreamHelpers.CreatePayload(payload, message, CommandCodes.GET_PERSONAL_ACCESS_TOKENS_CODE); - - using IMemoryOwner responseBuffer = await SendWithResponseAsync(payload, token); + using IMemoryOwner responseBuffer + = await SendWithResponseAsync(CommandCodes.GET_PERSONAL_ACCESS_TOKENS_CODE, message, token: token); if (responseBuffer.Memory.Length == 0) { @@ -828,10 +718,8 @@ public async Task> GetPersonalAccessT CancellationToken token = default) { var message = TcpContracts.CreatePersonalAccessToken(name, DurationHelpers.ToDuration(expiry)); - var payload = new byte[4 + BufferSizes.INITIAL_BYTES_LENGTH + message.Length]; - TcpMessageStreamHelpers.CreatePayload(payload, message, CommandCodes.CREATE_PERSONAL_ACCESS_TOKEN_CODE); - - using IMemoryOwner responseBuffer = await SendWithResponseAsync(payload, token); + using IMemoryOwner responseBuffer + = await SendWithResponseAsync(CommandCodes.CREATE_PERSONAL_ACCESS_TOKEN_CODE, message, token: token); if (responseBuffer.Memory.Length == 0) { @@ -845,10 +733,7 @@ public async Task> GetPersonalAccessT public async Task DeletePersonalAccessTokenAsync(string name, CancellationToken token = default) { var message = TcpContracts.DeletePersonalRequestToken(name); - var payload = new byte[4 + BufferSizes.INITIAL_BYTES_LENGTH + message.Length]; - TcpMessageStreamHelpers.CreatePayload(payload, message, CommandCodes.DELETE_PERSONAL_ACCESS_TOKEN_CODE); - - await SendAckAsync(payload, token); + await SendAckAsync(CommandCodes.DELETE_PERSONAL_ACCESS_TOKEN_CODE, message, token); } /// @@ -865,7 +750,7 @@ public async Task DeletePersonalAccessTokenAsync(string name, CancellationToken /// private async Task ConnectAsync(bool autoLogin, CancellationToken token) { - if (_state is ConnectionState.Connected + if (State is ConnectionState.Connected or ConnectionState.Authenticating or ConnectionState.Authenticated) { @@ -877,7 +762,7 @@ or ConnectionState.Authenticating Interlocked.Exchange(ref _isConnecting, 1); try { - if (_state is ConnectionState.Connected + if (State is ConnectionState.Connected or ConnectionState.Authenticating or ConnectionState.Authenticated) { @@ -889,7 +774,7 @@ or ConnectionState.Authenticating await Task.Delay(_configuration.ReconnectionSettings.InitialDelay, token); } - SetConnectionStateAsync(ConnectionState.Connecting); + SetConnectionState(ConnectionState.Connecting); await TryEstablishConnectionAsync(autoLogin, token); } finally @@ -904,18 +789,16 @@ private async Task PollPartitionMessagesRentedAsync(Identi CancellationToken token) { var messageBufferSize = CalculateMessageBufferSize(streamId, topicId, consumer); - var payloadBufferSize = CalculatePayloadBufferSize(messageBufferSize); - var payload = ArrayPool.Shared.Rent(payloadBufferSize); + var payload = ArrayPool.Shared.Rent(messageBufferSize); IMemoryOwner? responseBuffer = null; try { - TcpContracts.GetMessages(payload.AsSpan().Slice(8, messageBufferSize), consumer, streamId, + TcpContracts.GetMessages(payload.AsSpan(0, messageBufferSize), consumer, streamId, topicId, pollingStrategy, count, autoCommit, partitionId); - BinaryPrimitives.WriteInt32LittleEndian(payload.AsSpan()[..4], messageBufferSize + 4); - BinaryPrimitives.WriteInt32LittleEndian(payload.AsSpan()[4..8], CommandCodes.POLL_MESSAGES_CODE); - responseBuffer = await SendWithResponseAsync(payload.AsMemory(0, payloadBufferSize), token); + responseBuffer = await SendWithResponseAsync(CommandCodes.POLL_MESSAGES_CODE, + payload.AsMemory(0, messageBufferSize), token: token); if (responseBuffer.Memory.Length == 0) { responseBuffer.Dispose(); @@ -959,15 +842,13 @@ private Task SendMessagesCoreAsync(Identifier streamId, Id + 2 + partitioning.Length + 4 + 4; var maxMessageBufferSize = TcpMessageStreamHelpers.CalculateMessageBytesCount(messages, encryptor) + metadataLength; - var maxPayloadBufferSize = CalculatePayloadBufferSize(maxMessageBufferSize); - IMemoryOwner payloadBuffer = MemoryPool.Shared.Rent(maxPayloadBufferSize); - int payloadBufferSize; + IMemoryOwner payloadBuffer = MemoryPool.Shared.Rent(maxMessageBufferSize); + int bodySize; try { - var messageBufferSize = FillSendMessagesPayload(payloadBuffer.Memory.Span, maxMessageBufferSize, - streamId, topicId, partitioning, messages, encryptor); - payloadBufferSize = CalculatePayloadBufferSize(messageBufferSize); + bodySize = TcpContracts.CreateMessage(payloadBuffer.Memory.Span[..maxMessageBufferSize], streamId, + topicId, partitioning, messages, encryptor); } catch { @@ -975,16 +856,16 @@ private Task SendMessagesCoreAsync(Identifier streamId, Id throw; } - return SendConfirmedAndDisposeAsync(payloadBuffer, payloadBufferSize, token); + return SendConfirmedAndDisposeAsync(payloadBuffer, bodySize, token); } private async Task SendConfirmedAndDisposeAsync(IMemoryOwner payloadBuffer, - int payloadBufferSize, CancellationToken token) + int bodySize, CancellationToken token) { try { - using IMemoryOwner responseBuffer = - await SendWithResponseAsync(payloadBuffer.Memory[..payloadBufferSize], token); + using IMemoryOwner responseBuffer = await SendWithResponseAsync(CommandCodes.SEND_MESSAGES_CODE, + payloadBuffer.Memory[..bodySize], token: token); return BinaryMapper.MapSendMessages(responseBuffer.Memory.Span); } finally @@ -1003,18 +884,6 @@ private static ReadOnlySpan AsSpan(IList messages) }; } - [MethodImpl(MethodImplOptions.AggressiveInlining)] - private static int FillSendMessagesPayload(Span buffer, int maxMessageBufferSize, - Identifier streamId, Identifier topicId, Partitioning partitioning, ReadOnlySpan messages, - IMessageEncryptor? encryptor) - { - var messageBufferSize = TcpContracts.CreateMessage(buffer.Slice(8, maxMessageBufferSize), streamId, topicId, - partitioning, messages, encryptor); - BinaryPrimitives.WriteInt32LittleEndian(buffer[..4], messageBufferSize + 4); - BinaryPrimitives.WriteInt32LittleEndian(buffer[4..8], CommandCodes.SEND_MESSAGES_CODE); - return messageBufferSize; - } - private async Task TryEstablishConnectionAsync(bool autoLogin, CancellationToken token) { var retryCount = 0; @@ -1022,12 +891,13 @@ private async Task TryEstablishConnectionAsync(bool autoLogin, CancellationToken var delay = _configuration.ReconnectionSettings.InitialDelay; do { - // The sending semaphore owns every write to _stream, so an in-flight request never observes the - // field changing between its write and its reply. + // The sending semaphore owns every write to _connection, so an in-flight request never observes + // the field changing between its write and its reply. await _sendingSemaphore.WaitAsync(token); try { - _stream?.Dispose(); + _connection?.Dispose(); + _connection = null; ResetConsensusSession(); } @@ -1063,27 +933,28 @@ private async Task TryEstablishConnectionAsync(bool autoLogin, CancellationToken _currentRemoteAddress = socket.RemoteEndPoint is IPEndPoint remote ? ServerAddress.HostPort(remote.Address.ToString(), (ushort)remote.Port) : string.Empty; + _connectedAddress = _currentAddress; socket.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.KeepAlive, true); socket.SetSocketOption(SocketOptionLevel.Tcp, SocketOptionName.TcpKeepAliveTime, 5); - var connectionStream = _configuration.TlsSettings.Enabled switch - { - true => await CreateSslStreamAndAuthenticate(socket, _configuration.TlsSettings), - false => new TcpConnectionStream(new NetworkStream(socket, true)) - }; + var connectionStream = _configuration.TlsSettings.Enabled + ? await CreateSslStreamAndAuthenticate(socket, _configuration.TlsSettings) + : new NetworkStream(socket, true); await _sendingSemaphore.WaitAsync(token); try { - _stream = connectionStream; + _connection = new VsrConnection(connectionStream, _consensusSession, + _configuration.MaxResponseFrameSize, VsrRequestTimeoutMs, DropVsrConnectionLocked, + _logger); } finally { _sendingSemaphore.Release(); } - SetConnectionStateAsync(ConnectionState.Connected); + SetConnectionState(ConnectionState.Connected); _lastConnectionTime = DateTimeOffset.UtcNow; socket = null; @@ -1091,7 +962,7 @@ private async Task TryEstablishConnectionAsync(bool autoLogin, CancellationToken // No pre-login roster read: the server auth-gates cluster metadata, so leadership settles after // a sign-in binds a session. A login dialed at a backup still succeeds because the server // forwards the register to the primary. - if (autoLogin && _configuration.AutoLoginSettings.Enabled && !ConsumeSkipAutoLogin()) + if (autoLogin && _configuration.AutoLoginSettings.Enabled) { _logger.LogInformation("Auto login enabled. Trying to login with credentials: {Username}", _configuration.AutoLoginSettings.Username); @@ -1117,7 +988,10 @@ await LoginUserAsync(_configuration.AutoLoginSettings.Username, (_configuration.ReconnectionSettings.MaxRetries > 0 && retryCount >= _configuration.ReconnectionSettings.MaxRetries)) { - SetConnectionStateAsync(ConnectionState.Disconnected); + // A failure past the socket handoff (TLS handshake, auto login, redirect probe) leaves + // _connection holding a live stream that nothing would ever close once this throw lands. + await DropVsrConnectionAsync(_connection); + SetConnectionState(ConnectionState.Disconnected); throw; } @@ -1149,7 +1023,7 @@ async Task BackoffOrThrowAsync() { if (++redirects > VsrMaxLeaderRedirects) { - SetConnectionStateAsync(ConnectionState.Disconnected); + SetConnectionState(ConnectionState.Disconnected); throw new MissingLeaderException(); } @@ -1159,23 +1033,7 @@ async Task BackoffOrThrowAsync() } } - /// - /// Whether this connect was triggered by a login or register request that will re-authenticate itself, - /// so the auto-login must sit this one out. Consumes the flag. - /// - private bool ConsumeSkipAutoLogin() - { - if (Interlocked.Exchange(ref _skipAutoLoginOnce, 0) == 0) - { - return false; - } - - _logger.LogInformation("Skipping auto login for a replayed register request"); - - return true; - } - - private async Task CreateSslStreamAndAuthenticate(Socket socket, TlsSettings tlsSettings) + private async Task CreateSslStreamAndAuthenticate(Socket socket, TlsSettings tlsSettings) { ValidateCertificatePath(tlsSettings.CertificatePath); @@ -1186,59 +1044,59 @@ private async Task CreateSslStreamAndAuthenticate(Socket so await sslStream.AuthenticateAsClientAsync(tlsSettings.Hostname); - return new TcpConnectionStream(sslStream); + return sslStream; } - private async Task SendAckAsync(ReadOnlyMemory payload, CancellationToken token = default) + private async Task SendAckAsync(int code, ReadOnlyMemory body, CancellationToken token) { - using IMemoryOwner _ = await SendWithResponseAsync(payload, token); + using IMemoryOwner _ = await SendWithResponseAsync(code, body, token: token); } - private async Task> SendWithResponseAsync(ReadOnlyMemory payload, - CancellationToken token = default) + private async Task> SendWithResponseAsync(int code, ReadOnlyMemory body, + bool autoLoginOnReconnect = true, CancellationToken token = default) { try { - return await SendRawAsync(payload, token); + return await SendRawAsync(code, body, token); } - catch (Exception e) when (IsConnectionException(e) && !IsConnecting && !_disposed) + catch (Exception e) when (VsrConnection.IsConnectionException(e) && !IsConnecting && !_disposed) { _logger.LogWarning("Connection lost"); if (!_configuration.ReconnectionSettings.Enabled) { _logger.LogWarning("Reconnection is disabled"); - SetConnectionStateAsync(ConnectionState.Disconnected); + SetConnectionState(ConnectionState.Disconnected); throw; } - return await HandleReconnectionAsync(payload, token); + return await HandleReconnectionAsync(code, body, autoLoginOnReconnect, token); } } - private async Task> HandleReconnectionAsync(ReadOnlyMemory payload, - CancellationToken token) + private async Task> HandleReconnectionAsync(int code, ReadOnlyMemory body, + bool autoLogin, CancellationToken token) { var currentTime = DateTimeOffset.UtcNow; await _connectionSemaphore.WaitAsync(token); try { - if (_state is ConnectionState.Connected or ConnectionState.Authenticated + if (State is ConnectionState.Connected or ConnectionState.Authenticated && _lastConnectionTime > currentTime) { _logger.LogInformation("Connection already established, sending payload"); - return await SendRawAsync(payload, token); + return await SendRawAsync(code, body, token); } - SetConnectionStateAsync(ConnectionState.Disconnected); + SetConnectionState(ConnectionState.Disconnected); _logger.LogInformation("Reconnecting to the server"); - await ConnectAsync(token); + await ConnectAsync(autoLogin, token); _logger.LogInformation("Reconnected to the server"); await Task.Delay(_configuration.ReconnectionSettings.WaitAfterReconnect, token); - return await SendRawAsync(payload, token); + return await SendRawAsync(code, body, token); } finally { @@ -1246,32 +1104,6 @@ private async Task> HandleReconnectionAsync(ReadOnlyMemory> SendRawAsync(ReadOnlyMemory payload, CancellationToken token) - { - ObjectDisposedException.ThrowIf(_disposed, this); - - if (_state is ConnectionState.Disconnected or ConnectionState.Connecting) - { - throw new NotConnectedException(); - } - - return SendRawVsrAsync(payload, token); - } - - private static bool IsConnectionException(Exception ex) - { - return ex is IggyZeroBytesException or - NotConnectedException or - SocketException or - IOException or - ObjectDisposedException; - } - - private static int CalculatePayloadBufferSize(int messageBufferSize) - { - return messageBufferSize + 4 + BufferSizes.INITIAL_BYTES_LENGTH; - } - private static int CalculateMessageBufferSize(Identifier streamId, Identifier topicId, Consumer consumer) { // Original: 14 + 5 + 2 + streamId.Length + 2 + topicId.Length + 2 + consumer.Id.Length @@ -1281,19 +1113,20 @@ private static int CalculateMessageBufferSize(Identifier streamId, Identifier to /// /// Sets the connection state and publishes a ConnectionStateChangedEventArgs to subscribers via the connection event - /// aggregator. + /// aggregator. Callers reach this holding different locks - the connect loop the connection semaphore, a drop the + /// sending one, the login path neither - so the swap has to be atomic: a read-modify-write would let a concurrent + /// transition publish a previous state that never preceded the current one. Only the thread that changed the value + /// publishes, so a state is never announced twice. /// /// The new connection state - private void SetConnectionStateAsync(ConnectionState newState) + private void SetConnectionState(ConnectionState newState) { - if (_state == newState) + var previousState = (ConnectionState)Interlocked.Exchange(ref _stateValue, (int)newState); + if (previousState == newState) { return; } - var previousState = _state; - _state = newState; - _logger.LogInformation("Connection state changed: {PreviousState} -> {CurrentState}", previousState, newState); _connectionEvents.Publish(new ConnectionStateChangedEventArgs(previousState, newState)); } @@ -1369,45 +1202,4 @@ private bool RemoteCertificateValidationCallback(object sender, X509Certificate? return false; } - - internal sealed class EmptyMemoryOwner : IMemoryOwner - { - public static readonly EmptyMemoryOwner Instance = new(); - - private EmptyMemoryOwner() - { - } - - public Memory Memory => Memory.Empty; - - public void Dispose() - { - } - } -} - -internal static class ArrayPoolHelper -{ - public static SlicedMemoryOwner Rent(int minimumLength, bool clearOnReturn = false) - { - return new SlicedMemoryOwner(minimumLength, clearOnReturn); - } - - internal sealed class SlicedMemoryOwner(int minimumLength, bool clearOnReturn = false) : IMemoryOwner - { - private readonly byte[] _value = ArrayPool.Shared.Rent(minimumLength); - private int _disposed; - - public Memory Memory => _value.AsMemory()[..minimumLength]; - - public void Dispose() - { - if (Interlocked.Exchange(ref _disposed, 1) != 0) - { - return; - } - - ArrayPool.Shared.Return(_value, clearOnReturn); - } - } } diff --git a/foreign/csharp/Iggy_SDK/Mappers/BinaryMapper.cs b/foreign/csharp/Iggy_SDK/Mappers/BinaryMapper.cs index f5be657f20..257d401930 100644 --- a/foreign/csharp/Iggy_SDK/Mappers/BinaryMapper.cs +++ b/foreign/csharp/Iggy_SDK/Mappers/BinaryMapper.cs @@ -25,9 +25,9 @@ using Apache.Iggy.Exceptions; using Apache.Iggy.Extensions; using Apache.Iggy.Headers; -using Apache.Iggy.IggyClient.Implementations; using Apache.Iggy.Messages; using Apache.Iggy.Utils; +using Apache.Iggy.Vsr; namespace Apache.Iggy.Mappers; @@ -563,7 +563,7 @@ internal static PolledMessagesRental ToRentedMessages(PolledMessages messages) }); } - return new PolledMessagesRental(TcpMessageStream.EmptyMemoryOwner.Instance) + return new PolledMessagesRental(EmptyMemoryOwner.Instance) { PartitionId = messages.PartitionId, CurrentOffset = messages.CurrentOffset, diff --git a/foreign/csharp/Iggy_SDK/Utils/BufferSizes.cs b/foreign/csharp/Iggy_SDK/Utils/ArrayPoolHelper.cs similarity index 51% rename from foreign/csharp/Iggy_SDK/Utils/BufferSizes.cs rename to foreign/csharp/Iggy_SDK/Utils/ArrayPoolHelper.cs index cd5a9e871f..0db7b1cce4 100644 --- a/foreign/csharp/Iggy_SDK/Utils/BufferSizes.cs +++ b/foreign/csharp/Iggy_SDK/Utils/ArrayPoolHelper.cs @@ -15,9 +15,32 @@ // specific language governing permissions and limitations // under the License. +using System.Buffers; + namespace Apache.Iggy.Utils; -internal static class BufferSizes +internal static class ArrayPoolHelper { - internal const int INITIAL_BYTES_LENGTH = 4; + public static SlicedMemoryOwner Rent(int minimumLength, bool clearOnReturn = false) + { + return new SlicedMemoryOwner(minimumLength, clearOnReturn); + } + + internal sealed class SlicedMemoryOwner(int minimumLength, bool clearOnReturn = false) : IMemoryOwner + { + private readonly byte[] _value = ArrayPool.Shared.Rent(minimumLength); + private int _disposed; + + public Memory Memory => _value.AsMemory()[..minimumLength]; + + public void Dispose() + { + if (Interlocked.Exchange(ref _disposed, 1) != 0) + { + return; + } + + ArrayPool.Shared.Return(_value, clearOnReturn); + } + } } diff --git a/foreign/csharp/Iggy_SDK/Utils/TcpMessageStreamHelpers.cs b/foreign/csharp/Iggy_SDK/Utils/TcpMessageStreamHelpers.cs index 6f0dc135ca..cbf00ccfa4 100644 --- a/foreign/csharp/Iggy_SDK/Utils/TcpMessageStreamHelpers.cs +++ b/foreign/csharp/Iggy_SDK/Utils/TcpMessageStreamHelpers.cs @@ -15,7 +15,6 @@ // specific language governing permissions and limitations // under the License. -using System.Buffers.Binary; using System.Runtime.CompilerServices; using Apache.Iggy.Contracts.Tcp; using Apache.Iggy.Encryption; @@ -26,24 +25,6 @@ namespace Apache.Iggy.Utils; internal static class TcpMessageStreamHelpers { - [MethodImpl(MethodImplOptions.AggressiveInlining)] - internal static void CreatePayload(Span result, Span message, int command) - { - var messageLength = message.Length + 4; - BinaryPrimitives.WriteInt32LittleEndian(result[..4], messageLength); - BinaryPrimitives.WriteInt32LittleEndian(result[4..8], command); - message.CopyTo(result[8..]); - } - - [MethodImpl(MethodImplOptions.AggressiveInlining)] - internal static (int Status, int Length) GetResponseLengthAndStatus(Span buffer) - { - var status = BinaryPrimitives.ReadInt32LittleEndian(buffer[..4]); - var length = BinaryPrimitives.ReadInt32LittleEndian(buffer[4..]); - - return (status, length); - } - internal static int CalculateMessageBytesCount(ReadOnlySpan messages, IMessageEncryptor? encryptor) { var bytesCount = 0; diff --git a/foreign/csharp/Iggy_SDK/Vsr/ConsensusSession.cs b/foreign/csharp/Iggy_SDK/Vsr/ConsensusSession.cs index 8cdbdcc16c..58e1bed51f 100644 --- a/foreign/csharp/Iggy_SDK/Vsr/ConsensusSession.cs +++ b/foreign/csharp/Iggy_SDK/Vsr/ConsensusSession.cs @@ -38,6 +38,7 @@ internal sealed class ConsensusSession private readonly object _gate = new(); #endif private UInt128 _clientId; + private ulong _generation; private bool _registerPending; private ulong _requestCounter; private ulong? _session; @@ -89,6 +90,23 @@ internal bool IsBound } } + /// + /// Generation of the identity, bumped on every re-arm. Anything scoped to a server session - a + /// consumer-group membership, say - compares the generation it was established under to detect that the + /// session it lived on is gone, without inferring it from connection-state edges. Distinct from the + /// fence epoch the wire protocol carries in . + /// + internal ulong Generation + { + get + { + lock (_gate) + { + return _generation; + } + } + } + internal ConsensusSession() : this(GenerateClientId()) { } @@ -205,6 +223,7 @@ private void ReArmLocked() _session = null; _requestCounter = 1; _registerPending = false; + _generation++; } private static UInt128 GenerateClientId() diff --git a/foreign/csharp/Iggy_SDK/ConnectionStream/IConnectionStream.cs b/foreign/csharp/Iggy_SDK/Vsr/ISessionGenerationProvider.cs similarity index 54% rename from foreign/csharp/Iggy_SDK/ConnectionStream/IConnectionStream.cs rename to foreign/csharp/Iggy_SDK/Vsr/ISessionGenerationProvider.cs index 288ba53c35..4b96e72baf 100644 --- a/foreign/csharp/Iggy_SDK/ConnectionStream/IConnectionStream.cs +++ b/foreign/csharp/Iggy_SDK/Vsr/ISessionGenerationProvider.cs @@ -15,12 +15,16 @@ // specific language governing permissions and limitations // under the License. -namespace Apache.Iggy.ConnectionStream; +namespace Apache.Iggy.Vsr; -internal interface IConnectionStream : IDisposable +/// +/// Exposes the consensus session generation of a transport, so session-scoped state - a consumer-group +/// membership - can detect the session it was established under is gone. A client that does not implement +/// it gets edge-based group rejoin from the connection-state events it publishes; a transport that +/// publishes no such events (the built-in HTTP client) gets neither and keeps its membership as-is. +/// +public interface ISessionGenerationProvider { - ValueTask SendAsync(ReadOnlyMemory payload, CancellationToken cancellationToken = default); - ValueTask ReadAsync(Memory buffer, CancellationToken cancellationToken = default); - Task FlushAsync(CancellationToken cancellationToken = default); - void Close(); + /// Generation of the transport's consensus session, bumped on every session re-arm. + ulong SessionGeneration { get; } } diff --git a/foreign/csharp/Iggy_SDK/Vsr/VsrConnection.cs b/foreign/csharp/Iggy_SDK/Vsr/VsrConnection.cs new file mode 100644 index 0000000000..3c6d3c325f --- /dev/null +++ b/foreign/csharp/Iggy_SDK/Vsr/VsrConnection.cs @@ -0,0 +1,372 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +using System.Buffers; +using System.Net.Sockets; +using Apache.Iggy.Exceptions; +using Microsoft.Extensions.Logging; + +namespace Apache.Iggy.Vsr; + +/// +/// One consensus-framed socket: encodes request frames, replays them while the server answers +/// transiently, and reads and decodes reply frames. The caller owns the sending lock that serialises +/// attempts, the leader redirection above this layer, and the reconnect that replaces a dropped +/// connection. +/// +internal sealed class VsrConnection : IDisposable +{ + /// Backoff between replays of a transiently refused request. + private const int TransientRetryIntervalMs = 50; + + private readonly ILogger _logger; + private readonly long _maxResponseFrameSize; + + /// + /// Tears this connection down at the transport level - session reset, state event - when a frame-level + /// failure makes the socket unusable. Runs under the sending lock the caller holds, and it is the + /// transport's job to ignore the call when a reconnect already replaced this connection. + /// + private readonly Action _onDropped; + + /// + /// Bodies up to this size go out as one write: header and body coalesced in + /// . With Nagle disabled every write is its own segment (and its own + /// TLS record under a wrapping stream), so the second write a small request would pay is measurable + /// even against the round trip the lockstep protocol already costs. + /// + private const int CoalescedBodyLimit = 4096 - VsrHeader.HEADER_SIZE; + + private readonly byte[] _replyHeaderBuffer = new byte[VsrHeader.HEADER_SIZE]; + private readonly byte[] _requestFrameBuffer = new byte[VsrHeader.HEADER_SIZE + CoalescedBodyLimit]; + private readonly int _requestTimeoutMs; + + /// The session identity requests on this connection encode from. + private readonly ConsensusSession _session; + + private readonly Stream _stream; + + internal VsrConnection(Stream stream, ConsensusSession session, long maxResponseFrameSize, int requestTimeoutMs, + Action onDropped, ILogger logger) + { + _stream = stream; + _session = session; + _maxResponseFrameSize = maxResponseFrameSize; + _requestTimeoutMs = requestTimeoutMs; + _onDropped = onDropped; + _logger = logger; + } + + public void Dispose() + { + _stream.Dispose(); + } + + internal static bool IsConnectionException(Exception ex) + { + return ex is IggyZeroBytesException or + NotConnectedException or + SocketException or + IOException or + ObjectDisposedException; + } + + /// + /// One attempt on this connection: encode the header, write the frame, and replay it - same session, + /// same request id - while the server answers transiently. The caller must hold the sending lock, + /// which also guards the reuse of the per-connection frame buffer. + /// + internal async ValueTask SendAttemptAsync(int code, ReadOnlyMemory body, + long transientDeadline, long readDeadline, bool clearSensitiveReply, CancellationToken token) + { + var encoded = false; + var requestStarted = false; + + try + { + VsrHeader.EncodeRequestHeader(_requestFrameBuffer, _session, code, body.Span); + + var frameSize = VsrHeader.HEADER_SIZE + body.Length; + var coalesced = body.Length <= CoalescedBodyLimit; + if (coalesced) + { + body.Span.CopyTo(_requestFrameBuffer.AsSpan(VsrHeader.HEADER_SIZE)); + } + + encoded = true; + + while (true) + { + try + { + // Everything that fails without reaching the socket has to fail before this point: past it a + // failure is reported as an outcome the server alone knows, which for a replicated write + // tells the caller its request may have committed twice. + token.ThrowIfCancellationRequested(); + requestStarted = true; + + if (coalesced) + { + await _stream.WriteAsync(_requestFrameBuffer.AsMemory(0, frameSize), token); + } + else + { + await _stream.WriteAsync(_requestFrameBuffer.AsMemory(0, VsrHeader.HEADER_SIZE), token); + await _stream.WriteAsync(body, token); + } + + await _stream.FlushAsync(token); + + IMemoryOwner response = await ReadReplyAsync(readDeadline, clearSensitiveReply, token); + + return VsrAttempt.Ok(response, this); + } + catch (IggyInvalidStatusCodeException e) when (IsReplayableTransient(e, transientDeadline, + readDeadline)) + { + var governingDeadline = e.StatusCode == VsrError.TRANSIENT_NOT_COMMITTED + ? readDeadline + : transientDeadline; + var remaining = governingDeadline - Environment.TickCount64; + await Task.Delay((int)Math.Clamp(remaining, 0, TransientRetryIntervalMs), token); + } + catch (Exception e) when (IsConnectionException(e)) + { + _onDropped(this); + + return VsrAttempt.Failed(encoded, e, requestStarted, this); + } + catch (OperationCanceledException e) + { + _onDropped(this); + + return VsrAttempt.Failed(encoded, e, requestStarted, this); + } + catch (Exception e) + { + return VsrAttempt.Failed(encoded, e, requestStarted, this); + } + } + } + catch (OperationCanceledException e) + { + if (encoded) + { + _onDropped(this); + } + + return VsrAttempt.Failed(encoded, e, requestStarted, this); + } + catch (Exception e) + { + return VsrAttempt.Failed(encoded, e, requestStarted, this); + } + } + + private async Task> ReadReplyAsync(long readDeadline, bool clearSensitiveReply, + CancellationToken token) + { + var remaining = readDeadline - Environment.TickCount64; + if (remaining <= 0) + { + throw new IOException($"Timed out after {_requestTimeoutMs} ms waiting for a consensus reply."); + } + + // One timer for the whole reply: the deadline covers the frame, not each partial read, so a per-read + // source would both re-arm the budget and allocate a timer per socket read. + using var readCancellation = CancellationTokenSource.CreateLinkedTokenSource(token); + readCancellation.CancelAfter((int)Math.Min(remaining, _requestTimeoutMs)); + + await ReadExactAsync(_replyHeaderBuffer, readCancellation.Token, token); + + var command = VsrHeader.PeekCommand(_replyHeaderBuffer); + if (command == Command2.Eviction) + { + var eviction = VsrHeader.ReadEviction(_replyHeaderBuffer); + _logger.LogWarning("Consensus session evicted by the server: {Reason}", eviction.Reason); + _onDropped(this); + + throw new VsrSessionEvictedException(VsrReplyDecoder.ToException(eviction)); + } + + if (command != Command2.Reply) + { + // Neither a reply nor an eviction: this frame was never an answer to the outstanding request, so + // whatever the peer does send for it would be read as the next request's reply and handed to the + // wrong caller. The size field of a frame the client cannot model is no basis for resynchronising. + _onDropped(this); + + throw VsrError.Exception(VsrError.INVALID_COMMAND, + $"Unexpected consensus frame {command} on a client connection."); + } + + int bodySize; + try + { + bodySize = VsrReplyDecoder.ReadBodySize(_replyHeaderBuffer); + if (VsrHeader.HEADER_SIZE + (long)bodySize > _maxResponseFrameSize) + { + throw VsrError.Exception(VsrError.INVALID_COMMAND, + $"Reply frame of {VsrHeader.HEADER_SIZE + bodySize} bytes exceeds the configured maximum of " + + $"{_maxResponseFrameSize} bytes."); + } + } + catch + { + // An announced size the client refuses to read - undersized, oversized - leaves the body on the + // wire, so the stream no longer sits on a frame boundary and the next reply would decode body bytes + // as a header. + _onDropped(this); + + throw; + } + + if (bodySize == 0) + { + VsrReplyDecoder.Decode(_replyHeaderBuffer, ReadOnlyMemory.Empty); + + return EmptyMemoryOwner.Instance; + } + + var buffer = ArrayPool.Shared.Rent(bodySize); + try + { + await ReadExactAsync(buffer.AsMemory(0, bodySize), readCancellation.Token, token); + ReadOnlyMemory decoded = VsrReplyDecoder.Decode(_replyHeaderBuffer, buffer.AsMemory(0, bodySize)); + if (decoded.IsEmpty) + { + ArrayPool.Shared.Return(buffer, clearSensitiveReply); + + return EmptyMemoryOwner.Instance; + } + + // The decoded payload is always a suffix of the body - the funnel only strips the leading + // committed result section. + return new PooledMemoryOwner(buffer, bodySize - decoded.Length, decoded.Length, clearSensitiveReply); + } + catch + { + ArrayPool.Shared.Return(buffer, clearSensitiveReply); + throw; + } + } + + private async ValueTask ReadExactAsync(Memory buffer, CancellationToken readToken, CancellationToken token) + { + var totalRead = 0; + while (totalRead < buffer.Length) + { + int readBytes; + try + { + readBytes = await _stream.ReadAsync(buffer[totalRead..], readToken); + } + catch (OperationCanceledException) when (!token.IsCancellationRequested) + { + throw new IOException($"Timed out after {_requestTimeoutMs} ms waiting for a consensus reply."); + } + + if (readBytes == 0) + { + throw new IggyZeroBytesException(); + } + + totalRead += readBytes; + } + } + + private static bool IsReplayableTransient(IggyInvalidStatusCodeException error, long transientDeadline, + long readDeadline) + { + if (!error.FromServer) + { + return false; + } + + return error.StatusCode switch + { + VsrError.TRANSIENT_NOT_COMMITTED => Environment.TickCount64 < readDeadline, + VsrError.TRANSIENT_NOT_ACCEPTED => Environment.TickCount64 < transientDeadline, + _ => false + }; + } +} + +/// Outcome of one call. +/// Whether the header was encoded, i.e. whether a request id may have been consumed. +/// The decoded reply payload, non-null exactly when is null. +/// The failure that ended the attempt, or null on success. +/// +/// Whether any byte of the frame was written, which makes the server-side outcome unknowable on failure. +/// +/// +/// The connection the attempt ran on, so a caller that drops it after releasing the sending lock can tell +/// its own connection from a replacement a reconnect installed since. +/// +internal readonly record struct VsrAttempt( + bool Encoded, + IMemoryOwner? Response, + Exception? Error, + bool RequestStarted, + VsrConnection? Connection) +{ + public static VsrAttempt Ok(IMemoryOwner response, VsrConnection connection) + { + return new VsrAttempt(true, response, null, true, connection); + } + + public static VsrAttempt Failed(bool encoded, Exception error, bool requestStarted, VsrConnection? connection) + { + return new VsrAttempt(encoded, null, error, requestStarted, connection); + } +} + +/// Shared empty reply payload; disposes to nothing, so it is safe to hand out repeatedly. +internal sealed class EmptyMemoryOwner : IMemoryOwner +{ + public static readonly EmptyMemoryOwner Instance = new(); + + private EmptyMemoryOwner() + { + } + + public Memory Memory => Memory.Empty; + + public void Dispose() + { + } +} + +/// +/// Owns a pooled buffer while exposing only the decoded payload slice inside it. A buffer that carried a +/// credential is zeroed on the way back to the pool so the secret cannot resurface in a later rental. +/// +internal sealed class PooledMemoryOwner(byte[] buffer, int start, int length, bool clearOnReturn = false) + : IMemoryOwner +{ + private int _disposed; + + public Memory Memory => buffer.AsMemory(start, length); + + public void Dispose() + { + if (Interlocked.Exchange(ref _disposed, 1) == 0) + { + ArrayPool.Shared.Return(buffer, clearOnReturn); + } + } +} diff --git a/foreign/csharp/Iggy_SDK_Tests/ConsumerTests/ConsumerSessionGenerationTests.cs b/foreign/csharp/Iggy_SDK_Tests/ConsumerTests/ConsumerSessionGenerationTests.cs new file mode 100644 index 0000000000..00fbd147f4 --- /dev/null +++ b/foreign/csharp/Iggy_SDK_Tests/ConsumerTests/ConsumerSessionGenerationTests.cs @@ -0,0 +1,245 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +using Apache.Iggy.Consumers; +using Apache.Iggy.Contracts; +using Apache.Iggy.Enums; +using Apache.Iggy.IggyClient; +using Apache.Iggy.Kinds; +using Apache.Iggy.Messages; +using Apache.Iggy.Vsr; +using Microsoft.Extensions.Logging.Abstractions; +using Moq; + +namespace Apache.Iggy.Tests.ConsumerTests; + +/// +/// Group membership on a transport that exposes a consensus session is keyed off the session generation +/// rather than off connection-state edges, so these cover both arms of that branch. +/// +public sealed class ConsumerSessionGenerationTests +{ + [Fact] + public async Task + given_group_consumer_when_reauthenticated_on_the_same_session_generation_should_not_rejoin_the_group() + { + var client = new GenerationClient(); + var consumer = new IggyConsumer(client.Object, BuildGroupConfig(), NullLoggerFactory.Instance); + await consumer.InitAsync(TestContext.Current.CancellationToken); + Assert.Equal(1, client.JoinCount); + + await client.RaiseAsync(ConnectionState.Connected, ConnectionState.Authenticated); + + Assert.Equal(1, client.JoinCount); + await consumer.DisposeAsync(); + } + + [Fact] + public async Task given_group_consumer_when_the_session_generation_moved_should_rejoin_the_group() + { + var client = new GenerationClient(); + var consumer = new IggyConsumer(client.Object, BuildGroupConfig(), NullLoggerFactory.Instance); + await consumer.InitAsync(TestContext.Current.CancellationToken); + Assert.Equal(1, client.JoinCount); + + // The session the membership was established under is gone. + client.SessionGeneration++; + await client.RaiseAsync(ConnectionState.Connected, ConnectionState.Authenticated); + + Assert.Equal(2, client.JoinCount); + + // The rejoin must re-stamp the membership generation: without it every later event would rejoin + // again, triggering a group-wide rebalance per reconnect. + await client.RaiseAsync(ConnectionState.Connected, ConnectionState.Authenticated); + + Assert.Equal(2, client.JoinCount); + await consumer.DisposeAsync(); + } + + /// + /// A disconnect re-arms the transport's session, so the generation moves and the reconnect rejoins. The + /// membership is not cleared on the Disconnected event itself: the poll gate compares generations + /// instead, which survives state events arriving out of order. + /// + [Fact] + public async Task given_group_consumer_when_disconnected_should_surrender_membership_and_rejoin_on_reconnect() + { + var client = new GenerationClient(); + var consumer = new IggyConsumer(client.Object, BuildGroupConfig(), NullLoggerFactory.Instance); + await consumer.InitAsync(TestContext.Current.CancellationToken); + Assert.Equal(1, client.JoinCount); + + await client.RaiseAsync(ConnectionState.Authenticated, ConnectionState.Disconnected); + // The transport re-arms the consensus session when the connection drops. + client.SessionGeneration++; + await client.RaiseAsync(ConnectionState.Connecting, ConnectionState.Authenticated); + + Assert.Equal(2, client.JoinCount); + await consumer.DisposeAsync(); + } + + /// + /// The event-time rejoin swallows its failures and the transport never republishes an unchanged state, + /// so the poll loop is the only place left that can restore a membership the event handler failed to. + /// + [Fact] + public async Task given_group_consumer_when_the_event_time_rejoin_fails_should_rejoin_from_the_poll_loop() + { + var client = new GenerationClient(); + var consumer = new IggyConsumer(client.Object, BuildGroupConfig(), NullLoggerFactory.Instance); + await consumer.InitAsync(TestContext.Current.CancellationToken); + Assert.Equal(1, client.JoinCount); + + client.SessionGeneration++; + client.FailNextJoin = true; + await client.RaiseAsync(ConnectionState.Connected, ConnectionState.Authenticated); + Assert.Equal(1, client.JoinCount); + + var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5)); + await using IAsyncEnumerator messages = consumer.ReceiveAsync(cts.Token) + .GetAsyncEnumerator(TestContext.Current.CancellationToken); + + Assert.True(await messages.MoveNextAsync()); + Assert.Equal(2, client.JoinCount); + await consumer.DisposeAsync(); + } + + /// + /// A plain consumer holds no membership, so nothing may clear its joined flag: no code path would ever + /// set it again and every later poll would be skipped. + /// + [Fact] + public async Task given_plain_consumer_when_disconnected_should_keep_polling_after_reconnect() + { + var client = new GenerationClient(); + var config = BuildGroupConfig(); + config.Consumer = Consumer.New(1); + var consumer = new IggyConsumer(client.Object, config, NullLoggerFactory.Instance); + await consumer.InitAsync(TestContext.Current.CancellationToken); + Assert.Equal(0, client.JoinCount); + + await client.RaiseAsync(ConnectionState.Authenticated, ConnectionState.Disconnected); + client.SessionGeneration++; + await client.RaiseAsync(ConnectionState.Connecting, ConnectionState.Authenticated); + + var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5)); + await using IAsyncEnumerator messages = consumer.ReceiveAsync(cts.Token) + .GetAsyncEnumerator(TestContext.Current.CancellationToken); + + Assert.True(await messages.MoveNextAsync()); + Assert.Equal(0, client.JoinCount); + await consumer.DisposeAsync(); + } + + private static IggyConsumerConfig BuildGroupConfig() + { + return new IggyConsumerConfig + { + StreamId = Identifier.Numeric(1), + TopicId = Identifier.Numeric(1), + Consumer = Consumer.Group("group-1"), + PollingStrategy = PollingStrategy.Next(), + BatchSize = 10, + AutoCommitMode = AutoCommitMode.Disabled, + AutoCommit = false, + PollingIntervalMs = 0 + }; + } + + /// + /// A client that carries a consensus session, counts group joins, and replays connection-state events on + /// demand so a test can drive the reconnection handler without a socket. + /// + private sealed class GenerationClient + { + private readonly List> _subscribers = []; + + public IIggyClient Object { get; } + + public ulong SessionGeneration { get; set; } + + public int JoinCount { get; private set; } + + /// The next join attempt throws instead of counting, then the flag disarms itself. + public bool FailNextJoin { get; set; } + + public GenerationClient() + { + var mock = new Mock(MockBehavior.Loose); + mock.As().SetupGet(c => c.SessionGeneration) + .Returns(() => SessionGeneration); + mock.Setup(c => c.ConnectAsync(It.IsAny())).Returns(Task.CompletedTask); + mock.Setup(c => c.SubscribeConnectionEvents(It.IsAny>())) + .Callback>(_subscribers.Add); + mock.Setup(c => c.UnsubscribeConnectionEvents(It.IsAny>())) + .Callback>(callback => _subscribers.Remove(callback)); + mock.Setup(c => c.GetConsumerGroupByIdAsync(It.IsAny(), It.IsAny(), + It.IsAny(), It.IsAny())) + .ReturnsAsync(new ConsumerGroupResponse + { + Id = 1, + Name = "group-1", + MembersCount = 1, + PartitionsCount = 1 + }); + mock.Setup(c => c.JoinConsumerGroupAsync(It.IsAny(), It.IsAny(), + It.IsAny(), It.IsAny())) + .Callback(() => + { + if (FailNextJoin) + { + FailNextJoin = false; + throw new InvalidOperationException("Injected join failure."); + } + + JoinCount++; + }) + .Returns(Task.CompletedTask); + mock.Setup(c => c.PollMessagesAsync(It.IsAny(), It.IsAny(), It.IsAny(), + It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny(), + It.IsAny())) + .ReturnsAsync(() => new PolledMessages + { + PartitionId = 1, + CurrentOffset = 0, + Messages = + [ + new MessageResponse + { + Header = new MessageHeader + { + Offset = 0, + PayloadLength = 1 + }, + Payload = new byte[] { 1 }, + UserHeaders = null + } + ] + }); + + Object = mock.Object; + } + + public async Task RaiseAsync(ConnectionState previousState, ConnectionState currentState) + { + foreach (Func subscriber in _subscribers.ToArray()) + { + await subscriber(new ConnectionStateChangedEventArgs(previousState, currentState)); + } + } + } +} diff --git a/foreign/csharp/Iggy_SDK_Tests/ConsumerTests/RentedConsumerTests.cs b/foreign/csharp/Iggy_SDK_Tests/ConsumerTests/RentedConsumerTests.cs index 7d65beed3a..f4adb538d1 100644 --- a/foreign/csharp/Iggy_SDK_Tests/ConsumerTests/RentedConsumerTests.cs +++ b/foreign/csharp/Iggy_SDK_Tests/ConsumerTests/RentedConsumerTests.cs @@ -22,9 +22,9 @@ using Apache.Iggy.Encryption; using Apache.Iggy.Exceptions; using Apache.Iggy.IggyClient; -using Apache.Iggy.IggyClient.Implementations; using Apache.Iggy.Kinds; using Apache.Iggy.Messages; +using Apache.Iggy.Vsr; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging.Abstractions; using Moq; @@ -377,7 +377,7 @@ internal static Mock BuildClientMock(Queue re return rentals.Dequeue(); } - return new PolledMessagesRental(TcpMessageStream.EmptyMemoryOwner.Instance) + return new PolledMessagesRental(EmptyMemoryOwner.Instance) { PartitionId = 1, CurrentOffset = 0, diff --git a/foreign/csharp/Iggy_SDK_Tests/MapperTests/BinaryMapper.cs b/foreign/csharp/Iggy_SDK_Tests/MapperTests/BinaryMapper.cs index e37aa35750..e0054919a3 100644 --- a/foreign/csharp/Iggy_SDK_Tests/MapperTests/BinaryMapper.cs +++ b/foreign/csharp/Iggy_SDK_Tests/MapperTests/BinaryMapper.cs @@ -23,13 +23,13 @@ using Apache.Iggy.Enums; using Apache.Iggy.Exceptions; using Apache.Iggy.Extensions; -using Apache.Iggy.IggyClient.Implementations; using Apache.Iggy.Shared; using Apache.Iggy.Tests.Utils; using Apache.Iggy.Tests.Utils.Groups; using Apache.Iggy.Tests.Utils.Messages; using Apache.Iggy.Tests.Utils.Stats; using Apache.Iggy.Tests.Utils.Topics; +using Apache.Iggy.Vsr; using StreamFactory = Apache.Iggy.Tests.Utils.Streams.StreamFactory; namespace Apache.Iggy.Tests.MapperTests; @@ -91,7 +91,7 @@ public void MapMessages_NoHeaders_ReturnsValidMessageResponses() // Act var responses - = Mappers.BinaryMapper.MapRentedMessages(combinedPayload, TcpMessageStream.EmptyMemoryOwner.Instance); + = Mappers.BinaryMapper.MapRentedMessages(combinedPayload, EmptyMemoryOwner.Instance); // Assert Assert.NotNull(responses); @@ -366,7 +366,7 @@ public void MapRentedMessages_WithEncryptor_DecryptsPayloadsAndHeadersIntoPooled frame1.CopyTo(combined.AsSpan(16)); frame2.CopyTo(combined.AsSpan(16 + frame1.Length)); - using var rental = Mappers.BinaryMapper.MapRentedMessages(combined, TcpMessageStream.EmptyMemoryOwner.Instance, + using var rental = Mappers.BinaryMapper.MapRentedMessages(combined, EmptyMemoryOwner.Instance, encryptor); Assert.Equal(7, rental.PartitionId); @@ -407,7 +407,7 @@ public void MapRentedMessages_WithEncryptor_NegativePayloadLength_ThrowsInsteadO frame.CopyTo(combined.AsSpan(16)); Assert.Throws(() => - Mappers.BinaryMapper.MapRentedMessages(combined, TcpMessageStream.EmptyMemoryOwner.Instance, encryptor)); + Mappers.BinaryMapper.MapRentedMessages(combined, EmptyMemoryOwner.Instance, encryptor)); } [Fact] @@ -425,7 +425,7 @@ public void MapRentedMessages_WithEncryptor_TamperedCiphertext_ThrowsMessageDecr frame.CopyTo(combined.AsSpan(16)); var ex = Assert.Throws(() => - Mappers.BinaryMapper.MapRentedMessages(combined, TcpMessageStream.EmptyMemoryOwner.Instance, encryptor)); + Mappers.BinaryMapper.MapRentedMessages(combined, EmptyMemoryOwner.Instance, encryptor)); Assert.Equal(42ul, ex.Offset); Assert.Equal(7u, ex.PartitionId); diff --git a/foreign/csharp/Iggy_SDK_Tests/UtilityTests/SlicedMemoryOwnerTests.cs b/foreign/csharp/Iggy_SDK_Tests/UtilityTests/SlicedMemoryOwnerTests.cs deleted file mode 100644 index 73e9ba3b9c..0000000000 --- a/foreign/csharp/Iggy_SDK_Tests/UtilityTests/SlicedMemoryOwnerTests.cs +++ /dev/null @@ -1,85 +0,0 @@ -// Licensed to the Apache Software Foundation (ASF) under one -// or more contributor license agreements. See the NOTICE file -// distributed with this work for additional information -// regarding copyright ownership. The ASF licenses this file -// to you under the Apache License, Version 2.0 (the -// "License"); you may not use this file except in compliance -// with the License. You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, -// software distributed under the License is distributed on an -// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY -// KIND, either express or implied. See the License for the -// specific language governing permissions and limitations -// under the License. - -using System.Buffers; -using System.Reflection; -using System.Runtime.CompilerServices; -using System.Runtime.InteropServices; -using Apache.Iggy.IggyClient.Implementations; - -namespace Apache.Iggy.Tests.UtilityTests; - -public class SlicedMemoryOwnerTests -{ - private const int BufferSize = 4096; - - [Fact] - public void Dispose_ReturnsBufferToPool() - { - var owner = ArrayPoolHelper.Rent(BufferSize); - byte[] underlying = GetUnderlyingArray(owner); - owner.Dispose(); - - using var second = ArrayPoolHelper.Rent(BufferSize); - Assert.Same(underlying, GetUnderlyingArray(second)); - } - - [Fact] - public void Dispose_IsIdempotent() - { - var owner = ArrayPoolHelper.Rent(BufferSize); - owner.Dispose(); - owner.Dispose(); - owner.Dispose(); - } - - [Fact] - public void FinalizerIsDeclared() - { - Type sliced = typeof(ArrayPoolHelper) - .GetNestedType("SlicedMemoryOwner", BindingFlags.NonPublic)!; - - MethodInfo? finalizer = sliced.GetMethod("Finalize", BindingFlags.NonPublic | BindingFlags.Instance); - - Assert.NotNull(finalizer); - } - - [Fact] - public void ForgotDispose_FinalizerRunsAndReclaimsInstance() - { - WeakReference weakRef = RentWeak(); - - GC.Collect(); - GC.WaitForPendingFinalizers(); - GC.Collect(); - - Assert.False(weakRef.IsAlive); - - [MethodImpl(MethodImplOptions.NoInlining)] - static WeakReference RentWeak() => new(ArrayPoolHelper.Rent(BufferSize)); - } - - private static byte[] GetUnderlyingArray(IMemoryOwner owner) - { - if (!MemoryMarshal.TryGetArray(owner.Memory, out var segment) || segment.Array is null) - { - throw new InvalidOperationException("SlicedMemoryOwner.Memory must be array-backed."); - } - - return segment.Array; - } -} diff --git a/foreign/csharp/Iggy_SDK_Tests/Utils/Errors/ErrorModelFactory.cs b/foreign/csharp/Iggy_SDK_Tests/Utils/Errors/ErrorModelFactory.cs deleted file mode 100644 index 137a2bb2d2..0000000000 --- a/foreign/csharp/Iggy_SDK_Tests/Utils/Errors/ErrorModelFactory.cs +++ /dev/null @@ -1,33 +0,0 @@ -// Licensed to the Apache Software Foundation (ASF) under one -// or more contributor license agreements. See the NOTICE file -// distributed with this work for additional information -// regarding copyright ownership. The ASF licenses this file -// to you under the Apache License, Version 2.0 (the -// "License"); you may not use this file except in compliance -// with the License. You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, -// software distributed under the License is distributed on an -// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY -// KIND, either express or implied. See the License for the -// specific language governing permissions and limitations -// under the License. - -using Apache.Iggy.Errors; - -namespace Apache.Iggy.Tests.Utils.Errors; - -public static class ErrorModelFactory -{ - public static ErrorModel CreateErrorModelBadRequest() - { - return new ErrorModel(69, "bad_request", "Bad Request"); - } - - public static ErrorModel CreateErrorModelNotFound() - { - return new ErrorModel(69, "not_found", "Not Found"); - } -} diff --git a/foreign/csharp/Iggy_SDK_Tests/VsrTests/ConsensusSessionTests.cs b/foreign/csharp/Iggy_SDK_Tests/VsrTests/ConsensusSessionTests.cs index 51e4d7f0a1..3ff5256c85 100644 --- a/foreign/csharp/Iggy_SDK_Tests/VsrTests/ConsensusSessionTests.cs +++ b/foreign/csharp/Iggy_SDK_Tests/VsrTests/ConsensusSessionTests.cs @@ -215,4 +215,24 @@ public void Reset_ClearsBindingAndCounter() Assert.Equal(1UL, session.RequestCounter); Assert.NotEqual((UInt128)1, session.ClientId); } + + [Fact] + public void Generation_AdvancesOnEveryReArmAndNotOnBind() + { + var session = new ConsensusSession(1); + var initialGeneration = session.Generation; + + session.Resolve(VsrOperation.Register); + session.Bind(10); + Assert.Equal(initialGeneration, session.Generation); + + session.Reset(); + Assert.Equal(initialGeneration + 1, session.Generation); + + // A register on a previously bound session re-arms the identity, which is a new generation too. + session.Resolve(VsrOperation.Register); + session.Bind(11); + session.Resolve(VsrOperation.Register); + Assert.Equal(initialGeneration + 2, session.Generation); + } }