Merge hw6-662 (#662 native-alarm feed survives data re-subscribes; reconcile retries alarms)
v2-ci / build (push) Successful in 5m1s
v2-ci / unit-tests (push) Failing after 3h10m35s

Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
This commit is contained in:
Joseph Doherty
2026-10-09 07:35:50 -04:00
7 changed files with 358 additions and 26 deletions
+5
View File
@@ -1162,6 +1162,11 @@ host logger factory (wired in `DriverFactoryBootstrap`; the FOCAS and TwinCAT fa
FOCAS's is PAUSED, its told record kept, #647) — the device-state half of `ShutdownAsync` only; the poll engine is kept and the actor re-subscribes FOCAS's is PAUSED, its told record kept, #647) — the device-state half of `ShutdownAsync` only; the poll engine is kept and the actor re-subscribes
after every successful init. AbCip publishes fresh tag tables by reference swap (a second init used to throw after every successful init. AbCip publishes fresh tag tables by reference swap (a second init used to throw
"tag name collision" on `AdoptConfigDuringInit`); S7 swaps `Plc` under `_gate`. "tag name collision" on `AdoptConfigDuringInit`); S7 swaps `Plc` under `_gate`.
- **Data and alarm subscriptions are independent in `DriverInstanceActor` (#662):** a data re-subscribe (every
deploy's bulk push to a kept driver) detaches DATA only (`DetachDataSubscription`) — it used to unhook
`OnAlarmEvent` too, silently killing the native-alarm feed until a respawn. The health-poll reconcile
re-attaches/re-subscribes alarms as well as data, so a failed alarm subscribe is retried every tick.
`docs/DriverLifecycle.md`.
- **S7 (#632/#638):** STRING/WSTRING/CHAR/WCHAR and DTL/DATE/TOD/TIME/S5TIME codecs (`S7ValueCodecs`; TOD/TIME/S5TIME - **S7 (#632/#638):** STRING/WSTRING/CHAR/WCHAR and DTL/DATE/TOD/TIME/S5TIME codecs (`S7ValueCodecs`; TOD/TIME/S5TIME
surface as Int32 ms); an undecodable buffer or unrepresentable write is `BadOutOfRange` with no health impact. surface as Int32 ms); an undecodable buffer or unrepresentable write is `BadOutOfRange` with no health impact.
Optional `localTsap`/`remoteTsap`/`connectionType` override the slot TSAP (`remoteTsap` ⊕ `connectionType`). Test Optional `localTsap`/`remoteTsap`/`connectionType` override the slot TSAP (`remoteTsap` ⊕ `connectionType`). Test
+13
View File
@@ -242,6 +242,19 @@ Things to know before touching any of this:
re-subscribing** so a filtering driver does not accumulate two overlapping ones. There is still re-subscribing** so a filtering driver does not accumulate two overlapping ones. There is still
deliberately no unsubscribe-on-**empty**: a removed alarm tag's condition node is torn down on the deliberately no unsubscribe-on-**empty**: a removed alarm tag's condition node is torn down on the
address-space rebuild and `ForwardNativeAlarm` drops any transition whose `ConditionId` no longer maps. address-space rebuild and `ForwardNativeAlarm` drops any transition whose `ConditionId` no longer maps.
- ⚠️ **A data re-subscribe must not touch the alarm feed (#662).** Every deploy's bulk push re-sends
`SetDesiredSubscriptions` to every kept driver, and `HandleSubscribeAsync` drops the prior DATA
subscription first. That teardown used to be the full `DetachSubscription`, which also unhooked
`OnAlarmEvent` and forgot the alarm handle — so on any deploy that did **not** respawn a driver (e.g. one
that only disabled another driver) its alarm subscription kept logging "established" while every event
reached nobody, until a reconnect or respawn. The data path now uses `DetachDataSubscription`; only a lost
connection (`DisconnectObserved` / `ForceReconnect`) and `PostStop` detach both. The **#488 periodic
reconcile covers alarms too** (`ReconcileAlarmSubscription`): it re-attaches a missing `OnAlarmEvent`
handler and re-subscribes when the live alarm handle does not match the desired set, so an alarm subscribe
that failed on (re)connect — a session not yet ready after an outage — is retried every health-poll tick
instead of gating the feed for the life of the connection. OpcUaClient removes a half-built alarm
subscription from its session when the subscribe throws, so those retries do not accumulate.
End-to-end pin: `NativeAlarmFeedSurvivesDeployTests`.
- **Restart tests must assert a FUNCTION, never existence.** `RestartDriver_respawns_the_child_without_an_actor_name_collision` - **Restart tests must assert a FUNCTION, never existence.** `RestartDriver_respawns_the_child_without_an_actor_name_collision`
asserts only that a child exists after a restart, and it passed green against the dead driver 03/S16 asserts only that a child exists after a restart, and it passed green against the dead driver 03/S16
describes. It is kept — its own subject, the `InvalidActorNameException`, is real — but the load-bearing describes. It is kept — its own subject, the `InvalidActorNameException`, is real — but the load-bearing
@@ -1701,6 +1701,10 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
AttributeId = Attributes.Value, AttributeId = Attributes.Value,
}); });
// Set by CreateAlarmEventItemAsync so a failure after the handler is attached can detach it (#662).
MonitoredItem? eventItem = null;
MonitoredItemNotificationEventHandler? notifHandler = null;
await _gate.WaitAsync(cancellationToken).ConfigureAwait(false); await _gate.WaitAsync(cancellationToken).ConfigureAwait(false);
try try
{ {
@@ -1710,9 +1714,44 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
?? throw new InvalidOperationException( ?? throw new InvalidOperationException(
"OpcUaClient session was lost before the alarm subscription could be created."); "OpcUaClient session was lost before the alarm subscription could be created.");
session.AddSubscription(subscription); session.AddSubscription(subscription);
try
{
await CreateAlarmEventItemAsync().ConfigureAwait(false);
}
catch
{
// #662: the host retries a failed alarm subscribe from its reconcile, so a half-built subscription
// left on the session would accumulate one per retry for the length of an outage. Detach the
// handler first — the remove is best-effort and can itself fail against a broken server — and
// bound it, since _gate is held. OnEventNotification also drops anything from an unregistered id.
if (eventItem is not null && notifHandler is not null)
{
try { eventItem.Notification -= notifHandler; }
catch { /* best-effort */ }
}
try
{
using var removeCts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
await session.RemoveSubscriptionAsync(subscription, removeCts.Token).ConfigureAwait(false);
}
catch { /* best-effort — the session may be the thing that failed */ }
throw;
}
// #619: a new event item only reports FUTURE transitions — a condition already active upstream would
// stay invisible until it next changed. ConditionRefresh re-delivers every retained condition's
// current state through the normal path (deduplicated against the last-told state).
await TryConditionRefreshAsync(subscription, cancellationToken).ConfigureAwait(false);
}
finally { _gate.Release(); }
return handle;
async Task CreateAlarmEventItemAsync()
{
await subscription.CreateAsync(cancellationToken).ConfigureAwait(false); await subscription.CreateAsync(cancellationToken).ConfigureAwait(false);
var eventItem = new MonitoredItem(telemetry: null!, new MonitoredItemOptions eventItem = new MonitoredItem(telemetry: null!, new MonitoredItemOptions
{ {
DisplayName = "Server/Events", DisplayName = "Server/Events",
StartNodeId = ObjectIds.Server, StartNodeId = ObjectIds.Server,
@@ -1729,22 +1768,14 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
// can `-=` it later. The lambda captures `handle` and `sourceFilter`, so without // can `-=` it later. The lambda captures `handle` and `sourceFilter`, so without
// the explicit detach the SDK's invocation list keeps the driver instance alive // the explicit detach the SDK's invocation list keeps the driver instance alive
// until the session itself is disposed. // until the session itself is disposed.
MonitoredItemNotificationEventHandler notifHandler = (mi, args) => notifHandler = (mi, args) =>
OnEventNotification(handle, sourceFilter, mi, args); OnEventNotification(handle, sourceFilter, mi, args);
eventItem.Notification += notifHandler; eventItem.Notification += notifHandler;
subscription.AddItem(eventItem); subscription.AddItem(eventItem);
await subscription.CreateItemsAsync(cancellationToken).ConfigureAwait(false); await subscription.CreateItemsAsync(cancellationToken).ConfigureAwait(false);
_alarmSubscriptions[id] = new RemoteAlarmSubscription(subscription, handle, eventItem, notifHandler); _alarmSubscriptions[id] = new RemoteAlarmSubscription(subscription, handle, eventItem, notifHandler);
// #619: a new event item only reports FUTURE transitions — a condition already active upstream would
// stay invisible until it next changed. ConditionRefresh re-delivers every retained condition's
// current state through the normal path (deduplicated against the last-told state).
await TryConditionRefreshAsync(subscription, cancellationToken).ConfigureAwait(false);
} }
finally { _gate.Release(); }
return handle;
} }
/// <inheritdoc /> /// <inheritdoc />
@@ -1882,6 +1913,10 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
MonitoredItemNotificationEventArgs args) MonitoredItemNotificationEventArgs args)
{ {
if (args.NotificationValue is not EventFieldList efl) return; if (args.NotificationValue is not EventFieldList efl) return;
// #662 review: only a subscription SubscribeAlarmsAsync completed (and nothing has since released) may
// deliver. A half-built one whose cleanup failed against a broken server can survive on the session and
// be transferred by an SDK reconnect; without this it would feed RouteAlarmEvent alongside the real one.
if (!_alarmSubscriptions.ContainsKey(handle.Id)) return;
RouteAlarmEvent(efl.EventFields, sourceFilter, handle); RouteAlarmEvent(efl.EventFields, sourceFilter, handle);
} }
@@ -281,6 +281,17 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
/// newly-added alarm for the life of the connection. Empty whenever no handle is held.</para></summary> /// newly-added alarm for the life of the connection. Empty whenever no handle is held.</para></summary>
private IReadOnlyList<string> _subscribedAlarmRefs = Array.Empty<string>(); private IReadOnlyList<string> _subscribedAlarmRefs = Array.Empty<string>();
/// <summary>#662 — consecutive <see cref="HandleSubscribeAlarmsAsync"/> failures; reset on success and on
/// <see cref="DetachAlarmSource"/> (a new connection starts a new streak).</summary>
private int _alarmSubscribeFailures;
/// <summary>#662 — health-poll ticks the alarm reconcile still skips before its next retry (backoff).</summary>
private int _alarmRetrySkipTicks;
/// <summary>#662 — backoff cap: at the production 30 s tick a failing alarm subscribe is retried at least
/// every 4 minutes.</summary>
private const int MaxAlarmRetrySkipTicks = 7;
/// <summary> /// <summary>
/// Gets or sets the timer scheduler for scheduling reconnection attempts. /// Gets or sets the timer scheduler for scheduling reconnection attempts.
/// </summary> /// </summary>
@@ -690,6 +701,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
/// </remarks> /// </remarks>
private void ReconcileSubscription() private void ReconcileSubscription()
{ {
ReconcileAlarmSubscription();
if (_driver is not ISubscribable) return; if (_driver is not ISubscribable) return;
if (_desiredRefs.Count == 0 || _subscriptionHandle is not null) return; if (_desiredRefs.Count == 0 || _subscriptionHandle is not null) return;
@@ -699,6 +711,43 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
Self.Tell(new Subscribe(_desiredRefs, _desiredInterval)); Self.Tell(new Subscribe(_desiredRefs, _desiredInterval));
} }
/// <summary>
/// #662 — the native-alarm half of <see cref="ReconcileSubscription"/>. The data reconcile above never
/// looked at alarms, so an alarm subscribe that failed on (re)connect — e.g. a session not yet ready
/// after an outage — left the feed gated for the life of the connection, with the driver Healthy and
/// its data streaming. Re-asserts both halves the feed needs: the <c>OnAlarmEvent</c> handler
/// (idempotent attach) and a subscription matching the desired set (<see cref="SubscribeDesiredAlarms"/>
/// is a no-op while the live handle already matches). Gated on <see cref="IAlarmSource"/> + a
/// non-empty desired set, so it costs nothing on a driver with no alarm tags.
/// <para>A persistently failing subscribe backs off: after the n-th consecutive failure the reconcile
/// skips 2^(n-1) − 1 ticks, capped at <see cref="MaxAlarmRetrySkipTicks"/>, because a HANGING subscribe
/// holds the mailbox for its full 10 s bound (writes queued behind it can miss the host's 8 s Ask). A
/// Connected entry or a deploy's desired-set push still retries immediately — only the timer backs off.</para>
/// </summary>
private void ReconcileAlarmSubscription()
{
if (_driver is not IAlarmSource || _desiredAlarmRefs.Count == 0) return;
if (_alarmEventHandler is null)
{
// Unreachable while every Connected entry attaches — so if this ever logs, something regressed #662.
_log.Warning("DriverInstance {Id}: connected with desired alarm refs but no native-alarm handler — re-attaching (periodic reconcile)",
_driverInstanceId);
AttachAlarmSource();
}
if (_alarmRetrySkipTicks > 0)
{
_alarmRetrySkipTicks--;
return;
}
if (AlarmSubscriptionIsStale())
{
_log.Info(
"DriverInstance {Id}: connected with {Count} desired alarm ref(s) but no matching native-alarm subscription — re-subscribing (periodic reconcile)",
_driverInstanceId, _desiredAlarmRefs.Count);
Self.Tell(new SubscribeAlarms());
}
}
private void Reconnecting() private void Reconnecting()
{ {
// #477 — announce comms loss to the host so it annotates this driver's native alarm conditions Bad // #477 — announce comms loss to the host so it annotates this driver's native alarm conditions Bad
@@ -1002,7 +1051,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
} }
catch (Exception ex) catch (Exception ex)
{ {
DetachSubscription(); DetachDataSubscription();
_log.Warning(ex, "DriverInstance {Id}: subscribe failed", _driverInstanceId); _log.Warning(ex, "DriverInstance {Id}: subscribe failed", _driverInstanceId);
replyTo.Tell(new SubscriptionFailed(ex.Message)); replyTo.Tell(new SubscriptionFailed(ex.Message));
} }
@@ -1012,7 +1061,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
{ {
if (_driver is not ISubscribable subscribable || _subscriptionHandle is null) if (_driver is not ISubscribable subscribable || _subscriptionHandle is null)
{ {
DetachSubscription(); DetachDataSubscription();
return; return;
} }
try try
@@ -1031,17 +1080,32 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
} }
finally finally
{ {
DetachSubscription(); DetachDataSubscription();
} }
} }
/// <summary>Tear down the data-change (per-tag AND batch) + native-alarm event handlers + null the handle. /// <summary>Tear down the data-change (per-tag AND batch) + native-alarm event handlers + null both handles.
/// Called from the Unsubscribe path, on PostStop, and on Connected → Reconnecting transitions so a stale /// Called on PostStop and on Connected → Reconnecting transitions so a stale handler doesn't push
/// handler doesn't push data-change / alarm events to an actor that has lost its driver connection. /// data-change / alarm events to an actor that has lost its driver connection.
/// <para>Not for the data Unsubscribe / re-subscribe path — that uses <see cref="DetachDataSubscription"/>
/// (#662).</para></summary>
private void DetachSubscription()
{
DetachDataSubscription();
DetachAlarmSource();
}
/// <summary>Tear down the data-change (per-tag AND batch) handlers + null the data handle, leaving the
/// native-alarm handler and handle alone.
/// <para>#662: the data Unsubscribe / re-subscribe path used to run the full <see cref="DetachSubscription"/>,
/// so every deploy that re-subscribed a Connected driver's DATA also unhooked <c>OnAlarmEvent</c> and
/// forgot the alarm handle — the alarm subscription still logged "established" but its events reached
/// nobody until a reconnect or respawn. The two subscriptions are independent at the driver; only a lost
/// connection ends both.</para>
/// <para>Both data-change fields are cleared unconditionally even though only one is ever attached — the /// <para>Both data-change fields are cleared unconditionally even though only one is ever attached — the
/// IDriver object outlives the actor when the host respawns a child around it, so a handler left on the /// IDriver object outlives the actor when the host respawns a child around it, so a handler left on the
/// driver's event would keep Telling a dead actor.</para></summary> /// driver's event would keep Telling a dead actor.</para></summary>
private void DetachSubscription() private void DetachDataSubscription()
{ {
if (_driver is ISubscribable subscribable && _dataChangeHandler is not null) if (_driver is ISubscribable subscribable && _dataChangeHandler is not null)
{ {
@@ -1054,7 +1118,6 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
} }
_batchChangeHandler = null; _batchChangeHandler = null;
_subscriptionHandle = null; _subscriptionHandle = null;
DetachAlarmSource();
} }
/// <summary>Subscribe the driver's native alarm event (if it is an <see cref="IAlarmSource"/>), /// <summary>Subscribe the driver's native alarm event (if it is an <see cref="IAlarmSource"/>),
@@ -1206,6 +1269,8 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
// across reconnects. The gate (Count > 0) and the one-shot fan-out are unchanged. // across reconnects. The gate (Count > 0) and the one-shot fan-out are unchanged.
_alarmSubscriptionHandle = null; _alarmSubscriptionHandle = null;
_subscribedAlarmRefs = Array.Empty<string>(); _subscribedAlarmRefs = Array.Empty<string>();
_alarmSubscribeFailures = 0;
_alarmRetrySkipTicks = 0;
} }
/// <summary>Establish the native-alarm subscription that un-gates an <see cref="IAlarmSource"/> driver's /// <summary>Establish the native-alarm subscription that un-gates an <see cref="IAlarmSource"/> driver's
@@ -1229,7 +1294,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
/// <summary>Calls the driver's <see cref="IAlarmSource.SubscribeAlarmsAsync"/> (bounded) to register the /// <summary>Calls the driver's <see cref="IAlarmSource.SubscribeAlarmsAsync"/> (bounded) to register the
/// alarm subscription that un-gates its native feed, caching the returned handle. Re-checks the guard /// alarm subscription that un-gates its native feed, caching the returned handle. Re-checks the guard
/// (the desired set may have cleared, or another SubscribeAlarms may have already established a handle, /// (the desired set may have cleared, or another SubscribeAlarms may have already established a handle,
/// while this was queued). Failures are logged and retried on the next Connected entry — the feed simply /// while this was queued). Failures are logged and retried on the next Connected entry or health-poll reconcile tick (#662) — the feed
/// stays gated until then.</summary> /// stays gated until then.</summary>
private async Task HandleSubscribeAlarmsAsync(SubscribeAlarms _) private async Task HandleSubscribeAlarmsAsync(SubscribeAlarms _)
{ {
@@ -1271,13 +1336,24 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
async ct => await src.SubscribeAlarmsAsync(refs, ct), async ct => await src.SubscribeAlarmsAsync(refs, ct),
cts.Token); cts.Token);
_subscribedAlarmRefs = refs; _subscribedAlarmRefs = refs;
_log.Info("DriverInstance {Id}: native-alarm subscription established for {Count} alarm ref(s) ({Diag})", _log.Info("DriverInstance {Id}: native-alarm subscription established for {Count} alarm ref(s) ({Diag}){After}",
_driverInstanceId, refs.Count, _alarmSubscriptionHandle.DiagnosticId); _driverInstanceId, refs.Count, _alarmSubscriptionHandle.DiagnosticId,
_alarmSubscribeFailures > 0 ? $" after {_alarmSubscribeFailures} failed attempt(s)" : "");
_alarmSubscribeFailures = 0;
_alarmRetrySkipTicks = 0;
} }
catch (Exception ex) catch (Exception ex)
{ {
_log.Warning(ex, "DriverInstance {Id}: native-alarm subscription failed — feed stays gated until reconnect", _alarmSubscribeFailures++;
_driverInstanceId); _alarmRetrySkipTicks = Math.Min((1 << Math.Min(_alarmSubscribeFailures - 1, 30)) - 1, MaxAlarmRetrySkipTicks);
// The stack once per failure streak; the retries that follow repeat the same cause, so they log
// just the message rather than a full trace every tick for the length of an outage.
if (_alarmSubscribeFailures == 1)
_log.Warning(ex, "DriverInstance {Id}: native-alarm subscription failed — feed stays gated until the periodic reconcile retries it",
_driverInstanceId);
else
_log.Warning("DriverInstance {Id}: native-alarm subscription retry {Attempt} failed ({Reason}) — next retry in {Ticks} health-poll tick(s)",
_driverInstanceId, _alarmSubscribeFailures, ex.Message, _alarmRetrySkipTicks + 1);
} }
} }
@@ -218,6 +218,88 @@ public sealed class DriverInstanceActorNativeAlarmTests : RuntimeActorTestBase
parent.ExpectNoMsg(TimeSpan.FromMilliseconds(100)); parent.ExpectNoMsg(TimeSpan.FromMilliseconds(100));
} }
/// <summary>
/// Gitea #662 case 1 — re-subscribing a Connected driver's DATA (what every deploy's
/// <c>SetDesiredSubscriptions</c> push does) must leave the native-alarm feed alone. The data
/// re-subscribe drops the prior data subscription first, and that teardown used to run the full
/// detach — unhooking <c>OnAlarmEvent</c> and forgetting the alarm handle — so the alarm subscription
/// stayed "established" while every event it carried reached nobody until a reconnect.
/// </summary>
[Fact]
public async Task A_data_re_subscribe_keeps_the_native_alarm_feed_attached()
{
var driver = new AlarmSourceStubDriver();
var parent = CreateTestProbe();
parent.IgnoreMessages(m => m is DriverInstanceActor.ConnectivityChanged);
var actor = parent.ChildActorOf(DriverInstanceActor.Props(driver));
actor.Tell(new DriverInstanceActor.InitializeRequested("{}"));
AwaitCondition(() => driver.AlarmSubscriberCount == 1, TimeSpan.FromSeconds(2));
actor.Tell(new DriverInstanceActor.SetDesiredSubscriptions(
new[] { "tag-a" }, TimeSpan.FromMilliseconds(100), new[] { "Temp.HiHi" }));
AwaitCondition(() => driver.SubscribeAlarmsCallCount == 1, TimeSpan.FromSeconds(2));
// A second data subscribe while the first handle is live — the drop-then-resubscribe path.
await actor.Ask<DriverInstanceActor.SubscriptionEstablished>(
new DriverInstanceActor.Subscribe(new[] { "tag-a", "tag-b" }, TimeSpan.FromMilliseconds(100)),
TimeSpan.FromSeconds(3));
driver.AlarmSubscriberCount.ShouldBe(1, "the data re-subscribe must not unhook OnAlarmEvent");
driver.SubscribeAlarmsCallCount.ShouldBe(1,
"the alarm handle is still live, so nothing should re-issue (and duplicate) the driver-side subscription");
driver.RaiseAlarm(new AlarmEventArgs(
new StubAlarmHandle(), "Temp.HiHi", "Temp.HiHi", "T", "after re-subscribe", AlarmSeverity.High,
DateTime.UtcNow, Kind: AlarmTransitionKind.Raise));
parent.FishForMessage<DriverInstanceActor.AttributeAlarmPublished>(_ => true, TimeSpan.FromSeconds(2))
.Args.ConditionId.ShouldBe("Temp.HiHi");
}
/// <summary>
/// Gitea #662 case 2 — an alarm subscribe that fails on (re)connect (a session not yet ready after an
/// outage) is retried by the periodic reconcile while the driver stays Connected. Before #662 the
/// reconcile looked at data only, so the feed stayed gated for the life of the connection. Nothing
/// else here can produce the second call: no reconnect, no deploy.
/// </summary>
/// <param name="failures">Leading failures; 3 crosses the backoff (skips 0, 1, then 3 ticks), proving the
/// retries keep coming rather than giving up.</param>
[Theory]
[InlineData(1)]
[InlineData(3)]
public void A_failed_alarm_subscribe_is_retried_by_the_periodic_reconcile(int failures)
{
var driver = new AlarmSourceStubDriver { AlarmSubscribeFailuresBeforeSuccess = failures };
var publisher = new CountingHealthPublisher();
var parent = CreateTestProbe();
parent.IgnoreMessages(m => m is DriverInstanceActor.ConnectivityChanged);
var actor = parent.ChildActorOf(DriverInstanceActor.Props(
driver, healthPublisher: publisher, healthPollInterval: TimeSpan.FromMilliseconds(100)));
actor.Tell(new DriverInstanceActor.SetDesiredSubscriptions(
Array.Empty<string>(), TimeSpan.FromMilliseconds(100), new[] { "Temp.HiHi" }));
actor.Tell(new DriverInstanceActor.InitializeRequested("{}"));
AwaitCondition(() => driver.SubscribeAlarmsCallCount >= 1, PresenceBudget);
AwaitCondition(() => driver.SubscribeAlarmsCallCount >= failures + 1, PresenceBudget);
driver.LastAlarmRefs.ShouldBe(new[] { "Temp.HiHi" });
driver.RaiseAlarm(new AlarmEventArgs(
new StubAlarmHandle(), "Temp.HiHi", "Temp.HiHi", "T", "after retry", AlarmSeverity.High,
DateTime.UtcNow, Kind: AlarmTransitionKind.Raise));
parent.FishForMessage<DriverInstanceActor.AttributeAlarmPublished>(_ => true, TimeSpan.FromSeconds(2))
.Args.ConditionId.ShouldBe("Temp.HiHi");
// Converged: a live, matching handle is left alone on later ticks. POSITIVE CONTROL: each tick
// publishes one health snapshot, so waiting for four more makes the unchanged count mean "the
// reconcile ran and declined", not "no tick had fired yet". It relies on the base StubDriver's
// GetHealth() stamping UtcNow, which defeats the publish dedup so EVERY tick publishes — a stub with
// stable health would publish once and this wait would time out (loudly, not vacuously).
var settled = driver.SubscribeAlarmsCallCount;
var before = publisher.Count;
AwaitCondition(() => publisher.Count >= before + 4, PresenceBudget);
driver.SubscribeAlarmsCallCount.ShouldBe(settled, "a live, matching alarm handle must not be re-subscribed every tick");
}
/// <summary> /// <summary>
/// After a full reconnect cycle (Connected → Reconnecting → Connected), a single raised alarm still /// After a full reconnect cycle (Connected → Reconnecting → Connected), a single raised alarm still
/// yields EXACTLY ONE <see cref="DriverInstanceActor.AttributeAlarmPublished"/> — the /// yields EXACTLY ONE <see cref="DriverInstanceActor.AttributeAlarmPublished"/> — the
@@ -354,10 +436,16 @@ public sealed class DriverInstanceActorNativeAlarmTests : RuntimeActorTestBase
IReadOnlyList<string> sourceNodeIds, CancellationToken cancellationToken) IReadOnlyList<string> sourceNodeIds, CancellationToken cancellationToken)
{ {
LastAlarmRefs = sourceNodeIds; LastAlarmRefs = sourceNodeIds;
Interlocked.Increment(ref _subscribeAlarmsCallCount); var call = Interlocked.Increment(ref _subscribeAlarmsCallCount);
if (call <= AlarmSubscribeFailuresBeforeSuccess)
throw new InvalidOperationException($"alarm subscribe failure {call} (test-injected)");
return Task.FromResult<IAlarmSubscriptionHandle>(new StubAlarmHandle()); return Task.FromResult<IAlarmSubscriptionHandle>(new StubAlarmHandle());
} }
/// <summary>How many leading <see cref="SubscribeAlarmsAsync"/> calls throw — the "session not ready
/// after an outage" shape from #662.</summary>
public int AlarmSubscribeFailuresBeforeSuccess { get; init; }
/// <summary>Cancels an alarm subscription.</summary> /// <summary>Cancels an alarm subscription.</summary>
public Task UnsubscribeAlarmsAsync(IAlarmSubscriptionHandle handle, CancellationToken cancellationToken) => public Task UnsubscribeAlarmsAsync(IAlarmSubscriptionHandle handle, CancellationToken cancellationToken) =>
Task.CompletedTask; Task.CompletedTask;
@@ -416,6 +504,28 @@ public sealed class DriverInstanceActorNativeAlarmTests : RuntimeActorTestBase
} }
} }
/// <summary>Counts health-snapshot publishes — one per processed <c>health-poll</c> tick.</summary>
private sealed class CountingHealthPublisher : IDriverHealthPublisher
{
private int _count;
/// <summary>Number of snapshots published so far.</summary>
public int Count => Volatile.Read(ref _count);
/// <inheritdoc />
public void Publish(
string clusterId,
string driverInstanceId,
DriverHealth health,
int errorCount5Min,
DateTime? rediscoveryNeededUtc = null,
string? rediscoveryReason = null,
IReadOnlyList<HostConnectivityStatus>? hostStatuses = null,
SkippedTagReport? skippedTags = null,
IReadOnlyDictionary<string, string>? diagnostics = null)
=> Interlocked.Increment(ref _count);
}
/// <summary>Minimal <see cref="IAlarmSubscriptionHandle"/> for building <see cref="AlarmEventArgs"/>.</summary> /// <summary>Minimal <see cref="IAlarmSubscriptionHandle"/> for building <see cref="AlarmEventArgs"/>.</summary>
private sealed class StubAlarmHandle : IAlarmSubscriptionHandle private sealed class StubAlarmHandle : IAlarmSubscriptionHandle
{ {
@@ -0,0 +1,87 @@
using Akka.Actor;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy;
using ZB.MOM.WW.OtOpcUa.Commons.Types;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Runtime.OpcUa;
using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness;
namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers;
/// <summary>
/// Gitea #662 — a native-alarm feed must survive a deploy that does NOT respawn its driver, end to end
/// through the real <see cref="Runtime.Drivers.DriverHostActor"/> and its real
/// <see cref="Runtime.Drivers.DriverInstanceActor"/> child.
/// </summary>
/// <remarks>
/// Found live on the docker-dev rig: a deploy that only disabled ANOTHER driver re-pushed the desired set to
/// an unchanged OpcUaClient driver, whose data re-subscribe tore down the alarm event handler too. The
/// condition then froze while upstream kept emitting, and only a respawn brought it back. Every existing
/// end-to-end alarm test drives the driver directly (the host's driver is <c>Enabled=false</c>), so none of
/// them could see the instance actor drop the feed.
/// </remarks>
public sealed class NativeAlarmFeedSurvivesDeployTests : RuntimeActorTestBase
{
private static readonly RevisionHash RevA = RevisionHash.Parse(new string('a', 64));
private static readonly RevisionHash RevB = RevisionHash.Parse(new string('b', 64));
private static readonly NodeId TestNode = NodeId.Parse("node-1:4053");
private const string DriverId = "DI-1";
private const string AlarmXPath = "Plant/Modbus/dev1/alarmX";
private const string AlarmConfig = """{"alarm":{"alarmType":"AlarmCondition","severity":500}}""";
private static readonly (string Driver, string DriverName, string Device, string Tag, string DataType, bool Writable)[]
RawTags =
[
(DriverId, "Modbus", "dev1", "speed", "Float", false),
(DriverId, "Modbus", "dev1", "alarmX", "Float", false),
];
private static readonly Dictionary<string, string> TagConfigs = new() { ["alarmX"] = AlarmConfig };
[Fact]
public void An_upstream_alarm_still_reaches_its_condition_after_a_non_respawning_deploy()
{
var db = NewInMemoryDbFactory();
var factory = new RecordingSubscribableDriverFactory("Modbus");
var revA = V3ArtifactSeeder.SeedV3Deployment(
db, RevA, RawTags, Array.Empty<(string, string, string, string, string?)>(),
enabled: true, tagConfigByTag: TagConfigs);
var (host, publish, _) = V3ArtifactSeeder.SpawnHostAndApply(this, db, TestNode, revA, RevA, factory);
RecordingSubscribableDriver driver = null!;
AwaitAssert(() =>
{
factory.Created.Count.ShouldBe(1);
driver = factory.Created[0];
driver.SubscribeCount.ShouldBeGreaterThanOrEqualTo(1);
driver.LastSubscribedAlarmRefs.ShouldBe(new[] { AlarmXPath });
driver.AlarmSubscriberCount.ShouldBe(1);
}, duration: PresenceBudget);
var dataSubscribesBefore = driver.SubscribeCount;
// Revision B only ADDS another driver: DI-1's merged DriverConfig is byte-identical, so it is kept,
// and the bulk push re-sends its (unchanged) desired set — which re-subscribes its data.
var revB = V3ArtifactSeeder.SeedV3Deployment(
db, RevB, RawTags, Array.Empty<(string, string, string, string, string?)>(),
enabled: true, tagConfigByTag: TagConfigs, taglessDrivers: ["DI-2"]);
host.Tell(new DispatchDeployment(revB, RevB, CorrelationId.NewId()));
AwaitAssert(() =>
{
driver.SubscribeCount.ShouldBeGreaterThan(dataSubscribesBefore,
"precondition: the deploy must re-subscribe DI-1's data, or this test exercises nothing");
}, duration: PresenceBudget);
driver.InitializeCount.ShouldBe(1, "precondition: DI-1 must NOT have been respawned or re-initialised");
driver.ShutdownCount.ShouldBe(0);
factory.Created[0].ShouldBeSameAs(driver);
driver.AlarmSubscriberCount.ShouldBe(1, "the alarm event handler must survive the data re-subscribe");
driver.FireAlarm(AlarmXPath, AlarmTransitionKind.Raise);
var update = publish.FishForMessage<OpcUaPublishActor.AlarmStateUpdate>(
m => m.AlarmNodeId == AlarmXPath && m.State.Active, PresenceBudget);
update.State.Acknowledged.ShouldBeFalse();
}
}
@@ -343,9 +343,15 @@ internal sealed class RecordingSubscribableDriver : IDriver, ISubscribable, IAla
/// <summary>Raises <see cref="OnAlarmEvent"/>; present so the event is not merely written-to-never-read.</summary> /// <summary>Raises <see cref="OnAlarmEvent"/>; present so the event is not merely written-to-never-read.</summary>
/// <param name="conditionId">The condition id to raise.</param> /// <param name="conditionId">The condition id to raise.</param>
public void FireAlarm(string conditionId) => /// <param name="kind">The transition kind to report.</param>
public void FireAlarm(string conditionId, AlarmTransitionKind kind = AlarmTransitionKind.Unspecified) =>
OnAlarmEvent?.Invoke(this, new AlarmEventArgs( OnAlarmEvent?.Invoke(this, new AlarmEventArgs(
_alarmHandle, conditionId, conditionId, "AlarmCondition", "test", AlarmSeverity.High, DateTime.UtcNow)); _alarmHandle, conditionId, conditionId, "AlarmCondition", "test", AlarmSeverity.High, DateTime.UtcNow,
Kind: kind));
/// <summary>Live subscribers on <see cref="OnAlarmEvent"/> — an alarm subscription with no handler
/// attached delivers nothing (#662), so this is the other half of "the feed is up".</summary>
public int AlarmSubscriberCount => OnAlarmEvent?.GetInvocationList().Length ?? 0;
private sealed class RecordingAlarmHandle : IAlarmSubscriptionHandle private sealed class RecordingAlarmHandle : IAlarmSubscriptionHandle
{ {