fix(mtconnect): stamp the /sample pump with its session generation and supersede a stale one (#602 STAB-20)

Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
This commit is contained in:
Joseph Doherty
2026-10-08 07:24:43 -04:00
parent 71881163df
commit e76df0741a
2 changed files with 135 additions and 1 deletions
@@ -327,6 +327,20 @@ public sealed class MTConnectDriver
/// <summary>The shared <c>/sample</c> pump task. Guarded by <see cref="_subscriptionLock"/>.</summary>
private Task? _pumpTask;
/// <summary>
/// The <see cref="_sessionGeneration"/> the running pump was started against (STAB-20), guarded by
/// <see cref="_subscriptionLock"/> like the rest of the control block. A pump whose generation is
/// not the current one is bound to a client a lifecycle restart has retired.
/// </summary>
private long _pumpGeneration;
/// <summary>
/// Bumped each time <see cref="StartCoreAsync"/> installs a NEW <see cref="AgentSession"/> (not on a
/// <c>with</c>-copy that only advances the cursor or the probe cache). Written under
/// <see cref="_subscriptionLock"/>.
/// </summary>
private long _sessionGeneration;
/// <summary>
/// The Agent answered <c>/sample</c> with something that cannot be streamed at all, so the
/// pump gave up. Latched — reconnecting reproduces the identical answer forever, and a
@@ -1808,7 +1822,23 @@ public sealed class MTConnectDriver
// revived by a resubscribe. The genuinely-unrepeatable case is the latch below, not this.
if (_pumpTask is { IsCompleted: false })
{
return;
if (_pumpGeneration == _sessionGeneration)
{
return;
}
// STAB-20: a Subscribe can land between StopSampleStreamAsync clearing the registry slots and
// the teardown that retires the client — StopSampleStreamAsync is still awaiting the OLD pump
// then — and starts a pump on the session about to be disposed. Treating that pump as "the
// running stream" would leave the freshly-installed session with no live stream at all
// (the stale pump just reads its disposed client as a dropped connection). Retire it: it is
// cancelled and can only fail against its disposed client.
try { _pumpCts?.Cancel(); }
catch (ObjectDisposedException) { /* already going away */ }
_logger.LogInformation(
"MTConnect driver {DriverInstanceId} replaced a /sample pump bound to a retired Agent session.",
_driverInstanceId);
}
if (_streamUnsupported)
@@ -1856,6 +1886,7 @@ public sealed class MTConnectDriver
var options = Volatile.Read(ref _options);
var cts = new CancellationTokenSource();
_pumpCts = cts;
_pumpGeneration = _sessionGeneration;
_pumpTask = Task.Run(
() => RunSampleStreamAsync(
session.Client, options, session.NextSequence, session.InstanceId, republishOnStart, cts.Token),
@@ -2695,6 +2726,11 @@ public sealed class MTConnectDriver
// One publish, so a concurrent lock-free reader sees the new client, the new probe
// cache, and the pump's opening cursor together or none of them.
lock (_subscriptionLock)
{
_sessionGeneration++;
}
Volatile.Write(
ref _session,
new AgentSession(client, probe, current.InstanceId, current.NextSequence));