You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by lz...@apache.org on 2026/08/10 19:52:58 UTC
(iggy) 01/01: refactor(csharp): cleanup tcp connection after vsr implementation
This is an automated email from the ASF dual-hosted git repository.
lukaszzborek pushed a commit to branch dotnet-vsr-cleanup
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 95289da7c313a08278f4d6f0c207e3aaa505c053
Author: Łukasz Zborek <lu...@gmail.com>
AuthorDate: Mon Aug 10 20:49:18 2026 +0200
refactor(csharp): cleanup tcp connection after vsr implementation
---
.../ConnectionStream/TcpConnectionStream.cs | 53 ---
foreign/csharp/Iggy_SDK/Consumers/IggyConsumer.cs | 46 ++
.../Implementations/TcpMessageStream.Vsr.cs | 420 +++---------------
.../IggyClient/Implementations/TcpMessageStream.cs | 477 +++++++--------------
foreign/csharp/Iggy_SDK/Iggy_SDK.csproj | 12 +-
foreign/csharp/Iggy_SDK/Mappers/BinaryMapper.cs | 4 +-
.../Utils/{BufferSizes.cs => ArrayPoolHelper.cs} | 27 +-
.../Iggy_SDK/Utils/TcpMessageStreamHelpers.cs | 19 -
foreign/csharp/Iggy_SDK/Vsr/ConsensusSession.cs | 18 +
.../ISessionEpochProvider.cs} | 14 +-
foreign/csharp/Iggy_SDK/Vsr/VsrConnection.cs | 355 +++++++++++++++
foreign/csharp/Iggy_SDK/Vsr/VsrHeader.cs | 12 +-
foreign/csharp/Iggy_SDK/Vsr/VsrNamespace.cs | 51 ++-
.../ConsumerTests/ConsumerSessionEpochTests.cs | 197 +++++++++
.../ConsumerTests/RentedConsumerTests.cs | 4 +-
.../Iggy_SDK_Tests/MapperTests/BinaryMapper.cs | 10 +-
.../UtilityTests/SlicedMemoryOwnerTests.cs | 17 +-
.../Utils/Errors/ErrorModelFactory.cs | 33 --
.../VsrTests/ConsensusSessionTests.cs | 20 +
.../Iggy_SDK_Tests/VsrTests/VsrHeaderTests.cs | 52 +--
.../Iggy_SDK_Tests/VsrTests/VsrNamespaceTests.cs | 53 ++-
21 files changed, 1047 insertions(+), 847 deletions(-)
diff --git a/foreign/csharp/Iggy_SDK/ConnectionStream/TcpConnectionStream.cs b/foreign/csharp/Iggy_SDK/ConnectionStream/TcpConnectionStream.cs
deleted file mode 100644
index ce016a70f..000000000
--- 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<byte> payload, CancellationToken cancellationToken = default)
- {
- return _stream.WriteAsync(payload, cancellationToken);
- }
-
- public ValueTask<int> ReadAsync(Memory<byte> 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.cs b/foreign/csharp/Iggy_SDK/Consumers/IggyConsumer.cs
index b5317cfde..f40cf6227 100644
--- a/foreign/csharp/Iggy_SDK/Consumers/IggyConsumer.cs
+++ b/foreign/csharp/Iggy_SDK/Consumers/IggyConsumer.cs
@@ -23,6 +23,7 @@ using Apache.Iggy.Exceptions;
using Apache.Iggy.IggyClient;
using Apache.Iggy.Kinds;
using Apache.Iggy.Utils;
+using Apache.Iggy.Vsr;
using Microsoft.Extensions.Logging;
namespace Apache.Iggy.Consumers;
@@ -46,6 +47,13 @@ public partial class IggyConsumer : IAsyncDisposable
private int _disposeState;
private volatile bool _isInitialized;
private volatile bool _joinedConsumerGroup;
+
+ /// <summary>
+ /// Consensus session epoch the group membership was established under. A later epoch means the server
+ /// session that held the membership is gone and the group must be rejoined.
+ /// </summary>
+ private ulong _joinedSessionEpoch;
+
private long _lastPolledAtMs;
/// <summary>Whether this consumer has been initialized via <see cref="InitAsync" />.</summary>
@@ -263,8 +271,13 @@ public partial class IggyConsumer : IAsyncDisposable
return;
}
+ // Captured before the join: a session reset racing the join lands the membership on a later epoch, and
+ // the mismatch triggers one redundant (idempotent) rejoin instead of a missed one.
+ var sessionEpoch = (_client as ISessionEpochProvider)?.SessionEpoch ?? 0;
+
if (_config.Consumer.Type == ConsumerType.Consumer)
{
+ _joinedSessionEpoch = sessionEpoch;
_joinedConsumerGroup = true;
return;
}
@@ -306,6 +319,7 @@ public partial class IggyConsumer : IAsyncDisposable
await _client.JoinConsumerGroupAsync(_config.StreamId, _config.TopicId,
Identifier.String(_consumerGroupName), ct);
+ _joinedSessionEpoch = sessionEpoch;
_joinedConsumerGroup = true;
LogConsumerGroupJoined(_consumerGroupName);
}
@@ -505,6 +519,38 @@ public partial class IggyConsumer : IAsyncDisposable
await _connectionStateSemaphore.WaitAsync();
try
{
+ if (_client is ISessionEpochProvider epochProvider)
+ {
+ // A plain consumer holds no membership, so its flag must never be cleared: nothing would ever
+ // set it again and every later poll would be skipped.
+ if (_config.Consumer.Type == ConsumerType.Consumer)
+ {
+ return;
+ }
+
+ if (e.CurrentState != ConnectionState.Authenticated)
+ {
+ // Polling under a dropped session would be refused by the server for as long as the
+ // reconnect takes to re-authenticate, so the membership is surrendered up front.
+ if (e.CurrentState == ConnectionState.Disconnected)
+ {
+ _joinedConsumerGroup = false;
+ }
+
+ return;
+ }
+
+ if (_joinedConsumerGroup && epochProvider.SessionEpoch == _joinedSessionEpoch)
+ {
+ 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 81898c099..7582bfaf7 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 <see cref="TcpMessageStream" />.
/// </summary>
-public sealed partial class TcpMessageStream
+public sealed partial class TcpMessageStream : ISessionEpochProvider
{
/// <summary>
/// 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
/// </summary>
private const int VsrRequestTimeoutMs = 30_000;
- /// <summary>Backoff between replays of a transiently refused request.</summary>
- private const int VsrTransientRetryIntervalMs = 50;
-
- /// <summary>
- /// 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.
- /// </summary>
- private const int VsrContiguousFrameLimit = 4 * 1024;
-
/// <summary>
/// How long a <see cref="VsrError.TRANSIENT_NOT_ACCEPTED" /> 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
/// <summary>
/// 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.
/// </summary>
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;
+ /// <inheritdoc />
+ ulong ISessionEpochProvider.SessionEpoch => _consensusSession.Epoch;
/// <summary>
/// Runs the consensus register handshake and binds the session it commits. Everything before the bind
@@ -138,16 +124,13 @@ public sealed partial class TcpMessageStream
SetConnectionStateAsync(ConnectionState.Connected);
}
- var payload = new byte[4 + BufferSizes.INITIAL_BYTES_LENGTH + message.Length];
- TcpMessageStreamHelpers.CreatePayload(payload, message, code);
-
SetConnectionStateAsync(ConnectionState.Authenticating);
LoginRegisterResponse response;
try
{
- Interlocked.Exchange(ref _skipAutoLoginOnce, 1);
- using IMemoryOwner<byte> responseBuffer = await SendWithResponseAsync(payload, token);
+ using IMemoryOwner<byte> responseBuffer =
+ await SendWithResponseAsync(code, 0, message, token, false);
response = LoginRegister.Deserialize(responseBuffer.Memory.Span);
_consensusSession.Bind(response.Session);
@@ -162,10 +145,6 @@ public sealed partial class TcpMessageStream
throw;
}
- finally
- {
- Interlocked.Exchange(ref _skipAutoLoginOnce, 0);
- }
_logger.LogInformation(
"Authenticated against the server, version {ServerVersion}, protocol version {ServerProtocolVersion}",
@@ -328,10 +307,8 @@ public sealed partial class TcpMessageStream
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<byte> responseBuffer = await SendWithResponseAsync(payload, token);
+ using IMemoryOwner<byte> responseBuffer =
+ await SendWithResponseAsync(CommandCodes.SYNC_CONSUMER_GROUP_CODE, message, token);
var key = new GroupKey(streamId, topicId, groupId);
if (responseBuffer.Memory.Length == 0)
@@ -373,9 +350,9 @@ public sealed partial class TcpMessageStream
/// <summary>
/// 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
+ /// <see cref="VsrMaxLeaderRedirects" />.
/// </summary>
private async Task<bool> RedirectAsync(CancellationToken token)
{
@@ -384,20 +361,12 @@ public sealed partial class TcpMessageStream
var currentLeaderNode = await GetCurrentLeaderNodeAsync(token);
if (currentLeaderNode == null)
{
- Interlocked.Exchange(ref _leaderRedirectCount, 0);
return false;
}
var leaderAddress = ServerAddress.HostPort(currentLeaderNode.Ip, currentLeaderNode.Endpoints.Tcp);
if (ServerAddress.IsSame(leaderAddress, _currentAddress))
{
- Interlocked.Exchange(ref _leaderRedirectCount, 0);
- return false;
- }
-
- if (Interlocked.Increment(ref _leaderRedirectCount) > VsrMaxLeaderRedirects)
- {
- _logger.LogWarning("Maximum leader redirections reached, continuing on {Address}", _currentAddress);
return false;
}
@@ -410,7 +379,7 @@ public sealed partial class TcpMessageStream
try
{
_currentAddress = leaderAddress;
- DropVsrConnectionLocked(_stream);
+ DropVsrConnectionLocked(_connection);
}
finally
{
@@ -423,12 +392,11 @@ public sealed partial class TcpMessageStream
private async Task<ClusterNode?> 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;
@@ -471,32 +439,52 @@ public sealed partial class TcpMessageStream
return null;
}
- finally
+ }
+
+ /// <summary>
+ /// 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.
+ /// </summary>
+ private async Task<ClusterMetadata?> ReadClusterMetadataNoRedirectAsync(CancellationToken token)
+ {
+ using IMemoryOwner<byte> responseBuffer = await SendRawAsync(CommandCodes.GET_CLUSTER_METADATA_CODE, 0,
+ ReadOnlyMemory<byte>.Empty, token, false);
+
+ if (responseBuffer.Memory.Length == 0)
{
- Interlocked.Decrement(ref _leaderProbeDepth);
+ return null;
}
+
+ return BinaryMapper.MapClusterMetadata(responseBuffer.Memory.Span);
}
/// <summary>
- /// Sends a consensus-framed request. The call sites still build the classic
- /// <c>[size u32][code u32][body]</c> 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: the body is written right after the 256-byte consensus header -
+ /// two writes, no concatenation.
/// </summary>
/// <remarks>
/// 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.
/// </remarks>
- private async Task<IMemoryOwner<byte>> SendRawVsrAsync(ReadOnlyMemory<byte> payload, CancellationToken token)
+ private async Task<IMemoryOwner<byte>> SendRawAsync(int code, ulong ns, ReadOnlyMemory<byte> body,
+ CancellationToken token, bool allowRedirect = true)
{
- var code = (int)BinaryPrimitives.ReadUInt32LittleEndian(payload.Span.Slice(4, 4));
- ReadOnlyMemory<byte> 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 overallDeadline = Environment.TickCount64 + VsrRequestTimeoutMs;
var headerBuffer = ArrayPool<byte>.Shared.Rent(VsrHeader.HEADER_SIZE);
Memory<byte> header = headerBuffer.AsMemory(0, VsrHeader.HEADER_SIZE);
var requestEncoded = false;
- TcpConnectionStream? lastStream = null;
+ var redirects = 0;
+ VsrConnection? lastConnection = null;
try
{
@@ -506,20 +494,13 @@ public sealed partial class TcpMessageStream
? overallDeadline
: Math.Min(overallDeadline, Environment.TickCount64 + VsrTransientFailoverCheckMs);
- var attempt = await SendVsrAttemptAsync(code, body, header, transientDeadline, overallDeadline,
+ var attempt = await SendVsrAttemptAsync(code, ns, body, header, transientDeadline, overallDeadline,
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!;
}
@@ -528,10 +509,12 @@ public sealed partial class TcpMessageStream
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);
}
@@ -562,7 +545,7 @@ public sealed partial class TcpMessageStream
{
if (requestEncoded)
{
- await DropVsrConnectionAsync(lastStream);
+ await DropVsrConnectionAsync(lastConnection);
}
throw;
@@ -588,236 +571,31 @@ public sealed partial class TcpMessageStream
}
/// <summary>
- /// One attempt on the current connection: encode the header into <paramref name="header" />, 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.
/// </summary>
- private async ValueTask<VsrAttempt> SendVsrAttemptAsync(int code, ReadOnlyMemory<byte> body, Memory<byte> header,
- long transientDeadline, long readDeadline, CancellationToken token)
+ private async ValueTask<VsrAttempt> SendVsrAttemptAsync(int code, ulong ns, ReadOnlyMemory<byte> body,
+ Memory<byte> header, long transientDeadline, long readDeadline, 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)
+ var connection = _connection;
+ if (connection is null)
{
- frameBuffer = ArrayPool<byte>.Shared.Rent(VsrHeader.HEADER_SIZE + body.Length);
+ return VsrAttempt.Failed(false, new NotConnectedException(), false, null);
}
- VsrHeader.EncodeRequestHeader(header.Span, _consensusSession, code, body.Span);
-
- encoded = true;
-
- var frame = Memory<byte>.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<byte> 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)
- {
- DropVsrConnectionLocked(stream);
- }
-
- return VsrAttempt.Failed(encoded, e, requestStarted, stream);
- }
- catch (Exception e)
- {
- return VsrAttempt.Failed(encoded, e, requestStarted, stream);
+ return await connection.SendAttemptAsync(code, ns, body, header, transientDeadline, readDeadline,
+ token);
}
finally
{
- if (frameBuffer is not null)
- {
- ArrayPool<byte>.Shared.Return(frameBuffer);
- }
-
_sendingSemaphore.Release();
}
}
- private async Task<IMemoryOwner<byte>> 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<byte>.Empty);
-
- return EmptyMemoryOwner.Instance;
- }
-
- var buffer = ArrayPool<byte>.Shared.Rent(bodySize);
- try
- {
- await ReadExactVsrAsync(stream, buffer.AsMemory(0, bodySize), readCancellation.Token, token);
- ReadOnlyMemory<byte> decoded = VsrReplyDecoder.Decode(_vsrReplyHeaderBuffer, buffer.AsMemory(0, bodySize));
- if (decoded.IsEmpty)
- {
- ArrayPool<byte>.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<byte>.Shared.Return(buffer);
- throw;
- }
- }
-
- private async ValueTask ReadExactVsrAsync(TcpConnectionStream stream, Memory<byte> 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;
- }
- }
-
/// <summary>
/// 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
@@ -873,27 +651,27 @@ public sealed partial class TcpMessageStream
/// <summary>
/// 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 <see cref="_sendingSemaphore" />,
- /// which owns every write to <see cref="_stream" />.
+ /// next request, so the socket cannot be reused. The caller must hold <see cref="_sendingSemaphore" />,
+ /// which owns every write to <see cref="_connection" />.
/// </summary>
- /// <param name="stream">
+ /// <param name="connection">
/// 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.
/// </param>
- 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();
+ connection.Close();
SetConnectionStateAsync(ConnectionState.Disconnected);
}
/// <summary>Drops the connection on behalf of a caller that no longer holds the sending lock.</summary>
- 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
@@ -906,73 +684,11 @@ public sealed partial class TcpMessageStream
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
- };
- }
-
- /// <summary>Outcome of one <see cref="SendVsrAttemptAsync" /> call on the current connection.</summary>
- /// <param name="Encoded">Whether the header was encoded, i.e. whether a request id may have been consumed.</param>
- /// <param name="Response">The decoded reply payload, non-null exactly when <paramref name="Error" /> is null.</param>
- /// <param name="Error">The failure that ended the attempt, or null on success.</param>
- /// <param name="RequestStarted">
- /// Whether any byte of the frame was written, which makes the server-side outcome unknowable on failure.
- /// </param>
- /// <param name="Stream">
- /// 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.
- /// </param>
- private readonly record struct VsrAttempt(
- bool Encoded,
- IMemoryOwner<byte>? Response,
- Exception? Error,
- bool RequestStarted,
- TcpConnectionStream? Stream)
- {
- public static VsrAttempt Ok(IMemoryOwner<byte> 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);
- }
- }
-
- /// <summary>Owns a pooled buffer while exposing only the decoded payload slice inside it.</summary>
- internal sealed class PooledMemoryOwner(byte[] buffer, int start, int length) : IMemoryOwner<byte>
- {
- private int _disposed;
-
- public Memory<byte> Memory => buffer.AsMemory(start, length);
-
- public void Dispose()
- {
- if (Interlocked.Exchange(ref _disposed, 1) == 0)
- {
- ArrayPool<byte>.Shared.Return(buffer);
- }
- }
- }
}
diff --git a/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.cs b/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.cs
index 146c93ee1..f0b2fc5cf 100644
--- a/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.cs
+++ b/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.cs
@@ -19,11 +19,9 @@ using System.Buffers;
using System.Buffers.Binary;
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;
@@ -46,8 +44,6 @@ namespace Apache.Iggy.IggyClient.Implementations;
/// </summary>
public sealed partial class TcpMessageStream : IIggyClient
{
- private const int InvalidCommandStatus = 3;
-
private static readonly HashSet<uint> SessionControlCodes =
[
CommandCodes.LOGIN_USER_CODE,
@@ -63,20 +59,13 @@ public sealed partial class TcpMessageStream : IIggyClient
private readonly SemaphoreSlim _connectionSemaphore;
private readonly ILogger<TcpMessageStream> _logger;
private readonly SemaphoreSlim _sendingSemaphore;
+ private VsrConnection? _connection;
private string _currentAddress = 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 bool IsConnecting => Volatile.Read(ref _isConnecting) != 0;
@@ -96,8 +85,8 @@ public sealed partial class TcpMessageStream : IIggyClient
public void Dispose()
{
_disposed = true;
- _stream?.Close();
- _stream?.Dispose();
+ _connection?.Close();
+ _connection?.Dispose();
SetConnectionStateAsync(ConnectionState.Disconnected);
_sendingSemaphore.Dispose();
@@ -131,10 +120,8 @@ public sealed partial class TcpMessageStream : IIggyClient
public async Task<StreamResponse?> 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<byte> responseBuffer = await SendWithResponseAsync(payload, token);
+ using IMemoryOwner<byte> responseBuffer
+ = await SendWithResponseAsync(CommandCodes.CREATE_STREAM_CODE, message, token);
if (responseBuffer.Memory.Length == 0)
{
@@ -148,10 +135,8 @@ public sealed partial class TcpMessageStream : IIggyClient
public async Task<StreamResponse?> 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<byte> responseBuffer = await SendWithResponseAsync(payload, token);
+ using IMemoryOwner<byte> responseBuffer
+ = await SendWithResponseAsync(CommandCodes.GET_STREAM_CODE, message, token);
if (responseBuffer.Memory.Length == 0)
{
@@ -165,10 +150,8 @@ public sealed partial class TcpMessageStream : IIggyClient
public async Task<IReadOnlyList<StreamResponse>> GetStreamsAsync(CancellationToken token = default)
{
var message = Array.Empty<byte>();
- var payload = new byte[4 + BufferSizes.INITIAL_BYTES_LENGTH + message.Length];
- TcpMessageStreamHelpers.CreatePayload(payload, message, CommandCodes.GET_STREAMS_CODE);
-
- using IMemoryOwner<byte> responseBuffer = await SendWithResponseAsync(payload, token);
+ using IMemoryOwner<byte> responseBuffer
+ = await SendWithResponseAsync(CommandCodes.GET_STREAMS_CODE, message, token);
if (responseBuffer.Memory.Length == 0)
{
@@ -182,30 +165,21 @@ public sealed partial class TcpMessageStream : IIggyClient
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);
}
/// <inheritdoc />
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);
}
/// <inheritdoc />
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);
}
/// <inheritdoc />
@@ -213,10 +187,8 @@ public sealed partial class TcpMessageStream : IIggyClient
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<byte> responseBuffer = await SendWithResponseAsync(payload, token);
+ using IMemoryOwner<byte> responseBuffer
+ = await SendWithResponseAsync(CommandCodes.GET_TOPICS_CODE, message, token);
if (responseBuffer.Memory.Length == 0)
{
@@ -231,10 +203,8 @@ public sealed partial class TcpMessageStream : IIggyClient
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<byte> responseBuffer = await SendWithResponseAsync(payload, token);
+ using IMemoryOwner<byte> responseBuffer
+ = await SendWithResponseAsync(CommandCodes.GET_TOPIC_CODE, message, token);
if (responseBuffer.Memory.Length == 0)
{
@@ -252,10 +222,8 @@ public sealed partial class TcpMessageStream : IIggyClient
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<byte> responseBuffer = await SendWithResponseAsync(payload, token);
+ using IMemoryOwner<byte> responseBuffer
+ = await SendWithResponseAsync(CommandCodes.CREATE_TOPIC_CODE, message, token);
if (responseBuffer.Memory.Length == 0)
{
@@ -274,20 +242,14 @@ public sealed partial class TcpMessageStream : IIggyClient
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);
}
/// <inheritdoc />
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));
}
@@ -295,10 +257,7 @@ public sealed partial class TcpMessageStream : IIggyClient
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);
}
@@ -371,11 +330,9 @@ public sealed partial class TcpMessageStream : IIggyClient
public async Task StoreOffsetAsync(Consumer consumer, Identifier streamId, Identifier topicId, ulong offset,
uint? partitionId, CancellationToken token = default)
{
+ var ns = ConsumerOffsetNamespace(streamId, topicId, partitionId);
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, ns, message, token);
}
/// <inheritdoc />
@@ -383,10 +340,8 @@ public sealed partial class TcpMessageStream : IIggyClient
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<byte> responseBuffer = await SendWithResponseAsync(payload, token);
+ using IMemoryOwner<byte> responseBuffer
+ = await SendWithResponseAsync(CommandCodes.GET_CONSUMER_OFFSET_CODE, message, token);
if (responseBuffer.Memory.Length == 0)
{
@@ -400,11 +355,9 @@ public sealed partial class TcpMessageStream : IIggyClient
public async Task DeleteOffsetAsync(Consumer consumer, Identifier streamId, Identifier topicId, uint? partitionId,
CancellationToken token = default)
{
+ var ns = ConsumerOffsetNamespace(streamId, topicId, partitionId);
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, ns, message, token);
}
/// <inheritdoc />
@@ -413,10 +366,8 @@ public sealed partial class TcpMessageStream : IIggyClient
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<byte> responseBuffer = await SendWithResponseAsync(payload, token);
+ using IMemoryOwner<byte> responseBuffer
+ = await SendWithResponseAsync(CommandCodes.GET_CONSUMER_GROUPS_CODE, message, token);
if (responseBuffer.Memory.Length == 0)
{
@@ -431,10 +382,8 @@ public sealed partial class TcpMessageStream : IIggyClient
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<byte> responseBuffer = await SendWithResponseAsync(payload, token);
+ using IMemoryOwner<byte> responseBuffer
+ = await SendWithResponseAsync(CommandCodes.GET_CONSUMER_GROUP_CODE, message, token);
if (responseBuffer.Memory.Length == 0)
{
@@ -449,10 +398,8 @@ public sealed partial class TcpMessageStream : IIggyClient
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<byte> responseBuffer = await SendWithResponseAsync(payload, token);
+ using IMemoryOwner<byte> responseBuffer
+ = await SendWithResponseAsync(CommandCodes.CREATE_CONSUMER_GROUP_CODE, message, token);
if (responseBuffer.Memory.Length == 0)
{
@@ -467,10 +414,7 @@ public sealed partial class TcpMessageStream : IIggyClient
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));
}
@@ -479,10 +423,7 @@ public sealed partial class TcpMessageStream : IIggyClient
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.
@@ -494,10 +435,7 @@ public sealed partial class TcpMessageStream : IIggyClient
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));
}
@@ -506,10 +444,7 @@ public sealed partial class TcpMessageStream : IIggyClient
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));
}
@@ -518,10 +453,7 @@ public sealed partial class TcpMessageStream : IIggyClient
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));
}
@@ -529,21 +461,16 @@ public sealed partial class TcpMessageStream : IIggyClient
public async Task DeleteSegmentsAsync(Identifier streamId, Identifier topicId, uint partitionId,
uint segmentsCount, CancellationToken token = default)
{
+ var ns = VsrNamespace.ForPartition(streamId, topicId, partitionId);
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, ns, message, token);
}
/// <inheritdoc />
public async Task<ClientResponse?> GetMeAsync(CancellationToken token = default)
{
var message = Array.Empty<byte>();
- var payload = new byte[4 + BufferSizes.INITIAL_BYTES_LENGTH + message.Length];
- TcpMessageStreamHelpers.CreatePayload(payload, message, CommandCodes.GET_ME_CODE);
-
- using IMemoryOwner<byte> responseBuffer = await SendWithResponseAsync(payload, token);
+ using IMemoryOwner<byte> responseBuffer = await SendWithResponseAsync(CommandCodes.GET_ME_CODE, message, token);
if (responseBuffer.Memory.Length == 0)
{
@@ -557,10 +484,8 @@ public sealed partial class TcpMessageStream : IIggyClient
public async Task<StatsResponse?> GetStatsAsync(CancellationToken token = default)
{
var message = Array.Empty<byte>();
- var payload = new byte[4 + BufferSizes.INITIAL_BYTES_LENGTH + message.Length];
- TcpMessageStreamHelpers.CreatePayload(payload, message, CommandCodes.GET_STATS_CODE);
-
- using IMemoryOwner<byte> responseBuffer = await SendWithResponseAsync(payload, token);
+ using IMemoryOwner<byte> responseBuffer
+ = await SendWithResponseAsync(CommandCodes.GET_STATS_CODE, message, token);
if (responseBuffer.Memory.Length == 0)
{
@@ -574,10 +499,8 @@ public sealed partial class TcpMessageStream : IIggyClient
public async Task<ClusterMetadata?> GetClusterMetadataAsync(CancellationToken token = default)
{
var message = Array.Empty<byte>();
- var payload = new byte[4 + BufferSizes.INITIAL_BYTES_LENGTH + message.Length];
- TcpMessageStreamHelpers.CreatePayload(payload, message, CommandCodes.GET_CLUSTER_METADATA_CODE);
-
- using IMemoryOwner<byte> responseBuffer = await SendWithResponseAsync(payload, token);
+ using IMemoryOwner<byte> responseBuffer
+ = await SendWithResponseAsync(CommandCodes.GET_CLUSTER_METADATA_CODE, message, token);
if (responseBuffer.Memory.Length == 0)
{
@@ -591,10 +514,7 @@ public sealed partial class TcpMessageStream : IIggyClient
public async Task PingAsync(CancellationToken token = default)
{
var message = Array.Empty<byte>();
- 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);
}
@@ -604,10 +524,7 @@ public sealed partial class TcpMessageStream : IIggyClient
IList<SystemSnapshotType> 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<byte> result = await SendWithResponseAsync(payload, token);
+ using IMemoryOwner<byte> result = await SendWithResponseAsync(CommandCodes.GET_SNAPSHOT_CODE, message, token);
return result.Memory.Span.ToArray();
}
@@ -617,14 +534,12 @@ public sealed partial class TcpMessageStream : IIggyClient
{
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<byte> result = await SendWithResponseAsync(buffer, token);
+ var ns = VsrNamespace.ForRequest((int)code, payload, VsrOperations.ForCode((int)code));
+ using IMemoryOwner<byte> result = await SendWithResponseAsync((int)code, ns, payload, token);
return result.Memory.Length <= 1 ? [] : result.Memory.Span.ToArray();
}
@@ -639,10 +554,8 @@ public sealed partial class TcpMessageStream : IIggyClient
public async Task<IReadOnlyList<ClientResponse>> GetClientsAsync(CancellationToken token = default)
{
var message = Array.Empty<byte>();
- var payload = new byte[4 + BufferSizes.INITIAL_BYTES_LENGTH + message.Length];
- TcpMessageStreamHelpers.CreatePayload(payload, message, CommandCodes.GET_CLIENTS_CODE);
-
- using IMemoryOwner<byte> responseBuffer = await SendWithResponseAsync(payload, token);
+ using IMemoryOwner<byte> responseBuffer
+ = await SendWithResponseAsync(CommandCodes.GET_CLIENTS_CODE, message, token);
if (responseBuffer.Memory.Length == 0)
{
@@ -656,10 +569,8 @@ public sealed partial class TcpMessageStream : IIggyClient
public async Task<ClientResponse?> 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<byte> responseBuffer = await SendWithResponseAsync(payload, token);
+ using IMemoryOwner<byte> responseBuffer
+ = await SendWithResponseAsync(CommandCodes.GET_CLIENT_CODE, message, token);
if (responseBuffer.Memory.Length == 0)
{
@@ -673,10 +584,8 @@ public sealed partial class TcpMessageStream : IIggyClient
public async Task<UserResponse?> 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<byte> responseBuffer = await SendWithResponseAsync(payload, token);
+ using IMemoryOwner<byte> responseBuffer
+ = await SendWithResponseAsync(CommandCodes.GET_USER_CODE, message, token);
if (responseBuffer.Memory.Length == 0)
{
@@ -690,10 +599,8 @@ public sealed partial class TcpMessageStream : IIggyClient
public async Task<IReadOnlyList<UserResponse>> GetUsersAsync(CancellationToken token = default)
{
var message = Array.Empty<byte>();
- var payload = new byte[4 + BufferSizes.INITIAL_BYTES_LENGTH + message.Length];
- TcpMessageStreamHelpers.CreatePayload(payload, message, CommandCodes.GET_USERS_CODE);
-
- using IMemoryOwner<byte> responseBuffer = await SendWithResponseAsync(payload, token);
+ using IMemoryOwner<byte> responseBuffer
+ = await SendWithResponseAsync(CommandCodes.GET_USERS_CODE, message, token);
if (responseBuffer.Memory.Length == 0)
{
@@ -708,10 +615,8 @@ public sealed partial class TcpMessageStream : IIggyClient
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<byte> responseBuffer = await SendWithResponseAsync(payload, token);
+ using IMemoryOwner<byte> responseBuffer
+ = await SendWithResponseAsync(CommandCodes.CREATE_USER_CODE, message, token);
if (responseBuffer.Memory.Length == 0)
{
@@ -725,10 +630,7 @@ public sealed partial class TcpMessageStream : IIggyClient
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);
}
/// <inheritdoc />
@@ -736,10 +638,7 @@ public sealed partial class TcpMessageStream : IIggyClient
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);
}
/// <inheritdoc />
@@ -747,10 +646,7 @@ public sealed partial class TcpMessageStream : IIggyClient
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);
}
/// <inheritdoc />
@@ -758,10 +654,7 @@ public sealed partial class TcpMessageStream : IIggyClient
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);
}
/// <inheritdoc />
@@ -779,13 +672,9 @@ public sealed partial class TcpMessageStream : IIggyClient
/// <inheritdoc />
public async Task LogoutUserAsync(CancellationToken token = default)
{
- var message = Array.Empty<byte>();
- 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<byte>.Empty, token);
}
finally
{
@@ -803,10 +692,8 @@ public sealed partial class TcpMessageStream : IIggyClient
CancellationToken token = default)
{
var message = Array.Empty<byte>();
- var payload = new byte[4 + BufferSizes.INITIAL_BYTES_LENGTH + message.Length];
- TcpMessageStreamHelpers.CreatePayload(payload, message, CommandCodes.GET_PERSONAL_ACCESS_TOKENS_CODE);
-
- using IMemoryOwner<byte> responseBuffer = await SendWithResponseAsync(payload, token);
+ using IMemoryOwner<byte> responseBuffer
+ = await SendWithResponseAsync(CommandCodes.GET_PERSONAL_ACCESS_TOKENS_CODE, message, token);
if (responseBuffer.Memory.Length == 0)
{
@@ -821,10 +708,8 @@ public sealed partial class TcpMessageStream : IIggyClient
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<byte> responseBuffer = await SendWithResponseAsync(payload, token);
+ using IMemoryOwner<byte> responseBuffer
+ = await SendWithResponseAsync(CommandCodes.CREATE_PERSONAL_ACCESS_TOKEN_CODE, message, token);
if (responseBuffer.Memory.Length == 0)
{
@@ -838,10 +723,7 @@ public sealed partial class TcpMessageStream : IIggyClient
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);
}
/// <inheritdoc />
@@ -897,18 +779,16 @@ public sealed partial class TcpMessageStream : IIggyClient
CancellationToken token)
{
var messageBufferSize = CalculateMessageBufferSize(streamId, topicId, consumer);
- var payloadBufferSize = CalculatePayloadBufferSize(messageBufferSize);
- var payload = ArrayPool<byte>.Shared.Rent(payloadBufferSize);
+ var payload = ArrayPool<byte>.Shared.Rent(messageBufferSize);
IMemoryOwner<byte>? 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);
if (responseBuffer.Memory.Length == 0)
{
responseBuffer.Dispose();
@@ -952,15 +832,14 @@ public sealed partial class TcpMessageStream : IIggyClient
+ 2 + partitioning.Length + 4 + 4;
var maxMessageBufferSize = TcpMessageStreamHelpers.CalculateMessageBytesCount(messages, encryptor)
+ metadataLength;
- var maxPayloadBufferSize = CalculatePayloadBufferSize(maxMessageBufferSize);
- IMemoryOwner<byte> payloadBuffer = MemoryPool<byte>.Shared.Rent(maxPayloadBufferSize);
- int payloadBufferSize;
+ var ns = SendMessagesNamespace(streamId, topicId, partitioning);
+ IMemoryOwner<byte> payloadBuffer = MemoryPool<byte>.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
{
@@ -968,16 +847,16 @@ public sealed partial class TcpMessageStream : IIggyClient
throw;
}
- return SendConfirmedAndDisposeAsync(payloadBuffer, payloadBufferSize, token);
+ return SendConfirmedAndDisposeAsync(ns, payloadBuffer, bodySize, token);
}
- private async Task<SendMessagesResponse> SendConfirmedAndDisposeAsync(IMemoryOwner<byte> payloadBuffer,
- int payloadBufferSize, CancellationToken token)
+ private async Task<SendMessagesResponse> SendConfirmedAndDisposeAsync(ulong ns, IMemoryOwner<byte> payloadBuffer,
+ int bodySize, CancellationToken token)
{
try
{
- using IMemoryOwner<byte> responseBuffer =
- await SendWithResponseAsync(payloadBuffer.Memory[..payloadBufferSize], token);
+ using IMemoryOwner<byte> responseBuffer = await SendWithResponseAsync(CommandCodes.SEND_MESSAGES_CODE,
+ ns, payloadBuffer.Memory[..bodySize], token);
return BinaryMapper.MapSendMessages(responseBuffer.Memory.Span);
}
finally
@@ -996,16 +875,42 @@ public sealed partial class TcpMessageStream : IIggyClient
};
}
- [MethodImpl(MethodImplOptions.AggressiveInlining)]
- private static int FillSendMessagesPayload(Span<byte> buffer, int maxMessageBufferSize,
- Identifier streamId, Identifier topicId, Partitioning partitioning, ReadOnlySpan<Message> messages,
- IMessageEncryptor? encryptor)
+ /// <summary>
+ /// The routing namespace for a send: callers resolve balanced and message-key partitioning to an
+ /// explicit partition before framing, so anything else reaching this point is a bug surfaced as the
+ /// same error the server-side router would raise.
+ /// </summary>
+ private static ulong SendMessagesNamespace(Identifier streamId, Identifier topicId, Partitioning partitioning)
{
- 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;
+ if (partitioning.Kind != Enums.Partitioning.PartitionId)
+ {
+ throw VsrError.Exception(VsrError.FEATURE_UNAVAILABLE,
+ "Under VSR the partition must be resolved by the client; balanced and message-key partitioning cannot be sent on the wire.");
+ }
+
+ if (partitioning.Value.Length < 4)
+ {
+ throw VsrError.Exception(VsrError.INVALID_COMMAND,
+ $"Partition id is {partitioning.Value.Length} bytes, expected 4.");
+ }
+
+ return VsrNamespace.ForPartition(streamId, topicId,
+ BinaryPrimitives.ReadUInt32LittleEndian(partitioning.Value));
+ }
+
+ /// <summary>
+ /// The routing namespace for a consumer-offset write. The broker routes explicit partitions only, so a
+ /// missing partition id fails client-side before a request id is consumed.
+ /// </summary>
+ private static ulong ConsumerOffsetNamespace(Identifier streamId, Identifier topicId, uint? partitionId)
+ {
+ if (partitionId is null)
+ {
+ throw VsrError.Exception(VsrError.INVALID_IDENTIFIER,
+ "Under VSR a consumer-offset request must carry an explicit partition id.");
+ }
+
+ return VsrNamespace.ForPartition(streamId, topicId, partitionId.Value);
}
private async Task TryEstablishConnectionAsync(bool autoLogin, CancellationToken token)
@@ -1015,12 +920,12 @@ public sealed partial class TcpMessageStream : IIggyClient
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();
ResetConsensusSession();
}
@@ -1056,16 +961,16 @@ public sealed partial class TcpMessageStream : IIggyClient
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
{
@@ -1083,7 +988,7 @@ public sealed partial class TcpMessageStream : IIggyClient
continue;
}
- 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);
@@ -1151,23 +1056,7 @@ public sealed partial class TcpMessageStream : IIggyClient
}
}
- /// <summary>
- /// 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.
- /// </summary>
- 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<TcpConnectionStream> CreateSslStreamAndAuthenticate(Socket socket, TlsSettings tlsSettings)
+ private async Task<Stream> CreateSslStreamAndAuthenticate(Socket socket, TlsSettings tlsSettings)
{
ValidateCertificatePath(tlsSettings.CertificatePath);
@@ -1178,22 +1067,33 @@ public sealed partial class TcpMessageStream : IIggyClient
await sslStream.AuthenticateAsClientAsync(tlsSettings.Hostname);
- return new TcpConnectionStream(sslStream);
+ return sslStream;
}
- private async Task SendAckAsync(ReadOnlyMemory<byte> payload, CancellationToken token = default)
+ private async Task SendAckAsync(int code, ReadOnlyMemory<byte> body, CancellationToken token)
{
- using IMemoryOwner<byte> _ = await SendWithResponseAsync(payload, token);
+ using IMemoryOwner<byte> _ = await SendWithResponseAsync(code, body, token);
}
- private async Task<IMemoryOwner<byte>> SendWithResponseAsync(ReadOnlyMemory<byte> payload,
- CancellationToken token = default)
+ private async Task SendAckAsync(int code, ulong ns, ReadOnlyMemory<byte> body, CancellationToken token)
+ {
+ using IMemoryOwner<byte> _ = await SendWithResponseAsync(code, ns, body, token);
+ }
+
+ private Task<IMemoryOwner<byte>> SendWithResponseAsync(int code, ReadOnlyMemory<byte> body,
+ CancellationToken token)
+ {
+ return SendWithResponseAsync(code, 0, body, token);
+ }
+
+ private async Task<IMemoryOwner<byte>> SendWithResponseAsync(int code, ulong ns, ReadOnlyMemory<byte> body,
+ CancellationToken token, bool autoLoginOnReconnect = true)
{
try
{
- return await SendRawAsync(payload, token);
+ return await SendRawAsync(code, ns, 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)
@@ -1203,12 +1103,12 @@ public sealed partial class TcpMessageStream : IIggyClient
throw;
}
- return await HandleReconnectionAsync(payload, token);
+ return await HandleReconnectionAsync(code, ns, body, autoLoginOnReconnect, token);
}
}
- private async Task<IMemoryOwner<byte>> HandleReconnectionAsync(ReadOnlyMemory<byte> payload,
- CancellationToken token)
+ private async Task<IMemoryOwner<byte>> HandleReconnectionAsync(int code, ulong ns, ReadOnlyMemory<byte> body,
+ bool autoLogin, CancellationToken token)
{
var currentTime = DateTimeOffset.UtcNow;
await _connectionSemaphore.WaitAsync(token);
@@ -1219,18 +1119,18 @@ public sealed partial class TcpMessageStream : IIggyClient
&& _lastConnectionTime > currentTime)
{
_logger.LogInformation("Connection already established, sending payload");
- return await SendRawAsync(payload, token);
+ return await SendRawAsync(code, ns, body, token);
}
SetConnectionStateAsync(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, ns, body, token);
}
finally
{
@@ -1238,32 +1138,6 @@ public sealed partial class TcpMessageStream : IIggyClient
}
}
- private Task<IMemoryOwner<byte>> SendRawAsync(ReadOnlyMemory<byte> 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
@@ -1361,45 +1235,4 @@ public sealed partial class TcpMessageStream : IIggyClient
return false;
}
-
- internal sealed class EmptyMemoryOwner : IMemoryOwner<byte>
- {
- public static readonly EmptyMemoryOwner Instance = new();
-
- private EmptyMemoryOwner()
- {
- }
-
- public Memory<byte> Memory => Memory<byte>.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<byte>
- {
- private readonly byte[] _value = ArrayPool<byte>.Shared.Rent(minimumLength);
- private int _disposed;
-
- public Memory<byte> Memory => _value.AsMemory()[..minimumLength];
-
- public void Dispose()
- {
- if (Interlocked.Exchange(ref _disposed, 1) != 0)
- {
- return;
- }
-
- ArrayPool<byte>.Shared.Return(_value, clearOnReturn);
- }
- }
}
diff --git a/foreign/csharp/Iggy_SDK/Iggy_SDK.csproj b/foreign/csharp/Iggy_SDK/Iggy_SDK.csproj
index 79a639036..84fd52285 100644
--- a/foreign/csharp/Iggy_SDK/Iggy_SDK.csproj
+++ b/foreign/csharp/Iggy_SDK/Iggy_SDK.csproj
@@ -27,7 +27,7 @@
<TargetFrameworks>net8.0;net10.0</TargetFrameworks>
<AssemblyName>Apache.Iggy</AssemblyName>
<RootNamespace>Apache.Iggy</RootNamespace>
- <Version>0.9.0-edge.1</Version>
+ <Version>0.9.0-edge.2</Version>
<GenerateDocumentationFile>true</GenerateDocumentationFile>
</PropertyGroup>
@@ -68,13 +68,15 @@
</ItemGroup>
<ItemGroup>
- <PackageReference Include="Microsoft.Extensions.Logging.Abstractions" />
- <PackageReference Include="System.IO.Hashing" />
+ <PackageReference Include="Microsoft.Extensions.Logging.Abstractions"/>
+ <PackageReference Include="System.IO.Hashing"/>
</ItemGroup>
<ItemGroup>
- <InternalsVisibleTo Include="Apache.Iggy.Tests" />
- <InternalsVisibleTo Include="Apache.Iggy.Tests.Integrations" />
+ <InternalsVisibleTo Include="Apache.Iggy.Tests"/>
+ <InternalsVisibleTo Include="Apache.Iggy.Tests.Integrations"/>
+ <!-- Lets the test mocking framework proxy internal interfaces such as ISessionEpochProvider. -->
+ <InternalsVisibleTo Include="DynamicProxyGenAssembly2"/>
</ItemGroup>
</Project>
diff --git a/foreign/csharp/Iggy_SDK/Mappers/BinaryMapper.cs b/foreign/csharp/Iggy_SDK/Mappers/BinaryMapper.cs
index f5be657f2..257d40193 100644
--- a/foreign/csharp/Iggy_SDK/Mappers/BinaryMapper.cs
+++ b/foreign/csharp/Iggy_SDK/Mappers/BinaryMapper.cs
@@ -25,9 +25,9 @@ using Apache.Iggy.Enums;
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 class BinaryMapper
});
}
- 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 cd5a9e871..0db7b1cce 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<byte>
+ {
+ private readonly byte[] _value = ArrayPool<byte>.Shared.Rent(minimumLength);
+ private int _disposed;
+
+ public Memory<byte> Memory => _value.AsMemory()[..minimumLength];
+
+ public void Dispose()
+ {
+ if (Interlocked.Exchange(ref _disposed, 1) != 0)
+ {
+ return;
+ }
+
+ ArrayPool<byte>.Shared.Return(_value, clearOnReturn);
+ }
+ }
}
diff --git a/foreign/csharp/Iggy_SDK/Utils/TcpMessageStreamHelpers.cs b/foreign/csharp/Iggy_SDK/Utils/TcpMessageStreamHelpers.cs
index 6f0dc135c..cbf00ccfa 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<byte> result, Span<byte> 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<byte> buffer)
- {
- var status = BinaryPrimitives.ReadInt32LittleEndian(buffer[..4]);
- var length = BinaryPrimitives.ReadInt32LittleEndian(buffer[4..]);
-
- return (status, length);
- }
-
internal static int CalculateMessageBytesCount(ReadOnlySpan<Message> 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 8cdbdcc16..5eda4d0d4 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 _epoch;
private bool _registerPending;
private ulong _requestCounter;
private ulong? _session;
@@ -89,6 +90,22 @@ internal sealed class ConsensusSession
}
}
+ /// <summary>
+ /// Generation of the identity, bumped on every re-arm. Anything scoped to a server session - a
+ /// consumer-group membership, say - compares the epoch it was established under to detect that the
+ /// session it lived on is gone, without inferring it from connection-state edges.
+ /// </summary>
+ internal ulong Epoch
+ {
+ get
+ {
+ lock (_gate)
+ {
+ return _epoch;
+ }
+ }
+ }
+
internal ConsensusSession() : this(GenerateClientId())
{
}
@@ -205,6 +222,7 @@ internal sealed class ConsensusSession
_session = null;
_requestCounter = 1;
_registerPending = false;
+ _epoch++;
}
private static UInt128 GenerateClientId()
diff --git a/foreign/csharp/Iggy_SDK/ConnectionStream/IConnectionStream.cs b/foreign/csharp/Iggy_SDK/Vsr/ISessionEpochProvider.cs
similarity index 64%
rename from foreign/csharp/Iggy_SDK/ConnectionStream/IConnectionStream.cs
rename to foreign/csharp/Iggy_SDK/Vsr/ISessionEpochProvider.cs
index 288ba53c3..f81d2ebf4 100644
--- a/foreign/csharp/Iggy_SDK/ConnectionStream/IConnectionStream.cs
+++ b/foreign/csharp/Iggy_SDK/Vsr/ISessionEpochProvider.cs
@@ -15,12 +15,14 @@
// specific language governing permissions and limitations
// under the License.
-namespace Apache.Iggy.ConnectionStream;
+namespace Apache.Iggy.Vsr;
-internal interface IConnectionStream : IDisposable
+/// <summary>
+/// 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. Transports without a consensus
+/// session (HTTP) do not implement it and their consumers fall back to connection-state edges.
+/// </summary>
+internal interface ISessionEpochProvider
{
- ValueTask SendAsync(ReadOnlyMemory<byte> payload, CancellationToken cancellationToken = default);
- ValueTask<int> ReadAsync(Memory<byte> buffer, CancellationToken cancellationToken = default);
- Task FlushAsync(CancellationToken cancellationToken = default);
- void Close();
+ ulong SessionEpoch { get; }
}
diff --git a/foreign/csharp/Iggy_SDK/Vsr/VsrConnection.cs b/foreign/csharp/Iggy_SDK/Vsr/VsrConnection.cs
new file mode 100644
index 000000000..302974710
--- /dev/null
+++ b/foreign/csharp/Iggy_SDK/Vsr/VsrConnection.cs
@@ -0,0 +1,355 @@
+// 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;
+
+/// <summary>
+/// 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.
+/// </summary>
+internal sealed class VsrConnection : IDisposable
+{
+ /// <summary>Backoff between replays of a transiently refused request.</summary>
+ private const int TransientRetryIntervalMs = 50;
+
+ private readonly ILogger _logger;
+ private readonly long _maxResponseFrameSize;
+
+ /// <summary>
+ /// 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.
+ /// </summary>
+ private readonly Action<VsrConnection> _onDropped;
+
+ private readonly byte[] _replyHeaderBuffer = new byte[VsrHeader.HEADER_SIZE];
+ private readonly int _requestTimeoutMs;
+ private readonly Stream _stream;
+
+ /// <summary>The session identity requests on this connection encode from.</summary>
+ internal ConsensusSession Session { get; }
+
+ internal VsrConnection(Stream stream, ConsensusSession session, long maxResponseFrameSize, int requestTimeoutMs,
+ Action<VsrConnection> onDropped, ILogger logger)
+ {
+ _stream = stream;
+ Session = session;
+ _maxResponseFrameSize = maxResponseFrameSize;
+ _requestTimeoutMs = requestTimeoutMs;
+ _onDropped = onDropped;
+ _logger = logger;
+ }
+
+ public void Dispose()
+ {
+ _stream.Dispose();
+ }
+
+ internal void Close()
+ {
+ _stream.Close();
+ }
+
+ internal static bool IsConnectionException(Exception ex)
+ {
+ return ex is IggyZeroBytesException or
+ NotConnectedException or
+ SocketException or
+ IOException or
+ ObjectDisposedException;
+ }
+
+ /// <summary>
+ /// One attempt on this connection: encode the header into <paramref name="header" />, write the frame,
+ /// and replay it - same session, same request id - while the server answers transiently. The caller
+ /// must hold the sending lock.
+ /// </summary>
+ internal async ValueTask<VsrAttempt> SendAttemptAsync(int code, ulong ns, ReadOnlyMemory<byte> body,
+ Memory<byte> header, long transientDeadline, long readDeadline, CancellationToken token)
+ {
+ var encoded = false;
+ var requestStarted = false;
+
+ try
+ {
+ VsrHeader.EncodeRequestHeader(header.Span, Session, code, ns, body.Length);
+
+ 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;
+
+ // Header and body go out as two writes, straight from the buffers they were encoded into:
+ // no frame-sized rent and no copy. The server frames by the announced size, and the
+ // lockstep protocol already pays a full round trip per request, so the extra segment a
+ // small request costs (with Nagle disabled) is noise against the reply wait.
+ await _stream.WriteAsync(header, token);
+ if (!body.IsEmpty)
+ {
+ await _stream.WriteAsync(body, token);
+ }
+
+ await _stream.FlushAsync(token);
+
+ IMemoryOwner<byte> response = await ReadReplyAsync(readDeadline, 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<IMemoryOwner<byte>> ReadReplyAsync(long readDeadline, 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<byte>.Empty);
+
+ return EmptyMemoryOwner.Instance;
+ }
+
+ var buffer = ArrayPool<byte>.Shared.Rent(bodySize);
+ try
+ {
+ await ReadExactAsync(buffer.AsMemory(0, bodySize), readCancellation.Token, token);
+ ReadOnlyMemory<byte> decoded = VsrReplyDecoder.Decode(_replyHeaderBuffer, buffer.AsMemory(0, bodySize));
+ if (decoded.IsEmpty)
+ {
+ ArrayPool<byte>.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<byte>.Shared.Return(buffer);
+ throw;
+ }
+ }
+
+ private async ValueTask ReadExactAsync(Memory<byte> 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
+ };
+ }
+}
+
+/// <summary>Outcome of one <see cref="VsrConnection.SendAttemptAsync" /> call.</summary>
+/// <param name="Encoded">Whether the header was encoded, i.e. whether a request id may have been consumed.</param>
+/// <param name="Response">The decoded reply payload, non-null exactly when <paramref name="Error" /> is null.</param>
+/// <param name="Error">The failure that ended the attempt, or null on success.</param>
+/// <param name="RequestStarted">
+/// Whether any byte of the frame was written, which makes the server-side outcome unknowable on failure.
+/// </param>
+/// <param name="Connection">
+/// 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.
+/// </param>
+internal readonly record struct VsrAttempt(
+ bool Encoded,
+ IMemoryOwner<byte>? Response,
+ Exception? Error,
+ bool RequestStarted,
+ VsrConnection? Connection)
+{
+ public static VsrAttempt Ok(IMemoryOwner<byte> 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);
+ }
+}
+
+/// <summary>Shared empty reply payload; disposes to nothing, so it is safe to hand out repeatedly.</summary>
+internal sealed class EmptyMemoryOwner : IMemoryOwner<byte>
+{
+ public static readonly EmptyMemoryOwner Instance = new();
+
+ private EmptyMemoryOwner()
+ {
+ }
+
+ public Memory<byte> Memory => Memory<byte>.Empty;
+
+ public void Dispose()
+ {
+ }
+}
+
+/// <summary>Owns a pooled buffer while exposing only the decoded payload slice inside it.</summary>
+internal sealed class PooledMemoryOwner(byte[] buffer, int start, int length) : IMemoryOwner<byte>
+{
+ private int _disposed;
+
+ public Memory<byte> Memory => buffer.AsMemory(start, length);
+
+ public void Dispose()
+ {
+ if (Interlocked.Exchange(ref _disposed, 1) == 0)
+ {
+ ArrayPool<byte>.Shared.Return(buffer);
+ }
+ }
+}
diff --git a/foreign/csharp/Iggy_SDK/Vsr/VsrHeader.cs b/foreign/csharp/Iggy_SDK/Vsr/VsrHeader.cs
index 5ebb96e83..dbd00f28e 100644
--- a/foreign/csharp/Iggy_SDK/Vsr/VsrHeader.cs
+++ b/foreign/csharp/Iggy_SDK/Vsr/VsrHeader.cs
@@ -49,8 +49,8 @@ internal static class VsrHeader
internal const int EVICTION_REASON_OFFSET = 255;
/// <summary>
- /// Encodes the request header for a classic command code and its body. Returns the total frame size
- /// (header plus body).
+ /// Encodes the request header for a command code, the namespace the caller resolved from its typed
+ /// identifiers, and the body length. Returns the total frame size (header plus body).
/// </summary>
/// <remarks>
/// Everything that can fail runs before a request id is consumed. The primary accepts any id above the
@@ -58,7 +58,7 @@ internal static class VsrHeader
/// different request would let the client table answer that request from the first one's cached reply.
/// </remarks>
internal static int EncodeRequestHeader(Span<byte> header, ConsensusSession session, int code,
- ReadOnlySpan<byte> payload)
+ ulong partitionNamespace, int bodyLength)
{
if (header.Length < HEADER_SIZE)
{
@@ -69,15 +69,15 @@ internal static class VsrHeader
header.Clear();
var operation = VsrOperations.ForCode(code);
- var ns = VsrNamespace.ForRequest(code, payload, operation);
+ var ns = VsrNamespace.ForOperation(operation, partitionNamespace);
- if (payload.Length > int.MaxValue - HEADER_SIZE)
+ if (bodyLength > int.MaxValue - HEADER_SIZE)
{
throw VsrError.Exception(VsrError.INVALID_COMMAND, "Request body exceeds the maximum frame size.");
}
var frame = session.Resolve(operation);
- var totalSize = HEADER_SIZE + payload.Length;
+ var totalSize = HEADER_SIZE + bodyLength;
BinaryPrimitives.WriteUInt32LittleEndian(header[SIZE_OFFSET..], (uint)totalSize);
header[COMMAND_OFFSET] = (byte)Command2.Request;
diff --git a/foreign/csharp/Iggy_SDK/Vsr/VsrNamespace.cs b/foreign/csharp/Iggy_SDK/Vsr/VsrNamespace.cs
index 34187591d..b27bcd038 100644
--- a/foreign/csharp/Iggy_SDK/Vsr/VsrNamespace.cs
+++ b/foreign/csharp/Iggy_SDK/Vsr/VsrNamespace.cs
@@ -56,9 +56,38 @@ internal static class VsrNamespace
private const byte ConsumerGroupKind = 2;
/// <summary>
- /// The namespace a request header must carry, peeking into the classic payload for the ops the server
- /// routes by partition. Named (string) stream or topic identifiers resolve to 0; the server resolves
- /// them itself.
+ /// The namespace a request header carries for its operation. Register and logout ride the metadata
+ /// consensus group; non-replicated and metadata operations route by 0; partition-plane operations carry
+ /// the namespace the caller packed from its typed identifiers.
+ /// </summary>
+ internal static ulong ForOperation(VsrOperation operation, ulong partitionNamespace)
+ {
+ if (operation is VsrOperation.Register or VsrOperation.Logout)
+ {
+ return METADATA_CONSENSUS_NAMESPACE;
+ }
+
+ if (operation == VsrOperation.NonReplicated || operation.IsMetadata())
+ {
+ return 0;
+ }
+
+ return partitionNamespace;
+ }
+
+ /// <summary>
+ /// Packs the namespace for a partition-plane request from its typed identifiers. Named (string) stream
+ /// or topic identifiers resolve to 0; the server resolves them itself.
+ /// </summary>
+ internal static ulong ForPartition(Identifier streamId, Identifier topicId, uint partitionId)
+ {
+ return FromPartition(NumericValue(streamId), NumericValue(topicId), partitionId);
+ }
+
+ /// <summary>
+ /// The namespace for a raw binary request, peeking into the serialized body for the ops the server
+ /// routes by partition. Only <c>SendBinaryRequestAsync</c> lands here: the typed
+ /// command surface packs its namespace from typed identifiers instead.
/// </summary>
internal static ulong ForRequest(int code, ReadOnlySpan<byte> payload, VsrOperation operation)
{
@@ -175,6 +204,22 @@ internal static class VsrNamespace
return FromPartition(streamId, topicId, ReadUInt32(payload, position));
}
+ /// <summary>Numeric identifier value, or <c>null</c> for a named identifier the server has to resolve.</summary>
+ private static uint? NumericValue(Identifier identifier)
+ {
+ if (identifier.Kind != IdKind.Numeric)
+ {
+ return null;
+ }
+
+ if (identifier.Length != 4 || identifier.Value.Length < 4)
+ {
+ throw Malformed();
+ }
+
+ return BinaryPrimitives.ReadUInt32LittleEndian(identifier.Value);
+ }
+
private static ulong FromPartition(uint? streamId, uint? topicId, uint partitionId)
{
if (streamId is null || topicId is null)
diff --git a/foreign/csharp/Iggy_SDK_Tests/ConsumerTests/ConsumerSessionEpochTests.cs b/foreign/csharp/Iggy_SDK_Tests/ConsumerTests/ConsumerSessionEpochTests.cs
new file mode 100644
index 000000000..8f0f47682
--- /dev/null
+++ b/foreign/csharp/Iggy_SDK_Tests/ConsumerTests/ConsumerSessionEpochTests.cs
@@ -0,0 +1,197 @@
+// 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;
+
+/// <summary>
+/// Group membership on a transport that exposes a consensus session is keyed off the session epoch rather
+/// than off connection-state edges, so these cover both arms of that branch.
+/// </summary>
+public sealed class ConsumerSessionEpochTests
+{
+ [Fact]
+ public async Task
+ given_group_consumer_when_reauthenticated_on_the_same_session_epoch_should_not_rejoin_the_group()
+ {
+ var client = new EpochClient();
+ 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_epoch_moved_should_rejoin_the_group()
+ {
+ var client = new EpochClient();
+ 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.SessionEpoch++;
+ await client.RaiseAsync(ConnectionState.Connected, ConnectionState.Authenticated);
+
+ Assert.Equal(2, client.JoinCount);
+ await consumer.DisposeAsync();
+ }
+
+ /// <summary>
+ /// A disconnect surrenders the membership even though the epoch has not moved yet. Holding it would let
+ /// the poll loop keep issuing requests the server refuses for as long as the reconnect takes.
+ /// </summary>
+ [Fact]
+ public async Task given_group_consumer_when_disconnected_should_surrender_membership_and_rejoin_on_reconnect()
+ {
+ var client = new EpochClient();
+ 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);
+ await client.RaiseAsync(ConnectionState.Connecting, ConnectionState.Authenticated);
+
+ Assert.Equal(2, client.JoinCount);
+ await consumer.DisposeAsync();
+ }
+
+ /// <summary>
+ /// 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.
+ /// </summary>
+ [Fact]
+ public async Task given_plain_consumer_when_disconnected_should_keep_polling_after_reconnect()
+ {
+ var client = new EpochClient();
+ 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.SessionEpoch++;
+ await client.RaiseAsync(ConnectionState.Connecting, ConnectionState.Authenticated);
+
+ var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
+ await using IAsyncEnumerator<ReceivedMessage> 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
+ };
+ }
+
+ /// <summary>
+ /// 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.
+ /// </summary>
+ private sealed class EpochClient
+ {
+ private readonly List<Func<ConnectionStateChangedEventArgs, Task>> _subscribers = [];
+
+ public IIggyClient Object { get; }
+
+ public ulong SessionEpoch { get; set; }
+
+ public int JoinCount { get; private set; }
+
+ public EpochClient()
+ {
+ var mock = new Mock<IIggyClient>(MockBehavior.Loose);
+ mock.As<ISessionEpochProvider>().SetupGet(c => c.SessionEpoch).Returns(() => SessionEpoch);
+ mock.Setup(c => c.ConnectAsync(It.IsAny<CancellationToken>())).Returns(Task.CompletedTask);
+ mock.Setup(c => c.SubscribeConnectionEvents(It.IsAny<Func<ConnectionStateChangedEventArgs, Task>>()))
+ .Callback<Func<ConnectionStateChangedEventArgs, Task>>(_subscribers.Add);
+ mock.Setup(c => c.UnsubscribeConnectionEvents(It.IsAny<Func<ConnectionStateChangedEventArgs, Task>>()))
+ .Callback<Func<ConnectionStateChangedEventArgs, Task>>(callback => _subscribers.Remove(callback));
+ mock.Setup(c => c.GetConsumerGroupByIdAsync(It.IsAny<Identifier>(), It.IsAny<Identifier>(),
+ It.IsAny<Identifier>(), It.IsAny<CancellationToken>()))
+ .ReturnsAsync(new ConsumerGroupResponse
+ {
+ Id = 1,
+ Name = "group-1",
+ MembersCount = 1,
+ PartitionsCount = 1
+ });
+ mock.Setup(c => c.JoinConsumerGroupAsync(It.IsAny<Identifier>(), It.IsAny<Identifier>(),
+ It.IsAny<Identifier>(), It.IsAny<CancellationToken>()))
+ .Callback(() => JoinCount++)
+ .Returns(Task.CompletedTask);
+ mock.Setup(c => c.PollMessagesAsync(It.IsAny<Identifier>(), It.IsAny<Identifier>(), It.IsAny<uint?>(),
+ It.IsAny<Consumer>(), It.IsAny<PollingStrategy>(), It.IsAny<uint>(), It.IsAny<bool>(),
+ It.IsAny<CancellationToken>()))
+ .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<ConnectionStateChangedEventArgs, Task> 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 7d65beed3..f4adb538d 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.Contracts;
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 @@ public class RentedConsumerTests
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 e37aa3575..e0054919a 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.Encryption;
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 sealed class BinaryMapper
// 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 sealed class BinaryMapper
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 sealed class BinaryMapper
frame.CopyTo(combined.AsSpan(16));
Assert.Throws<MalformedResponseException>(() =>
- Mappers.BinaryMapper.MapRentedMessages(combined, TcpMessageStream.EmptyMemoryOwner.Instance, encryptor));
+ Mappers.BinaryMapper.MapRentedMessages(combined, EmptyMemoryOwner.Instance, encryptor));
}
[Fact]
@@ -425,7 +425,7 @@ public sealed class BinaryMapper
frame.CopyTo(combined.AsSpan(16));
var ex = Assert.Throws<MessageDecryptionException>(() =>
- 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
index 73e9ba3b9..7463b1df7 100644
--- a/foreign/csharp/Iggy_SDK_Tests/UtilityTests/SlicedMemoryOwnerTests.cs
+++ b/foreign/csharp/Iggy_SDK_Tests/UtilityTests/SlicedMemoryOwnerTests.cs
@@ -19,7 +19,7 @@ using System.Buffers;
using System.Reflection;
using System.Runtime.CompilerServices;
using System.Runtime.InteropServices;
-using Apache.Iggy.IggyClient.Implementations;
+using Apache.Iggy.Utils;
namespace Apache.Iggy.Tests.UtilityTests;
@@ -31,7 +31,7 @@ public class SlicedMemoryOwnerTests
public void Dispose_ReturnsBufferToPool()
{
var owner = ArrayPoolHelper.Rent(BufferSize);
- byte[] underlying = GetUnderlyingArray(owner);
+ var underlying = GetUnderlyingArray(owner);
owner.Dispose();
using var second = ArrayPoolHelper.Rent(BufferSize);
@@ -50,10 +50,10 @@ public class SlicedMemoryOwnerTests
[Fact]
public void FinalizerIsDeclared()
{
- Type sliced = typeof(ArrayPoolHelper)
+ var sliced = typeof(ArrayPoolHelper)
.GetNestedType("SlicedMemoryOwner", BindingFlags.NonPublic)!;
- MethodInfo? finalizer = sliced.GetMethod("Finalize", BindingFlags.NonPublic | BindingFlags.Instance);
+ var finalizer = sliced.GetMethod("Finalize", BindingFlags.NonPublic | BindingFlags.Instance);
Assert.NotNull(finalizer);
}
@@ -61,7 +61,7 @@ public class SlicedMemoryOwnerTests
[Fact]
public void ForgotDispose_FinalizerRunsAndReclaimsInstance()
{
- WeakReference weakRef = RentWeak();
+ var weakRef = RentWeak();
GC.Collect();
GC.WaitForPendingFinalizers();
@@ -70,12 +70,15 @@ public class SlicedMemoryOwnerTests
Assert.False(weakRef.IsAlive);
[MethodImpl(MethodImplOptions.NoInlining)]
- static WeakReference RentWeak() => new(ArrayPoolHelper.Rent(BufferSize));
+ static WeakReference RentWeak()
+ {
+ return new WeakReference(ArrayPoolHelper.Rent(BufferSize));
+ }
}
private static byte[] GetUnderlyingArray(IMemoryOwner<byte> owner)
{
- if (!MemoryMarshal.TryGetArray<byte>(owner.Memory, out var segment) || segment.Array is null)
+ if (!MemoryMarshal.TryGetArray<byte>(owner.Memory, out ArraySegment<byte> segment) || segment.Array is null)
{
throw new InvalidOperationException("SlicedMemoryOwner.Memory must be array-backed.");
}
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 137a2bb2d..000000000
--- 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 51e4d7f0a..aaf438213 100644
--- a/foreign/csharp/Iggy_SDK_Tests/VsrTests/ConsensusSessionTests.cs
+++ b/foreign/csharp/Iggy_SDK_Tests/VsrTests/ConsensusSessionTests.cs
@@ -215,4 +215,24 @@ public sealed class ConsensusSessionTests
Assert.Equal(1UL, session.RequestCounter);
Assert.NotEqual((UInt128)1, session.ClientId);
}
+
+ [Fact]
+ public void Epoch_AdvancesOnEveryReArmAndNotOnBind()
+ {
+ var session = new ConsensusSession(1);
+ var initialEpoch = session.Epoch;
+
+ session.Resolve(VsrOperation.Register);
+ session.Bind(10);
+ Assert.Equal(initialEpoch, session.Epoch);
+
+ session.Reset();
+ Assert.Equal(initialEpoch + 1, session.Epoch);
+
+ // A register on a previously bound session re-arms the identity, which is a new epoch too.
+ session.Resolve(VsrOperation.Register);
+ session.Bind(11);
+ session.Resolve(VsrOperation.Register);
+ Assert.Equal(initialEpoch + 2, session.Epoch);
+ }
}
diff --git a/foreign/csharp/Iggy_SDK_Tests/VsrTests/VsrHeaderTests.cs b/foreign/csharp/Iggy_SDK_Tests/VsrTests/VsrHeaderTests.cs
index f35843ff3..3f06e7003 100644
--- a/foreign/csharp/Iggy_SDK_Tests/VsrTests/VsrHeaderTests.cs
+++ b/foreign/csharp/Iggy_SDK_Tests/VsrTests/VsrHeaderTests.cs
@@ -33,10 +33,11 @@ public sealed class VsrHeaderTests
return consensusSession;
}
- private static byte[] Encode(ConsensusSession session, int code, byte[] payload, out int totalSize)
+ private static byte[] Encode(ConsensusSession session, int code, int bodyLength, out int totalSize,
+ ulong ns = 0)
{
var header = new byte[VsrHeader.HEADER_SIZE];
- totalSize = VsrHeader.EncodeRequestHeader(header, session, code, payload);
+ totalSize = VsrHeader.EncodeRequestHeader(header, session, code, ns, bodyLength);
return header;
}
@@ -52,27 +53,20 @@ public sealed class VsrHeaderTests
}
/// <summary>
- /// A namespace failure must leave the session exactly as it was. Consuming the request id would gap the
- /// next metadata request, and dropping the binding would leave the client unable to send or re-login.
+ /// The caller resolves the routing namespace from its typed identifiers, but only partition-plane
+ /// operations carry it: a stray namespace on a metadata or non-replicated request would route it off
+ /// the metadata consensus group.
/// </summary>
[Fact]
- public void Encode_NamespaceFailureConsumesNothingAndKeepsTheSession()
+ public void Encode_NonPartitionOpsIgnoreTheCallerNamespace()
{
var session = BoundSession();
- var payload = VsrTestPayloads.ConsumerOffset(VsrTestPayloads.NumericIdentifier(4),
- VsrTestPayloads.NumericIdentifier(5), null);
- var header = new byte[VsrHeader.HEADER_SIZE];
-
- var exception = Assert.Throws<IggyInvalidStatusCodeException>(() =>
- VsrHeader.EncodeRequestHeader(header, session, CommandCodes.STORE_CONSUMER_OFFSET_CODE, payload));
- Assert.Equal(VsrError.INVALID_IDENTIFIER, exception.StatusCode);
- Assert.True(session.IsBound);
- Assert.Equal(1UL, session.RequestCounter);
-
- VsrHeader.EncodeRequestHeader(header, session, CommandCodes.CREATE_STREAM_CODE, [1]);
+ var metadataHeader = Encode(session, CommandCodes.CREATE_STREAM_CODE, 1, out _, VsrNamespace.Pack(1, 2, 3));
+ var nonReplicatedHeader = Encode(session, CommandCodes.PING_CODE, 0, out _, VsrNamespace.Pack(1, 2, 3));
- Assert.Equal(1UL, ReadUInt64(header, VsrHeader.REQUEST_ID_OFFSET));
+ Assert.Equal(0UL, ReadUInt64(metadataHeader, VsrHeader.REQUEST_NAMESPACE_OFFSET));
+ Assert.Equal(0UL, ReadUInt64(nonReplicatedHeader, VsrHeader.REQUEST_NAMESPACE_OFFSET));
}
[Fact]
@@ -81,7 +75,7 @@ public sealed class VsrHeaderTests
var session = new ConsensusSession(7);
var payload = LoginRegister.Serialize("admin", "secret");
- var header = Encode(session, CommandCodes.LOGIN_REGISTER_CODE, payload, out var totalSize);
+ var header = Encode(session, CommandCodes.LOGIN_REGISTER_CODE, payload.Length, out var totalSize);
Assert.Equal(VsrHeader.HEADER_SIZE + payload.Length, totalSize);
Assert.Equal((uint)totalSize, ReadUInt32(header, VsrHeader.SIZE_OFFSET));
@@ -100,7 +94,7 @@ public sealed class VsrHeaderTests
{
var session = new ConsensusSession(new UInt128(0xAABB_CCDD_EEFF_0011, 0x1122_3344_5566_7788));
- var header = Encode(session, CommandCodes.PING_CODE, [], out _);
+ var header = Encode(session, CommandCodes.PING_CODE, 0, out _);
Assert.Equal(0x1122_3344_5566_7788UL, ReadUInt64(header, VsrHeader.REQUEST_CLIENT_OFFSET));
Assert.Equal(0xAABB_CCDD_EEFF_0011UL, ReadUInt64(header, VsrHeader.REQUEST_CLIENT_OFFSET + 8));
@@ -111,7 +105,7 @@ public sealed class VsrHeaderTests
{
var session = BoundSession();
- var header = Encode(session, CommandCodes.PING_CODE, [], out _);
+ var header = Encode(session, CommandCodes.PING_CODE, 0, out _);
Assert.Equal((byte)VsrOperation.NonReplicated, header[VsrHeader.REQUEST_OPERATION_OFFSET]);
Assert.Equal(1UL, ReadUInt64(header, VsrHeader.REQUEST_ID_OFFSET));
@@ -126,7 +120,7 @@ public sealed class VsrHeaderTests
{
var session = BoundSession();
- var header = Encode(session, 9999, [], out _);
+ var header = Encode(session, 9999, 0, out _);
Assert.Equal((byte)VsrOperation.NonReplicated, header[VsrHeader.REQUEST_OPERATION_OFFSET]);
Assert.Equal(9999u, ReadUInt32(header, VsrHeader.REQUEST_RESERVED_OFFSET));
@@ -137,7 +131,7 @@ public sealed class VsrHeaderTests
{
var session = new ConsensusSession(1);
- var header = Encode(session, CommandCodes.PING_CODE, [], out _);
+ var header = Encode(session, CommandCodes.PING_CODE, 0, out _);
Assert.Equal(0UL, ReadUInt64(header, VsrHeader.REQUEST_SESSION_OFFSET));
}
@@ -147,7 +141,7 @@ public sealed class VsrHeaderTests
{
var session = BoundSession();
- var header = Encode(session, CommandCodes.CREATE_STREAM_CODE, [1, 2, 3], out _);
+ var header = Encode(session, CommandCodes.CREATE_STREAM_CODE, 3, out _);
Assert.Equal((byte)VsrOperation.CreateStream, header[VsrHeader.REQUEST_OPERATION_OFFSET]);
Assert.Equal(1UL, ReadUInt64(header, VsrHeader.REQUEST_ID_OFFSET));
@@ -161,7 +155,7 @@ public sealed class VsrHeaderTests
{
var session = BoundSession();
- var header = Encode(session, CommandCodes.LOGOUT_USER_CODE, [], out _);
+ var header = Encode(session, CommandCodes.LOGOUT_USER_CODE, 0, out _);
Assert.Equal((byte)VsrOperation.Logout, header[VsrHeader.REQUEST_OPERATION_OFFSET]);
Assert.Equal(1UL, ReadUInt64(header, VsrHeader.REQUEST_ID_OFFSET));
@@ -173,9 +167,8 @@ public sealed class VsrHeaderTests
public void Encode_PartitionOpDoesNotAdvanceTheCounter()
{
var session = BoundSession();
- var payload = VsrTestPayloads.SendMessagesToPartition(2, 3, 4);
- var header = Encode(session, CommandCodes.SEND_MESSAGES_CODE, payload, out _);
+ var header = Encode(session, CommandCodes.SEND_MESSAGES_CODE, 16, out _, VsrNamespace.Pack(2, 3, 4));
Assert.Equal((byte)VsrOperation.SendMessages, header[VsrHeader.REQUEST_OPERATION_OFFSET]);
Assert.Equal(1UL, ReadUInt64(header, VsrHeader.REQUEST_ID_OFFSET));
@@ -189,7 +182,7 @@ public sealed class VsrHeaderTests
var session = new ConsensusSession(1);
var exception = Assert.Throws<IggyInvalidStatusCodeException>(() =>
- Encode(session, CommandCodes.CREATE_STREAM_CODE, [1], out _));
+ Encode(session, CommandCodes.CREATE_STREAM_CODE, 1, out _));
Assert.Equal(VsrError.UNAUTHENTICATED, exception.StatusCode);
Assert.Equal(1UL, session.RequestCounter);
@@ -202,7 +195,7 @@ public sealed class VsrHeaderTests
Array.Fill(header, (byte)0xFF);
var session = BoundSession();
- VsrHeader.EncodeRequestHeader(header, session, CommandCodes.CREATE_STREAM_CODE, [1]);
+ VsrHeader.EncodeRequestHeader(header, session, CommandCodes.CREATE_STREAM_CODE, 0, 1);
Assert.Equal(0u, ReadUInt32(header, VsrHeader.REQUEST_RESERVED_OFFSET));
Assert.All(header[..VsrHeader.SIZE_OFFSET], stale => Assert.Equal(0, stale));
@@ -214,7 +207,8 @@ public sealed class VsrHeaderTests
var session = BoundSession();
Assert.Throws<ArgumentException>(() =>
- VsrHeader.EncodeRequestHeader(new byte[VsrHeader.HEADER_SIZE - 1], session, CommandCodes.PING_CODE, []));
+ VsrHeader.EncodeRequestHeader(new byte[VsrHeader.HEADER_SIZE - 1], session, CommandCodes.PING_CODE, 0,
+ 0));
}
[Fact]
diff --git a/foreign/csharp/Iggy_SDK_Tests/VsrTests/VsrNamespaceTests.cs b/foreign/csharp/Iggy_SDK_Tests/VsrTests/VsrNamespaceTests.cs
index 3ce709560..5527e6262 100644
--- a/foreign/csharp/Iggy_SDK_Tests/VsrTests/VsrNamespaceTests.cs
+++ b/foreign/csharp/Iggy_SDK_Tests/VsrTests/VsrNamespaceTests.cs
@@ -39,6 +39,57 @@ public sealed class VsrNamespaceTests
Assert.True(VsrNamespace.METADATA_CONSENSUS_NAMESPACE > packedMax);
}
+ [Fact]
+ public void ForPartition_PacksNumericIdentifiers()
+ {
+ Assert.Equal(VsrNamespace.Pack(2, 3, 4),
+ VsrNamespace.ForPartition(Identifier.Numeric(2), Identifier.Numeric(3), 4));
+ }
+
+ [Fact]
+ public void ForPartition_NamedIdentifierResolvesToZero()
+ {
+ Assert.Equal(0UL, VsrNamespace.ForPartition(Identifier.String("orders"), Identifier.Numeric(3), 4));
+ Assert.Equal(0UL, VsrNamespace.ForPartition(Identifier.Numeric(2), Identifier.String("events"), 4));
+ }
+
+ [Theory]
+ [InlineData((uint)VsrNamespace.MAX_STREAMS, 3u, 4u)]
+ [InlineData(2u, (uint)VsrNamespace.MAX_TOPICS, 4u)]
+ [InlineData(2u, 3u, (uint)VsrNamespace.MAX_PARTITIONS)]
+ public void ForPartition_RejectsIdentifiersOutsideThePackableRange(uint streamId, uint topicId, uint partitionId)
+ {
+ var exception = Assert.Throws<IggyInvalidStatusCodeException>(() =>
+ VsrNamespace.ForPartition(Identifier.Numeric(streamId), Identifier.Numeric(topicId), partitionId));
+
+ Assert.Equal(VsrError.INVALID_IDENTIFIER, exception.StatusCode);
+ }
+
+ [Theory]
+ [InlineData((byte)VsrOperation.Register)]
+ [InlineData((byte)VsrOperation.Logout)]
+ public void ForOperation_ControlPlaneOpsTargetTheMetadataReplica(byte operation)
+ {
+ Assert.Equal(VsrNamespace.METADATA_CONSENSUS_NAMESPACE,
+ VsrNamespace.ForOperation((VsrOperation)operation, VsrNamespace.Pack(1, 2, 3)));
+ }
+
+ [Fact]
+ public void ForOperation_MetadataAndNonReplicatedOpsIgnoreThePartitionNamespace()
+ {
+ Assert.Equal(0UL, VsrNamespace.ForOperation(VsrOperation.CreateStream, VsrNamespace.Pack(1, 2, 3)));
+ Assert.Equal(0UL, VsrNamespace.ForOperation(VsrOperation.NonReplicated, VsrNamespace.Pack(1, 2, 3)));
+ }
+
+ [Fact]
+ public void ForOperation_PartitionOpsCarryTheCallerNamespace()
+ {
+ Assert.Equal(VsrNamespace.Pack(1, 2, 3),
+ VsrNamespace.ForOperation(VsrOperation.SendMessages, VsrNamespace.Pack(1, 2, 3)));
+ Assert.Equal(VsrNamespace.Pack(1, 2, 3),
+ VsrNamespace.ForOperation(VsrOperation.StoreConsumerOffset, VsrNamespace.Pack(1, 2, 3)));
+ }
+
[Theory]
[InlineData((byte)VsrOperation.Register)]
[InlineData((byte)VsrOperation.Logout)]
@@ -160,7 +211,7 @@ public sealed class VsrNamespaceTests
[Fact]
public void ForRequest_ConsumerOffsetWithoutAPartitionIsRejectedEvenWhenTheBodyEndsThere()
{
- byte[] full = VsrTestPayloads.ConsumerOffset(VsrTestPayloads.NumericIdentifier(4),
+ var full = VsrTestPayloads.ConsumerOffset(VsrTestPayloads.NumericIdentifier(4),
VsrTestPayloads.NumericIdentifier(5), null);
// Drop the four padding bytes that follow the absent-partition flag.