From ecf1e864641a24f89b6230ae1f72bb862ec01e97 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 22 Oct 2026 15:10:21 -0400 Subject: [PATCH] fix(s7): lifecycle gate + generation-stamped health and publishes (R07:H5) Overlapping InitializeAsync calls (the actor's fire-and-forget init plus a retry tick) could leak an opened PLC connection and orphan a probe loop: no gate guarded the lifecycle, only _gate for I/O. - _lifecycleGate (the Modbus pattern) serialises Initialize / Reinitialize / Shutdown / Dispose; teardown waits a bounded time, then tears down anyway. - Every initialize and teardown starts a lifecycle generation. Health writes (init, read/write passes, poll errors, shutdown) and the init publish carry it and are dropped once it has moved on, so an abandoned init never publishes over a later shutdown and a shutdown's final write never clobbers a newer init. A superseded init disposes what it opened and throws. - The init publish (options, tag tables, Plc, probe loop) happens under _gate; the tag tables are built fresh and swapped by one reference write instead of being cleared in place under live poll passes. - Read/write passes re-check initialization under _gate, so a pass that waited out a teardown cannot reopen a connection on a shut-down driver. - The semaphores are no longer disposed: Dispose can race a running shutdown. Refs #742 Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT --- .../ZB.MOM.WW.OtOpcUa.Driver.S7/S7Driver.cs | 439 +++++++++++++----- .../S7LifecycleGateTests.cs | 143 ++++++ 2 files changed, 467 insertions(+), 115 deletions(-) create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7.Tests/S7LifecycleGateTests.cs diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7/S7Driver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7/S7Driver.cs index 347a0b19..acade74e 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7/S7Driver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7/S7Driver.cs @@ -59,7 +59,7 @@ public sealed class S7Driver _plcFactory = plcFactory ?? new S7PlcFactory(); _logger = logger ?? NullLogger.Instance; _resolver = new EquipmentTagRefResolver( - r => _tagsByRawPath.TryGetValue(r, out var t) ? t : null); + r => TagsByRawPath.TryGetValue(r, out var t) ? t : null); // The polling overlay runs on the shared PollGroupEngine (structural array diff + // drain-before-dispose + capped-exponential failure backoff + caller-token-filtered OCE), // retiring the bespoke S7 poll fork. minInterval 100 ms matches the S7 scan-mailbox floor; @@ -128,13 +128,6 @@ public sealed class S7Driver /// OPC UA StatusCode used for a genuine device fault (CPU error, hardware fault). private const uint StatusBadDeviceFailure = 0x808B0000u; - // v3: the RawPath → definition table. Keyed by RawPath (the v3 tag identity, == def.Name), - // case-sensitive ordinal. Built in InitializeAsync from _options.RawTags via the pure factory. - private readonly Dictionary _tagsByRawPath = new(StringComparer.Ordinal); - // Pre-parsed S7 address per tag, keyed by RawPath (== def.Name). Populated alongside - // _tagsByRawPath at init so a config typo fails fast here rather than as a per-read fault. - private readonly Dictionary _parsedByRawPath = new(StringComparer.Ordinal); - // Resolves a read/write/subscribe fullReference (a RawPath in v3) to a tag definition by a single // authored-table hit; a miss is a miss (the driver surfaces BadNodeIdUnknown, never a parse attempt). private readonly EquipmentTagRefResolver _resolver; @@ -225,19 +218,50 @@ public sealed class S7Driver /// public async Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken) { - // InitializeAsync is reachable on a live driver with no teardown first (the IDriver contract - // allows it, and a reconnect retry can overlap a running attempt). Stop the previous probe loop and drop the previous connection - // before they are overwritten below. Deliberately NOT ShutdownAsync: that also drops _poll and the - // subscriptions, which are owned by the shutdown path (the actor re-subscribes after every successful init). - await ReleaseConnectionAsync().ConfigureAwait(false); - Health = new DriverHealth(DriverState.Initializing, null, null); + // Serialised against every other lifecycle operation: DriverInstanceActor dispatches initialize + // on a fire-and-forget Task.Run and a reconnect retry can start a second attempt while the + // first is still inside OpenAsync. Without the gate attempt A's freshly opened Plc and probe + // loop were overwritten by attempt B and never released (R07:H5) — the Modbus pattern. + await _lifecycleGate.WaitAsync(cancellationToken).ConfigureAwait(false); + try + { + await InitializeCoreAsync(driverConfigJson, cancellationToken).ConfigureAwait(false); + } + finally + { + _lifecycleGate.Release(); + } + } + + /// The lifecycle-gated body of . + /// The driver config. + /// Caller cancellation. + /// A task that completes when the attempt has published, been superseded, or failed. + private async Task InitializeCoreAsync(string driverConfigJson, CancellationToken cancellationToken) + { + // This attempt's generation. Every health write and table/connection publish below is stamped + // with it and dropped once the generation has moved on: an attempt abandoned by its caller can + // still be running when a ShutdownAsync (which only waits a bounded time for the lifecycle + // gate) starts, and it must never publish over it. + var gen = BeginGeneration(); + + // InitializeAsync is reachable on a live driver with no teardown first (the IDriver contract + // allows it). Stop the previous probe loop and drop the previous connection before they are + // replaced below. Deliberately NOT a full teardown: that also drops _poll and the subscriptions, + // which are owned by the shutdown path (the actor re-subscribes after every successful init). + await ReleaseConnectionAsync(gen).ConfigureAwait(false); + TryWriteHealth(gen, new DriverHealth(DriverState.Initializing, null, null)); + + // Re-parse the supplied DriverConfig JSON so a config change delivered through the IDriver + // contract is honoured. An empty / placeholder document (e.g. the "{}" some unit tests pass) + // keeps the constructor-supplied options. Local until published: a superseded attempt must + // not leave its options behind for the live connection to reopen against. + var options = _options; + IS7Plc? plc = null; try { - // Re-parse the supplied DriverConfig JSON so a config change delivered through the - // IDriver contract is honoured. An empty / placeholder document - // (e.g. the "{}" some unit tests pass) keeps the constructor-supplied options. if (HasConfigBody(driverConfigJson)) - _options = S7DriverFactoryExtensions.ParseOptions(_driverInstanceId, driverConfigJson); + options = S7DriverFactoryExtensions.ParseOptions(_driverInstanceId, driverConfigJson); // v3: build the RawPath → definition table from the authored raw tags. Each entry's TagConfig // blob is mapped by the pure factory, which also applies S7TagShapeRules (the address must @@ -246,36 +270,32 @@ public sealed class S7Driver // through ISkippedTagReporter, reads BadNodeIdUnknown), never thrown: these used to be init // guards that faulted every tag on the PLC and made the actor retry the same init forever. // The entry's WriteIdempotent flag is threaded onto the def (it lives on the RawTagEntry). - _tagsByRawPath.Clear(); - _parsedByRawPath.Clear(); - _resolver.Clear(); // no-op in v3 (the resolver reads the live table by closure); kept for symmetry - _skippedTags.BeginRebuild(_options.RawTags.Count); - foreach (var entry in _options.RawTags) + // Built into fresh dictionaries and published by ONE reference swap — a retained poll pass + // reading the old table never sees it half-cleared. + var tags = new Dictionary(StringComparer.Ordinal); + var parsed = new Dictionary(StringComparer.Ordinal); + var skipped = new List<(string RawPath, string Reason)>(); + foreach (var entry in options.RawTags) { if (S7TagDefinitionFactory.FromTagConfig(entry.TagConfig, entry.RawPath, out var def, out var shapeRejection) && S7AddressParser.TryParse(def.Address, out var parsedAddress)) { - _tagsByRawPath[entry.RawPath] = def with { WriteIdempotent = entry.WriteIdempotent }; - _parsedByRawPath[entry.RawPath] = parsedAddress; + tags[entry.RawPath] = def with { WriteIdempotent = entry.WriteIdempotent }; + parsed[entry.RawPath] = parsedAddress; } else { - var reason = shapeRejection ?? "TagConfig did not map to a definition"; - _skippedTags.Skip(entry.RawPath, reason); - _logger.LogWarning( - "S7 tag skipped: {Reason} Driver={DriverInstanceId} RawPath={RawPath}", - reason, _driverInstanceId, entry.RawPath); + skipped.Add((entry.RawPath, shapeRejection ?? "TagConfig did not map to a definition")); } } - _skippedTags.Commit(); // Build the connection through the factory (the same seam EnsureConnectedAsync's reopen // uses) so the read/write timeouts are baked in identically on connect and reconnect. - var plc = _plcFactory.Create(_options); + plc = _plcFactory.Create(options); try { using var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); - cts.CancelAfter(_options.Timeout); + cts.CancelAfter(options.Timeout); await plc.OpenAsync(cts.Token).ConfigureAwait(false); } catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested) @@ -283,88 +303,157 @@ public sealed class S7Driver // The linked CancelAfter fired — a connect TIMEOUT, not a caller cancellation. // Surface it as a TimeoutException so init-time and reconnect-time timeouts report // the same exception type (for consistency); the outer catch still faults health. - try { plc.Dispose(); } catch { } throw new TimeoutException( - $"S7 connect to {_options.Host}:{_options.Port} timed out after " + - $"{(int)_options.Timeout.TotalMilliseconds} ms."); + $"S7 connect to {options.Host}:{options.Port} timed out after " + + $"{(int)options.Timeout.TotalMilliseconds} ms."); } - catch + + // Publish everything this attempt built under _gate (no pass can be mid-flight against a + // half-published state) and only while this attempt is still the current generation. + await _gate.WaitAsync(cancellationToken).ConfigureAwait(false); + try { - // Dispose the partially-constructed connection so a failed init doesn't leak it - // before the outer catch nulls Plc (Plc isn't assigned yet, so it can't). - try { plc.Dispose(); } catch { } - throw; + if (!IsCurrentGeneration(gen)) + { + // The outer finally disposes the unpublished plc; the catch writes no health (the + // generation moved on). Thrown, not returned: a caller must never read "initialized" + // into an attempt that published nothing. + throw new InvalidOperationException( + "S7 initialize was superseded by a newer lifecycle operation before it published; " + + "its connection was discarded."); + } + + _options = options; + Volatile.Write(ref _tables, new TagTables(tags, parsed)); + _skippedTags.BeginRebuild(options.RawTags.Count); + foreach (var (rawPath, reason) in skipped) + { + _skippedTags.Skip(rawPath, reason); + _logger.LogWarning( + "S7 tag skipped: {Reason} Driver={DriverInstanceId} RawPath={RawPath}", + reason, _driverInstanceId, rawPath); + } + _skippedTags.Commit(); + + Plc = plc; + plc = null; // published — no longer this method's to dispose + _plcDead = false; + _connectBackoff.RecordSuccess(); + Volatile.Write(ref _negotiatedPduSize, Plc.MaxPduSize); + _initialized = true; + TryWriteHealth(gen, new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null)); + + // Kick off the probe loop once the connection is up. Initial HostState stays Unknown + // until the first probe tick succeeds — avoids broadcasting a premature Running + // transition before any PDU round-trip has happened. Tracked (not fire-and-forget) so + // a teardown can await it before disposing its CTS; Task.Run gets None so the handle + // is always awaitable — the loop's own token check handles cancellation. + if (options.Probe.Enabled) + { + var probeCts = new CancellationTokenSource(); + _probeCts = probeCts; + _probeTask = Task.Run(() => ProbeLoopAsync(probeCts.Token), CancellationToken.None); + } + } + finally + { + _gate.Release(); } - Plc = plc; - _plcDead = false; - _connectBackoff.RecordSuccess(); - Volatile.Write(ref _negotiatedPduSize, plc.MaxPduSize); - - _initialized = true; - Health = new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null); _logger.LogInformation("S7Driver connected. Driver={DriverInstanceId} Host={Host} CPU={CpuType} PduSize={PduSize} Tags={TagCount}", - _driverInstanceId, _options.Host, _options.CpuType, _negotiatedPduSize, _tagsByRawPath.Count); - - // Kick off the probe loop once the connection is up. Initial HostState stays - // Unknown until the first probe tick succeeds — avoids broadcasting a premature - // Running transition before any PDU round-trip has happened. - if (_options.Probe.Enabled) - { - _probeCts = new CancellationTokenSource(); - // Track the probe Task (not fire-and-forget) so ShutdownAsync can await it - // before disposing _gate / _probeCts. Pass None to Task.Run so - // the delegate always runs and the handle is always awaitable; the loop's own - // token check handles cancellation. - _probeTask = Task.Run(() => ProbeLoopAsync(_probeCts.Token), CancellationToken.None); - } + _driverInstanceId, options.Host, options.CpuType, _negotiatedPduSize, tags.Count); } catch (Exception ex) { - // Clean up a partially-constructed Plc so a retry from the caller doesn't leak - // the TcpClient. Dispose is best-effort and idempotent. - try { Plc?.Dispose(); } catch { } - Plc = null; - Volatile.Write(ref _negotiatedPduSize, 0); - _initialized = false; - _plcDead = false; - Health = new DriverHealth(DriverState.Faulted, null, ex.Message); - _logger.LogError(ex, "S7Driver connect failed. Driver={DriverInstanceId} Host={Host}", _driverInstanceId, _options.Host); + // A superseded attempt reports nothing: the generation that replaced it owns health. + if (TryWriteHealth(gen, new DriverHealth(DriverState.Faulted, null, ex.Message))) + { + _options = options; + _logger.LogError(ex, "S7Driver connect failed. Driver={DriverInstanceId} Host={Host}", _driverInstanceId, options.Host); + } throw; } + finally + { + // A connection this attempt opened but did not publish (failed or superseded) is released + // here, so a retry from the caller never leaks the TcpClient. + if (plc is not null) { try { plc.Dispose(); } catch { /* best-effort */ } } + } } /// public async Task ReinitializeAsync(string driverConfigJson, CancellationToken cancellationToken) { - // InitializeAsync re-parses driverConfigJson, so a config change delivered here is + // Holds the lifecycle gate across BOTH halves so no other lifecycle operation slips between + // them. InitializeCoreAsync re-parses driverConfigJson, so a config change delivered here is // applied in place rather than silently discarded. - await ShutdownAsync(cancellationToken).ConfigureAwait(false); - await InitializeAsync(driverConfigJson, cancellationToken).ConfigureAwait(false); + await _lifecycleGate.WaitAsync(cancellationToken).ConfigureAwait(false); + try + { + await ShutdownCoreAsync().ConfigureAwait(false); + await InitializeCoreAsync(driverConfigJson, cancellationToken).ConfigureAwait(false); + } + finally + { + _lifecycleGate.Release(); + } } /// public async Task ShutdownAsync(CancellationToken cancellationToken) { - // Signal cancellation to the probe + poll loops first, collect their Task handles, - // then await all of them with a bounded timeout BEFORE disposing the shared semaphore - // and CTS objects. Without the drain, a loop iteration mid-_gate would call Release() - // on (or WaitAsync against) a disposed semaphore. + // Bounded wait for the lifecycle gate: a wedged initialize (a connect that never returns, or an + // attempt its caller abandoned) must not turn shutdown into a hang. Past the bound we tear down + // anyway — the generation bump makes the still-running attempt discard instead of publish. + var gated = false; + try { gated = await _lifecycleGate.WaitAsync(LifecycleGateTimeout, cancellationToken).ConfigureAwait(false); } + catch (OperationCanceledException) { /* the caller stopped waiting for the gate — still tear down */ } + if (!gated) + _logger.LogWarning( + "S7 lifecycle gate still held at shutdown; tearing down anyway. Driver={DriverInstanceId}", + _driverInstanceId); + try + { + await ShutdownCoreAsync().ConfigureAwait(false); + } + finally + { + if (gated) _lifecycleGate.Release(); + } + } + + /// + /// The complete teardown: the poll engine (every subscription), the probe loop, the connection, + /// the sticky write fault and the per-generation memos. It is its own generation, so its final + /// health write can never clobber an initialize that started after it. + /// + /// A task that completes once everything is released. + private async Task ShutdownCoreAsync() + { + var gen = BeginGeneration(); // Drain the polling overlay first — the engine cancels each loop and awaits it before - // disposing its CTS (drain-before-dispose), so no poll iteration touches _gate - // after this returns. - await _poll.DisposeAsync().ConfigureAwait(false); - await ReleaseConnectionAsync().ConfigureAwait(false); - _writeFault = null; - _connectBackoff.RecordSuccess(); - Health = new DriverHealth(DriverState.Unknown, Health.LastSuccessfulRead, null); + // disposing its CTS (drain-before-dispose), so no poll iteration touches _gate after this. + try { await _poll.DisposeAsync().ConfigureAwait(false); } + catch (Exception ex) { _logger.LogWarning(ex, "S7 poll engine drain failed. Driver={DriverInstanceId}", _driverInstanceId); } + await ReleaseConnectionAsync(gen).ConfigureAwait(false); + lock (_healthLock) + { + if (_generation != gen) return; // a newer initialize owns the state now + _writeFault = null; + _connectBackoff.RecordSuccess(); + Health = new DriverHealth(DriverState.Unknown, Health.LastSuccessfulRead, null); + } } /// /// Cancels + bounded-joins the probe loop, then disposes its CTS and the PLC connection. The - /// connection part of (everything but the poll engine and health). + /// connection part of a teardown (everything but the poll engine and health). Releases only + /// while is still current — a teardown that ran past the lifecycle gate + /// must not dispose a connection a NEWER initialize already published. /// - private async Task ReleaseConnectionAsync() + /// The generation of the lifecycle operation doing the release. + /// A task that completes once the prior state is released. + private async Task ReleaseConnectionAsync(long gen) { var probeCts = _probeCts; var probeTask = _probeTask; @@ -379,11 +468,6 @@ public sealed class S7Driver catch { /* loop faults are already surfaced via health; teardown continues */ } } - // The probe loop has now observed cancellation and released _gate — safe to dispose the CTS. - probeCts?.Dispose(); - _probeCts = null; - _probeTask = null; - // Swap the connection out under _gate so a Read/Write/poll pass that already holds the gate (they // re-read Plc inside it) can never see a disposed Plc mid-swap. Bounded: a wedged holder must not hang // teardown, in which case we proceed exactly as the un-gated path always did. @@ -395,6 +479,15 @@ public sealed class S7Driver _driverInstanceId, DrainTimeout); try { + if (!IsCurrentGeneration(gen)) return; // superseded: the newer generation released (or owns) this state + + // The probe loop has now observed cancellation and released _gate — safe to dispose the CTS. + if (ReferenceEquals(_probeCts, probeCts)) + { + probeCts?.Dispose(); + _probeCts = null; + _probeTask = null; + } _initialized = false; _plcDead = false; try { Plc?.Dispose(); } catch { /* best-effort — tearing down anyway */ } @@ -407,6 +500,101 @@ public sealed class S7Driver } } + // ---- lifecycle generation ---- + + /// + /// Guards / / + /// / against each other (R07:H5, the + /// Modbus _lifecycleGate pattern). Distinct from , which serialises I/O. + /// Never disposed: a Dispose racing a still-running ShutdownAsync must not make it throw. + /// + private readonly SemaphoreSlim _lifecycleGate = new(1, 1); + + /// + /// Bounded wait the teardown paths take on before tearing down anyway. + /// Internal so a test can shorten it instead of waiting the production 10 s. + /// + internal TimeSpan LifecycleGateTimeout { get; set; } = TimeSpan.FromSeconds(10); + + /// Serialises generation bumps against generation-stamped health writes. + private readonly object _healthLock = new(); + + /// + /// Lifecycle generation: bumped by every initialize and teardown. Health writes and + /// table/connection publishes carry the generation they were computed under and are dropped + /// once it has moved on. Written under . + /// + private long _generation; + + /// Starts a new lifecycle generation. + /// The new generation. + private long BeginGeneration() + { + lock (_healthLock) return ++_generation; + } + + /// The current lifecycle generation. + private long CurrentGeneration => Interlocked.Read(ref _generation); + + /// True while is still the current lifecycle generation. + /// The generation to test. + /// when nothing has superseded it. + private bool IsCurrentGeneration(long gen) => CurrentGeneration == gen; + + /// Writes only while is current. + /// The generation the health was computed under. + /// The health to publish. + /// when written. + private bool TryWriteHealth(long gen, DriverHealth health) + { + lock (_healthLock) + { + if (_generation != gen) return false; + Health = health; + return true; + } + } + + /// + /// Derives and writes a new health from the current one, atomically, only while + /// is current — for the passes that preserve LastSuccessfulRead. + /// + /// The generation the update was computed under. + /// Maps the current health to the new one; returning null writes nothing. + /// when written. + private bool TryUpdateHealth(long gen, Func update) + { + lock (_healthLock) + { + if (_generation != gen) return false; + if (update(Health) is not { } next) return false; + Health = next; + return true; + } + } + + /// The published RawPath → definition and RawPath → parsed-address tables, swapped as one. + /// RawPath → definition. + /// RawPath → pre-parsed address. + private sealed record TagTables( + IReadOnlyDictionary Tags, + IReadOnlyDictionary Parsed) + { + /// The empty tables a driver starts with. + public static readonly TagTables Empty = new( + new Dictionary(StringComparer.Ordinal), + new Dictionary(StringComparer.Ordinal)); + } + + /// The live tag tables; replaced by reference at init, never mutated in place. + private TagTables _tables = TagTables.Empty; + + /// The live RawPath → definition table. + private IReadOnlyDictionary TagsByRawPath => Volatile.Read(ref _tables).Tags; + + /// The live RawPath → pre-parsed address table. + private IReadOnlyDictionary ParsedByRawPath => Volatile.Read(ref _tables).Parsed; + /// public DriverHealth GetHealth() => Health; @@ -488,8 +676,15 @@ public sealed class S7Driver var outcome = new ReadPassOutcome(); await _gate.WaitAsync(cancellationToken).ConfigureAwait(false); + // The lifecycle generation this pass runs in — its health write is dropped if a teardown or + // re-initialize has moved on by the time it lands. Captured under _gate, where the init + // publish happens, and re-checked against _initialized: a pass that passed RequireInitialized + // and then waited out a teardown must not reopen a connection on a shut-down driver. + var gen = CurrentGeneration; try { + RequireInitialized(); + // (Re)open the connection under the gate. A failed (re)connect degrades the whole // batch to BadCommunicationError rather than throwing — the poll loop / next read // retries, and EnsureConnectedAsync repairs on the following call. @@ -506,7 +701,7 @@ public sealed class S7Driver { for (var i = 0; i < fullReferences.Count; i++) results[i] = new DataValueSnapshot(null, StatusBadCommunicationError, null, now); - Health = new DriverHealth(DriverState.Degraded, Health.LastSuccessfulRead, ex.Message); + TryUpdateHealth(gen, h => new DriverHealth(DriverState.Degraded, h.LastSuccessfulRead, ex.Message)); return results; } @@ -635,10 +830,10 @@ public sealed class S7Driver state = wf.State; error = wf.Message; } - Health = new DriverHealth( + TryUpdateHealth(gen, h => new DriverHealth( state, - outcome.AnySuccess ? now : Health.LastSuccessfulRead, - error); + outcome.AnySuccess ? now : h.LastSuccessfulRead, + error)); } _gate.Release(); @@ -1068,7 +1263,7 @@ public sealed class S7Driver /// The parsed address, valid only when this returns true. /// true when the address is known or parses. private bool TryGetParsedAddress(S7TagDefinition tag, out S7ParsedAddress addr) - => _parsedByRawPath.TryGetValue(tag.Name, out addr) || S7AddressParser.TryParse(tag.Address, out addr); + => ParsedByRawPath.TryGetValue(tag.Name, out addr) || S7AddressParser.TryParse(tag.Address, out addr); /// /// The read-failure classification ladder, shared by the per-tag path and the span path so @@ -1137,10 +1332,10 @@ public sealed class S7Driver private async Task ReadOneAsync(IS7Plc plc, S7TagDefinition tag, CancellationToken ct) { - // Every tag pre-parses its address at init (_parsedByRawPath); the on-demand + // Every tag pre-parses its address at init (ParsedByRawPath); the on-demand // S7AddressParser.Parse is a defensive fallback for a tag not in that table. Parse throws // FormatException on a bad address, which the caller's catch maps to BadCommunicationError. - var addr = _parsedByRawPath.TryGetValue(tag.Name, out var parsed) + var addr = ParsedByRawPath.TryGetValue(tag.Name, out var parsed) ? parsed : S7AddressParser.Parse(tag.Address); @@ -1691,8 +1886,11 @@ public sealed class S7Driver var outcome = new WritePassOutcome(); await _gate.WaitAsync(cancellationToken).ConfigureAwait(false); + var gen = CurrentGeneration; // see ReadAsync try { + RequireInitialized(); + // (Re)open under the gate; a failed (re)connect degrades every write to // BadCommunicationError rather than throwing. IS7Plc plc; @@ -1733,7 +1931,7 @@ public sealed class S7Driver // Timer/Counter, but the mapper doesn't run that guard). // TryParse (not Parse) so a malformed address falls through to the existing // error path rather than throwing here. - if (_parsedByRawPath.TryGetValue(tag.Name, out var tcParsed) + if (ParsedByRawPath.TryGetValue(tag.Name, out var tcParsed) ? tcParsed.Area is S7Area.Timer or S7Area.Counter : (S7AddressParser.TryParse(tag.Address, out tcParsed) && tcParsed.Area is S7Area.Timer or S7Area.Counter)) { @@ -1806,7 +2004,7 @@ public sealed class S7Driver { // THE single health write of the batch (a finally, so an OCE mid-batch still flushes the // evidence gathered so far — a pass that observed nothing writes nothing). - ApplyWritePassHealth(outcome); + ApplyWritePassHealth(gen, outcome); _gate.Release(); } return results; @@ -1818,10 +2016,11 @@ public sealed class S7Driver /// successful sibling. Maintains , the sticky Faulted a read pass /// must not erase. Caller holds . /// + /// The lifecycle generation the batch ran in; a superseded batch writes nothing. /// The batch's accumulated outcomes. - private void ApplyWritePassHealth(WritePassOutcome outcome) + private void ApplyWritePassHealth(long gen, WritePassOutcome outcome) { - if (!outcome.HasObservations) return; + if (!outcome.HasObservations || !IsCurrentGeneration(gen)) return; var worst = outcome.WorstFailure; var priorWriteFault = _writeFault; @@ -1836,9 +2035,10 @@ public sealed class S7Driver // side) were holding — a clean write is the only evidence the alert has cleared. // Only a Faulted the write side owns: a read-side Faulted (e.g. access denied on another DB) is the // read pass's to clear, and Health still carries the write fault's own message iff it is ours. - if (outcome.AnySuccess && _writeFault is null && priorWriteFault is { } owned - && Health.State == DriverState.Faulted && Health.LastError == owned.Message) - Health = new DriverHealth(DriverState.Healthy, Health.LastSuccessfulRead, null); + if (outcome.AnySuccess && _writeFault is null && priorWriteFault is { } owned) + TryUpdateHealth(gen, h => h.State == DriverState.Faulted && h.LastError == owned.Message + ? new DriverHealth(DriverState.Healthy, h.LastSuccessfulRead, null) + : null); return; } @@ -1850,7 +2050,7 @@ public sealed class S7Driver state = wf.State; error = wf.Message; } - Health = new DriverHealth(state, Health.LastSuccessfulRead, error); + TryUpdateHealth(gen, h => new DriverHealth(state, h.LastSuccessfulRead, error)); } /// @@ -1881,10 +2081,10 @@ public sealed class S7Driver private async Task WriteOneAsync(IS7Plc plc, S7TagDefinition tag, object? value, CancellationToken ct) { // Parse the address the same way ReadOneAsync does: every tag pre-parses at init - // (_parsedByRawPath), with an on-demand parse as a defensive fallback. Needed here so the + // (ParsedByRawPath), with an on-demand parse as a defensive fallback. Needed here so the // wide-type write can byte-address the block (the narrow path below addresses by the raw // address string instead). - var addr = _parsedByRawPath.TryGetValue(tag.Name, out var parsed) + var addr = ParsedByRawPath.TryGetValue(tag.Name, out var parsed) ? parsed : S7AddressParser.Parse(tag.Address); @@ -2243,7 +2443,7 @@ public sealed class S7Driver { ArgumentNullException.ThrowIfNull(builder); var folder = builder.Folder("S7", "S7"); - foreach (var t in _tagsByRawPath.Values) + foreach (var t in TagsByRawPath.Values) { // A tag carrying a non-null array count (>= 1) surfaces as a 1-D OPC UA array node. // A null count stays scalar. A count of 1 IS a valid 1-element array: the foundation @@ -2308,8 +2508,9 @@ public sealed class S7Driver private void HandlePollError(Exception ex) { _logger.LogWarning(ex, "S7 poll reader failed. Driver={DriverInstanceId}", _driverInstanceId); - if (Health.State != DriverState.Faulted) - Health = new DriverHealth(DriverState.Degraded, Health.LastSuccessfulRead, ex.Message); + TryUpdateHealth(CurrentGeneration, h => h.State == DriverState.Faulted + ? null + : new DriverHealth(DriverState.Degraded, h.LastSuccessfulRead, ex.Message)); } // ---- IHostConnectivityProbe ---- @@ -2418,7 +2619,10 @@ public sealed class S7Driver if (_disposed) return; _disposed = true; SynchronousTeardown(); - _gate.Dispose(); + // _gate / _lifecycleGate are deliberately NOT disposed: a Dispose can race a still-running + // ShutdownAsync (the runtime disposes right after a bounded shutdown), which would then throw + // ObjectDisposedException mid-teardown. A SemaphoreSlim whose wait handle was never requested + // holds no unmanaged resource. } /// Asynchronously disposes the driver and releases resources. @@ -2427,9 +2631,10 @@ public sealed class S7Driver { if (_disposed) return; _disposed = true; + // Safe while a ShutdownAsync is still running: it waits (bounded) for the lifecycle gate, and + // the teardown is idempotent. The semaphores are not disposed (see Dispose). try { await ShutdownAsync(CancellationToken.None).ConfigureAwait(false); } catch { /* disposal is best-effort */ } - _gate.Dispose(); } /// @@ -2439,6 +2644,10 @@ public sealed class S7Driver /// private void SynchronousTeardown() { + // Its own lifecycle generation, like ShutdownCoreAsync: an initialize still running past this + // point discards what it built instead of publishing it over the torn-down driver. + var gen = BeginGeneration(); + // Drain the polling overlay (drain-before-dispose). try { _poll.DisposeAsync().AsTask().Wait(DrainTimeout); } catch { /* timeouts/loop faults are tolerated — teardown continues */ } @@ -2463,6 +2672,6 @@ public sealed class S7Driver try { Plc?.Dispose(); } catch { /* best-effort — tearing down anyway */ } Plc = null; Volatile.Write(ref _negotiatedPduSize, 0); // a stale post-teardown reading would read as "online" - Health = new DriverHealth(DriverState.Unknown, Health.LastSuccessfulRead, null); + TryUpdateHealth(gen, h => new DriverHealth(DriverState.Unknown, h.LastSuccessfulRead, null)); } } diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7.Tests/S7LifecycleGateTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7.Tests/S7LifecycleGateTests.cs new file mode 100644 index 00000000..d856e307 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7.Tests/S7LifecycleGateTests.cs @@ -0,0 +1,143 @@ +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.Driver.S7.Tests; + +/// +/// R07:H5 — overlapping lifecycle calls. DriverInstanceActor dispatches initialize on a +/// fire-and-forget task and a reconnect retry can start a second attempt while the first is still in +/// OpenAsync; an abandoned initialize can also still be running when the next shutdown starts. +/// Every wait here is a condition wait on the fake's own signals, never a sleep. +/// +[Trait("Category", "Unit")] +public sealed class S7LifecycleGateTests +{ + private static readonly TimeSpan Bound = TimeSpan.FromSeconds(10); + + private static S7DriverOptions Options(bool probe = false) => new() + { + Host = "192.0.2.1", + Timeout = TimeSpan.FromSeconds(30), // the fake's gated open is released by the test, never timed out + Probe = new S7ProbeOptions { Enabled = probe, Interval = TimeSpan.FromSeconds(30) }, + RawTags = S7RawTags.Entries(new S7TagDefinition("W0", "DB1.DBW0", S7DataType.UInt16, Writable: false)), + }; + + /// A factory whose FIRST open blocks until released; later opens complete immediately. + private sealed class FirstOpenBlocks + { + public readonly ScriptedS7PlcFactory Factory = new(); + public readonly TaskCompletionSource FirstOpenEntered = new(TaskCreationOptions.RunContinuationsAsynchronously); + public readonly TaskCompletionSource ReleaseFirstOpen = new(TaskCreationOptions.RunContinuationsAsynchronously); + + public FirstOpenBlocks() + { + Factory.Open = async (index, ct) => + { + if (index != 0) return; + FirstOpenEntered.TrySetResult(); + await ReleaseFirstOpen.Task.WaitAsync(ct).ConfigureAwait(false); + }; + } + } + + /// + /// A second initialize waits for the first instead of running beside it, then releases the first + /// attempt's connection: exactly one live connection remains. Before the gate, attempt B opened + /// its own connection while A was still in OpenAsync, and A's published connection (and + /// probe loop) was overwritten and never released. + /// + [Fact] + public async Task Overlapping_initializes_are_serialised_and_leave_one_live_connection() + { + var ct = TestContext.Current.CancellationToken; + var f = new FirstOpenBlocks(); + using var drv = new S7Driver(Options(), "s7-overlap", f.Factory); + + var a = drv.InitializeAsync("{}", ct); + await f.FirstOpenEntered.Task.WaitAsync(Bound, ct); + var b = drv.InitializeAsync("{}", ct); + + // Positive control first: B is genuinely pending (not finished, not faulted)... + b.IsCompleted.ShouldBeFalse(); + // ...and while A holds the gate B has not created a connection of its own. B can only reach + // its open after A completes, so this is not a race the test can lose by being slow. + f.Factory.Created.Count.ShouldBe(1, "a second initialize must not open beside the first"); + + f.ReleaseFirstOpen.SetResult(); + await a.WaitAsync(Bound, ct); + await b.WaitAsync(Bound, ct); + + f.Factory.Created.Count.ShouldBe(2); + f.Factory.Created[0].Disposed.ShouldBeTrue("B releases A's published connection before replacing it"); + f.Factory.Created[1].Disposed.ShouldBeFalse(); + drv.GetHealth().State.ShouldBe(DriverState.Healthy); + (await drv.ReadAsync(["W0"], ct))[0].StatusCode.ShouldBe(0u); + } + + /// + /// An initialize still running when a shutdown gives up waiting for it must not publish its + /// connection, its health or its probe loop over the shut-down driver; it fails and disposes what + /// it opened. Before the fix the late attempt published Healthy + a live connection after the + /// shutdown returned. + /// + [Fact] + public async Task An_initialize_superseded_by_shutdown_publishes_nothing() + { + var ct = TestContext.Current.CancellationToken; + var f = new FirstOpenBlocks(); + using var drv = new S7Driver(Options(probe: true), "s7-superseded", f.Factory) + { + LifecycleGateTimeout = TimeSpan.FromSeconds(1), + }; + + var init = drv.InitializeAsync("{}", ct); + await f.FirstOpenEntered.Task.WaitAsync(Bound, ct); + + // The shutdown cannot get the lifecycle gate (the attempt holds it), gives up after its bound + // and tears down anyway. + await drv.ShutdownAsync(ct).WaitAsync(Bound, ct); + drv.GetHealth().State.ShouldBe(DriverState.Unknown); + + // The abandoned attempt now completes its open... + f.ReleaseFirstOpen.SetResult(); + var ex = await Should.ThrowAsync(() => init.WaitAsync(Bound, ct)); + ex.Message.ShouldContain("superseded"); + + // ...and published nothing: no connection kept, health untouched, the driver still shut down. + f.Factory.Created.ShouldHaveSingleItem().Disposed.ShouldBeTrue(); + drv.GetHealth().State.ShouldBe(DriverState.Unknown, "a superseded attempt must not write Healthy or Faulted"); + await Should.ThrowAsync(() => drv.ReadAsync(["W0"], ct)); + + // Positive control: the driver is not wedged — a fresh initialize publishes normally. + await drv.InitializeAsync("{}", ct).WaitAsync(Bound, ct); + drv.GetHealth().State.ShouldBe(DriverState.Healthy); + (await drv.ReadAsync(["W0"], ct))[0].StatusCode.ShouldBe(0u); + } + + /// + /// Dispose racing a shutdown that is still running is safe: neither throws, and the driver ends + /// torn down. The semaphores are no longer disposed, so the running shutdown cannot hit an + /// ObjectDisposedException. + /// + [Fact] + public async Task Dispose_while_a_shutdown_is_still_running_is_safe() + { + var ct = TestContext.Current.CancellationToken; + var f = new FirstOpenBlocks(); + var drv = new S7Driver(Options(), "s7-dispose-race", f.Factory) { LifecycleGateTimeout = TimeSpan.FromSeconds(1) }; + + var init = drv.InitializeAsync("{}", ct); + await f.FirstOpenEntered.Task.WaitAsync(Bound, ct); + + var shutdown = drv.ShutdownAsync(ct); // waits on the gate the blocked init holds + var dispose = drv.DisposeAsync().AsTask(); + await shutdown.WaitAsync(Bound, ct); + await dispose.WaitAsync(Bound, ct); + + f.ReleaseFirstOpen.SetResult(); + await Should.ThrowAsync(() => init.WaitAsync(Bound, ct)); + f.Factory.Created.ShouldHaveSingleItem().Disposed.ShouldBeTrue(); + drv.GetHealth().State.ShouldBe(DriverState.Unknown); + } +}