From 1e71a25e3db2f78f7175f6dd44054b01019a5546 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Tue, 13 Oct 2026 14:29:00 -0400 Subject: [PATCH] fix: age DB-less ServiceLevel snapshots periodically --- .../OpcUa/OpcUaPublishActor.cs | 68 +++++--- .../OpcUa/OpcUaPublishActorFreshnessTests.cs | 162 ++++++++++++++++++ 2 files changed, 204 insertions(+), 26 deletions(-) create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorFreshnessTests.cs diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/OpcUa/OpcUaPublishActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/OpcUa/OpcUaPublishActor.cs index dfd1de73..eb28bdf7 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/OpcUa/OpcUaPublishActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/OpcUa/OpcUaPublishActor.cs @@ -147,6 +147,7 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers private readonly TimeSpan _staleWindow; private readonly TimeSpan _probeFreshnessWindow; private readonly TimeSpan _healthTickInterval; + private readonly Func _utcNow; private readonly Akka.Cluster.Cluster _cluster = Akka.Cluster.Cluster.Get(Context.System); private readonly ILoggingAdapter _log = Context.GetLogger(); @@ -200,9 +201,10 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers /// The window beyond which a peer's OPC UA probe verdict about /// this node is considered stale (and thus given the benefit of the doubt rather than demoting); /// defaults to 30 seconds. - /// The period between self-driven DB-health refresh ticks (each - /// Asks for its cached status); defaults to 5 seconds. No timer is - /// started when is null. + /// The period between freshness ticks; defaults to 5 seconds. + /// DB-backed nodes Ask the probe for cached status; DB-less nodes recompute snapshot freshness. + /// Legacy nodes without a probe do not start a timer. + /// The UTC clock used for freshness; defaults to . /// Per-cluster mesh Phase 4: this is a driver-only node with no ConfigDb, so the /// ServiceLevel DB-reachability input is retired (DbReachable treated constant-true, only /// staleness demotes). No is wired on such a node. @@ -219,7 +221,8 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers TimeSpan? probeFreshnessWindow = null, TimeSpan? healthTickInterval = null, bool dbLess = false, - IDataPlaneAuthoritySink? dataPlaneAuthority = null) => + IDataPlaneAuthoritySink? dataPlaneAuthority = null, + Func? utcNow = null) => Akka.Actor.Props.Create(() => new OpcUaPublishActor( sink ?? NullOpcUaAddressSpaceSink.Instance, serviceLevel ?? NullServiceLevelPublisher.Instance, @@ -232,7 +235,8 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers probeFreshnessWindow, healthTickInterval, dbLess, - dataPlaneAuthority)).WithDispatcher(DispatcherId); + dataPlaneAuthority, + utcNow)).WithDispatcher(DispatcherId); /// Test-only Props that omits the pinned-dispatcher requirement and skips the /// DPS subscribe so unit tests can spin up the actor on a vanilla TestKit cluster. @@ -249,9 +253,10 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers /// The window beyond which a peer's OPC UA probe verdict about /// this node is considered stale (and thus given the benefit of the doubt rather than demoting); /// defaults to 30 seconds. - /// The period between self-driven DB-health refresh ticks (each - /// Asks for its cached status); defaults to 5 seconds. No timer is - /// started when is null. + /// The period between freshness ticks; defaults to 5 seconds. + /// DB-backed nodes Ask the probe for cached status; DB-less nodes recompute snapshot freshness. + /// Legacy nodes without a probe do not start a timer. + /// The UTC clock used for freshness; defaults to . /// Per-cluster mesh Phase 4: this is a driver-only node with no ConfigDb, so the /// ServiceLevel DB-reachability input is retired (DbReachable treated constant-true, only /// staleness demotes). No is wired on such a node. @@ -269,7 +274,8 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers TimeSpan? probeFreshnessWindow = null, TimeSpan? healthTickInterval = null, bool dbLess = false, - IDataPlaneAuthoritySink? dataPlaneAuthority = null) => + IDataPlaneAuthoritySink? dataPlaneAuthority = null, + Func? utcNow = null) => Akka.Actor.Props.Create(() => new OpcUaPublishActor( sink ?? NullOpcUaAddressSpaceSink.Instance, serviceLevel ?? NullServiceLevelPublisher.Instance, @@ -282,7 +288,8 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers probeFreshnessWindow, healthTickInterval, dbLess, - dataPlaneAuthority)); + dataPlaneAuthority, + utcNow)); /// Initializes a new instance of the class. /// The OPC UA address space sink. @@ -298,9 +305,10 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers /// The window beyond which a peer's OPC UA probe verdict about /// this node is considered stale (and thus given the benefit of the doubt rather than demoting); /// defaults to 30 seconds. - /// The period between self-driven DB-health refresh ticks (each - /// Asks for its cached status); defaults to 5 seconds. No timer is - /// started when is null. + /// The period between freshness ticks; defaults to 5 seconds. + /// DB-backed nodes Ask the probe for cached status; DB-less nodes recompute snapshot freshness. + /// Legacy nodes without a probe do not start a timer. + /// The UTC clock used for freshness; defaults to . /// Per-cluster mesh Phase 4: this is a driver-only node with no ConfigDb, so the /// ServiceLevel DB-reachability input is retired (DbReachable treated constant-true, only /// staleness demotes). No is wired on such a node. @@ -316,7 +324,8 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers TimeSpan? probeFreshnessWindow = null, TimeSpan? healthTickInterval = null, bool dbLess = false, - IDataPlaneAuthoritySink? dataPlaneAuthority = null) + IDataPlaneAuthoritySink? dataPlaneAuthority = null, + Func? utcNow = null) { _sink = sink; _dataPlaneAuthority = dataPlaneAuthority; @@ -330,6 +339,7 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers _staleWindow = staleWindow ?? TimeSpan.FromSeconds(30); _probeFreshnessWindow = probeFreshnessWindow ?? TimeSpan.FromSeconds(30); _healthTickInterval = healthTickInterval ?? TimeSpan.FromSeconds(5); + _utcNow = utcNow ?? (() => DateTime.UtcNow); Receive(HandleAttributeUpdate); Receive(HandleAttributeBatchUpdate); @@ -353,11 +363,10 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers DistributedPubSub.Get(Context.System).Mediator.Tell(new Subscribe(RedundancyStateTopic, Self)); } - // Production delivery of DB-health: when a probe is wired, kick an immediate refresh + start the - // periodic tick so _lastDbHealth gets populated and kept fresh without any external pump. Gated - // ONLY on the probe being present (independent of the DPS subscribe) so the calculator path works - // in tests too. No probe → stay on the legacy role-only seam (no timer). - if (_dbHealthProbe is not null) + // DB-backed nodes refresh their cached probe status; DB-less nodes still need periodic + // snapshot/probe-verdict age checks even when no new messages arrive. Legacy nodes keep + // their role-only behavior without a timer. + if (_dbLess || _dbHealthProbe is not null) { Self.Tell(HealthTick.Instance); // immediate first refresh Timers.StartPeriodicTimer("db-health", HealthTick.Instance, _healthTickInterval); @@ -1004,7 +1013,8 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers RecomputeServiceLevel(); } - /// Periodic self-tick: Asks the local for its cached + /// Periodic self-tick: recomputes freshness directly when no DB probe exists; otherwise + /// Asks the local for its cached /// status and pipes the reply back to , where /// caches it + recomputes. A hung/late probe (Ask timeout) pipes a Reachable=false status so /// the node FAIL-SAFE-DEMOTES rather than freezing the last-known-good. The continuation only @@ -1013,11 +1023,17 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers /// touched here. private void OnHealthTick() { - _dbHealthProbe!.Ask( + if (_dbHealthProbe is null) + { + RecomputeServiceLevel(); + return; + } + + _dbHealthProbe.Ask( DbHealthProbeActor.GetStatus.Instance, TimeSpan.FromSeconds(1)) .ContinueWith(t => t.IsCompletedSuccessfully ? t.Result - : new DbHealthProbeActor.DbHealthStatus(false, DateTime.UtcNow, "db-health ask timeout")) + : new DbHealthProbeActor.DbHealthStatus(false, _utcNow(), "db-health ask timeout")) .PipeTo(Self); } @@ -1030,7 +1046,7 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers { // The result targets the probed node. If it isn't me, it's about a peer — ignore it. if (_localNode is null || r.NodeId != _localNode.Value) return; - _probeAboutMe = (r.Ok, DateTime.UtcNow); + _probeAboutMe = (r.Ok, _utcNow()); RecomputeServiceLevel(); } @@ -1042,7 +1058,7 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers private bool OpcUaProbeOk() { if (_probeAboutMe is not { } verdict) return true; - if (DateTime.UtcNow - verdict.At > _probeFreshnessWindow) return true; + if (_utcNow() - verdict.At > _probeFreshnessWindow) return true; return verdict.Ok; } @@ -1079,7 +1095,7 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers // wins, so a DB-less Detached node stays 0. if (_dbLess) { - var utcNow = DateTime.UtcNow; + var utcNow = _utcNow(); var dbLessInputs = new NodeHealthInputs( MemberState: SafeSelfStatus(), DbReachable: true, @@ -1099,7 +1115,7 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers return; } - var now = DateTime.UtcNow; + var now = _utcNow(); var inputs = new NodeHealthInputs( MemberState: SafeSelfStatus(), DbReachable: _lastDbHealth.Reachable, diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorFreshnessTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorFreshnessTests.cs new file mode 100644 index 00000000..347279c6 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorFreshnessTests.cs @@ -0,0 +1,162 @@ +using System.Collections.Concurrent; +using Akka.Actor; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Redundancy; +using ZB.MOM.WW.OtOpcUa.Commons.Types; +using ZB.MOM.WW.OtOpcUa.Commons.OpcUa; +using ZB.MOM.WW.OtOpcUa.OpcUaServer; +using ZB.MOM.WW.OtOpcUa.Runtime.OpcUa; +using ZB.MOM.WW.OtOpcUa.Runtime.Health; +using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.OpcUa; + +public sealed class OpcUaPublishActorFreshnessTests : RuntimeActorTestBase +{ + [Theory] + [InlineData(false, 240, 200)] + [InlineData(true, 250, 210)] + public void DbLess_timer_demotes_without_incoming_messages(bool primary, byte fresh, byte stale) + { + var publisher = new RecordingPublisher(); + var clock = new ControlledClock(); + var local = NodeId.Parse("freshness-node"); + var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests( + serviceLevel: publisher, localNode: local, dbLess: true, + staleWindow: TimeSpan.FromSeconds(1), healthTickInterval: TimeSpan.FromMilliseconds(25), + utcNow: clock.Read)); + actor.Tell(Snapshot(local, primary, clock.Now)); + AwaitAssert(() => publisher.Levels.ShouldBe(new[] { fresh }), PresenceBudget); + // Equality remains fresh. Observe several real ticks with a fixed clock; no actor Tell. + clock.Advance(TimeSpan.FromSeconds(1)); + var reads = clock.Reads; + AwaitCondition(() => clock.Reads >= reads + 4, PresenceBudget); + // Drain earlier self-enqueued publications without triggering a health recomputation. + actor.Tell(new Identify("freshness-barrier"), TestActor); + ExpectMsg().Subject.ShouldBe(actor); + publisher.Levels.ShouldBe(new[] { fresh }); + clock.Advance(TimeSpan.FromTicks(1)); + // No further Tell: only the production timer can notice age crossing the window. + AwaitAssert(() => publisher.Levels.ShouldBe(new[] { fresh, stale }), PresenceBudget); + // A new fresh snapshot restores service, after which ticks continue to age that snapshot. + actor.Tell(Snapshot(local, primary, clock.Now)); + AwaitAssert(() => publisher.Levels.ShouldBe(new[] { fresh, stale, fresh }), PresenceBudget); + clock.Advance(TimeSpan.FromSeconds(1) + TimeSpan.FromTicks(1)); + AwaitAssert(() => publisher.Levels.ShouldBe(new[] { fresh, stale, fresh, stale }), PresenceBudget); + } + + [Fact] + public void DbLess_timer_expires_negative_peer_verdict_only_after_exact_boundary() + { + var publisher = new RecordingPublisher(); + var clock = new ControlledClock(); + var local = NodeId.Parse("peer-verdict-node"); + var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests( + serviceLevel: publisher, localNode: local, dbLess: true, + staleWindow: TimeSpan.FromSeconds(10), probeFreshnessWindow: TimeSpan.FromSeconds(2), + healthTickInterval: TimeSpan.FromMilliseconds(25), utcNow: clock.Read)); + actor.Tell(Snapshot(local, false, clock.Now)); + AwaitAssert(() => publisher.Levels.ShouldBe(new byte[] { 240 }), PresenceBudget); + actor.Tell(new PeerOpcUaProbeActor.OpcUaProbeResult(local, Ok: false)); + AwaitAssert(() => publisher.Levels.ShouldBe(new byte[] { 240, 0 }), PresenceBudget); + clock.Advance(TimeSpan.FromSeconds(2)); + var reads = clock.Reads; + AwaitCondition(() => clock.Reads >= reads + 4, PresenceBudget); + // Drain earlier self-enqueued publications without triggering a health recomputation. + actor.Tell(new Identify("freshness-barrier"), TestActor); + ExpectMsg().Subject.ShouldBe(actor); + publisher.Levels.ShouldBe(new byte[] { 240, 0 }); + clock.Advance(TimeSpan.FromTicks(1)); + AwaitAssert(() => publisher.Levels.ShouldBe(new byte[] { 240, 0, 240 }), PresenceBudget); + } + + [Fact] + public void Stopping_DbLess_actor_cancels_freshness_timer_and_publications() + { + var publisher = new RecordingPublisher(); + var clock = new ControlledClock(); + var local = NodeId.Parse("stopped-node"); + var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests( + serviceLevel: publisher, localNode: local, dbLess: true, + staleWindow: TimeSpan.FromSeconds(1), healthTickInterval: TimeSpan.FromMilliseconds(25), + utcNow: clock.Read)); + actor.Tell(Snapshot(local, false, clock.Now)); + AwaitAssert(() => publisher.Levels.ShouldBe(new byte[] { 240 }), PresenceBudget); + var reads = clock.Reads; + AwaitCondition(() => clock.Reads >= reads + 4, PresenceBudget); + Watch(actor); + Sys.Stop(actor); + ExpectTerminated(actor); + reads = clock.Reads; + clock.Advance(TimeSpan.FromSeconds(2)); + ExpectNoMsg(TimeSpan.FromMilliseconds(150)); // Six timer periods after confirmed termination. + clock.Reads.ShouldBe(reads); + publisher.Levels.ShouldBe(new byte[] { 240 }); + } + + [Fact] + public void DbBacked_timer_refreshes_probe_status_periodically() + { + var publisher = new RecordingPublisher(); + var clock = new ControlledClock(); + var local = NodeId.Parse("db-backed-periodic-node"); + var db = CreateTestProbe(); + var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests( + serviceLevel: publisher, localNode: local, dbHealthProbe: db.Ref, + healthTickInterval: TimeSpan.FromMilliseconds(100), utcNow: clock.Read)); + actor.Tell(Snapshot(local, false, clock.Now)); + db.ExpectMsg(); + db.Reply(new DbHealthProbeActor.DbHealthStatus(true, clock.Now, null)); + AwaitAssert(() => publisher.Levels.ShouldContain((byte)240), PresenceBudget); + // A second request comes from the periodic timer, not the immediate startup tick. + db.ExpectMsg(); + db.Reply(new DbHealthProbeActor.DbHealthStatus(false, clock.Now, "unreachable")); + AwaitAssert(() => publisher.Levels.Last().ShouldBe((byte)100), PresenceBudget); + } + + [Fact] + public void DbBacked_timer_ask_timeout_demotes_last_known_healthy_node() + { + var publisher = new RecordingPublisher(); + var clock = new ControlledClock(); + var local = NodeId.Parse("db-backed-timeout-node"); + var db = CreateTestProbe(); + var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests( + serviceLevel: publisher, localNode: local, dbHealthProbe: db.Ref, + healthTickInterval: TimeSpan.FromMilliseconds(100), utcNow: clock.Read)); + actor.Tell(Snapshot(local, true, clock.Now)); + db.ExpectMsg(); + db.Reply(new DbHealthProbeActor.DbHealthStatus(true, clock.Now, null)); + AwaitAssert(() => publisher.Levels.ShouldContain((byte)250), PresenceBudget); + // Leave subsequent real Ask requests unanswered: the timeout status must reach the actor. + db.ExpectMsg(); + AwaitAssert(() => publisher.Levels.Last().ShouldBe((byte)110), PresenceBudget); + } + + private sealed class ControlledClock + { + private long _ticks = new DateTime(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc).Ticks; + private int _reads; + public DateTime Now => new(Interlocked.Read(ref _ticks), DateTimeKind.Utc); + public int Reads => Volatile.Read(ref _reads); + public DateTime Read() + { + Interlocked.Increment(ref _reads); + return Now; + } + public void Advance(TimeSpan age) => Interlocked.Add(ref _ticks, age.Ticks); + } + + private static RedundancyStateChanged Snapshot(NodeId local, bool primary, DateTime asOf) => + new(new[] { new NodeRedundancyState(local, + primary ? RedundancyRole.Primary : RedundancyRole.Secondary, + IsClusterLeader: primary, IsDriverPrimary: primary, asOf) }, CorrelationId.NewId()); + + private sealed class RecordingPublisher : IServiceLevelPublisher + { + private readonly ConcurrentQueue _levels = new(); + public byte[] Levels => _levels.ToArray(); + public void Publish(byte serviceLevel) => _levels.Enqueue(serviceLevel); + } +}