diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT/AdsTwinCATClient.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT/AdsTwinCATClient.cs index 9eecbd34..649998b0 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT/AdsTwinCATClient.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT/AdsTwinCATClient.cs @@ -1098,10 +1098,13 @@ internal sealed class AdsTwinCATClient : ITwinCATClient return d; } + // An explicit infinity — a float/double infinity, or a string that SPELLS one ("Infinity", "-inf", "∞") + // — is the caller's own value and passes; a finite number that overflows ("1e400") does not. private static bool IsExplicitInfinity(object? value) => value switch { float f => float.IsInfinity(f), double d => double.IsInfinity(d), + string s => s.Contains("inf", StringComparison.OrdinalIgnoreCase) || s.Contains('∞'), _ => false, }; diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT/TwinCATDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT/TwinCATDriver.cs index 4d9be07b..8b6c0927 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT/TwinCATDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT/TwinCATDriver.cs @@ -27,7 +27,28 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery private readonly string _driverInstanceId; private readonly ITwinCATClientFactory _clientFactory; private readonly ILogger _logger; - private readonly PollGroupEngine _poll; + // The poll engine belongs to a teardown GENERATION (#739 fix round 2): ShutdownAsync / Dispose retire the + // current engine and install a fresh one in the same swap, then dispose the retired one — possibly in the + // background after a cancelled or over-budget shutdown. PollGroupEngine.DisposeAsync clears whatever its + // own table holds when it finishes, so a subscription made after a later InitializeAsync must never land + // in a retiring engine, or the old teardown would dispose its CTS without cancelling it (a poll loop no + // UnsubscribeAsync can find, running forever). + private volatile PollGeneration _poll; + private long _nextPollGeneration; + + /// One generation's poll engine plus its serial (the serial is part of every handle it issues). + private sealed record PollGeneration(long Serial, PollGroupEngine Engine); + + /// + /// A poll subscription handle, qualified by the engine generation that issued it. Each engine numbers its + /// handles from 1, so without the generation a retired engine's late callback would carry a handle EQUAL + /// to a new subscription's and be taken for it. + /// + private sealed record TwinCATPollHandle(long Generation, ISubscriptionHandle Inner) : ISubscriptionHandle + { + /// + public string DiagnosticId => $"twincat-poll-g{Generation}-{Inner.DiagnosticId}"; + } // Device + tag tables are BUILD-AND-SWAP (#739 L8): InitializeAsync builds fresh tables and publishes // them by reference under _lifecycleLock; ShutdownAsync / Dispose swap in empty ones and release what @@ -120,18 +141,27 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery _logger = logger ?? NullLogger.Instance; _resolver = new EquipmentTagRefResolver( r => _tagsByRawPath.TryGetValue(r, out var t) ? t : null); - _poll = new PollGroupEngine( + _poll = CreatePollEngine(); + } + + /// Builds one generation's poll engine; its callbacks publish generation-qualified handles. + private PollGeneration CreatePollEngine() + { + var driverInstanceId = _driverInstanceId; + var serial = Interlocked.Increment(ref _nextPollGeneration); + return new PollGeneration(serial, new PollGroupEngine( reader: ReadAsync, onChange: (handle, tagRef, snapshot) => - OnDataChange?.Invoke(this, new DataChangeEventArgs(handle, tagRef, snapshot)), + OnDataChange?.Invoke(this, new DataChangeEventArgs(new TwinCATPollHandle(serial, handle), tagRef, snapshot)), onError: HandlePollError, backoffCap: PollBackoffCap, - onBatch: (handle, items) => OnDataChangeBatch?.Invoke(this, new DataChangeBatchEventArgs(handle, items)), + onBatch: (handle, items) => OnDataChangeBatch?.Invoke(this, + new DataChangeBatchEventArgs(new TwinCATPollHandle(serial, handle), items)), // The engine owns no Meter (Core.Abstractions takes no metrics dependency), so this // closure IS the production consumer of its overrun signal. onOverrun: _ => OtOpcUaTelemetry.PollOverruns.Add(1, new KeyValuePair(OtOpcUaTelemetry.PollOverrunDriverTypeTag, DriverTypeNames.TwinCAT), - new KeyValuePair(OtOpcUaTelemetry.PollOverrunDriverInstanceTag, driverInstanceId))); + new KeyValuePair(OtOpcUaTelemetry.PollOverrunDriverInstanceTag, driverInstanceId)))); } /// Upper bound on the poll-loop failure backoff — adopts the S7-proven 30 s cap fleet-wide. @@ -327,8 +357,12 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery } } - /// Swaps out every table under the lifecycle lock and returns the device set the caller now owns. - private ICollection TakeAllForTeardown() + /// + /// Swaps out every table AND the poll engine under the lifecycle lock and returns what the caller now + /// owns. Everything returned belongs to the retired generation only; nothing a later + /// / creates is reachable from it. + /// + private (ICollection devices, PollGroupEngine poll) TakeAllForTeardown() { lock (_lifecycleLock) { @@ -336,10 +370,16 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery var taken = _devices; _devices = NewDeviceTable(); _tagsByRawPath = EmptyTags; - return taken.Values; + var poll = _poll; + _poll = CreatePollEngine(); + return (taken.Values, poll.Engine); } } + /// The most recent teardown's background work (poll-engine dispose + device release). Test + /// visibility only: a cancelled shutdown leaves it running. + internal Task? LastTeardownForTest { get; private set; } + /// public async Task ReinitializeAsync(string driverConfigJson, CancellationToken cancellationToken) { @@ -362,11 +402,13 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery foreach (var r in sub.Registrations) { try { r.Dispose(); } catch { } } _nativeSubs.Clear(); - var taken = TakeAllForTeardown(); + var (taken, retiredPoll) = TakeAllForTeardown(); // The poll loops and the device release wind down in parallel (releasing a client can unblock a - // poll loop stuck in a read on it); the whole teardown is bounded. - var teardown = Task.WhenAll(_poll.DisposeAsync().AsTask(), ReleaseDevicesAsync(taken)); + // poll loop stuck in a read on it); the whole teardown is bounded. Both halves own ONLY the retired + // generation, so finishing in the background after a cancelled wait is safe. + var teardown = Task.WhenAll(retiredPoll.DisposeAsync().AsTask(), ReleaseDevicesAsync(taken)); + LastTeardownForTest = teardown; var cancelled = false; try { await teardown.WaitAsync(ShutdownBudget, cancellationToken).ConfigureAwait(false); } catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) @@ -515,6 +557,19 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery return new DataValueSnapshot(value, status, now, now); } + /// + /// Applies the health contract to a write the client returned a status for: Good is the device + /// answering (clears that device's comms fault, advances LastSuccessfulRead); a comms-class + /// status is a comms fault; anything else is a refusal — the device answered, the state is unchanged. + /// + private void NoteWriteOutcome(DeviceState device, uint status, string reference) + { + var now = DateTime.UtcNow; + if (status == TwinCATStatusMapper.Good) RecordDeviceAnswered(now, device, force: true); + else if (IsCommsStatus(status)) MarkCommsFault($"ADS status {status:X8} writing {reference}", device); + else RecordDeviceAnsweredKeepState(now); + } + /// /// True for a status that means the request did not get a device answer (router / transport / /// timeout). Every other non-Good status is the device refusing this one tag (symbol not found, type @@ -675,6 +730,7 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery { gate.Release(); } + NoteWriteOutcome(device, results[i].StatusCode, w.FullReference); continue; } @@ -685,12 +741,14 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery // array — the inverse of the array read. results[i] = new WriteResult(await client.WriteArrayAsync( symbolName, def.DataType, elements, cancellationToken).ConfigureAwait(false)); + NoteWriteOutcome(device, results[i].StatusCode, w.FullReference); continue; } var status = await client.WriteValueAsync( symbolName, def.DataType, parsed?.BitIndex, w.Value, cancellationToken).ConfigureAwait(false); results[i] = new WriteResult(status); + NoteWriteOutcome(device, status, w.FullReference); } catch (OperationCanceledException) { throw; } catch (NotSupportedException) @@ -826,7 +884,17 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery IReadOnlyList fullReferences, TimeSpan publishingInterval, CancellationToken cancellationToken) { if (!_options.UseNativeNotifications) - return _poll.Subscribe(fullReferences, publishingInterval); + { + // A teardown racing this call may retire the engine we read: re-home the subscription on the + // current one so it never lives in an engine that is being disposed. + while (true) + { + var generation = _poll; + var inner = generation.Engine.Subscribe(fullReferences, publishingInterval); + if (ReferenceEquals(generation, _poll)) return new TwinCATPollHandle(generation.Serial, inner); + generation.Engine.Unsubscribe(inner); + } + } var id = Interlocked.Increment(ref _nextNativeSubId); var handle = new NativeSubscriptionHandle(id); @@ -987,6 +1055,10 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery /// ADS state is not Run (coordinator ruling: a stopped / config-mode PLC is Degraded with a /// reason even though it answers), or an uncleared comms fault seen on it. /// + /// + /// The ADS state comes from the probe's ReadState, so PLC-not-in-RUN detection needs the probe + /// (Probe.Enabled, the default); with the probe off the state is unknown and never degrades. + /// private static string? DeviceDegradedReason(DeviceState d) { if (d.ConsecutiveProbeFailures >= ProbeFailuresBeforeDegraded) @@ -1042,9 +1114,20 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery /// or retried by the reconnect replay; a comms-class failure stays on the every-tick retry. /// /// True when the published status changes (the caller republishes the tag). - private static bool NoteDeadFailure(NativeRegistration reg, uint status, bool refused, DateTime now) + /// The registration. + /// The status the failure maps to. + /// True for a refusal (backs off), false for a comms-class failure. + /// The failure time. + /// The registration's when the attempt + /// started; when a re-arm happened while it was in flight the backoff is NOT re-imposed (the re-arm wins). + private static bool NoteDeadFailure(NativeRegistration reg, uint status, bool refused, DateTime now, + int? epochAtAttempt = null) { - if (refused) + if (epochAtAttempt is int epoch && epoch != reg.RearmEpoch) + { + // A symbol-version change re-armed this registration mid-attempt: keep it re-armed. + } + else if (refused) { reg.RefusalBackoff = reg.RefusalBackoff == TimeSpan.Zero ? RefusedRetryInitial @@ -1141,7 +1224,10 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery } // Async teardown: the engine drains the in-flight poll without blocking the caller's // thread — this runs inside a DriverInstanceActor mailbox turn. - await _poll.UnsubscribeAsync(handle).ConfigureAwait(false); + // A handle of a retired generation has nothing left to stop: its engine was disposed with it. + var current = _poll; + if (handle is TwinCATPollHandle poll && poll.Generation == current.Serial) + await current.Engine.UnsubscribeAsync(poll.Inner).ConfigureAwait(false); } private sealed record NativeSubscriptionHandle(long Id) : ISubscriptionHandle @@ -1255,9 +1341,25 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery /// Earliest time the probe-tick retry may re-attempt this dead registration. public DateTime NextRetryUtc { get; set; } - /// Current refusal backoff (zero after a comms-class failure or a success). + /// Current refusal backoff (zero after a comms-class failure, a success or a re-arm). public TimeSpan RefusalBackoff { get; set; } + private int _rearmEpoch; + + /// Bumped by every symbol-version re-arm; a retry that started before it must not re-impose a backoff. + public int RearmEpoch => Volatile.Read(ref _rearmEpoch); + + /// + /// Re-arms the retry after a PLC program change: retry on the next tick and restart the backoff + /// from (the refusal it backed off from may no longer apply). + /// + public void Rearm() + { + Interlocked.Increment(ref _rearmEpoch); + RefusalBackoff = TimeSpan.Zero; + NextRetryUtc = DateTime.MinValue; + } + private volatile bool _disposed; /// Gets a value indicating whether this registration has been disposed (its callbacks are dropped). @@ -1793,6 +1895,7 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery if (reg.HasLiveHandle) continue; // A refused registration waits out its backoff (a symbol-version change re-arms it). if (now < reg.NextRetryUtc) continue; + var epoch = reg.RearmEpoch; try { var newHandle = await client.AddNotificationAsync( @@ -1816,7 +1919,7 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery // or the reverse); the same failure again publishes nothing. var refusal = ex as TwinCATNotificationRefusedException; var status = refusal?.StatusCode ?? TwinCATStatusMapper.BadCommunicationError; - if (NoteDeadFailure(reg, status, refusal is not null, now)) + if (NoteDeadFailure(reg, status, refusal is not null, now, epoch)) try { reg.OnRefused(status); } catch { /* never abort the rest */ } _logger.LogDebug(ex, "TwinCAT driver '{DriverInstanceId}' probe-tick retry of dead notification " + @@ -1880,7 +1983,7 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery foreach (var d in _devices.Values) if (ReferenceEquals(d.Client, sender)) foreach (var reg in d.NativeRegistrations.Values) - reg.NextRetryUtc = DateTime.MinValue; + reg.Rearm(); OnRediscoveryNeeded?.Invoke(this, new RediscoveryEventArgs( "TwinCAT symbol-version-changed (DeviceSymbolVersionInvalid 0x0711) — PLC program re-downloaded", ScopeHint: "TwinCAT")); @@ -1902,12 +2005,12 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery foreach (var r in sub.Registrations) { try { r.Dispose(); } catch { } } _nativeSubs.Clear(); - var taken = TakeAllForTeardown(); + var (taken, retiredPoll) = TakeAllForTeardown(); var probes = BeginRelease(taken); // PollGroupEngine.DisposeAsync awaits loop tasks; drive it synchronously with a bounded Wait() on // the returned task (no sync-context capture), in parallel with the probe join. - var poll = _poll.DisposeAsync().AsTask(); + var poll = retiredPoll.DisposeAsync().AsTask(); try { Task.WhenAll([poll, .. probes]).Wait(ProbeJoinTimeout); } catch { /* cancel races are expected */ } EndRelease(taken); diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT.Tests/FakeTwinCATClient.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT.Tests/FakeTwinCATClient.cs index 49a6deea..8559f9af 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT.Tests/FakeTwinCATClient.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT.Tests/FakeTwinCATClient.cs @@ -83,16 +83,31 @@ internal class FakeTwinCATClient : ITwinCATClient public virtual Task<(object? value, uint status)> ReadValueAsync( string symbolPath, TwinCATDataType type, int? bitIndex, int? arrayCount, CancellationToken ct) { + if (BlockRead is { } block) + { + // A read wedged in the (uncancellable) wire call; completes with the normal answer once released. + return block.Task.ContinueWith(_ => + { + BlockRead = null; + return ReadValueAsync(symbolPath, type, bitIndex, arrayCount, CancellationToken.None).Result; + }, TaskScheduler.Default); + } if (ThrowOnRead) throw Exception ?? new InvalidOperationException(); - SingleReadCount++; + Interlocked.Increment(ref _singleReadCount); LastReadArrayCount = arrayCount; var status = ReadStatuses.TryGetValue(symbolPath, out var s) ? s : TwinCATStatusMapper.Good; var value = Values.TryGetValue(symbolPath, out var v) ? v : null; return Task.FromResult((value, status)); } + /// When set, does not complete (and ignores cancellation) until this is + /// completed — a poll read wedged in the wire call. + public TaskCompletionSource? BlockRead { get; set; } + + private int _singleReadCount; + /// Gets the number of individual calls served. - public int SingleReadCount { get; private set; } + public int SingleReadCount { get => Volatile.Read(ref _singleReadCount); private set => Volatile.Write(ref _singleReadCount, value); } /// Gets the size of each batch the driver issued, in order (#645). public List SumReadBatches { get; } = new(); /// Gets or sets an exception throws instead of answering — a @@ -264,6 +279,9 @@ internal class FakeTwinCATClient : ITwinCATClient public int AttemptsFor(string symbolPath) => _addAttempts.TryGetValue(symbolPath, out var n) ? n : 0; /// Records the most recently-supplied maxDelayMs for Driver.TwinCAT-014 tests. public int LastMaxDelayMs { get; private set; } + /// When set, awaits this gate BEFORE applying + /// — wedges an attempt in flight whose outcome is a failure. + public TaskCompletionSource? PreAttemptGate { get; set; } /// When set, awaits this gate before creating the handle — /// lets a test wedge a replay/register mid-flight (STAB-17 orphan race). public TaskCompletionSource? AddNotificationGate { get; set; } @@ -283,6 +301,7 @@ internal class FakeTwinCATClient : ITwinCATClient int maxDelayMs, Action onChange, CancellationToken cancellationToken) { _addAttempts.AddOrUpdate(symbolPath, 1, (_, n) => n + 1); + if (PreAttemptGate is { } pre) await pre.Task.ConfigureAwait(false); if (ThrowOnAddNotification) throw Exception ?? new InvalidOperationException("fake AddNotification failure"); if (AddNotificationErrors.TryGetValue(symbolPath, out var perSymbol)) diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT.Tests/TwinCATReviewFixRound1Tests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT.Tests/TwinCATReviewFixRound1Tests.cs index 35ff945b..a25d5292 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT.Tests/TwinCATReviewFixRound1Tests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT.Tests/TwinCATReviewFixRound1Tests.cs @@ -210,7 +210,15 @@ public sealed class TwinCATReviewFixRound1Tests await Should.ThrowAsync(() => drv.ShutdownAsync(cts.Token)); sw.Elapsed.ShouldBeLessThan(TwinCATDriver.ProbeJoinTimeout, "the wait ended on the token, not the join bound"); drv.DeviceCount.ShouldBe(0); + drv.GetHealth().State.ShouldBe(DriverState.Unknown); + + // The background teardown finishes once the wedge releases, and disposes the client the wedged + // connect produced (no dangling teardown left behind the test). + var teardown = drv.LastTeardownForTest.ShouldNotBeNull(); wedge.SetResult(); + await teardown.WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken); + await WaitForAsync(() => factory.Clients[0].DisposeCount > 0, TimeSpan.FromSeconds(3)); + factory.Clients[0].DisposeCount.ShouldBeGreaterThan(0); } // ---- MINOR 7: the real STRING / array encode path through the client planning helpers ---- diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT.Tests/TwinCATReviewFixRound2Tests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT.Tests/TwinCATReviewFixRound2Tests.cs new file mode 100644 index 00000000..0acc860f --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT.Tests/TwinCATReviewFixRound2Tests.cs @@ -0,0 +1,262 @@ +using System.Collections.Concurrent; +using Shouldly; +using TwinCAT.TypeSystem; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Driver.TwinCAT; + +namespace ZB.MOM.WW.OtOpcUa.Driver.TwinCAT.Tests; + +/// +/// #739 review fix round 2: a cancelled shutdown's background teardown is confined to its own generation +/// (I1); write outcomes feed per-device health (M1); a symbol-version re-arm survives an in-flight retry +/// and restarts the backoff (M2/M3); WSTRING / multi-dimensional encode plans (M5); infinity strings (M6). +/// +[Trait("Category", "Unit")] +public sealed class TwinCATReviewFixRound2Tests +{ + private const string Host = "ads://5.23.91.23.1.1:851"; + + private static TwinCATProbeOptions FastProbe => new() + { + Enabled = true, + Interval = TimeSpan.FromMilliseconds(30), + Timeout = TimeSpan.FromMilliseconds(100), + }; + + private static TwinCATDriver NewDriver(FakeTwinCATClientFactory factory, TwinCATProbeOptions probe, bool native, + params TwinCATTagDefinition[] tags) => + new(new TwinCATDriverOptions + { + Devices = [new TwinCATDeviceOptions(Host)], + RawTags = TwinCATRawTags.Entries(tags), + Probe = probe, + UseNativeNotifications = native, + }, "drv-1", factory); + + private static async Task WaitForAsync(Func condition, TimeSpan timeout) + { + var deadline = DateTime.UtcNow + timeout; + while (!condition() && DateTime.UtcNow < deadline) + await Task.Delay(10); + } + + // ---- I1 ---- + + /// + /// The exact reviewed sequence: a poll read is wedged; ShutdownAsync is cancelled (its teardown + /// continues in the background); the same instance is re-initialised and subscribed; the wedge is + /// released and the OLD teardown completes. The NEW subscription keeps polling and delivering (positive + /// control), UnsubscribeAsync finds and stops it, and a final shutdown leaves nothing polling. + /// + [Fact] + public async Task A_cancelled_shutdowns_background_teardown_never_touches_the_next_generations_subscription() + { + var wedge = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var counter = 0; + var clients = new ConcurrentQueue(); + var factory = new FakeTwinCATClientFactory + { + Customise = () => + { + // Client 1 (generation 1) wedges its first read; later clients answer an ever-changing value + // so every poll publishes. + var c = new ChangingValueFake(() => Interlocked.Increment(ref counter)) + { + BlockRead = clients.IsEmpty ? wedge : null, + }; + clients.Enqueue(c); + return c; + }, + }; + var tag = new TwinCATTagDefinition("A", Host, "MAIN.A", TwinCATDataType.DInt); + var drv = NewDriver(factory, new TwinCATProbeOptions { Enabled = false }, native: false, tag); + await drv.InitializeAsync("{}", TestContext.Current.CancellationToken); + var events = new ConcurrentQueue(); + drv.OnDataChange += (_, e) => events.Enqueue(e); + + _ = await drv.SubscribeAsync(["A"], TimeSpan.FromMilliseconds(50), TestContext.Current.CancellationToken); + await WaitForAsync(() => !clients.IsEmpty, TimeSpan.FromSeconds(2)); // gen-1 poll read is wedged + + using (var cts = new CancellationTokenSource(TimeSpan.FromMilliseconds(100))) + await Should.ThrowAsync(() => drv.ShutdownAsync(cts.Token)); + var oldTeardown = drv.LastTeardownForTest.ShouldNotBeNull(); + oldTeardown.IsCompleted.ShouldBeFalse("the old teardown must still be running in the background"); + + await drv.InitializeAsync("{}", TestContext.Current.CancellationToken); + var handle = await drv.SubscribeAsync(["A"], TimeSpan.FromMilliseconds(50), TestContext.Current.CancellationToken); + + wedge.SetResult(); + await oldTeardown.WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken); + + // Positive control: the new subscription delivers AFTER the old teardown finished. + var afterTeardown = events.Count(e => e.SubscriptionHandle.Equals(handle)); + await WaitForAsync(() => events.Count(e => e.SubscriptionHandle.Equals(handle)) >= afterTeardown + 3, TimeSpan.FromSeconds(3)); + events.Count(e => e.SubscriptionHandle.Equals(handle)).ShouldBeGreaterThanOrEqualTo(afterTeardown + 3); + + // UnsubscribeAsync can still find and stop it. + await drv.UnsubscribeAsync(handle, TestContext.Current.CancellationToken); + var stoppedAt = events.Count(e => e.SubscriptionHandle.Equals(handle)); + // Measured window: a sibling driver's poll publishes 5 times; the stopped loop must publish nothing. + var witnessFactory = new FakeTwinCATClientFactory { Customise = () => new ChangingValueFake(() => Environment.TickCount) }; + var witnessDrv = NewDriver(witnessFactory, new TwinCATProbeOptions { Enabled = false }, native: false, tag); + await witnessDrv.InitializeAsync("{}", TestContext.Current.CancellationToken); + var witnessEvents = 0; + witnessDrv.OnDataChange += (_, _) => Interlocked.Increment(ref witnessEvents); + _ = await witnessDrv.SubscribeAsync(["A"], TimeSpan.FromMilliseconds(50), TestContext.Current.CancellationToken); + await WaitForAsync(() => Volatile.Read(ref witnessEvents) >= 5, TimeSpan.FromSeconds(3)); + Volatile.Read(ref witnessEvents).ShouldBeGreaterThanOrEqualTo(5); + await witnessDrv.ShutdownAsync(TestContext.Current.CancellationToken); + + events.Count(e => e.SubscriptionHandle.Equals(handle)).ShouldBe(stoppedAt, "the unsubscribed loop must have stopped"); + await drv.ShutdownAsync(TestContext.Current.CancellationToken); + } + + /// Answers each read with a new value so every poll cycle publishes. + private sealed class ChangingValueFake(Func next) : FakeTwinCATClient + { + public override Task<(object? value, uint status)> ReadValueAsync( + string symbolPath, TwinCATDataType type, int? bitIndex, int? arrayCount, CancellationToken ct) + { + if (BlockRead is not null) return base.ReadValueAsync(symbolPath, type, bitIndex, arrayCount, ct); + Values[symbolPath] = next(); + return base.ReadValueAsync(symbolPath, type, bitIndex, arrayCount, ct); + } + } + + // ---- M1 ---- + + /// + /// A write-only device with the probe off: a wire-failed write degrades; a later Good write is the device + /// answering — Healthy again with the stamp advanced. A refused write (BadNotWritable) changes nothing. + /// + [Fact] + public async Task A_good_write_clears_that_devices_comms_fault() + { + var fake = new FakeTwinCATClient(); + var drv = NewDriver(new FakeTwinCATClientFactory { Customise = () => fake }, + new TwinCATProbeOptions { Enabled = false }, native: false, + new TwinCATTagDefinition("W", Host, "MAIN.W", TwinCATDataType.DInt)); + await drv.InitializeAsync("{}", TestContext.Current.CancellationToken); + var initStamp = drv.GetHealth().LastSuccessfulRead!.Value; + + fake.ThrowOnWrite = true; + _ = await drv.WriteAsync([new WriteRequest("W", 1)], TestContext.Current.CancellationToken); + drv.GetHealth().State.ShouldBe(DriverState.Degraded); + + fake.ThrowOnWrite = false; + fake.WriteStatuses["MAIN.W"] = TwinCATStatusMapper.BadNotWritable; + _ = await drv.WriteAsync([new WriteRequest("W", 1)], TestContext.Current.CancellationToken); + drv.GetHealth().State.ShouldBe(DriverState.Degraded, "a refusal says nothing about the comms fault"); + + await WaitForAsync(() => DateTime.UtcNow > initStamp, TimeSpan.FromSeconds(1)); + fake.WriteStatuses.Remove("MAIN.W"); + _ = await drv.WriteAsync([new WriteRequest("W", 1)], TestContext.Current.CancellationToken); + drv.GetHealth().State.ShouldBe(DriverState.Healthy); + drv.GetHealth().LastSuccessfulRead!.Value.ShouldBeGreaterThan(initStamp); + } + + // ---- M2 + M3 ---- + + /// + /// A symbol-version re-arm that lands while a retry of the refused registration is in flight is not + /// overwritten by that retry's failure: the next tick re-attempts at once. (Without the epoch the + /// failure re-imposed a 30 s+ backoff.) + /// + [Fact] + public async Task A_rearm_during_an_in_flight_retry_is_not_lost() + { + var fake = new FakeTwinCATClient(); + fake.AddNotificationErrors["MAIN.T"] = new TwinCATTypeMismatchException("MAIN.T", "PLC STRING"); + var drv = NewDriver(new FakeTwinCATClientFactory { Customise = () => fake }, FastProbe, native: true, + new TwinCATTagDefinition("T", Host, "MAIN.T", TwinCATDataType.DInt)); + await drv.InitializeAsync("{}", TestContext.Current.CancellationToken); + _ = await drv.SubscribeAsync(["T"], TimeSpan.FromMilliseconds(100), TestContext.Current.CancellationToken); + fake.AttemptsFor("MAIN.T").ShouldBe(1); + + var inFlight = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + fake.PreAttemptGate = inFlight; + fake.FireSymbolVersionChanged(); // re-arm: the next tick starts a retry, which wedges on the gate + await WaitForAsync(() => fake.AttemptsFor("MAIN.T") >= 2, TimeSpan.FromSeconds(3)); + fake.FireSymbolVersionChanged(); // a second program change while that retry is in flight + fake.PreAttemptGate = null; + inFlight.SetResult(); // the in-flight retry now fails (still refused) + + await WaitForAsync(() => fake.AttemptsFor("MAIN.T") >= 3, TimeSpan.FromSeconds(3)); + fake.AttemptsFor("MAIN.T").ShouldBeGreaterThanOrEqualTo(3); + await drv.ShutdownAsync(TestContext.Current.CancellationToken); + } + + /// A re-arm restarts the refusal backoff from the initial delay instead of doubling the old one. + [Fact] + public async Task A_rearm_restarts_the_refusal_backoff() + { + var fake = new FakeTwinCATClient(); + fake.AddNotificationErrors["MAIN.T"] = new TwinCATTypeMismatchException("MAIN.T", "PLC STRING"); + var drv = NewDriver(new FakeTwinCATClientFactory { Customise = () => fake }, FastProbe, native: true, + new TwinCATTagDefinition("T", Host, "MAIN.T", TwinCATDataType.DInt)); + await drv.InitializeAsync("{}", TestContext.Current.CancellationToken); + _ = await drv.SubscribeAsync(["T"], TimeSpan.FromMilliseconds(100), TestContext.Current.CancellationToken); + var reg = drv.GetDeviceState(Host)!.NativeRegistrations.Values.Single(); + reg.RefusalBackoff.ShouldBe(TwinCATDriver.RefusedRetryInitial); + + fake.FireSymbolVersionChanged(); + await WaitForAsync(() => fake.AttemptsFor("MAIN.T") >= 2, TimeSpan.FromSeconds(3)); + await WaitForAsync(() => reg.RefusalBackoff != TimeSpan.Zero, TimeSpan.FromSeconds(1)); + reg.RefusalBackoff.ShouldBe(TwinCATDriver.RefusedRetryInitial); + await drv.ShutdownAsync(TestContext.Current.CancellationToken); + } + + // ---- M5 ---- + + /// A WSTRING tag over a WSTRING symbol writes UTF-16LE with an aligned 2-byte terminator, zero-padded. + [Fact] + public async Task A_WSTRING_write_goes_through_the_client_encode_plan() + { + var fake = new FakeTwinCATClient(); + fake.SymbolShapes["MAIN.W"] = TwinCATPlcSymbolShape.Classify("WSTRING(3)", DataTypeCategory.String, 8); + var drv = NewDriver(new FakeTwinCATClientFactory { Customise = () => fake }, + new TwinCATProbeOptions { Enabled = false }, native: false, + new TwinCATTagDefinition("W", Host, "MAIN.W", TwinCATDataType.WString)); + await drv.InitializeAsync("{}", TestContext.Current.CancellationToken); + + var results = await drv.WriteAsync( + [new WriteRequest("W", "中x"), new WriteRequest("W", "abcd")], TestContext.Current.CancellationToken); + + results[0].StatusCode.ShouldBe(TwinCATStatusMapper.Good); + results[1].StatusCode.ShouldBe(TwinCATStatusMapper.BadOutOfRange); // 4 chars do not fit WSTRING(3) + fake.RawWrites["MAIN.W"].ShouldBe(new byte[] { 0x2D, 0x4E, (byte)'x', 0, 0, 0, 0, 0 }); + } + + /// A multi-dimensional array is written flattened in PLC memory order (last index fastest). + [Fact] + public async Task A_multi_dimensional_array_write_is_flattened_row_major() + { + var fake = new FakeTwinCATClient(); + var element = TwinCATPlcSymbolShape.Classify("INT", DataTypeCategory.Primitive, 2); + fake.SymbolShapes["MAIN.M"] = TwinCATPlcSymbolShape.Classify("ARRAY [0..1,0..2] OF INT", DataTypeCategory.Array, 12); + fake.ArrayShapes["MAIN.M"] = new TwinCATPlcArrayShape([2, 3], 12, element); + var drv = NewDriver(new FakeTwinCATClientFactory { Customise = () => fake }, + new TwinCATProbeOptions { Enabled = false }, native: false, + new TwinCATTagDefinition("M", Host, "MAIN.M", TwinCATDataType.Int, ArrayLength: 6)); + await drv.InitializeAsync("{}", TestContext.Current.CancellationToken); + + var result = await drv.WriteAsync([new WriteRequest("M", new[] { 1, 2, 3, 4, 5, 6 })], TestContext.Current.CancellationToken); + + result.Single().StatusCode.ShouldBe(TwinCATStatusMapper.Good); + fake.RawWrites["MAIN.M"].ShouldBe(new byte[] { 1, 0, 2, 0, 3, 0, 4, 0, 5, 0, 6, 0 }); + TwinCATArrayCodec.Decode(TwinCATDataType.Int, fake.RawWrites["MAIN.M"], new TwinCATArrayLayout(6, 2), false) + .ShouldBe(new short[] { 1, 2, 3, 4, 5, 6 }); + } + + // ---- M6 ---- + + /// An infinity spelled as a string passes like a float/double infinity; a finite overflow does not. + [Fact] + public void Infinity_strings_are_explicit_infinities() + { + AdsTwinCATClient.ConvertForWrite(TwinCATDataType.Real, "Infinity").ShouldBe(float.PositiveInfinity); + AdsTwinCATClient.ConvertForWrite(TwinCATDataType.LReal, "-Infinity").ShouldBe(double.NegativeInfinity); + Should.Throw(() => AdsTwinCATClient.ConvertForWrite(TwinCATDataType.Real, "1e300")); + } +}