dohertj2/wwtools
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

mbproxy: Wave 1 fixes from 2026-05-14 code review

Browse Source 
Resolves the four critical correctness defects + the ShutdownCoordinator
double-stop ordering bug called out in codereviews/2026-05-14/Overview.md.
Tests: 362 pass / 0 fail (baseline 358 + 4 new W1 regression tests).

W1.1 — Context swap on running multiplexer.
  PlcMultiplexer._ctx becomes volatile with a new ReplaceContext() method
  that re-registers the cache stats provider on the (preserved) counters.
  PlcListener exposes its multiplexer; PlcListenerSupervisor.ReplaceContextAsync
  swaps the running mux first, then disposes the old cache. Hot-reload
  tag-list changes and the cache-flush-on-reload contract now actually take
  effect on the next PDU instead of waiting for the next listener fault.

W1.2 — Coalescing factory leak.
  When the InFlightByKey factory soft-fails (allocator saturation or duplicate
  TxId), the cleanup path now TryRemoves the stub and walks every party on it
  (including late attachers) to deliver Modbus exception 0x04. Previously
  only the leader got the exception; late attachers waited forever for a
  response that no backend round-trip would ever fire.

W1.3 — Backend-reader head-of-line block.
  UpstreamPipe gains TrySendResponse for non-blocking enqueue. The per-PLC
  backend reader's fan-out loop uses it instead of awaiting SendResponseAsync,
  so a wedged upstream's full bounded response channel can no longer stall
  the single backend reader and starve every other client on that PLC. New
  responseDropForFullUpstream counter on ProxyCounters / CounterSnapshot
  records the drops.

W1.4 — Stranded outbound frames after cascade.
  TearDownBackendAsync acquires _connectGate and drains any frames left in
  _outboundChannel after the writer task faulted/cancelled, releasing their
  proxy TxIds back to the allocator. Without this, a fresh
  EnsureBackendConnectedAsync racing the cascade would send stranded frames
  with old TxIds onto the new backend socket; the responses would arrive
  with no correlation entry and the upstream peers would hang on the
  watchdog until BackendRequestTimeoutMs.

W1.5 — Delete ShutdownCoordinator (Option B).
  Drain logic moved into ProxyWorker.StopAsync. AdminEndpointHost is no
  longer registered as IHostedService; ProxyWorker drives its lifecycle
  directly so admin starts after listeners are bound and stops AFTER the
  in-flight drain (the design's documented contract). Admin is resolved
  lazily in ExecuteAsync to break the circular DI graph
  (Admin -> StatusSnapshotBuilder -> ProxyWorker). GracefulShutdownTimeoutMs
  is now read fresh from IOptionsMonitor.CurrentValue at stop time, so a
  hot-reloaded value is honoured. Removes ShutdownCoordinator + tests.

New tests:
  PlcMultiplexerTests.ReplaceContext_NewTagMap_VisibleOnNextPdu
  PlcMultiplexerTests.ReplaceContext_NewCache_NextReadGoesToBackend_NotOldCache
  UpstreamPipeTests.TrySendResponse_WhenChannelFull_ReturnsFalse_WithoutBlocking
  UpstreamPipeTests.TrySendResponse_AfterDispose_ReturnsFalse

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Joseph Doherty
2026-05-14 05:16:13 -04:00
parent f2c6669444
commit ce32c5cee8
14 changed files with 614 additions and 532 deletions
Download Patch File Download Diff File Expand all files Collapse all files
mbproxy/src/Mbproxy/Proxy/Multiplexing/PlcMultiplexer.cs
+164 -69
View File
@@ -47,7 +47,12 @@ internal sealed class PlcMultiplexer : IAsyncDisposable, IMultiplexCountersProvi
private readonly PlcOptions _plc;
private readonly ConnectionOptions _connectionOptions;
private readonly IPduPipeline _pipeline;
private readonly PerPlcContext _ctx;
// Phase 12 (W1.1) — `_ctx` is volatile so a hot-reload reseat can swap it on the running
// multiplexer. Each method that uses the context snapshots it into a local at the start
// of the operation so a single PDU sees a consistent (TagMap, Cache) pair even if the
// swap fires mid-PDU. ReplaceContext is the single mutator.
private volatile PerPlcContext _ctx;
private readonly ILogger<PlcMultiplexer> _logger;
private readonly ResiliencePipeline? _backendConnectPipeline;
// Phase 10: live read-coalescing config accessor. The accessor is read per-PDU on the
@@ -145,6 +150,35 @@ internal sealed class PlcMultiplexer : IAsyncDisposable, IMultiplexCountersProvi
_pipes[pipe.Id] = pipe;
}
/// <summary>
/// Phase 12 (W1.1) — atomically swaps the per-PLC context on a running multiplexer.
/// Called by <see cref="Supervision.PlcListenerSupervisor.ReplaceContextAsync"/> when a
/// hot-reload tag-list change is applied to a PLC whose listener is already bound.
///
/// <para>The new context's tag map and (optional) response cache become visible on the
/// next PDU through the volatile <c>_ctx</c> field. Counters are PRESERVED across reseat
/// (the supervisor builds the new context with the running counters), so we only need
/// to re-register the cache stats provider — the multiplex provider already points at
/// this same instance.</para>
///
/// <para>Existing per-call snapshots of the old context held by in-flight PDUs (via
/// <c>WithCurrentRequest</c>) finish on the old map. New PDUs after this call see the
/// new map. Per the design contract a one-PDU "old map" tail is acceptable; partial-BCD
/// rewrites mid-request would be worse.</para>
/// </summary>
public void ReplaceContext(PerPlcContext newContext)
{
if (_disposed) return;
_ctx = newContext;
// Re-register the cache stats provider on the (preserved) counters so the status
// page sees the new cache's count/bytes immediately. Pass null when the new context
// opted out of caching to clear any stale provider from the previous context.
newContext.Counters.SetCacheStatsProvider(
newContext.Cache is not null ? new CacheStatsAdapter(newContext.Cache) : null);
}
/// <summary>
/// Starts the read+write tasks for <paramref name="pipe"/> and returns a task that
/// completes when the pipe's read loop ends. The multiplexer detaches the pipe when
@@ -284,73 +318,98 @@ internal sealed class PlcMultiplexer : IAsyncDisposable, IMultiplexCountersProvi
private async Task TearDownBackendAsync(string reason, bool cascadeUpstreams)
{
Socket? oldSocket;
CancellationTokenSource? oldCts;
Task? writer, reader;
lock (_backendLock)
// Phase 12 (W1.4) — serialise tear-down vs connect-up via the connect gate. Without
// this, a fresh EnsureBackendConnectedAsync racing with the channel drain below
// could see stranded frames sent on its new socket with old (already-released) TxIds,
// producing orphaned responses that hang upstream peers via the watchdog.
await _connectGate.WaitAsync().ConfigureAwait(false);
try
{
oldSocket = _backendSocket;
oldCts = _backendCts;
writer = _backendWriterTask;
reader = _backendReaderTask;
_backendSocket = null;
_backendCts = null;
_backendWriterTask = null;
_backendReaderTask = null;
}
if (oldSocket is null && oldCts is null) return;
try { oldCts?.Cancel(); } catch { /* best effort */ }
try { oldSocket?.Shutdown(SocketShutdown.Both); } catch { /* already closed */ }
try { oldSocket?.Dispose(); } catch { /* best effort */ }
// Drain correlation map; cascade-close every interested upstream pipe.
var dropped = _correlation.DrainAll();
var cascadeIds = new HashSet<Guid>();
foreach (var kvp in dropped)
{
_allocator.Release(kvp.Key);
foreach (var party in kvp.Value.InterestedParties)
cascadeIds.Add(party.Pipe.Id);
}
// Phase 10 — also drain the in-flight-by-key map so a brand-new identical request
// through the freshly-reconnected backend is treated as a miss (no stale entries
// outlive the backend they were destined for).
_inFlightByKey.DrainAll();
int upstreamCount = 0;
if (cascadeUpstreams)
{
// Close every attached pipe that had a request in flight; the others will
// simply re-issue on next request through a fresh backend connect.
// Per the design doc, ALL attached upstreams cascade on backend disconnect.
upstreamCount = _pipes.Count;
// Snapshot keys before disposal modifies the dictionary indirectly.
var pipeList = _pipes.Values.ToArray();
foreach (var pipe in pipeList)
Socket? oldSocket;
CancellationTokenSource? oldCts;
Task? writer, reader;
lock (_backendLock)
{
try { await pipe.DisposeAsync().ConfigureAwait(false); }
catch { /* best effort */ }
oldSocket = _backendSocket;
oldCts = _backendCts;
writer = _backendWriterTask;
reader = _backendReaderTask;
_backendSocket = null;
_backendCts = null;
_backendWriterTask = null;
_backendReaderTask = null;
}
_pipes.Clear();
_ctx.Counters.AddDisconnectCascades(upstreamCount);
if (oldSocket is null && oldCts is null) return;
try { oldCts?.Cancel(); } catch { /* best effort */ }
try { oldSocket?.Shutdown(SocketShutdown.Both); } catch { /* already closed */ }
try { oldSocket?.Dispose(); } catch { /* best effort */ }
// Drain correlation map; cascade-close every interested upstream pipe.
var dropped = _correlation.DrainAll();
foreach (var kvp in dropped)
{
_allocator.Release(kvp.Key);
}
// Phase 10 — also drain the in-flight-by-key map so a brand-new identical request
// through the freshly-reconnected backend is treated as a miss (no stale entries
// outlive the backend they were destined for).
_inFlightByKey.DrainAll();
int upstreamCount = 0;
if (cascadeUpstreams)
{
// Close every attached pipe that had a request in flight; the others will
// simply re-issue on next request through a fresh backend connect.
// Per the design doc, ALL attached upstreams cascade on backend disconnect.
upstreamCount = _pipes.Count;
// Snapshot keys before disposal modifies the dictionary indirectly.
var pipeList = _pipes.Values.ToArray();
foreach (var pipe in pipeList)
{
try { await pipe.DisposeAsync().ConfigureAwait(false); }
catch { /* best effort */ }
}
_pipes.Clear();
_ctx.Counters.AddDisconnectCascades(upstreamCount);
}
// Phase 12 (W1.4) — drain any stranded frames left in the outbound channel by
// the writer task that just faulted/cancelled. Released their proxy TxIds back
// to the allocator. By the time we reach this line the writer has stopped
// reading from the channel (cancelled CTS) and the upstream pipes have been
// cascaded (no more enqueues), so the channel state is stable.
int strandedDropped = 0;
while (_outboundChannel.Reader.TryRead(out byte[]? stranded))
{
if (stranded.Length >= 2)
{
ushort strandedTxId = (ushort)((stranded[0] << 8) | stranded[1]);
_allocator.Release(strandedTxId);
}
strandedDropped++;
}
// Best-effort join.
try { if (writer is not null) await writer.WaitAsync(TimeSpan.FromSeconds(2)).ConfigureAwait(false); } catch { /* swallow */ }
try { if (reader is not null) await reader.WaitAsync(TimeSpan.FromSeconds(2)).ConfigureAwait(false); } catch { /* swallow */ }
oldCts?.Dispose();
if (upstreamCount > 0 || dropped.Count > 0 || strandedDropped > 0)
MultiplexerLogEvents.BackendDisconnected(_logger, _plc.Name, upstreamCount, dropped.Count + strandedDropped, reason);
}
finally
{
_connectGate.Release();
}
// Best-effort join.
try { if (writer is not null) await writer.WaitAsync(TimeSpan.FromSeconds(2)).ConfigureAwait(false); } catch { /* swallow */ }
try { if (reader is not null) await reader.WaitAsync(TimeSpan.FromSeconds(2)).ConfigureAwait(false); } catch { /* swallow */ }
oldCts?.Dispose();
if (upstreamCount > 0 || dropped.Count > 0)
MultiplexerLogEvents.BackendDisconnected(_logger, _plc.Name, upstreamCount, dropped.Count, reason);
}
// ── Backend writer / reader tasks ─────────────────────────────────────────
@@ -513,6 +572,13 @@ internal sealed class PlcMultiplexer : IAsyncDisposable, IMultiplexCountersProvi
// Phase 9: always exactly one party. Phase 10: N parties (read coalescing).
// Note: the InFlightByKey TryRemove above (for FC03/FC04) guarantees no
// further attaches can occur — the parties list is now a stable snapshot.
//
// Phase 12 (W1.3) — non-blocking fan-out via `TrySendResponse`. The
// single backend reader task must NEVER `await` a per-upstream channel
// write: a wedged upstream (full bounded response channel) would otherwise
// stall the reader and starve every other client on this PLC. A drop here
// is recorded via `responseDropForFullUpstream`; the wedged upstream loses
// its own response and will be reaped by its own socket-close path.
foreach (var party in inFlight.InterestedParties)
{
if (!party.Pipe.IsAlive)
@@ -542,7 +608,10 @@ internal sealed class PlcMultiplexer : IAsyncDisposable, IMultiplexCountersProvi
outFrame[0] = (byte)(party.OriginalTxId >> 8);
outFrame[1] = (byte)(party.OriginalTxId & 0xFF);
await party.Pipe.SendResponseAsync(outFrame, ct).ConfigureAwait(false);
if (!party.Pipe.TrySendResponse(outFrame))
{
_ctx.Counters.IncrementResponseDropForFullUpstream();
}
}
}
@@ -723,12 +792,38 @@ internal sealed class PlcMultiplexer : IAsyncDisposable, IMultiplexCountersProvi
if (inFlightForSend is null)
{
// The factory hit the allocator-saturation path or a duplicate-key race.
// Surface a Modbus exception 04 to the upstream and clean up.
// Phase 12 (W1.2) — the factory hit the allocator-saturation path or a
// duplicate-key race and stored a stub `InFlightRequest` under `key`. Late
// attachers may have joined the stub between the factory call and this
// cleanup; we must deliver the saturation exception to ALL of them, not just
// the leader, otherwise the late attachers wait forever for a response that
// never comes (the stub has no proxy TxId, so no backend round-trip will
// ever fire).
MultiplexerLogEvents.Saturated(_logger, _plc.Name, pipe.RemoteEp?.ToString() ?? "?");
byte[] excFrame = BuildExceptionFrame(originalTxId, unitId, fcByte, exceptionCode: 4);
_inFlightByKey.TryRemove(key, out _);
await pipe.SendResponseAsync(excFrame, ct).ConfigureAwait(false);
if (_inFlightByKey.TryRemove(key, out var stub))
{
foreach (var party in stub.InterestedParties)
{
byte[] excFrame = BuildExceptionFrame(party.OriginalTxId, unitId, fcByte, exceptionCode: 4);
try
{
await party.Pipe.SendResponseAsync(excFrame, ct).ConfigureAwait(false);
}
catch
{
// Best-effort delivery. A dead pipe will be collected by its own
// socket close path; nothing more we can do here.
}
}
}
else
{
// The stub was already removed by another path (extremely unlikely, but
// defensive). Surface the exception to the original requester.
byte[] excFrame = BuildExceptionFrame(originalTxId, unitId, fcByte, exceptionCode: 4);
await pipe.SendResponseAsync(excFrame, ct).ConfigureAwait(false);
}
return;
}