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:57 UTC
(iggy) branch dotnet-vsr-cleanup created (now 95289da7c)
This is an automated email from the ASF dual-hosted git repository.
lukaszzborek pushed a change to branch dotnet-vsr-cleanup
in repository https://gitbox.apache.org/repos/asf/iggy.git
at 95289da7c refactor(csharp): cleanup tcp connection after vsr implementation
This branch includes the following new commits:
new 95289da7c refactor(csharp): cleanup tcp connection after vsr implementation
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(iggy) 01/01: refactor(csharp): cleanup tcp connection after vsr implementation
Posted by lz...@apache.org.
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.