fix(opcua): carry native alarm condition state across a full rebuild (#641)

A full RebuildAddressSpace recreated every native Part 9 condition inactive +
acknowledged, and when the alarm-tag set was unchanged (e.g. one
UnsTagReference added to an alarm-bearing raw tag) nothing re-told the new node
its state: the host does not re-subscribe for an unchanged set, and Galaxy /
OpcUaClient have no snapshot to replay anyway. Active alarms read normal until
their next device transition.

OtOpcUaNodeManager now captures each NATIVE condition's live state (from the
node, so a client-applied ack is kept) before the rebuild clears, restores it
when the same id is re-materialised native in the same apply, and reports ONE
condition event per restored off-normal condition at the new
IOpcUaAddressSpaceSink.CompleteAddressSpaceApply (after notifier wiring).
Unclaimed captures are discarded there, so a later re-add is fresh. Quality is
carried silently for normal conditions too (it is only re-annotated on a
connectivity edge). Scripted conditions are untouched (#487 re-asserts them).
OpcUaPublishActor calls CompleteAddressSpaceApply at the end of every apply.

Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
This commit is contained in:
Joseph Doherty
2026-10-08 17:02:12 -04:00
parent a8b07ae12d
commit ddd7678269
9 changed files with 611 additions and 3 deletions
@@ -0,0 +1,125 @@
using Opc.Ua;
using Opc.Ua.Client;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
using ZB.MOM.WW.OtOpcUa.Configuration.Enums;
using static ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests.ServerEventRig;
namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests;
/// <summary>
/// Gitea #641 over the wire — a full address-space rebuild that leaves the native alarm set unchanged must not
/// leave an active condition reading inactive. A Server-object (<c>i=2253</c>) item, which outlives the rebuild
/// (#586), receives exactly ONE condition event carrying the restored state — and an acknowledgement a client
/// applied before the rebuild survives it.
/// </summary>
/// <remarks>
/// The rebuild is driven exactly as <c>OpcUaPublishActor</c> drives one: <c>RebuildAddressSpace</c>, the
/// re-materialise + notifier wiring, then <c>CompleteAddressSpaceApply</c>. The count is read after a later
/// sentinel transition arrives on the same item, so a duplicate would already be queued ahead of it.
/// The native ack router echoes nothing back, like MTConnect and FOCAS: the acknowledgement lives only on the node.
/// </remarks>
public sealed class NativeAlarmRebuildCarryOverWireTests
{
private const NodePermissions All =
NodePermissions.Browse | NodePermissions.Read | NodePermissions.Subscribe | NodePermissions.AlarmRead
| NodePermissions.AlarmAcknowledge | NodePermissions.MethodCall;
private static AlarmConditionSnapshot Fault(string message) => new(
Active: true, Acknowledged: false, Confirmed: true, Enabled: true,
Shelving: AlarmShelvingKind.Unshelved, Severity: 800, Message: message);
[Theory]
[InlineData(false)]
[InlineData(true)]
public async Task A_rebuild_with_an_unchanged_alarm_set_reports_the_restored_state_once(bool clientAcked)
{
var ct = TestContext.Current.CancellationToken;
var rig = await StartAsync($"Rebuild641{clientAcked}", encrypted: false, ct);
try
{
rig.Enforce(false, [Key()], ClusterGrant(KeyPrincipal, "acl-key", All));
rig.Nm.NativeAlarmAckRouter = _ => { }; // a driver whose ack is a no-op: nothing echoes back
rig.MaterialiseNative();
using var session = await rig.ConnectAsync(anonymous: false, ct);
var events = new RecordedEvents();
var subscription = await RecordedEvents.SubscribeAsync(session, events, ct);
var serverItem = RecordedEvents.AddEventItem(subscription, ObjectIds.Server);
await subscription.ApplyChangesAsync(ct);
rig.Project(native: true, Fault("fault"));
await RecordedEvents.WaitUntilAsync(() => events.Has(serverItem, "fault"));
var expectedBefore = 1;
if (clientAcked)
{
await session.CallAsync(
ConditionNodeId(session, native: true), MethodIds.AcknowledgeableConditionType_Acknowledge, ct,
events.LatestEventId(serverItem, "fault"), new LocalizedText("ack"));
await RecordedEvents.WaitUntilAsync(() => events.ConditionEvents(serverItem, "fault").Count == 2);
expectedBefore = 2;
}
// The deploy: a rebuild whose composition carries the same alarm (e.g. one UnsTagReference added).
rig.Sink.RebuildAddressSpace();
rig.MaterialiseNative();
rig.Sink.CompleteAddressSpaceApply();
rig.Project(native: true, Fault("sentinel") with { Acknowledged = clientAcked });
await RecordedEvents.WaitUntilAsync(() => events.Has(serverItem, "sentinel"));
var faultEvents = events.ConditionEvents(serverItem, "fault");
faultEvents.Count.ShouldBe(expectedBefore + 1, "the pre-rebuild event(s) plus exactly one restored event");
var restored = faultEvents[^1];
restored[RecordedEvents.ActiveField].Value.ShouldBe(true);
restored[RecordedEvents.AckedField].Value.ShouldBe(clientAcked);
restored[RecordedEvents.RetainField].Value.ShouldBe(true);
restored[RecordedEvents.SeverityField].Value.ShouldBe((ushort)800);
// And a ConditionRefresh-style read of the live node agrees.
var active = await session.ReadValueAsync(
new NodeId($"{RawAlarmPath}/ActiveState/Id", ConditionNodeId(session, native: true).NamespaceIndex), ct);
active.Value.ShouldBe(true);
await subscription.DeleteAsync(true, ct);
}
finally { await rig.DisposeAsync(); }
}
/// <summary>Falsification control: a NORMAL native condition produces no event across the rebuild.</summary>
[Fact]
public async Task A_rebuild_reports_nothing_for_a_normal_condition()
{
var ct = TestContext.Current.CancellationToken;
var rig = await StartAsync("Rebuild641Normal", encrypted: false, ct);
try
{
rig.Enforce(false, [Key()], ClusterGrant(KeyPrincipal, "acl-key", All));
rig.MaterialiseNative();
using var session = await rig.ConnectAsync(anonymous: false, ct);
var events = new RecordedEvents();
var subscription = await RecordedEvents.SubscribeAsync(session, events, ct);
var serverItem = RecordedEvents.AddEventItem(subscription, ObjectIds.Server);
await subscription.ApplyChangesAsync(ct);
rig.Project(native: true, Fault("cleared") with { Active = false, Acknowledged = true });
await RecordedEvents.WaitUntilAsync(() => events.Has(serverItem, "cleared"));
rig.Sink.RebuildAddressSpace();
rig.MaterialiseNative();
rig.Sink.CompleteAddressSpaceApply();
rig.Project(native: true, Fault("sentinel"));
await RecordedEvents.WaitUntilAsync(() => events.Has(serverItem, "sentinel"));
events.ConditionEvents(serverItem, "cleared").Count.ShouldBe(1);
events.ConditionEvents(serverItem, "temp_hi").ShouldBeEmpty("the fresh node is not reported");
await subscription.DeleteAsync(true, ct);
}
finally { await rig.DisposeAsync(); }
}
}
@@ -0,0 +1,255 @@
using Opc.Ua;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests;
/// <summary>
/// Gitea #641 — a full <see cref="OtOpcUaNodeManager.RebuildAddressSpace"/> recreates every native condition
/// inactive, and when the alarm-tag set is unchanged nothing tells the new node its state again (the driver
/// does not re-subscribe, and Galaxy / OpcUaClient have nothing to replay anyway). The node manager therefore
/// carries each NATIVE condition's live state across the rebuild: captured before the clear, restored when
/// the same id is re-materialised native in the same apply, and reported as ONE condition event when the
/// apply completes (<see cref="OtOpcUaNodeManager.CompleteAddressSpaceApply"/>).
/// </summary>
/// <remarks>
/// "One event" is observed here through the condition's <c>EventId</c>, which every report stamps fresh.
/// The exact count over the wire is pinned by <c>NativeAlarmRebuildCarryOverWireTests</c>.
/// </remarks>
public sealed class NodeManagerNativeAlarmRebuildCarryOverTests : IDisposable
{
private const string Device = "Plant/MTConnect/dev1";
private const string Alarm = "Plant/MTConnect/dev1/Xtravel";
private static CancellationToken Ct => TestContext.Current.CancellationToken;
private readonly string _pkiRoot = Path.Combine(Path.GetTempPath(), $"otopcua-641-{Guid.NewGuid():N}");
private static AlarmConditionSnapshot Fault(bool acked) => new(
Active: true, Acknowledged: acked, Confirmed: true, Enabled: true,
Shelving: AlarmShelvingKind.Unshelved, Severity: 800, Message: "Xtravel FAULT 1001",
StatusCode: OpcUaStatus.Good);
[Trait("Category", "Unit")]
[Fact]
public async Task An_active_unacked_native_condition_survives_a_rebuild_and_is_reported_once()
{
var (host, nm) = await BootAsync();
var raisedAt = new DateTime(2026, 10, 8, 9, 0, 0, DateTimeKind.Utc);
Materialise(nm, native: true);
nm.WriteAlarmCondition(Alarm, Fault(acked: false), raisedAt, AddressSpaceRealm.Raw);
nm.RebuildAddressSpace();
Materialise(nm, native: true);
var restored = Condition(nm);
var idBeforeComplete = (byte[]?)restored.EventId.Value?.Clone();
// State is back as soon as the node exists — a Read or a ConditionRefresh in between already sees it.
restored.ActiveState.Id.Value.ShouldBeTrue();
restored.AckedState.Id.Value.ShouldBeFalse();
restored.Retain.Value.ShouldBeTrue();
restored.Severity.Value.ShouldBe((ushort)800);
restored.Message.Value.Text.ShouldBe("Xtravel FAULT 1001");
restored.Quality.Value.Code.ShouldBe(StatusCodes.Good);
nm.CompleteAddressSpaceApply();
// The report happens at the END of the apply (after the notifier wiring), stamped with the instant the
// state came about, not the deploy time.
restored.EventId.Value.ShouldNotBe(idBeforeComplete);
restored.Time.Value.ShouldBe(raisedAt);
// ...and only once: a second completion with nothing pending reports nothing.
var idAfterComplete = (byte[]?)restored.EventId.Value?.Clone();
nm.CompleteAddressSpaceApply();
restored.EventId.Value.ShouldBe(idAfterComplete);
await host.DisposeAsync();
}
/// <summary>
/// The acknowledgement a CLIENT applied lives only on the node (the SDK applied it; no driver or projector
/// saw it) — the reason the carry-over reads the node rather than any host-side cache.
/// </summary>
[Trait("Category", "Unit")]
[Fact]
public async Task A_client_acknowledgement_survives_the_rebuild()
{
var (host, nm) = await BootAsync();
Materialise(nm, native: true);
nm.WriteAlarmCondition(Alarm, Fault(acked: false), DateTime.UtcNow, AddressSpaceRealm.Raw);
// What the SDK's Acknowledge handler does to the node; nothing else hears of it.
Condition(nm).SetAcknowledgedState(nm.SystemContext, true);
nm.RebuildAddressSpace();
Materialise(nm, native: true);
nm.CompleteAddressSpaceApply();
var restored = Condition(nm);
restored.ActiveState.Id.Value.ShouldBeTrue();
restored.AckedState.Id.Value.ShouldBeTrue();
restored.Retain.Value.ShouldBeTrue();
await host.DisposeAsync();
}
/// <summary>Falsification control: a condition in the no-event position comes back exactly as a fresh
/// materialise builds it, and reports nothing.</summary>
[Trait("Category", "Unit")]
[Fact]
public async Task A_normal_native_condition_reports_nothing_and_keeps_the_fresh_state()
{
var (host, nm) = await BootAsync();
Materialise(nm, native: true);
nm.WriteAlarmCondition(Alarm, Fault(acked: false) with { Active = false, Acknowledged = true, Message = "cleared" },
DateTime.UtcNow, AddressSpaceRealm.Raw);
nm.RebuildAddressSpace();
Materialise(nm, native: true);
var fresh = Condition(nm);
var idBefore = (byte[]?)fresh.EventId.Value?.Clone();
nm.CompleteAddressSpaceApply();
fresh.EventId.Value.ShouldBe(idBefore);
fresh.ActiveState.Id.Value.ShouldBeFalse();
fresh.AckedState.Id.Value.ShouldBeTrue();
fresh.Retain.Value.ShouldBeFalse();
fresh.Message.Value.Text.ShouldBe("Xtravel", "a normal condition keeps the materialise's message");
fresh.Severity.Value.ShouldBe((ushort)700, "and its authored severity");
await host.DisposeAsync();
}
/// <summary>
/// The source-data quality is carried for a NORMAL condition too, silently: quality is annotated only on a
/// driver's connectivity edge, so without the carry a rebuild left every normal native condition reading
/// <c>BadWaitingForInitialData</c> until the driver next flapped.
/// </summary>
[Trait("Category", "Unit")]
[Fact]
public async Task A_normal_condition_keeps_its_quality_without_an_event()
{
var (host, nm) = await BootAsync();
Materialise(nm, native: true);
nm.WriteAlarmQuality(Alarm, OpcUaStatus.Good, DateTime.UtcNow, AddressSpaceRealm.Raw);
nm.RebuildAddressSpace();
Materialise(nm, native: true);
var fresh = Condition(nm);
var idBefore = (byte[]?)fresh.EventId.Value?.Clone();
nm.CompleteAddressSpaceApply();
fresh.Quality.Value.Code.ShouldBe(StatusCodes.Good);
fresh.EventId.Value.ShouldBe(idBefore);
await host.DisposeAsync();
}
/// <summary>A condition the rebuild did NOT re-create has its captured state discarded when the apply
/// completes, so a later deploy that adds the same RawPath back gets a fresh node, not a resurrection.</summary>
[Trait("Category", "Unit")]
[Fact]
public async Task A_condition_removed_by_the_rebuild_is_not_resurrected_by_a_later_add()
{
var (host, nm) = await BootAsync();
Materialise(nm, native: true);
nm.WriteAlarmCondition(Alarm, Fault(acked: false), DateTime.UtcNow, AddressSpaceRealm.Raw);
nm.RebuildAddressSpace();
nm.CompleteAddressSpaceApply(); // the alarm was removed: nothing re-materialised it
Materialise(nm, native: true); // a later, surgical re-add of the same RawPath
nm.CompleteAddressSpaceApply();
var readded = Condition(nm);
readded.ActiveState.Id.Value.ShouldBeFalse();
readded.AckedState.Id.Value.ShouldBeTrue();
await host.DisposeAsync();
}
/// <summary>A native condition re-materialised as SCRIPTED under the same id is a different alarm: nothing
/// is restored onto it (and the scripted engine's own #487 re-assert owns scripted state).</summary>
[Trait("Category", "Unit")]
[Fact]
public async Task A_kind_swap_on_the_same_id_is_not_restored()
{
var (host, nm) = await BootAsync();
Materialise(nm, native: true);
nm.WriteAlarmCondition(Alarm, Fault(acked: false), DateTime.UtcNow, AddressSpaceRealm.Raw);
nm.RebuildAddressSpace();
Materialise(nm, native: false);
nm.CompleteAddressSpaceApply();
Condition(nm).ActiveState.Id.Value.ShouldBeFalse();
await host.DisposeAsync();
}
/// <summary>Scripted conditions are not captured — <c>ScriptedAlarmHostActor.ReassertConditionNodes</c>
/// (#487) already re-projects them after a reload, and a second source would double-report.</summary>
[Trait("Category", "Unit")]
[Fact]
public async Task A_scripted_condition_is_not_carried()
{
var (host, nm) = await BootAsync();
Materialise(nm, native: false);
nm.WriteAlarmCondition(Alarm, Fault(acked: false), DateTime.UtcNow, AddressSpaceRealm.Raw);
nm.RebuildAddressSpace();
Materialise(nm, native: false);
nm.CompleteAddressSpaceApply();
Condition(nm).ActiveState.Id.Value.ShouldBeFalse();
await host.DisposeAsync();
}
private static void Materialise(OtOpcUaNodeManager nm, bool native)
{
nm.EnsureFolder(Device, parentNodeId: null, displayName: "dev1", realm: AddressSpaceRealm.Raw);
nm.MaterialiseAlarmCondition(Alarm, Device, "Xtravel", "OffNormalAlarm", 700, AddressSpaceRealm.Raw, isNative: native);
}
private static Opc.Ua.AlarmConditionState Condition(OtOpcUaNodeManager nm) =>
nm.TryGetAlarmCondition(Alarm, AddressSpaceRealm.Raw).ShouldNotBeNull();
private async Task<(OpcUaApplicationHost Host, OtOpcUaNodeManager Nm)> BootAsync()
{
var host = new OpcUaApplicationHost(
new OpcUaApplicationHostOptions
{
ApplicationName = "OtOpcUa.Gitea641Test",
ApplicationUri = $"urn:OtOpcUa.Gitea641Test:{Guid.NewGuid():N}",
OpcUaPort = AllocateFreePort(),
PublicHostname = "localhost",
PkiStoreRoot = _pkiRoot,
},
Microsoft.Extensions.Logging.Abstractions.NullLogger<OpcUaApplicationHost>.Instance);
var server = new OtOpcUaSdkServer();
await host.StartAsync(server, Ct);
return (host, server.NodeManager!);
}
private static int AllocateFreePort()
{
using var listener = new System.Net.Sockets.TcpListener(System.Net.IPAddress.Loopback, 0);
listener.Start();
var port = ((System.Net.IPEndPoint)listener.LocalEndpoint).Port;
listener.Stop();
return port;
}
/// <summary>Cleans up the PKI root directory.</summary>
public void Dispose()
{
if (Directory.Exists(_pkiRoot))
{
try { Directory.Delete(_pkiRoot, recursive: true); }
catch { /* best-effort cleanup */ }
}
}
}
@@ -61,6 +61,41 @@ public sealed class OpcUaPublishActorRebuildTests : RuntimeActorTestBase
sink.RebuildCalls.ShouldBe(0);
}
/// <summary>
/// Gitea #641 — every apply ends with <c>CompleteAddressSpaceApply</c>, AFTER the last materialise / announce
/// call: that is where the node manager reports the conditions a rebuild carried over (their notifier
/// folders are wired by then) and drops any captured state no materialise claimed.
/// </summary>
[Fact]
public void An_apply_completes_after_every_materialise_pass()
{
var db = NewInMemoryDbFactory();
var sink = new RecordingSink();
var applier = new AddressSpaceApplier(sink, NullLogger<AddressSpaceApplier>.Instance);
SeedDeployment(db, equipmentIds: new[] { "eq-1", "eq-2" }, driverIds: new[] { "drv-1" });
var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(sink: sink, dbFactory: db, applier: applier));
actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId()));
AwaitAssert(() => sink.CompletedAfterCallCount.Count.ShouldBe(1), duration: PresenceBudget);
sink.CompletedAfterCallCount.Single().ShouldBe(sink.Calls.Count, "nothing may follow the completion");
sink.Calls.ShouldContain("NA:line-1");
}
/// <summary>Gitea #641 — the raw-sink fallback rebuild (no applier) completes too, so the state it captured
/// is discarded rather than left for a later add to resurrect.</summary>
[Fact]
public void The_raw_sink_fallback_rebuild_completes_the_apply()
{
var sink = new RecordingSink();
var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(sink: sink));
actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId()));
AwaitAssert(() => sink.CompletedAfterCallCount.Count.ShouldBe(1), duration: PresenceBudget);
sink.RebuildCalls.ShouldBe(1);
}
/// <summary>Tests that rebuild with no artifact is idempotent no-op.</summary>
[Fact]
public void Rebuild_with_no_artifact_is_idempotent_no_op()
@@ -1053,6 +1088,12 @@ public sealed class OpcUaPublishActorRebuildTests : RuntimeActorTestBase
}
/// <summary>Records a rebuild address space call.</summary>
public void RebuildAddressSpace() => Interlocked.Increment(ref RebuildCalls);
/// <summary>Gets, per <see cref="CompleteAddressSpaceApply"/> call, how many <see cref="Calls"/> had been
/// recorded at that moment — so a test can prove the completion came after every pass (Gitea #641).
/// Kept out of <see cref="Calls"/> so existing call-count assertions are unaffected.</summary>
public ConcurrentQueue<int> CompletedAfterCallCount { get; } = new();
/// <summary>Records the end of an apply pass (Gitea #641).</summary>
public void CompleteAddressSpaceApply() => CompletedAfterCallCount.Enqueue(Calls.Count);
/// <summary>Records a NodeAdded model-change announcement.</summary>
/// <param name="affectedNodeId">The node under which discovered nodes were added.</param>
public void RaiseNodesAddedModelChange(string affectedNodeId, AddressSpaceRealm realm = AddressSpaceRealm.Uns) => Calls.Enqueue($"NA:{affectedNodeId}");