diff --git a/docs/plans/2026-11-11-warm-redundancy-717-plan.md b/docs/plans/2026-11-11-warm-redundancy-717-plan.md index 6c9752a98..de06eb4ee 100644 --- a/docs/plans/2026-11-11-warm-redundancy-717-plan.md +++ b/docs/plans/2026-11-11-warm-redundancy-717-plan.md @@ -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. diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/WarmSingletonHandoverTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/WarmSingletonHandoverTests.cs new file mode 100644 index 000000000..531f23a5f --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/WarmSingletonHandoverTests.cs @@ -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; + +/// +/// Gitea #717 spec §4.2 / plan T3 on a REAL two-node cluster (pattern: ): +/// the redundancy singleton runs under a real 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 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. +/// +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 _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 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; + + /// The singleton's broadcast seam on each node: record, and relay to the subscriber unless held back. + 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"); + } + + /// + /// 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. + /// + [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 log) : IServiceLevelPublisher + { + public void Publish(byte serviceLevel) => log.Enqueue($"SL:{serviceLevel}"); + } + + private sealed class LogProjection(ConcurrentQueue 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; } + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/WarmRedundancyPooledWireTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/WarmRedundancyPooledWireTests.cs new file mode 100644 index 000000000..c6ea042be --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/WarmRedundancyPooledWireTests.cs @@ -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; + +/// +/// 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. +/// +[CollectionDefinition(Name, DisableParallelization = true)] +public sealed class WarmRedundancyProcessWideCollection +{ + /// The collection name. + public const string Name = "warm-redundancy-process-wide"; +} + +/// +/// 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), standby.projected_notifications and +/// standby.suppressed_alarm_events, each selected by instrument INSTANCE (#503). +/// +[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; + + /// + /// SDK 2.0.1 returns acknowledged (and evicted) notification payloads to a process-wide pool, and + /// Initialize() 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 + /// (i=2267): 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). + /// + [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.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(); + 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("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); + } + + /// Sums every measurement of ONE counter instance. + 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((_, value, _, _) => Interlocked.Add(ref _total, value)); + _listener.Start(); + } + + public long Total => Interlocked.Read(ref _total); + + public void Dispose() => _listener.Dispose(); + } + + /// + /// standby.projected_notifications counts each notification the Standby rewrites, at Publish and again at + /// Republish (each is a delivery), and nothing while Active. + /// + [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.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"); + } + + /// + /// standby.suppressed_alarm_events counts one per new alarm event withheld on the Standby, and none once + /// Active (where the same transition is delivered). + /// + [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); + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/WarmRedundancyWireTests.Deferred.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/WarmRedundancyWireTests.Deferred.cs new file mode 100644 index 000000000..08e599d6f --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/WarmRedundancyWireTests.Deferred.cs @@ -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; + +/// +/// 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 +/// ), and an ACL-stood-down item across a transition +/// (T8). +/// +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 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]; + } + + /// + /// 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. + /// + [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(); + body.TryGetValue(out var eu).ShouldBeTrue(); + return eu!; + } + + // ------------------------------------------------------------------ T7 item deleted / reused between queue and publish + + /// + /// 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. + /// + [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 + + /// + /// 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). + /// + [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); + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/WarmRedundancyWireTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/WarmRedundancyWireTests.cs index c401f3a53..ee84f9026 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/WarmRedundancyWireTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/WarmRedundancyWireTests.cs @@ -16,7 +16,7 @@ namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests; /// in the spec §4.7 step order; the actor that owns that order has its own tests. /// [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 StartModeAsync(string name, string mode, CancellationToken ct) => + internal static Task 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 ReadAsync(ISession session, NodeId node, CancellationToken ct, uint attribute = Attributes.Value, + internal static async Task 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 TranslateAsync(ISession session, NodeId start, string path, CancellationToken ct) + internal static async Task 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 /// 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. - 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 AddItemAsync(SessionClient client, uint sub, NodeId node, uint handle, uint queue, CancellationToken ct, + internal static async Task 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 PublishDataAsync(SessionClient client, CancellationToken ct) + internal static async Task 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> CollectAsync(SessionClient client, CancellationToken ct, TimeSpan window) + internal static async Task> CollectAsync(SessionClient client, CancellationToken ct, TimeSpan window) { var collected = new List<(uint, DataValue)>(); var until = DateTime.UtcNow + window; diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/WarmStandbyCloneMatrixTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/WarmStandbyCloneMatrixTests.cs new file mode 100644 index 000000000..7ced37f27 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/WarmStandbyCloneMatrixTests.cs @@ -0,0 +1,209 @@ +using Opc.Ua; +using Shouldly; +using Xunit; + +namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests; + +/// The SDK's notification-payload pool is process-wide; the recycled-payload test must not race another +/// test renting from it. +[CollectionDefinition(Name, DisableParallelization = true)] +public sealed class SdkPayloadPoolCollection +{ + /// The collection name. + public const string Name = "sdk-payload-pool"; +} + +/// +/// Gitea #717 T7, the payload matrix the D4 rewrite meets at delivery: SDK 2.0.1 hands each Publish and Republish a +/// CoreUtils.Clone of the cached message, which may carry several s (with +/// s), an and a +/// 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. +/// +[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; + } + + /// 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. + private static NotificationMessage Cached() => new() + { + SequenceNumber = 42, + PublishTime = Server, + NotificationData = new List + { + new(new DataChangeNotification + { + MonitoredItems = new List + { + Marked(1, 1.5), + new() { ClientHandle = 2, Value = Good(2.5) }, // i=2267 or a condition child: unmarked + }, + DiagnosticInfos = new List + { + new() { SymbolicId = 11, AdditionalInfo = "first" }, + new() { SymbolicId = 12, AdditionalInfo = "second" }, + }, + }), + new(new EventNotificationList + { + Events = new List + { + new() { ClientHandle = 9, EventFields = new List { new("event-id"), new("message") } }, + }, + }), + new(new StatusChangeNotification { Status = StatusCodes.BadTimeout }), + new(new DataChangeNotification + { + MonitoredItems = new List { Marked(3, 3.5) }, + }), + }, + }; + + private static T Body(NotificationMessage m, int index) where T : class, IEncodeable + { + m.NotificationData[index].TryGetValue(out var body).ShouldBeTrue($"payload {index} is a {typeof(T).Name}"); + return body!; + } + + private static MonitoredItemNotification Item(NotificationMessage m, int payload, int item) => + Body(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(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(projected, 1).IsEqual(Body(cached, 1)).ShouldBeTrue("events pass through"); + Body(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(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(decoded, 0).DiagnosticInfos.Count.ShouldBe(2); + } + + /// A Republish is a second clone of the same cached message: projecting it gives the same bytes. + [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)); + } + + /// A message with no marked value comes back as the same instance (nothing rebuilt, nothing counted). + [Fact] + public void A_message_with_only_events_status_and_unmarked_values_is_returned_as_is() + { + var cached = Cached(); + cached.NotificationData = new List { cached.NotificationData[1], cached.NotificationData[2] }; + var delivered = CoreUtils.Clone(cached)!; + ReferenceEquals(StandbyNotificationRewriter.Project(delivered, out var rewritten), delivered).ShouldBeTrue(); + rewritten.ShouldBe(0); + } + + /// + /// The SDK's own pool, not a hand-made stale marker: a payload stamped for an ordinary tag and returned (as + /// SentMessageQueue 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. + /// + [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? recycled = null; + for (var i = 0; i <= PooledEncodeableType.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("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 + { + new(new DataChangeNotification { MonitoredItems = new List { 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(); + } + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/WarmRedundancyMetricsTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/WarmRedundancyMetricsTests.cs new file mode 100644 index 000000000..a5026b4d9 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/WarmRedundancyMetricsTests.cs @@ -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; + +/// +/// 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. +/// +[CollectionDefinition(Name, DisableParallelization = true)] +public sealed class WarmRedundancyMetricsCollection +{ + /// The collection name. + public const string Name = "warm-redundancy-metrics"; +} + +/// +/// Gitea #717 T10: each Runtime-side Warm instrument fires on its own path — warm_transition with its +/// direction/reason tags, the warm_role gauge, and snapshot_rejected per reason from both +/// subscribers ( and ). Listeners select by +/// instrument INSTANCE, never by meter name (#503). mode_mismatch lives in +/// . +/// +[Collection(WarmRedundancyMetricsCollection.Name)] +public sealed class WarmRedundancyMetricsTests : RuntimeActorTestBase +{ + private static readonly NodeId Other = NodeId.Parse("10.255.255.1:9"); + + private readonly ConcurrentQueue _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); + + /// Records every measurement of ONE instrument, with its tags. + private sealed class Recorder : IDisposable where T : struct + { + private readonly MeterListener _listener = new(); + private readonly ConcurrentQueue<(T Value, Dictionary Tags)> _seen = new(); + + public Recorder(Instrument instrument) + { + _listener.InstrumentPublished = (published, l) => + { + if (ReferenceEquals(published, instrument)) l.EnableMeasurementEvents(published); + }; + _listener.SetMeasurementEventCallback((_, value, tags, _) => + { + var copy = new Dictionary(); + foreach (var tag in tags) copy[tag.Key] = tag.Value; + _seen.Enqueue((value, copy)); + }); + _listener.Start(); + } + + public IReadOnlyList<(T Value, Dictionary 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(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(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(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); + } + + /// + /// The 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. + /// + [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(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 _published = []; + private readonly Lock _gate = new(); + + public IReadOnlyList 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); + } + } +}