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

fix(r2-10): periodic DPS publisher for tracker snapshots

Browse Source
This commit is contained in:
Joseph Doherty
2026-07-13 10:40:54 -04:00
parent 2cb3f962b9
commit 37f180c5a3
3 changed files with 150 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
archreview/plans/R2-10-resilience-observability-plan.md.tasks.json
+1 -1
View File
@@ -6,7 +6,7 @@
{ "id": "T2", "subject": "Builder: OnClosed records breaker-close on the tracker (TDD in DriverResiliencePipelineBuilderTests)", "status": "completed", "blockedBy": ["T1"] },
{ "id": "T3", "subject": "Invoker: successful Execute* resets ConsecutiveFailures (TDD in CapabilityInvokerTests)", "status": "completed", "blockedBy": ["T1"] },
{ "id": "T4", "subject": "Commons: DriverResilienceStatusChanged record + driver-resilience-status TopicName", "status": "completed", "blockedBy": [] },
{ "id": "T5", "subject": "Host: DriverResilienceStatusPublisherService (BuildMessages mapping + 5s DPS timer shell; TDD mapping in Host.IntegrationTests)", "status": "pending", "blockedBy": ["T1", "T4"] },
{ "id": "T5", "subject": "Host: DriverResilienceStatusPublisherService (BuildMessages mapping + 5s DPS timer shell; TDD mapping in Host.IntegrationTests)", "status": "completed", "blockedBy": ["T1", "T4"] },
{ "id": "T6", "subject": "Host DI: register the publisher in AddOtOpcUaDriverFactories + ResilienceStatusReaderRegistrationTests wiring guard (tracker HAS a production reader)", "status": "pending", "blockedBy": ["T5"] },
{ "id": "T7", "subject": "AdminUI: IDriverResilienceStatusStore + InMemory impl + AddOtOpcUaDriverStatusServices registration (TDD mirroring DriverStatusSnapshotStoreTests)", "status": "pending", "blockedBy": ["T4"] },
{ "id": "T8", "subject": "AdminUI: DriverResilienceStatusBridge DPS actor + spawn in WithOtOpcUaSignalRBridges", "status": "pending", "blockedBy": ["T7"] },
src/Server/ZB.MOM.WW.OtOpcUa.Host/Drivers/DriverResilienceStatusPublisherService.cs
+93
View File
@@ -0,0 +1,93 @@
using Akka.Actor;
using Akka.Cluster.Tools.PublishSubscribe;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Drivers;
using ZB.MOM.WW.OtOpcUa.Core.Resilience;
namespace ZB.MOM.WW.OtOpcUa.Host.Drivers;
/// <summary>
/// The operator-facing reader for <see cref="DriverResilienceStatusTracker"/>. Runs on driver-role
/// nodes (registered next to the tracker it reads) and every <see cref="_interval"/> publishes the
/// full tracker snapshot — one <see cref="DriverResilienceStatusChanged"/> per <c>(instance, host)</c>
/// — to the <c>driver-resilience-status</c> DistributedPubSub topic. The AdminUI
/// <c>DriverResilienceStatusBridge</c> subscribes and feeds the resilience section of the driver
/// status panel. Mirrors the driver-health flow; publishing the full set every tick self-heals
/// late-subscribing admin nodes and gives the panel staleness detection for free.
/// </summary>
public sealed class DriverResilienceStatusPublisherService : BackgroundService
{
private static readonly TimeSpan DefaultInterval = TimeSpan.FromSeconds(5);
private readonly DriverResilienceStatusTracker _tracker;
private readonly Func<ActorSystem> _actorSystemAccessor;
private readonly ILogger<DriverResilienceStatusPublisherService> _logger;
private readonly TimeSpan _interval;
/// <summary>Initializes a new instance of <see cref="DriverResilienceStatusPublisherService"/>.</summary>
/// <param name="tracker">The process-singleton resilience-status tracker to read each tick.</param>
/// <param name="actorSystemAccessor">Lazy accessor for the Akka <see cref="ActorSystem"/> whose DPS mediator publishes the snapshots.</param>
/// <param name="logger">Logger for publish diagnostics.</param>
/// <param name="interval">Publish cadence; defaults to 5 s when null.</param>
public DriverResilienceStatusPublisherService(
DriverResilienceStatusTracker tracker,
Func<ActorSystem> actorSystemAccessor,
ILogger<DriverResilienceStatusPublisherService> logger,
TimeSpan? interval = null)
{
_tracker = tracker;
_actorSystemAccessor = actorSystemAccessor;
_logger = logger;
_interval = interval ?? DefaultInterval;
}
/// <summary>
/// Map the tracker's current snapshot to one wire message per <c>(instance, host)</c> pair.
/// Pure + unit-testable; the timer/publish shell in <see cref="ExecuteAsync"/> calls this each tick.
/// </summary>
/// <param name="tracker">The tracker to read.</param>
/// <param name="publishedUtc">The publish timestamp stamped on every emitted message.</param>
/// <returns>One <see cref="DriverResilienceStatusChanged"/> per tracked pair (empty when the tracker is empty).</returns>
public static IReadOnlyList<DriverResilienceStatusChanged> BuildMessages(
DriverResilienceStatusTracker tracker, DateTime publishedUtc) =>
tracker.Snapshot()
.Select(entry => new DriverResilienceStatusChanged(
entry.DriverInstanceId,
entry.HostName,
entry.Snapshot.IsBreakerOpen,
entry.Snapshot.ConsecutiveFailures,
entry.Snapshot.CurrentInFlight,
entry.Snapshot.LastBreakerOpenUtc,
entry.Snapshot.LastSampledUtc,
publishedUtc))
.ToList();
/// <inheritdoc />
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
using var timer = new PeriodicTimer(_interval);
while (await timer.WaitForNextTickAsync(stoppingToken).ConfigureAwait(false))
{
try
{
var messages = BuildMessages(_tracker, DateTime.UtcNow);
if (messages.Count == 0)
continue;
var mediator = DistributedPubSub.Get(_actorSystemAccessor()).Mediator;
foreach (var message in messages)
mediator.Tell(new Publish(DriverResilienceStatusChanged.TopicName, message));
}
catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
{
break;
}
catch (Exception ex)
{
// A transient DPS / actor-system hiccup must never kill the publisher — the next tick retries.
_logger.LogWarning(ex, "DriverResilienceStatusPublisherService: publish tick failed; retrying next interval");
}
}
}
}
tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/DriverResilienceStatusPublisherServiceTests.cs
+56
View File
@@ -0,0 +1,56 @@
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Resilience;
using ZB.MOM.WW.OtOpcUa.Host.Drivers;
namespace ZB.MOM.WW.OtOpcUa.Host.IntegrationTests;
/// <summary>
/// Pure-DI (no fixture) coverage of the publisher's snapshot→message mapping — the only unit-testable
/// seam of <see cref="DriverResilienceStatusPublisherService"/>. The DPS <c>Publish</c> shell is
/// verified by the R2-10 live gate; this asserts the mapping the shell publishes.
/// </summary>
public sealed class DriverResilienceStatusPublisherServiceTests
{
private static readonly DateTime Published = new(2026, 7, 13, 9, 0, 0, DateTimeKind.Utc);
private static readonly DateTime Sampled = new(2026, 7, 13, 8, 59, 0, DateTimeKind.Utc);
[Fact]
public void BuildMessages_EmptyTracker_ReturnsEmpty()
{
var tracker = new DriverResilienceStatusTracker();
var messages = DriverResilienceStatusPublisherService.BuildMessages(tracker, Published);
messages.ShouldBeEmpty();
}
[Fact]
public void BuildMessages_MapsTrackerSnapshot_OneMessagePerInstanceHost()
{
var tracker = new DriverResilienceStatusTracker();
// (drv-1, host-a): an open breaker with a standing failure count.
tracker.RecordFailure("drv-1", "host-a", Sampled);
tracker.RecordFailure("drv-1", "host-a", Sampled);
tracker.RecordBreakerOpen("drv-1", "host-a", Sampled);
// (drv-1, host-b): a healthy host with in-flight calls.
tracker.RecordCallStart("drv-1", "host-b");
var messages = DriverResilienceStatusPublisherService.BuildMessages(tracker, Published);
messages.Count.ShouldBe(2);
var open = messages.Single(m => m.HostName == "host-a");
open.DriverInstanceId.ShouldBe("drv-1");
open.BreakerOpen.ShouldBeTrue();
open.ConsecutiveFailures.ShouldBe(2);
open.LastBreakerOpenUtc.ShouldBe(Sampled);
open.PublishedUtc.ShouldBe(Published);
var healthy = messages.Single(m => m.HostName == "host-b");
healthy.BreakerOpen.ShouldBeFalse();
healthy.ConsecutiveFailures.ShouldBe(0);
healthy.CurrentInFlight.ShouldBe(1);
healthy.PublishedUtc.ShouldBe(Published);
}
}