fix(r2-10): in-process resilience status store

This commit is contained in:
Joseph Doherty
2026-07-13 10:43:47 -04:00
parent 12b2ce078b
commit 99ce5b401e
5 changed files with 158 additions and 1 deletions
@@ -8,7 +8,7 @@
{ "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": "completed", "blockedBy": ["T1", "T4"] },
{ "id": "T6", "subject": "Host DI: register the publisher in AddOtOpcUaDriverFactories + ResilienceStatusReaderRegistrationTests wiring guard (tracker HAS a production reader)", "status": "completed", "blockedBy": ["T5"] },
{ "id": "T7", "subject": "AdminUI: IDriverResilienceStatusStore + InMemory impl + AddOtOpcUaDriverStatusServices registration (TDD mirroring DriverStatusSnapshotStoreTests)", "status": "pending", "blockedBy": ["T4"] },
{ "id": "T7", "subject": "AdminUI: IDriverResilienceStatusStore + InMemory impl + AddOtOpcUaDriverStatusServices registration (TDD mirroring DriverStatusSnapshotStoreTests)", "status": "completed", "blockedBy": ["T4"] },
{ "id": "T8", "subject": "AdminUI: DriverResilienceStatusBridge DPS actor + spawn in WithOtOpcUaSignalRBridges", "status": "pending", "blockedBy": ["T7"] },
{ "id": "T9", "subject": "AdminUI: resilience chip section in DriverStatusPanel.razor (in-process store read; no bUnit — verified by T10)", "status": "pending", "blockedBy": ["T7"] },
{ "id": "T10", "subject": "Live-/run gate on docker-dev: rebuild BOTH centrals, S7 dead-endpoint breaker-open, both-replica chips, recovery clear, staleness dim", "status": "pending", "blockedBy": ["T6", "T8", "T9"] },
@@ -28,6 +28,7 @@ public static class HubServiceCollectionExtensions
public static IServiceCollection AddOtOpcUaDriverStatusServices(this IServiceCollection services)
{
services.AddSingleton<IDriverStatusSnapshotStore, InMemoryDriverStatusSnapshotStore>();
services.AddSingleton<IDriverResilienceStatusStore, InMemoryDriverResilienceStatusStore>();
services.AddSingleton(typeof(IInProcessBroadcaster<>), typeof(InProcessBroadcaster<>));
return services;
}
@@ -0,0 +1,36 @@
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Drivers;
namespace ZB.MOM.WW.OtOpcUa.AdminUI.Hubs;
/// <summary>
/// Singleton last-snapshot-per-<c>(instance, host)</c> cache of driver resilience status
/// (breaker open / consecutive failures / in-flight depth). Populated by
/// <c>DriverResilienceStatusBridge</c> as it forwards <c>driver-resilience-status</c> DPS
/// messages; read in-process by the driver status panel's resilience section via
/// <see cref="SnapshotChanged"/> + <see cref="GetForInstance"/>. Mirrors
/// <see cref="IDriverStatusSnapshotStore"/> — the panel reads this singleton directly instead of
/// opening a self-targeted SignalR connection (which a server-side Blazor component cannot reach
/// behind a reverse proxy).
/// </summary>
public interface IDriverResilienceStatusStore
{
/// <summary>Stores or replaces the last-known resilience snapshot for a <c>(instance, host)</c> pair.</summary>
/// <param name="snapshot">The resilience status snapshot to store.</param>
void Upsert(DriverResilienceStatusChanged snapshot);
/// <summary>Returns every stored host snapshot for one driver instance (empty if none).</summary>
/// <param name="driverInstanceId">The driver instance's stable logical ID.</param>
/// <returns>The per-host resilience snapshots for that instance.</returns>
IReadOnlyList<DriverResilienceStatusChanged> GetForInstance(string driverInstanceId);
/// <summary>Returns a point-in-time snapshot of every tracked <c>(instance, host)</c> pair.</summary>
/// <returns>A read-only collection of the last-known resilience snapshot for each pair.</returns>
IReadOnlyCollection<DriverResilienceStatusChanged> GetAll();
/// <summary>
/// Raised after every <see cref="Upsert"/> with the just-stored snapshot. Lets the Blazor Server
/// driver status panel receive live updates by reading this singleton directly. Handlers run on
/// the caller's thread (the bridge actor), so subscribers must marshal to their own sync context.
/// </summary>
event Action<DriverResilienceStatusChanged>? SnapshotChanged;
}
@@ -0,0 +1,33 @@
using System.Collections.Concurrent;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Drivers;
namespace ZB.MOM.WW.OtOpcUa.AdminUI.Hubs;
/// <summary>
/// Thread-safe in-memory implementation of <see cref="IDriverResilienceStatusStore"/>.
/// Keyed by <c>(DriverInstanceId, HostName)</c>; last write wins.
/// </summary>
public sealed class InMemoryDriverResilienceStatusStore : IDriverResilienceStatusStore
{
private readonly ConcurrentDictionary<(string Instance, string Host), DriverResilienceStatusChanged> _byPair = new();
/// <inheritdoc />
public event Action<DriverResilienceStatusChanged>? SnapshotChanged;
/// <inheritdoc />
public void Upsert(DriverResilienceStatusChanged snapshot)
{
_byPair[(snapshot.DriverInstanceId, snapshot.HostName)] = snapshot;
// Capture-then-invoke so a concurrent unsubscribe can't null the delegate mid-raise.
SnapshotChanged?.Invoke(snapshot);
}
/// <inheritdoc />
public IReadOnlyList<DriverResilienceStatusChanged> GetForInstance(string driverInstanceId) =>
_byPair.Values
.Where(s => string.Equals(s.DriverInstanceId, driverInstanceId, StringComparison.Ordinal))
.ToList();
/// <inheritdoc />
public IReadOnlyCollection<DriverResilienceStatusChanged> GetAll() => _byPair.Values.ToArray();
}
@@ -0,0 +1,87 @@
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.AdminUI.Hubs;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Drivers;
namespace ZB.MOM.WW.OtOpcUa.AdminUI.Tests;
/// <summary>
/// Covers the in-process push contract the Blazor Server driver status panel's resilience section
/// relies on: <see cref="IDriverResilienceStatusStore.SnapshotChanged"/> fires on every
/// <see cref="IDriverResilienceStatusStore.Upsert"/>, keyed per <c>(instance, host)</c> with
/// last-write-wins, and <c>GetForInstance</c> returns only the queried instance's hosts. Mirrors
/// <see cref="DriverStatusSnapshotStoreTests"/>; the panel reads this store directly (no
/// self-targeted SignalR connection).
/// </summary>
public sealed class DriverResilienceStatusStoreTests
{
private static DriverResilienceStatusChanged Msg(
string instance, string host, bool breakerOpen = false, int failures = 0) =>
new(instance, host, breakerOpen, failures, 0, null,
new DateTime(2026, 7, 13, 0, 0, 0, DateTimeKind.Utc),
new DateTime(2026, 7, 13, 0, 0, 0, DateTimeKind.Utc));
[Fact]
public void Upsert_raises_SnapshotChanged_with_the_stored_snapshot()
{
var store = new InMemoryDriverResilienceStatusStore();
var received = new List<DriverResilienceStatusChanged>();
store.SnapshotChanged += received.Add;
var msg = Msg("drv-1", "host-a", breakerOpen: true);
store.Upsert(msg);
received.Count.ShouldBe(1);
received[0].ShouldBeSameAs(msg);
}
[Fact]
public void Upsert_is_last_write_wins_per_instance_host()
{
var store = new InMemoryDriverResilienceStatusStore();
store.Upsert(Msg("drv-1", "host-a", failures: 3));
store.Upsert(Msg("drv-1", "host-a", failures: 0));
var hosts = store.GetForInstance("drv-1");
hosts.Count.ShouldBe(1);
hosts[0].ConsecutiveFailures.ShouldBe(0);
}
[Fact]
public void GetForInstance_returns_only_the_queried_instances_hosts()
{
var store = new InMemoryDriverResilienceStatusStore();
store.Upsert(Msg("drv-1", "host-a"));
store.Upsert(Msg("drv-1", "host-b"));
store.Upsert(Msg("drv-2", "host-a"));
var hosts = store.GetForInstance("drv-1");
hosts.Count.ShouldBe(2);
hosts.ShouldAllBe(h => h.DriverInstanceId == "drv-1");
}
[Fact]
public void GetForInstance_returns_empty_for_unknown_instance()
{
var store = new InMemoryDriverResilienceStatusStore();
store.Upsert(Msg("drv-1", "host-a"));
store.GetForInstance("nope").ShouldBeEmpty();
}
[Fact]
public void Unsubscribed_handler_stops_receiving_after_removal()
{
var store = new InMemoryDriverResilienceStatusStore();
var count = 0;
void Handler(DriverResilienceStatusChanged _) => count++;
store.SnapshotChanged += Handler;
store.Upsert(Msg("drv-1", "host-a"));
store.SnapshotChanged -= Handler;
store.Upsert(Msg("drv-1", "host-a"));
count.ShouldBe(1);
}
}