dohertj2/lmxopcua
1
0
Fork 0
Code Issues 5 Pull Requests Actions Packages Projects Releases Wiki Activity

test(717): deferred Warm coverage - 2-node handover, DataEncoding, deleted/reused items, clone matrix, ACL stand-down, per-counter metrics
v2-ci / unit-tests (push) Has been cancelled
Details

Browse Source 
- T3 WarmSingletonHandoverTests: real 2-node cluster, singleton under ClusterSingletonManager on both nodes; the old
  incarnation's held-back publish (older revision, higher sequence) reaches a real Warm OpcUaPublishActor after the
  handover and never demotes the new Primary. Pin: a removed node cannot rejoin from its old ActorSystem; a new one
  rejoins younger (higher up-number).
- T6 DataEncoding: Binary/XML encoded reads of a structured tag projected on the Standby; scalar encoding error kept;
  a DefaultXml item is projected, then promoted through its own encoding (TypeId i=888).
- T7 item deleted between queue and publish; ClientHandle reused by i=2267 on a payload recycled through the SDK 2.0.1
  pool (precondition asserted); WarmStandbyCloneMatrixTests over DataChange+DiagnosticInfos/EventNotificationList/
  StatusChange, clone/cached immutability, encode round-trip, real-pool stale marker.
- T8 ACL-stood-down item receives nothing from demotion or promotion; restored grant reaches it again.
- T10 per-counter MeterListener tests (by instrument instance): warm_transition, warm_role gauge, snapshot_rejected
  (publish actor both reasons + DriverHostActor's ordering gate, first actor test of it), projected_notifications,
  suppressed_alarm_events. Process-wide state tests run in non-parallel collections.
- Every test mutation-checked (26 mutations killed); no product bug found. Plan status updated.

Claude-Session: https://claude.ai/code/session_01GT67ACmpMR7b9LGRPLijMi
This commit is contained in:
Joseph Doherty
2026-11-13 04:31:11 -05:00
parent 0c19f26c81
commit 687256d68e
7 changed files with 1152 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/plans/2026-11-11-warm-redundancy-717-plan.md
+59 -5
View File
@@ -307,9 +307,63 @@ Test homes: `WarmStandbyUnitTests` (OpcUaServer.Tests: marker, rewrite, decision
option rules), `WarmRedundancyWireTests` (OpcUaServer.IntegrationTests: metadata per mode, read projection, write and
Part 9 refusal, the three T0 demotion timings, live-item queue, alarm suppression/refresh, effective-URI XML load),
`WarmRedundancyRevisionTests` + `WarmServiceLevelBandTests` (ControlPlane.Tests), `WarmRoleAuthorityTests` +
`OpcUaPublishActorWarmTests` (Runtime.Tests), `OpcUaRedundancyValidationTests` (Host.IntegrationTests).
`OpcUaPublishActorWarmTests` (Runtime.Tests), `OpcUaRedundancyValidationTests` (Host.IntegrationTests). The deferred
depth below adds `WarmSingletonHandoverTests` (ControlPlane.Tests), `WarmRedundancyMetricsTests` (Runtime.Tests),
`WarmStandbyCloneMatrixTests` (OpcUaServer.Tests), `WarmRedundancyWireTests.Deferred.cs` and `WarmRedundancyPooledWireTests`
(OpcUaServer.IntegrationTests).
Not yet covered by a test (left for T13 or a follow-up): the real 2-node singleton-handover test and the
"removed node cannot rejoin" pin (T3; ordering is covered at gate and actor level), a DataEncoding read, a
deleted/reused monitored item between queue and publish and the ExtensionObject/DiagnosticInfo clone matrix (T7), the
ACL-stood-down item on transition (T8), and per-counter MeterListener tests beyond `mode_mismatch` (T10).
**Deferred coverage added (2026-11-13, in-process, no docker).** Each test was mutation-checked against a deliberately
broken implementation and failed. No product bug was found.
- **T3:** `WarmSingletonHandoverTests` (ControlPlane.Tests). A real 2-node cluster runs the redundancy singleton under
`ClusterSingletonManager` on both nodes. The old incarnation's last publish is held back through the broadcast seam,
its node is downed, and the singleton hands over. The held publish is then delivered to a real Warm
`OpcUaPublishActor` on the survivor. That publish is the older revision but has the higher sequence. The survivor
stays Active. The second test pins that a removed node cannot rejoin from its old ActorSystem. A new ActorSystem on
the same address rejoins as a younger member, with a higher up-number. Mutations killed:
- the gate accepts older revisions;
- the revision drops the up-number and incarnation (sequence only);
- for the pin, skipping the Down (a harness negative control).
- **T6:** `A_DataEncoding_read_and_item_are_projected_…` (wire). On the Standby, a Binary or XML DataEncoding read of
a structured (EUInformation) tag is projected. An encoding a scalar cannot take still returns the base read's own
`BadDataEncodingUnsupported`. An item created with `DefaultXml` gets the projected value on the Standby, and on
promotion it receives the value read through its own encoding (TypeId `i=888`). Mutations killed:
- DataEncoding reads are not projected;
- the promotion queue ignores the item's DataEncoding.
- **T7:**
- `An_item_deleted_between_queue_and_publish_…` (wire). The deleted item's queued samples never surface. Its
sibling is still first-delivered projected, and a Republish gives the same result.
- `WarmRedundancyPooledWireTests.A_client_handle_reused_after_delete_…` (wire, non-parallel collection). A marked
payload goes back through SDK 2.0.1's process-wide payload pool, and its ClientHandle is reused by an `i=2267`
item. That item stays Good. The test first asserts that the pool really hands out a marked payload.
- `WarmStandbyCloneMatrixTests` (unit). One delivered clone carries two DataChangeNotifications (one with
DiagnosticInfos), an EventNotificationList and a StatusChangeNotification. Only marked values are rewritten.
Diagnostics, events and status pass through in order. Neither the clone nor the cached message is mutated, and
the result encodes. Two clones project to identical bytes. A recycled payload's stale marker, rented through the
real pool with the same handle, never validates.
- Mutations killed:
- the marker is not validated (wire and unit);
- the marker is not set;
- DiagnosticInfos are dropped;
- non-data payloads are dropped;
- the rewrite happens in place;
- only the first DataChangeNotification is rewritten.
- **T8:** `An_ACL_stood_down_item_receives_nothing_…` (wire). An item that `RevalidateMonitoredItems` stood down gets
nothing from a demotion or a promotion (`QueueCurrentToLiveItems` returns 1, not 2). Its permitted sibling gets
both. Once the grant is restored, the next demotion reaches the item. Mutation killed: stood-down items are not
skipped.
- **T10:**
- `WarmRedundancyMetricsTests` (Runtime.Tests, non-parallel collection) checks four things:
- `warm_transition`, with direction and reason (`agreed` / `snapshot`), and no count on a same-role snapshot;
- the `warm_role` gauge, -1, then 0, then 1, then 0;
- `snapshot_rejected` at the publish actor for `older_revision` and `unrevisioned`;
- at `DriverHostActor`, a stale snapshot is dropped under Warm and counted, and its role view does not move. The
Unset control does move. This is the first actor test of the driver host's ordering gate.
- `WarmRedundancyPooledWireTests` checks `standby.projected_notifications`: it equals the rewritten count, counts
again on Republish, and stays 0 while Active. It also checks `standby.suppressed_alarm_events`: one per withheld
transition, none once Active.
- Every listener selects by instrument instance.
- Mutations killed: each counter's `Add` was removed, the direction tag inverted, the gauge never set, and the
driver-host gate ignored.
Still open (live-only, T13): the durable restore, TransferSubscriptions under one identity, and the rig legs.
tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/WarmSingletonHandoverTests.cs
+222
View File
@@ -0,0 +1,222 @@
using System.Collections.Concurrent;
using Akka.Actor;
using Akka.Cluster;
using Akka.Cluster.Tools.Singleton;
using Akka.Configuration;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Redundancy;
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
using ZB.MOM.WW.OtOpcUa.ControlPlane.Redundancy;
using ZB.MOM.WW.OtOpcUa.Runtime.OpcUa;
using CommonsNodeId = ZB.MOM.WW.OtOpcUa.Commons.Types.NodeId;
namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Tests;
/// <summary>
/// Gitea #717 spec §4.2 / plan T3 on a REAL two-node cluster (pattern: <see cref="RedundancyPrimaryElectionTests"/>):
/// the redundancy singleton runs under a real <see cref="ClusterSingletonManager"/> on both nodes, the old
/// incarnation's last publish is held back through the broadcast seam while the singleton hands over, and the held
/// publish is delivered to a real Warm <see cref="OpcUaPublishActor"/> on the surviving node AFTER the new
/// incarnation's. The subscriber must keep the new Primary. Then the ordering assumption the revision rests on is
/// pinned: a removed node cannot come back with its old up-number — its ActorSystem's cluster is gone for good, and
/// only a NEW ActorSystem rejoins, as a younger member.
/// </summary>
public sealed class WarmSingletonHandoverTests : IAsyncLifetime
{
// Fixed ports: the rejoin control re-binds the first node's port with a new ActorSystem.
private const int FirstPort = 19_570; // seed: forms the cluster, oldest, hosts the first singleton incarnation
private const int SecondPort = 19_571; // joins second: the survivor
private const string SystemName = "warm-handover";
private static readonly TimeSpan Budget = TimeSpan.FromSeconds(30);
private ActorSystem? _first;
private ActorSystem? _second;
private ActorSystem? _rejoined;
private readonly ConcurrentQueue<(int Port, RedundancyStateChanged Msg)> _published = new();
private readonly ConcurrentQueue<string> _log = new();
private volatile bool _holdFirst;
private IActorRef? _subscriber;
private static Config NodeConfig(int port, bool seedOnFirst = true) => ConfigurationFactory.ParseString($$"""
akka {
loglevel = "WARNING"
actor.provider = "Akka.Cluster.ClusterActorRefProvider, Akka.Cluster"
remote.dot-netty.tcp {
hostname = "127.0.0.1"
port = {{port}}
}
cluster {
seed-nodes = [{{(seedOnFirst ? $"\"akka.tcp://{SystemName}@127.0.0.1:{FirstPort}\"" : string.Empty)}}]
roles = ["admin", "driver"]
min-nr-of-members = 1
run-coordinated-shutdown-when-down = off
downing-provider-class = ""
}
}
""");
public async ValueTask InitializeAsync()
{
_first = ActorSystem.Create(SystemName, NodeConfig(FirstPort));
await WaitForAsync(() => UpCount(_first) >= 1, "the first node Up");
_second = ActorSystem.Create(SystemName, NodeConfig(SecondPort));
await WaitForAsync(() => UpCount(_first) == 2 && UpCount(_second) == 2, "both nodes Up on both");
}
public async ValueTask DisposeAsync()
{
foreach (var system in new[] { _rejoined, _second, _first })
if (system is not null) await system.Terminate();
}
private static int UpCount(ActorSystem system) =>
Akka.Cluster.Cluster.Get(system).State.Members.Count(m => m.Status == MemberStatus.Up);
private static async Task WaitForAsync(Func<bool> condition, string what)
{
var deadline = DateTime.UtcNow + Budget;
while (DateTime.UtcNow < deadline)
{
if (condition()) return;
await Task.Delay(50);
}
throw new TimeoutException($"timed out waiting for {what}");
}
private static CommonsNodeId NodeOf(int port) => CommonsNodeId.Parse($"127.0.0.1:{port}");
private static CommonsNodeId? PrimaryOf(RedundancyStateChanged msg) =>
msg.Nodes.FirstOrDefault(n => n.Role == RedundancyRole.Primary)?.NodeId;
/// <summary>The singleton's broadcast seam on each node: record, and relay to the subscriber unless held back.</summary>
private void Broadcast(int port, object msg)
{
if (msg is not RedundancyStateChanged changed) return;
_published.Enqueue((port, changed));
if (port == FirstPort && _holdFirst) return;
_subscriber?.Tell(changed);
}
private void StartSingleton(ActorSystem system, int port) =>
system.ActorOf(ClusterSingletonManager.Props(
RedundancyStateActor.Props(msg => Broadcast(port, msg), TimeSpan.FromMilliseconds(100)),
PoisonPill.Instance,
ClusterSingletonManagerSettings.Create(system).WithRole("driver")),
"redundancy-state");
private RedundancyStateChanged[] PublishedBy(int port) => _published.Where(p => p.Port == port).Select(p => p.Msg).ToArray();
[Fact]
public async Task A_held_back_publish_of_the_old_incarnation_never_demotes_the_new_Primary_after_handover()
{
// The subscriber: a real Warm OpcUaPublishActor on the node that will survive.
_subscriber = _second!.ActorOf(OpcUaPublishActor.PropsForTests(
serviceLevel: new LogPublisher(_log), localNode: NodeOf(SecondPort), dbLess: true,
healthTickInterval: TimeSpan.FromMinutes(5), redundancy: new WarmRedundancySettings(OpcUaRedundancyMode.Warm),
standbyProjection: new LogProjection(_log)));
StartSingleton(_first!, FirstPort);
StartSingleton(_second!, SecondPort);
// 1. The first incarnation (on the oldest node) names it Primary; the survivor is Standby.
await WaitForAsync(() => PublishedBy(FirstPort).Any(m => PrimaryOf(m) == NodeOf(FirstPort)), "the first incarnation's publish");
await WaitForAsync(() => _log.Contains("SL:90"), "the survivor publishing its Standby level");
PublishedBy(SecondPort).ShouldBeEmpty("one singleton incarnation at a time");
_log.ShouldNotContain("SL:250");
// 2. Let the old incarnation run well ahead on sequence (heartbeats), then hold back its publishes and keep the
// last one.
await WaitForAsync(() => PublishedBy(FirstPort).Length >= 40, "the old incarnation's heartbeats");
_holdFirst = true;
var heldFrom = PublishedBy(FirstPort).Length;
await WaitForAsync(() => PublishedBy(FirstPort).Length > heldFrom, "a held-back heartbeat of the old incarnation");
// 3. The oldest node is downed (a crash, seen from the survivor); the singleton hands over to the survivor.
Akka.Cluster.Cluster.Get(_second!).Down(Akka.Cluster.Cluster.Get(_first!).SelfAddress);
await WaitForAsync(() => Akka.Cluster.Cluster.Get(_second!).State.Members.All(m => m.Address.Port != FirstPort),
"the downed node removed from the survivor's view");
await WaitForAsync(() => PublishedBy(SecondPort).Any(m => PrimaryOf(m) == NodeOf(SecondPort)), "the new incarnation's publish");
await WaitForAsync(() => _log.LastOrDefault() == "SL:250", "the survivor promoting");
var held = PublishedBy(FirstPort).Last();
var newest = PublishedBy(SecondPort).Last();
// Artifact first: the held publish is a real, revisioned, OLDER snapshot that would demote the survivor.
PrimaryOf(held).ShouldBe(NodeOf(FirstPort));
held.Revision.ShouldNotBeNull();
newest.Revision.ShouldNotBeNull();
held.Revision!.PublisherUpNumber.ShouldBeLessThan(newest.Revision!.PublisherUpNumber,
"the old incarnation ran on the older member");
held.Revision.CompareTo(newest.Revision).ShouldBeLessThan(0);
held.Revision.Sequence.ShouldBeGreaterThan(newest.Revision.Sequence,
"the old incarnation published more often, so a sequence-only order would let it win");
// 4. The late publish arrives after the handover — and is dropped. The survivor stays Active.
var before = _log.Count;
_subscriber.Tell(held);
await Task.Delay(1500, TestContext.Current.CancellationToken); // absence window: ~15 heartbeats of the new incarnation are relayed meanwhile
var after = _log.Skip(before).ToArray();
after.ShouldNotContain("delivery:True", "the held publish must not demote the new Primary");
after.ShouldNotContain("SL:90");
_log.Last().ShouldBe("SL:250");
}
/// <summary>
/// The assumption under the up-number order (plan T3): a member the cluster has removed cannot rejoin from the
/// same ActorSystem — its cluster extension has shut down — so an incarnation hosted there can never publish
/// again with its old, lower up-number. A NEW ActorSystem on the same address joins as a new, younger member
/// (positive control) whose up-number exceeds the survivor's.
/// </summary>
[Fact]
public async Task A_removed_node_cannot_rejoin_without_a_new_ActorSystem_and_comes_back_younger()
{
var survivor = Akka.Cluster.Cluster.Get(_second!);
var removed = Akka.Cluster.Cluster.Get(_first!);
var oldUpNumber = RedundancyStateActor.ReadUpNumber(removed.SelfMember)!.Value;
var survivorUpNumber = RedundancyStateActor.ReadUpNumber(survivor.SelfMember)!.Value;
oldUpNumber.ShouldBeLessThan(survivorUpNumber);
survivor.Down(removed.SelfAddress);
await WaitForAsync(() => survivor.State.Members.All(m => m.Address.Port != FirstPort), "the node removed");
await WaitForAsync(() => removed.IsTerminated, "the removed node's cluster extension shutting down");
_first!.WhenTerminated.IsCompleted.ShouldBeFalse("the ActorSystem itself is still alive (run-coordinated-shutdown-when-down = off)");
// The same ActorSystem asks to join again: nothing happens.
try { removed.Join(survivor.SelfAddress); }
catch (Exception) { /* a terminated cluster may refuse outright; either way it must not come back */ }
var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(4);
while (DateTime.UtcNow < deadline)
{
survivor.State.Members.ShouldNotContain(m => m.Address.Port == FirstPort,
"a removed member must never reappear from its old ActorSystem");
await Task.Delay(100, TestContext.Current.CancellationToken);
}
// Positive control: a NEW ActorSystem on the same address does join, as a younger member.
await _first.Terminate();
_first = null;
_rejoined = ActorSystem.Create(SystemName, NodeConfig(FirstPort, seedOnFirst: false));
Akka.Cluster.Cluster.Get(_rejoined).Join(survivor.SelfAddress);
await WaitForAsync(() => survivor.State.Members.Any(m => m.Address.Port == FirstPort && m.Status == MemberStatus.Up),
"the new ActorSystem Up");
var rejoinedMember = survivor.State.Members.Single(m => m.Address.Port == FirstPort);
RedundancyStateActor.ReadUpNumber(rejoinedMember)!.Value.ShouldBeGreaterThan(survivorUpNumber,
"the address comes back only as a younger member, so its publishes outrank nothing of the old incarnation's");
rejoinedMember.UniqueAddress.Uid.ShouldNotBe(removed.SelfUniqueAddress.Uid);
}
private sealed class LogPublisher(ConcurrentQueue<string> log) : IServiceLevelPublisher
{
public void Publish(byte serviceLevel) => log.Enqueue($"SL:{serviceLevel}");
}
private sealed class LogProjection(ConcurrentQueue<string> log) : IStandbyProjection
{
public void SetDataProjection(bool projected) => log.Enqueue($"proj:{projected}");
public void SetStandbyDelivery(bool standby) => log.Enqueue($"delivery:{standby}");
public void SetAlarmSuppression(bool suppressed) => log.Enqueue($"alarms:{suppressed}");
public int QueueCurrentToLiveItems(bool projected) { log.Enqueue($"queue:{projected}"); return 0; }
public int RefreshAllEventItems() { log.Enqueue("refresh"); return 0; }
}
}
tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/WarmRedundancyPooledWireTests.cs
+210
View File
@@ -0,0 +1,210 @@
using System.Collections.Concurrent;
using System.Diagnostics.Metrics;
using Opc.Ua;
using Opc.Ua.Client;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.Observability;
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
using ZB.MOM.WW.OtOpcUa.Configuration.Enums;
using static ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests.ServerEventRig;
using static ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests.WarmRedundancyWireTests;
namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests;
/// <summary>
/// Process-wide state these tests observe — SDK 2.0.1's notification-payload pool and the Warm instruments — is
/// shared by every server and client in the test process, so they run alone.
/// </summary>
[CollectionDefinition(Name, DisableParallelization = true)]
public sealed class WarmRedundancyProcessWideCollection
{
/// <summary>The collection name.</summary>
public const string Name = "warm-redundancy-process-wide";
}
/// <summary>
/// Gitea #717 over the wire, where the observation is process-wide: a monitored-item ClientHandle reused after a
/// delete while the SDK's payload pool hands out notifications that still carry an old ordinary-tag marker (T7),
/// and the two OpcUaServer-side Warm instruments (T10), <c>standby.projected_notifications</c> and
/// <c>standby.suppressed_alarm_events</c>, each selected by instrument INSTANCE (#503).
/// </summary>
[Trait("Category", "Integration")]
[Collection(WarmRedundancyProcessWideCollection.Name)]
public sealed class WarmRedundancyPooledWireTests
{
private const NodePermissions All =
NodePermissions.Browse | NodePermissions.Read | NodePermissions.Subscribe | NodePermissions.AlarmRead
| NodePermissions.AlarmAcknowledge | NodePermissions.AlarmConfirm | NodePermissions.AlarmShelve
| NodePermissions.MethodCall | NodePermissions.WriteOperate;
/// <summary>
/// SDK 2.0.1 returns acknowledged (and evicted) notification payloads to a process-wide pool, and
/// <c>Initialize()</c> clears only ClientHandle and Value, so a payload stamped for an ordinary tag goes back out
/// to the next item that samples — with its old marker. Here an ordinary-tag item's payloads are delivered and
/// acknowledged in one go, the item is deleted, and its ClientHandle is reused by an item on a NON-ordinary node
/// (<c>i=2267</c>): on the Standby every value of that item must stay Good, because the stale marker names
/// another value. The test first proves the pool really does hold a marked payload at that moment, and an
/// ordinary-tag item reusing the handle afterwards is projected (positive control).
/// </summary>
[Fact]
public async Task A_client_handle_reused_after_delete_by_a_non_ordinary_item_is_never_projected_from_a_recycled_payload()
{
var ct = Deadline();
await using var rig = await StartModeAsync("Warm717Reuse", "Warm", ct);
var projection = rig.Server.CreateStandbyProjection()!;
var raw = rig.EnsureRawVariable("reuse", "Double", writable: false);
rig.Sink.WriteValue(raw, 0.0, OpcUaStatus.Good, DateTime.UtcNow, AddressSpaceRealm.Raw);
Promote(projection);
var serviceLevel = new SdkServiceLevelPublisher(rig.Server.CurrentInstance!,
Microsoft.Extensions.Logging.Abstractions.NullLogger<SdkServiceLevelPublisher>.Instance);
const uint handle = 7;
using var session = await rig.ConnectAsync(anonymous: true, ct);
var client = (SessionClient)session;
var sub = (await client.CreateSubscriptionAsync(H(), 50, 300, 10, 0, true, 0, ct)).SubscriptionId;
var tagItem = await AddItemAsync(client, sub, Raw(session, raw), handle, 1, ct);
// Deliver a run of marked notifications WITHOUT acknowledging, then acknowledge them all at once, so a batch of
// marked payloads is back in the pool together (evicted ones are returned too).
var delivered = new List<uint>();
for (var i = 1; i <= 24; i++)
{
rig.Sink.WriteValue(raw, i, OpcUaStatus.Good, DateTime.UtcNow, AddressSpaceRealm.Raw);
var r = await PublishDataAsync(client, ct);
delivered.Add(r.NotificationMessage.SequenceNumber);
}
StatusCode.IsGood((await client.DeleteMonitoredItemsAsync(H(), sub, new[] { tagItem }, ct)).Results[0]).ShouldBeTrue();
await client.PublishAsync(H(), delivered.Select(seq => new SubscriptionAcknowledgement { SubscriptionId = sub, SequenceNumber = seq }).ToArray(), ct);
var levelItem = await AddItemAsync(client, sub, VariableIds.Server_ServiceLevel, handle, 10, ct);
await CollectAsync(client, ct, TimeSpan.FromMilliseconds(500)); // the initial sample, delivered while Active
DemoteDeliveryOnly(projection);
// Artifact before outcome: the pool's next payload really is a recycled ordinary-tag one. Peek and put it back.
var peeked = (MonitoredItemNotification)MonitoredItemNotificationActivator.Instance.CreateInstance();
peeked.Message.ShouldBeOfType<OrdinaryTagMarker>("the precondition: the SDK pool hands out a payload still carrying a marker");
peeked.Reuse();
var levels = new byte[] { 91, 92, 93, 94, 95, 96, 97, 98 };
foreach (var level in levels)
{
serviceLevel.Publish(level);
await Task.Delay(60, ct);
}
var standby = await CollectAsync(client, ct, TimeSpan.FromSeconds(1));
standby.Where(x => x.Handle == handle).Select(x => Convert.ToByte(x.Value.ClrValue())).ShouldBe(levels,
"every i=2267 change reaches the client Good: a recycled payload's stale marker never validates");
standby.ShouldAllBe(x => x.Value.StatusCode == StatusCodes.Good);
// Positive control: an ordinary-tag item reusing the handle IS projected on the same Standby.
StatusCode.IsGood((await client.DeleteMonitoredItemsAsync(H(), sub, new[] { levelItem }, ct)).Results[0]).ShouldBeTrue();
rig.Sink.WriteValue(raw, 99.0, OpcUaStatus.Good, DateTime.UtcNow, AddressSpaceRealm.Raw);
await AddItemAsync(client, sub, Raw(session, raw), handle, 10, ct);
var control = await CollectAsync(client, ct, TimeSpan.FromSeconds(1));
control.Where(x => x.Handle == handle).ShouldNotBeEmpty();
control.Where(x => x.Handle == handle).ShouldAllBe(x => x.Value.StatusCode == StatusCodes.BadOutOfService);
}
/// <summary>Sums every measurement of ONE counter instance.</summary>
private sealed class CounterTotal : IDisposable
{
private readonly MeterListener _listener = new();
private long _total;
public CounterTotal(Instrument instrument)
{
_listener.InstrumentPublished = (published, l) =>
{
if (ReferenceEquals(published, instrument)) l.EnableMeasurementEvents(published);
};
_listener.SetMeasurementEventCallback<long>((_, value, _, _) => Interlocked.Add(ref _total, value));
_listener.Start();
}
public long Total => Interlocked.Read(ref _total);
public void Dispose() => _listener.Dispose();
}
/// <summary>
/// <c>standby.projected_notifications</c> counts each notification the Standby rewrites, at Publish and again at
/// Republish (each is a delivery), and nothing while Active.
/// </summary>
[Fact]
public async Task Projected_notifications_counts_each_rewritten_notification_per_delivery_and_nothing_while_Active()
{
var ct = Deadline();
await using var rig = await StartModeAsync("Warm717ProjMetric", "Warm", ct);
var projection = rig.Server.CreateStandbyProjection()!;
var raw = rig.EnsureRawVariable("m", "Double", writable: false);
rig.Sink.WriteValue(raw, 1.0, OpcUaStatus.Good, DateTime.UtcNow, AddressSpaceRealm.Raw);
Promote(projection);
using var counter = new CounterTotal(OtOpcUaTelemetry.StandbyProjectedNotifications);
using var session = await rig.ConnectAsync(anonymous: true, ct);
var client = (SessionClient)session;
var sub = (await client.CreateSubscriptionAsync(H(), 50, 300, 10, 0, true, 0, ct)).SubscriptionId;
await AddItemAsync(client, sub, Raw(session, raw), 1, 10, ct);
await AddItemAsync(client, sub, VariableIds.Server_ServiceLevel, 2, 1, ct);
await PublishDataAsync(client, ct);
counter.Total.ShouldBe(0, "nothing is rewritten while Active");
await client.SetPublishingModeAsync(H(), false, new[] { sub }, ct);
rig.Sink.WriteValue(raw, 10.0, OpcUaStatus.Good, DateTime.UtcNow, AddressSpaceRealm.Raw);
rig.Sink.WriteValue(raw, 11.0, OpcUaStatus.Good, DateTime.UtcNow, AddressSpaceRealm.Raw);
rig.Sink.WriteValue(raw, 12.0, OpcUaStatus.Good, DateTime.UtcNow, AddressSpaceRealm.Raw);
new SdkServiceLevelPublisher(rig.Server.CurrentInstance!,
Microsoft.Extensions.Logging.Abstractions.NullLogger<SdkServiceLevelPublisher>.Instance).Publish(90);
DemoteDeliveryOnly(projection);
await client.SetPublishingModeAsync(H(), true, new[] { sub }, ct);
var msg = await PublishDataAsync(client, ct);
var changes = DataChanges(msg.NotificationMessage);
var rewritten = changes.Count(x => x.Handle == 1 && x.Value.StatusCode == StatusCodes.BadOutOfService);
rewritten.ShouldBe(3, "the three queued Good samples");
changes.ShouldContain(x => x.Handle == 2 && x.Value.StatusCode == StatusCodes.Good, "i=2267 is not rewritten, nor counted");
counter.Total.ShouldBe(rewritten);
await client.RepublishAsync(H(), sub, msg.NotificationMessage.SequenceNumber, ct);
counter.Total.ShouldBe(2 * rewritten, "a Republish is another delivery, rewritten and counted again");
}
/// <summary>
/// <c>standby.suppressed_alarm_events</c> counts one per new alarm event withheld on the Standby, and none once
/// Active (where the same transition is delivered).
/// </summary>
[Fact]
public async Task Suppressed_alarm_events_counts_each_withheld_event_on_the_Standby_and_none_once_Active()
{
var ct = Deadline();
await using var rig = await StartModeAsync("Warm717AlarmMetric", "Warm", ct);
var projection = rig.Server.CreateStandbyProjection()!;
rig.Enforce(false, [Key()], ClusterGrant(KeyPrincipal, "acl-key", All));
rig.MaterialiseScripted();
using var counter = new CounterTotal(OtOpcUaTelemetry.StandbySuppressedAlarmEvents);
using var session = await rig.ConnectAsync(anonymous: false, ct);
var events = new RecordedEvents();
var subscription = await RecordedEvents.SubscribeAsync(session, events, ct);
var serverItem = RecordedEvents.AddEventItem(subscription, ObjectIds.Server);
await subscription.ApplyChangesAsync(ct);
rig.Project(native: false, new AlarmConditionSnapshot(Active: true, Acknowledged: false, Confirmed: true, Enabled: true,
Shelving: AlarmShelvingKind.Unshelved, Severity: 500, Message: "standby-1"));
rig.Project(native: false, new AlarmConditionSnapshot(Active: true, Acknowledged: false, Confirmed: true, Enabled: true,
Shelving: AlarmShelvingKind.Unshelved, Severity: 600, Message: "standby-2"));
counter.Total.ShouldBe(2, "one per withheld transition");
Promote(projection);
await RecordedEvents.WaitUntilAsync(() => events.HasType(serverItem, ObjectTypeIds.RefreshEndEventType));
rig.Project(native: false, new AlarmConditionSnapshot(Active: true, Acknowledged: false, Confirmed: true, Enabled: true,
Shelving: AlarmShelvingKind.Unshelved, Severity: 700, Message: "active"));
await RecordedEvents.WaitUntilAsync(() => events.Has(serverItem, "active"));
counter.Total.ShouldBe(2, "a delivered event is not counted");
events.Has(serverItem, "standby-1").ShouldBeFalse("withheld events are never delivered as transitions");
await subscription.DeleteAsync(true, ct);
}
}
tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/WarmRedundancyWireTests.Deferred.cs
+207
View File
@@ -0,0 +1,207 @@
using Opc.Ua;
using Opc.Ua.Client;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
using ZB.MOM.WW.OtOpcUa.Configuration.Enums;
using static ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests.ServerEventRig;
namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests;
/// <summary>
/// Gitea #717 test depth deferred from the first landing (plan "Implementation status"): the DataEncoding read (T6),
/// a monitored item deleted between queue and publish (T7; the reused-payload case is in
/// <see cref="WarmRedundancyPooledWireTests"/>), and an ACL-stood-down item across a transition
/// (T8).
/// </summary>
public sealed partial class WarmRedundancyWireTests
{
// ------------------------------------------------------------------ T6 DataEncoding
private static readonly QualifiedName DefaultBinary = new(BrowseNames.DefaultBinary);
private static readonly QualifiedName DefaultXml = new(BrowseNames.DefaultXml);
private static async Task<DataValue> ReadEncodedAsync(ISession session, NodeId node, QualifiedName encoding, CancellationToken ct)
{
var r = await ((SessionClient)session).ReadAsync(H(), 0, TimestampsToReturn.Both,
new[] { new ReadValueId { NodeId = node, AttributeId = Attributes.Value, DataEncoding = encoding } }, ct);
return r.Results[0];
}
/// <summary>
/// Spec §4.4: a Read naming a DataEncoding is projected like any other value read on the Standby — a structured
/// value never leaves through the encoding path — while an encoding the value cannot take is still reported as
/// the base read's own error (the projection runs after it). A live item created with a DataEncoding is projected
/// too, and promotion queues it the current value read through the item's own encoding.
/// </summary>
[Fact]
public async Task A_DataEncoding_read_and_item_are_projected_on_the_Standby_and_read_through_the_encoding_once_promoted()
{
var ct = Deadline();
await using var rig = await StartModeAsync("Warm717Encoding", "Warm", ct);
var projection = rig.Server.CreateStandbyProjection()!;
var structured = rig.EnsureRawVariable("eu", "Structure", writable: false);
var scalar = rig.EnsureRawVariable("dbl", "Double", writable: false);
var eu = new EUInformation { NamespaceUri = "urn:717", UnitId = 4408652, DisplayName = new LocalizedText("degC") };
rig.Sink.WriteValue(structured, new ExtensionObject(eu), OpcUaStatus.Good, DateTime.UtcNow, AddressSpaceRealm.Raw);
rig.Sink.WriteValue(scalar, 3.5, OpcUaStatus.Good, DateTime.UtcNow, AddressSpaceRealm.Raw);
using var session = await rig.ConnectAsync(anonymous: true, ct);
var standbyRead = await ReadEncodedAsync(session, Raw(session, structured), DefaultBinary, ct);
standbyRead.StatusCode.ShouldBe(StatusCodes.BadOutOfService);
standbyRead.WrappedValue.IsNull.ShouldBeTrue("the encoded structure must not leave the Standby");
(await ReadEncodedAsync(session, Raw(session, structured), DefaultXml, ct)).StatusCode.ShouldBe(StatusCodes.BadOutOfService);
var scalarWithEncoding = await ReadEncodedAsync(session, Raw(session, scalar), DefaultBinary, ct);
scalarWithEncoding.StatusCode.ShouldBe(StatusCodes.BadDataEncodingUnsupported,
"an encoding a scalar cannot take is the base read's own error, not masked as BadOutOfService");
scalarWithEncoding.WrappedValue.IsNull.ShouldBeTrue();
var client = (SessionClient)session;
var sub = (await client.CreateSubscriptionAsync(H(), 50, 300, 10, 0, true, 0, ct)).SubscriptionId;
var created = await client.CreateMonitoredItemsAsync(H(), sub, TimestampsToReturn.Both, new[]
{
new MonitoredItemCreateRequest
{
ItemToMonitor = new ReadValueId { NodeId = Raw(session, structured), AttributeId = Attributes.Value, DataEncoding = DefaultXml },
MonitoringMode = MonitoringMode.Reporting,
RequestedParameters = new MonitoringParameters { ClientHandle = 5, SamplingInterval = 0, QueueSize = 10, DiscardOldest = true },
},
}, ct);
StatusCode.IsGood(created.Results[0].StatusCode).ShouldBeTrue(created.Results[0].StatusCode.ToString());
var initial = await CollectAsync(client, ct, TimeSpan.FromSeconds(1));
initial.Where(x => x.Handle == 5).ShouldNotBeEmpty();
initial.Where(x => x.Handle == 5).ShouldAllBe(x => x.Value.StatusCode == StatusCodes.BadOutOfService && x.Value.WrappedValue.IsNull,
"the encoded item's initial sample is projected on the Standby");
Promote(projection);
var promoted = await CollectAsync(client, ct, TimeSpan.FromSeconds(1));
var good = promoted.Where(x => x.Handle == 5 && x.Value.StatusCode == StatusCodes.Good).ShouldHaveSingleItem(
"promotion queues the current value to the item through its own DataEncoding");
DecodeEu(good.Value).UnitId.ShouldBe(eu.UnitId);
((ExtensionObject)good.Value.WrappedValue.AsBoxedObject()!).TypeId.ShouldBe(
new ExpandedNodeId(ObjectIds.EUInformation_Encoding_DefaultXml), "the item's own (XML) encoding, not the default binary one");
var activeRead = await ReadEncodedAsync(session, Raw(session, structured), DefaultBinary, ct);
activeRead.StatusCode.ShouldBe(StatusCodes.Good);
DecodeEu(activeRead).DisplayName.Text.ShouldBe("degC");
((ExtensionObject)activeRead.WrappedValue.AsBoxedObject()!).TypeId.ShouldBe(new ExpandedNodeId(ObjectIds.EUInformation_Encoding_DefaultBinary));
}
private static EUInformation DecodeEu(DataValue value)
{
var body = value.WrappedValue.AsBoxedObject().ShouldBeOfType<ExtensionObject>();
body.TryGetValue<EUInformation>(out var eu).ShouldBeTrue();
return eu!;
}
// ------------------------------------------------------------------ T7 item deleted / reused between queue and publish
/// <summary>
/// An ordinary-tag item deleted after its Good samples were queued but before the Publish that would carry
/// them: the deleted item's notifications never surface, a sibling item queued in the same window is still
/// first-delivered projected (D4), and Republish returns the same recorded choice.
/// </summary>
[Fact]
public async Task An_item_deleted_between_queue_and_publish_leaves_nothing_behind_and_its_sibling_is_still_projected()
{
var ct = Deadline();
await using var rig = await StartModeAsync("Warm717Delete", "Warm", ct);
var projection = rig.Server.CreateStandbyProjection()!;
var gone = rig.EnsureRawVariable("gone", "Double", writable: false);
var kept = rig.EnsureRawVariable("kept", "Double", writable: false);
rig.Sink.WriteValue(gone, 1.0, OpcUaStatus.Good, DateTime.UtcNow, AddressSpaceRealm.Raw);
rig.Sink.WriteValue(kept, 1.0, OpcUaStatus.Good, DateTime.UtcNow, AddressSpaceRealm.Raw);
Promote(projection);
using var session = await rig.ConnectAsync(anonymous: true, ct);
var client = (SessionClient)session;
var sub = (await client.CreateSubscriptionAsync(H(), 50, 300, 10, 0, true, 0, ct)).SubscriptionId;
var goneItem = await AddItemAsync(client, sub, Raw(session, gone), 1, 10, ct);
await AddItemAsync(client, sub, Raw(session, kept), 2, 10, ct);
(await CollectAsync(client, ct, TimeSpan.FromSeconds(1))).Select(x => x.Handle).Distinct().Order().ShouldBe(new uint[] { 1, 2 });
await client.SetPublishingModeAsync(H(), false, new[] { sub }, ct);
rig.Sink.WriteValue(gone, 10.0, OpcUaStatus.Good, DateTime.UtcNow, AddressSpaceRealm.Raw);
rig.Sink.WriteValue(kept, 20.0, OpcUaStatus.Good, DateTime.UtcNow, AddressSpaceRealm.Raw);
rig.Sink.WriteValue(gone, 11.0, OpcUaStatus.Good, DateTime.UtcNow, AddressSpaceRealm.Raw);
var deleted = await client.DeleteMonitoredItemsAsync(H(), sub, new[] { goneItem }, ct);
StatusCode.IsGood(deleted.Results[0]).ShouldBeTrue();
DemoteDeliveryOnly(projection);
await client.SetPublishingModeAsync(H(), true, new[] { sub }, ct);
var msg = await PublishDataAsync(client, ct);
var changes = DataChanges(msg.NotificationMessage);
changes.ShouldNotContain(x => x.Handle == 1, "a deleted item's queued samples are never delivered");
changes.Where(x => x.Handle == 2).ShouldNotBeEmpty();
changes.Where(x => x.Handle == 2).ShouldAllBe(x => x.Value.StatusCode == StatusCodes.BadOutOfService && x.Value.WrappedValue.IsNull);
var again = await client.RepublishAsync(H(), sub, msg.NotificationMessage.SequenceNumber, ct);
DataChanges(again.NotificationMessage).Select(x => (x.Handle, x.Value.StatusCode))
.ShouldBe(changes.Select(x => (x.Handle, x.Value.StatusCode)), "Republish applies the recorded choice to the same payload");
}
// ------------------------------------------------------------------ T8 ACL-stood-down item on transition
/// <summary>
/// Plan T8: an item the ACL revalidation has stood down (#562) receives nothing from either transition — no
/// BadOutOfService on demotion and no current value on promotion — while a permitted sibling receives both. A
/// restored grant re-attaches the item and the next transition reaches it again (the skip is the stand-down,
/// not the node).
/// </summary>
[Fact]
public async Task An_ACL_stood_down_item_receives_nothing_from_a_demotion_or_a_promotion()
{
var ct = Deadline();
await using var rig = await StartModeAsync("Warm717AclLive", "Warm", ct);
var projection = rig.Server.CreateStandbyProjection()!;
rig.EnsureEquipment();
rig.Sink.EnsureVariable("EqA/a", EqA, "a", "Double", writable: false, AddressSpaceRealm.Uns);
rig.Sink.EnsureVariable("EqB/b", EqB, "b", "Double", writable: false, AddressSpaceRealm.Uns);
rig.Sink.WriteValue("EqA/a", 1.0, OpcUaStatus.Good, DateTime.UtcNow, AddressSpaceRealm.Uns);
rig.Sink.WriteValue("EqB/b", 1.0, OpcUaStatus.Good, DateTime.UtcNow, AddressSpaceRealm.Uns);
const NodePermissions DataRead = NodePermissions.Browse | NodePermissions.Read | NodePermissions.Subscribe;
rig.Enforce(true, [Key()], EquipmentGrant(EqA, DataRead), EquipmentGrant(EqB, DataRead));
Promote(projection);
using var session = await rig.ConnectAsync(anonymous: true, ct);
var client = (SessionClient)session;
var sub = (await client.CreateSubscriptionAsync(H(), 50, 300, 10, 0, true, 0, ct)).SubscriptionId;
await AddItemAsync(client, sub, UnsNodeId(session, "EqA/a"), 1, 10, ct);
await AddItemAsync(client, sub, UnsNodeId(session, "EqB/b"), 2, 10, ct);
(await CollectAsync(client, ct, TimeSpan.FromSeconds(1))).Select(x => x.Handle).Distinct().Order().ShouldBe(new uint[] { 1, 2 });
rig.Enforce(true, [Key()], EquipmentGrant(EqA, DataRead)); // EqB revoked
rig.Nm.RevalidateMonitoredItems();
var revoked = await CollectAsync(client, ct, TimeSpan.FromSeconds(1));
revoked.ShouldContain(x => x.Handle == 2 && x.Value.StatusCode == StatusCodes.BadUserAccessDenied, "the item is told, then stood down");
// Demotion, in the spec §4.7 order, counting what step 5 queued.
projection.SetStandbyDelivery(true);
projection.SetAlarmSuppression(true);
projection.SetDataProjection(true);
projection.QueueCurrentToLiveItems(true).ShouldBe(1, "only the permitted item is queued to");
rig.Sink.WriteValue("EqB/b", 5.0, OpcUaStatus.Good, DateTime.UtcNow, AddressSpaceRealm.Uns);
var demoted = await CollectAsync(client, ct, TimeSpan.FromSeconds(1));
demoted.Where(x => x.Handle == 1).ShouldHaveSingleItem().Value.StatusCode.ShouldBe(StatusCodes.BadOutOfService);
demoted.ShouldNotContain(x => x.Handle == 2, "a stood-down item receives no demotion value");
rig.Sink.WriteValue("EqA/a", 7.0, OpcUaStatus.Good, DateTime.UtcNow, AddressSpaceRealm.Uns);
rig.Sink.WriteValue("EqB/b", 8.0, OpcUaStatus.Good, DateTime.UtcNow, AddressSpaceRealm.Uns);
projection.SetDataProjection(false);
projection.QueueCurrentToLiveItems(false).ShouldBe(1);
projection.SetStandbyDelivery(false);
projection.SetAlarmSuppression(false);
projection.RefreshAllEventItems();
var promoted = await CollectAsync(client, ct, TimeSpan.FromSeconds(1));
promoted.ShouldContain(x => x.Handle == 1 && x.Value.StatusCode == StatusCodes.Good && Equals(x.Value.ClrValue(), 7.0));
promoted.ShouldNotContain(x => x.Handle == 2, "a stood-down item receives no promotion value");
// Positive control: the restored grant re-attaches the item and the next demotion reaches it.
rig.Enforce(true, [Key()], EquipmentGrant(EqA, DataRead), EquipmentGrant(EqB, DataRead));
rig.Nm.RevalidateMonitoredItems();
await CollectAsync(client, ct, TimeSpan.FromMilliseconds(500));
Demote(projection);
var restored = await CollectAsync(client, ct, TimeSpan.FromSeconds(1));
restored.ShouldContain(x => x.Handle == 2 && x.Value.StatusCode == StatusCodes.BadOutOfService);
}
}
tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/WarmRedundancyWireTests.cs
+11 -11
View File
@@ -16,7 +16,7 @@ namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests;
/// <see cref="IStandbyProjection"/> in the spec §4.7 step order; the actor that owns that order has its own tests.
/// </summary>
[Trait("Category", "Integration")]
public sealed class WarmRedundancyWireTests
public sealed partial class WarmRedundancyWireTests
{
private const string Peer = "urn:OtOpcUa:peer-717";
@@ -25,7 +25,7 @@ public sealed class WarmRedundancyWireTests
| NodePermissions.AlarmAcknowledge | NodePermissions.AlarmConfirm | NodePermissions.AlarmShelve
| NodePermissions.MethodCall | NodePermissions.WriteOperate;
private static Task<ServerEventRig> StartModeAsync(string name, string mode, CancellationToken ct) =>
internal static Task<ServerEventRig> StartModeAsync(string name, string mode, CancellationToken ct) =>
ServerEventRig.StartAsync(name, encrypted: false, ct, o =>
{
o.Redundancy = new OpcUaRedundancyOptions { Mode = mode };
@@ -51,19 +51,19 @@ public sealed class WarmRedundancyWireTests
p.QueueCurrentToLiveItems(true);
}
private static CancellationToken Deadline(int seconds = 90)
internal static CancellationToken Deadline(int seconds = 90)
{
var cts = CancellationTokenSource.CreateLinkedTokenSource(TestContext.Current.CancellationToken);
cts.CancelAfter(TimeSpan.FromSeconds(seconds));
return cts.Token;
}
private static RequestHeader H() => new() { Timestamp = DateTimeUtc.Now, TimeoutHint = 10_000 };
internal static RequestHeader H() => new() { Timestamp = DateTimeUtc.Now, TimeoutHint = 10_000 };
private static NodeId Raw(ISession session, string id) =>
internal static NodeId Raw(ISession session, string id) =>
new(id, (ushort)session.NamespaceUris.GetIndex(V3NodeIds.RawNamespaceUri));
private static async Task<DataValue> ReadAsync(ISession session, NodeId node, CancellationToken ct, uint attribute = Attributes.Value,
internal static async Task<DataValue> ReadAsync(ISession session, NodeId node, CancellationToken ct, uint attribute = Attributes.Value,
string? indexRange = null)
{
var r = await ((SessionClient)session).ReadAsync(H(), 0, TimestampsToReturn.Both,
@@ -332,7 +332,7 @@ public sealed class WarmRedundancyWireTests
}
}
private static async Task<NodeId> TranslateAsync(ISession session, NodeId start, string path, CancellationToken ct)
internal static async Task<NodeId> TranslateAsync(ISession session, NodeId start, string path, CancellationToken ct)
{
var elements = path.Split('/').Select(name => new RelativePathElement
{
@@ -714,11 +714,11 @@ public sealed class WarmRedundancyWireTests
/// <summary>The D4 boundary alone (spec §4.7 demotion step 2), without the projection that follows it, so the
/// decision record is what the test observes.</summary>
private static void DemoteDeliveryOnly(IStandbyProjection p) => p.SetStandbyDelivery(true);
internal static void DemoteDeliveryOnly(IStandbyProjection p) => p.SetStandbyDelivery(true);
// ------------------------------------------------------------------ raw client helpers
private static async Task<uint> AddItemAsync(SessionClient client, uint sub, NodeId node, uint handle, uint queue, CancellationToken ct,
internal static async Task<uint> AddItemAsync(SessionClient client, uint sub, NodeId node, uint handle, uint queue, CancellationToken ct,
MonitoringFilter? filter = null)
{
var r = await client.CreateMonitoredItemsAsync(H(), sub, TimestampsToReturn.Both, new[]
@@ -750,7 +750,7 @@ public sealed class WarmRedundancyWireTests
return list;
}
private static async Task<PublishResponse> PublishDataAsync(SessionClient client, CancellationToken ct)
internal static async Task<PublishResponse> PublishDataAsync(SessionClient client, CancellationToken ct)
{
for (var i = 0; i < 40; i++)
{
@@ -761,7 +761,7 @@ public sealed class WarmRedundancyWireTests
throw new InvalidOperationException("no data message");
}
private static async Task<List<(uint Handle, DataValue Value)>> CollectAsync(SessionClient client, CancellationToken ct, TimeSpan window)
internal static async Task<List<(uint Handle, DataValue Value)>> CollectAsync(SessionClient client, CancellationToken ct, TimeSpan window)
{
var collected = new List<(uint, DataValue)>();
var until = DateTime.UtcNow + window;
tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/WarmStandbyCloneMatrixTests.cs
+209
View File
@@ -0,0 +1,209 @@
using Opc.Ua;
using Shouldly;
using Xunit;
namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests;
/// <summary>The SDK's notification-payload pool is process-wide; the recycled-payload test must not race another
/// test renting from it.</summary>
[CollectionDefinition(Name, DisableParallelization = true)]
public sealed class SdkPayloadPoolCollection
{
/// <summary>The collection name.</summary>
public const string Name = "sdk-payload-pool";
}
/// <summary>
/// Gitea #717 T7, the payload matrix the D4 rewrite meets at delivery: SDK 2.0.1 hands each Publish and Republish a
/// <c>CoreUtils.Clone</c> of the cached message, which may carry several <see cref="DataChangeNotification"/>s (with
/// <see cref="DiagnosticInfo"/>s), an <see cref="EventNotificationList"/> and a <see cref="StatusChangeNotification"/>
/// side by side. Every payload kind must survive the clone, only marked ordinary-tag values may change, nothing the
/// SDK owns may be mutated, and the result must still encode. The stale-marker case is driven through the SDK's real
/// payload pool.
/// </summary>
[Collection(SdkPayloadPoolCollection.Name)]
public sealed class WarmStandbyCloneMatrixTests
{
private static readonly DateTimeUtc Source = (DateTimeUtc)new DateTime(2026, 11, 13, 8, 0, 0, DateTimeKind.Utc);
private static readonly DateTimeUtc Server = (DateTimeUtc)new DateTime(2026, 11, 13, 8, 0, 1, DateTimeKind.Utc);
private static DataValue Good(double v) => new(new Variant(v), StatusCodes.Good, Source, Server);
private static MonitoredItemNotification Marked(uint handle, double v)
{
var n = new MonitoredItemNotification { ClientHandle = handle, Value = Good(v) };
n.Message = new OrdinaryTagMarker(handle, n.Value);
return n;
}
/// <summary>The cached message as the SDK builds it: two data-change payloads (the first with per-item diagnostics),
/// an event list and a status change, in that order.</summary>
private static NotificationMessage Cached() => new()
{
SequenceNumber = 42,
PublishTime = Server,
NotificationData = new List<ExtensionObject>
{
new(new DataChangeNotification
{
MonitoredItems = new List<MonitoredItemNotification>
{
Marked(1, 1.5),
new() { ClientHandle = 2, Value = Good(2.5) }, // i=2267 or a condition child: unmarked
},
DiagnosticInfos = new List<DiagnosticInfo>
{
new() { SymbolicId = 11, AdditionalInfo = "first" },
new() { SymbolicId = 12, AdditionalInfo = "second" },
},
}),
new(new EventNotificationList
{
Events = new List<EventFieldList>
{
new() { ClientHandle = 9, EventFields = new List<Variant> { new("event-id"), new("message") } },
},
}),
new(new StatusChangeNotification { Status = StatusCodes.BadTimeout }),
new(new DataChangeNotification
{
MonitoredItems = new List<MonitoredItemNotification> { Marked(3, 3.5) },
}),
},
};
private static T Body<T>(NotificationMessage m, int index) where T : class, IEncodeable
{
m.NotificationData[index].TryGetValue<T>(out var body).ShouldBeTrue($"payload {index} is a {typeof(T).Name}");
return body!;
}
private static MonitoredItemNotification Item(NotificationMessage m, int payload, int item) =>
Body<DataChangeNotification>(m, payload).MonitoredItems[item];
private static byte[] Encode(NotificationMessage m) =>
BinaryEncoder.EncodeMessage(m, ServiceMessageContext.Create(DefaultTelemetry.Create(_ => { })));
[Fact]
public void Every_payload_kind_survives_the_delivery_clone_and_only_marked_values_are_rewritten()
{
var cached = Cached();
var cachedBytes = Encode(cached);
var delivered = CoreUtils.Clone(cached)!;
var deliveredBytes = Encode(delivered);
// The clone carries every marker, each still valid against its own (cloned) notification.
OrdinaryTagMarker.IsOrdinaryTag(Item(delivered, 0, 0)).ShouldBeTrue();
OrdinaryTagMarker.IsOrdinaryTag(Item(delivered, 0, 1)).ShouldBeFalse();
OrdinaryTagMarker.IsOrdinaryTag(Item(delivered, 3, 0)).ShouldBeTrue();
var projected = StandbyNotificationRewriter.Project(delivered, out var rewritten);
rewritten.ShouldBe(2, "both marked values, across both data-change payloads");
projected.SequenceNumber.ShouldBe(42u);
projected.PublishTime.ShouldBe(Server);
projected.NotificationData.Count.ShouldBe(4, "no payload is dropped or added");
Item(projected, 0, 0).Value.StatusCode.ShouldBe(StatusCodes.BadOutOfService);
Item(projected, 0, 0).Value.WrappedValue.IsNull.ShouldBeTrue();
Item(projected, 0, 0).Value.SourceTimestamp.ShouldBe(DateTimeUtc.MinValue);
Item(projected, 0, 0).Value.ServerTimestamp.ShouldBe(Server);
Item(projected, 0, 0).ClientHandle.ShouldBe(1u);
Item(projected, 0, 1).Value.ShouldBe(Good(2.5));
Item(projected, 3, 0).Value.StatusCode.ShouldBe(StatusCodes.BadOutOfService);
var diagnostics = Body<DataChangeNotification>(projected, 0).DiagnosticInfos;
diagnostics.Count.ShouldBe(2, "per-item diagnostics stay aligned with the items");
diagnostics[0].AdditionalInfo.ShouldBe("first");
diagnostics[1].AdditionalInfo.ShouldBe("second");
Body<EventNotificationList>(projected, 1).IsEqual(Body<EventNotificationList>(cached, 1)).ShouldBeTrue("events pass through");
Body<StatusChangeNotification>(projected, 2).Status.ShouldBe(StatusCodes.BadTimeout);
// Nothing the SDK owns moved: neither the per-delivery clone handed in nor the cached original.
Encode(delivered).ShouldBe(deliveredBytes, "the delivered clone is not mutated");
Encode(cached).ShouldBe(cachedBytes, "the cached message is not mutated");
OrdinaryTagMarker.IsOrdinaryTag(Item(cached, 0, 0)).ShouldBeTrue("the cached original keeps its marker for a later Republish");
// The projected message is wire-valid and the marker never reaches the wire.
var decoded = BinaryDecoder.DecodeMessage<NotificationMessage>(Encode(projected),
ServiceMessageContext.Create(DefaultTelemetry.Create(_ => { })))!;
decoded.NotificationData.Count.ShouldBe(4);
Item(decoded, 0, 0).Value.StatusCode.ShouldBe(StatusCodes.BadOutOfService);
Item(decoded, 0, 0).Message.ShouldBeNull();
Body<DataChangeNotification>(decoded, 0).DiagnosticInfos.Count.ShouldBe(2);
}
/// <summary>A Republish is a second clone of the same cached message: projecting it gives the same bytes.</summary>
[Fact]
public void Projecting_two_delivery_clones_of_one_cached_message_gives_identical_payloads()
{
var cached = Cached();
var first = StandbyNotificationRewriter.Project(CoreUtils.Clone(cached)!, out var a);
var second = StandbyNotificationRewriter.Project(CoreUtils.Clone(cached)!, out var b);
a.ShouldBe(b);
Encode(second).ShouldBe(Encode(first));
}
/// <summary>A message with no marked value comes back as the same instance (nothing rebuilt, nothing counted).</summary>
[Fact]
public void A_message_with_only_events_status_and_unmarked_values_is_returned_as_is()
{
var cached = Cached();
cached.NotificationData = new List<ExtensionObject> { cached.NotificationData[1], cached.NotificationData[2] };
var delivered = CoreUtils.Clone(cached)!;
ReferenceEquals(StandbyNotificationRewriter.Project(delivered, out var rewritten), delivered).ShouldBeTrue();
rewritten.ShouldBe(0);
}
/// <summary>
/// The SDK's own pool, not a hand-made stale marker: a payload stamped for an ordinary tag and returned (as
/// <c>SentMessageQueue</c> does on acknowledge) comes back out with its marker, here even reusing the SAME
/// ClientHandle. With a different value it is not an ordinary-tag notification and is never rewritten.
/// </summary>
[Fact]
public void A_payload_recycled_through_the_sdk_pool_keeps_its_stale_marker_which_never_validates()
{
var stamped = (MonitoredItemNotification)MonitoredItemNotificationActivator.Instance.CreateInstance();
stamped.ClientHandle = 7;
stamped.Value = Good(1.0);
stamped.Message = new OrdinaryTagMarker(7, stamped.Value);
stamped.Reuse();
// Rent until the stamped payload comes back (other pooled payloads may sit ahead of it), then return the rest.
var others = new List<MonitoredItemNotification>();
MonitoredItemNotification? recycled = null;
for (var i = 0; i <= PooledEncodeableType<MonitoredItemNotification>.DefaultMaximumRetained && recycled is null; i++)
{
var rented = (MonitoredItemNotification)MonitoredItemNotificationActivator.Instance.CreateInstance();
if (ReferenceEquals(rented, stamped)) recycled = rented;
else others.Add(rented);
}
foreach (var other in others) other.Reuse();
recycled.ShouldNotBeNull("precondition: the pool hands the stamped payload back out");
try
{
recycled.Message.ShouldBeOfType<OrdinaryTagMarker>("precondition: Initialize() does not clear the helper");
recycled.ClientHandle = 7; // the client reused the handle for a non-ordinary item
recycled.Value = new DataValue(new Variant((byte)90), StatusCodes.Good, Source, Server);
OrdinaryTagMarker.IsOrdinaryTag(recycled).ShouldBeFalse();
var message = new NotificationMessage
{
SequenceNumber = 1,
NotificationData = new List<ExtensionObject>
{
new(new DataChangeNotification { MonitoredItems = new List<MonitoredItemNotification> { recycled } }),
},
};
var projected = StandbyNotificationRewriter.Project(CoreUtils.Clone(message)!, out var rewritten);
rewritten.ShouldBe(0);
Item(projected, 0, 0).Value.StatusCode.ShouldBe(StatusCodes.Good);
}
finally
{
recycled.Message = null!;
recycled.Reuse();
}
}
}
tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/WarmRedundancyMetricsTests.cs
+234
View File
@@ -0,0 +1,234 @@
using System.Collections.Concurrent;
using System.Diagnostics.Metrics;
using Akka.Actor;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Redundancy;
using ZB.MOM.WW.OtOpcUa.Commons.Observability;
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
using ZB.MOM.WW.OtOpcUa.Commons.Types;
using ZB.MOM.WW.OtOpcUa.Runtime.Drivers;
using ZB.MOM.WW.OtOpcUa.Runtime.OpcUa;
using ZB.MOM.WW.OtOpcUa.Runtime.Redundancy;
using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness;
using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Redundancy;
namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.OpcUa;
/// <summary>
/// The Warm instruments are process-wide statics (the role gauge's backing value too), so these tests run alone:
/// another Warm actor test running at the same time would move what is asserted.
/// </summary>
[CollectionDefinition(Name, DisableParallelization = true)]
public sealed class WarmRedundancyMetricsCollection
{
/// <summary>The collection name.</summary>
public const string Name = "warm-redundancy-metrics";
}
/// <summary>
/// Gitea #717 T10: each Runtime-side Warm instrument fires on its own path — <c>warm_transition</c> with its
/// direction/reason tags, the <c>warm_role</c> gauge, and <c>snapshot_rejected</c> per reason from both
/// subscribers (<see cref="OpcUaPublishActor"/> and <see cref="DriverHostActor"/>). Listeners select by
/// instrument INSTANCE, never by meter name (#503). <c>mode_mismatch</c> lives in
/// <see cref="OpcUaPublishActorWarmTests"/>.
/// </summary>
[Collection(WarmRedundancyMetricsCollection.Name)]
public sealed class WarmRedundancyMetricsTests : RuntimeActorTestBase
{
private static readonly NodeId Other = NodeId.Parse("10.255.255.1:9");
private readonly ConcurrentQueue<string> _log = new();
private long _seq;
private NodeId Self
{
get
{
var cluster = Akka.Cluster.Cluster.Get(Sys);
AwaitCondition(() => cluster.SelfMember.Status == Akka.Cluster.MemberStatus.Up, PresenceBudget);
return NodeId.Parse($"{cluster.SelfAddress.Host}:{cluster.SelfAddress.Port}");
}
}
private IActorRef StartWarm() =>
Sys.ActorOf(OpcUaPublishActor.PropsForTests(
serviceLevel: new LoggingPublisher(_log), localNode: Self, dbLess: true,
healthTickInterval: TimeSpan.FromMinutes(5), redundancy: new WarmRedundancySettings(OpcUaRedundancyMode.Warm),
standbyProjection: new RecordingProjection(_log)));
private RedundancyStateChanged Snapshot(NodeId primary, bool revisioned = true, long upNumber = 1) =>
new(new[]
{
new NodeRedundancyState(Self, primary == Self ? RedundancyRole.Primary : RedundancyRole.Secondary,
false, primary == Self, DateTime.UtcNow),
new NodeRedundancyState(Other, primary == Other ? RedundancyRole.Primary : RedundancyRole.Secondary,
false, primary == Other, DateTime.UtcNow),
},
CorrelationId.NewId(),
revisioned ? new SnapshotRevision(upNumber, 1, Interlocked.Increment(ref _seq)) : null);
/// <summary>Records every measurement of ONE instrument, with its tags.</summary>
private sealed class Recorder<T> : IDisposable where T : struct
{
private readonly MeterListener _listener = new();
private readonly ConcurrentQueue<(T Value, Dictionary<string, object?> Tags)> _seen = new();
public Recorder(Instrument instrument)
{
_listener.InstrumentPublished = (published, l) =>
{
if (ReferenceEquals(published, instrument)) l.EnableMeasurementEvents(published);
};
_listener.SetMeasurementEventCallback<T>((_, value, tags, _) =>
{
var copy = new Dictionary<string, object?>();
foreach (var tag in tags) copy[tag.Key] = tag.Value;
_seen.Enqueue((value, copy));
});
_listener.Start();
}
public IReadOnlyList<(T Value, Dictionary<string, object?> Tags)> Seen => _seen.ToArray();
public void Observe() => _listener.RecordObservableInstruments();
public void Dispose() => _listener.Dispose();
}
private void AwaitLogEndsWith(params string[] tail) =>
AwaitAssert(() => _log.TakeLast(tail.Length).ShouldBe(tail, string.Join(" ", _log)), PresenceBudget,
cancellationToken: TestContext.Current.CancellationToken);
[Fact]
public void Warm_transition_counts_each_promotion_and_demotion_with_direction_and_reason()
{
using var transitions = new Recorder<long>(OtOpcUaTelemetry.WarmTransition);
var actor = StartWarm();
actor.Tell(Snapshot(Self));
AwaitLogEndsWith("SL:250");
AwaitAssert(() => transitions.Seen.Count.ShouldBe(1), PresenceBudget, cancellationToken: TestContext.Current.CancellationToken);
transitions.Seen[0].Value.ShouldBe(1);
transitions.Seen[0].Tags["direction"].ShouldBe("promote");
transitions.Seen[0].Tags["reason"].ShouldBe("agreed", "membership and the accepted snapshot both name this node");
actor.Tell(Snapshot(Other));
AwaitLogEndsWith("queue:True");
AwaitAssert(() => transitions.Seen.Count.ShouldBe(2), PresenceBudget, cancellationToken: TestContext.Current.CancellationToken);
transitions.Seen[1].Tags["direction"].ShouldBe("demote");
transitions.Seen[1].Tags["reason"].ShouldBe("snapshot", "membership still names this node; the snapshot does not");
// A same-role snapshot (a heartbeat) is not a transition.
actor.Tell(Snapshot(Other));
ExpectNoMsg(TimeSpan.FromMilliseconds(300), TestContext.Current.CancellationToken);
transitions.Seen.Count.ShouldBe(2);
}
[Fact]
public void Warm_role_gauge_reads_Standby_at_boot_Active_after_promotion_and_Standby_after_demotion()
{
using var role = new Recorder<int>(OtOpcUaTelemetry.WarmRole);
int Current()
{
var before = role.Seen.Count;
role.Observe();
return role.Seen.Skip(before).Single().Value;
}
OtOpcUaTelemetry.SetWarmRole(-1);
Current().ShouldBe(-1, "not Warm until a Warm actor starts");
var actor = StartWarm();
AwaitAssert(() => Current().ShouldBe(0), PresenceBudget, cancellationToken: TestContext.Current.CancellationToken);
actor.Tell(Snapshot(Self));
AwaitLogEndsWith("SL:250");
AwaitAssert(() => Current().ShouldBe(1), PresenceBudget, cancellationToken: TestContext.Current.CancellationToken);
actor.Tell(Snapshot(Other));
AwaitLogEndsWith("queue:True");
AwaitAssert(() => Current().ShouldBe(0), PresenceBudget, cancellationToken: TestContext.Current.CancellationToken);
Sys.Stop(actor);
OtOpcUaTelemetry.SetWarmRole(-1);
}
[Fact]
public void Snapshot_rejected_counts_older_revisions_and_unrevisioned_snapshots_by_reason_at_the_publish_actor()
{
using var rejected = new Recorder<long>(OtOpcUaTelemetry.SnapshotRejected);
var actor = StartWarm();
actor.Tell(Snapshot(Self, upNumber: 5));
AwaitLogEndsWith("SL:250");
rejected.Seen.ShouldBeEmpty("an accepted snapshot is not counted");
actor.Tell(Snapshot(Other, upNumber: 4));
AwaitAssert(() => rejected.Seen.Count.ShouldBe(1), PresenceBudget, cancellationToken: TestContext.Current.CancellationToken);
rejected.Seen[0].Tags["reason"].ShouldBe("older_revision");
actor.Tell(Snapshot(Other, revisioned: false));
AwaitAssert(() => rejected.Seen.Count.ShouldBe(2), PresenceBudget, cancellationToken: TestContext.Current.CancellationToken);
rejected.Seen[1].Tags["reason"].ShouldBe("unrevisioned");
rejected.Seen.ShouldAllBe(m => m.Value == 1);
}
/// <summary>
/// The <see cref="DriverHostActor"/> half of spec §4.2 (its Primary gates): under Warm a stale snapshot is
/// dropped — the role view does not move — and counted; under Unset (no ordering) the same sequence moves it.
/// The view publishes once per accepted snapshot, so Primary@5, Secondary@4 (stale), Primary@6, Secondary@7
/// publishes exactly [true, true, false] when the stale one is dropped and [true, false, true, false] when not.
/// </summary>
[Theory]
[InlineData(true)]
[InlineData(false)]
public void Snapshot_rejected_counts_an_older_revision_at_the_driver_host_and_its_roles_do_not_move(bool warm)
{
using var rejected = new Recorder<long>(OtOpcUaTelemetry.SnapshotRejected);
var node = NodeId.Parse("driver-warm-order:4053");
var peer = NodeId.Parse("the-peer:4053");
var view = new PublishedRoleView();
var host = Sys.ActorOf(DriverHostActor.Props(
NewInMemoryDbFactory(), node, ackRouter: null, driverMemberCountProvider: () => 2,
redundancyRoleView: view, replicationPeerHost: "the-peer", enforceSnapshotOrdering: warm));
RedundancyStateChanged Snap(bool primary, long upNumber) => new(
[
new NodeRedundancyState(node, primary ? RedundancyRole.Primary : RedundancyRole.Secondary, primary, primary, DateTime.UtcNow),
new NodeRedundancyState(peer, primary ? RedundancyRole.Secondary : RedundancyRole.Primary, !primary, !primary, DateTime.UtcNow),
],
CorrelationId.NewId(), new SnapshotRevision(upNumber, 1, upNumber));
host.Tell(Snap(primary: true, 5));
host.Tell(Snap(primary: false, 4)); // a departed older incarnation's late heartbeat
host.Tell(Snap(primary: true, 6));
host.Tell(Snap(primary: false, 7));
bool[] expected = warm ? [true, true, false] : [true, false, true, false];
AwaitAssert(() => view.Published.ShouldBe(expected), PresenceBudget, cancellationToken: TestContext.Current.CancellationToken);
ExpectNoMsg(TimeSpan.FromMilliseconds(200), TestContext.Current.CancellationToken);
view.Published.ShouldBe(expected);
rejected.Seen.Count.ShouldBe(warm ? 1 : 0);
if (warm) rejected.Seen[0].Tags["reason"].ShouldBe("older_revision");
}
private sealed class PublishedRoleView : IRedundancyRoleView
{
private readonly List<bool> _published = [];
private readonly Lock _gate = new();
public IReadOnlyList<bool> Published
{
get { lock (_gate) return _published.ToArray(); }
}
public bool ShouldDrainAlarmHistory
{
get { lock (_gate) return _published.Count == 0 || _published[^1]; }
}
public void Publish(bool shouldDrainAlarmHistory)
{
lock (_gate) _published.Add(shouldDrainAlarmHistory);
}
}
}