dohertj2/lmxopcua
1
0
Fork 0
Code Issues 17 Pull Requests Actions Packages Projects Releases Wiki Activity

feat(mqtt): Sparkplug UntilStable discovery + rediscover-on-DBIRTH

Browse Source 
RediscoverPolicy is now mode-dependent: UntilStable in Sparkplug B (a tag's
dataType is optional -- the birth certificate declares it, so the discovered
tree fills in asynchronously), Once in Plain (nothing about the authored set
arrives on the wire).

DiscoverAsync resolves each Sparkplug tag's datatype per pass from the live
BirthCache, with the ingest path's own precedence: authored dataType wins,
else the birth's, else the record default. An unsupported Sparkplug type
(DataSet/Template/PropertySet/Unknown) falls back rather than blanking the
tag. The authored tag SET is never conditional on a birth -- an authored tag
is part of the declared configuration, not something the plant grants by
publishing.

SparkplugIngestor.BirthObserved is wired to RaiseRediscoveryNeeded behind a
two-part change gate, which is the anti-storm mechanism rather than an
optimisation: fire only for a scope an authored tag binds, and only when the
birth's metric-name SET (ordered-distinct, ordinal) differs from the last one
seen for that scope. Edge nodes re-birth freely -- on their own timer, on
every reconnect via the late-join rebirth fan-out, and once per rebirth NCMD
the gap policy sends -- and firing on each of those would make a healthy plant
a permanent address-space-rebuild loop. No extra debounce: what survives the
gate is a real edit to the served tree, and delaying it would need its own
trailing-edge flush to avoid dropping the last change.

The signature map is NOT cleared on death/reconnect/ingest rebuild (the
ingestor drops its whole birth cache on reconnect, but "the driver forgot" is
not "the address space changed"); it is pruned when a redeploy stops
authoring a scope.

ScopeHint is the "Mqtt" discovery folder, deliberately not the Sparkplug scope
path the plan named: the discovered tree is flat, so an edge-node/device path
would name a subtree that does not exist and a consumer scoping a rebuild on
it would rebuild nothing. The scope is carried in Reason instead.

Falsifiability: always-fire reddens the identical-rebirth + reordered-rebirth
pins; always-authored-datatype reddens the birth-fill pin; dropping the
authored-scope filter reddens the unauthored-device pin. 524/524 MQTT unit
tests green (was 513).

Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW
This commit is contained in:
Joseph Doherty
2026-07-24 22:28:08 -04:00
parent 8538fb798b
commit 6917d8805d
2 changed files with 561 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttDriverDiscoveryTests.cs
+300
View File
@@ -1,7 +1,11 @@
using System.Text;
using Google.Protobuf;
using Org.Eclipse.Tahu.Protobuf;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Sparkplug;
using TahuDataType = Org.Eclipse.Tahu.Protobuf.DataType;
namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests;
@@ -364,6 +368,302 @@ public sealed class MqttDriverDiscoveryTests
statuses[0].State.ShouldBe(HostState.Unknown);
}
// =================================================================================
// Task 22 — Sparkplug discovery policy, birth-filled datatypes, rediscovery trigger
// =================================================================================
/// <summary>
/// Sparkplug's discovered shape is not fully known at connect: a tag authored without a
/// <c>dataType</c> takes its type from the birth certificate, which arrives asynchronously.
/// That is precisely the <see cref="DiscoveryRediscoverPolicy.UntilStable"/> contract — the
/// host re-runs discovery until the captured set settles.
/// </summary>
[Fact]
public void SparkplugMode_RediscoverPolicy_IsUntilStable()
=> SparkplugDriver().RediscoverPolicy.ShouldBe(DiscoveryRediscoverPolicy.UntilStable);
/// <summary>
/// The mode gate, asserted from the other side: plain MQTT's authored set is fully known
/// synchronously, so it must stay <see cref="DiscoveryRediscoverPolicy.Once"/>. A blanket flip
/// to <c>UntilStable</c> would make every plain deployment re-run discovery on a timer for a
/// tree that can never change.
/// </summary>
[Fact]
public void PlainMode_RediscoverPolicy_StaysOnce()
=> PlainDriver(Tag("Plant/Mqtt/dev1/Temp", "f/t")).RediscoverPolicy.ShouldBe(DiscoveryRediscoverPolicy.Once);
/// <summary>
/// A DBIRTH for an authored device the driver has never held a birth for introduces the
/// metrics' datatypes, so the discovered tree genuinely changed: rediscovery fires.
/// </summary>
[Fact]
public void NewDbirth_FiresOnRediscoveryNeeded()
{
var driver = SparkplugDriver(SpTag(SpTempPath, SpBlob(SpDevice, "Temperature")));
var fired = 0;
((IRediscoverable)driver).OnRediscoveryNeeded += (_, _) => Interlocked.Increment(ref fired);
FeedNodeBirth(driver);
FeedDeviceBirth(driver, seq: 1, ("Temperature", 5UL, TahuDataType.Float));
fired.ShouldBeGreaterThan(0);
}
/// <summary>
/// <b>The primary anti-storm pin.</b> An edge node that re-births on a timer (or a flapping
/// connection driving the late-join rebirth) republishes the same metric set over and over. If
/// every birth fired rediscovery, each one would trigger an address-space rebuild — expensive
/// and disruptive on a running server, forever, for a tree that never changed.
/// </summary>
[Fact]
public void IdenticalRebirth_DoesNotFireRediscovery()
{
var driver = SparkplugDriver(SpTag(SpTempPath, SpBlob(SpDevice, "Temperature")));
FeedNodeBirth(driver);
FeedDeviceBirth(driver, seq: 1, ("Temperature", 5UL, TahuDataType.Float));
// Subscribe only AFTER the first birth so the count below is purely the rebirths'.
var fired = 0;
((IRediscoverable)driver).OnRediscoveryNeeded += (_, _) => Interlocked.Increment(ref fired);
for (var i = 0; i < 5; i++)
{
FeedNodeBirth(driver, seq: 0, bdSeq: (ulong)(i + 2));
FeedDeviceBirth(driver, seq: 1, ("Temperature", 5UL, TahuDataType.Float));
}
fired.ShouldBe(0);
}
/// <summary>
/// The other half of the gate: a rebirth that actually changes the metric set (a metric added,
/// removed or renamed) DOES change what discovery can resolve, and must fire.
/// </summary>
[Fact]
public void RebirthWithChangedMetricSet_FiresRediscovery()
{
var driver = SparkplugDriver(
SpTag(SpTempPath, SpBlob(SpDevice, "Temperature")),
SpTag(SpPressPath, SpBlob(SpDevice, "Pressure")));
FeedNodeBirth(driver);
FeedDeviceBirth(driver, seq: 1, ("Temperature", 5UL, TahuDataType.Float));
var fired = 0;
((IRediscoverable)driver).OnRediscoveryNeeded += (_, _) => Interlocked.Increment(ref fired);
FeedDeviceBirth(driver, seq: 2, ("Temperature", 5UL, TahuDataType.Float), ("Pressure", 6UL, TahuDataType.Float));
fired.ShouldBe(1);
}
/// <summary>
/// Metric ORDER is not the metric SET. A birth that lists the same metrics in a different order
/// describes the same address space; firing on it would reintroduce the rebuild storm through
/// the back door, because nothing obliges an edge node to keep its birth order stable.
/// </summary>
[Fact]
public void RebirthReorderingTheSameMetrics_DoesNotFireRediscovery()
{
var driver = SparkplugDriver(
SpTag(SpTempPath, SpBlob(SpDevice, "Temperature")),
SpTag(SpPressPath, SpBlob(SpDevice, "Pressure")));
FeedNodeBirth(driver);
FeedDeviceBirth(driver, seq: 1, ("Temperature", 5UL, TahuDataType.Float), ("Pressure", 6UL, TahuDataType.Float));
var fired = 0;
((IRediscoverable)driver).OnRediscoveryNeeded += (_, _) => Interlocked.Increment(ref fired);
FeedDeviceBirth(driver, seq: 2, ("Pressure", 6UL, TahuDataType.Float), ("Temperature", 5UL, TahuDataType.Float));
fired.ShouldBe(0);
}
/// <summary>
/// A birth for a device NO authored tag names cannot change <see cref="MqttDriver.DiscoverAsync"/>'s
/// output — discovery replays the authored set. A large plant DBIRTHs devices this deployment
/// never reads on their own schedules; firing for those is a rebuild storm sourced from traffic
/// the configuration has no opinion about.
/// </summary>
[Fact]
public void BirthForAnUnauthoredDevice_DoesNotFireRediscovery()
{
var driver = SparkplugDriver(SpTag(SpTempPath, SpBlob(SpDevice, "Temperature")));
FeedNodeBirth(driver);
var fired = 0;
((IRediscoverable)driver).OnRediscoveryNeeded += (_, _) => Interlocked.Increment(ref fired);
// Same (authored) edge node, a device nothing binds — the ingestor still applies the birth.
FeedDeviceBirth(driver, "Filler99", seq: 1, ("Temperature", 5UL, TahuDataType.Float));
fired.ShouldBe(0);
}
/// <summary>
/// The rediscovery event names the subtree this driver's discovery actually emits (the
/// <c>Mqtt</c> folder) so a consumer scoping a rebuild on it scopes to exactly the tree that
/// changed, and carries the Sparkplug scope in the diagnostic reason.
/// </summary>
[Fact]
public void RediscoveryEvent_ScopeHintIsTheDiscoveryFolder_ReasonNamesTheScope()
{
var driver = SparkplugDriver(SpTag(SpTempPath, SpBlob(SpDevice, "Temperature")));
RediscoveryEventArgs? args = null;
((IRediscoverable)driver).OnRediscoveryNeeded += (_, e) => args = e;
FeedNodeBirth(driver);
FeedDeviceBirth(driver, seq: 1, ("Temperature", 5UL, TahuDataType.Float));
args.ShouldNotBeNull();
args.ScopeHint.ShouldBe("Mqtt");
args.Reason.ShouldContain($"{SpGroup}/{SpNode}/{SpDevice}");
}
/// <summary>
/// <b>The datatype fill-in.</b> <c>dataType</c> is optional in the Sparkplug tag shape — the
/// birth certificate declares it. Before any birth the tag can only report the record default;
/// once the DBIRTH lands, discovery must re-stream it with the type the birth declared.
/// </summary>
[Fact]
public async Task DiscoverAsync_Sparkplug_UnauthoredDataType_FillsInFromTheBirth()
{
var driver = SparkplugDriver(SpTag(SpTempPath, SpBlob(SpDevice, "Temperature")));
var before = new CapturingAddressSpaceBuilder();
await driver.DiscoverAsync(before, TestContext.Current.CancellationToken);
// The node exists from the first pass — an authored tag is part of the declared configuration,
// not something the plant grants by publishing.
before.Variables.Select(v => v.Info.FullName).ShouldBe([SpTempPath]);
before.Variables[0].Info.DriverDataType.ShouldBe(DriverDataType.String); // the record default
FeedNodeBirth(driver);
FeedDeviceBirth(driver, seq: 1, ("Temperature", 5UL, TahuDataType.Float));
var after = new CapturingAddressSpaceBuilder();
await driver.DiscoverAsync(after, TestContext.Current.CancellationToken);
after.Variables.Select(v => v.Info.FullName).ShouldBe([SpTempPath]);
after.Variables[0].Info.DriverDataType.ShouldBe(DriverDataType.Float32);
}
/// <summary>
/// The authored <c>dataType</c> wins over the birth's — the same precedence the ingest path
/// applies when it coerces a value (<c>DataTypeAuthored ? authored : birth</c>). A discovery
/// surface that disagreed with the publish surface would report a type no value ever arrives as.
/// </summary>
[Fact]
public async Task DiscoverAsync_Sparkplug_AuthoredDataType_WinsOverTheBirth()
{
var driver = SparkplugDriver(SpTag(SpTempPath, SpBlob(SpDevice, "Temperature", "Int32")));
FeedNodeBirth(driver);
FeedDeviceBirth(driver, seq: 1, ("Temperature", 5UL, TahuDataType.Float));
var b = new CapturingAddressSpaceBuilder();
await driver.DiscoverAsync(b, TestContext.Current.CancellationToken);
b.Variables[0].Info.DriverDataType.ShouldBe(DriverDataType.Int32);
}
/// <summary>
/// A birth declaring a Sparkplug type this driver cannot map (DataSet / Template / PropertySet
/// / Unknown) must not blank the tag's type — <c>ToDriverDataType()</c> returns null for those,
/// and discovery falls back to the authored/default type rather than to nothing.
/// </summary>
[Fact]
public async Task DiscoverAsync_Sparkplug_UnsupportedBirthType_FallsBackToTheDefault()
{
var driver = SparkplugDriver(SpTag(SpTempPath, SpBlob(SpDevice, "Temperature")));
FeedNodeBirth(driver);
FeedDeviceBirth(driver, seq: 1, ("Temperature", 5UL, TahuDataType.DataSet));
var b = new CapturingAddressSpaceBuilder();
await driver.DiscoverAsync(b, TestContext.Current.CancellationToken);
b.Variables.Count.ShouldBe(1);
b.Variables[0].Info.DriverDataType.ShouldBe(DriverDataType.String);
}
// ---- Sparkplug fixtures ----
private const string SpGroup = "Plant1";
private const string SpNode = "EdgeA";
private const string SpDevice = "Filler1";
private const string SpTempPath = "Plant/Mqtt/spb/Filler1Temp";
private const string SpPressPath = "Plant/Mqtt/spb/Filler1Press";
/// <summary>A Sparkplug tag blob. <c>dataType</c> is omitted unless supplied — it is optional.</summary>
private static string SpBlob(string? device, string metric, string? dataType = null)
{
var deviceKey = device is null ? "" : $"""{'"'}deviceId{'"'}:"{device}",""";
var typeKey = dataType is null ? "" : $""","dataType":"{dataType}" """;
return $$"""{"groupId":"{{SpGroup}}","edgeNodeId":"{{SpNode}}",{{deviceKey}}"metricName":"{{metric}}"{{typeKey}}}""";
}
private static RawTagEntry SpTag(string rawPath, string blob) => new(rawPath, blob, WriteIdempotent: false);
private static MqttDriver SparkplugDriver(params RawTagEntry[] tags)
=> new(
new MqttDriverOptions
{
Mode = MqttMode.SparkplugB,
Sparkplug = new MqttSparkplugOptions { GroupId = SpGroup },
RawTags = tags,
},
"d",
null);
/// <summary>
/// Feeds a real NBIRTH through the real codec + ingest state machine — no broker, and no
/// test-only shortcut on the driver. <c>Dispatch</c> is the method MQTTnet's dispatcher thread
/// reaches, so this exercises the production path end to end.
/// </summary>
private static void FeedNodeBirth(MqttDriver driver, ulong seq = 0, ulong bdSeq = 1)
{
var payload = new Payload { Seq = seq, Timestamp = 1721822400000UL };
payload.Metrics.Add(new Payload.Types.Metric
{
Name = "bdSeq",
Datatype = (uint)TahuDataType.Uint64,
LongValue = bdSeq,
});
payload.Metrics.Add(SpBirthMetric("Uptime", 7UL, TahuDataType.Int64));
driver.Sparkplug!.Dispatch(
new SparkplugTopic(SparkplugMessageType.NBIRTH, SpGroup, SpNode, null, null),
SparkplugCodec.Decode(payload.ToByteArray()));
}
private static void FeedDeviceBirth(
MqttDriver driver,
ulong seq,
params (string Name, ulong Alias, TahuDataType Type)[] metrics)
=> FeedDeviceBirth(driver, SpDevice, seq, metrics);
private static void FeedDeviceBirth(
MqttDriver driver,
string device,
ulong seq,
params (string Name, ulong Alias, TahuDataType Type)[] metrics)
{
var payload = new Payload { Seq = seq, Timestamp = 1721822400000UL };
foreach (var (name, alias, type) in metrics)
{
payload.Metrics.Add(SpBirthMetric(name, alias, type));
}
driver.Sparkplug!.Dispatch(
new SparkplugTopic(SparkplugMessageType.DBIRTH, SpGroup, SpNode, device, null),
SparkplugCodec.Decode(payload.ToByteArray()));
}
/// <summary>
/// A birth metric declaring name/alias/datatype and <b>no value</b> — a birth's job here is to
/// declare the catalog, and these tests assert on the catalog, never on a published value.
/// </summary>
private static Payload.Types.Metric SpBirthMetric(string name, ulong alias, TahuDataType type)
=> new() { Name = name, Alias = alias, Datatype = (uint)type };
// ---- test doubles ----
/// <summary>