fix(runtime): a changed alarm set re-subscribes instead of latching (review F1)
The alarm subscribe gated on 'a handle exists', which is correct for an idempotent re-push and wrong for a deploy that widens/narrows the set. Combined with the spawn-time desired set, a respawning deploy could latch the PREVIOUS revision's alarm refs for the life of the connection; sourceNodeIds-filtering drivers (OpcUaClient/AbCip/Focas) would never deliver the new alarm. Seen RED: LastSubscribedAlarmRefs stayed [alarmX] after a deploy widening to [alarmX, alarmY].
This commit is contained in:
@@ -211,6 +211,19 @@ Things to know before touching any of this:
|
||||
(`HandleSubscribeAsync` drops the prior subscription before establishing the new one); a real device
|
||||
driver, whose initialise opens a socket, rarely wins that race. Withholding the spawn-time send to avoid
|
||||
it would simply restore 03/S16, since the Restart and supervision routes have no bulk push coming.
|
||||
- ⚠️ **Convergence is not free on the ALARM side, and "a subscription already exists" is the wrong guard
|
||||
there.** The value path converges by construction — `StoreDesiredSubscriptions` is a full replace and
|
||||
`HandleSubscribeAsync` drops the prior subscription before establishing the new one — but
|
||||
`SubscribeDesiredAlarms` / `HandleSubscribeAlarmsAsync` used to gate purely on
|
||||
`_alarmSubscriptionHandle is not null`, so once a handle existed the alarm set **latched for the life of
|
||||
the connection**. Combined with the transient above, a deploy widening a driver's alarm refs `{X}` →
|
||||
`{X, Y}` could leave a fast-initialising driver subscribed to `{X}` forever, and the drivers that filter
|
||||
their feed on the subscribed `sourceNodeIds` (OPC UA Client, AB CIP, FOCAS) would simply never deliver
|
||||
the new alarm — silently. The guard is now "the desired refs differ from the ones the live handle was
|
||||
established with" (`_subscribedAlarmRefs`), and a changed set **releases the old subscription before
|
||||
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
|
||||
address-space rebuild and `ForwardNativeAlarm` drops any transition whose `ConditionId` no longer maps.
|
||||
- **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
|
||||
describes. It is kept — its own subject, the `InvalidActorNameException`, is real — but the load-bearing
|
||||
|
||||
@@ -271,6 +271,15 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
|
||||
/// idempotent across repeated <see cref="SetDesiredSubscriptions"/> pushes.</summary>
|
||||
private IAlarmSubscriptionHandle? _alarmSubscriptionHandle;
|
||||
|
||||
/// <summary>The alarm references the live <see cref="_alarmSubscriptionHandle"/> was actually established
|
||||
/// with, so a CHANGED desired set can be told apart from a repeat of the same one.
|
||||
/// <para>Without this the subscription latched: both <see cref="SubscribeDesiredAlarms"/> and
|
||||
/// <see cref="HandleSubscribeAlarmsAsync"/> gate on "a handle already exists", which is the right guard
|
||||
/// for an idempotent re-push and the WRONG one for a deploy that widens or narrows the alarm set —
|
||||
/// drivers that filter on <c>sourceNodeIds</c> (OPC UA Client, AB CIP, FOCAS) would then never deliver a
|
||||
/// newly-added alarm for the life of the connection. Empty whenever no handle is held.</para></summary>
|
||||
private IReadOnlyList<string> _subscribedAlarmRefs = Array.Empty<string>();
|
||||
|
||||
/// <summary>
|
||||
/// Gets or sets the timer scheduler for scheduling reconnection attempts.
|
||||
/// </summary>
|
||||
@@ -521,10 +530,14 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
|
||||
// Native-alarm analogue: un-gate the IAlarmSource feed when alarm tags are (now) present. The
|
||||
// common live path — a deploy delivers SetDesiredSubscriptions while the driver is already
|
||||
// Connected — flows through HERE, so the alarm subscribe must happen on this message, not only
|
||||
// on Connected entry. Unlike the value path above, there is deliberately no unsubscribe-on-empty:
|
||||
// a removed alarm tag's condition node is torn down on the address-space rebuild and
|
||||
// DriverHostActor.ForwardNativeAlarm drops any transition whose ConditionId no longer maps, so a
|
||||
// lingering session-less alarm subscription can never surface a removed alarm.
|
||||
// on Connected entry. A CHANGED set re-subscribes (release-then-subscribe inside
|
||||
// HandleSubscribeAlarmsAsync): unlike the value path, "a subscription already exists" is NOT
|
||||
// sufficient, because several drivers filter their feed on the subscribed sourceNodeIds, so a
|
||||
// widened set that never re-subscribed would silently never deliver the new alarm.
|
||||
// Still deliberately no unsubscribe-on-EMPTY: a removed alarm tag's condition node is torn down
|
||||
// on the address-space rebuild and DriverHostActor.ForwardNativeAlarm drops any transition whose
|
||||
// ConditionId no longer maps, so a lingering session-less alarm subscription can never surface a
|
||||
// removed alarm.
|
||||
SubscribeDesiredAlarms();
|
||||
});
|
||||
ReceiveAsync<SubscribeAlarms>(HandleSubscribeAlarmsAsync);
|
||||
@@ -1167,6 +1180,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
|
||||
// (GalaxyDriver.SubscribeAlarmsAsync clears the set before adding), so handles no longer accumulate
|
||||
// across reconnects. The gate (Count > 0) and the one-shot fan-out are unchanged.
|
||||
_alarmSubscriptionHandle = null;
|
||||
_subscribedAlarmRefs = Array.Empty<string>();
|
||||
}
|
||||
|
||||
/// <summary>Establish the native-alarm subscription that un-gates an <see cref="IAlarmSource"/> driver's
|
||||
@@ -1176,10 +1190,17 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
|
||||
/// desired, and no subscription is yet established.</summary>
|
||||
private void SubscribeDesiredAlarms()
|
||||
{
|
||||
if (_driver is IAlarmSource && _desiredAlarmRefs.Count > 0 && _alarmSubscriptionHandle is null)
|
||||
if (_driver is IAlarmSource && _desiredAlarmRefs.Count > 0 && AlarmSubscriptionIsStale())
|
||||
Self.Tell(new SubscribeAlarms());
|
||||
}
|
||||
|
||||
/// <summary>True when the desired alarm refs are not the ones the live handle was established with —
|
||||
/// either because there is no handle yet, or because a deploy changed the set. The second case is the
|
||||
/// one a bare "handle is null" check misses; see <see cref="_subscribedAlarmRefs"/>.</summary>
|
||||
private bool AlarmSubscriptionIsStale() =>
|
||||
_alarmSubscriptionHandle is null ||
|
||||
!_subscribedAlarmRefs.SequenceEqual(_desiredAlarmRefs, StringComparer.Ordinal);
|
||||
|
||||
/// <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
|
||||
/// (the desired set may have cleared, or another SubscribeAlarms may have already established a handle,
|
||||
@@ -1187,10 +1208,36 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
|
||||
/// stays gated until then.</summary>
|
||||
private async Task HandleSubscribeAlarmsAsync(SubscribeAlarms _)
|
||||
{
|
||||
if (_driver is not IAlarmSource src || _desiredAlarmRefs.Count == 0 || _alarmSubscriptionHandle is not null)
|
||||
if (_driver is not IAlarmSource src || _desiredAlarmRefs.Count == 0 || !AlarmSubscriptionIsStale())
|
||||
return;
|
||||
var refs = _desiredAlarmRefs;
|
||||
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10));
|
||||
|
||||
// A handle held against a DIFFERENT set must be released before re-subscribing, or a
|
||||
// sourceNodeIds-filtering driver accumulates two overlapping subscriptions. Best-effort and
|
||||
// deliberately non-fatal: the re-subscribe below is what actually matters, and a driver whose
|
||||
// unsubscribe throws must not be left on the stale set because of it. Cleared FIRST so a throw
|
||||
// cannot leave a handle that the staleness check would keep treating as live.
|
||||
if (_alarmSubscriptionHandle is { } stale)
|
||||
{
|
||||
_alarmSubscriptionHandle = null;
|
||||
_subscribedAlarmRefs = Array.Empty<string>();
|
||||
try
|
||||
{
|
||||
await _invoker.ExecuteAsync(
|
||||
DriverCapability.AlarmSubscribe,
|
||||
_driverInstanceId,
|
||||
async ct => await src.UnsubscribeAlarmsAsync(stale, ct),
|
||||
cts.Token);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_log.Warning(ex,
|
||||
"DriverInstance {Id}: releasing the previous native-alarm subscription ({Diag}) threw; re-subscribing anyway",
|
||||
_driverInstanceId, stale.DiagnosticId);
|
||||
}
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
_alarmSubscriptionHandle = await _invoker.ExecuteAsync(
|
||||
@@ -1198,6 +1245,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
|
||||
_driverInstanceId,
|
||||
async ct => await src.SubscribeAlarmsAsync(refs, ct),
|
||||
cts.Token);
|
||||
_subscribedAlarmRefs = refs;
|
||||
_log.Info("DriverInstance {Id}: native-alarm subscription established for {Count} alarm ref(s) ({Diag})",
|
||||
_driverInstanceId, refs.Count, _alarmSubscriptionHandle.DiagnosticId);
|
||||
}
|
||||
|
||||
@@ -64,6 +64,11 @@ public sealed class DriverHostActorRestartTests : RuntimeActorTestBase
|
||||
private const string DriverId = "DI-1";
|
||||
private const string SpeedPath = "Plant/Modbus/dev1/speed";
|
||||
private const string TempPath = "Plant/Modbus/dev1/temp";
|
||||
private const string AlarmXPath = "Plant/Modbus/dev1/alarmX";
|
||||
private const string AlarmYPath = "Plant/Modbus/dev1/alarmY";
|
||||
|
||||
/// <summary>TagConfig marking a raw tag as alarm-bearing (parsed by <c>TagConfigIntent</c>).</summary>
|
||||
private const string AlarmConfig = """{"alarm":{"alarmType":"AlarmCondition","severity":500}}""";
|
||||
|
||||
/// <summary>Seeds + applies a v3 deployment whose single Modbus driver is ENABLED, so a real child is
|
||||
/// spawned around a <see cref="RecordingSubscribableDriver"/>, and waits until that child has connected
|
||||
@@ -285,6 +290,69 @@ public sealed class DriverHostActorRestartTests : RuntimeActorTestBase
|
||||
duration: PresenceBudget);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// <b>Review finding F1 — the alarm subscription must FOLLOW a changed desired set, not latch.</b>
|
||||
/// The value side converges by construction (<c>HandleSubscribeAsync</c> drops the prior subscription
|
||||
/// before establishing the new one, and <c>StoreDesiredSubscriptions</c> is a full replace), but the
|
||||
/// alarm side is guarded by <c>_alarmSubscriptionHandle is not null</c> in BOTH
|
||||
/// <c>SubscribeDesiredAlarms</c> and <c>HandleSubscribeAlarmsAsync</c> — so once a handle exists it
|
||||
/// was never re-established for the life of the connection.
|
||||
/// <para>
|
||||
/// That became reachable when every spawn started carrying a desired set: a tag-set edit forces a
|
||||
/// respawn (raw tags fold into the merged <c>DriverConfig</c>, #516), and a fast-initialising
|
||||
/// driver connects and subscribes the transient spawn-time set — the PREVIOUS revision's alarm
|
||||
/// refs — before the bulk push lands. The bulk push then early-returns, latching the old set for
|
||||
/// the connection's life. Drivers that FILTER on <c>sourceNodeIds</c> (OPC UA Client, AB CIP,
|
||||
/// FOCAS) would silently never deliver the newly-added alarm.
|
||||
/// </para>
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void A_deploy_that_widens_the_alarm_set_re_subscribes_the_new_refs()
|
||||
{
|
||||
var db = NewInMemoryDbFactory();
|
||||
var (host, factory) = DeployOneDriver(
|
||||
db,
|
||||
new[] { (DriverId, "Modbus", "dev1", "alarmX", "Float", false) },
|
||||
new Dictionary<string, string> { ["alarmX"] = AlarmConfig });
|
||||
|
||||
AwaitAssert(
|
||||
() =>
|
||||
{
|
||||
factory.Created.Count.ShouldBe(1);
|
||||
factory.Created[0].LastSubscribedAlarmRefs.ShouldBe(new[] { AlarmXPath }, ignoreOrder: true);
|
||||
},
|
||||
duration: PresenceBudget);
|
||||
|
||||
// Revision B WIDENS the alarm set: X stays, Y is added.
|
||||
var revB = V3ArtifactSeeder.SeedV3Deployment(
|
||||
db, RevB,
|
||||
new[]
|
||||
{
|
||||
(DriverId, "Modbus", "dev1", "alarmX", "Float", false),
|
||||
(DriverId, "Modbus", "dev1", "alarmY", "Float", false),
|
||||
},
|
||||
Array.Empty<(string, string, string, string, string?)>(),
|
||||
enabled: true,
|
||||
tagConfigByTag: new Dictionary<string, string>
|
||||
{
|
||||
["alarmX"] = AlarmConfig,
|
||||
["alarmY"] = AlarmConfig,
|
||||
});
|
||||
host.Tell(new Commons.Messages.Deploy.DispatchDeployment(revB, RevB, CorrelationId.NewId()));
|
||||
|
||||
AwaitAssert(
|
||||
() =>
|
||||
{
|
||||
var current = factory.Created[^1];
|
||||
current.LastSubscribedAlarmRefs.ShouldBe(
|
||||
new[] { AlarmXPath, AlarmYPath },
|
||||
ignoreOrder: true,
|
||||
"the driver is still subscribed to the PREVIOUS revision's alarm refs — the alarm " +
|
||||
"subscription latched on the transient spawn-time set and never followed the new one");
|
||||
},
|
||||
duration: PresenceBudget);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// <b>The repo's first genuine Akka SUPERVISION-restart test (03/U6).</b> Not a stop, not a stub: the
|
||||
/// driver raises a malformed <c>OnDataChange</c>, the actor NREs on its own thread, and Akka's default
|
||||
|
||||
Reference in New Issue
Block a user