Joseph Doherty
a8284f499e
N1: a full alarm-state snapshot also sets the condition's Quality, so it now supersedes an earlier quality-only write for the same id in the deferred buffer; a quality arriving after the state is kept and replays after it. N2 / M1: the replay is not inert on a node the apply KEPT - an SDK-applied Acknowledge/Confirm/Shelve or an optimistic client Write in the window would be rolled back (with spurious Part 9 events). The replay is now scoped to nodes the apply created: variables the applier ensured during this apply (AddressSpaceApplier.BeginRecordingEnsuredVariables) or every variable after a full rebuild; conditions in the applied composition but not the previous one (also on a rebuild, which carries a kept condition's state over). Option (b), a sink that reports whether a live write landed, needs the Commons DeferredAddressSpaceSink seam outside this lane (follow-up F7). M2: the benchmark remark now matches its Category=Benchmark trait. M3: TestHooks and the seam-carrying constructor are internal; PropsForTests produces the actor through an IIndirectActorProducer. M4: pinned replay-nothing-and-drop-the-buffer after a failed plan. M5: PreRestart answers every queued rebuild requester (degraded). Refs #732 Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
662 lines
34 KiB
C#
662 lines
34 KiB
C#
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;
|
|
|
|
/// <summary>
|
|
/// 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.
|
|
/// </summary>
|
|
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); // non-vacuity: it really missed
|
|
|
|
gate.Set();
|
|
ExpectMsg<OpcUaPublishActor.RebuildCompleted>(PlanBudget).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<OpcUaPublishActor.RebuildCompleted>(PlanBudget);
|
|
|
|
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<OpcUaPublishActor.RebuildCompleted>(PlanBudget);
|
|
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);
|
|
sink.Exists(NewTag).ShouldBeFalse("the plan is still in flight");
|
|
|
|
gate.Set();
|
|
ExpectMsg<OpcUaPublishActor.RebuildCompleted>(PlanBudget);
|
|
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);
|
|
|
|
gate.Set();
|
|
ExpectMsg<OpcUaPublishActor.RebuildCompleted>(PlanBudget).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<AddressSpaceApplier>.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);
|
|
|
|
gate.Set();
|
|
ExpectMsg<OpcUaPublishActor.RebuildCompleted>(PlanBudget);
|
|
|
|
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<OpcUaPublishActor.RebuildCompleted>(PlanBudget);
|
|
var replayed = sink.ReplayedItems;
|
|
replayed.ShouldBe(1);
|
|
|
|
// A later plan replays only what arrived during IT.
|
|
actor.Tell(Rebuild(RawTagArtifact(3)));
|
|
ExpectMsg<OpcUaPublishActor.RebuildCompleted>(PlanBudget);
|
|
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<AddressSpaceApplier>.Instance),
|
|
new OpcUaPublishActor.TestHooks { ParseAuthority = (_, _) => throw new InvalidOperationException("bad grants") },
|
|
dataPlaneAuthority: authority));
|
|
|
|
actor.Tell(Rebuild(RawTagArtifact(2)));
|
|
|
|
var completed = ExpectMsg<OpcUaPublishActor.RebuildCompleted>(PlanBudget);
|
|
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<AddressSpaceApplier>.Instance),
|
|
hooks: null, dataPlaneAuthority: authority));
|
|
healthy.Tell(Rebuild(RawTagArtifact(2)));
|
|
ExpectMsg<OpcUaPublishActor.RebuildCompleted>(PlanBudget).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)); // the bytes-less one must not overtake the held plan
|
|
|
|
gate.Set();
|
|
var a = ExpectMsg<OpcUaPublishActor.RebuildCompleted>(PlanBudget);
|
|
var b = ExpectMsg<OpcUaPublishActor.RebuildCompleted>(PlanBudget);
|
|
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<AddressSpaceApplier>.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<IActorRef>();
|
|
|
|
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));
|
|
|
|
gate.Set();
|
|
var answer = ExpectMsg<OpcUaPublishActor.RebuildCompleted>(PlanBudget);
|
|
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<OpcUaPublishActor.RebuildCompleted>(PlanBudget).Degraded.ShouldBeFalse();
|
|
sink.Exists(NewTag).ShouldBeTrue();
|
|
}
|
|
|
|
// ───────────────────── fix round 2: N1 / N2 / M1 / M4 / M5 ─────────────────────
|
|
|
|
/// <summary>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).</summary>
|
|
[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);
|
|
gate.Set();
|
|
ExpectMsg<OpcUaPublishActor.RebuildCompleted>(PlanBudget);
|
|
|
|
sink.AlarmQualityOf("al-1").ShouldBe(OpcUaStatus.Good, "the later snapshot's quality is what live would leave");
|
|
sink.AlarmStateOf("al-1")!.Active.ShouldBeTrue();
|
|
}
|
|
|
|
/// <summary>N1 positive control — the reverse order: a quality-only write AFTER the state still lands last.</summary>
|
|
[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);
|
|
gate.Set();
|
|
ExpectMsg<OpcUaPublishActor.RebuildCompleted>(PlanBudget);
|
|
|
|
sink.AlarmQualityOf("al-1").ShouldBe(OpcUaStatus.BadNoCommunication);
|
|
}
|
|
|
|
/// <summary>
|
|
/// 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. Driven over
|
|
/// both apply shapes: a surgical pure-add, and a full rebuild (which carries the condition's state over).
|
|
/// </summary>
|
|
[Theory]
|
|
[InlineData(false)]
|
|
[InlineData(true)]
|
|
public void An_out_of_band_ack_on_a_kept_condition_is_not_reverted_by_the_replay(bool rebuild)
|
|
{
|
|
var sink = new ExistenceSink();
|
|
using var gate = new ManualResetEventSlim(true);
|
|
var actor = Spawn(sink, gate);
|
|
actor.Tell(Rebuild(AlarmArtifact()));
|
|
ExpectMsg<OpcUaPublishActor.RebuildCompleted>(PlanBudget).Degraded.ShouldBeFalse();
|
|
sink.ConditionExists("al-1").ShouldBeTrue();
|
|
|
|
gate.Reset();
|
|
// A surgical pure-add (new raw tags) or a full rebuild (an equipment rename) that keeps al-1.
|
|
actor.Tell(Rebuild(rebuild ? AlarmArtifact(equipmentName: "Eq-renamed") : 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); // the live write landed
|
|
sink.AcknowledgeOutOfBand("al-1"); // an operator acks in the window
|
|
|
|
gate.Set();
|
|
ExpectMsg<OpcUaPublishActor.RebuildCompleted>(PlanBudget).Degraded.ShouldBeFalse();
|
|
|
|
sink.SawRebuild.ShouldBe(rebuild, "non-vacuity: the intended apply shape ran");
|
|
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");
|
|
}
|
|
|
|
/// <summary>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.</summary>
|
|
[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<OpcUaPublishActor.RebuildCompleted>(PlanBudget);
|
|
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);
|
|
sink.SetValueOutOfBand(kept, 9); // a client Write in the window
|
|
|
|
gate.Set();
|
|
ExpectMsg<OpcUaPublishActor.RebuildCompleted>(PlanBudget);
|
|
|
|
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");
|
|
}
|
|
|
|
/// <summary>M1 — a FULL rebuild recreates every variable, so every deferred value replays (the live write landed
|
|
/// on a node the rebuild discarded).</summary>
|
|
[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<OpcUaPublishActor.RebuildCompleted>(PlanBudget);
|
|
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);
|
|
|
|
gate.Set();
|
|
ExpectMsg<OpcUaPublishActor.RebuildCompleted>(PlanBudget);
|
|
|
|
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");
|
|
}
|
|
|
|
/// <summary>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.</summary>
|
|
[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<OpcUaPublishActor.RebuildCompleted>(PlanBudget);
|
|
|
|
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);
|
|
// 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<OpcUaPublishActor.RebuildCompleted>(PlanBudget).Degraded.ShouldBeTrue();
|
|
sink.ReplayedItems.ShouldBe(replayedBefore, "nothing was applied, so nothing is replayed");
|
|
|
|
actor.Tell(Rebuild(RawTagArtifact(3)));
|
|
ExpectMsg<OpcUaPublishActor.RebuildCompleted>(PlanBudget).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");
|
|
}
|
|
|
|
/// <summary>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.</summary>
|
|
[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<AddressSpaceApplier>.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<IActorRef>();
|
|
|
|
var inFlight = Rebuild(RawTagArtifact(2));
|
|
var queued = Rebuild(RawTagArtifact(3));
|
|
child.Tell(inFlight);
|
|
child.Tell(queued);
|
|
child.Tell(Kill.Instance);
|
|
|
|
var first = ExpectMsg<OpcUaPublishActor.RebuildCompleted>(PlanBudget);
|
|
first.Correlation.ShouldBe(queued.Correlation, "the queued requester is answered at the restart");
|
|
first.Degraded.ShouldBeTrue();
|
|
|
|
gate.Set();
|
|
var second = ExpectMsg<OpcUaPublishActor.RebuildCompleted>(PlanBudget);
|
|
second.Correlation.ShouldBe(inFlight.Correlation);
|
|
second.Degraded.ShouldBeTrue();
|
|
}
|
|
|
|
private static AlarmConditionSnapshot ActiveUnacked(uint quality) =>
|
|
new(Active: true, Acknowledged: false, Confirmed: true, Enabled: true, AlarmShelvingKind.Unshelved, 500, "high", quality);
|
|
|
|
// ───────────────────────────── helpers ─────────────────────────────
|
|
|
|
private IActorRef Spawn(ExistenceSink sink, ManualResetEventSlim gate) =>
|
|
Sys.ActorOf(OpcUaPublishActor.PropsForTests(
|
|
sink, new AddressSpaceApplier(sink, NullLogger<AddressSpaceApplier>.Instance),
|
|
new OpcUaPublishActor.TestHooks { BeforePlan = () => gate.Wait() }));
|
|
|
|
private static OpcUaPublishActor.RebuildAddressSpace Rebuild(byte[] artifact) =>
|
|
new(CorrelationId.NewId(), DeploymentId.NewId(), artifact, ReportOutcome: true);
|
|
|
|
private static byte[] RawTagArtifact(int tagCount) =>
|
|
JsonSerializer.SerializeToUtf8Bytes(new
|
|
{
|
|
UnsAreas = Array.Empty<object>(),
|
|
UnsLines = Array.Empty<object>(),
|
|
Equipment = Array.Empty<object>(),
|
|
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<object>(),
|
|
Tags = Enumerable.Range(0, tagCount).Select(i => new
|
|
{
|
|
TagId = $"t-{i}", DeviceId = "dev-1", TagGroupId = (string?)null, Name = $"T{i}",
|
|
DataType = "Float", AccessLevel = 0, TagConfig = "{}",
|
|
}).ToArray(),
|
|
UnsTagReferences = Array.Empty<object>(),
|
|
ScriptedAlarms = Array.Empty<object>(),
|
|
});
|
|
|
|
private static byte[] AlarmArtifact(string equipmentName = "Eq", int rawTags = 0) =>
|
|
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 = equipmentName, MachineCode = "EQ-1" },
|
|
},
|
|
RawFolders = rawTags == 0
|
|
? Array.Empty<object>()
|
|
: new object[] { new { RawFolderId = "rf-1", ParentRawFolderId = (string?)null, Name = "Plant", ClusterId = "c1" } },
|
|
DriverInstances = rawTags == 0
|
|
? Array.Empty<object>()
|
|
: new object[] { new { DriverInstanceId = "drv-1", DriverType = "Modbus", DriverConfig = "{}", Name = "Modbus", RawFolderId = "rf-1", ClusterId = "c1" } },
|
|
Devices = rawTags == 0
|
|
? Array.Empty<object>()
|
|
: new object[] { new { DeviceId = "dev-1", DriverInstanceId = "drv-1", Name = "dev1", DeviceConfig = "{}" } },
|
|
TagGroups = Array.Empty<object>(),
|
|
Tags = Enumerable.Range(0, rawTags).Select(i => (object)new
|
|
{
|
|
TagId = $"t-{i}", DeviceId = "dev-1", TagGroupId = (string?)null, Name = $"T{i}",
|
|
DataType = "Float", AccessLevel = 0, TagConfig = "{}",
|
|
}).ToArray(),
|
|
ScriptedAlarms = new[]
|
|
{
|
|
new
|
|
{
|
|
ScriptedAlarmId = "al-1", EquipmentId = "eq-1", Name = "High", AlarmType = "AlarmCondition",
|
|
Severity = 500, MessageTemplate = "high", PredicateScriptId = "s-1", HistorizeToAveva = false,
|
|
Retain = true, Enabled = true,
|
|
},
|
|
},
|
|
Scripts = new[] { new { ScriptId = "s-1", SourceCode = "return false;" } },
|
|
});
|
|
|
|
/// <summary>Restarts its one child on any failure, and lets the test reach the child.</summary>
|
|
private sealed class RestartingParent : ReceiveActor
|
|
{
|
|
public sealed class GetChild { public static readonly GetChild Instance = new(); }
|
|
|
|
public RestartingParent(Props childProps)
|
|
{
|
|
var child = Context.ActorOf(childProps, "publish");
|
|
Receive<GetChild>(_ => Sender.Tell(child));
|
|
}
|
|
|
|
protected override SupervisorStrategy SupervisorStrategy() =>
|
|
new OneForOneStrategy(_ => Directive.Restart);
|
|
}
|
|
|
|
/// <summary>An authority sink that counts installs.</summary>
|
|
private sealed class RecordingAuthoritySink : IDataPlaneAuthoritySink
|
|
{
|
|
private int _applies;
|
|
public int Applies => Volatile.Read(ref _applies);
|
|
|
|
public DataPlaneAuthoritySnapshot Apply(
|
|
string? clusterId, string revision, IReadOnlyList<ZB.MOM.WW.OtOpcUa.Configuration.Entities.ApiKey> keys,
|
|
IReadOnlyList<ZB.MOM.WW.OtOpcUa.Configuration.Entities.NodeAcl> grants,
|
|
IReadOnlyDictionary<AclScopeRef, IReadOnlyList<NodeAclPath>> scopePaths)
|
|
{
|
|
Interlocked.Increment(ref _applies);
|
|
return new DataPlaneAuthorityState().Apply(clusterId, revision, keys, grants, scopePaths);
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Models the node manager's non-creating writes: a write to a node that was never ensured (or was removed)
|
|
/// is skipped and recorded; a full sink rebuild forgets every node. <see cref="ReplayedItems"/> counts the
|
|
/// items written after an apply closed and before the next apply starts — the actor's replay, since the
|
|
/// tests send nothing between releasing a plan and its answer.
|
|
/// </summary>
|
|
private sealed class ExistenceSink : IOpcUaAddressSpaceSink, ISurgicalAddressSpaceSink
|
|
{
|
|
private readonly object _gate = new();
|
|
private readonly HashSet<string> _variables = new();
|
|
private readonly HashSet<string> _conditions = new();
|
|
private readonly Dictionary<string, (object? Value, uint Status, DateTime At)> _values = new();
|
|
private readonly Dictionary<string, AlarmConditionSnapshot> _alarmStates = new();
|
|
private readonly Dictionary<string, uint> _alarmQualities = new();
|
|
|
|
public ConcurrentQueue<string> SkippedWrites { get; } = new();
|
|
private readonly Dictionary<string, int> _alarmWrites = new();
|
|
private volatile bool _sawRebuild;
|
|
|
|
/// <summary>Whether any apply called <see cref="RebuildAddressSpace"/> (the full-rebuild shape).</summary>
|
|
public bool SawRebuild => _sawRebuild;
|
|
public int AlarmWritesTo(string id) { lock (_gate) return _alarmWrites.GetValueOrDefault(id); }
|
|
public void SetValueOutOfBand(string id, object? value)
|
|
{
|
|
lock (_gate) { var c = _values.GetValueOrDefault(id); _values[id] = (value, c.Status, c.At); }
|
|
}
|
|
|
|
/// <summary>What an SDK-applied Part 9 Acknowledge does: mutates the condition directly, no sink write.</summary>
|
|
public void AcknowledgeOutOfBand(string id)
|
|
{
|
|
lock (_gate) _alarmStates[id] = _alarmStates[id] with { Acknowledged = true };
|
|
}
|
|
public ConcurrentQueue<string> SkippedAlarms { get; } = new();
|
|
public int ReplayedItems;
|
|
|
|
public bool Exists(string id) { lock (_gate) return _variables.Contains(id); }
|
|
public bool ConditionExists(string id) { lock (_gate) return _conditions.Contains(id); }
|
|
public (object? Value, uint Status, DateTime At)? ValueOf(string id) { lock (_gate) return _values.TryGetValue(id, out var v) ? v : null; }
|
|
public AlarmConditionSnapshot? AlarmStateOf(string id) { lock (_gate) return _alarmStates.GetValueOrDefault(id); }
|
|
public uint? AlarmQualityOf(string id) { lock (_gate) return _alarmQualities.TryGetValue(id, out var q) ? q : null; }
|
|
|
|
public void WriteValue(string nodeId, object? value, uint statusCode, DateTime ts, AddressSpaceRealm realm) =>
|
|
WriteValues([new ValueWrite(nodeId, value, statusCode, ts, realm)]);
|
|
|
|
public void WriteValues(IReadOnlyList<ValueWrite> values)
|
|
{
|
|
lock (_gate)
|
|
{
|
|
if (_applyClosed) ReplayedItems += values.Count;
|
|
foreach (var v in values)
|
|
{
|
|
if (!_variables.Contains(v.NodeId)) { SkippedWrites.Enqueue(v.NodeId); continue; }
|
|
_values[v.NodeId] = (v.Value, v.StatusCode, v.SourceTimestampUtc);
|
|
}
|
|
}
|
|
}
|
|
|
|
public void WriteStatuses(IReadOnlyList<StatusWrite> statuses)
|
|
{
|
|
lock (_gate)
|
|
{
|
|
if (_applyClosed) ReplayedItems += statuses.Count;
|
|
foreach (var s in statuses)
|
|
{
|
|
if (!_variables.Contains(s.NodeId)) { SkippedWrites.Enqueue(s.NodeId); continue; }
|
|
var current = _values.GetValueOrDefault(s.NodeId);
|
|
_values[s.NodeId] = (current.Value, s.StatusCode, s.SourceTimestampUtc);
|
|
}
|
|
}
|
|
}
|
|
|
|
public void WriteAlarmCondition(string alarmNodeId, AlarmConditionSnapshot state, DateTime ts, AddressSpaceRealm realm)
|
|
{
|
|
lock (_gate)
|
|
{
|
|
if (!_conditions.Contains(alarmNodeId)) { SkippedAlarms.Enqueue(alarmNodeId); return; }
|
|
_alarmWrites[alarmNodeId] = _alarmWrites.GetValueOrDefault(alarmNodeId) + 1;
|
|
// The node manager projects the snapshot's StatusCode onto the condition's Quality.
|
|
_alarmStates[alarmNodeId] = state;
|
|
_alarmQualities[alarmNodeId] = state.StatusCode;
|
|
}
|
|
}
|
|
|
|
public void WriteAlarmQuality(string alarmNodeId, uint statusCode, DateTime ts, AddressSpaceRealm realm)
|
|
{
|
|
lock (_gate)
|
|
{
|
|
if (!_conditions.Contains(alarmNodeId)) { SkippedAlarms.Enqueue(alarmNodeId); return; }
|
|
_alarmQualities[alarmNodeId] = statusCode;
|
|
}
|
|
}
|
|
|
|
public void MaterialiseAlarmCondition(string alarmNodeId, string equipmentNodeId, string displayName, string alarmType, int severity, AddressSpaceRealm realm, bool isNative = false)
|
|
{
|
|
lock (_gate) { _applyClosed = false; _conditions.Add(alarmNodeId); }
|
|
}
|
|
|
|
public void EnsureFolder(string folderNodeId, string? parentNodeId, string displayName, AddressSpaceRealm realm)
|
|
{
|
|
lock (_gate) _applyClosed = false;
|
|
}
|
|
|
|
public void EnsureVariable(string variableNodeId, string? parentFolderNodeId, string displayName, string dataType, bool writable, AddressSpaceRealm realm, string? historianTagname = null, bool isArray = false, uint? arrayLength = null, bool refuseArrayOnScalar = false)
|
|
{
|
|
lock (_gate) { _applyClosed = false; _variables.Add(variableNodeId); }
|
|
}
|
|
|
|
public void RebuildAddressSpace()
|
|
{
|
|
lock (_gate)
|
|
{
|
|
// Variables are recreated empty; a condition's state is captured and carried over to its
|
|
// re-materialised instance (the node manager's rebuild carry-over).
|
|
_applyClosed = false;
|
|
_sawRebuild = true;
|
|
_variables.Clear();
|
|
_conditions.Clear();
|
|
_values.Clear();
|
|
}
|
|
}
|
|
|
|
private bool _applyClosed;
|
|
|
|
public void CompleteAddressSpaceApply()
|
|
{
|
|
lock (_gate) _applyClosed = true;
|
|
}
|
|
|
|
public void RaiseNodesAddedModelChange(string affectedNodeId, AddressSpaceRealm realm) { }
|
|
public void WireAlarmNotifiers(string alarmNodeId, AddressSpaceRealm alarmRealm, IReadOnlyList<string> notifierFolderNodeIds, AddressSpaceRealm notifierFolderRealm) { }
|
|
public void AddReference(string sourceNodeId, AddressSpaceRealm sourceRealm, string targetNodeId, AddressSpaceRealm targetRealm, string referenceType = "Organizes") { }
|
|
public bool UpdateTagAttributes(string variableNodeId, bool writable, string? historianTagname, string dataType, bool isArray, uint? arrayLength, AddressSpaceRealm realm) => true;
|
|
public bool UpdateFolderDisplayName(string folderNodeId, string displayName, AddressSpaceRealm realm) => true;
|
|
public bool RemoveVariableNode(string variableNodeId, AddressSpaceRealm realm) { lock (_gate) return _variables.Remove(variableNodeId); }
|
|
public bool RemoveAlarmConditionNode(string alarmNodeId, AddressSpaceRealm realm) { lock (_gate) return _conditions.Remove(alarmNodeId); }
|
|
public bool RemoveEquipmentSubtree(string equipmentNodeId, AddressSpaceRealm realm) => true;
|
|
}
|
|
}
|