fix(twincat): isolate native registrations per tag and per device (H11)

Native subscribe was all-or-nothing: any registration failure other than a
type refusal disposed every registration and rethrew, so one typo'd symbol or
one unreachable PLC in a multi-device driver served zero data, churned
register/delete every reconcile tick, and left health Healthy.

Now each registration fails on its own: a refusal (the device answered: symbol
not found, PLC not ready, type mismatch) publishes its status with no health
write; any other per-tag failure publishes BadCommunicationError and degrades;
a connect failure marks the device unreachable for the rest of the call and
stores its tags as dead intents. Every failed tag is kept as a dead intent that
the probe-tick retry and the reconnect replay re-register. The ADS client now
classifies a failed symbol resolve / AddDeviceNotification by status instead of
throwing InvalidOperationException for all of them. Cancellation still tears
the partial subscription down.

Refs #739
Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
This commit is contained in:
Joseph Doherty
2026-10-22 14:00:38 -04:00
parent de21a54949
commit 17f4bf6d33
4 changed files with 278 additions and 38 deletions
@@ -523,8 +523,7 @@ internal sealed class AdsTwinCATClient : ITwinCATClient
var (parent, _, _, parentStatus) = await DescribeSymbolAsync(registerPath, cancellationToken)
.ConfigureAwait(false);
if (parentStatus != TwinCATStatusMapper.Good)
throw new InvalidOperationException(
$"Resolving {registerPath} for a notification failed with status 0x{parentStatus:X8}");
throw RegistrationFailure(registerPath, "resolving the parent word", parentStatus);
if (TwinCATBitAccess.CheckParent(bit, parent) is { } bitReason)
throw new TwinCATTypeMismatchException(symbolPath, bitReason);
clrType = typeof(byte[]);
@@ -538,8 +537,7 @@ internal sealed class AdsTwinCATClient : ITwinCATClient
var (shape, array, _, symbolStatus) = await DescribeSymbolAsync(symbolPath, cancellationToken)
.ConfigureAwait(false);
if (symbolStatus != TwinCATStatusMapper.Good)
throw new InvalidOperationException(
$"Resolving {symbolPath} for a notification failed with status 0x{symbolStatus:X8}");
throw RegistrationFailure(symbolPath, "resolving the symbol", symbolStatus);
if (arrayCount is int count)
{
@@ -582,8 +580,8 @@ internal sealed class AdsTwinCATClient : ITwinCATClient
registerPath, settings, userData: null, clrType, anyTypeArgs, cancellationToken)
.ConfigureAwait(false);
if (result.ErrorCode != AdsErrorCode.NoError)
throw new InvalidOperationException(
$"AddDeviceNotificationExAsync failed with ADS error {result.ErrorCode} for {registerPath}");
throw RegistrationFailure(registerPath, $"AddDeviceNotificationEx (ADS error {result.ErrorCode})",
MapAndSignal((uint)result.ErrorCode));
var reg = new NotificationRegistration(symbolPath, type, bitIndex, onChange, this, result.Handle)
{
@@ -594,6 +592,17 @@ internal sealed class AdsTwinCATClient : ITwinCATClient
return reg;
}
/// <summary>
/// Classifies a failed notification registration. A status proving the device ANSWERED (symbol not
/// found, invalid access, PLC not ready, …) is a per-tag refusal the driver isolates without a health
/// write; a comms-class status (<c>BadCommunicationError</c> / <c>BadTimeout</c>) stays an ordinary
/// exception — a comms fault the driver degrades on and retries.
/// </summary>
internal static Exception RegistrationFailure(string symbolPath, string step, uint status) =>
status is TwinCATStatusMapper.BadCommunicationError or TwinCATStatusMapper.BadTimeout
? new InvalidOperationException($"{step} for {symbolPath} failed with status 0x{status:X8}")
: new TwinCATNotificationRefusedException(symbolPath, $"{step} failed with status 0x{status:X8}", status);
/// <summary>
/// Runs on the <see cref="AdsClient"/> AMS router thread. Does the cheap decode (bit
/// extraction, string, array) then enqueues — no driver logic, no consumer callbacks — so a slow consumer
@@ -661,6 +661,9 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery
var id = Interlocked.Increment(ref _nextNativeSubId);
var handle = new NativeSubscriptionHandle(id);
var registrations = new List<NativeRegistration>(fullReferences.Count);
// Devices whose connect failed during THIS call: the rest of their tags are stored as dead
// intents straight away rather than each paying (and logging) its own failed connect.
HashSet<DeviceState>? unreachable = null;
try
{
@@ -686,44 +689,69 @@ 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) => OnDataChange?.Invoke(this,
new DataChangeEventArgs(handle, reference, new DataValueSnapshot(
value, TwinCATStatusMapper.Good, DateTime.UtcNow, DateTime.UtcNow))),
onChange: (_, value) => RaiseDataChange(handle, reference, new DataValueSnapshot(
value, TwinCATStatusMapper.Good, DateTime.UtcNow, DateTime.UtcNow)),
// 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: () => OnDataChange?.Invoke(this,
new DataChangeEventArgs(handle, reference, new DataValueSnapshot(
null, TwinCATStatusMapper.BadCommunicationError, null, DateTime.UtcNow))));
onUnavailable: () => RaiseDataChange(handle, reference, new DataValueSnapshot(
null, TwinCATStatusMapper.BadCommunicationError, null, DateTime.UtcNow)));
registrations.Add(reg);
if (unreachable?.Contains(device) == true)
{
// Same device already failed to connect in this call: keep the intent dead for the
// probe-tick retry / reconnect replay and surface Bad now.
device.NativeRegistrations[reg.Id] = reg;
RaiseDataChange(handle, reference, new DataValueSnapshot(
null, TwinCATStatusMapper.BadCommunicationError, null, DateTime.UtcNow));
continue;
}
try
{
await RegisterNotificationAsync(device, reg, cancellationToken).ConfigureAwait(false);
}
catch (OperationCanceledException) { throw; }
catch (TwinCATNotificationRefusedException refused)
{
// The declared type / array length cannot decode this PLC symbol (ignitionoee audit), or the array is over the native-notification size cap: isolate
// THIS tag to Bad — one WARN, one snapshot carrying the refusal's status
// (BadTypeMismatch / BadOutOfRange) — and keep the rest of the subscription. Its
// intent stays on the device with no live handle, so the probe-tick retry re-checks
// it after a PLC program change.
// The device answered and refused THIS tag (declared type cannot decode the symbol,
// array over the notification cap, symbol not found): isolate it — one WARN, one
// snapshot carrying the refusal's status, no health write (a refusal is not a comms
// fault). Its intent stays on the device with no live handle, so the probe-tick retry
// re-checks it after a PLC program change.
_logger.LogWarning(
"TwinCAT driver '{DriverInstanceId}' refused native notification for '{Reference}': {Reason}",
_driverInstanceId, reference, refused.Message);
OnDataChange?.Invoke(this, new DataChangeEventArgs(handle, reference, new DataValueSnapshot(
null, refused.StatusCode, null, DateTime.UtcNow)));
RaiseDataChange(handle, reference, new DataValueSnapshot(
null, refused.StatusCode, null, DateTime.UtcNow));
}
catch (DeviceUnavailableException unavailable)
{
// The device could not be connected (offline, router down, backoff window open).
// Every tag of this device goes Bad + dead for the retry; other devices carry on.
(unreachable ??= []).Add(device);
MarkCommsFault(unavailable.InnerException?.Message ?? unavailable.Message);
RaiseDataChange(handle, reference, new DataValueSnapshot(
null, TwinCATStatusMapper.BadCommunicationError, null, DateTime.UtcNow));
}
catch (Exception ex)
{
// A per-tag registration failure that is not a refusal (the AddDeviceNotification
// round-trip failed): this tag goes Bad + dead for the retry; the rest keep going.
_logger.LogWarning(ex,
"TwinCAT driver '{DriverInstanceId}' native-notification registration failed for " +
"'{Reference}'; it will be retried on the next healthy probe tick",
_driverInstanceId, reference);
MarkCommsFault(ex.Message);
RaiseDataChange(handle, reference, new DataValueSnapshot(
null, TwinCATStatusMapper.BadCommunicationError, null, DateTime.UtcNow));
}
registrations.Add(reg);
}
}
catch (Exception ex)
catch (OperationCanceledException)
{
// On any registration failure, tear down everything we got so far + rethrow. Leaves
// the subscription in a clean "never existed" state rather than a half-registered
// state the caller has to clean up.
_logger.LogWarning(ex,
"TwinCAT driver '{DriverInstanceId}' native-notification registration failed; " +
"tearing down {Count} partial registrations",
_driverInstanceId, registrations.Count);
// Cancelled mid-subscribe: leave the subscription in a clean "never existed" state rather
// than a half-registered one the caller has to clean up.
foreach (var r in registrations) { try { r.Dispose(); } catch { } }
throw;
}
@@ -732,6 +760,21 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery
return handle;
}
/// <summary>The connect step of a native registration failed — the whole device is unreachable.</summary>
private sealed class DeviceUnavailableException(Exception inner)
: Exception("TwinCAT device unreachable: " + inner.Message, inner);
/// <summary>
/// Records a comms fault: <see cref="DriverState.Degraded"/> preserving <c>LastSuccessfulRead</c>.
/// Never downgrades <see cref="DriverState.Faulted"/>.
/// </summary>
private void MarkCommsFault(string message)
{
var current = _health;
if (current.State != DriverState.Faulted)
_health = new DriverHealth(DriverState.Degraded, current.LastSuccessfulRead, message);
}
/// <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));
@@ -748,7 +791,20 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery
await device.ConnectGate.WaitAsync(ct).ConfigureAwait(false);
try
{
var client = await EnsureConnectedUnderGateAsync(device, ct, timeoutOverride: null).ConfigureAwait(false);
ITwinCATClient client;
try
{
client = await EnsureConnectedUnderGateAsync(device, ct, timeoutOverride: null).ConfigureAwait(false);
}
catch (OperationCanceledException) { throw; }
catch (Exception ex)
{
// Keep the intent dead (no live handle): the replay after the next successful connect
// re-registers it. The caller short-circuits the rest of this device's tags.
device.NativeRegistrations[reg.Id] = reg;
throw new DeviceUnavailableException(ex);
}
ITwinCATNotificationHandle newHandle;
try
{
@@ -756,7 +812,8 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery
reg.SymbolName, reg.DataType, reg.BitIndex, reg.ArrayLength, reg.Interval,
_options.NotificationMaxDelayMs, reg.OnChange, ct).ConfigureAwait(false);
}
catch (TwinCATNotificationRefusedException)
catch (OperationCanceledException) { throw; }
catch
{
// Keep the intent with no live handle (a "dead" registration) so the probe-tick
// retry re-checks it and UnsubscribeAsync still removes it; the caller isolates it.
@@ -56,4 +56,170 @@ public sealed class TwinCATNativeIsolationTests
good.Snapshot.StatusCode.ShouldBe(TwinCATStatusMapper.Good);
good.Snapshot.Value.ShouldBe(42);
}
/// <summary>
/// H11: one tag whose AddDeviceNotification round-trip fails (not a refusal) goes Bad on its own and
/// the subscribe call SUCCEEDS — before, the exception disposed every registration and rethrew, so one
/// bad tag served zero data for the whole driver. The good tags keep delivering (positive control).
/// </summary>
[Fact]
public async Task One_failing_registration_does_not_tear_down_the_subscription()
{
var fake = new FakeTwinCATClient();
fake.AddNotificationErrors["MAIN.B"] = new InvalidOperationException("AddDeviceNotificationEx blew up");
var factory = new FakeTwinCATClientFactory { Customise = () => fake };
var drv = NewNativeDriver(factory,
new TwinCATTagDefinition("A", Host, "MAIN.A", TwinCATDataType.DInt),
new TwinCATTagDefinition("B", Host, "MAIN.B", TwinCATDataType.DInt),
new TwinCATTagDefinition("C", Host, "MAIN.C", TwinCATDataType.DInt));
await drv.InitializeAsync("{}", TestContext.Current.CancellationToken);
var events = new ConcurrentQueue<DataChangeEventArgs>();
drv.OnDataChange += (_, e) => events.Enqueue(e);
var handle = await drv.SubscribeAsync(["A", "B", "C"], TimeSpan.FromMilliseconds(100),
TestContext.Current.CancellationToken);
fake.Notifications.Select(n => n.SymbolPath).ShouldBe(["MAIN.A", "MAIN.C"], ignoreOrder: true);
events.ShouldHaveSingleItem().FullReference.ShouldBe("B");
events.Single().Snapshot.StatusCode.ShouldBe(TwinCATStatusMapper.BadCommunicationError);
fake.FireNotification("MAIN.C", 3);
events.Last().FullReference.ShouldBe("C");
events.Last().Snapshot.StatusCode.ShouldBe(TwinCATStatusMapper.Good);
// The failed tag is kept as a dead intent and unsubscribe still removes everything.
drv.GetDeviceState(Host)!.NativeRegistrations.Count.ShouldBe(3);
await drv.UnsubscribeAsync(handle, TestContext.Current.CancellationToken);
drv.GetDeviceState(Host)!.NativeRegistrations.ShouldBeEmpty();
fake.Notifications.ShouldBeEmpty();
}
/// <summary>
/// H11: a refusal (symbol not found — the device answered) isolates the tag with the refusal's own
/// status and leaves driver health alone; the good tag delivers.
/// </summary>
[Fact]
public async Task A_refused_symbol_is_isolated_without_a_health_write()
{
var fake = new FakeTwinCATClient();
fake.AddNotificationErrors["MAIN.Typo"] = new TwinCATNotificationRefusedException(
"MAIN.Typo", "resolving the symbol failed", TwinCATStatusMapper.BadNodeIdUnknown);
var factory = new FakeTwinCATClientFactory { Customise = () => fake };
var drv = NewNativeDriver(factory,
new TwinCATTagDefinition("Typo", Host, "MAIN.Typo", TwinCATDataType.DInt),
new TwinCATTagDefinition("Good", Host, "MAIN.Good", TwinCATDataType.DInt));
await drv.InitializeAsync("{}", TestContext.Current.CancellationToken);
var events = new ConcurrentQueue<DataChangeEventArgs>();
drv.OnDataChange += (_, e) => events.Enqueue(e);
_ = await drv.SubscribeAsync(["Typo", "Good"], TimeSpan.FromMilliseconds(100), TestContext.Current.CancellationToken);
events.ShouldHaveSingleItem().Snapshot.StatusCode.ShouldBe(TwinCATStatusMapper.BadNodeIdUnknown);
drv.GetHealth().State.ShouldBe(DriverState.Healthy);
fake.FireNotification("MAIN.Good", 1);
events.Last().Snapshot.StatusCode.ShouldBe(TwinCATStatusMapper.Good);
}
/// <summary>
/// H11: in a two-device driver an unreachable device only blacks out ITS tags (Bad + dead intents,
/// health Degraded); the reachable device's tags register and deliver.
/// </summary>
[Fact]
public async Task An_unreachable_device_blacks_out_only_its_own_tags()
{
const string hostB = "ads://5.23.91.24.1.1:851";
var build = 0;
var factory = new FakeTwinCATClientFactory
{
// Device A's client is created first and works; device B's connect throws.
Customise = () => new FakeTwinCATClient { ThrowOnConnect = ++build == 2 },
};
var drv = new TwinCATDriver(new TwinCATDriverOptions
{
Devices = [new TwinCATDeviceOptions(Host), new TwinCATDeviceOptions(hostB)],
RawTags = TwinCATRawTags.Entries(
[
new TwinCATTagDefinition("A1", Host, "MAIN.A1", TwinCATDataType.DInt),
new TwinCATTagDefinition("B1", hostB, "MAIN.B1", TwinCATDataType.DInt),
new TwinCATTagDefinition("B2", hostB, "MAIN.B2", TwinCATDataType.DInt),
new TwinCATTagDefinition("A2", Host, "MAIN.A2", TwinCATDataType.DInt),
]),
Probe = new TwinCATProbeOptions { Enabled = false },
UseNativeNotifications = true,
}, "drv-1", factory);
await drv.InitializeAsync("{}", TestContext.Current.CancellationToken);
var events = new ConcurrentQueue<DataChangeEventArgs>();
drv.OnDataChange += (_, e) => events.Enqueue(e);
_ = await drv.SubscribeAsync(["A1", "B1", "B2", "A2"], TimeSpan.FromMilliseconds(100),
TestContext.Current.CancellationToken);
factory.Clients.Count.ShouldBe(2); // B2 did not attempt a second connect
factory.Clients[0].Notifications.Select(n => n.SymbolPath).ShouldBe(["MAIN.A1", "MAIN.A2"], ignoreOrder: true);
events.Select(e => e.FullReference).ShouldBe(["B1", "B2"], ignoreOrder: true);
events.ShouldAllBe(e => e.Snapshot.StatusCode == TwinCATStatusMapper.BadCommunicationError);
drv.GetDeviceState(hostB)!.NativeRegistrations.Count.ShouldBe(2);
drv.GetHealth().State.ShouldBe(DriverState.Degraded);
factory.Clients[0].FireNotification("MAIN.A2", 22);
events.Last().FullReference.ShouldBe("A2");
events.Last().Snapshot.StatusCode.ShouldBe(TwinCATStatusMapper.Good);
}
/// <summary>
/// H11: the dead intent left by a failed registration is re-registered by the probe-tick retry, and
/// its first push then arrives Good — the subscription heals without a re-subscribe.
/// </summary>
[Fact]
public async Task A_failed_registration_is_recovered_by_the_probe_tick_retry()
{
var fake = new FakeTwinCATClient();
fake.AddNotificationErrors["MAIN.B"] = new InvalidOperationException("transient");
var factory = new FakeTwinCATClientFactory { Customise = () => fake };
var drv = new TwinCATDriver(new TwinCATDriverOptions
{
Devices = [new TwinCATDeviceOptions(Host)],
RawTags = TwinCATRawTags.Entries([new TwinCATTagDefinition("B", Host, "MAIN.B", TwinCATDataType.DInt)]),
Probe = new TwinCATProbeOptions
{
Enabled = true,
Interval = TimeSpan.FromMilliseconds(40),
Timeout = TimeSpan.FromMilliseconds(100),
},
UseNativeNotifications = true,
}, "drv-1", factory);
await drv.InitializeAsync("{}", TestContext.Current.CancellationToken);
var events = new ConcurrentQueue<DataChangeEventArgs>();
drv.OnDataChange += (_, e) => events.Enqueue(e);
_ = await drv.SubscribeAsync(["B"], TimeSpan.FromMilliseconds(100), TestContext.Current.CancellationToken);
fake.AddNotificationErrors.Clear();
await WaitForAsync(() => fake.Notifications.Count == 1, TimeSpan.FromSeconds(3));
fake.Notifications.Count.ShouldBe(1);
fake.FireNotification("MAIN.B", 5);
events.Last().Snapshot.StatusCode.ShouldBe(TwinCATStatusMapper.Good);
await drv.ShutdownAsync(TestContext.Current.CancellationToken);
}
/// <summary>A comms-class registration status stays an exception; any other status is a per-tag refusal.</summary>
[Fact]
public void Registration_failures_are_classified_by_status()
{
AdsTwinCATClient.RegistrationFailure("MAIN.X", "step", TwinCATStatusMapper.BadNodeIdUnknown)
.ShouldBeOfType<TwinCATNotificationRefusedException>().StatusCode.ShouldBe(TwinCATStatusMapper.BadNodeIdUnknown);
AdsTwinCATClient.RegistrationFailure("MAIN.X", "step", TwinCATStatusMapper.BadOutOfService)
.ShouldBeOfType<TwinCATNotificationRefusedException>();
AdsTwinCATClient.RegistrationFailure("MAIN.X", "step", TwinCATStatusMapper.BadCommunicationError)
.ShouldBeOfType<InvalidOperationException>();
AdsTwinCATClient.RegistrationFailure("MAIN.X", "step", TwinCATStatusMapper.BadTimeout)
.ShouldBeOfType<InvalidOperationException>();
}
private static async Task WaitForAsync(Func<bool> condition, TimeSpan timeout)
{
var deadline = DateTime.UtcNow + timeout;
while (!condition() && DateTime.UtcNow < deadline)
await Task.Delay(20);
}
}
@@ -98,13 +98,14 @@ public sealed class TwinCATNativeNotificationTests
events.Count.ShouldBe(afterUnsub);
}
/// <summary>Verifies that native subscribe failure mid-registration cleans up partial state.</summary>
/// <summary>
/// A registration failing mid-subscribe isolates that one tag (#739 H11): the first registration
/// stays live, the second is kept as a dead intent and goes Bad, and the call does not throw.
/// </summary>
[Fact]
public async Task Native_subscribe_failure_mid_registration_cleans_up_partial_state()
public async Task Native_subscribe_failure_mid_registration_isolates_the_failed_tag()
{
// Fail-on-second-call fake — first AddNotificationAsync succeeds, second throws.
// Subscribe's catch block must tear the first one down before rethrowing so no zombie
// notification lingers.
var fake = new FailAfterNAddsFake(new AbTagParamsIrrelevant(), succeedBefore: 1);
var factory = new FakeTwinCATClientFactory { Customise = () => fake };
var drv = new TwinCATDriver(new TwinCATDriverOptions
@@ -120,12 +121,19 @@ public sealed class TwinCATNativeNotificationTests
}, "drv-1", factory);
await drv.InitializeAsync("{}", CancellationToken.None);
await Should.ThrowAsync<InvalidOperationException>(() =>
drv.SubscribeAsync(["A", "B"], TimeSpan.FromMilliseconds(100), CancellationToken.None));
var events = new ConcurrentQueue<DataChangeEventArgs>();
drv.OnDataChange += (_, e) => events.Enqueue(e);
var handle = await drv.SubscribeAsync(["A", "B"], TimeSpan.FromMilliseconds(100), CancellationToken.None);
// First registration succeeded then got torn down by the catch; second threw.
fake.AddCallCount.ShouldBe(2);
fake.Notifications.Count.ShouldBe(0); // partial handle cleaned up
fake.Notifications.ShouldHaveSingleItem().SymbolPath.ShouldBe("MAIN.A");
var bad = events.ShouldHaveSingleItem();
bad.FullReference.ShouldBe("B");
bad.Snapshot.StatusCode.ShouldBe(TwinCATStatusMapper.BadCommunicationError);
await drv.UnsubscribeAsync(handle, CancellationToken.None);
fake.Notifications.ShouldBeEmpty();
}
/// <summary>