fix(dcl): advance the session generation when a reconnect attempt is issued, not when its result is applied (review 2026-10-21 02/C1 follow-up)
This commit is contained in:
@@ -402,7 +402,7 @@ The DCL manages connection lifecycle automatically:
|
||||
3. **Connection state transitions**: The DCL tracks each connection's state as `connected`, `disconnected`, or `reconnecting`. All transitions are logged to Site Event Logging (event type `connection`): the initial connect (`Connection established on … endpoint`, Info — added by review 2026-10-21, it previously logged nothing), connection lost / reconfigured (`Warning` / `Info`), restored on reconnect (`Info`) and failover (`Warning`).
|
||||
4. **Transparent re-subscribe**: On successful reconnection, the DCL automatically re-establishes all previously active subscriptions for that connection. Instance Actors require no action — they simply see quality return to `good` as fresh values arrive from restored subscriptions.
|
||||
5. **Clean client replacement on (re)connect**: On every (re)connect the adapter detaches and disposes the previous protocol client (session, keep-alive timers, event loop) before creating a new one — a stale client can neither leak nor signal a disconnect against the new session.
|
||||
6. **Session generation on same-adapter reconnect** (review 2026-10-21, finding C1): a plain drop + reconnect reuses the adapter object, so `SwapAdapter`'s generation bump never runs. On a successful reconnect the actor therefore advances `_adapterGeneration` itself — **before** `ReSubscribeAll` — and re-hooks the adapter's `Disconnected` handler for the new generation (new handler hooked before the old one is removed, so a disconnect in between is never lost). Every subscribe/alarm completion, value callback and seed started against the dead session then takes the existing stale path. That path releases a stale completion's handles against the adapter that created them, **except** when that adapter is still the current one: the handles died with the dead session, and releasing by handle against the adapter's NEW session could tear down a live item. Before the fix a dead-session initial-subscribe completion (in flight up to `SeedOverallTimeout`) was applied against the new session — parking a dead handle that made the live re-subscribe row look redundant and get released (the tag read resolved and stayed silent), or releasing a handle number the new MxGateway session had reused, and a connection-level failure row in it knocked the fresh link back into Reconnecting.
|
||||
6. **Session generation on same-adapter reconnect** (review 2026-10-21, finding C1): a plain drop + reconnect reuses the adapter object, so `SwapAdapter`'s generation bump never runs. Each **reconnect attempt** therefore advances `_adapterGeneration` itself — before its `ConnectAsync` is issued — and re-hooks the adapter's `Disconnected` handler for the new generation (new handler hooked before the old one is removed). The attempt's `ConnectResult`, the new session's `Disconnected` and the `ReSubscribeAll` that follows a success all share that generation, so a disconnect queued right behind a successful result is still acted on (advancing the generation only when the result was applied would have dropped it as stale and wedged the actor Connected on a dead session — the adapter raises `Disconnected` once per session). The initial connect does not advance it: no session has existed yet. Every subscribe/alarm completion, value callback and seed started against the dead session then takes the existing stale path. That path releases a stale completion's handles against the adapter that created them, **except** when that adapter is still the current one: the handles died with the dead session, and releasing by handle against the adapter's NEW session could tear down a live item. Before the fix a dead-session initial-subscribe completion (in flight up to `SeedOverallTimeout`) was applied against the new session — parking a dead handle that made the live re-subscribe row look redundant and get released (the tag read resolved and stayed silent), or releasing a handle number the new MxGateway session had reused, and a connection-level failure row in it knocked the fresh link back into Reconnecting.
|
||||
7. **Session-scoped MxGateway subscription ids** (findings C1 + L1): gateway item handles are plain ints that restart per session, so `MxGatewayDataConnection` hands out ids of the form `"{session}:{handle}"` and only ever sends a handle to the gateway when its session is the live one — a dead session's id is dropped silently instead of unadvising whatever the new session gave that number. The adapter's routing maps are cleared on every `ConnectAsync`, `DisconnectAsync` and `DisposeAsync` (they previously kept every dead session's entries, growing per flap), and a subscribe completing after a reconnect does not repopulate them. An alarm unsubscribe for an id the adapter does not know no longer decrements the shared alarm stream's reference count.
|
||||
|
||||
### Disconnect Detection Pattern
|
||||
|
||||
@@ -269,10 +269,12 @@ public class DataConnectionActor : UntypedActor, IWithTimers
|
||||
/// </para>
|
||||
///
|
||||
/// <para>
|
||||
/// Residual (accepted, and in the safe direction): a disconnect raised by the PREVIOUS
|
||||
/// session but delivered after the next attempt has already cleared the flag is attributed to
|
||||
/// the new attempt, costing one extra reconnect cycle. Being wedged Connected on a dead
|
||||
/// adapter is the failure this trades against.
|
||||
/// Each attempt advances the session generation before it is issued, so a disconnect the
|
||||
/// PREVIOUS session already raised (stamped with its generation) is dropped as stale rather
|
||||
/// than attributed to the new attempt. Residual (accepted, and in the safe direction): a
|
||||
/// previous-session disconnect raised only AFTER the next attempt re-hooked the handler is
|
||||
/// attributed to the new attempt, costing one extra reconnect cycle. Being wedged Connected
|
||||
/// on a dead adapter is the failure this trades against.
|
||||
/// </para>
|
||||
/// </summary>
|
||||
private bool _disconnectDuringConnectAttempt;
|
||||
@@ -408,7 +410,7 @@ public class DataConnectionActor : UntypedActor, IWithTimers
|
||||
switch (message)
|
||||
{
|
||||
case AttemptConnect:
|
||||
HandleAttemptConnect();
|
||||
HandleAttemptConnect(advanceSession: false);
|
||||
break;
|
||||
case ConnectResult result:
|
||||
HandleConnectResult(result);
|
||||
@@ -736,6 +738,9 @@ public class DataConnectionActor : UntypedActor, IWithTimers
|
||||
/// <summary>
|
||||
/// Advances <see cref="_adapterGeneration"/> for a new SESSION on the SAME adapter — a plain
|
||||
/// drop + reconnect, where <see cref="SwapAdapter"/> never runs (review 2026-10-21 02/C1).
|
||||
/// Called by <see cref="HandleAttemptConnect"/> before each RECONNECT attempt is issued, so the
|
||||
/// attempt's result, the new session's <c>Disconnected</c> and everything started after it
|
||||
/// share one generation, and anything started against an earlier session is stale.
|
||||
///
|
||||
/// <para>
|
||||
/// Without it, an initial subscribe issued against the dead session (in flight for up to
|
||||
@@ -751,9 +756,9 @@ public class DataConnectionActor : UntypedActor, IWithTimers
|
||||
/// <para>
|
||||
/// The <c>Disconnected</c> handler stamps the generation it was hooked with, so it is
|
||||
/// re-hooked for the new generation. The new handler is hooked BEFORE the old one is
|
||||
/// removed: a disconnect raised in between is then delivered twice (old stamp dropped as
|
||||
/// stale, new stamp acted on) rather than lost, which would wedge the actor Connected on a
|
||||
/// dead session.
|
||||
/// removed, so no window exists with no handler at all. Because this runs before the
|
||||
/// attempt's <c>ConnectAsync</c>, the only disconnects that can carry the OLD stamp belong
|
||||
/// to the previous (already-dead) session and are correctly dropped as stale.
|
||||
/// </para>
|
||||
/// </summary>
|
||||
private void AdvanceSessionGeneration()
|
||||
@@ -852,7 +857,7 @@ public class DataConnectionActor : UntypedActor, IWithTimers
|
||||
switch (message)
|
||||
{
|
||||
case AttemptConnect:
|
||||
HandleAttemptConnect();
|
||||
HandleAttemptConnect(advanceSession: true);
|
||||
break;
|
||||
case ConnectResult result:
|
||||
HandleReconnectResult(result);
|
||||
@@ -944,13 +949,27 @@ public class DataConnectionActor : UntypedActor, IWithTimers
|
||||
|
||||
// ── Connection Management ──
|
||||
|
||||
private void HandleAttemptConnect()
|
||||
/// <param name="advanceSession">
|
||||
/// True from Reconnecting: a session has existed on this adapter, so the attempt opens a NEW
|
||||
/// one and gets its own generation (see <see cref="AdvanceSessionGeneration"/>). False from
|
||||
/// the initial Connecting state, where no session has ever been established and nothing is
|
||||
/// in flight against one, so the generation the handler and result share is already right.
|
||||
/// </param>
|
||||
private void HandleAttemptConnect(bool advanceSession)
|
||||
{
|
||||
_log.Debug("[{0}] Attempting connection...", _connectionName);
|
||||
// Scope the pending-disconnect flag to THIS attempt. Anything noted before now
|
||||
// belongs to a session that is already over and must not be replayed against the one
|
||||
// about to start.
|
||||
_disconnectDuringConnectAttempt = false;
|
||||
// Every reconnect attempt opens a NEW session, so it gets its own generation (review 2026-10-21
|
||||
// 02/C1), advanced — and the Disconnected handler re-hooked — BEFORE ConnectAsync is
|
||||
// issued. The new session's Disconnected therefore always carries the same generation
|
||||
// as its ConnectResult: a disconnect that lands in the mailbox right behind a successful
|
||||
// result is acted on, not dropped as stale (which would wedge the actor Connected on a
|
||||
// dead session — the adapter raises Disconnected only once per session).
|
||||
if (advanceSession)
|
||||
AdvanceSessionGeneration();
|
||||
var self = Self;
|
||||
// Stamp the generation this attempt belongs to, exactly as every other piped
|
||||
// completion does. A failover or an in-place reconfigure landing while ConnectAsync
|
||||
@@ -1146,12 +1165,9 @@ public class DataConnectionActor : UntypedActor, IWithTimers
|
||||
$"Connection restored on {_activeEndpoint} endpoint", null);
|
||||
}
|
||||
|
||||
// The adapter object survived the drop but its SESSION did not. Advance the
|
||||
// generation BEFORE re-subscribing so every completion, value and alarm callback
|
||||
// started against the dead session takes the existing stale path (review
|
||||
// 2026-10-21 02/C1) — exactly as it would after a swap.
|
||||
AdvanceSessionGeneration();
|
||||
|
||||
// The session generation was already advanced when this attempt was issued
|
||||
// (HandleAttemptConnect), so ReSubscribeAll below runs on the NEW session's
|
||||
// generation and every completion from the dead one takes the stale path.
|
||||
// Transparent re-subscribe — re-establish all active subscriptions
|
||||
ReSubscribeAll();
|
||||
// Re-establish native alarm feeds (source replays a snapshot).
|
||||
|
||||
+81
-2
@@ -187,6 +187,58 @@ public class DataConnectionSameAdapterReconnectTests : TestKit
|
||||
Assert.DoesNotContain("live-2", adapter.ReleasedIds);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The connect-result/disconnect race (adversarial review of the C1 fix). The (re)connect
|
||||
/// succeeds and its <c>ConnectResult</c> is enqueued; before the actor processes it the new
|
||||
/// session dies, so the adapter's once-only <c>Disconnected</c> lands in the mailbox AFTER the
|
||||
/// result. That disconnect belongs to the session the result describes and must drive the
|
||||
/// actor back to Reconnecting — if it were stamped with a generation the result's processing
|
||||
/// then advanced past, it would be dropped as stale and the actor would sit Connected on a
|
||||
/// dead session that never raises Disconnected again.
|
||||
/// </summary>
|
||||
/// <param name="reconnect">False: the initial connect; true: a same-adapter reconnect.</param>
|
||||
[Theory]
|
||||
[InlineData(false)]
|
||||
[InlineData(true)]
|
||||
public void DisconnectQueuedBehindASuccessfulConnectResult_IsNotLost(bool reconnect)
|
||||
{
|
||||
var name = reconnect ? "race-reconnect" : "race-initial";
|
||||
var adapter = new GatedAdapter();
|
||||
var connectGate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
adapter.NextConnectGate = connectGate.Task;
|
||||
var actor = CreateConnectionActor(name, adapter);
|
||||
var expectedConnectCalls = 1;
|
||||
|
||||
if (reconnect)
|
||||
{
|
||||
connectGate.SetResult();
|
||||
AwaitConnectedCount(name, 1);
|
||||
connectGate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
adapter.NextConnectGate = connectGate.Task;
|
||||
adapter.RaiseDisconnected();
|
||||
expectedConnectCalls = 2;
|
||||
}
|
||||
|
||||
AwaitCondition(() => Volatile.Read(ref adapter.ConnectCalls) == expectedConnectCalls, TimeSpan.FromSeconds(5));
|
||||
var disconnectsBefore = HealthTransitions(name, ConnectionHealth.Disconnected);
|
||||
|
||||
// Hold the actor inside a health-report read …
|
||||
using var hold = new ManualResetEventSlim(false);
|
||||
adapter.HoldStatus = hold;
|
||||
actor.Tell(new DataConnectionActor.GetHealthReport());
|
||||
Assert.True(adapter.StatusEntered.Wait(TimeSpan.FromSeconds(5)));
|
||||
|
||||
// … while the connect succeeds (its result queues behind the held message) …
|
||||
connectGate.SetResult();
|
||||
Thread.Sleep(300);
|
||||
// … and the new session dies at once, queuing its disconnect behind the result.
|
||||
adapter.RaiseDisconnected();
|
||||
hold.Set();
|
||||
|
||||
AwaitCondition(() => HealthTransitions(name, ConnectionHealth.Disconnected) > disconnectsBefore,
|
||||
TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(20));
|
||||
}
|
||||
|
||||
/// <summary>One gated <c>SubscribeBatchAsync</c> call.</summary>
|
||||
internal sealed class PendingBatch
|
||||
{
|
||||
@@ -219,7 +271,33 @@ public class DataConnectionSameAdapterReconnectTests : TestKit
|
||||
|
||||
public readonly ConcurrentQueue<string> ReleasedIds = new();
|
||||
|
||||
public ConnectionHealth Status { get; private set; } = ConnectionHealth.Disconnected;
|
||||
private ConnectionHealth _status = ConnectionHealth.Disconnected;
|
||||
|
||||
/// <summary>When set, the NEXT read of <see cref="Status"/> signals <see cref="StatusEntered"/> and blocks on it — a latch that holds the actor mid-message.</summary>
|
||||
public ManualResetEventSlim? HoldStatus;
|
||||
|
||||
/// <summary>Set once a held <see cref="Status"/> read has started.</summary>
|
||||
public readonly ManualResetEventSlim StatusEntered = new(false);
|
||||
|
||||
/// <summary>When set, the next <see cref="ConnectAsync"/> returns this task instead of completing at once.</summary>
|
||||
public Task? NextConnectGate;
|
||||
|
||||
/// <summary>Number of <see cref="ConnectAsync"/> calls so far.</summary>
|
||||
public int ConnectCalls;
|
||||
|
||||
public ConnectionHealth Status
|
||||
{
|
||||
get
|
||||
{
|
||||
if (Interlocked.Exchange(ref HoldStatus, null) is { } hold)
|
||||
{
|
||||
StatusEntered.Set();
|
||||
hold.Wait(TimeSpan.FromSeconds(10));
|
||||
}
|
||||
return _status;
|
||||
}
|
||||
private set => _status = value;
|
||||
}
|
||||
|
||||
public event Action? Disconnected;
|
||||
|
||||
@@ -244,8 +322,9 @@ public class DataConnectionSameAdapterReconnectTests : TestKit
|
||||
|
||||
public Task ConnectAsync(IDictionary<string, string> connectionDetails, CancellationToken cancellationToken = default)
|
||||
{
|
||||
Interlocked.Increment(ref ConnectCalls);
|
||||
Status = ConnectionHealth.Connected;
|
||||
return System.Threading.Tasks.Task.CompletedTask;
|
||||
return Interlocked.Exchange(ref NextConnectGate, null) ?? System.Threading.Tasks.Task.CompletedTask;
|
||||
}
|
||||
|
||||
public Task DisconnectAsync(CancellationToken cancellationToken = default) =>
|
||||
|
||||
Reference in New Issue
Block a user