feat(drivers): consume IRediscoverable as an operator re-browse prompt (§8.2, #518)
Nine drivers raise IRediscoverable.OnRediscoveryNeeded when they observe that a remote's tag set may have changed — a Galaxy redeploy, a TwinCAT symbol-version bump, an MTConnect agent restart, a Sparkplug rebirth. Nothing in src/ subscribed, so every raise went into the void. Drivers even wrap the invoke in try/catch for "subscriber threw"; there has never been a subscriber. DriverInstanceActor now attaches the event in PreStart and detaches in PostStop, mirroring AttachAlarmSource/DetachAlarmSource. Attach is per-actor, not per-connect: the raise has no connection affinity and can land while the driver is between connects. The detach is load-bearing rather than tidy — the IDriver instance outlives the actor when the host respawns a child around the same driver object, so a missing -= accumulates one handler per respawn, each holding a dead Self. The signal rides the existing driver-health snapshot to /hosts as a "re-browse" chip. It is ADVISORY: the served address space is deliberately not rebuilt, because v3 authors raw tags explicitly through the /raw browse-commit flow and a runtime graft would materialise nodes no operator approved and no deployment artifact records. Two things worth knowing: - PublishHealthSnapshot dedups on a health fingerprint, and a rediscovery raise on an otherwise-healthy driver changes none of the four fields it hashed. Without adding the timestamp to that tuple the dedup swallows the exact publish carrying the signal. Reverting that one line turns 3 of the 5 new tests red, which is how it was confirmed rather than assumed. - The new test stub implements IDriver from scratch instead of deriving from the shared StubDriver, whose GetHealth() returns DateTime.UtcNow — its fingerprint differs on every call, so the dedup never engages and the dedup test would have passed vacuously either way. The planned discovery-result drift detector was NOT built. Modbus, S7, MQTT, AbLegacy and Sql all echo authored config from DiscoverAsync rather than browsing the device, so a discovered-vs-authored diff is permanently empty for them — it would have been another plausible-looking inert seam, which is the class this audit exists to remove. IRediscoverable is the trustworthy signal because the driver asserts it deliberately. DriverHealthChanged, IDriverHealthPublisher.Publish and HostsDriverRow gain defaulted optional parameters, so no existing call site changed. telemetry.proto gains fields 8/9 and both Phase-5 mappers carry them. Runtime.Tests 512 passed / 31 skipped.
This commit is contained in:
Joseph Doherty
@@ -13,6 +13,19 @@ namespace ZB.MOM.WW.OtOpcUa.Commons.Messages.Drivers;
|
||||
/// <param name="LastError">Latest error message; null when none.</param>
|
||||
/// <param name="ErrorCount5Min">Number of state-transitions into Faulted in the last 5 minutes.</param>
|
||||
/// <param name="PublishedUtc">Timestamp this snapshot was published.</param>
|
||||
/// <param name="RediscoveryNeededUtc">
|
||||
/// When the driver last raised <c>IRediscoverable.OnRediscoveryNeeded</c> — i.e. it observed that the
|
||||
/// remote's tag set may have changed (a Galaxy redeploy, a TwinCAT symbol-version bump, an MTConnect
|
||||
/// agent restart, a Sparkplug rebirth). Null when the driver has never raised it, or does not
|
||||
/// implement <c>IRediscoverable</c>.
|
||||
/// <para><b>Advisory only.</b> The served address space is NOT rebuilt in response — raw tags are
|
||||
/// authored explicitly through the <c>/raw</c> browse-commit flow. This is a prompt for an operator
|
||||
/// to re-browse the device and commit whatever changed.</para>
|
||||
/// </param>
|
||||
/// <param name="RediscoveryReason">
|
||||
/// The driver-supplied reason string from the same event (e.g. <c>"deploy-time-changed"</c>), shown
|
||||
/// to the operator alongside the prompt. Null when <paramref name="RediscoveryNeededUtc"/> is null.
|
||||
/// </param>
|
||||
public sealed record DriverHealthChanged(
|
||||
string ClusterId,
|
||||
string DriverInstanceId,
|
||||
@@ -20,7 +33,9 @@ public sealed record DriverHealthChanged(
|
||||
DateTime? LastSuccessfulReadUtc,
|
||||
string? LastError,
|
||||
int ErrorCount5Min,
|
||||
DateTime PublishedUtc)
|
||||
DateTime PublishedUtc,
|
||||
DateTime? RediscoveryNeededUtc = null,
|
||||
string? RediscoveryReason = null)
|
||||
{
|
||||
/// <summary>
|
||||
/// DPS topic name. Both the runtime <c>AkkaDriverHealthPublisher</c> and the AdminUI
|
||||
|
||||
@@ -68,6 +68,8 @@ message DriverHealth {
|
||||
optional string last_error = 5; // nullable in the record
|
||||
int32 error_count_5min = 6;
|
||||
google.protobuf.Timestamp published_utc = 7;
|
||||
google.protobuf.Timestamp rediscovery_needed_utc = 8; // DateTime? — absent Timestamp encodes null
|
||||
optional string rediscovery_reason = 9; // nullable in the record
|
||||
}
|
||||
|
||||
// Mirrors ZB.MOM.WW.OtOpcUa.Commons.Messages.Drivers.DriverResilienceStatusChanged.
|
||||
|
||||
@@ -15,11 +15,20 @@ public interface IDriverHealthPublisher
|
||||
/// <param name="driverInstanceId">The driver instance the snapshot describes.</param>
|
||||
/// <param name="health">The current health snapshot.</param>
|
||||
/// <param name="errorCount5Min">The number of errors observed in the trailing 5-minute window.</param>
|
||||
/// <param name="rediscoveryNeededUtc">
|
||||
/// When the driver last raised <see cref="IRediscoverable.OnRediscoveryNeeded"/>, or null if never
|
||||
/// (or if it is not an <see cref="IRediscoverable"/>). Advisory: it prompts an operator to
|
||||
/// re-browse, and never rebuilds the served address space.
|
||||
/// </param>
|
||||
/// <param name="rediscoveryReason">The driver-supplied reason from that event; null when
|
||||
/// <paramref name="rediscoveryNeededUtc"/> is null.</param>
|
||||
void Publish(
|
||||
string clusterId,
|
||||
string driverInstanceId,
|
||||
DriverHealth health,
|
||||
int errorCount5Min);
|
||||
int errorCount5Min,
|
||||
DateTime? rediscoveryNeededUtc = null,
|
||||
string? rediscoveryReason = null);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
@@ -38,6 +47,8 @@ public sealed class NullDriverHealthPublisher : IDriverHealthPublisher
|
||||
string clusterId,
|
||||
string driverInstanceId,
|
||||
DriverHealth health,
|
||||
int errorCount5Min)
|
||||
int errorCount5Min,
|
||||
DateTime? rediscoveryNeededUtc = null,
|
||||
string? rediscoveryReason = null)
|
||||
{ /* no-op */ }
|
||||
}
|
||||
|
||||
@@ -191,6 +191,16 @@ else
|
||||
{
|
||||
<span class="mono small">@d.DriverInstanceId</span>
|
||||
}
|
||||
@if (d.RediscoveryNeededUtc is not null)
|
||||
{
|
||||
@* The driver reported its remote's tag set may have changed. Advisory
|
||||
only — v3 authors raw tags via /raw browse-commit, so nothing is
|
||||
grafted at runtime and an operator must re-browse to pick it up. *@
|
||||
<span class="chip chip-warn ms-1"
|
||||
title="@($"Reported {d.RediscoveryNeededUtc:u}: {d.RediscoveryReason ?? "tag set may have changed"}. The served address space is unchanged — re-browse this device under /raw and commit anything new.")">
|
||||
re-browse
|
||||
</span>
|
||||
}
|
||||
</td>
|
||||
<td>@(d.DriverType ?? "—")</td>
|
||||
<td><span class="chip @DriverChipClass(d.State)">@d.State</span></td>
|
||||
|
||||
@@ -35,9 +35,15 @@ public sealed record HostsDriverInstanceInfo(string DriverInstanceId, string Clu
|
||||
/// <param name="LastError">Latest error message; null when none.</param>
|
||||
/// <param name="ErrorCount5Min">Faulted-transition count in the last 5 minutes.</param>
|
||||
/// <param name="PublishedUtc">Timestamp the snapshot was published.</param>
|
||||
/// <param name="RediscoveryNeededUtc">When the driver last reported that the remote's tag set may have
|
||||
/// changed; null when never (or when the driver cannot report it). Advisory — the served address space
|
||||
/// is unchanged and an operator must re-browse the device via <c>/raw</c> to pick anything up.</param>
|
||||
/// <param name="RediscoveryReason">The driver-supplied reason for that report; null when
|
||||
/// <paramref name="RediscoveryNeededUtc"/> is null.</param>
|
||||
public sealed record HostsDriverRow(
|
||||
string DriverInstanceId, string? Name, string? DriverType, string State,
|
||||
DateTime? LastSuccessfulReadUtc, string? LastError, int ErrorCount5Min, DateTime PublishedUtc);
|
||||
DateTime? LastSuccessfulReadUtc, string? LastError, int ErrorCount5Min, DateTime PublishedUtc,
|
||||
DateTime? RediscoveryNeededUtc = null, string? RediscoveryReason = null);
|
||||
|
||||
/// <summary>
|
||||
/// One cluster's section on the <c>/hosts</c> page: its configured nodes plus its enriched
|
||||
@@ -110,7 +116,9 @@ public static class HostsDriverView
|
||||
s.LastSuccessfulReadUtc,
|
||||
s.LastError,
|
||||
s.ErrorCount5Min,
|
||||
s.PublishedUtc);
|
||||
s.PublishedUtc,
|
||||
s.RediscoveryNeededUtc,
|
||||
s.RediscoveryReason);
|
||||
})
|
||||
.OrderBy(d => d.Name ?? d.DriverInstanceId, StringComparer.OrdinalIgnoreCase)
|
||||
.ThenBy(d => d.DriverInstanceId, StringComparer.OrdinalIgnoreCase)
|
||||
|
||||
@@ -118,7 +118,9 @@ public static class TelemetryProtoMapCentral
|
||||
LastSuccessfulReadUtc: msg.LastSuccessfulReadUtc?.ToDateTime(),
|
||||
LastError: msg.HasLastError ? msg.LastError : null,
|
||||
ErrorCount5Min: msg.ErrorCount5Min,
|
||||
PublishedUtc: Required(msg.PublishedUtc, "DriverHealth", "published_utc"));
|
||||
PublishedUtc: Required(msg.PublishedUtc, "DriverHealth", "published_utc"),
|
||||
RediscoveryNeededUtc: msg.RediscoveryNeededUtc?.ToDateTime(),
|
||||
RediscoveryReason: msg.HasRediscoveryReason ? msg.RediscoveryReason : null);
|
||||
}
|
||||
|
||||
/// <summary>Projects a <see cref="DriverResilienceStatus"/> onto a <see cref="DriverResilienceStatusChanged"/>.</summary>
|
||||
|
||||
@@ -128,6 +128,10 @@ public static class TelemetryProtoMapNode
|
||||
msg.LastSuccessfulReadUtc = ToUtcTimestamp(e.LastSuccessfulReadUtc.Value);
|
||||
if (e.LastError is not null)
|
||||
msg.LastError = e.LastError;
|
||||
if (e.RediscoveryNeededUtc is not null)
|
||||
msg.RediscoveryNeededUtc = ToUtcTimestamp(e.RediscoveryNeededUtc.Value);
|
||||
if (e.RediscoveryReason is not null)
|
||||
msg.RediscoveryReason = e.RediscoveryReason;
|
||||
|
||||
return msg;
|
||||
}
|
||||
|
||||
@@ -30,7 +30,13 @@ public sealed class AkkaDriverHealthPublisher : IDriverHealthPublisher
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public void Publish(string clusterId, string driverInstanceId, DriverHealth health, int errorCount5Min)
|
||||
public void Publish(
|
||||
string clusterId,
|
||||
string driverInstanceId,
|
||||
DriverHealth health,
|
||||
int errorCount5Min,
|
||||
DateTime? rediscoveryNeededUtc = null,
|
||||
string? rediscoveryReason = null)
|
||||
{
|
||||
var msg = new DriverHealthChanged(
|
||||
clusterId,
|
||||
@@ -39,7 +45,9 @@ public sealed class AkkaDriverHealthPublisher : IDriverHealthPublisher
|
||||
health.LastSuccessfulRead,
|
||||
health.LastError,
|
||||
errorCount5Min,
|
||||
DateTime.UtcNow);
|
||||
DateTime.UtcNow,
|
||||
rediscoveryNeededUtc,
|
||||
rediscoveryReason);
|
||||
DistributedPubSub.Get(_system).Mediator.Tell(new Publish(TopicName, msg));
|
||||
// Phase 5: fan the same snapshot into the node-local live-telemetry hub (no-op until a gRPC
|
||||
// client subscribes). The DPS publish above is unchanged — the hub is a strictly additive tap.
|
||||
|
||||
@@ -139,6 +139,13 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
|
||||
/// </summary>
|
||||
public sealed record TriggerRediscovery;
|
||||
|
||||
/// <summary>Self-sent when the wrapped driver raises <see cref="IRediscoverable.OnRediscoveryNeeded"/> —
|
||||
/// it observed that the remote's tag set may have changed. Marshals the event off the driver's thread
|
||||
/// onto the actor thread. Handled in EVERY behaviour, including Stubbed and Reconnecting: the raise has no
|
||||
/// connection affinity (a Galaxy redeploy or a TwinCAT symbol-version bump can land while the driver is
|
||||
/// between connects), and dropping it in one state would lose the signal silently.</summary>
|
||||
private sealed record RediscoveryRaised(RediscoveryEventArgs Args);
|
||||
|
||||
/// <summary>Internal self-tick driving bounded post-connect re-discovery (FixedTree populates ~0–2s after connect).
|
||||
/// <paramref name="PreviousSignature"/> is the ordered-distinct full-reference signature of the prior pass's
|
||||
/// captured set (empty string on the first tick); re-discovery stops once a non-empty set repeats it.</summary>
|
||||
@@ -203,6 +210,16 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
|
||||
private ISubscriptionHandle? _subscriptionHandle;
|
||||
private EventHandler<DataChangeEventArgs>? _dataChangeHandler;
|
||||
private EventHandler<AlarmEventArgs>? _alarmEventHandler;
|
||||
private EventHandler<RediscoveryEventArgs>? _rediscoveryHandler;
|
||||
|
||||
/// <summary>When the driver last raised <see cref="IRediscoverable.OnRediscoveryNeeded"/>, and the reason
|
||||
/// it gave. Null until the first raise. Carried on every subsequent health snapshot so the AdminUI can
|
||||
/// prompt an operator to re-browse the device.
|
||||
/// <para>Deliberately <b>sticky</b> — it is not cleared on reconnect or on a later clean pass. The remote's
|
||||
/// tag set stayed changed; only an operator re-browsing and committing resolves it, and this actor cannot
|
||||
/// observe that happening. A redeploy respawns the child, which clears it naturally.</para></summary>
|
||||
private DateTime? _rediscoveryNeededUtc;
|
||||
private string? _rediscoveryReason;
|
||||
|
||||
/// <summary>The references the host wants kept subscribed (set by <see cref="SetDesiredSubscriptions"/>).
|
||||
/// Re-applied on every entry into <c>Connected</c> so values resume after a reconnect or redeploy.</summary>
|
||||
@@ -344,6 +361,9 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
|
||||
// Warm up the snapshot store immediately so AdminUI sees current state as soon as the
|
||||
// actor starts, before any state transition fires. Also start the periodic heartbeat so
|
||||
// long-lived Healthy drivers keep their snapshot fresh for newly-joined SignalR clients.
|
||||
// Attach the rediscovery signal before the first publish. Not per-connect: an IRediscoverable raise
|
||||
// has no connection affinity, and a driver can observe a remote change while disconnected.
|
||||
AttachRediscoverySource();
|
||||
PublishHealthSnapshot();
|
||||
Timers.StartPeriodicTimer("health-poll", HealthPollTick.Instance, _healthPollInterval);
|
||||
}
|
||||
@@ -365,6 +385,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
|
||||
Receive<RediscoverTick>(_ => { });
|
||||
// A TriggerRediscovery is meaningless to a stubbed (never-Connected) driver — silently ignore it.
|
||||
Receive<TriggerRediscovery>(_ => { });
|
||||
Receive<RediscoveryRaised>(HandleRediscoveryRaised);
|
||||
Receive<HealthPollTick>(_ => PublishHealthSnapshot());
|
||||
}
|
||||
|
||||
@@ -424,6 +445,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
|
||||
// A TriggerRediscovery arriving while not Connected is a deliberate no-op — the (re)connect path
|
||||
// re-runs discovery anyway. Swallow it so it stays a clean silent no-op (no Unhandled event).
|
||||
Receive<TriggerRediscovery>(_ => { });
|
||||
Receive<RediscoveryRaised>(HandleRediscoveryRaised);
|
||||
Receive<HealthPollTick>(_ => PublishHealthSnapshot());
|
||||
}
|
||||
|
||||
@@ -499,6 +521,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
|
||||
// to Self (HandleSubscribeAsync already logged the underlying cause). Swallow it so it doesn't dead-letter.
|
||||
Receive<SubscriptionFailed>(msg =>
|
||||
_log.Debug("DriverInstance {Id}: resubscribe reported failure: {Reason}", _driverInstanceId, msg.Reason));
|
||||
Receive<RediscoveryRaised>(HandleRediscoveryRaised);
|
||||
Receive<HealthPollTick>(_ =>
|
||||
{
|
||||
PublishHealthSnapshot();
|
||||
@@ -608,6 +631,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
|
||||
// A TriggerRediscovery arriving while not Connected is a deliberate no-op — the (re)connect path
|
||||
// re-runs discovery anyway. Swallow it so it stays a clean silent no-op (no Unhandled event).
|
||||
Receive<TriggerRediscovery>(_ => { });
|
||||
Receive<RediscoveryRaised>(HandleRediscoveryRaised);
|
||||
Receive<HealthPollTick>(_ => PublishHealthSnapshot());
|
||||
Timers.StartPeriodicTimer("retry-connect", RetryConnect.Instance, _reconnectInterval);
|
||||
}
|
||||
@@ -853,6 +877,44 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
|
||||
src.OnAlarmEvent += _alarmEventHandler;
|
||||
}
|
||||
|
||||
/// <summary>Subscribe the driver's <see cref="IRediscoverable.OnRediscoveryNeeded"/> (if it is one),
|
||||
/// marshaling each raise to the actor thread. Idempotent; mirrors <see cref="AttachAlarmSource"/>.
|
||||
/// <para>Attached once in <c>PreStart</c> rather than per-connect, because the interesting raises
|
||||
/// (a Galaxy redeploy, a TwinCAT symbol-version bump) can happen while the driver is between
|
||||
/// connects, and the event carries no connection affinity.</para></summary>
|
||||
private void AttachRediscoverySource()
|
||||
{
|
||||
if (_driver is not IRediscoverable src || _rediscoveryHandler is not null) return;
|
||||
var self = Self;
|
||||
_rediscoveryHandler = (_, e) => self.Tell(new RediscoveryRaised(e));
|
||||
src.OnRediscoveryNeeded += _rediscoveryHandler;
|
||||
}
|
||||
|
||||
/// <summary>Symmetric teardown, called from PostStop. Load-bearing: the <see cref="IDriver"/> instance
|
||||
/// can OUTLIVE this actor (the host respawns a child around the same driver object), so a missing
|
||||
/// unsubscribe would accumulate one handler per respawn, each holding a dead <c>Self</c>.</summary>
|
||||
private void DetachRediscoverySource()
|
||||
{
|
||||
if (_driver is IRediscoverable src && _rediscoveryHandler is not null)
|
||||
src.OnRediscoveryNeeded -= _rediscoveryHandler;
|
||||
_rediscoveryHandler = null;
|
||||
}
|
||||
|
||||
/// <summary>Records the driver's rediscovery raise and re-publishes health so the signal reaches the
|
||||
/// AdminUI promptly rather than waiting for the next 30 s heartbeat.
|
||||
/// <para><b>Advisory only.</b> The served address space is deliberately NOT rebuilt: v3 authors raw tags
|
||||
/// explicitly through the <c>/raw</c> browse-commit flow, so a runtime graft would create nodes no
|
||||
/// operator approved and that no deployment artifact records. This prompts a human to re-browse.</para></summary>
|
||||
private void HandleRediscoveryRaised(RediscoveryRaised msg)
|
||||
{
|
||||
_rediscoveryNeededUtc = DateTime.UtcNow;
|
||||
_rediscoveryReason = msg.Args.Reason;
|
||||
_log.Info(
|
||||
"DriverInstance {Id}: driver reports its tag set may have changed ({Reason}) — surfaced for operator re-browse; the served address space is unchanged",
|
||||
_driverInstanceId, msg.Args.Reason);
|
||||
PublishHealthSnapshot();
|
||||
}
|
||||
|
||||
/// <summary>Symmetric teardown — called from <see cref="DetachSubscription"/> and PostStop so a stale
|
||||
/// handler never pushes to a disconnected actor.</summary>
|
||||
private void DetachAlarmSource()
|
||||
@@ -1081,11 +1143,16 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
|
||||
{
|
||||
var health = _driver.GetHealth();
|
||||
var errorCount = ErrorCount5Min();
|
||||
var fingerprint = (health.State, health.LastSuccessfulRead, health.LastError, errorCount);
|
||||
// _rediscoveryNeededUtc is PART OF THE FINGERPRINT on purpose. A rediscovery raise on an
|
||||
// otherwise-unchanged Healthy driver leaves (state, lastSuccess, lastError, errorCount)
|
||||
// identical, so without it the dedup below would swallow the very publish that carries the
|
||||
// signal and the operator would never see the prompt.
|
||||
var fingerprint = (health.State, health.LastSuccessfulRead, health.LastError, errorCount, _rediscoveryNeededUtc);
|
||||
if (_lastPublishedFingerprint is { } prev && prev.Equals(fingerprint))
|
||||
return;
|
||||
_lastPublishedFingerprint = fingerprint;
|
||||
_healthPublisher.Publish(_clusterId, _driverInstanceId, health, errorCount);
|
||||
_healthPublisher.Publish(
|
||||
_clusterId, _driverInstanceId, health, errorCount, _rediscoveryNeededUtc, _rediscoveryReason);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
@@ -1094,12 +1161,15 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
|
||||
}
|
||||
|
||||
/// <summary>Fingerprint of the last <see cref="PublishHealthSnapshot"/> call; null until first publish.</summary>
|
||||
private (DriverState State, DateTime? LastSuccess, string? LastError, int ErrorCount)? _lastPublishedFingerprint;
|
||||
private (DriverState State, DateTime? LastSuccess, string? LastError, int ErrorCount, DateTime? RediscoveryNeededUtc)? _lastPublishedFingerprint;
|
||||
|
||||
/// <inheritdoc />
|
||||
protected override void PostStop()
|
||||
{
|
||||
DetachSubscription();
|
||||
// MUST happen: the IDriver instance can outlive this actor (the host respawns a child around the
|
||||
// same driver object), so a missing unsubscribe accumulates a handler per respawn holding a dead Self.
|
||||
DetachRediscoverySource();
|
||||
try { _driver.ShutdownAsync(CancellationToken.None).GetAwaiter().GetResult(); }
|
||||
catch (Exception ex) { _log.Warning(ex, "DriverInstance {Id}: ShutdownAsync threw on PostStop", _driverInstanceId); }
|
||||
OtOpcUaTelemetry.DriverInstanceLifecycle.Add(1,
|
||||
|
||||
+249
@@ -0,0 +1,249 @@
|
||||
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="IRediscoverable"/> consumer (Gitea #518). Nine drivers raise
|
||||
/// <see cref="IRediscoverable.OnRediscoveryNeeded"/> when they observe that a remote's tag set may have
|
||||
/// changed — a Galaxy redeploy, a TwinCAT symbol-version bump, an MTConnect agent restart, a Sparkplug
|
||||
/// rebirth. Before this wiring NOTHING in <c>src/</c> subscribed, so every one of those raises went into
|
||||
/// the void.
|
||||
/// <para><b>The signal is advisory.</b> It does not rebuild the served address space: v3 authors raw tags
|
||||
/// explicitly through the <c>/raw</c> browse-commit flow, so a runtime graft would create nodes no
|
||||
/// operator approved. It rides the driver-health snapshot to the AdminUI as a re-browse prompt.</para>
|
||||
/// </summary>
|
||||
[Trait("Category", "Unit")]
|
||||
public sealed class DriverInstanceActorRediscoverySignalTests : RuntimeActorTestBase
|
||||
{
|
||||
/// <summary>
|
||||
/// The load-bearing case: a driver raises rediscovery while otherwise perfectly healthy, and the
|
||||
/// signal reaches the health publisher carrying the driver's reason.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void Rediscovery_raise_reaches_the_health_publisher_with_its_reason()
|
||||
{
|
||||
var driver = new RediscoverableStubDriver();
|
||||
var publisher = new RecordingHealthPublisher();
|
||||
var parent = CreateTestProbe();
|
||||
parent.IgnoreMessages(_ => true);
|
||||
var actor = parent.ChildActorOf(DriverInstanceActor.Props(driver, healthPublisher: publisher));
|
||||
|
||||
actor.Tell(new DriverInstanceActor.InitializeRequested("{}"));
|
||||
AwaitAssert(() => publisher.Published.Count.ShouldBeGreaterThan(0), TimeSpan.FromSeconds(3));
|
||||
|
||||
driver.RaiseRediscovery("deploy-time-changed");
|
||||
|
||||
AwaitAssert(
|
||||
() =>
|
||||
{
|
||||
var withSignal = publisher.Published.LastOrDefault(p => p.RediscoveryNeededUtc is not null);
|
||||
withSignal.ShouldNotBeNull();
|
||||
withSignal!.RediscoveryReason.ShouldBe("deploy-time-changed");
|
||||
},
|
||||
TimeSpan.FromSeconds(3));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// <b>Regression guard for the dedup trap.</b> <c>PublishHealthSnapshot</c> suppresses a publish whose
|
||||
/// (state, lastSuccessfulRead, lastError, errorCount) tuple repeats. A rediscovery raise on an
|
||||
/// otherwise-unchanged Healthy driver leaves all four identical, so unless the rediscovery timestamp
|
||||
/// is part of the fingerprint the dedup swallows the very publish carrying the signal and the operator
|
||||
/// never sees the prompt.
|
||||
/// <para>Positive control: the assertion is that the count STRICTLY INCREASES across the raise. An
|
||||
/// "eventually non-null" assertion alone would pass on the warm-up publish and prove nothing. Revert
|
||||
/// <c>_rediscoveryNeededUtc</c> out of the fingerprint tuple and this test must fail.</para>
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void Rediscovery_raise_is_not_swallowed_by_the_unchanged_health_dedup()
|
||||
{
|
||||
var driver = new RediscoverableStubDriver();
|
||||
var publisher = new RecordingHealthPublisher();
|
||||
var parent = CreateTestProbe();
|
||||
parent.IgnoreMessages(_ => true);
|
||||
var actor = parent.ChildActorOf(DriverInstanceActor.Props(driver, healthPublisher: publisher));
|
||||
|
||||
actor.Tell(new DriverInstanceActor.InitializeRequested("{}"));
|
||||
AwaitAssert(() => publisher.Published.Count.ShouldBeGreaterThan(0), TimeSpan.FromSeconds(3));
|
||||
|
||||
// Let the actor settle so nothing else is in flight, then capture the baseline. Everything about
|
||||
// the driver's health is now stable — only the rediscovery flag will change.
|
||||
ExpectNoMsg(TimeSpan.FromMilliseconds(200));
|
||||
var before = publisher.Published.Count;
|
||||
|
||||
driver.RaiseRediscovery("symbol-version-changed");
|
||||
|
||||
AwaitAssert(
|
||||
() => publisher.Published.Count.ShouldBeGreaterThan(before),
|
||||
TimeSpan.FromSeconds(3));
|
||||
publisher.Published[^1].RediscoveryNeededUtc.ShouldNotBeNull();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The signal is sticky: a later health publish still carries it. The remote's tag set stayed
|
||||
/// changed, and only an operator re-browsing resolves that — which this actor cannot observe. A
|
||||
/// flag that cleared itself on the next heartbeat would be a prompt the operator could miss entirely.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void Rediscovery_flag_persists_on_subsequent_health_publishes()
|
||||
{
|
||||
var driver = new RediscoverableStubDriver();
|
||||
var publisher = new RecordingHealthPublisher();
|
||||
var parent = CreateTestProbe();
|
||||
parent.IgnoreMessages(_ => true);
|
||||
var actor = parent.ChildActorOf(DriverInstanceActor.Props(driver, healthPublisher: publisher));
|
||||
|
||||
actor.Tell(new DriverInstanceActor.InitializeRequested("{}"));
|
||||
AwaitAssert(() => publisher.Published.Count.ShouldBeGreaterThan(0), TimeSpan.FromSeconds(3));
|
||||
driver.RaiseRediscovery("agent-instance-changed");
|
||||
AwaitAssert(
|
||||
() => publisher.Published.Any(p => p.RediscoveryNeededUtc is not null),
|
||||
TimeSpan.FromSeconds(3));
|
||||
|
||||
// Force a further publish by changing something else about the driver's health.
|
||||
driver.SetLastError("transient read failure");
|
||||
AwaitAssert(
|
||||
() => publisher.Published.Any(p => p.LastError == "transient read failure"),
|
||||
TimeSpan.FromSeconds(3));
|
||||
|
||||
var latest = publisher.Published[^1];
|
||||
latest.RediscoveryNeededUtc.ShouldNotBeNull();
|
||||
latest.RediscoveryReason.ShouldBe("agent-instance-changed");
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// <b>Leak guard.</b> The <see cref="IDriver"/> instance can OUTLIVE the actor — the host stops a
|
||||
/// child and respawns another 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>. Asserts the
|
||||
/// driver's invocation list is empty again after the actor stops.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void Stopping_the_actor_detaches_the_rediscovery_handler()
|
||||
{
|
||||
var driver = new RediscoverableStubDriver();
|
||||
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 that does NOT implement <see cref="IRediscoverable"/> must publish a null flag —
|
||||
/// the field is absent, not defaulted to "needs re-browse".</summary>
|
||||
[Fact]
|
||||
public void A_non_rediscoverable_driver_never_sets_the_flag()
|
||||
{
|
||||
var publisher = new RecordingHealthPublisher();
|
||||
var parent = CreateTestProbe();
|
||||
parent.IgnoreMessages(_ => true);
|
||||
var actor = parent.ChildActorOf(DriverInstanceActor.Props(new StubDriver(), healthPublisher: publisher));
|
||||
|
||||
actor.Tell(new DriverInstanceActor.InitializeRequested("{}"));
|
||||
AwaitAssert(() => publisher.Published.Count.ShouldBeGreaterThan(0), TimeSpan.FromSeconds(3));
|
||||
|
||||
publisher.Published.ShouldAllBe(p => p.RediscoveryNeededUtc == null);
|
||||
}
|
||||
|
||||
/// <summary>Captures every health publish so a test can assert on the rediscovery fields.</summary>
|
||||
private sealed record HealthPublish(
|
||||
string ClusterId,
|
||||
string DriverInstanceId,
|
||||
DriverHealth Health,
|
||||
int ErrorCount5Min,
|
||||
DateTime? RediscoveryNeededUtc,
|
||||
string? RediscoveryReason)
|
||||
{
|
||||
public string? LastError => Health.LastError;
|
||||
}
|
||||
|
||||
private sealed class RecordingHealthPublisher : IDriverHealthPublisher
|
||||
{
|
||||
private readonly List<HealthPublish> _published = [];
|
||||
|
||||
/// <summary>Thread-safe snapshot — <c>Publish</c> is called from 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)
|
||||
{
|
||||
lock (_published)
|
||||
{
|
||||
_published.Add(new HealthPublish(
|
||||
clusterId, driverInstanceId, health, errorCount5Min, rediscoveryNeededUtc, rediscoveryReason));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A stub driver exposing <see cref="IRediscoverable"/>, a hook to raise it, and the live subscriber
|
||||
/// count (for the detach guard).
|
||||
/// <para><b>Implemented from scratch rather than derived from the shared <c>StubDriver</c> on
|
||||
/// purpose.</b> That one returns <c>GetHealth() => new(Healthy, DateTime.UtcNow, null)</c>, so its
|
||||
/// health fingerprint differs on EVERY call and the dedup under test never engages — which would make
|
||||
/// <see cref="Rediscovery_raise_is_not_swallowed_by_the_unchanged_health_dedup"/> pass vacuously
|
||||
/// whether or not the fix is present. This stub's health is STABLE until a test changes it.</para>
|
||||
/// </summary>
|
||||
private sealed class RediscoverableStubDriver : IDriver, IRediscoverable
|
||||
{
|
||||
private static readonly DateTime FixedLastRead = new(2026, 7, 27, 12, 0, 0, DateTimeKind.Utc);
|
||||
private volatile string? _lastError;
|
||||
|
||||
/// <inheritdoc />
|
||||
public event EventHandler<RediscoveryEventArgs>? OnRediscoveryNeeded;
|
||||
|
||||
/// <inheritdoc />
|
||||
public string DriverInstanceId => "rediscoverable-stub-1";
|
||||
|
||||
/// <inheritdoc />
|
||||
public string DriverType => "Stub";
|
||||
|
||||
/// <summary>Number of live subscribers on <see cref="OnRediscoveryNeeded"/>.</summary>
|
||||
public int SubscriberCount => OnRediscoveryNeeded?.GetInvocationList().Length ?? 0;
|
||||
|
||||
/// <summary>Raises the event exactly as a real driver's watcher does.</summary>
|
||||
public void RaiseRediscovery(string reason)
|
||||
=> OnRediscoveryNeeded?.Invoke(this, new RediscoveryEventArgs(reason, ScopeHint: null));
|
||||
|
||||
/// <summary>Changes the reported health so a test can force a publish that is NOT the rediscovery one.</summary>
|
||||
public void SetLastError(string error) => _lastError = error;
|
||||
|
||||
/// <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, _lastError);
|
||||
|
||||
/// <inheritdoc />
|
||||
public long GetMemoryFootprint() => 0;
|
||||
|
||||
/// <inheritdoc />
|
||||
public Task FlushOptionalCachesAsync(CancellationToken cancellationToken) => Task.CompletedTask;
|
||||
}
|
||||
}
|
||||
+7
-1
@@ -116,7 +116,13 @@ public sealed class DriverInstanceActorSubscriptionReconcileTests : RuntimeActor
|
||||
public int Count => Volatile.Read(ref _count);
|
||||
|
||||
/// <inheritdoc />
|
||||
public void Publish(string clusterId, string driverInstanceId, DriverHealth health, int errorCount5Min)
|
||||
public void Publish(
|
||||
string clusterId,
|
||||
string driverInstanceId,
|
||||
DriverHealth health,
|
||||
int errorCount5Min,
|
||||
DateTime? rediscoveryNeededUtc = null,
|
||||
string? rediscoveryReason = null)
|
||||
=> Interlocked.Increment(ref _count);
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user