refactor(csharp): cleanup tcp connection after vsr implementation - #3858
refactor(csharp): cleanup tcp connection after vsr implementation#3858lukaszzborek wants to merge 5 commits into
Conversation
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## master #3858 +/- ##
============================================
- Coverage 82.70% 82.64% -0.07%
Complexity 1296 1296
============================================
Files 1199 1200 +1
Lines 159941 159840 -101
Branches 129545 129651 +106
============================================
- Hits 132281 132092 -189
+ Misses 24163 24151 -12
- Partials 3497 3597 +100
🚀 New features to boost your workflow:
|
|
If possible, merge #3836 first. Then I will remove things with namespace which I refactoring 😅 |
hubcio
left a comment
hubcio
left a comment
There was a problem hiding this comment.
solid direction overall - classic framing gone, the VsrConnection extraction, typed namespaces and the epoch rejoin are real improvements. blocking items are the probe losing its reconnect, the never-nulled _connection, the fail-open namespace and the edge-based Disconnected clear; the rest is small.
one thing outside the diff: SetConnectionStateAsync (TcpMessageStream.cs:1153) is a non-atomic check-then-set with the publish after the write, reachable from callers holding different locks - concurrent transitions can publish duplicates or a PreviousState that never actually preceded CurrentState. it predates this PR, and fixing it properly means moving the publish behind an ordered queue (holding a lock across the publish is not an option - the handler chain runs synchronously into socket I/O when the semaphores are uncontended, so a sync write fault re-enters the same lock). follow-up material, but it is the substrate under the Disconnected-clear race flagged inline.
| /// 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) |
There was a problem hiding this comment.
if the probe's socket dies, DropVsrConnectionLocked has already reset the session and set Disconnected, but the catch in GetCurrentLeaderNodeAsync turns that into "no leader" and callers carry on: LoginRegisterAsync returns a success AuthResponse on a disconnected, unbound client (with auto-login off nothing ever re-authenticates), and the send loop keeps retrying the dead connection and surfaces UNAUTHENTICATED with no reconnect. let connection exceptions propagate from that catch (e is not OperationCanceledException && !VsrConnection.IsConnectionException(e)) - a dead socket is not "no leader" - and re-check state before LoginRegisterAsync returns success.
| try | ||
| { | ||
| _stream?.Dispose(); | ||
| _connection?.Dispose(); |
There was a problem hiding this comment.
_connection is never set back to null, here or in DropVsrConnectionLocked, so a disposed connection stays reachable behind a Connected state for the whole reconnect window, and a request hitting it can surface VsrRequestOutcomeUnknownException for bytes that never left the process. it also leaves the connection is null branch in SendVsrAttemptAsync dead. set _connection = null after dispose/close at both sites (still under the sending semaphore) - that branch then turns these into NotConnectedException, which the reconnect path already handles.
| // reconnect takes to re-authenticate, so the membership is surrendered up front. | ||
| if (e.CurrentState == ConnectionState.Disconnected) | ||
| { | ||
| _joinedConsumerGroup = false; |
There was a problem hiding this comment.
this clear is still edge-based inside the branch that exists to stop depending on edges. state events can be published concurrently (SetConnectionStateAsync is a plain read-modify-write with the publish after the write), so a late Disconnected can land after the Authenticated arm already rejoined and stamped the new epoch - the flag stays false and polling silently stops until the next Authenticated event, which on a healthy connection never comes. gating the poll on _joinedSessionEpoch == provider.SessionEpoch for group consumers removes the dependence on event arrival; plain consumers must keep the early return above (they never re-stamp), and the epoch read then happens outside _connectionStateSemaphore, so access it with Interlocked.Read/Exchange(ref ulong).
| client.SessionEpoch++; | ||
| await client.RaiseAsync(ConnectionState.Connected, ConnectionState.Authenticated); | ||
|
|
||
| Assert.Equal(2, client.JoinCount); |
There was a problem hiding this comment.
nothing pins that the rejoin re-stamps _joinedSessionEpoch - delete that stamp in the consumer and all four tests stay green while every reconnect triggers a rejoin (a group-wide rebalance each time). raise a second (Connected, Authenticated) here and assert JoinCount is still 2.
|
|
||
| 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); |
There was a problem hiding this comment.
the request header is rented per request and threaded through two signatures, while the reply header is already a per-connection field. a _requestHeaderBuffer field on VsrConnection next to _replyHeaderBuffer drops the rent/return per request and a parameter off both signatures - it is re-encoded at the top of every attempt under the sending lock, so the reuse is safe.
| private readonly Stream _stream; | ||
|
|
||
| /// <summary>The session identity requests on this connection encode from.</summary> | ||
| internal ConsensusSession Session { get; } |
There was a problem hiding this comment.
Session is only read inside this class - a private readonly field is enough.
| _stream.Dispose(); | ||
| } | ||
|
|
||
| internal void Close() |
There was a problem hiding this comment.
Close() and Dispose() are the same operation (Stream.Close() is Dispose()), and TcpMessageStream.Dispose() calls both back to back. one of them can go.
| return; | ||
| } | ||
|
|
||
| if (e.CurrentState == ConnectionState.Disconnected) |
There was a problem hiding this comment.
this fallback is unreachable in-tree: HttpMessageStream.SubscribeConnectionEvents has an empty body and never publishes, and an out-of-tree IIggyClient cannot implement the internal ISessionEpochProvider to take the branch above. the ISessionEpochProvider doc also says HTTP consumers fall back to connection-state edges, which is not true - they get no events at all. either make the interface public, or delete this and document group rejoin as built-in-TCP-only; if it stays, it needs the same plain-consumer carve-out as the epoch arm.
github-actions
Bot
added
S-waiting-on-author
# Conflicts: # foreign/csharp/Iggy_SDK/Vsr/VsrHeader.cs # foreign/csharp/Iggy_SDK/Vsr/VsrNamespace.cs # foreign/csharp/Iggy_SDK_Tests/VsrTests/VsrHeaderTests.cs # foreign/csharp/Iggy_SDK_Tests/VsrTests/VsrNamespaceTests.cs
2 participants
Follow-up cleanup of the .NET SDK TCP transport after the VSR support landed