Joseph Doherty
50b55ee4b9
RedundancyStateActor derived the driver Primary from ClusterState.RoleLeader("driver").
Akka offers two different notions of a "first" member and they are not the same
one: role leader is the lowest-ADDRESSED Up member (host, then port), while
ClusterSingletonManager places singletons on the OLDEST — lowest up-number.
They agree on a freshly-formed cluster, which is why every existing test passed.
They diverge after any restart: the restarted node re-joins as the youngest while
keeping its address, so if it holds the lower address it becomes role leader while
the singletons stay put. The snapshot would then name a Primary that is not
hosting the work, and every Primary-gated surface follows it — inbound device
writes, native-alarm acks, the fleet-wide alerts emit, and the alarm-history
drain would all enable on the wrong node while the node actually running the
singletons stayed gated off.
BuildSnapshot now selects the oldest Up member carrying the driver role, matching
singleton placement. Leaving members are excluded: a node handing its singletons
over must not be named Primary.
NodeRedundancyState.IsRoleLeaderForDriver is renamed IsDriverPrimary, and
NodeHealthInputs.IsDriverRoleLeader likewise. Keeping the old names would have
left the wire contract asserting a derivation the code no longer uses — the same
drift that made this defect invisible.
Proven by a real two-node cluster rather than a mock. RedundancyPrimaryElectionTests
binds the first-joining node to the HIGHER port, so oldest and lowest-address name
different nodes, and includes a fixture assertion that the divergence actually
occurred — without it the real assertion could pass for the wrong reason. Positive
control: restoring the RoleLeader derivation turns exactly the two election tests
red while the fixture check stays green.
Runtime.Tests 440 passed, ControlPlane.Tests 82, Cluster.Tests 36.
Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW
691 lines
36 KiB
C#
691 lines
36 KiB
C#
using System.Collections.Concurrent;
|
|
using Akka.Actor;
|
|
using Microsoft.Extensions.Logging.Abstractions;
|
|
using Shouldly;
|
|
using Xunit;
|
|
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Redundancy;
|
|
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
|
|
using ZB.MOM.WW.OtOpcUa.Commons.Types;
|
|
using ZB.MOM.WW.OtOpcUa.OpcUaServer;
|
|
using ZB.MOM.WW.OtOpcUa.Runtime.Health;
|
|
using ZB.MOM.WW.OtOpcUa.Runtime.OpcUa;
|
|
using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness;
|
|
|
|
namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.OpcUa;
|
|
|
|
public sealed class OpcUaPublishActorTests : RuntimeActorTestBase
|
|
{
|
|
/// <summary>Verifies that message contracts are accepted without a pinned dispatcher in tests.</summary>
|
|
[Fact]
|
|
public void Accepts_message_contracts_without_pinned_dispatcher_in_tests()
|
|
{
|
|
var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests());
|
|
actor.Tell(new OpcUaPublishActor.AttributeValueUpdate("ns=2;s=Tag1", 42.0, OpcUaQuality.Good, DateTime.UtcNow, Realm: AddressSpaceRealm.Uns));
|
|
actor.Tell(new OpcUaPublishActor.AlarmStateUpdate("ns=2;s=Alarm1", Snapshot(active: true), DateTime.UtcNow, Realm: AddressSpaceRealm.Uns));
|
|
actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId()));
|
|
actor.Tell(new OpcUaPublishActor.ServiceLevelChanged(240));
|
|
|
|
ExpectNoMsg(TimeSpan.FromMilliseconds(200));
|
|
}
|
|
|
|
/// <summary>Verifies that production props target the OPC UA synchronized dispatcher.</summary>
|
|
[Fact]
|
|
public void Production_Props_targets_opcua_synchronized_dispatcher()
|
|
{
|
|
var props = OpcUaPublishActor.Props();
|
|
props.Dispatcher.ShouldBe(OpcUaPublishActor.DispatcherId);
|
|
}
|
|
|
|
/// <summary>Verifies that AttributeValueUpdate routes to sink WriteValue.</summary>
|
|
[Fact]
|
|
public void AttributeValueUpdate_routes_to_sink_WriteValue()
|
|
{
|
|
var sink = new RecordingSink();
|
|
var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(sink: sink));
|
|
|
|
actor.Tell(new OpcUaPublishActor.AttributeValueUpdate("ns=2;s=T1", 3.14, OpcUaQuality.Good, DateTime.UtcNow, Realm: AddressSpaceRealm.Uns));
|
|
actor.Tell(new OpcUaPublishActor.AttributeValueUpdate("ns=2;s=T2", "abc", OpcUaQuality.Uncertain, DateTime.UtcNow, Realm: AddressSpaceRealm.Uns));
|
|
|
|
AwaitAssert(() =>
|
|
{
|
|
sink.Values.Count.ShouldBe(2);
|
|
sink.Values[0].NodeId.ShouldBe("ns=2;s=T1");
|
|
sink.Values[0].Value.ShouldBe(3.14);
|
|
sink.Values[0].Quality.ShouldBe(OpcUaQuality.Good);
|
|
sink.Values[1].Quality.ShouldBe(OpcUaQuality.Uncertain);
|
|
}, duration: TimeSpan.FromMilliseconds(500));
|
|
}
|
|
|
|
/// <summary>Verifies that AlarmStateUpdate routes to sink WriteAlarmCondition with the full snapshot.</summary>
|
|
[Fact]
|
|
public void AlarmStateUpdate_routes_to_sink_WriteAlarmCondition()
|
|
{
|
|
var sink = new RecordingSink();
|
|
var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(sink: sink));
|
|
|
|
actor.Tell(new OpcUaPublishActor.AlarmStateUpdate(
|
|
"ns=2;s=A1", Snapshot(active: true, acknowledged: false, severity: 700), DateTime.UtcNow, Realm: AddressSpaceRealm.Uns));
|
|
|
|
AwaitAssert(() =>
|
|
{
|
|
sink.Alarms.Count.ShouldBe(1);
|
|
sink.Alarms[0].AlarmNodeId.ShouldBe("ns=2;s=A1");
|
|
sink.Alarms[0].State.Active.ShouldBeTrue();
|
|
sink.Alarms[0].State.Acknowledged.ShouldBeFalse();
|
|
sink.Alarms[0].State.Severity.ShouldBe((ushort)700);
|
|
}, duration: TimeSpan.FromMilliseconds(500));
|
|
}
|
|
|
|
/// <summary>#477 — AlarmQualityUpdate routes to sink.WriteAlarmQuality with the quality + realm.</summary>
|
|
[Fact]
|
|
public void AlarmQualityUpdate_routes_to_sink_WriteAlarmQuality()
|
|
{
|
|
var sink = new RecordingSink();
|
|
var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(sink: sink));
|
|
|
|
actor.Tell(new OpcUaPublishActor.AlarmQualityUpdate(
|
|
"Plant/Modbus/dev1/temp_hi", OpcUaQuality.Bad, DateTime.UtcNow, Realm: AddressSpaceRealm.Raw));
|
|
|
|
AwaitAssert(() =>
|
|
{
|
|
sink.AlarmQualityQueue.Count.ShouldBe(1);
|
|
sink.AlarmQualityQueue.TryPeek(out var q).ShouldBeTrue();
|
|
q.AlarmNodeId.ShouldBe("Plant/Modbus/dev1/temp_hi");
|
|
q.Quality.ShouldBe(OpcUaQuality.Bad);
|
|
q.Realm.ShouldBe(AddressSpaceRealm.Raw);
|
|
}, duration: TimeSpan.FromMilliseconds(500));
|
|
}
|
|
|
|
/// <summary>Builds a test <see cref="AlarmConditionSnapshot"/> with sensible defaults so each test
|
|
/// only specifies the fields it cares about.</summary>
|
|
private static AlarmConditionSnapshot Snapshot(
|
|
bool active = false,
|
|
bool acknowledged = true,
|
|
bool confirmed = true,
|
|
bool enabled = true,
|
|
AlarmShelvingKind shelving = AlarmShelvingKind.Unshelved,
|
|
ushort severity = 500,
|
|
string message = "test") =>
|
|
new(active, acknowledged, confirmed, enabled, shelving, severity, message);
|
|
|
|
/// <summary>Verifies that RebuildAddressSpace calls sink Rebuild.</summary>
|
|
[Fact]
|
|
public void RebuildAddressSpace_calls_sink_Rebuild()
|
|
{
|
|
var sink = new RecordingSink();
|
|
var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(sink: sink));
|
|
|
|
actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId()));
|
|
|
|
AwaitAssert(() => sink.RebuildCalls.ShouldBe(1), duration: TimeSpan.FromMilliseconds(500));
|
|
}
|
|
|
|
/// <summary>Verifies that <see cref="OpcUaPublishActor.MaterialiseDiscoveredNodes"/> forwards to the
|
|
/// applier, which drives the sink to ensure the discovered folder + (read-only) variable and announce a
|
|
/// NodeAdded model-change under the equipment root — proving the message → handler → applier → sink path
|
|
/// end to end (mirrors the real-applier-over-recording-sink harness in
|
|
/// <c>OpcUaPublishActorRebuildTests</c>).</summary>
|
|
[Fact]
|
|
public void MaterialiseDiscoveredNodes_routes_through_applier_to_sink()
|
|
{
|
|
var sink = new RecordingSink();
|
|
var applier = new AddressSpaceApplier(sink, NullLogger<AddressSpaceApplier>.Instance);
|
|
var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(sink: sink, applier: applier));
|
|
|
|
var folders = new[] { new DiscoveredFolder("EQ-1/Axes", "EQ-1", "Axes") };
|
|
var variables = new[]
|
|
{
|
|
new DiscoveredVariable("EQ-1/Axes/X", "EQ-1/Axes", "X", "Double",
|
|
Writable: false, IsArray: false, ArrayLength: null),
|
|
};
|
|
|
|
actor.Tell(new OpcUaPublishActor.MaterialiseDiscoveredNodes("EQ-1", folders, variables));
|
|
|
|
AwaitAssert(() =>
|
|
{
|
|
sink.Folders.ShouldContain(("EQ-1/Axes", "EQ-1", "Axes"));
|
|
sink.Variables.ShouldContain(("EQ-1/Axes/X", "EQ-1/Axes", "X", "Double", false));
|
|
sink.ModelChanges.ShouldContain("EQ-1");
|
|
}, duration: TimeSpan.FromMilliseconds(500));
|
|
}
|
|
|
|
/// <summary>Verifies that ServiceLevelChanged publishes to IServiceLevelPublisher once per unique level.</summary>
|
|
[Fact]
|
|
public void ServiceLevelChanged_publishes_to_IServiceLevelPublisher_once_per_unique_level()
|
|
{
|
|
var publisher = new RecordingPublisher();
|
|
var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(serviceLevel: publisher));
|
|
|
|
actor.Tell(new OpcUaPublishActor.ServiceLevelChanged(240));
|
|
actor.Tell(new OpcUaPublishActor.ServiceLevelChanged(240)); // dedup
|
|
actor.Tell(new OpcUaPublishActor.ServiceLevelChanged(100));
|
|
|
|
AwaitAssert(() => publisher.Levels.ShouldBe(new byte[] { 240, 100 }),
|
|
duration: TimeSpan.FromMilliseconds(500));
|
|
}
|
|
|
|
/// <summary>Verifies that the very first computed ServiceLevel is always published even when it is
|
|
/// 0 — a node starting Detached (or role-less) must publish 0 rather than let the SDK default
|
|
/// (255 = full service) stand. The dedup against the <c>byte</c>-default <c>0</c> must not swallow
|
|
/// the first publish. Drives a Detached local entry (first computed level = 0) with no DB-health
|
|
/// probe and asserts the publisher saw exactly [0].</summary>
|
|
[Fact]
|
|
public void First_service_level_zero_is_published_not_deduped()
|
|
{
|
|
var publisher = new RecordingPublisher();
|
|
var local = NodeId.Parse("detached-node");
|
|
var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(
|
|
serviceLevel: publisher, localNode: local));
|
|
|
|
actor.Tell(new RedundancyStateChanged(
|
|
Nodes: new[]
|
|
{
|
|
new NodeRedundancyState(local, RedundancyRole.Detached,
|
|
IsClusterLeader: false, IsDriverPrimary: false, DateTime.UtcNow),
|
|
},
|
|
CorrelationId.NewId()));
|
|
|
|
AwaitAssert(() => publisher.Levels.ShouldBe(new byte[] { 0 }),
|
|
duration: TimeSpan.FromMilliseconds(500));
|
|
}
|
|
|
|
/// <summary>Verifies that RedundancyStateChanged drives local ServiceLevel publish for primary leader.</summary>
|
|
[Fact]
|
|
public void RedundancyStateChanged_drives_local_ServiceLevel_publish_for_primary_leader()
|
|
{
|
|
var publisher = new RecordingPublisher();
|
|
var local = NodeId.Parse("primary-node");
|
|
var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(
|
|
serviceLevel: publisher, localNode: local));
|
|
|
|
var snapshot = new RedundancyStateChanged(
|
|
Nodes: new[]
|
|
{
|
|
new NodeRedundancyState(local, RedundancyRole.Primary,
|
|
IsClusterLeader: true, IsDriverPrimary: true, DateTime.UtcNow),
|
|
new NodeRedundancyState(NodeId.Parse("other-node"), RedundancyRole.Secondary,
|
|
IsClusterLeader: false, IsDriverPrimary: false, DateTime.UtcNow),
|
|
},
|
|
CorrelationId.NewId());
|
|
actor.Tell(snapshot);
|
|
|
|
AwaitAssert(() => publisher.Levels.ShouldBe(new byte[] { 240 }),
|
|
duration: TimeSpan.FromMilliseconds(500));
|
|
}
|
|
|
|
/// <summary>Verifies that RedundancyStateChanged for secondary publishes 100.</summary>
|
|
[Fact]
|
|
public void RedundancyStateChanged_for_secondary_publishes_100()
|
|
{
|
|
var publisher = new RecordingPublisher();
|
|
var local = NodeId.Parse("secondary-node");
|
|
var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(serviceLevel: publisher, localNode: local));
|
|
|
|
var snapshot = new RedundancyStateChanged(
|
|
Nodes: new[]
|
|
{
|
|
new NodeRedundancyState(local, RedundancyRole.Secondary,
|
|
IsClusterLeader: false, IsDriverPrimary: false, DateTime.UtcNow),
|
|
},
|
|
CorrelationId.NewId());
|
|
actor.Tell(snapshot);
|
|
|
|
AwaitAssert(() => publisher.Levels.ShouldBe(new byte[] { 100 }),
|
|
duration: TimeSpan.FromMilliseconds(500));
|
|
}
|
|
|
|
/// <summary>Verifies that the calculator path computes 250 for a healthy primary role-leader
|
|
/// (basis 240 from DB-reachable + probe-ok + fresh, +10 driver-role-leader bonus).</summary>
|
|
[Fact]
|
|
public void Calculator_path_healthy_primary_leader_publishes_250()
|
|
{
|
|
var publisher = new RecordingPublisher();
|
|
var local = NodeId.Parse("primary-node");
|
|
var probe = Sys.ActorOf(Akka.Actor.Props.Create(() =>
|
|
new StubDbHealth(new DbHealthProbeActor.DbHealthStatus(true, DateTime.UtcNow, null))));
|
|
var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(
|
|
serviceLevel: publisher, localNode: local, dbHealthProbe: probe));
|
|
|
|
actor.Tell(new DbHealthProbeActor.DbHealthStatus(true, DateTime.UtcNow, null));
|
|
actor.Tell(new RedundancyStateChanged(
|
|
Nodes: new[]
|
|
{
|
|
new NodeRedundancyState(local, RedundancyRole.Primary,
|
|
IsClusterLeader: true, IsDriverPrimary: true, DateTime.UtcNow),
|
|
},
|
|
CorrelationId.NewId()));
|
|
|
|
AwaitAssert(() => publisher.Levels.ShouldContain((byte)250),
|
|
duration: TimeSpan.FromMilliseconds(500));
|
|
}
|
|
|
|
/// <summary>Verifies that the calculator path computes 240 for a healthy non-leader secondary
|
|
/// (basis 240 from DB-reachable + probe-ok + fresh, no leader bonus). Documented change from the
|
|
/// legacy role-only path, which mapped Secondary → 100.</summary>
|
|
[Fact]
|
|
public void Calculator_path_healthy_secondary_publishes_240()
|
|
{
|
|
var publisher = new RecordingPublisher();
|
|
var local = NodeId.Parse("secondary-node");
|
|
var probe = Sys.ActorOf(Akka.Actor.Props.Create(() =>
|
|
new StubDbHealth(new DbHealthProbeActor.DbHealthStatus(true, DateTime.UtcNow, null))));
|
|
var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(
|
|
serviceLevel: publisher, localNode: local, dbHealthProbe: probe));
|
|
|
|
actor.Tell(new DbHealthProbeActor.DbHealthStatus(true, DateTime.UtcNow, null));
|
|
actor.Tell(new RedundancyStateChanged(
|
|
Nodes: new[]
|
|
{
|
|
new NodeRedundancyState(local, RedundancyRole.Secondary,
|
|
IsClusterLeader: false, IsDriverPrimary: false, DateTime.UtcNow),
|
|
},
|
|
CorrelationId.NewId()));
|
|
|
|
AwaitAssert(() => publisher.Levels.ShouldContain((byte)240),
|
|
duration: TimeSpan.FromMilliseconds(500));
|
|
}
|
|
|
|
/// <summary>Verifies that the calculator path computes 100 when the DB is unreachable
|
|
/// ((false,_,true) basis, stale via !DbReachable) for a non-leader secondary (no bonus).</summary>
|
|
[Fact]
|
|
public void Calculator_path_db_unreachable_publishes_100()
|
|
{
|
|
var publisher = new RecordingPublisher();
|
|
var local = NodeId.Parse("secondary-node");
|
|
var probe = Sys.ActorOf(Akka.Actor.Props.Create(() =>
|
|
new StubDbHealth(new DbHealthProbeActor.DbHealthStatus(false, DateTime.UtcNow, "down"))));
|
|
var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(
|
|
serviceLevel: publisher, localNode: local, dbHealthProbe: probe));
|
|
|
|
actor.Tell(new DbHealthProbeActor.DbHealthStatus(false, DateTime.UtcNow, "down"));
|
|
actor.Tell(new RedundancyStateChanged(
|
|
Nodes: new[]
|
|
{
|
|
new NodeRedundancyState(local, RedundancyRole.Secondary,
|
|
IsClusterLeader: false, IsDriverPrimary: false, DateTime.UtcNow),
|
|
},
|
|
CorrelationId.NewId()));
|
|
|
|
AwaitAssert(() => publisher.Levels.ShouldContain((byte)100),
|
|
duration: TimeSpan.FromMilliseconds(500));
|
|
}
|
|
|
|
/// <summary>Verifies that the calculator path computes 200 for a stale snapshot when the DB is
|
|
/// reachable + fresh but the redundancy entry's AsOfUtc is older than the stale window
|
|
/// ((true,_,true) basis) for a non-leader secondary (no bonus).</summary>
|
|
[Fact]
|
|
public void Calculator_path_stale_snapshot_publishes_200()
|
|
{
|
|
var publisher = new RecordingPublisher();
|
|
var local = NodeId.Parse("secondary-node");
|
|
var probe = Sys.ActorOf(Akka.Actor.Props.Create(() =>
|
|
new StubDbHealth(new DbHealthProbeActor.DbHealthStatus(true, DateTime.UtcNow, null))));
|
|
var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(
|
|
serviceLevel: publisher, localNode: local, dbHealthProbe: probe,
|
|
staleWindow: TimeSpan.FromSeconds(2)));
|
|
|
|
actor.Tell(new DbHealthProbeActor.DbHealthStatus(true, DateTime.UtcNow, null));
|
|
actor.Tell(new RedundancyStateChanged(
|
|
Nodes: new[]
|
|
{
|
|
new NodeRedundancyState(local, RedundancyRole.Secondary,
|
|
IsClusterLeader: false, IsDriverPrimary: false,
|
|
DateTime.UtcNow - TimeSpan.FromMinutes(1)),
|
|
},
|
|
CorrelationId.NewId()));
|
|
|
|
AwaitAssert(() => publisher.Levels.ShouldContain((byte)200),
|
|
duration: TimeSpan.FromMilliseconds(500));
|
|
}
|
|
|
|
/// <summary>Verifies that a detached local node publishes 0 (the calculator does not model
|
|
/// Detached, so the handler guards it before the calculator path). The node first goes healthy
|
|
/// (250) so the dedup'd transition down to 0 is observable.</summary>
|
|
[Fact]
|
|
public void Calculator_path_detached_publishes_0()
|
|
{
|
|
var publisher = new RecordingPublisher();
|
|
var local = NodeId.Parse("detached-node");
|
|
var probe = Sys.ActorOf(Akka.Actor.Props.Create(() =>
|
|
new StubDbHealth(new DbHealthProbeActor.DbHealthStatus(true, DateTime.UtcNow, null))));
|
|
var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(
|
|
serviceLevel: publisher, localNode: local, dbHealthProbe: probe));
|
|
|
|
actor.Tell(new DbHealthProbeActor.DbHealthStatus(true, DateTime.UtcNow, null));
|
|
// First go healthy primary-leader (250) so the transition down to 0 is not dedup'd against
|
|
// the initial 0.
|
|
actor.Tell(new RedundancyStateChanged(
|
|
Nodes: new[]
|
|
{
|
|
new NodeRedundancyState(local, RedundancyRole.Primary,
|
|
IsClusterLeader: true, IsDriverPrimary: true, DateTime.UtcNow),
|
|
},
|
|
CorrelationId.NewId()));
|
|
AwaitAssert(() => publisher.Levels.ShouldContain((byte)250),
|
|
duration: TimeSpan.FromMilliseconds(500));
|
|
|
|
// Now detach — expect the guard to drive ServiceLevel down to 0.
|
|
actor.Tell(new RedundancyStateChanged(
|
|
Nodes: new[]
|
|
{
|
|
new NodeRedundancyState(local, RedundancyRole.Detached,
|
|
IsClusterLeader: false, IsDriverPrimary: false, DateTime.UtcNow),
|
|
},
|
|
CorrelationId.NewId()));
|
|
|
|
AwaitAssert(() => publisher.Levels.ShouldContain((byte)0),
|
|
duration: TimeSpan.FromMilliseconds(500));
|
|
}
|
|
|
|
/// <summary>Verifies that an actively-observed, recent peer probe of MY endpoint that came back
|
|
/// <c>Ok==false</c> demotes the calculator basis to 0. A non-leader Secondary entry (no +10 bonus)
|
|
/// is used so the assertion is unambiguous: inputs (DbReachable=true, OpcUaProbeOk=false,
|
|
/// Stale=false) match Compute's <c>_ => 0</c> arm → 0, +0 (non-leader) → 0.</summary>
|
|
[Fact]
|
|
public void Probe_false_about_me_with_healthy_db_publishes_0()
|
|
{
|
|
var publisher = new RecordingPublisher();
|
|
var local = NodeId.Parse("secondary-node");
|
|
var probe = Sys.ActorOf(Akka.Actor.Props.Create(() =>
|
|
new StubDbHealth(new DbHealthProbeActor.DbHealthStatus(true, DateTime.UtcNow, null))));
|
|
var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(
|
|
serviceLevel: publisher, localNode: local, dbHealthProbe: probe,
|
|
staleWindow: TimeSpan.FromSeconds(30), probeFreshnessWindow: TimeSpan.FromSeconds(30)));
|
|
|
|
actor.Tell(new DbHealthProbeActor.DbHealthStatus(true, DateTime.UtcNow, null));
|
|
actor.Tell(new PeerOpcUaProbeActor.OpcUaProbeResult(local, Ok: false));
|
|
actor.Tell(new RedundancyStateChanged(
|
|
Nodes: new[]
|
|
{
|
|
new NodeRedundancyState(local, RedundancyRole.Secondary,
|
|
IsClusterLeader: false, IsDriverPrimary: false, DateTime.UtcNow),
|
|
},
|
|
CorrelationId.NewId()));
|
|
|
|
AwaitAssert(() => publisher.Levels.ShouldContain((byte)0),
|
|
duration: TimeSpan.FromMilliseconds(500));
|
|
}
|
|
|
|
/// <summary>Verifies branch (3) of <c>OpcUaProbeOk()</c>: a peer's NEGATIVE verdict about this node
|
|
/// that has aged past <c>_probeFreshnessWindow</c> is given the benefit of the doubt (the peer's
|
|
/// verdict aged out → don't demote), so the node still publishes the HEALTHY level. With a 1ms
|
|
/// freshness window and a 30ms wait before the recompute, the cached <c>Ok==false</c> verdict is
|
|
/// reliably stale by the time the level is computed, so <c>OpcUaProbeOk()</c> returns true and a
|
|
/// healthy primary-leader computes inputs (true, true, false) → 240, +10 leader → 250 — NOT the 0
|
|
/// that a still-fresh negative verdict would produce (cf. <see cref="Probe_false_about_me_with_healthy_db_publishes_0"/>).</summary>
|
|
[Fact]
|
|
public void Stale_probe_verdict_is_not_demoted_publishes_healthy()
|
|
{
|
|
var publisher = new RecordingPublisher();
|
|
var local = NodeId.Parse("primary-node");
|
|
var probe = Sys.ActorOf(Akka.Actor.Props.Create(() =>
|
|
new StubDbHealth(new DbHealthProbeActor.DbHealthStatus(true, DateTime.UtcNow, null))));
|
|
var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(
|
|
serviceLevel: publisher, localNode: local, dbHealthProbe: probe,
|
|
staleWindow: TimeSpan.FromSeconds(30),
|
|
probeFreshnessWindow: TimeSpan.FromMilliseconds(1)));
|
|
|
|
// Seed a healthy DB sample and a NEGATIVE peer verdict about me. (No RedundancyStateChanged yet,
|
|
// so no level is computed from these — the verdict just gets stamped with the receive time.)
|
|
actor.Tell(new DbHealthProbeActor.DbHealthStatus(true, DateTime.UtcNow, null));
|
|
actor.Tell(new PeerOpcUaProbeActor.OpcUaProbeResult(local, Ok: false));
|
|
|
|
// Let the verdict age WELL past the 1ms freshness window before any level is computed. The 30ms
|
|
// wait is >> 1ms, so the verdict is reliably stale by the time OpcUaProbeOk() runs — no reliance
|
|
// on sub-ms clock ties.
|
|
ExpectNoMsg(TimeSpan.FromMilliseconds(30));
|
|
|
|
// Now trigger the recompute with a fresh healthy primary-leader snapshot. By now _probeAboutMe is
|
|
// >1ms old, so branch (3) ignores the negative verdict and OpcUaProbeOk() returns true.
|
|
actor.Tell(new RedundancyStateChanged(
|
|
Nodes: new[]
|
|
{
|
|
new NodeRedundancyState(local, RedundancyRole.Primary,
|
|
IsClusterLeader: true, IsDriverPrimary: true, DateTime.UtcNow),
|
|
},
|
|
CorrelationId.NewId()));
|
|
|
|
// Healthy (250), NOT 0 — proves an aged negative verdict does not demote.
|
|
AwaitAssert(() => publisher.Levels.ShouldContain((byte)250),
|
|
duration: TimeSpan.FromMilliseconds(500));
|
|
}
|
|
|
|
/// <summary>Verifies that with no peer probe result ever received, <c>OpcUaProbeOk()</c> defaults
|
|
/// to <c>true</c> (benefit of the doubt / single-node). A healthy primary-leader thus computes
|
|
/// inputs (true, true, false) → 240, +10 leader → 250.</summary>
|
|
[Fact]
|
|
public void No_probe_result_defaults_ok_true_publishes_250()
|
|
{
|
|
var publisher = new RecordingPublisher();
|
|
var local = NodeId.Parse("primary-node");
|
|
var probe = Sys.ActorOf(Akka.Actor.Props.Create(() =>
|
|
new StubDbHealth(new DbHealthProbeActor.DbHealthStatus(true, DateTime.UtcNow, null))));
|
|
var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(
|
|
serviceLevel: publisher, localNode: local, dbHealthProbe: probe,
|
|
staleWindow: TimeSpan.FromSeconds(30), probeFreshnessWindow: TimeSpan.FromSeconds(30)));
|
|
|
|
actor.Tell(new DbHealthProbeActor.DbHealthStatus(true, DateTime.UtcNow, null));
|
|
actor.Tell(new RedundancyStateChanged(
|
|
Nodes: new[]
|
|
{
|
|
new NodeRedundancyState(local, RedundancyRole.Primary,
|
|
IsClusterLeader: true, IsDriverPrimary: true, DateTime.UtcNow),
|
|
},
|
|
CorrelationId.NewId()));
|
|
|
|
AwaitAssert(() => publisher.Levels.ShouldContain((byte)250),
|
|
duration: TimeSpan.FromMilliseconds(500));
|
|
}
|
|
|
|
/// <summary>Verifies that a later <c>Ok==true</c> peer probe supersedes an earlier <c>Ok==false</c>
|
|
/// (recovery). A non-leader Secondary entry is used (no +10 bonus): after the true supersedes,
|
|
/// inputs (true, true, false) → 240, +0 → 240.</summary>
|
|
[Fact]
|
|
public void Probe_true_supersedes_earlier_false_publishes_240()
|
|
{
|
|
var publisher = new RecordingPublisher();
|
|
var local = NodeId.Parse("secondary-node");
|
|
var probe = Sys.ActorOf(Akka.Actor.Props.Create(() =>
|
|
new StubDbHealth(new DbHealthProbeActor.DbHealthStatus(true, DateTime.UtcNow, null))));
|
|
var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(
|
|
serviceLevel: publisher, localNode: local, dbHealthProbe: probe,
|
|
staleWindow: TimeSpan.FromSeconds(30), probeFreshnessWindow: TimeSpan.FromSeconds(30)));
|
|
|
|
actor.Tell(new DbHealthProbeActor.DbHealthStatus(true, DateTime.UtcNow, null));
|
|
actor.Tell(new PeerOpcUaProbeActor.OpcUaProbeResult(local, Ok: false));
|
|
actor.Tell(new PeerOpcUaProbeActor.OpcUaProbeResult(local, Ok: true));
|
|
actor.Tell(new RedundancyStateChanged(
|
|
Nodes: new[]
|
|
{
|
|
new NodeRedundancyState(local, RedundancyRole.Secondary,
|
|
IsClusterLeader: false, IsDriverPrimary: false, DateTime.UtcNow),
|
|
},
|
|
CorrelationId.NewId()));
|
|
|
|
AwaitAssert(() => publisher.Levels.ShouldContain((byte)240),
|
|
duration: TimeSpan.FromMilliseconds(500));
|
|
}
|
|
|
|
/// <summary>Verifies that a peer probe result about a DIFFERENT node is ignored — it does not
|
|
/// affect MY <c>OpcUaProbeOk()</c>, which stays at its default <c>true</c>. A healthy
|
|
/// primary-leader thus still computes (true, true, false) → 240, +10 → 250.</summary>
|
|
[Fact]
|
|
public void Probe_about_a_different_node_is_ignored()
|
|
{
|
|
var publisher = new RecordingPublisher();
|
|
var local = NodeId.Parse("primary-node");
|
|
var probe = Sys.ActorOf(Akka.Actor.Props.Create(() =>
|
|
new StubDbHealth(new DbHealthProbeActor.DbHealthStatus(true, DateTime.UtcNow, null))));
|
|
var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(
|
|
serviceLevel: publisher, localNode: local, dbHealthProbe: probe,
|
|
staleWindow: TimeSpan.FromSeconds(30), probeFreshnessWindow: TimeSpan.FromSeconds(30)));
|
|
|
|
actor.Tell(new DbHealthProbeActor.DbHealthStatus(true, DateTime.UtcNow, null));
|
|
actor.Tell(new PeerOpcUaProbeActor.OpcUaProbeResult(NodeId.Parse("someone-else"), Ok: false));
|
|
actor.Tell(new RedundancyStateChanged(
|
|
Nodes: new[]
|
|
{
|
|
new NodeRedundancyState(local, RedundancyRole.Primary,
|
|
IsClusterLeader: true, IsDriverPrimary: true, DateTime.UtcNow),
|
|
},
|
|
CorrelationId.NewId()));
|
|
|
|
AwaitAssert(() => publisher.Levels.ShouldContain((byte)250),
|
|
duration: TimeSpan.FromMilliseconds(500));
|
|
}
|
|
|
|
/// <summary>Verifies the legacy back-compat seam: with no DB-health probe wired, the handler
|
|
/// falls back to the old role-only switch (Primary + leader → 240).</summary>
|
|
[Fact]
|
|
public void Legacy_path_no_db_probe_keeps_role_only()
|
|
{
|
|
var publisher = new RecordingPublisher();
|
|
var local = NodeId.Parse("primary-node");
|
|
var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(
|
|
serviceLevel: publisher, localNode: local));
|
|
|
|
actor.Tell(new RedundancyStateChanged(
|
|
Nodes: new[]
|
|
{
|
|
new NodeRedundancyState(local, RedundancyRole.Primary,
|
|
IsClusterLeader: true, IsDriverPrimary: true, DateTime.UtcNow),
|
|
},
|
|
CorrelationId.NewId()));
|
|
|
|
AwaitAssert(() => publisher.Levels.ShouldContain((byte)240),
|
|
duration: TimeSpan.FromMilliseconds(500));
|
|
}
|
|
|
|
/// <summary>Verifies that the periodic <c>HealthTick</c> Asks the local <see cref="DbHealthProbeActor"/>
|
|
/// for status and pipes the reply back to itself — populating <c>_lastDbHealth</c> in production WITHOUT
|
|
/// any external pump (no direct <see cref="DbHealthProbeActor.DbHealthStatus"/> Tell). With a healthy DB
|
|
/// reply and a primary-leader snapshot, the calculator path therefore reaches 250.</summary>
|
|
[Fact]
|
|
public void HealthTick_asks_db_probe_and_publishes_calculator_byte()
|
|
{
|
|
var publisher = new RecordingPublisher();
|
|
var local = NodeId.Parse("primary-node");
|
|
var db = Sys.ActorOf(Akka.Actor.Props.Create(() => new StubDbHealth(
|
|
new DbHealthProbeActor.DbHealthStatus(true, DateTime.UtcNow, null))));
|
|
var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(
|
|
serviceLevel: publisher, localNode: local, dbHealthProbe: db,
|
|
staleWindow: TimeSpan.FromSeconds(30), probeFreshnessWindow: TimeSpan.FromSeconds(30),
|
|
healthTickInterval: TimeSpan.FromMilliseconds(100)));
|
|
actor.Tell(new RedundancyStateChanged(new[]
|
|
{
|
|
new NodeRedundancyState(local, RedundancyRole.Primary,
|
|
IsClusterLeader: true, IsDriverPrimary: true, DateTime.UtcNow),
|
|
}, CorrelationId.NewId()));
|
|
// No direct DbHealthStatus Tell — the periodic HealthTick Ask must populate it → 250.
|
|
AwaitAssert(() => publisher.Levels.ShouldContain((byte)250), TimeSpan.FromSeconds(2));
|
|
}
|
|
|
|
/// <summary>Stub DB-health probe actor that answers <see cref="DbHealthProbeActor.GetStatus"/>
|
|
/// with a fixed status, so the calculator path is deterministic without a real timer.</summary>
|
|
private sealed class StubDbHealth : Akka.Actor.ReceiveActor
|
|
{
|
|
/// <summary>Initializes a new instance of the <see cref="StubDbHealth"/> class.</summary>
|
|
/// <param name="status">The fixed DB-health status to reply with.</param>
|
|
public StubDbHealth(DbHealthProbeActor.DbHealthStatus status) =>
|
|
Receive<DbHealthProbeActor.GetStatus>(_ => Sender.Tell(status));
|
|
}
|
|
|
|
/// <summary>Test implementation of IOpcUaAddressSpaceSink that records calls.</summary>
|
|
private sealed class RecordingSink : IOpcUaAddressSpaceSink
|
|
{
|
|
/// <summary>Gets the queue of recorded value updates.</summary>
|
|
public ConcurrentQueue<(string NodeId, object? Value, OpcUaQuality Quality, DateTime Ts)> ValueQueue { get; } = new();
|
|
/// <summary>Gets the queue of recorded alarm condition updates.</summary>
|
|
public ConcurrentQueue<(string AlarmNodeId, AlarmConditionSnapshot State, DateTime Ts)> AlarmQueue { get; } = new();
|
|
/// <summary>Gets the queue of recorded alarm-quality annotations (#477).</summary>
|
|
public ConcurrentQueue<(string AlarmNodeId, OpcUaQuality Quality, DateTime Ts, AddressSpaceRealm Realm)> AlarmQualityQueue { get; } = new();
|
|
/// <summary>Count of rebuild calls.</summary>
|
|
public int RebuildCalls;
|
|
/// <summary>Gets the queue of recorded EnsureFolder calls.</summary>
|
|
public ConcurrentQueue<(string NodeId, string? ParentNodeId, string DisplayName)> FolderQueue { get; } = new();
|
|
/// <summary>Gets the queue of recorded EnsureVariable calls.</summary>
|
|
public ConcurrentQueue<(string NodeId, string? ParentNodeId, string DisplayName, string DataType, bool Writable)> VariableQueue { get; } = new();
|
|
/// <summary>Gets the queue of recorded RaiseNodesAddedModelChange announcements.</summary>
|
|
public ConcurrentQueue<string> ModelChangeQueue { get; } = new();
|
|
|
|
/// <summary>Gets the list of recorded value updates.</summary>
|
|
public List<(string NodeId, object? Value, OpcUaQuality Quality, DateTime Ts)> Values =>
|
|
ValueQueue.ToList();
|
|
/// <summary>Gets the list of recorded alarm condition updates.</summary>
|
|
public List<(string AlarmNodeId, AlarmConditionSnapshot State, DateTime Ts)> Alarms =>
|
|
AlarmQueue.ToList();
|
|
/// <summary>Gets the list of recorded EnsureFolder calls.</summary>
|
|
public List<(string NodeId, string? ParentNodeId, string DisplayName)> Folders =>
|
|
FolderQueue.ToList();
|
|
/// <summary>Gets the list of recorded EnsureVariable calls.</summary>
|
|
public List<(string NodeId, string? ParentNodeId, string DisplayName, string DataType, bool Writable)> Variables =>
|
|
VariableQueue.ToList();
|
|
/// <summary>Gets the list of recorded RaiseNodesAddedModelChange announcements.</summary>
|
|
public List<string> ModelChanges => ModelChangeQueue.ToList();
|
|
|
|
/// <summary>Records a value update.</summary>
|
|
/// <param name="nodeId">The OPC UA node identifier.</param>
|
|
/// <param name="value">The attribute value.</param>
|
|
/// <param name="quality">The OPC UA quality code.</param>
|
|
/// <param name="ts">The timestamp of the update.</param>
|
|
public void WriteValue(string nodeId, object? value, OpcUaQuality quality, DateTime ts, AddressSpaceRealm realm = AddressSpaceRealm.Uns) =>
|
|
ValueQueue.Enqueue((nodeId, value, quality, ts));
|
|
|
|
/// <summary>Records an alarm condition update.</summary>
|
|
/// <param name="alarmNodeId">The OPC UA alarm node identifier.</param>
|
|
/// <param name="state">The full condition state snapshot.</param>
|
|
/// <param name="ts">The timestamp of the update.</param>
|
|
public void WriteAlarmCondition(string alarmNodeId, AlarmConditionSnapshot state, DateTime ts, AddressSpaceRealm realm = AddressSpaceRealm.Uns) =>
|
|
AlarmQueue.Enqueue((alarmNodeId, state, ts));
|
|
|
|
/// <summary>Records an alarm-quality annotation (#477).</summary>
|
|
public void WriteAlarmQuality(string alarmNodeId, OpcUaQuality quality, DateTime ts, AddressSpaceRealm realm = AddressSpaceRealm.Uns) =>
|
|
AlarmQualityQueue.Enqueue((alarmNodeId, quality, ts, realm));
|
|
|
|
/// <summary>Materialises an alarm condition (no-op in test).</summary>
|
|
/// <param name="alarmNodeId">The alarm node ID.</param>
|
|
/// <param name="equipmentNodeId">The equipment folder node ID.</param>
|
|
/// <param name="displayName">The condition display name.</param>
|
|
/// <param name="alarmType">The domain alarm type.</param>
|
|
/// <param name="severity">The domain severity.</param>
|
|
public void MaterialiseAlarmCondition(string alarmNodeId, string equipmentNodeId, string displayName, string alarmType, int severity, AddressSpaceRealm realm, bool isNative = false) { }
|
|
|
|
/// <summary>Records a folder ensure call.</summary>
|
|
/// <param name="folderNodeId">The OPC UA folder node identifier.</param>
|
|
/// <param name="parentNodeId">The parent folder node identifier, or null for root.</param>
|
|
/// <param name="displayName">The display name of the folder.</param>
|
|
public void EnsureFolder(string folderNodeId, string? parentNodeId, string displayName, AddressSpaceRealm realm = AddressSpaceRealm.Uns) =>
|
|
FolderQueue.Enqueue((folderNodeId, parentNodeId, displayName));
|
|
|
|
/// <summary>Records a variable ensure call.</summary>
|
|
/// <param name="variableNodeId">The OPC UA variable node identifier.</param>
|
|
/// <param name="parentFolderNodeId">The parent folder node identifier, or null for root.</param>
|
|
/// <param name="displayName">The display name of the variable.</param>
|
|
/// <param name="dataType">The OPC UA built-in type name.</param>
|
|
/// <param name="writable">Whether the node is created read/write.</param>
|
|
/// <param name="historianTagname">The resolved historian tagname (null ⇒ not historized).</param>
|
|
public void EnsureVariable(string variableNodeId, string? parentFolderNodeId, string displayName, string dataType, bool writable, AddressSpaceRealm realm, string? historianTagname = null, bool isArray = false, uint? arrayLength = null) =>
|
|
VariableQueue.Enqueue((variableNodeId, parentFolderNodeId, displayName, dataType, writable));
|
|
|
|
/// <summary>Records a rebuild call.</summary>
|
|
public void RebuildAddressSpace() => Interlocked.Increment(ref RebuildCalls);
|
|
|
|
/// <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) => ModelChangeQueue.Enqueue(affectedNodeId);
|
|
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") { }
|
|
}
|
|
|
|
/// <summary>Test implementation of IServiceLevelPublisher that records publishes.</summary>
|
|
private sealed class RecordingPublisher : IServiceLevelPublisher
|
|
{
|
|
private readonly ConcurrentQueue<byte> _q = new();
|
|
/// <summary>Gets the recorded service levels.</summary>
|
|
public byte[] Levels => _q.ToArray();
|
|
/// <summary>Records a service level publish.</summary>
|
|
/// <param name="serviceLevel">The service level value to publish.</param>
|
|
public void Publish(byte serviceLevel) => _q.Enqueue(serviceLevel);
|
|
}
|
|
}
|