diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriver.cs
index cd495944..a6814efb 100644
--- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriver.cs
+++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriver.cs
@@ -1,3 +1,5 @@
+using System.Collections.Concurrent;
+using System.Collections.Frozen;
using System.Text.Json;
using Microsoft.Extensions.Logging;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
@@ -18,9 +20,10 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt;
/// therefore replays exactly the raw tags the deploy artifact
/// authored () and nothing else, and
/// stays so the
-/// universal browser never treats broker traffic as a tag catalogue. The policy is
-/// : the authored set is fully known
-/// synchronously, so re-running discovery on a timer can only produce the same tree.
+/// universal browser never treats broker traffic as a tag catalogue. In
+/// the policy is :
+/// the authored set — and every authored tag's datatype — is fully known synchronously, so
+/// re-running discovery on a timer can only produce the same tree.
///
///
/// Two ingest paths, one driver. routes through
@@ -32,12 +35,18 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt;
/// driver-owned , so IReadable reads one store either way.
///
///
-/// still never fires. Sparkplug B is the mode that
-/// eventually raises it — a DBIRTH can introduce metrics the previous birth did not carry —
-/// but the policy flip to and the wiring
-/// from to
-/// are Task 22. The seams exist on both sides; nothing
-/// connects them yet.
+/// fires in Sparkplug B only, and only on a real change.
+/// A Sparkplug tag's dataType is optional — the birth certificate declares it — so the
+/// discovered tree fills in asynchronously and the policy is
+/// .
+/// is wired to through
+/// , which fires only when the birth's metric-name SET
+/// differs from the last one seen for that scope (and only for scopes an authored tag actually
+/// binds). That gate is the anti-storm mechanism, not an optimisation: rediscovery triggers an
+/// address-space rebuild, edge nodes re-birth freely — on their own timer, on every reconnect
+/// via the late-join rebirth, and once per NCMD the ingestor's gap policy sends — and firing on
+/// each of those would turn a healthy plant into a permanent rebuild loop. Plain mode never
+/// raises it at all; its authored set changes only by redeploy.
///
///
/// Composition order is load-bearing.
@@ -65,6 +74,13 @@ public sealed class MqttDriver
///
private const long ApproxBytesPerAuthoredTag = 512;
+ ///
+ /// The single folder streams every variable under — and therefore
+ /// the only subtree path a rediscovery ScopeHint can name truthfully. See
+ /// .
+ ///
+ private const string DiscoveryFolderName = "Mqtt";
+
/// How often the host-connectivity poll samples .
private static readonly TimeSpan HostProbeInterval = TimeSpan.FromSeconds(1);
@@ -80,6 +96,23 @@ public sealed class MqttDriver
private readonly object _hostLock = new();
+ ///
+ /// Per authored Sparkplug scope, the signature of the metric-name set its most recent birth
+ /// declared — the rediscovery change gate's memory. Written on MQTTnet's dispatcher thread and
+ /// read nowhere else, but concurrent because a reconnect's birth flood and the dispatcher are
+ /// not guaranteed to be the same thread forever.
+ ///
+ ///
+ /// Deliberately NOT cleared on a death, a reconnect or an ingest rebuild. The ingestor
+ /// drops its whole cache on every reconnect (an alias may
+ /// have been rebound while the driver was away) — but "the driver forgot" is not "the address
+ /// space changed". Clearing this alongside it would make every reconnect's late-join rebirth
+ /// flood look like a tree full of brand-new scopes and fire one rediscovery per authored scope
+ /// on every single flap, which is the storm this map exists to prevent. It survives so that an
+ /// identical rebirth stays silent.
+ ///
+ private readonly ConcurrentDictionary _birthSignatures = new();
+
/// Guards / / against each other.
private readonly SemaphoreSlim _lifecycleGate = new(1, 1);
@@ -109,6 +142,23 @@ public sealed class MqttDriver
///
private IReadOnlyList _authoredRawPaths = [];
+ ///
+ /// The Sparkplug scopes at least one authored tag binds — empty outside
+ /// . Rebuilt wholesale by
+ /// alongside the authored table it is derived from.
+ ///
+ ///
+ /// The rediscovery gate's first question. The ingestor applies a birth for any device
+ /// under an authored edge node (it has to — a DBIRTH is a sequenced member of that node's
+ /// stream), so on a large plant it observes births for devices this deployment never reads. Such
+ /// a birth cannot change 's output by construction, because discovery
+ /// replays the authored set — so firing rediscovery for it would rebuild the address space in
+ /// response to traffic the configuration has no opinion about. Gating on this set also bounds
+ /// by configuration rather than by whatever the plant
+ /// happens to publish — the same rule the ingestor's own authored-edge-node filter follows.
+ ///
+ private FrozenSet _authoredScopes = FrozenSet.Empty;
+
private MqttConnection? _connection;
private CancellationTokenSource? _hostProbeCts;
@@ -357,11 +407,29 @@ public sealed class MqttDriver
///
///
- /// Plain mode discovers synchronously from the authored set, so one pass is all there is.
- /// Sparkplug B (P2) fills its metric set in asynchronously from birth certificates and
- /// switches this to .
+ ///
+ /// Plain mode discovers synchronously from the authored set — the RawPaths and their
+ /// datatypes are both in the deploy artifact — so one pass is all there is.
+ ///
+ ///
+ /// Sparkplug B is because half of what
+ /// it discovers arrives on the wire: dataType is optional in the Sparkplug tag shape
+ /// (the birth certificate declares it), so a tag authored with nothing but its
+ /// (group, node, device, metric) tuple can only be streamed with the record's default
+ /// type until its NBIRTH/DBIRTH lands. The host's UntilStable contract — re-run
+ /// discovery until the captured set is non-empty and its signature repeats — is exactly the
+ /// retry that lets those passes catch a birth that had not yet arrived at connect.
+ ///
+ ///
+ /// The retry is a floor, not the mechanism. The host's stability signature is built
+ /// from FullReferences alone, so a pass whose only change is a filled-in datatype still
+ /// looks "stable" and the loop stops. That is why the same fill-in also fires
+ /// (see ): a birth arriving
+ /// after the loop settled is the case the timer cannot cover.
+ ///
///
- public DiscoveryRediscoverPolicy RediscoverPolicy => DiscoveryRediscoverPolicy.Once;
+ public DiscoveryRediscoverPolicy RediscoverPolicy =>
+ IsSparkplug ? DiscoveryRediscoverPolicy.UntilStable : DiscoveryRediscoverPolicy.Once;
///
///
@@ -375,10 +443,21 @@ public sealed class MqttDriver
/// Streams the authored tag set — and only that — into .
///
///
- /// The enumeration source is the driver's own list of registered RawPaths, never anything
- /// observed on the wire, so no amount of broker traffic can add a node. A RawPath whose
- /// TagConfig failed to map is absent from the manager and is skipped here too — it surfaces
- /// as BadNodeIdUnknown on read rather than as a wrongly-typed variable.
+ ///
+ /// The enumeration source is the driver's own list of registered RawPaths, never anything
+ /// observed on the wire, so no amount of broker traffic can add a node. A RawPath whose
+ /// TagConfig failed to map is absent from the manager and is skipped here too — it surfaces
+ /// as BadNodeIdUnknown on read rather than as a wrongly-typed variable.
+ ///
+ ///
+ /// Sparkplug B: the datatype is resolved per pass, the tag set never is. An authored
+ /// tag is streamed on every pass, birth or no birth — it is part of the declared
+ /// configuration, not something the plant grants by publishing, and hiding it until a birth
+ /// arrives would make the node's existence depend on wire traffic (and, on a deployment
+ /// whose other tags do carry an authored type, would hide it behind a "non-empty and stable"
+ /// verdict the host reaches without it). What a birth changes is the type: see
+ /// .
+ ///
///
/// The address space builder to stream discovered nodes into.
/// Cancellation for the discovery operation.
@@ -388,7 +467,7 @@ public sealed class MqttDriver
ArgumentNullException.ThrowIfNull(builder);
cancellationToken.ThrowIfCancellationRequested();
- var folder = builder.Folder("Mqtt", "Mqtt");
+ var folder = builder.Folder(DiscoveryFolderName, DiscoveryFolderName);
foreach (var rawPath in _authoredRawPaths)
{
if (!TryResolve(rawPath, out var def))
@@ -398,7 +477,7 @@ public sealed class MqttDriver
folder.Variable(def.Name, def.Name, new DriverAttributeInfo(
FullName: def.Name,
- DriverDataType: def.DataType,
+ DriverDataType: ResolveDiscoveredDataType(def),
IsArray: false,
ArrayDim: null,
// MQTT ingest is one-way in P1 — this driver implements no IWritable leg, so every
@@ -412,6 +491,43 @@ public sealed class MqttDriver
return Task.CompletedTask;
}
+ ///
+ /// The datatype reports for one authored tag: the authored type when
+ /// the operator declared one, otherwise the type the metric's live birth certificate declared,
+ /// otherwise the record's default.
+ ///
+ ///
+ ///
+ /// The precedence is the ingest path's, restated. SparkplugIngestor.PublishMetric
+ /// coerces every value with DataTypeAuthored ? authored : birth; a discovery surface
+ /// that disagreed with it would advertise a type no value ever arrives as. The two must be
+ /// changed together.
+ ///
+ ///
+ /// An unsupported Sparkplug type (DataSet / Template / PropertySet / Unknown) maps to
+ /// and falls back rather than blanking the tag — the ingest path
+ /// already refuses those metrics once, loudly, and a discovery pass is not the place to
+ /// relitigate it.
+ ///
+ ///
+ /// The authored tag definition.
+ /// The effective data type for this pass.
+ private DriverDataType ResolveDiscoveredDataType(MqttTagDefinition def)
+ {
+ if (def.DataTypeAuthored || _sparkplug is not { } sparkplug)
+ {
+ return def.DataType;
+ }
+
+ if (def.GroupId is not { } groupId || def.EdgeNodeId is not { } edgeNodeId || def.MetricName is not { } metric)
+ {
+ return def.DataType;
+ }
+
+ var table = sparkplug.Births.Find(new SparkplugScope(groupId, edgeNodeId, def.DeviceId));
+ return table?.ResolveByName(metric)?.DataType.ToDriverDataType() ?? def.DataType;
+ }
+
// ---- ISubscribable (straight through to the manager) ----
///
@@ -495,6 +611,85 @@ public sealed class MqttDriver
internal void RaiseRediscoveryNeeded(string reason, string? scopeHint = null)
=> OnRediscoveryNeeded?.Invoke(this, new RediscoveryEventArgs(reason, scopeHint));
+ ///
+ /// The rediscovery decision: a birth changed what can produce only
+ /// when it is for an authored scope and its metric-name set differs from the last
+ /// one seen for that scope. Anything else is dropped.
+ ///
+ ///
+ ///
+ /// Both conditions are load-bearing, and both are anti-storm. Rediscovery triggers an
+ /// address-space rebuild — expensive and disruptive on a running server. An edge node
+ /// re-births freely: on its own schedule, on every reconnect (the ingestor's late-join
+ /// rebirth fan-out), and once per rebirth NCMD its gap policy sends. Firing on each of those
+ /// is a permanent rebuild loop against a plant that is behaving perfectly. Firing on
+ /// changed metric set only is the primary defence; the authored-scope filter is the
+ /// second, and removes the whole class of churn sourced from devices this deployment does
+ /// not read (see ).
+ ///
+ ///
+ /// The signature is order-insensitive — ordered-distinct, ordinal. Nothing obliges an
+ /// edge node to keep its birth's metric ordering stable, and a reordered birth describes the
+ /// same address space; comparing the raw sequence would leak the storm back in.
+ ///
+ ///
+ /// No additional debounce or coalescing. One was considered and rejected: the change
+ /// gate already collapses the unbounded sources (rebirth timers, reconnect floods, NCMD
+ /// answers) to zero, and what survives it is a genuine metric-set change — of which a given
+ /// scope has a handful in its lifetime, each one a real edit to the served tree. A timer on
+ /// top would only delay a rebuild that must happen, and would need its own trailing-edge
+ /// flush to avoid dropping the last change entirely. If a consumer ever needs coalescing
+ /// across many scopes birthing at once (a whole-plant restart), that belongs in the consumer,
+ /// which alone knows what a rebuild costs it — today there is no such consumer at all.
+ ///
+ ///
+ /// Runs on MQTTnet's dispatcher thread (via the ingestor's own contained raise), so it
+ /// does no I/O and never throws — catches a throwing
+ /// subscriber, but relying on that would still have cost the rest of the birth's fan-out.
+ ///
+ ///
+ private void OnBirthObserved(object? sender, SparkplugBirthObservedEventArgs e)
+ {
+ if (!_authoredScopes.Contains(e.Scope))
+ {
+ // A birth for a scope no authored tag binds cannot change the authored-only discovery tree.
+ return;
+ }
+
+ var signature = BirthSignature(e.MetricNames);
+ if (_birthSignatures.TryGetValue(e.Scope, out var previous)
+ && string.Equals(previous, signature, StringComparison.Ordinal))
+ {
+ return; // Same metric set as last time — nothing to rediscover.
+ }
+
+ _birthSignatures[e.Scope] = signature;
+
+ var reason = previous is null
+ ? $"Sparkplug birth observed for '{e.Scope}' ({e.MetricNames.Count} metric(s))"
+ : $"Sparkplug rebirth changed the metric set for '{e.Scope}' ({e.MetricNames.Count} metric(s))";
+
+ _logger?.LogInformation(
+ "MQTT driver '{DriverId}': {Reason}; requesting rediscovery.",
+ _driverInstanceId,
+ reason);
+
+ // ScopeHint names the folder DiscoverAsync actually streams under — deliberately NOT the
+ // Sparkplug scope path. The discovered tree is flat (one 'Mqtt' folder of RawPath-named
+ // variables); it has no edge-node/device folders, so a hint of 'Plant1/EdgeA/Filler1' would
+ // name a subtree that does not exist and a consumer scoping a rebuild on it would rebuild
+ // nothing. The scope belongs in the Reason, which is the documented diagnostic field.
+ RaiseRediscoveryNeeded(reason, DiscoveryFolderName);
+ }
+
+ /// An order-insensitive, ordinal signature of a birth's metric-name set.
+ /// The names the birth declared.
+ /// The comparable signature.
+ private static string BirthSignature(IReadOnlyList metricNames)
+ => string.Join(
+ '\u0001', // A separator no Sparkplug metric name can contain, so two sets cannot alias.
+ metricNames.Distinct(StringComparer.Ordinal).Order(StringComparer.Ordinal));
+
// ---- disposal ----
///
@@ -764,6 +959,12 @@ public sealed class MqttDriver
// The published reference is the RawPath the ingestor already stamped on the args; the driver
// only re-raises with itself as sender.
ingestor.OnDataChange += (_, args) => OnDataChange?.Invoke(this, args);
+
+ // The ingestor detects a birth; the driver decides whether it changed the address space. Wired
+ // here rather than in InitializeAsync so a driver rebuilt for a config delta re-arms it — an
+ // ingest rebuild that silently dropped this subscription would leave a Sparkplug driver whose
+ // datatypes fill in and whose rediscovery never fires again, with nothing to show for it.
+ ingestor.BirthObserved += OnBirthObserved;
return ingestor;
}
@@ -779,6 +980,20 @@ public sealed class MqttDriver
// Enumerate in authoring order; the manager owns which of these actually mapped, and
// DiscoverAsync skips the rest.
_authoredRawPaths = [.. options.RawTags.Select(t => t.RawPath)];
+ _authoredScopes = BuildAuthoredScopes();
+
+ // Drop change-gate memory for scopes this deploy stopped authoring — the same housekeeping
+ // SparkplugIngestor.Register does for its trackers, and for the same reason: without it a scope
+ // nobody reads keeps a slot across every future redeploy. Re-authoring such a scope later means
+ // its next birth reads as new and fires once, which is correct — a newly authored scope IS a
+ // change to the discovered tree.
+ foreach (var scope in _birthSignatures.Keys)
+ {
+ if (!_authoredScopes.Contains(scope))
+ {
+ _birthSignatures.TryRemove(scope, out _);
+ }
+ }
if (mapped != options.RawTags.Count)
{
@@ -791,6 +1006,33 @@ public sealed class MqttDriver
}
}
+ ///
+ /// The Sparkplug scopes the currently registered authored tags bind, derived from the same
+ /// definitions enumerates — empty outside
+ /// , and empty for any tag whose TagConfig did not map.
+ ///
+ /// The authored scope set.
+ private FrozenSet BuildAuthoredScopes()
+ {
+ if (_sparkplug is null)
+ {
+ return FrozenSet.Empty;
+ }
+
+ var scopes = new HashSet();
+ foreach (var rawPath in _authoredRawPaths)
+ {
+ if (TryResolve(rawPath, out var def)
+ && def.GroupId is { } groupId
+ && def.EdgeNodeId is { } edgeNodeId)
+ {
+ scopes.Add(new SparkplugScope(groupId, edgeNodeId, def.DeviceId));
+ }
+ }
+
+ return scopes.ToFrozenSet();
+ }
+
///
/// Whether two option sets describe the same broker session. Note the explicit
/// : the identity projections are object-typed
diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttDriverDiscoveryTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttDriverDiscoveryTests.cs
index 4fbd1c96..60a10cbd 100644
--- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttDriverDiscoveryTests.cs
+++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttDriverDiscoveryTests.cs
@@ -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
+ // =================================================================================
+
+ ///
+ /// Sparkplug's discovered shape is not fully known at connect: a tag authored without a
+ /// dataType takes its type from the birth certificate, which arrives asynchronously.
+ /// That is precisely the contract — the
+ /// host re-runs discovery until the captured set settles.
+ ///
+ [Fact]
+ public void SparkplugMode_RediscoverPolicy_IsUntilStable()
+ => SparkplugDriver().RediscoverPolicy.ShouldBe(DiscoveryRediscoverPolicy.UntilStable);
+
+ ///
+ /// The mode gate, asserted from the other side: plain MQTT's authored set is fully known
+ /// synchronously, so it must stay . A blanket flip
+ /// to UntilStable would make every plain deployment re-run discovery on a timer for a
+ /// tree that can never change.
+ ///
+ [Fact]
+ public void PlainMode_RediscoverPolicy_StaysOnce()
+ => PlainDriver(Tag("Plant/Mqtt/dev1/Temp", "f/t")).RediscoverPolicy.ShouldBe(DiscoveryRediscoverPolicy.Once);
+
+ ///
+ /// 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.
+ ///
+ [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);
+ }
+
+ ///
+ /// The primary anti-storm pin. 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.
+ ///
+ [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);
+ }
+
+ ///
+ /// 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.
+ ///
+ [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);
+ }
+
+ ///
+ /// 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.
+ ///
+ [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);
+ }
+
+ ///
+ /// A birth for a device NO authored tag names cannot change '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.
+ ///
+ [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);
+ }
+
+ ///
+ /// The rediscovery event names the subtree this driver's discovery actually emits (the
+ /// Mqtt 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.
+ ///
+ [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}");
+ }
+
+ ///
+ /// The datatype fill-in. dataType 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.
+ ///
+ [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);
+ }
+
+ ///
+ /// The authored dataType wins over the birth's — the same precedence the ingest path
+ /// applies when it coerces a value (DataTypeAuthored ? authored : birth). A discovery
+ /// surface that disagreed with the publish surface would report a type no value ever arrives as.
+ ///
+ [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);
+ }
+
+ ///
+ /// A birth declaring a Sparkplug type this driver cannot map (DataSet / Template / PropertySet
+ /// / Unknown) must not blank the tag's type — ToDriverDataType() returns null for those,
+ /// and discovery falls back to the authored/default type rather than to nothing.
+ ///
+ [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";
+
+ /// A Sparkplug tag blob. dataType is omitted unless supplied — it is optional.
+ 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);
+
+ ///
+ /// Feeds a real NBIRTH through the real codec + ingest state machine — no broker, and no
+ /// test-only shortcut on the driver. Dispatch is the method MQTTnet's dispatcher thread
+ /// reaches, so this exercises the production path end to end.
+ ///
+ 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()));
+ }
+
+ ///
+ /// A birth metric declaring name/alias/datatype and no value — a birth's job here is to
+ /// declare the catalog, and these tests assert on the catalog, never on a published value.
+ ///
+ private static Payload.Types.Metric SpBirthMetric(string name, ulong alias, TahuDataType type)
+ => new() { Name = name, Alias = alias, Datatype = (uint)type };
+
// ---- test doubles ----
///