Joseph Doherty
230 lines
11 KiB
C#
230 lines
11 KiB
C#
using Shouldly;
|
|
using Xunit;
|
|
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
|
|
|
namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests;
|
|
|
|
/// <summary>
|
|
/// The push-model edge behaviours <see cref="MqttSubscriptionManager"/> lacked (archreview
|
|
/// 05/STAB-18): an <b>initial push</b> of the cached snapshot on every (re)subscribe, and a
|
|
/// <b>connection-loss stale</b> to <c>BadNoCommunication</c> with the value retained.
|
|
///
|
|
/// <para><b>Why both, and why the identical payload is load-bearing.</b> The manager's steady-state
|
|
/// dedup suppresses a message whose value AND status equal the cached one, and that cache is
|
|
/// driver-owned — it survives a reconnect and a redeploy. So a broker that replays its retained
|
|
/// message unchanged after an outage raised nothing, and every node the host had latched Bad stayed
|
|
/// Bad until the plant value happened to change. A recovery test that replays a <em>different</em>
|
|
/// payload passes against the pre-fix code and proves nothing; every recovery assertion here
|
|
/// therefore replays the <b>identical</b> payload.</para>
|
|
///
|
|
/// <para>The steady-state suppression itself must survive — it is what keeps a 10 Hz topic from
|
|
/// re-triggering the whole downstream fan-out per message — so each recovery case is paired with a
|
|
/// control that asserts an unchanged republish is still suppressed.</para>
|
|
/// </summary>
|
|
public sealed class MqttSubscriptionManagerRecoveryTests
|
|
{
|
|
// OPC UA status codes, verified against Opc.Ua.StatusCodes rather than recalled.
|
|
private const uint Good = 0x00000000u;
|
|
private const uint BadNoCommunication = 0x80310000u;
|
|
private const uint BadWaitingForInitialData = 0x80320000u;
|
|
|
|
private const string TempPath = "Plant/Mqtt/broker1/OvenTemp";
|
|
private const string PressPath = "Plant/Mqtt/broker1/OvenPressure";
|
|
private const string TempTopic = "f/oven/temp";
|
|
private const string PressTopic = "f/oven/press";
|
|
|
|
private static RawTagEntry Tag(string rawPath, string topic) =>
|
|
new(rawPath, $$"""{"topic":"{{topic}}","payloadFormat":"Json","jsonPath":"$.value","dataType":"Float64"}""",
|
|
WriteIdempotent: false);
|
|
|
|
private static MqttSubscriptionManager NewManager()
|
|
{
|
|
var mgr = new MqttSubscriptionManager(new MqttDriverOptions { Mode = MqttMode.Plain }, "mqtt-1");
|
|
mgr.Register([Tag(TempPath, TempTopic), Tag(PressPath, PressTopic)]);
|
|
return mgr;
|
|
}
|
|
|
|
private sealed record Raised(string Reference, object? Value, uint StatusCode, ISubscriptionHandle Handle);
|
|
|
|
private static List<Raised> Record(MqttSubscriptionManager mgr)
|
|
{
|
|
var raised = new List<Raised>();
|
|
mgr.OnDataChange += (_, e) =>
|
|
raised.Add(new Raised(e.FullReference, e.Snapshot.Value, e.Snapshot.StatusCode, e.SubscriptionHandle));
|
|
return raised;
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------------
|
|
// (A) Initial push on (re)subscribe — 05/STAB-18 shape (b)
|
|
// ---------------------------------------------------------------------------------
|
|
|
|
/// <summary>
|
|
/// A ref observed before the subscription existed is pushed once, with its cached snapshot,
|
|
/// attributed to the new handle. This is the push-model equivalent of the poll engine's
|
|
/// <c>forceRaise: true</c> on the first poll of a new subscription, and it is what refills a
|
|
/// node the address-space rebuild reseeded <c>BadWaitingForInitialData</c> on a quiet topic.
|
|
/// </summary>
|
|
[Fact]
|
|
public async Task SubscribeAsync_pushes_the_cached_snapshot_for_an_observed_ref()
|
|
{
|
|
var mgr = NewManager();
|
|
mgr.HandleMessage(TempTopic, """{"value":21.5}"""u8.ToArray(), retained: false); // cached; no handle yet
|
|
|
|
var raised = Record(mgr);
|
|
var handle = await mgr.SubscribeAsync([TempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken);
|
|
|
|
raised.Count.ShouldBe(1);
|
|
raised[0].Reference.ShouldBe(TempPath);
|
|
raised[0].Value.ShouldBe(21.5);
|
|
raised[0].StatusCode.ShouldBe(Good);
|
|
raised[0].Handle.ShouldBe(handle);
|
|
}
|
|
|
|
/// <summary>
|
|
/// A ref that has never been observed carries only the materialiser's
|
|
/// <c>BadWaitingForInitialData</c> placeholder — pushing that would claim an observation the
|
|
/// driver has not made, so the initial push skips it.
|
|
/// </summary>
|
|
[Fact]
|
|
public async Task SubscribeAsync_does_not_push_a_never_observed_ref()
|
|
{
|
|
var mgr = NewManager();
|
|
mgr.HandleMessage(TempTopic, """{"value":21.5}"""u8.ToArray(), retained: false);
|
|
|
|
var raised = Record(mgr);
|
|
await mgr.SubscribeAsync([TempPath, PressPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken);
|
|
|
|
raised.Select(r => r.Reference).ShouldBe([TempPath]);
|
|
mgr.Values.Read(PressPath).StatusCode.ShouldBe(BadWaitingForInitialData);
|
|
}
|
|
|
|
/// <summary>
|
|
/// <b>Control.</b> The initial push must not weaken steady-state suppression: an identical
|
|
/// message arriving after the subscribe is still swallowed. Without this, "recovery works"
|
|
/// could simply mean "nothing is ever deduped any more".
|
|
/// </summary>
|
|
[Fact]
|
|
public async Task An_identical_message_after_the_initial_push_is_still_suppressed()
|
|
{
|
|
var mgr = NewManager();
|
|
mgr.HandleMessage(TempTopic, """{"value":21.5}"""u8.ToArray(), retained: false);
|
|
|
|
var raised = Record(mgr);
|
|
await mgr.SubscribeAsync([TempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken);
|
|
mgr.HandleMessage(TempTopic, """{"value":21.5}"""u8.ToArray(), retained: false);
|
|
|
|
raised.Count.ShouldBe(1); // the initial push only
|
|
}
|
|
|
|
/// <summary>
|
|
/// The rebuild shape: unsubscribe and subscribe again — as every deploy does through
|
|
/// <c>DriverInstanceActor</c>'s unsubscribe-then-subscribe — and the cached value is pushed
|
|
/// again onto the new handle rather than leaving the rebuilt node waiting for the plant.
|
|
/// </summary>
|
|
[Fact]
|
|
public async Task A_resubscribe_pushes_again()
|
|
{
|
|
var mgr = NewManager();
|
|
mgr.HandleMessage(TempTopic, """{"value":21.5}"""u8.ToArray(), retained: false);
|
|
var first = await mgr.SubscribeAsync([TempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken);
|
|
await mgr.UnsubscribeAsync(first, TestContext.Current.CancellationToken);
|
|
|
|
var raised = Record(mgr);
|
|
var second = await mgr.SubscribeAsync([TempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken);
|
|
|
|
raised.Count.ShouldBe(1);
|
|
raised[0].Value.ShouldBe(21.5);
|
|
raised[0].Handle.ShouldBe(second);
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------------
|
|
// (B) Connection-loss stale — 05/STAB-18 shape (a)
|
|
// ---------------------------------------------------------------------------------
|
|
|
|
/// <summary>
|
|
/// A lost broker session makes every observed reference <c>BadNoCommunication</c> — with its
|
|
/// last value retained, the Part 8 last-known-value convention a SCADA client expects. MQTT
|
|
/// previously marked nothing at all on its own disconnect edge, so the tags stayed Good until
|
|
/// the host's 90 s comms-loss watchdog noticed (if it noticed at all — MQTTnet usually
|
|
/// reconnects well inside that window).
|
|
/// </summary>
|
|
[Fact]
|
|
public async Task OnConnectionLost_stales_every_observed_ref_keeping_its_value()
|
|
{
|
|
var mgr = NewManager();
|
|
mgr.HandleMessage(TempTopic, """{"value":21.5}"""u8.ToArray(), retained: false);
|
|
await mgr.SubscribeAsync([TempPath, PressPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken);
|
|
|
|
var raised = Record(mgr); // recorded AFTER the subscribe, so the initial push is not counted
|
|
mgr.OnConnectionLost();
|
|
|
|
raised.Count.ShouldBe(1, "the never-observed ref has nothing to stale");
|
|
raised[0].Reference.ShouldBe(TempPath);
|
|
raised[0].StatusCode.ShouldBe(BadNoCommunication);
|
|
raised[0].Value.ShouldBe(21.5, "the last known value is retained, not nulled");
|
|
mgr.Values.Read(TempPath).StatusCode.ShouldBe(BadNoCommunication);
|
|
mgr.Values.Read(TempPath).Value.ShouldBe(21.5);
|
|
}
|
|
|
|
/// <summary>
|
|
/// <b>The contract test for this seam.</b> After the stale, the broker's retained replay of the
|
|
/// <b>identical</b> payload is raised Good — because the cached status now differs even though
|
|
/// the value does not. Replaying a <em>different</em> payload here would pass against the
|
|
/// pre-fix build and prove nothing; the identical payload is the whole point.
|
|
/// </summary>
|
|
[Fact]
|
|
public async Task After_the_stale_the_identical_retained_replay_is_raised_Good()
|
|
{
|
|
var mgr = NewManager();
|
|
mgr.HandleMessage(TempTopic, """{"value":21.5}"""u8.ToArray(), retained: false);
|
|
await mgr.SubscribeAsync([TempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken);
|
|
mgr.OnConnectionLost();
|
|
|
|
var raised = Record(mgr);
|
|
mgr.HandleMessage(TempTopic, """{"value":21.5}"""u8.ToArray(), retained: true); // the retained seed
|
|
|
|
raised.Count.ShouldBe(1);
|
|
raised[0].Reference.ShouldBe(TempPath);
|
|
raised[0].Value.ShouldBe(21.5);
|
|
raised[0].StatusCode.ShouldBe(Good);
|
|
}
|
|
|
|
/// <summary>
|
|
/// A second loss with no intervening message raises nothing: the ref is already Bad, and a
|
|
/// re-stamp would be pure churn on the downstream fan-out.
|
|
/// </summary>
|
|
[Fact]
|
|
public async Task OnConnectionLost_does_not_re_raise_an_already_bad_ref()
|
|
{
|
|
var mgr = NewManager();
|
|
mgr.HandleMessage(TempTopic, """{"value":21.5}"""u8.ToArray(), retained: false);
|
|
await mgr.SubscribeAsync([TempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken);
|
|
mgr.OnConnectionLost();
|
|
|
|
var raised = Record(mgr);
|
|
mgr.OnConnectionLost();
|
|
|
|
raised.ShouldBeEmpty();
|
|
}
|
|
|
|
/// <summary>
|
|
/// The stale runs on MQTTnet's dispatcher thread, so it must complete synchronously and never
|
|
/// await: parking that thread stalls delivery for every other subscription on the connection.
|
|
/// </summary>
|
|
[Fact]
|
|
public async Task OnConnectionLost_completes_on_the_calling_thread()
|
|
{
|
|
var mgr = NewManager();
|
|
mgr.HandleMessage(TempTopic, """{"value":21.5}"""u8.ToArray(), retained: false);
|
|
await mgr.SubscribeAsync([TempPath], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken);
|
|
|
|
var callingThread = Environment.CurrentManagedThreadId;
|
|
var raisedOn = 0;
|
|
mgr.OnDataChange += (_, _) => raisedOn = Environment.CurrentManagedThreadId;
|
|
|
|
mgr.OnConnectionLost();
|
|
|
|
raisedOn.ShouldBe(callingThread);
|
|
}
|
|
}
|