using System.Collections.Concurrent;
using System.Text.Json;
using Akka.Actor;
using Microsoft.Extensions.Logging.Abstractions;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
using ZB.MOM.WW.OtOpcUa.Commons.Types;
using ZB.MOM.WW.OtOpcUa.Core.Authorization;
using ZB.MOM.WW.OtOpcUa.OpcUaServer;
using ZB.MOM.WW.OtOpcUa.Runtime.OpcUa;
using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness;
namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.OpcUa;
///
/// Lane B review fix round 1. C1: the plan runs off the pinned thread (R02:P5) while writes are non-creating
/// (R01:L1), so a value, status or alarm write for a node the IN-FLIGHT plan adds used to be skipped — and a
/// change-driven driver never republishes it. The actor now remembers the latest write per node while a plan
/// is in flight and replays it after the apply. Every test holds the plan open with a gate, so "during the
/// plan" is deterministic, and runs against a sink that — like the real node manager — drops writes for nodes
/// that do not exist.
/// Also here: I2 (an authority parse throw never fails the apply), the bytes-less request queued behind an
/// in-flight plan, and a plan piped back to a restarted actor.
///
public sealed class OpcUaPublishActorDeferredWriteTests : RuntimeActorTestBase
{
private const string NewTag = "Plant/Modbus/dev1/T1";
private static readonly TimeSpan PlanBudget = TimeSpan.FromSeconds(30);
[Fact]
public void A_value_for_a_node_the_in_flight_plan_adds_lands_on_that_node_after_the_apply()
{
var sink = new ExistenceSink();
using var gate = new ManualResetEventSlim(false);
var actor = Spawn(sink, gate);
actor.Tell(Rebuild(RawTagArtifact(3)));
var at = DateTime.UtcNow;
actor.Tell(new OpcUaPublishActor.AttributeValueBatchUpdate(
[new ValueWrite(NewTag, 7, OpcUaStatus.Good, at, AddressSpaceRealm.Raw)]));
AwaitAssert(() => sink.SkippedWrites.ShouldContain(NewTag), PlanBudget, cancellationToken: TestContext.Current.CancellationToken); // non-vacuity: it really missed
gate.Set();
ExpectMsg(PlanBudget, cancellationToken: TestContext.Current.CancellationToken).Degraded.ShouldBeFalse();
sink.Exists(NewTag).ShouldBeTrue("the plan materialised the node");
sink.ValueOf(NewTag).ShouldBe((7, OpcUaStatus.Good, at), "the value sent while the plan was in flight must end up on the node");
}
[Fact]
public void A_status_after_a_value_composes_onto_the_new_node_in_order()
{
var sink = new ExistenceSink();
using var gate = new ManualResetEventSlim(false);
var actor = Spawn(sink, gate);
actor.Tell(Rebuild(RawTagArtifact(3)));
var t1 = DateTime.UtcNow;
var t2 = t1.AddSeconds(1);
actor.Tell(new OpcUaPublishActor.AttributeValueBatchUpdate([new ValueWrite(NewTag, 7, OpcUaStatus.Good, t1, AddressSpaceRealm.Raw)]));
actor.Tell(new OpcUaPublishActor.AttributeStatusBatchUpdate([new StatusWrite(NewTag, OpcUaStatus.BadNoCommunication, t2, AddressSpaceRealm.Raw)]));
gate.Set();
ExpectMsg(PlanBudget, cancellationToken: TestContext.Current.CancellationToken);
sink.ValueOf(NewTag).ShouldBe((7, OpcUaStatus.BadNoCommunication, t2), "last known value under the later status");
}
[Fact]
public void Existing_nodes_keep_updating_while_the_plan_computes()
{
var sink = new ExistenceSink();
using var gate = new ManualResetEventSlim(false);
var actor = Spawn(sink, gate);
actor.Tell(Rebuild(RawTagArtifact(1)));
gate.Set();
ExpectMsg(PlanBudget, cancellationToken: TestContext.Current.CancellationToken);
var existing = "Plant/Modbus/dev1/T0";
sink.Exists(existing).ShouldBeTrue();
gate.Reset();
actor.Tell(Rebuild(RawTagArtifact(3)));
var at = DateTime.UtcNow;
actor.Tell(new OpcUaPublishActor.AttributeValueBatchUpdate([new ValueWrite(existing, 5, OpcUaStatus.Good, at, AddressSpaceRealm.Raw)]));
// Applied immediately, while the plan is still held open.
AwaitAssert(() => sink.ValueOf(existing).ShouldBe((5, OpcUaStatus.Good, at)), PlanBudget, cancellationToken: TestContext.Current.CancellationToken);
sink.Exists(NewTag).ShouldBeFalse("the plan is still in flight");
gate.Set();
ExpectMsg(PlanBudget, cancellationToken: TestContext.Current.CancellationToken);
sink.ValueOf(existing).ShouldBe((5, OpcUaStatus.Good, at));
}
[Fact]
public void An_alarm_state_sent_during_the_plan_lands_on_the_condition_it_materialises()
{
var sink = new ExistenceSink();
using var gate = new ManualResetEventSlim(false);
var actor = Spawn(sink, gate);
actor.Tell(Rebuild(AlarmArtifact()));
var state = new AlarmConditionSnapshot(Active: true, Acknowledged: false, Confirmed: true, Enabled: true, AlarmShelvingKind.Unshelved, 500, "high");
actor.Tell(new OpcUaPublishActor.AlarmStateUpdate("al-1", state, DateTime.UtcNow, AddressSpaceRealm.Uns));
actor.Tell(new OpcUaPublishActor.AlarmQualityUpdate("al-1", OpcUaStatus.BadNoCommunication, DateTime.UtcNow, AddressSpaceRealm.Uns));
AwaitAssert(() => sink.SkippedAlarms.ShouldContain("al-1"), PlanBudget, cancellationToken: TestContext.Current.CancellationToken);
gate.Set();
ExpectMsg(PlanBudget, cancellationToken: TestContext.Current.CancellationToken).Degraded.ShouldBeFalse();
sink.ConditionExists("al-1").ShouldBeTrue("the plan materialised the condition");
sink.AlarmStateOf("al-1").ShouldBeSameAs(state);
sink.AlarmQualityOf("al-1").ShouldBe(OpcUaStatus.BadNoCommunication);
}
[Fact]
public void The_deferred_buffer_is_bounded_under_a_flood_and_keeps_the_earliest_keys()
{
var sink = new ExistenceSink();
using var gate = new ManualResetEventSlim(false);
var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(
sink, new AddressSpaceApplier(sink, NullLogger.Instance),
new OpcUaPublishActor.TestHooks { BeforePlan = () => gate.Wait() }, deferredWriteCapacity: 2));
actor.Tell(Rebuild(RawTagArtifact(5)));
var at = DateTime.UtcNow;
actor.Tell(new OpcUaPublishActor.AttributeValueBatchUpdate(Enumerable.Range(0, 5)
.Select(i => new ValueWrite($"Plant/Modbus/dev1/T{i}", i, OpcUaStatus.Good, at, AddressSpaceRealm.Raw)).ToList()));
// An existing key still updates when the buffer is full.
actor.Tell(new OpcUaPublishActor.AttributeValueBatchUpdate([new ValueWrite("Plant/Modbus/dev1/T0", 100, OpcUaStatus.Good, at, AddressSpaceRealm.Raw)]));
AwaitAssert(() => sink.SkippedWrites.Count.ShouldBe(6), PlanBudget, cancellationToken: TestContext.Current.CancellationToken);
gate.Set();
ExpectMsg(PlanBudget, cancellationToken: TestContext.Current.CancellationToken);
sink.ValueOf("Plant/Modbus/dev1/T0")!.Value.Value.ShouldBe(100);
sink.ValueOf("Plant/Modbus/dev1/T1")!.Value.Value.ShouldBe(1);
for (var i = 2; i < 5; i++)
sink.ValueOf($"Plant/Modbus/dev1/T{i}").ShouldBeNull("beyond the bound a new key is not held");
sink.ReplayedItems.ShouldBe(2);
}
[Fact]
public void Writes_after_the_pipeline_drains_are_not_replayed_again()
{
var sink = new ExistenceSink();
using var gate = new ManualResetEventSlim(false);
var actor = Spawn(sink, gate);
actor.Tell(Rebuild(RawTagArtifact(3)));
actor.Tell(new OpcUaPublishActor.AttributeValueBatchUpdate([new ValueWrite(NewTag, 7, OpcUaStatus.Good, DateTime.UtcNow, AddressSpaceRealm.Raw)]));
gate.Set();
ExpectMsg(PlanBudget, cancellationToken: TestContext.Current.CancellationToken);
var replayed = sink.ReplayedItems;
replayed.ShouldBe(1);
// A later plan replays only what arrived during IT.
actor.Tell(Rebuild(RawTagArtifact(3)));
ExpectMsg(PlanBudget, cancellationToken: TestContext.Current.CancellationToken);
sink.ReplayedItems.ShouldBe(replayed);
}
[Fact]
public void A_throwing_authority_parse_does_not_fail_the_apply()
{
var sink = new ExistenceSink();
var authority = new RecordingAuthoritySink();
var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(
sink, new AddressSpaceApplier(sink, NullLogger.Instance),
new OpcUaPublishActor.TestHooks { ParseAuthority = (_, _) => throw new InvalidOperationException("bad grants") },
dataPlaneAuthority: authority));
actor.Tell(Rebuild(RawTagArtifact(2)));
var completed = ExpectMsg(PlanBudget, cancellationToken: TestContext.Current.CancellationToken);
completed.Degraded.ShouldBeFalse(completed.DegradedReason);
sink.Exists("Plant/Modbus/dev1/T1").ShouldBeTrue("the address space is applied");
authority.Applies.ShouldBe(0, "nothing is installed from an authority that could not be read");
// Positive control: the same actor installs an authority that CAN be read.
var healthy = Sys.ActorOf(OpcUaPublishActor.PropsForTests(
new ExistenceSink(), new AddressSpaceApplier(new ExistenceSink(), NullLogger.Instance),
hooks: null, dataPlaneAuthority: authority));
healthy.Tell(Rebuild(RawTagArtifact(2)));
ExpectMsg(PlanBudget, cancellationToken: TestContext.Current.CancellationToken).Degraded.ShouldBeFalse();
authority.Applies.ShouldBe(1);
}
[Fact]
public void A_bytes_less_rebuild_arriving_mid_plan_answers_after_the_plan_ahead_of_it()
{
var sink = new ExistenceSink();
using var gate = new ManualResetEventSlim(false);
var actor = Spawn(sink, gate);
var first = Rebuild(RawTagArtifact(2));
var second = new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId(), DeploymentId.NewId(), Artifact: null, ReportOutcome: true);
actor.Tell(first);
actor.Tell(second);
ExpectNoMsg(TimeSpan.FromMilliseconds(300), TestContext.Current.CancellationToken); // the bytes-less one must not overtake the held plan
gate.Set();
var a = ExpectMsg(PlanBudget, cancellationToken: TestContext.Current.CancellationToken);
var b = ExpectMsg(PlanBudget, cancellationToken: TestContext.Current.CancellationToken);
a.Correlation.ShouldBe(first.Correlation);
a.Degraded.ShouldBeFalse();
b.Correlation.ShouldBe(second.Correlation);
b.Degraded.ShouldBeTrue();
}
[Fact]
public void A_plan_piped_back_to_a_restarted_actor_is_not_applied()
{
var sink = new ExistenceSink();
using var gate = new ManualResetEventSlim(false);
var applier = new AddressSpaceApplier(sink, NullLogger.Instance);
var childProps = OpcUaPublishActor.PropsForTests(sink, applier, new OpcUaPublishActor.TestHooks { BeforePlan = () => gate.Wait() });
var parent = Sys.ActorOf(Props.Create(() => new RestartingParent(childProps)));
parent.Tell(RestartingParent.GetChild.Instance);
var child = ExpectMsg(cancellationToken: TestContext.Current.CancellationToken);
var stale = Rebuild(RawTagArtifact(3));
child.Tell(stale);
child.Tell(Kill.Instance); // the parent's strategy RESTARTS on any failure, ActorKilledException included
ExpectNoMsg(TimeSpan.FromMilliseconds(300), TestContext.Current.CancellationToken);
gate.Set();
var answer = ExpectMsg(PlanBudget, cancellationToken: TestContext.Current.CancellationToken);
answer.Correlation.ShouldBe(stale.Correlation);
answer.Degraded.ShouldBeTrue("the requester is still answered");
sink.Exists(NewTag).ShouldBeFalse("a plan from the previous incarnation must not be applied");
// Positive control: the restarted instance applies a fresh rebuild.
child.Tell(Rebuild(RawTagArtifact(3)));
ExpectMsg(PlanBudget, cancellationToken: TestContext.Current.CancellationToken).Degraded.ShouldBeFalse();
sink.Exists(NewTag).ShouldBeTrue();
}
// ───────────────────── fix round 2: N1 / N2 / M1 / M4 / M5 ─────────────────────
/// N1 — a full state snapshot carries quality, so it supersedes an EARLIER quality-only write for the
/// same condition (the engine sends a transition with the recovered worst quality and no QualityChanged).
[Fact]
public void A_state_after_a_quality_only_write_supersedes_it_on_replay()
{
var sink = new ExistenceSink();
using var gate = new ManualResetEventSlim(false);
var actor = Spawn(sink, gate);
actor.Tell(Rebuild(AlarmArtifact()));
actor.Tell(new OpcUaPublishActor.AlarmQualityUpdate("al-1", OpcUaStatus.BadNoCommunication, DateTime.UtcNow, AddressSpaceRealm.Uns));
actor.Tell(new OpcUaPublishActor.AlarmStateUpdate("al-1", ActiveUnacked(OpcUaStatus.Good), DateTime.UtcNow, AddressSpaceRealm.Uns));
AwaitAssert(() => sink.SkippedAlarms.Count.ShouldBe(2), PlanBudget, cancellationToken: TestContext.Current.CancellationToken);
gate.Set();
ExpectMsg(PlanBudget, cancellationToken: TestContext.Current.CancellationToken);
sink.AlarmQualityOf("al-1").ShouldBe(OpcUaStatus.Good, "the later snapshot's quality is what live would leave");
sink.AlarmStateOf("al-1")!.Active.ShouldBeTrue();
}
/// N1 positive control — the reverse order: a quality-only write AFTER the state still lands last.
[Fact]
public void A_quality_only_write_after_a_state_still_lands_after_it_on_replay()
{
var sink = new ExistenceSink();
using var gate = new ManualResetEventSlim(false);
var actor = Spawn(sink, gate);
actor.Tell(Rebuild(AlarmArtifact()));
actor.Tell(new OpcUaPublishActor.AlarmStateUpdate("al-1", ActiveUnacked(OpcUaStatus.Good), DateTime.UtcNow, AddressSpaceRealm.Uns));
actor.Tell(new OpcUaPublishActor.AlarmQualityUpdate("al-1", OpcUaStatus.BadNoCommunication, DateTime.UtcNow, AddressSpaceRealm.Uns));
AwaitAssert(() => sink.SkippedAlarms.Count.ShouldBe(2), PlanBudget, cancellationToken: TestContext.Current.CancellationToken);
gate.Set();
ExpectMsg(PlanBudget, cancellationToken: TestContext.Current.CancellationToken);
sink.AlarmQualityOf("al-1").ShouldBe(OpcUaStatus.BadNoCommunication);
}
///
/// N2 — a condition the plan KEEPS took the live write already; replaying it would roll back an operator's
/// Acknowledge applied in the window (the SDK mutates the node directly) and fire spurious events. The surgical
/// apply shape; the rebuild shapes are the two tests below (scripted: replayed; native: carried over).
///
[Fact]
public void An_out_of_band_ack_on_a_kept_condition_is_not_reverted_by_the_replay()
{
var sink = new ExistenceSink();
using var gate = new ManualResetEventSlim(true);
var actor = Spawn(sink, gate);
actor.Tell(Rebuild(AlarmArtifact()));
ExpectMsg(PlanBudget, cancellationToken: TestContext.Current.CancellationToken).Degraded.ShouldBeFalse();
sink.ConditionExists("al-1").ShouldBeTrue();
gate.Reset();
// A surgical pure-add (new raw tags) that keeps al-1.
actor.Tell(Rebuild(AlarmArtifact(rawTags: 2)));
actor.Tell(new OpcUaPublishActor.AlarmStateUpdate("al-1", ActiveUnacked(OpcUaStatus.Good), DateTime.UtcNow, AddressSpaceRealm.Uns));
AwaitAssert(() => sink.AlarmWritesTo("al-1").ShouldBe(1), PlanBudget, cancellationToken: TestContext.Current.CancellationToken); // the live write landed
sink.AcknowledgeOutOfBand("al-1"); // an operator acks in the window
gate.Set();
ExpectMsg(PlanBudget, cancellationToken: TestContext.Current.CancellationToken).Degraded.ShouldBeFalse();
sink.SawRebuild.ShouldBeFalse("non-vacuity: a surgical apply");
sink.AlarmStateOf("al-1")!.Acknowledged.ShouldBeTrue("the operator's ack must survive the apply");
sink.AlarmWritesTo("al-1").ShouldBe(1, "no replayed write (and so no extra Part 9 event) on a kept condition");
}
/// M1 — a variable the surgical plan keeps took the live write; replaying it would revert a client
/// Write applied in the window. Positive control in the same apply: the node the plan ADDS gets its value.
[Fact]
public void A_surgical_apply_replays_values_only_onto_the_variables_it_added()
{
var sink = new ExistenceSink();
using var gate = new ManualResetEventSlim(true);
var actor = Spawn(sink, gate);
actor.Tell(Rebuild(RawTagArtifact(1)));
ExpectMsg(PlanBudget, cancellationToken: TestContext.Current.CancellationToken);
const string kept = "Plant/Modbus/dev1/T0";
gate.Reset();
actor.Tell(Rebuild(RawTagArtifact(3)));
var at = DateTime.UtcNow;
actor.Tell(new OpcUaPublishActor.AttributeValueBatchUpdate(
[
new ValueWrite(kept, 5, OpcUaStatus.Good, at, AddressSpaceRealm.Raw),
new ValueWrite(NewTag, 7, OpcUaStatus.Good, at, AddressSpaceRealm.Raw),
]));
AwaitAssert(() => sink.ValueOf(kept)!.Value.Value.ShouldBe(5), PlanBudget, cancellationToken: TestContext.Current.CancellationToken);
sink.SetValueOutOfBand(kept, 9); // a client Write in the window
gate.Set();
ExpectMsg(PlanBudget, cancellationToken: TestContext.Current.CancellationToken);
sink.SawRebuild.ShouldBeFalse("non-vacuity: a surgical apply");
sink.ValueOf(kept)!.Value.Value.ShouldBe(9, "the kept node's client write is not reverted");
sink.ValueOf(NewTag)!.Value.Value.ShouldBe(7, "positive control: the added node gets its deferred value");
}
/// M1 — a FULL rebuild recreates every variable, so every deferred value replays (the live write landed
/// on a node the rebuild discarded).
[Fact]
public void A_rebuild_replays_every_deferred_value()
{
var sink = new ExistenceSink();
using var gate = new ManualResetEventSlim(true);
var actor = Spawn(sink, gate);
actor.Tell(Rebuild(AlarmArtifact(rawTags: 1)));
ExpectMsg(PlanBudget, cancellationToken: TestContext.Current.CancellationToken);
const string kept = "Plant/Modbus/dev1/T0";
sink.Exists(kept).ShouldBeTrue();
gate.Reset();
// An equipment rename is a ChangedEquipment delta ⇒ full rebuild; the raw tag keeps its path.
actor.Tell(Rebuild(AlarmArtifact(equipmentName: "Eq-renamed", rawTags: 1)));
var at = DateTime.UtcNow;
actor.Tell(new OpcUaPublishActor.AttributeValueBatchUpdate([new ValueWrite(kept, 5, OpcUaStatus.Good, at, AddressSpaceRealm.Raw)]));
AwaitAssert(() => sink.ValueOf(kept)!.Value.Value.ShouldBe(5), PlanBudget, cancellationToken: TestContext.Current.CancellationToken);
gate.Set();
ExpectMsg(PlanBudget, cancellationToken: TestContext.Current.CancellationToken);
sink.SawRebuild.ShouldBeTrue("non-vacuity: a full rebuild ran");
sink.ValueOf(kept)!.Value.Value.ShouldBe(5, "the recreated node gets the value its predecessor took");
}
/// M4 — a FAILED plan applies nothing, so nothing is replayed, and the buffer is dropped once the
/// pipeline is idle: a later plan that adds the node does not receive a write from the failed window.
[Fact]
public void A_failed_plan_replays_nothing_and_drops_the_buffer()
{
var sink = new ExistenceSink();
using var gate = new ManualResetEventSlim(true);
var actor = Spawn(sink, gate);
actor.Tell(Rebuild(RawTagArtifact(2)));
ExpectMsg(PlanBudget, cancellationToken: TestContext.Current.CancellationToken);
gate.Reset();
actor.Tell(Rebuild("not json"u8.ToArray()));
actor.Tell(new OpcUaPublishActor.AttributeValueBatchUpdate(
[new ValueWrite("Plant/Modbus/dev1/T2", 42, OpcUaStatus.Good, DateTime.UtcNow, AddressSpaceRealm.Raw)]));
AwaitAssert(() => sink.SkippedWrites.ShouldContain("Plant/Modbus/dev1/T2"), PlanBudget, cancellationToken: TestContext.Current.CancellationToken);
// Snapshot AFTER the live write (the sink's counter cannot tell a live write after an apply from a replay).
var replayedBefore = sink.ReplayedItems;
gate.Set();
ExpectMsg(PlanBudget, cancellationToken: TestContext.Current.CancellationToken).Degraded.ShouldBeTrue();
sink.ReplayedItems.ShouldBe(replayedBefore, "nothing was applied, so nothing is replayed");
actor.Tell(Rebuild(RawTagArtifact(3)));
ExpectMsg(PlanBudget, cancellationToken: TestContext.Current.CancellationToken).Degraded.ShouldBeFalse();
sink.Exists("Plant/Modbus/dev1/T2").ShouldBeTrue("positive control: the later plan adds it");
sink.ValueOf("Plant/Modbus/dev1/T2").ShouldBeNull("the failed window's buffer was dropped");
}
/// M5 — a restart answers every QUEUED requester (degraded) instead of leaving it to the deploy timeout;
/// the in-flight one is answered by the incarnation check.
[Fact]
public void A_restart_answers_queued_rebuild_requesters()
{
var sink = new ExistenceSink();
using var gate = new ManualResetEventSlim(false);
var applier = new AddressSpaceApplier(sink, NullLogger.Instance);
var childProps = OpcUaPublishActor.PropsForTests(sink, applier, new OpcUaPublishActor.TestHooks { BeforePlan = () => gate.Wait() });
var parent = Sys.ActorOf(Props.Create(() => new RestartingParent(childProps)));
parent.Tell(RestartingParent.GetChild.Instance);
var child = ExpectMsg(cancellationToken: TestContext.Current.CancellationToken);
var inFlight = Rebuild(RawTagArtifact(2));
var queued = Rebuild(RawTagArtifact(3));
child.Tell(inFlight);
child.Tell(queued);
child.Tell(Kill.Instance);
var first = ExpectMsg(PlanBudget, cancellationToken: TestContext.Current.CancellationToken);
first.Correlation.ShouldBe(queued.Correlation, "the queued requester is answered at the restart");
first.Degraded.ShouldBeTrue();
gate.Set();
var second = ExpectMsg(PlanBudget, cancellationToken: TestContext.Current.CancellationToken);
second.Correlation.ShouldBe(inFlight.Correlation);
second.Degraded.ShouldBeTrue();
}
///
/// N3 (i) — a full rebuild recreates every KEPT SCRIPTED condition fresh (only native conditions are carried
/// over), so the engine's re-assert handled in the window — landing on the discarded instance — must be
/// replayed onto the new one, or the active alarm reads Normal until its next transition (#487).
///
[Fact]
public void A_rebuild_replays_the_window_state_onto_every_kept_scripted_condition()
{
var sink = new ExistenceSink();
using var gate = new ManualResetEventSlim(true);
var actor = Spawn(sink, gate);
actor.Tell(Rebuild(AlarmArtifact()));
ExpectMsg(PlanBudget, cancellationToken: TestContext.Current.CancellationToken).Degraded.ShouldBeFalse();
gate.Reset();
actor.Tell(Rebuild(AlarmArtifact(equipmentName: "Eq-renamed"))); // ChangedEquipment ⇒ full rebuild, al-1 kept
actor.Tell(new OpcUaPublishActor.AlarmStateUpdate("al-1", ActiveUnacked(OpcUaStatus.Good), DateTime.UtcNow, AddressSpaceRealm.Uns));
AwaitAssert(() => sink.AlarmWritesTo("al-1").ShouldBe(1), PlanBudget, cancellationToken: TestContext.Current.CancellationToken); // landed live on the old instance
gate.Set();
ExpectMsg(PlanBudget, cancellationToken: TestContext.Current.CancellationToken).Degraded.ShouldBeFalse();
sink.SawRebuild.ShouldBeTrue("non-vacuity: a full rebuild ran");
sink.AlarmStateOf("al-1").ShouldNotBeNull("the re-assert must reach the recreated condition")
.Active.ShouldBeTrue();
}
///
/// N3 (ii) — the native counterpart: a kept NATIVE condition is restored from the node manager's carry-over
/// (which already holds the live write and the operator's ack), so it is not replayed — the ack survives and no
/// extra write lands.
///
[Fact]
public void A_rebuild_does_not_replay_onto_a_kept_native_condition()
{
var sink = new ExistenceSink();
using var gate = new ManualResetEventSlim(true);
var actor = Spawn(sink, gate);
actor.Tell(Rebuild(AlarmArtifact(nativeAlarm: true)));
ExpectMsg(PlanBudget, cancellationToken: TestContext.Current.CancellationToken).Degraded.ShouldBeFalse();
sink.ConditionExists(NativeAlarm).ShouldBeTrue();
gate.Reset();
actor.Tell(Rebuild(AlarmArtifact(equipmentName: "Eq-renamed", nativeAlarm: true)));
actor.Tell(new OpcUaPublishActor.AlarmStateUpdate(NativeAlarm, ActiveUnacked(OpcUaStatus.Good), DateTime.UtcNow, AddressSpaceRealm.Raw));
AwaitAssert(() => sink.AlarmWritesTo(NativeAlarm).ShouldBe(1), PlanBudget, cancellationToken: TestContext.Current.CancellationToken);
sink.AcknowledgeOutOfBand(NativeAlarm);
gate.Set();
ExpectMsg(PlanBudget, cancellationToken: TestContext.Current.CancellationToken).Degraded.ShouldBeFalse();
sink.SawRebuild.ShouldBeTrue("non-vacuity: a full rebuild ran");
sink.AlarmStateOf(NativeAlarm)!.Acknowledged.ShouldBeTrue("the carried-over ack survives");
sink.AlarmWritesTo(NativeAlarm).ShouldBe(1, "no replayed write on a carried-over native condition");
}
///
/// N4 — a surgical in-place DataType edit resets the raw node AND its UNS mirror (the node manager drops an
/// old-typed value), and neither is ensured by the apply. The window's value is replayed onto both, fitted to
/// the new type, so the two NodeIds of one value agree.
///
[Fact]
public void A_surgical_shape_edit_replays_the_window_value_fitted_onto_the_raw_node_and_its_mirror()
{
var sink = new ExistenceSink();
using var gate = new ManualResetEventSlim(true);
var actor = Spawn(sink, gate);
actor.Tell(Rebuild(ShapeArtifact("Float")));
ExpectMsg(PlanBudget, cancellationToken: TestContext.Current.CancellationToken).Degraded.ShouldBeFalse();
sink.Exists(RawSpeed).ShouldBeTrue();
sink.Exists(UnsSpeed).ShouldBeTrue("non-vacuity: the mirror exists");
gate.Reset();
actor.Tell(Rebuild(ShapeArtifact("Int32")));
var at = DateTime.UtcNow;
actor.Tell(new OpcUaPublishActor.AttributeValueBatchUpdate(
[
new ValueWrite(RawSpeed, 5.0f, OpcUaStatus.Good, at, AddressSpaceRealm.Raw),
new ValueWrite(UnsSpeed, 5.0f, OpcUaStatus.Good, at, AddressSpaceRealm.Uns),
]));
AwaitAssert(() => sink.ValueOf(RawSpeed)!.Value.Value.ShouldBe(5.0f), PlanBudget, cancellationToken: TestContext.Current.CancellationToken);
gate.Set();
ExpectMsg(PlanBudget, cancellationToken: TestContext.Current.CancellationToken).Degraded.ShouldBeFalse();
sink.SawRebuild.ShouldBeFalse("non-vacuity: a surgical in-place edit");
sink.ReshapedNodes.ShouldContain(RawSpeed, "non-vacuity: the raw node was reset");
sink.ReshapedNodes.ShouldContain(UnsSpeed, "non-vacuity: the mirror was reset");
var raw = sink.ValueOf(RawSpeed).ShouldNotBeNull();
raw.Value.ShouldBeOfType().ShouldBe(5);
sink.ValueOf(UnsSpeed).ShouldBe(raw, "the raw node and its UNS mirror agree");
}
/// N4 — a window value that cannot be fitted to the new shape is never written: both NodeIds stay
/// value-less (BadWaitingForInitialData) until the driver's next value.
[Fact]
public void A_surgical_shape_edit_never_replays_a_value_that_does_not_fit_the_new_shape()
{
var sink = new ExistenceSink();
using var gate = new ManualResetEventSlim(true);
var actor = Spawn(sink, gate);
actor.Tell(Rebuild(ShapeArtifact("String")));
ExpectMsg(PlanBudget, cancellationToken: TestContext.Current.CancellationToken);
gate.Reset();
actor.Tell(Rebuild(ShapeArtifact("Int32")));
var at = DateTime.UtcNow;
actor.Tell(new OpcUaPublishActor.AttributeValueBatchUpdate(
[
new ValueWrite(RawSpeed, "fast", OpcUaStatus.Good, at, AddressSpaceRealm.Raw),
new ValueWrite(UnsSpeed, "fast", OpcUaStatus.Good, at, AddressSpaceRealm.Uns),
]));
AwaitAssert(() => sink.ValueOf(RawSpeed)!.Value.Value.ShouldBe("fast"), PlanBudget, cancellationToken: TestContext.Current.CancellationToken);
gate.Set();
ExpectMsg(PlanBudget, cancellationToken: TestContext.Current.CancellationToken);
sink.ReshapedNodes.ShouldContain(RawSpeed, "non-vacuity: the raw node was reset");
sink.ValueOf(RawSpeed).ShouldBeNull("\"fast\" is not an Int32 — the node keeps waiting");
sink.ValueOf(UnsSpeed).ShouldBeNull();
}
private const string NativeAlarm = "Plant/Modbus/dev1/Hi";
private const string RawSpeed = "Plant/Modbus/dev1/Speed";
private const string UnsSpeed = "Area/Line/Eq/Speed";
/// One raw tag (Speed) of and one UNS reference to it under Area/Line/Eq.
private static byte[] ShapeArtifact(string dataType) =>
JsonSerializer.SerializeToUtf8Bytes(new
{
UnsAreas = new[] { new { UnsAreaId = "area-1", ClusterId = "c1", Name = "Area" } },
UnsLines = new[] { new { UnsLineId = "line-1", UnsAreaId = "area-1", Name = "Line" } },
Equipment = new[] { new { EquipmentId = "eq-1", DriverInstanceId = (string?)null, UnsLineId = "line-1", Name = "Eq", MachineCode = "EQ-1" } },
RawFolders = new[] { new { RawFolderId = "rf-1", ParentRawFolderId = (string?)null, Name = "Plant", ClusterId = "c1" } },
DriverInstances = new[] { new { DriverInstanceId = "drv-1", DriverType = "Modbus", DriverConfig = "{}", Name = "Modbus", RawFolderId = "rf-1", ClusterId = "c1" } },
Devices = new[] { new { DeviceId = "dev-1", DriverInstanceId = "drv-1", Name = "dev1", DeviceConfig = "{}" } },
TagGroups = Array.Empty