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

fix(twincat): native/probe evidence advances LastSuccessfulRead; probe failures degrade (H12)

Browse Source 
In native-notification mode LastSuccessfulRead never moved past the init
stamp, so one Degraded write on a connection older than 90 s latched the
comms-loss watchdog into a full reconnect (every notification on every device
deleted and re-registered). Conversely an unreachable AMS router was swallowed
by the probe and never left Healthy.

- a native push and a successful probe now stamp LastSuccessfulRead (throttled
  to one health record per 250 ms) and restore Healthy unless a device is still
  probe-unreachable;
- 3 consecutive failed probe ticks write Degraded with the stamp preserved;
- a NotSupportedException write is BadNotSupported with no health write (#611);
- a reconnect replay the PLC refuses republishes the refusal's status without
  degrading (it used to report BadCommunicationError + Degraded on every
  reconnect for a correctly-isolated type-mismatch tag).

Refs #739
Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
This commit is contained in:
Joseph Doherty
2026-10-22 14:03:28 -04:00
parent 17f4bf6d33
commit ecdd6dbecb
2 changed files with 252 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT/TwinCATDriver.cs
+91 -15
View File
@@ -346,7 +346,7 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery
{
if (status == TwinCATStatusMapper.Good)
{
_health = new DriverHealth(DriverState.Healthy, now, null);
RecordDeviceAnswered(now, force: true);
return new DataValueSnapshot(value, status, now, now);
}
@@ -524,10 +524,11 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery
results[i] = new WriteResult(status);
}
catch (OperationCanceledException) { throw; }
catch (NotSupportedException nse)
catch (NotSupportedException)
{
// A write the client cannot express is a refusal, not a comms fault (#611): no health
// write, or the comms-loss watchdog would reconnect a healthy device over a config issue.
results[i] = new WriteResult(TwinCATStatusMapper.BadNotSupported);
_health = new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, nse.Message);
}
catch (Exception ex) when (ex is FormatException or InvalidCastException)
{
@@ -540,7 +541,7 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery
catch (Exception ex)
{
results[i] = new WriteResult(TwinCATStatusMapper.BadCommunicationError);
_health = new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, ex.Message);
MarkCommsFault(ex.Message);
}
}
@@ -689,12 +690,18 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery
var reg = new NativeRegistration(
Interlocked.Increment(ref _nextRegId), device, reference, symbolName,
def.DataType, bitIndex, def.ArrayLength, publishingInterval,
onChange: (_, value) => RaiseDataChange(handle, reference, new DataValueSnapshot(
value, TwinCATStatusMapper.Good, DateTime.UtcNow, DateTime.UtcNow)),
onChange: (_, value) =>
{
// A push is the device answering: stamp LastSuccessfulRead so a notification-only
// driver does not look silent to the comms-loss watchdog.
var now = DateTime.UtcNow;
RecordDeviceAnswered(now);
RaiseDataChange(handle, reference, new DataValueSnapshot(value, TwinCATStatusMapper.Good, now, now));
},
// When the handle goes dead (a failed replay) push a Bad snapshot so
// subscribers see the tag go Bad instead of a frozen-Good-stale value.
onUnavailable: () => RaiseDataChange(handle, reference, new DataValueSnapshot(
null, TwinCATStatusMapper.BadCommunicationError, null, DateTime.UtcNow)));
onUnavailable: status => RaiseDataChange(handle, reference, new DataValueSnapshot(
null, status, null, DateTime.UtcNow)));
registrations.Add(reg);
if (unreachable?.Contains(device) == true)
@@ -775,6 +782,43 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery
_health = new DriverHealth(DriverState.Degraded, current.LastSuccessfulRead, message);
}
/// <summary>
/// Consecutive failed probe ticks (connect failure, failed ReadState) after which a device counts as
/// unreachable and the driver goes <see cref="DriverState.Degraded"/> with <c>LastSuccessfulRead</c>
/// preserved — so an unreachable AMS router reaches the comms-loss watchdog instead of leaving the
/// driver Healthy with stale-Good values. Three ticks ride out a single dropped ReadState.
/// </summary>
internal const int ProbeFailuresBeforeDegraded = 3;
/// <summary>Minimum spacing between <c>LastSuccessfulRead</c> stamps from the notification path.</summary>
private static readonly TimeSpan AnsweredStampInterval = TimeSpan.FromMilliseconds(250);
/// <summary>
/// Records that a device answered (a native push, a successful probe, a Good read): advances
/// <c>LastSuccessfulRead</c> and restores <see cref="DriverState.Healthy"/> unless a device is still
/// probe-unreachable (then the state stays Degraded, the stamp still advances). Throttled so a
/// notification burst does not allocate a health record per value; never touches Faulted.
/// </summary>
/// <param name="now">The time the device answered.</param>
/// <param name="force">Write even inside the throttle window (an explicit read / probe outcome).</param>
private void RecordDeviceAnswered(DateTime now, bool force = false)
{
var current = _health;
if (current.State == DriverState.Faulted) return;
// Hot path (a notification on a healthy driver inside the window): no device scan, no allocation.
if (!force && current.State == DriverState.Healthy && current.LastSuccessfulRead is { } recent
&& now - recent < AnsweredStampInterval)
return;
var unreachable = _devices.Values.FirstOrDefault(d => d.ConsecutiveProbeFailures >= ProbeFailuresBeforeDegraded);
var state = unreachable is null ? DriverState.Healthy : DriverState.Degraded;
if (!force && current.State == state && current.LastSuccessfulRead is { } last
&& now - last < AnsweredStampInterval)
return;
_health = new DriverHealth(state, now, unreachable is null
? null
: $"TwinCAT device {unreachable.Options.HostAddress} unreachable ({unreachable.ConsecutiveProbeFailures} consecutive failed probes)");
}
/// <summary>Raises one data change for a native subscription.</summary>
private void RaiseDataChange(ISubscriptionHandle handle, string reference, DataValueSnapshot snapshot) =>
OnDataChange?.Invoke(this, new DataChangeEventArgs(handle, reference, snapshot));
@@ -888,7 +932,14 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery
/// the <see cref="OnChange"/> shape has no quality channel. Closes over the same
/// (handle, reference) pair as <see cref="OnChange"/>.
/// </summary>
public Action OnUnavailable { get; }
public Action<uint> OnUnavailableStatus { get; }
/// <summary>Publishes <c>BadCommunicationError</c> for a handle that went dead on a comms failure.</summary>
public void OnUnavailable() => OnUnavailableStatus(TwinCATStatusMapper.BadCommunicationError);
/// <summary>Publishes the status of a refusal (the PLC answered and refused this tag).</summary>
/// <param name="status">The refusal's OPC UA status.</param>
public void OnRefused(uint status) => OnUnavailableStatus(status);
private ITwinCATNotificationHandle? _handle;
@@ -908,10 +959,10 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery
/// <param name="arrayLength">Authored array length for an array tag; null for a scalar.</param>
/// <param name="interval">Requested notification interval.</param>
/// <param name="onChange">Value-change callback (closes over the owning subscription handle + reference).</param>
/// <param name="onUnavailable">Bad-quality callback fired when the handle goes dead.</param>
/// <param name="onUnavailable">Bad-quality callback (given the status to publish) fired when the handle goes dead.</param>
public NativeRegistration(long id, DeviceState device, string reference, string symbolName,
TwinCATDataType dataType, int? bitIndex, int? arrayLength, TimeSpan interval, Action<string, object?> onChange,
Action onUnavailable)
Action<uint> onUnavailable)
{
Id = id;
Device = device;
@@ -922,7 +973,7 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery
ArrayLength = arrayLength;
Interval = interval;
OnChange = onChange;
OnUnavailable = onUnavailable;
OnUnavailableStatus = onUnavailable;
}
/// <summary>Installs a freshly-registered handle and returns the previous one (null on first set).</summary>
@@ -996,6 +1047,17 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery
catch (OperationCanceledException) when (ct.IsCancellationRequested) { break; }
}
if (success)
{
state.ConsecutiveProbeFailures = 0;
RecordDeviceAnswered(DateTime.UtcNow, force: true);
}
else if (++state.ConsecutiveProbeFailures >= ProbeFailuresBeforeDegraded)
{
MarkCommsFault($"TwinCAT device {state.Options.HostAddress} unreachable " +
$"({state.ConsecutiveProbeFailures} consecutive failed probes)");
}
TransitionDeviceState(state, success ? HostState.Running : HostState.Stopped);
// Opt-in PLC state / task-cycle sampling for /hosts. After the host-state transition so a
@@ -1360,6 +1422,19 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery
}
replayed++;
}
catch (TwinCATNotificationRefusedException refused)
{
// The PLC answered and refused this tag again (it was already isolated at subscribe, or a
// program change made it undecodable): republish the refusal's status, keep it dead for the
// probe-tick retry, and leave health alone — a refusal is not a comms fault.
_logger.LogDebug(
"TwinCAT driver '{DriverInstanceId}' native notification '{Symbol}' on device '{HostAddress}' " +
"still refused after reconnect: {Reason}",
_driverInstanceId, reg.SymbolName, device.Options.HostAddress, refused.Message);
var dead = reg.MarkHandleDead();
try { dead?.Dispose(); } catch { /* already-dead handle from the disposed client */ }
try { reg.OnRefused(refused.StatusCode); } catch { /* never abort recovery of the rest */ }
}
catch (Exception ex)
{
_logger.LogWarning(ex,
@@ -1373,9 +1448,7 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery
var dead = reg.MarkHandleDead();
try { dead?.Dispose(); } catch { /* already-dead handle from the disposed client */ }
try { reg.OnUnavailable(); } catch { /* never let a subscriber callback abort recovery of the rest */ }
if (_health.State != DriverState.Faulted)
_health = new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead,
$"native-notification replay failed for '{reg.SymbolName}' on {device.Options.HostAddress}");
MarkCommsFault($"native-notification replay failed for '{reg.SymbolName}' on {device.Options.HostAddress}");
}
}
@@ -1586,6 +1659,9 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery
/// <summary>True once an unexpected sum-read chunk failure was logged for this device.</summary>
public volatile bool SumChunkFailureWarned;
/// <summary>Consecutive failed probe ticks; reset by a successful probe. Written only by this device's probe loop.</summary>
public volatile int ConsecutiveProbeFailures;
/// <summary>Latest probe-sampled PLC diagnostics; null before the first sample.</summary>
public TwinCATDeviceDiagnostics? LastDiagnostics { get; set; }
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT.Tests/TwinCATHealthContractTests.cs
+161
View File
@@ -0,0 +1,161 @@
using System.Collections.Concurrent;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Driver.TwinCAT;
namespace ZB.MOM.WW.OtOpcUa.Driver.TwinCAT.Tests;
/// <summary>
/// The driver health contract (#739 H12): data proving the device answered — a native notification or
/// a successful probe — advances <c>LastSuccessfulRead</c>; sustained probe/connect failure degrades with
/// it preserved; a refusal (a write the client cannot express, a replay the PLC refuses) is no comms fault.
/// </summary>
[Trait("Category", "Unit")]
public sealed class TwinCATHealthContractTests
{
private const string Host = "ads://5.23.91.23.1.1:851";
private static TwinCATDriver NewDriver(FakeTwinCATClientFactory factory, TwinCATProbeOptions probe, bool native = true) =>
new(new TwinCATDriverOptions
{
Devices = [new TwinCATDeviceOptions(Host)],
RawTags = TwinCATRawTags.Entries([new TwinCATTagDefinition("Speed", Host, "MAIN.Speed", TwinCATDataType.DInt)]),
Probe = probe,
UseNativeNotifications = native,
}, "drv-1", factory);
private static TwinCATProbeOptions FastProbe => new()
{
Enabled = true,
Interval = TimeSpan.FromMilliseconds(30),
Timeout = TimeSpan.FromMilliseconds(100),
};
/// <summary>A native push advances <c>LastSuccessfulRead</c> and restores Healthy after a comms-fault write.</summary>
[Fact]
public async Task A_native_notification_advances_LastSuccessfulRead_and_restores_Healthy()
{
var fake = new FakeTwinCATClient();
var factory = new FakeTwinCATClientFactory { Customise = () => fake };
var drv = NewDriver(factory, new TwinCATProbeOptions { Enabled = false });
await drv.InitializeAsync("{}", TestContext.Current.CancellationToken);
_ = await drv.SubscribeAsync(["Speed"], TimeSpan.FromMilliseconds(100), TestContext.Current.CancellationToken);
var initStamp = drv.GetHealth().LastSuccessfulRead;
initStamp.ShouldNotBeNull();
// A write that fails on the wire degrades (comms fault) with the stamp preserved.
fake.ThrowOnWrite = true;
_ = await drv.WriteAsync([new WriteRequest("Speed", 1)], TestContext.Current.CancellationToken);
drv.GetHealth().State.ShouldBe(DriverState.Degraded);
drv.GetHealth().LastSuccessfulRead.ShouldBe(initStamp);
await Task.Delay(300, TestContext.Current.CancellationToken);
fake.FireNotification("MAIN.Speed", 7);
var after = drv.GetHealth();
after.State.ShouldBe(DriverState.Healthy);
after.LastSuccessfulRead!.Value.ShouldBeGreaterThan(initStamp!.Value);
}
/// <summary>A successful probe advances <c>LastSuccessfulRead</c> in a driver that serves only notifications.</summary>
[Fact]
public async Task A_successful_probe_advances_LastSuccessfulRead()
{
var factory = new FakeTwinCATClientFactory();
var drv = NewDriver(factory, FastProbe);
await drv.InitializeAsync("{}", TestContext.Current.CancellationToken);
var initStamp = drv.GetHealth().LastSuccessfulRead!.Value;
await WaitForAsync(() => drv.GetHealth().LastSuccessfulRead > initStamp.AddMilliseconds(250), TimeSpan.FromSeconds(3));
drv.GetHealth().LastSuccessfulRead!.Value.ShouldBeGreaterThan(initStamp.AddMilliseconds(250));
drv.GetHealth().State.ShouldBe(DriverState.Healthy);
await drv.ShutdownAsync(TestContext.Current.CancellationToken);
}
/// <summary>
/// An unreachable AMS router (every probe connect throws) degrades the driver after
/// <see cref="TwinCATDriver.ProbeFailuresBeforeDegraded"/> consecutive failures, preserving the stamp —
/// before, the probe caught it and health stayed Healthy forever.
/// </summary>
[Fact]
public async Task Consecutive_probe_failures_degrade_preserving_LastSuccessfulRead()
{
var factory = new FakeTwinCATClientFactory { Customise = () => new FakeTwinCATClient { ThrowOnConnect = true } };
var drv = NewDriver(factory, FastProbe);
await drv.InitializeAsync("{}", TestContext.Current.CancellationToken);
var initStamp = drv.GetHealth().LastSuccessfulRead;
await WaitForAsync(() => drv.GetHealth().State == DriverState.Degraded, TimeSpan.FromSeconds(3));
var health = drv.GetHealth();
health.State.ShouldBe(DriverState.Degraded);
health.LastSuccessfulRead.ShouldBe(initStamp);
factory.Clients.Count.ShouldBeGreaterThanOrEqualTo(TwinCATDriver.ProbeFailuresBeforeDegraded);
await drv.ShutdownAsync(TestContext.Current.CancellationToken);
}
/// <summary>
/// A write the client refuses with <see cref="NotSupportedException"/> is <c>BadNotSupported</c> with no
/// health write (#611: a refusal is not a comms fault). Positive control: a wire failure does degrade.
/// </summary>
[Fact]
public async Task A_NotSupported_write_leaves_health_alone()
{
var fake = new FakeTwinCATClient { ThrowOnWrite = true, Exception = new NotSupportedException("no") };
var factory = new FakeTwinCATClientFactory { Customise = () => fake };
var drv = NewDriver(factory, new TwinCATProbeOptions { Enabled = false }, native: false);
await drv.InitializeAsync("{}", TestContext.Current.CancellationToken);
var result = await drv.WriteAsync([new WriteRequest("Speed", 1)], TestContext.Current.CancellationToken);
result.Single().StatusCode.ShouldBe(TwinCATStatusMapper.BadNotSupported);
drv.GetHealth().State.ShouldBe(DriverState.Healthy);
fake.Exception = new InvalidOperationException("wire");
result = await drv.WriteAsync([new WriteRequest("Speed", 1)], TestContext.Current.CancellationToken);
result.Single().StatusCode.ShouldBe(TwinCATStatusMapper.BadCommunicationError);
drv.GetHealth().State.ShouldBe(DriverState.Degraded);
}
/// <summary>
/// A reconnect replay the PLC REFUSES (an isolated type-mismatch tag re-attempted on every reconnect)
/// republishes the refusal's status and does not degrade. Positive control: a replay that fails on the
/// wire degrades (covered for the same flow in <c>TwinCATReconnectReplayTests</c>).
/// </summary>
[Fact]
public async Task A_refused_replay_republishes_the_refusal_status_without_degrading()
{
var build = 0;
var factory = new FakeTwinCATClientFactory
{
Customise = () =>
{
var c = new FakeTwinCATClient();
if (++build >= 2)
c.AddNotificationErrors["MAIN.Speed"] = new TwinCATTypeMismatchException("MAIN.Speed", "now a STRING");
return c;
},
};
var drv = NewDriver(factory, new TwinCATProbeOptions { Enabled = false });
await drv.InitializeAsync("{}", TestContext.Current.CancellationToken);
var events = new ConcurrentQueue<DataChangeEventArgs>();
drv.OnDataChange += (_, e) => events.Enqueue(e);
_ = await drv.SubscribeAsync(["Speed"], TimeSpan.FromMilliseconds(100), TestContext.Current.CancellationToken);
factory.Clients[0].SimulateWireDrop();
// A write reconnects (and replays) without itself touching health on success.
_ = await drv.WriteAsync([new WriteRequest("Speed", 1)], TestContext.Current.CancellationToken);
factory.Clients.Count.ShouldBe(2);
events.ShouldHaveSingleItem().Snapshot.StatusCode.ShouldBe(TwinCATStatusMapper.BadTypeMismatch);
drv.GetHealth().State.ShouldBe(DriverState.Healthy);
}
private static async Task WaitForAsync(Func<bool> condition, TimeSpan timeout)
{
var deadline = DateTime.UtcNow + timeout;
while (!condition() && DateTime.UtcNow < deadline)
await Task.Delay(20);
}
}