Joseph Doherty
238 lines
12 KiB
C#
238 lines
12 KiB
C#
using System.Diagnostics.Metrics;
|
|
using Akka.Actor;
|
|
using Shouldly;
|
|
using Xunit;
|
|
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Admin;
|
|
using ZB.MOM.WW.OtOpcUa.Commons.Observability;
|
|
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
|
|
using ZB.MOM.WW.OtOpcUa.Commons.Types;
|
|
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
|
using ZB.MOM.WW.OtOpcUa.Runtime.Drivers;
|
|
using ZB.MOM.WW.OtOpcUa.Runtime.OpcUa;
|
|
using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness;
|
|
using static ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness.V3ArtifactSeeder;
|
|
|
|
namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers;
|
|
|
|
/// <summary>
|
|
/// <b>The 03/S15 contract test: comms loss → value nodes Bad → the driver's own re-publish clears
|
|
/// them.</b> A real <see cref="DriverInstanceActor"/> is spawned by a real
|
|
/// <see cref="DriverHostActor"/> around a scripted driver, so the whole seam runs: the child's
|
|
/// <c>Reconnecting</c> entry announces comms loss, the host stamps the value nodes
|
|
/// <c>BadNoCommunication</c>, the child reconnects and re-subscribes, and the driver's initial push is
|
|
/// what makes the nodes Good again.
|
|
///
|
|
/// <para>⚠️ <b>Vacuity rule — the pushed value is IDENTICAL to the pre-outage one.</b> The host writes
|
|
/// nothing Good on reconnect by design, so recovery depends entirely on a re-publish landing; and every
|
|
/// dedup on that path (a driver's last-value cache, a consumer's last-published memory) keys on value.
|
|
/// A recovery test whose driver pushes a <em>different</em> value therefore passes against a build with
|
|
/// no force-republish anywhere — the value change alone defeats every dedup — and is not this
|
|
/// contract's guard. The same-value push is the whole point.</para>
|
|
///
|
|
/// <para>The falsification control is
|
|
/// <see cref="A_silent_resubscribe_leaves_the_nodes_Bad_and_trips_the_audit"/>: a driver whose
|
|
/// <c>SubscribeAsync</c> pushes nothing (the pre-STAB-18 MQTT shape) leaves the nodes Bad and is caught
|
|
/// only by the audit. It proves the test above can fail.</para>
|
|
/// </summary>
|
|
public sealed class DriverHostCommsLossRecoveryTests : RuntimeActorTestBase
|
|
{
|
|
private static readonly NodeId TestNode = NodeId.Parse("driver-commsloss-recovery");
|
|
private static readonly RevisionHash RevA = RevisionHash.Parse(new string('d', 64));
|
|
|
|
/// <summary>Audit window measured from the RECONNECT, not from the outage — the initial push lands
|
|
/// within milliseconds of Connected, so this is generous.</summary>
|
|
private static readonly TimeSpan AuditDelay = TimeSpan.FromSeconds(2);
|
|
|
|
/// <summary>The child's reconnect retry runs on <c>DriverInstanceActor.DefaultReconnectInterval</c>
|
|
/// (10 s) and the host exposes no seam to shorten it, so every budget here clears that.</summary>
|
|
private static readonly TimeSpan ReconnectBudget = TimeSpan.FromSeconds(25);
|
|
|
|
private const string SpeedRef = "Plant/Modbus/dev1/speed";
|
|
private const string UnsSpeedNode = "filling/line1/station1/speed";
|
|
|
|
[Fact]
|
|
public void An_identical_value_pushed_on_resubscribe_recovers_the_nodes()
|
|
{
|
|
var driver = new ScriptedSubscribeDriver { InstanceId = "drv-recover", InitialPush = (SpeedRef, 5.0, OpcUaStatus.Good) };
|
|
var (host, publish) = Arrange(driver, "drv-recover");
|
|
using var meter = new MeterRecorder("otopcua.driver.comms_latched_refs");
|
|
|
|
// Steady state: the driver's FIRST subscribe already pushed 5 Good, and the host fanned it to
|
|
// raw + UNS. (The push on subscribe is the behaviour under test, so it is also how the pre-outage
|
|
// value gets there — no separate seeding call, which would leave an extra batch in the probe.)
|
|
var first = publish.ExpectMsg<OpcUaPublishActor.AttributeValueBatchUpdate>(PresenceBudget);
|
|
first.Items.Select(i => i.NodeId).ShouldBe(new[] { SpeedRef, UnsSpeedNode }, ignoreOrder: true);
|
|
first.Items.ShouldAllBe(i => Equals(i.Value, 5.0));
|
|
|
|
// The outage: the child re-enters Reconnecting and announces comms loss to the host.
|
|
host.Tell(new ReconnectDriver("c1", "drv-recover", "test-operator", Guid.NewGuid()));
|
|
|
|
var staled = publish.ExpectMsg<OpcUaPublishActor.AttributeStatusBatchUpdate>(ReconnectBudget);
|
|
staled.Items.Select(i => i.NodeId).ShouldBe(new[] { SpeedRef, UnsSpeedNode }, ignoreOrder: true);
|
|
staled.Items.ShouldAllBe(i => i.StatusCode == OpcUaStatus.BadNoCommunication);
|
|
|
|
// Recovery: the reconnect re-subscribes, and the driver pushes the SAME value it held before.
|
|
var recovered = publish.ExpectMsg<OpcUaPublishActor.AttributeValueBatchUpdate>(ReconnectBudget);
|
|
recovered.Items.Select(i => i.NodeId).ShouldBe(new[] { SpeedRef, UnsSpeedNode }, ignoreOrder: true);
|
|
recovered.Items.ShouldAllBe(i => i.StatusCode == OpcUaStatus.Good);
|
|
recovered.Items.ShouldAllBe(i => Equals(i.Value, 5.0));
|
|
|
|
// …and the audit finds nothing outstanding.
|
|
EventFilter.Warning(contains: "still BadNoCommunication")
|
|
.Expect(0, AuditDelay + TimeSpan.FromSeconds(3), () => { });
|
|
meter.SumWithTag("driver_instance", "drv-recover").ShouldBe(0);
|
|
}
|
|
|
|
/// <summary>
|
|
/// <b>The falsification control.</b> Same scenario, but the driver's <c>SubscribeAsync</c> pushes
|
|
/// nothing — exactly what MQTT did before 05/STAB-18, and what any push driver without an initial
|
|
/// push does. The nodes stay Bad (the host writes no Good of its own), and the ONLY thing that says
|
|
/// so is the recovery audit.
|
|
/// </summary>
|
|
[Fact]
|
|
public void A_silent_resubscribe_leaves_the_nodes_Bad_and_trips_the_audit()
|
|
{
|
|
var driver = new ScriptedSubscribeDriver { InstanceId = "drv-silent", InitialPush = null };
|
|
var (host, publish) = Arrange(driver, "drv-silent");
|
|
using var meter = new MeterRecorder("otopcua.driver.comms_latched_refs");
|
|
|
|
driver.FireDataChange(SpeedRef, 5.0, OpcUaStatus.Good);
|
|
publish.ExpectMsg<OpcUaPublishActor.AttributeValueBatchUpdate>(PresenceBudget);
|
|
|
|
EventFilter.Warning(contains: "still BadNoCommunication").ExpectOne(ReconnectBudget, () =>
|
|
{
|
|
host.Tell(new ReconnectDriver("c1", "drv-silent", "test-operator", Guid.NewGuid()));
|
|
publish.ExpectMsg<OpcUaPublishActor.AttributeStatusBatchUpdate>(ReconnectBudget);
|
|
|
|
// Nothing Good ever arrives — assert it rather than merely not looking.
|
|
publish.ExpectNoMsg(TimeSpan.FromSeconds(1));
|
|
});
|
|
|
|
// One ref (the raw tag); its UNS mirror shares it.
|
|
AwaitCondition(() => meter.SumWithTag("driver_instance", "drv-silent") == 1, TimeSpan.FromSeconds(5));
|
|
driver.SubscribeCount.ShouldBeGreaterThanOrEqualTo(2,
|
|
"the child must actually have re-subscribed — otherwise this proves nothing about the push");
|
|
}
|
|
|
|
private (IActorRef Host, Akka.TestKit.TestProbe Publish) Arrange(ScriptedSubscribeDriver driver, string driverId)
|
|
{
|
|
var db = NewInMemoryDbFactory();
|
|
var deploymentId = SeedV3Deployment(db, RevA,
|
|
rawTags: new[]
|
|
{
|
|
(Driver: driverId, DriverName: "Modbus", Device: "dev1", Tag: "speed", DataType: "Double", Writable: false),
|
|
},
|
|
unsRefs: new[] { (Area: "filling", Line: "line1", Equip: "station1", BackingTag: "speed", Effective: (string?)null) },
|
|
enabled: true);
|
|
|
|
var (host, publish, _) = SpawnHostAndApply(
|
|
this, db, TestNode, deploymentId, RevA,
|
|
commsRecoveryAuditDelay: AuditDelay,
|
|
driverFactory: new SingleDriverFactory(driver));
|
|
|
|
// The child is spawned + initialised by the apply; wait for its first subscribe so the scripted
|
|
// pushes below are attributed to a live subscription.
|
|
AwaitCondition(() => driver.SubscribeCount >= 1, PresenceBudget);
|
|
return (host, publish);
|
|
}
|
|
|
|
/// <summary>Hands the host one scripted driver, whatever type/instance it asks for.</summary>
|
|
private sealed class SingleDriverFactory(IDriver driver) : IDriverFactory
|
|
{
|
|
public IDriver? TryCreate(string driverType, string driverInstanceId, string driverConfigJson) => driver;
|
|
|
|
public IReadOnlyCollection<string> SupportedTypes { get; } = ["Modbus"];
|
|
}
|
|
|
|
/// <summary>
|
|
/// A subscribable driver whose <see cref="SubscribeAsync"/> optionally performs an <b>initial
|
|
/// push</b> — the poll engine's <c>forceRaise: true</c> and MQTT's post-STAB-18 behaviour, modelled
|
|
/// at the seam the host actually observes. With <see cref="InitialPush"/> null it is the silent
|
|
/// shape the control test needs.
|
|
/// </summary>
|
|
private sealed class ScriptedSubscribeDriver : IDriver, ISubscribable
|
|
{
|
|
private readonly StubHandle _handle = new();
|
|
private int _subscribeCount;
|
|
|
|
public event EventHandler<DataChangeEventArgs>? OnDataChange;
|
|
|
|
/// <summary>The value re-published on every (re)subscribe, or null for a silent driver.</summary>
|
|
public (string Ref, object? Value, uint StatusCode)? InitialPush { get; init; }
|
|
|
|
/// <summary>The instance id the child actor adopts — <c>DriverInstanceActor</c> takes it from the
|
|
/// DRIVER, not from the spec, and the host's ref map is keyed by the spec's id. They must agree or
|
|
/// every published value is dropped as unmapped.</summary>
|
|
public required string InstanceId { get; init; }
|
|
|
|
public int SubscribeCount => Volatile.Read(ref _subscribeCount);
|
|
|
|
public string DriverInstanceId => InstanceId;
|
|
|
|
public string DriverType => "Modbus";
|
|
|
|
public Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken) => Task.CompletedTask;
|
|
|
|
public Task ReinitializeAsync(string driverConfigJson, CancellationToken cancellationToken) => Task.CompletedTask;
|
|
|
|
public Task ShutdownAsync(CancellationToken cancellationToken) => Task.CompletedTask;
|
|
|
|
// Stable health: the #527 watchdog must not fire on its own here — the outage in these tests is
|
|
// driven explicitly, and a health-driven trip would make the timing non-deterministic.
|
|
public DriverHealth GetHealth() => new(DriverState.Healthy, DateTime.UtcNow, null);
|
|
|
|
public Task<ISubscriptionHandle> SubscribeAsync(
|
|
IReadOnlyList<string> fullReferences, TimeSpan publishingInterval, CancellationToken cancellationToken)
|
|
{
|
|
Interlocked.Increment(ref _subscribeCount);
|
|
if (InitialPush is { } push && fullReferences.Contains(push.Ref))
|
|
FireDataChange(push.Ref, push.Value, push.StatusCode);
|
|
|
|
return Task.FromResult<ISubscriptionHandle>(_handle);
|
|
}
|
|
|
|
public Task UnsubscribeAsync(ISubscriptionHandle handle, CancellationToken cancellationToken) => Task.CompletedTask;
|
|
|
|
public void FireDataChange(string fullRef, object? value, uint statusCode)
|
|
{
|
|
var snapshot = new DataValueSnapshot(value, statusCode, DateTime.UtcNow, DateTime.UtcNow);
|
|
OnDataChange?.Invoke(this, new DataChangeEventArgs(_handle, fullRef, snapshot));
|
|
}
|
|
}
|
|
|
|
/// <summary>Tag-filtered counter recorder — see the note in
|
|
/// <c>DriverHostActorCommsLossValueNodesTests</c> on why the instrument name alone is not enough.</summary>
|
|
private sealed class MeterRecorder : IDisposable
|
|
{
|
|
private readonly MeterListener _listener;
|
|
private readonly List<(long Value, KeyValuePair<string, object?>[] Tags)> _measurements = new();
|
|
private readonly object _gate = new();
|
|
|
|
public MeterRecorder(string instrumentName)
|
|
{
|
|
_listener = new MeterListener
|
|
{
|
|
InstrumentPublished = (instrument, listener) =>
|
|
{
|
|
if (instrument.Meter.Name == OtOpcUaTelemetry.MeterName && instrument.Name == instrumentName)
|
|
listener.EnableMeasurementEvents(instrument);
|
|
},
|
|
};
|
|
_listener.SetMeasurementEventCallback<long>((_, value, tags, _) =>
|
|
{
|
|
lock (_gate) _measurements.Add((value, tags.ToArray()));
|
|
});
|
|
_listener.Start();
|
|
}
|
|
|
|
public long SumWithTag(string key, string value)
|
|
{
|
|
lock (_gate)
|
|
return _measurements.Where(m => m.Tags.Any(t => t.Key == key && Equals(t.Value, value)))
|
|
.Sum(m => m.Value);
|
|
}
|
|
|
|
public void Dispose() => _listener.Dispose();
|
|
}
|
|
}
|