feat(hosts): surface per-host connectivity on /hosts; drop the unwritable table (#521)
`IHostConnectivityProbe` was a dead surface: eleven drivers implement it, `GetHostStatuses()` had ZERO production call sites, and `OnHostStatusChanged` had no subscriber outside the Galaxy driver's own aggregator. Per-host connectivity was computed by every driver and read by nobody. The issue offered "build the publisher or delete it". The publisher as its entity doc described it — driver nodes upserting `DriverHostStatus` rows — is not buildable: per-cluster mesh Phase 4 gates `AddOtOpcUaConfigDb` on the `admin` role, so a driver-only node has no ConfigDb connection to write rows with. So the capability is kept and the transport changed. `DriverHealthChanged.HostStatuses` now carries the probe result to `/hosts` as a Hosts column. That channel already reached the page, already survives the mesh split via the Phase 5 gRPC telemetry stream, and already replays a last-value snapshot on re-subscribe — so per-host state re-primes after a reconnect without a durable store. Both halves of the interface finally do what they are for: the event triggers a prompt publish, the pull is the source of truth. The point of the column is the case the driver-level state chip structurally cannot express: a multi-device driver stays aggregate-Healthy while ONE of its devices is unreachable. Two traps, both pinned by tests that were falsified against the prod code: - The host digest MUST be in the publish fingerprint. On a single-host-down transition every other fingerprint component is unchanged, so the dedup would swallow exactly the publish carrying the news — the trap that already bit the rediscovery signal. Removing it turns the guard test red, verified. - null (no probe) must stay distinct from empty (probe with no hosts). proto3 cannot tell an absent repeated field from an empty one, hence the explicit `has_host_statuses` flag; collapsing them would render every probe-less driver as one whose devices are all fine. Dropped: the DriverHostStatus entity, enum, DbSet, model config and table (migration DropDriverHostStatusTable — empty on every deployment, so the scaffolder's data-loss warning is moot, and Down() recreates it exactly). Found en route, NOT fixed here: `DriverInstanceResilienceStatus` is the identical defect — no writer, no reader, only a DbSet declaration, while the live data rides the `driver-resilience-status` telemetry channel. Its doc-comment now states that rather than describing the sampler and AdminUI join that were never built. Filed as #524 rather than widening this schema change beyond what was asked. Claude-Session: https://claude.ai/code/session_015p7wGqy3YpZNCpDzTpGMKo
This commit is contained in:
Joseph Doherty
+298
@@ -0,0 +1,298 @@
|
||||
using Akka.Actor;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
using ZB.MOM.WW.OtOpcUa.Runtime.Drivers;
|
||||
using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers;
|
||||
|
||||
/// <summary>
|
||||
/// Covers the <see cref="IHostConnectivityProbe"/> consumer (Gitea #521). Eleven drivers implement the
|
||||
/// capability; before this wiring <c>GetHostStatuses()</c> had ZERO production call sites and
|
||||
/// <c>OnHostStatusChanged</c> had no subscriber outside the Galaxy driver's own aggregator, so per-host
|
||||
/// connectivity was computed by every driver and read by nobody.
|
||||
/// <para><b>Why it does not go to the DB.</b> The <c>DriverHostStatus</c> table's doc-comment described a
|
||||
/// publisher hosted service that upserted rows from each driver node. It was never built, and
|
||||
/// per-cluster mesh Phase 4 made it unbuildable as described — <c>AddOtOpcUaConfigDb</c> is gated on the
|
||||
/// <c>admin</c> role, so a driver-only node has no ConfigDb connection. The table was dropped; the data
|
||||
/// rides the driver-health snapshot instead.</para>
|
||||
/// </summary>
|
||||
[Trait("Category", "Unit")]
|
||||
public sealed class DriverInstanceActorHostStatusTests : RuntimeActorTestBase
|
||||
{
|
||||
/// <summary>The base case: a probe driver's hosts reach the health publisher.</summary>
|
||||
[Fact]
|
||||
public void Host_statuses_reach_the_health_publisher()
|
||||
{
|
||||
var driver = new ProbeStubDriver();
|
||||
driver.SetHost("plc-a", HostState.Running);
|
||||
driver.SetHost("plc-b", HostState.Running);
|
||||
var publisher = new RecordingHealthPublisher();
|
||||
var actor = SpawnDriverActor(driver, publisher);
|
||||
|
||||
actor.Tell(new DriverInstanceActor.InitializeRequested("{}"));
|
||||
|
||||
AwaitAssert(
|
||||
() =>
|
||||
{
|
||||
var latest = publisher.Published.LastOrDefault();
|
||||
latest.ShouldNotBeNull();
|
||||
latest!.HostStatuses.ShouldNotBeNull();
|
||||
latest.HostStatuses!.Select(h => h.HostName).OrderBy(n => n, StringComparer.Ordinal)
|
||||
.ShouldBe(["plc-a", "plc-b"]);
|
||||
},
|
||||
TimeSpan.FromSeconds(3));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// <b>The load-bearing case, and the entire reason this channel exists.</b> A multi-device driver
|
||||
/// stays aggregate-<c>Healthy</c> when ONE of its devices drops — the driver-level state chip cannot
|
||||
/// express it. So the transition must reach the operator through the per-host detail.
|
||||
/// <para>This is also the dedup trap that already bit the rediscovery signal once.
|
||||
/// <c>PublishHealthSnapshot</c> suppresses a publish whose fingerprint repeats, and on this
|
||||
/// transition (state, lastSuccessfulRead, lastError, errorCount) are ALL unchanged. Unless the
|
||||
/// host-status digest is part of the fingerprint, the dedup swallows exactly the publish that
|
||||
/// carries the news.</para>
|
||||
/// <para><b>Falsifiability:</b> the assertion is that the publish count STRICTLY INCREASES across the
|
||||
/// transition — an "eventually shows Stopped" assertion would be satisfied by the warm-up publish
|
||||
/// plus a later 30 s heartbeat and would prove nothing. Drop <c>HostStatusDigest</c> from the
|
||||
/// fingerprint tuple in <c>DriverInstanceActor</c> and this test must go red. Verified by doing so.</para>
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void A_single_host_going_down_is_not_swallowed_by_the_unchanged_health_dedup()
|
||||
{
|
||||
var driver = new ProbeStubDriver();
|
||||
driver.SetHost("plc-a", HostState.Running);
|
||||
driver.SetHost("plc-b", HostState.Running);
|
||||
var publisher = new RecordingHealthPublisher();
|
||||
var actor = SpawnDriverActor(driver, publisher);
|
||||
|
||||
actor.Tell(new DriverInstanceActor.InitializeRequested("{}"));
|
||||
AwaitAssert(() => publisher.Published.Count.ShouldBeGreaterThan(0), TimeSpan.FromSeconds(3));
|
||||
|
||||
// Settle, so the baseline is a quiet actor: from here only the host state changes. The driver's
|
||||
// OWN health stays Healthy throughout — that is the point.
|
||||
ExpectNoMsg(TimeSpan.FromMilliseconds(200));
|
||||
var before = publisher.Published.Count;
|
||||
|
||||
driver.SetHost("plc-b", HostState.Stopped, raise: true);
|
||||
|
||||
AwaitAssert(
|
||||
() => publisher.Published.Count.ShouldBeGreaterThan(before),
|
||||
TimeSpan.FromSeconds(3));
|
||||
|
||||
var latest = publisher.Published[^1];
|
||||
latest.Health.State.ShouldBe(DriverState.Healthy, "the driver itself never faulted — only one of its devices did");
|
||||
latest.HostStatuses.ShouldNotBeNull();
|
||||
latest.HostStatuses!.Single(h => h.HostName == "plc-b").State.ShouldBe(HostState.Stopped);
|
||||
latest.HostStatuses.Single(h => h.HostName == "plc-a").State.ShouldBe(HostState.Running);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The other half of the dedup contract: when nothing changes, the digest must NOT churn. A digest
|
||||
/// built over an <c>IReadOnlyList</c> by reference, or one sensitive to the order a driver happens to
|
||||
/// enumerate its hosts in, would differ on every call and re-publish on every 30 s heartbeat forever
|
||||
/// — turning the dedup off without anyone noticing.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void Unchanged_hosts_do_not_defeat_the_dedup()
|
||||
{
|
||||
var driver = new ProbeStubDriver();
|
||||
driver.SetHost("plc-a", HostState.Running);
|
||||
driver.SetHost("plc-b", HostState.Running);
|
||||
var publisher = new RecordingHealthPublisher();
|
||||
var actor = SpawnDriverActor(driver, publisher);
|
||||
|
||||
actor.Tell(new DriverInstanceActor.InitializeRequested("{}"));
|
||||
AwaitAssert(() => publisher.Published.Count.ShouldBeGreaterThan(0), TimeSpan.FromSeconds(3));
|
||||
ExpectNoMsg(TimeSpan.FromMilliseconds(200));
|
||||
var before = publisher.Published.Count;
|
||||
|
||||
// The driver re-shuffles its host order without changing any state. A real driver builds this list
|
||||
// from a Dictionary, so enumeration order is not guaranteed stable between calls.
|
||||
driver.ReverseHostOrder();
|
||||
// Poke the actor into re-publishing without changing anything material.
|
||||
driver.RaiseHostStatusChanged();
|
||||
|
||||
ExpectNoMsg(TimeSpan.FromMilliseconds(500));
|
||||
publisher.Published.Count.ShouldBe(before, "an order flip with no state change must be deduped, not re-published");
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// <b>Leak guard.</b> The <see cref="IDriver"/> can OUTLIVE the actor — the host respawns a child
|
||||
/// around the same driver object — so a missing <c>-=</c> in <c>PostStop</c> accumulates one handler
|
||||
/// per respawn, each holding a dead <c>Self</c>.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void Stopping_the_actor_detaches_the_host_status_handler()
|
||||
{
|
||||
var driver = new ProbeStubDriver();
|
||||
var parent = CreateTestProbe();
|
||||
parent.IgnoreMessages(_ => true);
|
||||
var actor = parent.ChildActorOf(DriverInstanceActor.Props(driver));
|
||||
|
||||
actor.Tell(new DriverInstanceActor.InitializeRequested("{}"));
|
||||
AwaitAssert(() => driver.SubscriberCount.ShouldBe(1), TimeSpan.FromSeconds(3));
|
||||
|
||||
Watch(actor);
|
||||
actor.Tell(PoisonPill.Instance);
|
||||
ExpectTerminated(actor, TimeSpan.FromSeconds(3));
|
||||
|
||||
driver.SubscriberCount.ShouldBe(0);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A driver with no probe publishes null host statuses — NOT an empty list. The two mean different
|
||||
/// things at the UI ("no per-host detail available" vs "a probe that currently knows no hosts") and
|
||||
/// collapsing them would render a driver with no probe as one whose devices are all fine.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void A_driver_without_a_probe_publishes_null_host_statuses()
|
||||
{
|
||||
var publisher = new RecordingHealthPublisher();
|
||||
var actor = SpawnDriverActor(new StubDriver(), publisher);
|
||||
|
||||
actor.Tell(new DriverInstanceActor.InitializeRequested("{}"));
|
||||
AwaitAssert(() => publisher.Published.Count.ShouldBeGreaterThan(0), TimeSpan.FromSeconds(3));
|
||||
|
||||
publisher.Published.ShouldAllBe(p => p.HostStatuses == null);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A probe that throws must not take the health publish down with it. <c>GetHostStatuses()</c> is
|
||||
/// documented as a pure in-memory snapshot (which is why the capability analyzer exempts it from
|
||||
/// the guarded-call rule), but a driver is free to violate that, and losing the whole health channel
|
||||
/// for one misbehaving probe would be a much worse outcome than losing the host detail.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void A_throwing_probe_degrades_to_null_without_killing_the_health_publish()
|
||||
{
|
||||
var driver = new ProbeStubDriver { ThrowOnGetHostStatuses = true };
|
||||
var publisher = new RecordingHealthPublisher();
|
||||
var actor = SpawnDriverActor(driver, publisher);
|
||||
|
||||
actor.Tell(new DriverInstanceActor.InitializeRequested("{}"));
|
||||
|
||||
AwaitAssert(() => publisher.Published.Count.ShouldBeGreaterThan(0), TimeSpan.FromSeconds(3));
|
||||
publisher.Published[^1].Health.State.ShouldBe(DriverState.Healthy);
|
||||
publisher.Published[^1].HostStatuses.ShouldBeNull();
|
||||
}
|
||||
|
||||
private IActorRef SpawnDriverActor(IDriver driver, IDriverHealthPublisher publisher)
|
||||
{
|
||||
var parent = CreateTestProbe();
|
||||
parent.IgnoreMessages(_ => true);
|
||||
return parent.ChildActorOf(DriverInstanceActor.Props(driver, healthPublisher: publisher));
|
||||
}
|
||||
|
||||
/// <summary>Captures every health publish so a test can assert on the host-status field.</summary>
|
||||
private sealed record HealthPublish(
|
||||
DriverHealth Health,
|
||||
IReadOnlyList<HostConnectivityStatus>? HostStatuses);
|
||||
|
||||
private sealed class RecordingHealthPublisher : IDriverHealthPublisher
|
||||
{
|
||||
private readonly List<HealthPublish> _published = [];
|
||||
|
||||
/// <summary>Thread-safe snapshot — <c>Publish</c> runs on the actor thread while the test asserts
|
||||
/// from its own.</summary>
|
||||
public IReadOnlyList<HealthPublish> Published
|
||||
{
|
||||
get { lock (_published) return _published.ToArray(); }
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public void Publish(
|
||||
string clusterId,
|
||||
string driverInstanceId,
|
||||
DriverHealth health,
|
||||
int errorCount5Min,
|
||||
DateTime? rediscoveryNeededUtc = null,
|
||||
string? rediscoveryReason = null,
|
||||
IReadOnlyList<HostConnectivityStatus>? hostStatuses = null)
|
||||
{
|
||||
lock (_published) _published.Add(new HealthPublish(health, hostStatuses));
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A stub driver exposing <see cref="IHostConnectivityProbe"/>.
|
||||
/// <para><b>Its own health is deliberately STABLE</b> (a fixed last-read timestamp), for the same
|
||||
/// reason <c>RediscoverableStubDriver</c> is: the shared <c>StubDriver</c> returns
|
||||
/// <c>DateTime.UtcNow</c> from <c>GetHealth()</c>, so its fingerprint differs on every call, the
|
||||
/// dedup never engages, and any test built on it passes vacuously whether or not the fix is
|
||||
/// present.</para>
|
||||
/// </summary>
|
||||
private sealed class ProbeStubDriver : IDriver, IHostConnectivityProbe
|
||||
{
|
||||
private static readonly DateTime FixedLastRead = new(2026, 7, 30, 12, 0, 0, DateTimeKind.Utc);
|
||||
private static readonly DateTime FixedChangedAt = new(2026, 7, 30, 11, 0, 0, DateTimeKind.Utc);
|
||||
|
||||
private readonly List<HostConnectivityStatus> _hosts = [];
|
||||
|
||||
/// <summary>When set, <see cref="GetHostStatuses"/> throws — the misbehaving-probe case.</summary>
|
||||
public bool ThrowOnGetHostStatuses { get; init; }
|
||||
|
||||
/// <inheritdoc />
|
||||
public event EventHandler<HostStatusChangedEventArgs>? OnHostStatusChanged;
|
||||
|
||||
/// <inheritdoc />
|
||||
public string DriverInstanceId => "probe-stub-1";
|
||||
|
||||
/// <inheritdoc />
|
||||
public string DriverType => "Stub";
|
||||
|
||||
/// <summary>Number of live subscribers on <see cref="OnHostStatusChanged"/>.</summary>
|
||||
public int SubscriberCount => OnHostStatusChanged?.GetInvocationList().Length ?? 0;
|
||||
|
||||
/// <summary>Adds or updates a host, optionally raising the transition event as a real probe loop does.</summary>
|
||||
public void SetHost(string hostName, HostState state, bool raise = false)
|
||||
{
|
||||
lock (_hosts)
|
||||
{
|
||||
var index = _hosts.FindIndex(h => h.HostName == hostName);
|
||||
var entry = new HostConnectivityStatus(hostName, state, FixedChangedAt);
|
||||
if (index >= 0) _hosts[index] = entry;
|
||||
else _hosts.Add(entry);
|
||||
}
|
||||
|
||||
if (raise) RaiseHostStatusChanged();
|
||||
}
|
||||
|
||||
/// <summary>Flips enumeration order without changing any host's state.</summary>
|
||||
public void ReverseHostOrder()
|
||||
{
|
||||
lock (_hosts) _hosts.Reverse();
|
||||
}
|
||||
|
||||
/// <summary>Raises the event exactly as a real probe loop does.</summary>
|
||||
public void RaiseHostStatusChanged()
|
||||
=> OnHostStatusChanged?.Invoke(this, new HostStatusChangedEventArgs("plc-b", HostState.Running, HostState.Stopped));
|
||||
|
||||
/// <inheritdoc />
|
||||
public IReadOnlyList<HostConnectivityStatus> GetHostStatuses()
|
||||
{
|
||||
if (ThrowOnGetHostStatuses) throw new InvalidOperationException("probe is broken");
|
||||
lock (_hosts) return _hosts.ToArray();
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken) => Task.CompletedTask;
|
||||
|
||||
/// <inheritdoc />
|
||||
public Task ReinitializeAsync(string driverConfigJson, CancellationToken cancellationToken) => Task.CompletedTask;
|
||||
|
||||
/// <inheritdoc />
|
||||
public Task ShutdownAsync(CancellationToken cancellationToken) => Task.CompletedTask;
|
||||
|
||||
/// <inheritdoc />
|
||||
public DriverHealth GetHealth() => new(DriverState.Healthy, FixedLastRead, null);
|
||||
|
||||
/// <inheritdoc />
|
||||
public long GetMemoryFootprint() => 0;
|
||||
|
||||
/// <inheritdoc />
|
||||
public Task FlushOptionalCachesAsync(CancellationToken cancellationToken) => Task.CompletedTask;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user