refactor(csharp): cleanup tcp connection after vsr implementation

foreign/csharp/Iggy_SDK/Consumers/IggyConsumer.Rented.cs

@@ -102,9 +102,10 @@ await _rentedChannel.Writer.WriteAsync(new ReceivedRentedMessage
     /// </summary>
     protected async Task PollRentedMessagesAsync(CancellationToken ct)
     {
-        if (!_joinedConsumerGroup)
+        if (!_joinedConsumerGroup || !IsGroupMembershipCurrent())
         {
             LogConsumerGroupNotJoinedYetSkippingPolling();
+            await TryRecoverGroupMembershipAsync(ct);
             return;
         }
 

foreign/csharp/Iggy_SDK/Consumers/IggyConsumer.cs

@@ -23,6 +23,7 @@
 using Apache.Iggy.IggyClient;
 using Apache.Iggy.Kinds;
 using Apache.Iggy.Utils;
+using Apache.Iggy.Vsr;
 using Microsoft.Extensions.Logging;
 
 namespace Apache.Iggy.Consumers;
@@ -33,6 +34,13 @@ namespace Apache.Iggy.Consumers;
 /// </summary>
 public partial class IggyConsumer : IAsyncDisposable
 {
+    /// <summary>
+    ///     Backoff between poll iterations that found the group membership missing. It bounds the retry rate of
+    ///     the poll-side rejoin below and keeps the receive loop from spinning while the membership is
+    ///     unrecoverable (client disconnected, group deleted).
+    /// </summary>
+    private const int GroupRejoinRetryDelayMs = 1_000;
+
     private readonly Channel<ReceivedMessage> _channel;
     private readonly IIggyClient _client;
     private readonly IggyConsumerConfig _config;
@@ -46,6 +54,13 @@ public partial class IggyConsumer : IAsyncDisposable
     private int _disposeState;
     private volatile bool _isInitialized;
     private volatile bool _joinedConsumerGroup;
+
+    /// <summary>
+    ///     Consensus session generation the group membership was established under. A later generation means the
+    ///     server session that held the membership is gone and the group must be rejoined.
+    /// </summary>
+    private ulong _joinedSessionGeneration;
+
     private long _lastPolledAtMs;
 
     /// <summary>Whether this consumer has been initialized via <see cref="InitAsync" />.</summary>
@@ -263,8 +278,13 @@ private async Task InitializeConsumerGroupAsync(CancellationToken ct = default)
             return;
         }
 
+        // Captured before the join: a session reset racing the join lands the membership on a later generation,
+        // and the mismatch triggers one redundant (idempotent) rejoin instead of a missed one.
+        var sessionGeneration = (_client as ISessionGenerationProvider)?.SessionGeneration ?? 0;
+
         if (_config.Consumer.Type == ConsumerType.Consumer)
         {
+            Interlocked.Exchange(ref _joinedSessionGeneration, sessionGeneration);
             _joinedConsumerGroup = true;
             return;
         }
@@ -306,6 +326,7 @@ private async Task InitializeConsumerGroupAsync(CancellationToken ct = default)
                 await _client.JoinConsumerGroupAsync(_config.StreamId, _config.TopicId,
                     Identifier.String(_consumerGroupName), ct);
 
+                Interlocked.Exchange(ref _joinedSessionGeneration, sessionGeneration);
                 _joinedConsumerGroup = true;
                 LogConsumerGroupJoined(_consumerGroupName);
             }
@@ -364,9 +385,10 @@ private void ThrowIfAutoCommitWithEncryptor()
     /// </summary>
     private async Task PollMessagesAsync(CancellationToken ct)
     {
-        if (!_joinedConsumerGroup)
+        if (!_joinedConsumerGroup || !IsGroupMembershipCurrent())
         {
             LogConsumerGroupNotJoinedYetSkippingPolling();
+            await TryRecoverGroupMembershipAsync(ct);
             return;
         }
 
@@ -452,6 +474,57 @@ private async Task PollMessagesAsync(CancellationToken ct)
         }
     }
 
+    /// <summary>
+    ///     Whether the session generation the group membership was stamped under is still the transport's
+    ///     current one. Gating the poll here instead of clearing the joined flag on a Disconnected event survives
+    ///     a late event landing after a rejoin already re-stamped the generation: state events are published
+    ///     outside the state lock, so their order is not guaranteed. Runs outside
+    ///     <see cref="_connectionStateSemaphore" />, hence the interlocked read.
+    /// </summary>
+    private bool IsGroupMembershipCurrent()
+    {
+        if (_config.Consumer.Type != ConsumerType.ConsumerGroup
+            || _client is not ISessionGenerationProvider generationProvider)
+        {
+            return true;
+        }
+
+        return generationProvider.SessionGeneration == Interlocked.Read(ref _joinedSessionGeneration);
+    }
+
+    /// <summary>
+    ///     Poll-side rejoin for a membership found missing or stamped under a dead session. The state-event
+    ///     rejoin swallows its failures so the event loop survives them, and the transport suppresses a repeat
+    ///     event for an unchanged state, so without this backstop one failed rejoin would park the consumer for
+    ///     good. The trailing delay keeps the receive loop from spinning while the membership stays gone.
+    /// </summary>
+    private async Task TryRecoverGroupMembershipAsync(CancellationToken ct)
+    {
+        if (_config.Consumer.Type == ConsumerType.ConsumerGroup && _config.JoinConsumerGroup)
+        {
+            await _connectionStateSemaphore.WaitAsync(ct);
+            try
+            {
+                if (!_joinedConsumerGroup || !IsGroupMembershipCurrent())
+                {
+                    _joinedConsumerGroup = false;
+                    await RejoinConsumerGroupOnReconnectionAsync();
+                }
+            }
+            finally
+            {
+                _connectionStateSemaphore.Release();
+            }
+
+            if (_joinedConsumerGroup && IsGroupMembershipCurrent())
+            {
+                return;
+            }
+        }
+
+        await Task.Delay(GroupRejoinRetryDelayMs, ct);
+    }
+
     /// <summary>
     ///     Implements polling interval throttling to avoid excessive server requests.
     ///     Uses monotonic time tracking to ensure proper intervals even with clock adjustments.
@@ -502,9 +575,33 @@ private async Task OnClientConnectionStateChangedAsync(ConnectionStateChangedEve
     {
         LogConnectionStateChanged(e.PreviousState, e.CurrentState);
 
+        if (_config.Consumer.Type == ConsumerType.Consumer)
+        {
+            return;
+        }
+
         await _connectionStateSemaphore.WaitAsync();
         try
         {
+            if (_client is ISessionGenerationProvider generationProvider)
+            {
+                if (e.CurrentState != ConnectionState.Authenticated)
+                {
+                    return;
+                }
+
+                if (_joinedConsumerGroup
+                    && generationProvider.SessionGeneration == Interlocked.Read(ref _joinedSessionGeneration))
+                {
+                    return;
+                }
+
+                _joinedConsumerGroup = false;
+                await RejoinConsumerGroupOnReconnectionAsync();
+
+                return;
+            }
+
             if (e.CurrentState == ConnectionState.Disconnected)
             {
                 _joinedConsumerGroup = false;