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

fix: age DB-less ServiceLevel snapshots periodically

Browse Source
This commit is contained in:
Joseph Doherty
2026-10-13 14:29:00 -04:00
parent 4f7a07fbd3
commit 1e71a25e3d
2 changed files with 204 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Server/ZB.MOM.WW.OtOpcUa.Runtime/OpcUa/OpcUaPublishActor.cs
+42 -26
View File
@@ -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<DateTime> _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
/// <param name="probeFreshnessWindow">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.</param>
/// <param name="healthTickInterval">The period between self-driven DB-health refresh ticks (each
/// Asks <paramref name="dbHealthProbe"/> for its cached status); defaults to 5 seconds. No timer is
/// started when <paramref name="dbHealthProbe"/> is null.</param>
/// <param name="healthTickInterval">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.</param>
/// <param name="utcNow">The UTC clock used for freshness; defaults to <see cref="DateTime.UtcNow"/>.</param>
/// <param name="dbLess">Per-cluster mesh Phase 4: this is a driver-only node with no ConfigDb, so the
/// ServiceLevel DB-reachability input is retired (<c>DbReachable</c> treated constant-true, only
/// staleness demotes). No <paramref name="dbHealthProbe"/> is wired on such a node.</param>
@@ -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<DateTime>? 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);
/// <summary>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.</summary>
@@ -249,9 +253,10 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
/// <param name="probeFreshnessWindow">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.</param>
/// <param name="healthTickInterval">The period between self-driven DB-health refresh ticks (each
/// Asks <paramref name="dbHealthProbe"/> for its cached status); defaults to 5 seconds. No timer is
/// started when <paramref name="dbHealthProbe"/> is null.</param>
/// <param name="healthTickInterval">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.</param>
/// <param name="utcNow">The UTC clock used for freshness; defaults to <see cref="DateTime.UtcNow"/>.</param>
/// <param name="dbLess">Per-cluster mesh Phase 4: this is a driver-only node with no ConfigDb, so the
/// ServiceLevel DB-reachability input is retired (<c>DbReachable</c> treated constant-true, only
/// staleness demotes). No <paramref name="dbHealthProbe"/> is wired on such a node.</param>
@@ -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<DateTime>? 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));
/// <summary>Initializes a new instance of the <see cref="OpcUaPublishActor"/> class.</summary>
/// <param name="sink">The OPC UA address space sink.</param>
@@ -298,9 +305,10 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
/// <param name="probeFreshnessWindow">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.</param>
/// <param name="healthTickInterval">The period between self-driven DB-health refresh ticks (each
/// Asks <paramref name="dbHealthProbe"/> for its cached status); defaults to 5 seconds. No timer is
/// started when <paramref name="dbHealthProbe"/> is null.</param>
/// <param name="healthTickInterval">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.</param>
/// <param name="utcNow">The UTC clock used for freshness; defaults to <see cref="DateTime.UtcNow"/>.</param>
/// <param name="dbLess">Per-cluster mesh Phase 4: this is a driver-only node with no ConfigDb, so the
/// ServiceLevel DB-reachability input is retired (<c>DbReachable</c> treated constant-true, only
/// staleness demotes). No <paramref name="dbHealthProbe"/> is wired on such a node.</param>
@@ -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<DateTime>? 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<AttributeValueUpdate>(HandleAttributeUpdate);
Receive<AttributeValueBatchUpdate>(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();
}
/// <summary>Periodic self-tick: Asks the local <see cref="DbHealthProbeActor"/> for its cached
/// <summary>Periodic self-tick: recomputes freshness directly when no DB probe exists; otherwise
/// Asks the local <see cref="DbHealthProbeActor"/> for its cached
/// status and pipes the reply back to <see cref="Self"/>, where <see cref="HandleDbHealthStatus"/>
/// caches it + recomputes. A hung/late probe (Ask timeout) pipes a <c>Reachable=false</c> 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.</summary>
private void OnHealthTick()
{
_dbHealthProbe!.Ask<DbHealthProbeActor.DbHealthStatus>(
if (_dbHealthProbe is null)
{
RecomputeServiceLevel();
return;
}
_dbHealthProbe.Ask<DbHealthProbeActor.DbHealthStatus>(
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,
tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorFreshnessTests.cs
+162
View File
@@ -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<ActorIdentity>().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<ActorIdentity>().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<DbHealthProbeActor.GetStatus>();
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<DbHealthProbeActor.GetStatus>();
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<DbHealthProbeActor.GetStatus>();
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<DbHealthProbeActor.GetStatus>();
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<byte> _levels = new();
public byte[] Levels => _levels.ToArray();
public void Publish(byte serviceLevel) => _levels.Enqueue(serviceLevel);
}
}