diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttTagDefinition.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttTagDefinition.cs index d4f6663b..15c3c3eb 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttTagDefinition.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttTagDefinition.cs @@ -9,12 +9,20 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt; /// See docs/plans/2026-07-15-mqtt-sparkplug-driver-design.md §5.2 / §5.3. /// /// The record carries both ingest shapes: the Plain-MQTT fields -/// ( … ), populated in P1, and the Sparkplug B -/// descriptor fields (, , -/// , ), which are deliberate stubs in P1 -/// — always until the P2 tasks (15–26) teach the parser to read them. -/// A Sparkplug tag is resolved by the stable (group, node, device, metricName) tuple, -/// never by the per-birth metric alias. +/// ( … ), populated by +/// , and the Sparkplug B descriptor fields +/// (, , , +/// ), populated by +/// (Task 21). A Sparkplug tag is +/// resolved by the stable (group, node, device, metricName) tuple, never by the +/// per-birth metric alias — an alias may be reused across a rebirth for a different metric. +/// +/// +/// Which factory runs is chosen by the driver's , never sniffed from +/// the blob. A blob carrying both shapes' keys is legal (the AdminUI editor preserves +/// unknown keys through a load→save, so a tag retyped from Plain to Sparkplug keeps its old +/// topic), and a heuristic would make the same authored tag mean two different things +/// depending on which key happened to survive. /// /// /// @@ -44,15 +52,23 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt; /// Whether the broker's retained message for seeds this tag's initial value /// on subscribe (the OPC UA initial-data convention). Defaults to . /// -/// P2 stub — the Sparkplug group id. Always in P1. -/// P2 stub — the Sparkplug edge-node id. Always in P1. +/// +/// Whether came from an authored dataType key or is merely this +/// record's default. Load-bearing in Sparkplug mode only, where dataType is optional: +/// a Sparkplug metric's type is declared by its own birth certificate, so an unauthored tag must +/// take the type the NBIRTH/DBIRTH declared rather than silently coercing every value to the +/// default. Both factories set it from the key's presence so it +/// means the same thing in either mode; nothing on the Plain ingest path reads it. +/// +/// The Sparkplug group id; for a Plain-mode tag. +/// The Sparkplug edge-node id; for a Plain-mode tag. /// -/// P2 stub — the Sparkplug device id (absent for node-level metrics). Always -/// in P1. +/// The Sparkplug device id, or for a metric published by the edge node +/// itself (and for every Plain-mode tag). /// /// -/// P2 stub — the Sparkplug metric name, the tag's stable identity across rebirths. Always -/// in P1. +/// The Sparkplug metric name — the tag's stable identity across rebirths, and the key the ingest +/// state machine binds by. for a Plain-mode tag. /// public sealed record MqttTagDefinition( string Name, @@ -62,6 +78,7 @@ public sealed record MqttTagDefinition( DriverDataType DataType, int? Qos, bool RetainSeed, + bool DataTypeAuthored = true, string? GroupId = null, string? EdgeNodeId = null, string? DeviceId = null, diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttTagDefinitionFactory.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttTagDefinitionFactory.cs index 697256b1..239d84ab 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttTagDefinitionFactory.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttTagDefinitionFactory.cs @@ -20,9 +20,13 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt; /// deploy instead of silently going dark at runtime. /// /// -/// Sparkplug-descriptor parsing (groupId / edgeNodeId / deviceId / -/// metricName) is a deliberate P1 stub — the fields exist on -/// but are never populated here until the P2 tasks fill them. +/// Two runtime entry points, one per ingest shape — (Plain) +/// and (Sparkplug B). The caller picks by the driver's +/// ; neither sniffs the blob. That is deliberate: the AdminUI editor +/// preserves unknown keys through a load→save, so a tag retyped from Plain to Sparkplug keeps +/// its old topic and a presence heuristic would make one authored blob mean two +/// different things depending on which key happened to survive. The driver's mode is +/// the single authority. /// /// /// No ToTagConfig inverse — a deliberate YAGNI call. The six sibling factories @@ -80,6 +84,7 @@ public static class MqttTagDefinitionFactory // (→ BadNodeIdUnknown) instead of silently defaulting to a wrong-typed Good. if (!TagConfigJson.TryReadEnumStrict(root, "payloadFormat", MqttPayloadFormat.Json, out var payloadFormat)) return false; + var dataTypeAuthored = root.TryGetProperty("dataType", out _); if (!TagConfigJson.TryReadEnumStrict(root, "dataType", DriverDataType.String, out var dataType)) return false; @@ -98,7 +103,94 @@ public static class MqttTagDefinitionFactory JsonPath: jsonPath, DataType: dataType, Qos: qos, - RetainSeed: ReadBoolOrDefault(root, "retainSeed", defaultValue: true)); + RetainSeed: ReadBoolOrDefault(root, "retainSeed", defaultValue: true), + DataTypeAuthored: dataTypeAuthored); + return true; + } + catch (JsonException) { return false; } + catch (FormatException) { return false; } + catch (InvalidOperationException) { return false; } + } + + /// + /// Maps an authored TagConfig object to a Sparkplug B definition keyed by + /// . The tag's binding identity is the stable + /// (groupId, edgeNodeId, deviceId?, metricName) tuple; deviceId is optional (absent + /// means the metric is published by the edge node itself), and there is no topic — + /// a Sparkplug driver subscribes one group-wide filter and routes by the decoded topic + birth + /// certificate, never by a per-tag topic. Never throws. + /// + /// + /// + /// dataType is OPTIONAL here, unlike in Plain mode. A Sparkplug metric declares + /// its own datatype in its birth certificate, so an unauthored tag legitimately takes the + /// type the birth declares — that is the whole point of Task 22's UntilStable + /// discovery. It is still read strictly when present (a typo'd value rejects the tag, + /// exactly as in Plain mode) and the outcome is recorded on + /// so the ingest path can tell "the operator + /// declared String" from "nobody declared anything and the record defaulted". + /// + /// + /// Plain-shape keys are read but not required. topic/payloadFormat/ + /// jsonPath are meaningless to the Sparkplug ingest path and are deliberately NOT + /// validated — a blob retyped in the editor keeps them, and rejecting the tag for a stale + /// leftover key would take a correctly-authored Sparkplug tag dark with no stated cause. + /// + /// + /// The authored equipment-tag TagConfig JSON. + /// The tag's RawPath — becomes the definition's identity (Name). + /// The mapped definition when this returns . + /// when the blob carries a usable Sparkplug descriptor. + public static bool FromSparkplugTagConfig(string tagConfig, string rawPath, out MqttTagDefinition def) + { + def = null!; + if (string.IsNullOrWhiteSpace(tagConfig) || tagConfig[0] != '{') return false; + try + { + using var doc = JsonDocument.Parse(tagConfig); + var root = doc.RootElement; + if (root.ValueKind != JsonValueKind.Object) return false; + + // The binding tuple. Without all three of these there is nothing a birth certificate could + // ever bind the tag to, so an absent/blank value is a hard reject (→ BadNodeIdUnknown) + // rather than a definition that can never resolve and reports nothing about why. + var groupId = ReadString(root, "groupId"); + var edgeNodeId = ReadString(root, "edgeNodeId"); + var metricName = ReadString(root, "metricName"); + if (string.IsNullOrWhiteSpace(groupId) + || string.IsNullOrWhiteSpace(edgeNodeId) + || string.IsNullOrWhiteSpace(metricName)) + { + return false; + } + + // Optional: a node-level metric has no device segment. Blank normalises to absent so + // "deviceId":"" and an omitted key describe the same scope rather than two. + var deviceId = ReadString(root, "deviceId"); + if (string.IsNullOrWhiteSpace(deviceId)) deviceId = null; + + // Strict, but only when present — see the remarks. + var dataTypeAuthored = root.TryGetProperty("dataType", out _); + if (!TagConfigJson.TryReadEnumStrict(root, "dataType", DriverDataType.String, out var dataType)) + return false; + + if (!TryReadQosStrict(root, out var qos)) return false; + + def = new MqttTagDefinition( + Name: rawPath, + // Plain-shape fields carried through verbatim; the Sparkplug ingest path reads none of + // them, and a retyped blob's leftovers must not change what the tag means. + Topic: ReadString(root, "topic"), + PayloadFormat: MqttPayloadFormat.Json, + JsonPath: RootJsonPath, + DataType: dataType, + Qos: qos, + RetainSeed: ReadBoolOrDefault(root, "retainSeed", defaultValue: true), + DataTypeAuthored: dataTypeAuthored, + GroupId: groupId, + EdgeNodeId: edgeNodeId, + DeviceId: deviceId, + MetricName: metricName); return true; } catch (JsonException) { return false; } diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttConnection.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttConnection.cs index ebb6a486..1733fefc 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttConnection.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttConnection.cs @@ -237,10 +237,17 @@ public sealed class MqttConnectRejectedException : Exception /// socket no later dispose could ever reach. Now the losing side of that race disposes the /// client it created and reports . /// -/// Sparkplug rebirth-on-reconnect (Task 21, which hangs off ) is -/// deliberately not implemented here. +/// +/// The publish leg is deliberately narrow. exists to satisfy +/// — whose entire production caller is +/// — not to make this a general MQTT publisher. This +/// driver has no IWritable leg in v1, so a Sparkplug rebirth NCMD is the only +/// outbound application message it ever sends. Sparkplug rebirth-on-reconnect itself is decided +/// by hanging off ; this type +/// only carries the bytes. +/// /// -public sealed class MqttConnection : IAsyncDisposable, IMqttSubscribeTransport +public sealed class MqttConnection : IAsyncDisposable, IMqttSubscribeTransport, Sparkplug.IMqttPublishTransport { private readonly string _driverId; @@ -994,6 +1001,86 @@ public sealed class MqttConnection : IAsyncDisposable, IMqttSubscribeTransport return outcomes; } + /// + /// Publishes one application message, bounded by + /// so a broker that accepts PUBLISH and + /// never completes it cannot park the caller — the same frozen-peer rule every other wait here + /// follows. + /// + /// + /// + /// A refused PUBLISH throws. MQTTnet reports a broker rejection as a + /// result (as it does for CONNACK — see ), + /// so a caller that treated "did not throw" as "published" would silently drop a rebirth + /// NCMD against a broker whose ACL denies the NCMD topic and then wait forever for the + /// birth it never asked for. + /// + /// + /// Publishing on a down session is refused up front rather than left to MQTTnet's + /// own exception, so the message names the driver and the broker. A rebirth request lost to + /// a reconnect is not a fault: the reconnect path re-requests one itself. + /// + /// + /// The concrete topic to publish on. + /// The message body. + /// Requested QoS, 0–2; out-of-range values clamp. + /// The MQTT retain flag. + /// Caller cancellation; linked with the publish deadline. + /// A task that completes when the broker has accepted the message. + /// The publish deadline elapsed. + /// was cancelled. + /// The connection was disposed. + /// There is no established session, or the broker refused the message. + public async Task PublishAsync( + string topic, + byte[] payload, + int qos, + bool retain, + CancellationToken cancellationToken) + { + ArgumentException.ThrowIfNullOrEmpty(topic); + ArgumentNullException.ThrowIfNull(payload); + ObjectDisposedException.ThrowIf(Disposed, this); + + var client = _client; + if (client is null || !client.IsConnected) + { + throw new InvalidOperationException( + $"MQTT driver '{_driverId}': cannot publish '{topic}' — there is no established session to " + + $"{_options.Host}:{_options.Port}."); + } + + var message = new MqttApplicationMessageBuilder() + .WithTopic(topic) + .WithPayload(payload) + .WithQualityOfServiceLevel(MapQos(qos)) + .WithRetainFlag(retain) + .Build(); + + using var deadline = new CancellationTokenSource(TimeSpan.FromSeconds(_options.ConnectTimeoutSeconds)); + using var linked = CancellationTokenSource.CreateLinkedTokenSource( + cancellationToken, + deadline.Token, + _lifetimeCts.Token); + + MqttClientPublishResult result; + try + { + result = await client.PublishAsync(message, linked.Token).ConfigureAwait(false); + } + catch (Exception ex) + { + throw Classify(ex, cancellationToken, deadline, operation: "publish"); + } + + if (result is not null && !result.IsSuccess) + { + throw new InvalidOperationException( + $"MQTT driver '{_driverId}': broker at {_options.Host}:{_options.Port} refused PUBLISH " + + $"'{topic}': {result.ReasonCode}."); + } + } + /// Maps a configured QoS integer onto MQTTnet's enum, clamping an out-of-range value. private static MqttQualityOfServiceLevel MapQos(int qos) => qos switch { 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 da500db5..cd495944 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriver.cs @@ -1,6 +1,7 @@ using System.Text.Json; using Microsoft.Extensions.Logging; using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Sparkplug; namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt; @@ -22,11 +23,21 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt; /// synchronously, so re-running discovery on a timer can only produce the same tree. /// /// -/// P1 scope — no Sparkplug. is implemented but -/// never fires in : the -/// authored set only changes by redeploy. Sparkplug B flips the policy to -/// and raises the event on DBIRTH — that -/// is a later task, and the seam here () exists for it. +/// Two ingest paths, one driver. routes through +/// (authored topics); +/// routes through (one group-wide filter, birth certificates, +/// alias resolution, seq gaps, rebirth NCMDs). Exactly one is live, selected by +/// and rebuilt whenever +/// changes anything either captures. They share the +/// 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. /// /// /// Composition order is load-bearing. @@ -83,6 +94,14 @@ public sealed class MqttDriver private MqttDriverOptions _options; private MqttSubscriptionManager _subscriptions; + /// + /// The Sparkplug B ingest path, or in . + /// Exactly one of this and is fed authored tags and attached to the + /// connection; the other exists but is never registered against, so it can neither route a + /// message nor answer a resolve. + /// + private SparkplugIngestor? _sparkplug; + /// /// RawPaths of the currently registered authored tags, in authoring order — the discovery /// enumeration set. The manager resolves a RawPath to its definition but does not enumerate, @@ -115,6 +134,7 @@ public sealed class MqttDriver _driverInstanceId = driverInstanceId; _logger = logger; _subscriptions = CreateSubscriptionManager(options); + _sparkplug = CreateSparkplugIngestor(options); RegisterAuthoredTags(options); } @@ -141,6 +161,17 @@ public sealed class MqttDriver /// internal MqttSubscriptionManager Subscriptions => _subscriptions; + /// + /// The Sparkplug B ingest path, or outside + /// . Internal: the shell tests drive + /// to simulate an edge node without a broker, and + /// Tasks 22/23 read its . + /// + internal SparkplugIngestor? Sparkplug => _sparkplug; + + /// Whether this driver instance ingests Sparkplug B rather than plain topics. + private bool IsSparkplug => _options.Mode == MqttMode.SparkplugB; + /// /// Test seam: awaited inside while the lifecycle gate is /// held, immediately before the broker connect. Lets a test park a lifecycle operation @@ -360,7 +391,7 @@ public sealed class MqttDriver var folder = builder.Folder("Mqtt", "Mqtt"); foreach (var rawPath in _authoredRawPaths) { - if (!_subscriptions.TryResolve(rawPath, out var def)) + if (!TryResolve(rawPath, out var def)) { continue; } @@ -388,11 +419,24 @@ public sealed class MqttDriver IReadOnlyList fullReferences, TimeSpan publishingInterval, CancellationToken cancellationToken) - => _subscriptions.SubscribeAsync(fullReferences, publishingInterval, cancellationToken); + => _sparkplug is { } sparkplug + ? sparkplug.SubscribeAsync(fullReferences, publishingInterval, cancellationToken) + : _subscriptions.SubscribeAsync(fullReferences, publishingInterval, cancellationToken); /// public Task UnsubscribeAsync(ISubscriptionHandle handle, CancellationToken cancellationToken) - => _subscriptions.UnsubscribeAsync(handle, cancellationToken); + => _sparkplug is { } sparkplug + ? sparkplug.UnsubscribeAsync(handle, cancellationToken) + : _subscriptions.UnsubscribeAsync(handle, cancellationToken); + + /// Resolves a RawPath through whichever ingest path this driver's mode selected. + /// The driver wire reference. + /// The definition when this returns . + /// when the reference is an authored tag of the live ingest path. + private bool TryResolve(string rawPath, out MqttTagDefinition def) + => _sparkplug is { } sparkplug + ? sparkplug.TryResolve(rawPath, out def) + : _subscriptions.TryResolve(rawPath, out def); // ---- IReadable ---- @@ -417,7 +461,9 @@ public sealed class MqttDriver var results = new DataValueSnapshot[fullReferences.Count]; for (var i = 0; i < fullReferences.Count; i++) { - results[i] = _subscriptions.Values.Read(fullReferences[i]); + // The driver-owned cache, not the live ingest path's: both are handed this same instance, + // and reading it directly means a mid-flight ingest rebuild can never blank a read. + results[i] = _values.Read(fullReferences[i]); } return Task.FromResult>(results); @@ -539,9 +585,16 @@ public sealed class MqttDriver var connection = new MqttConnection(_options, _driverInstanceId, _logger); // MUST precede the connect: this wires message delivery AND the reconnect re-subscribe. - // The manager's Reconnected handler is passed through unwrapped on purpose — its - // throw-on-total-failure is what tears a deaf session down and retries it. - _subscriptions.AttachTo(connection); + // The handler is passed through unwrapped on purpose — its throw-on-total-failure is what + // tears a deaf session down and retries it. + if (_sparkplug is { } attaching) + { + attaching.AttachTo(connection); + } + else + { + _subscriptions.AttachTo(connection); + } if (BeforeConnectHookForTests is { } hook) { @@ -561,6 +614,29 @@ public sealed class MqttDriver throw new ObjectDisposedException(nameof(MqttDriver)); } + // Sparkplug subscribes ONCE, here, rather than on the OPC UA server's first subscribe: the + // driver must see birth certificates whether or not anything is subscribed yet, because a + // birth is where every alias and datatype comes from. This also issues the late-join + // rebirth (§3.6 #5) — the initial connect does not fire Reconnected. + // + // It runs BEFORE `_connection` is published, and disposes the session it was given if it + // fails. Publishing first would leave a live, supervised connection behind a throw: the + // driver-host resilience layer re-runs this method, the retry overwrites `_connection`, and + // the first session becomes an orphan no later dispose can reach — the exact leak + // MqttConnection's own remarks describe having already fixed once. + if (_sparkplug is { } sparkplug) + { + try + { + await sparkplug.EstablishAsync(cancellationToken).ConfigureAwait(false); + } + catch + { + await connection.DisposeAsync().ConfigureAwait(false); + throw; + } + } + _connection = connection; WriteHealth(new DriverHealth(DriverState.Healthy, connection.LastMessageUtc, null)); @@ -623,6 +699,7 @@ public sealed class MqttDriver if (!SameIngest(next, _options)) { _subscriptions = CreateSubscriptionManager(next); + _sparkplug = CreateSparkplugIngestor(next); } _options = next; @@ -645,10 +722,59 @@ public sealed class MqttDriver return manager; } + /// + /// Builds the Sparkplug B ingest path for , or + /// outside . Shares the driver-owned + /// for the same reason the plain manager does: a rebuild for + /// changed settings must not blank every observed value. + /// + private SparkplugIngestor? CreateSparkplugIngestor(MqttDriverOptions options) + { + if (options.Mode != MqttMode.SparkplugB) + { + return null; + } + + if (options.Sparkplug is { ActAsPrimaryHost: true }) + { + // Loud, because the flag reads as if it does something. Being a Sparkplug Primary Host is a + // three-part contract — a retained ONLINE after CONNECT, a matching retained OFFLINE + // registered as the MQTT Last Will AT CONNECT TIME, and an explicit OFFLINE on clean + // shutdown — and edge nodes stop publishing while their host is offline. Shipping the + // ONLINE half without the Will is strictly worse than shipping neither: a crashed node + // would leave a retained ONLINE asserting forever that a dead host is alive, which is the + // exact state the mechanism exists to prevent. The Will belongs in + // MqttConnection.BuildClientOptions; until it is there this flag stays inert and said so. + _logger?.LogWarning( + "MQTT driver '{DriverId}': Sparkplug actAsPrimaryHost is set but primary-host STATE " + + "publishing is not implemented — this driver observes STATE and never publishes one. " + + "Edge nodes gated on this host id will not see it come online.", + _driverInstanceId); + } + + var ingestor = new SparkplugIngestor( + options, + _driverInstanceId, + subscribeTransport: null, + publishTransport: null, + logger: _logger, + cache: _values, + maxPayloadBytes: options.MaxPayloadBytes); + + // 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); + return ingestor; + } + /// Registers the authored raw tags wholesale and refreshes the discovery enumeration set. private void RegisterAuthoredTags(MqttDriverOptions options) { - var mapped = _subscriptions.Register(options.RawTags); + // Wholesale into the LIVE path only. Feeding Sparkplug tags to the plain parser would log a + // "did not map" warning for every one of them (they carry no topic) and vice versa. + var mapped = _sparkplug is { } sparkplug + ? sparkplug.Register(options.RawTags) + : _subscriptions.Register(options.RawTags); // Enumerate in authoring order; the manager owns which of these actually mapped, and // DiscoverAsync skips the rest. @@ -676,16 +802,19 @@ public sealed class MqttDriver private static bool SameSession(MqttDriverOptions a, MqttDriverOptions b) => SessionIdentity(a).Equals(SessionIdentity(b)); - /// Whether two option sets produce an identical . + /// + /// Whether two option sets produce an identical ingest path — + /// in Plain mode, in Sparkplug B. + /// /// - /// ⚠️ P2 (Sparkplug, tasks 21/22) must extend . It names - /// only the settings the P1 manager captures at construction — Mode, - /// Plain.DefaultQos, MaxPayloadBytes — and deliberately no - /// field, because nothing reads one yet. The moment a - /// Sparkplug handler captures its own options (group id, host id, rebirth policy, birth - /// window) at construction, a delta changing one of them would be judged "same ingest", - /// applied in place, and silently never reach the component that needed rebuilding — the - /// driver would keep subscribing to the OLD group id and report perfect health. + /// ⚠️ Every setting an ingest path captures at construction MUST be named by + /// . A setting that is captured but unnamed makes a delta + /// changing it compare "same ingest", get applied in place, and silently never reach the + /// component that needed rebuilding — the driver would keep subscribing to the OLD Sparkplug + /// group id and report perfect health. Task 21 closed the Sparkplug half of that gap by naming + /// the whole record (it has value equality, so a member + /// added there is covered automatically — which is the point of naming the record rather than + /// enumerating its fields). /// private static bool SameIngest(MqttDriverOptions a, MqttDriverOptions b) => IngestIdentity(a).Equals(IngestIdentity(b)); @@ -714,12 +843,18 @@ public sealed class MqttDriver Ingest = IngestIdentity(o), }; - /// The settings captures at construction. + /// The settings the live ingest path captures at construction. + /// + /// is named as a whole record rather than field-by-field: it + /// has structural value equality, so a member added to it in a later task is covered here with + /// no edit — and the failure mode of forgetting one is silent (see ). + /// private static object IngestIdentity(MqttDriverOptions o) => new { o.Mode, DefaultQos = o.Plain?.DefaultQos, o.MaxPayloadBytes, + o.Sparkplug, }; /// Shared teardown for , and the session rebuild. diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/SparkplugIngestor.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/SparkplugIngestor.cs new file mode 100644 index 00000000..0133813d --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/SparkplugIngestor.cs @@ -0,0 +1,1775 @@ +using System.Collections.Concurrent; +using System.Collections.Frozen; +using System.Globalization; +using System.Text; +using System.Text.Json; +using Microsoft.Extensions.Logging; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Sparkplug; + +/// +/// The Sparkplug B ingest state machine — design doc §3.6, the driver's correctness core. It owns +/// the authored (group, node, device?, metric) → RawPath index, the live +/// , one per edge node, and the rebirth +/// policy; it turns every decoded NBIRTH/DBIRTH/NDATA/DDATA/NDEATH/DDEATH/STATE into +/// notifications plus updates, or into a +/// bounded, debounced rebirth NCMD. +/// +/// +/// +/// The published reference is the RawPath. Every this +/// type raises carries — the tag's RawPath — never a +/// composed Sparkplug reference such as Plant1/EdgeA/Filler1:Temperature and never the +/// topic. DriverHostActor's dual-namespace fan-out is keyed by RawPath, so publishing +/// under a Sparkplug-shaped key produces a driver that satisfies a naive unit test and delivers +/// nothing at all in production. The Sparkplug tuple is the lookup; the RawPath is the +/// identity. +/// +/// +/// Bind by metric NAME, resolve by alias (§3.6 #1). A DATA metric carries only an alias; +/// the alias is resolved through the scope's — rebuilt wholesale by +/// every birth — back to the stable metric name, and that name is what the authored index +/// is keyed by. Binding a tag to an alias silently mis-routes Pressure into a Temperature tag +/// the first time an edge node reuses an alias across a rebirth: good quality, plausible values, +/// completely wrong. +/// +/// +/// Every signed integer goes through . +/// Sparkplug carries Int8/16/32/64 as two's complement in an unsigned proto field, so a +/// metric whose value is -42 arrives as 4294967254. Skipping the call publishes a plausible, +/// Good-quality, completely wrong number. +/// +/// +/// An invalid payload never mutates state. An undecodable body advances no sequence +/// tracker, clears no alias table and evicts no birth — including for an NDEATH. Garbage bytes +/// are not evidence of anything, and a hostile or merely broken publisher on a group-wide +/// # subscription must not be able to blank the address space with them. +/// +/// +/// NDEATH is not part of the sequence. It is the broker-published Last Will and carries +/// no seq; it is tied to its birth by +/// instead. Feeding it to reports a gap on every death and +/// answers a node that has just died with a rebirth request it can never serve. +/// +/// +/// Rebirth requests are debounced per edge node. A single lost message on a busy node +/// produces one unknown-alias or data-before-birth event per metric per message until the birth +/// lands; without a floor between requests that is the NCMD storm the design warns about. See +/// . +/// +/// +/// Unauthored edge nodes are ignored outright. The driver subscribes +/// spBv1.0/{group}/#, so a large plant delivers traffic for every node in the group. Only +/// messages whose (group, edgeNode) appears in the authored index are processed at all — +/// which is what bounds the sequence-tracker and birth-cache maps by configuration rather +/// than by whatever the plant happens to publish. Device-level filtering happens later, at the +/// metric-resolution step, because an unauthored device's DDATA still advances its edge node's +/// sequence and dropping it early would manufacture a gap. +/// +/// +/// Nothing here throws on MQTTnet's dispatcher thread. runs on +/// it: every failure degrades a tag or drops a message, a throwing +/// subscriber is contained, and the rebirth publish is dispatched off-thread. +/// +/// +/// Primary-host STATE publishing is deliberately NOT implemented — see +/// and the remarks on +/// . This type observes STATE; it never publishes one. +/// +/// +public sealed class SparkplugIngestor +{ + /// + /// Minimum interval between two rebirth NCMDs aimed at the same edge node, when the caller does + /// not supply one. Sized against the observation that a rebirth is a full metadata republish an + /// edge node needs a moment to assemble: repeatedly commanding a node that is already + /// republishing is the storm, not the cure. + /// + public static readonly TimeSpan DefaultRebirthDebounce = TimeSpan.FromSeconds(10); + + // OPC UA status codes (values verified against Opc.Ua.StatusCodes, not recalled). + private const uint StatusGood = 0x00000000u; + + /// + /// The source has announced it is offline (NDEATH/DDEATH), or a birth voided the scope that was + /// feeding the tag. BadCommunicationError rather than the arguably more literal + /// BadNoCommunication (0x80310000) because it is what this repo's drivers actually + /// produce for a link that has stopped delivering — Modbus, S7, TwinCAT, FOCAS, + /// OpcUaClient and this driver's own plain-mode path all emit it, while no driver emits + /// BadNoCommunication at all (it appears once, in a CLI formatter's label map). Following the + /// precedent keeps one code meaning "not communicating" across the whole fleet. + /// + private const uint StatusBadCommunicationError = 0x80050000u; + + /// The metric's value does not fit the tag's effective data type — including an explicit Sparkplug null. + private const uint StatusBadTypeMismatch = 0x80740000u; + + /// The subscribed reference is not an authored Sparkplug tag. + private const uint StatusBadNodeIdUnknown = 0x80340000u; + + private readonly string _driverId; + private readonly ILogger? _logger; + private readonly TimeSpan _rebirthDebounce; + private readonly Func _utcNow; + + /// RawPath → the handle of the subscription currently covering it. + private readonly ConcurrentDictionary _handleByRawPath = new(StringComparer.Ordinal); + + /// Handle → the references it was created for, so unsubscribe can undo exactly its own. + private readonly ConcurrentDictionary _refsByHandle = new(); + + /// One stream-continuity tracker per authored edge node; bounded by the authored set. + private readonly ConcurrentDictionary _trackers = new(); + + /// Last rebirth-NCMD instant per edge node, in UTC ticks — the debounce floor. + private readonly ConcurrentDictionary _lastRebirthTicks = new(); + + /// Keys whose anomaly has already been logged loudly once, cleared by . + private readonly ConcurrentDictionary _warned = new(StringComparer.Ordinal); + + /// In-flight rebirth publishes, so teardown and tests can drain rather than race them. + private readonly List _pendingRebirths = []; + + private readonly EquipmentTagRefResolver _resolver; + + private long _handleSeq; + + /// The authored table + its Sparkplug indexes, swapped atomically by . + private volatile AuthoredTable _authored = AuthoredTable.Empty; + + private readonly MqttDriverOptions _options; + private volatile SparkplugHostState? _hostState; + private IMqttSubscribeTransport? _subscribeTransport; + private IMqttPublishTransport? _publishTransport; + + /// Initializes a new instance of the class. + /// + /// Driver configuration. supplies the group id, the + /// host id and the rebirth policy; a section is treated as the defaults. + /// + /// Driver instance id, used only for log correlation. + /// + /// The SUBSCRIBE seam. records the group filter without dialling a broker + /// (the shape tests use); supplies the live connection. + /// + /// + /// The NCMD publish seam. disables rebirth requests, which are then + /// logged and skipped rather than throwing on the dispatcher thread. + /// + /// Optional logger; never receives credentials or payload bodies. + /// The last-value store backing IReadable; defaults to a fresh one. + /// + /// Ceiling on an inbound message body. Non-positive falls back to + /// — "unbounded" is not offered, + /// because the bound protects a shared dispatcher thread. + /// + /// + /// Minimum interval between rebirth NCMDs aimed at one edge node. Defaults to + /// ; disables the floor (tests). + /// + /// Clock seam for the debounce; defaults to . + public SparkplugIngestor( + MqttDriverOptions options, + string driverId, + IMqttSubscribeTransport? subscribeTransport = null, + IMqttPublishTransport? publishTransport = null, + ILogger? logger = null, + LastValueCache? cache = null, + int maxPayloadBytes = MqttSubscriptionManager.DefaultMaxPayloadBytes, + TimeSpan? rebirthDebounce = null, + Func? utcNow = null) + { + ArgumentNullException.ThrowIfNull(options); + ArgumentException.ThrowIfNullOrWhiteSpace(driverId); + + _options = options; + _driverId = driverId; + _subscribeTransport = subscribeTransport; + _publishTransport = publishTransport; + _logger = logger; + Values = cache ?? new LastValueCache(); + MaxPayloadBytes = maxPayloadBytes > 0 ? maxPayloadBytes : MqttSubscriptionManager.DefaultMaxPayloadBytes; + _rebirthDebounce = rebirthDebounce is { } d && d >= TimeSpan.Zero ? d : DefaultRebirthDebounce; + _utcNow = utcNow ?? (() => DateTime.UtcNow); + _resolver = new EquipmentTagRefResolver( + rawPath => _authored.ByRawPath.TryGetValue(rawPath, out var def) ? def : null); + } + + /// + public event EventHandler? OnDataChange; + + /// + /// Raised after a birth certificate has been applied — the seam Task 22 wires to + /// IRediscoverable.OnRediscoveryNeeded so a DBIRTH introducing metrics the previous birth + /// did not carry rebuilds the address space. Deliberately not wired here: this type detects, the + /// driver decides. + /// + public event EventHandler? BirthObserved; + + /// Raised when a primary-host STATE message is observed. Informational; gates nothing. + public event EventHandler? HostStateObserved; + + /// The last observed value per RawPath — the push→poll bridge IReadable serves from. + public LastValueCache Values { get; } + + /// The live birth state: one per observed scope. Read by Tasks 22/23. + public BirthCache Births { get; } = new(); + + /// Ceiling on an inbound message body; a larger message is dropped before any decode. + public int MaxPayloadBytes { get; } + + /// + /// The last observed primary-host STATE, or when none has been seen. + /// + public SparkplugHostState? HostState => _hostState; + + /// The group-wide filter this ingestor subscribes, or when no group is configured. + public string? GroupFilter + { + get + { + var group = _options.Sparkplug?.GroupId; + return string.IsNullOrWhiteSpace(group) ? null : $"spBv1.0/{group}/#"; + } + } + + /// + /// The primary-host STATE filter this ingestor subscribes, or when no host + /// id is configured. Deliberately narrowed to the configured host id rather than + /// spBv1.0/STATE/#: this driver has no use for a stranger's host state, and a wildcard + /// would grow an unbounded observation map off traffic nobody asked for. + /// + public string? StateFilter + { + get + { + var hostId = _options.Sparkplug?.HostId; + return string.IsNullOrWhiteSpace(hostId) ? null : SparkplugTopic.FormatState(hostId); + } + } + + /// + /// Wires this ingestor onto a live connection: inbound messages route to + /// , the connection becomes both transports, and every reconnect runs + /// (re-subscribe + birth-state reset + late-join rebirth). + /// + /// The connection to observe. + public void AttachTo(MqttConnection connection) + { + ArgumentNullException.ThrowIfNull(connection); + + _subscribeTransport = connection; + _publishTransport = connection; + connection.MessageReceived += HandleMessage; + connection.Reconnected += OnReconnectedAsync; + } + + /// + /// Replaces the authored tag table with the deploy's raw tags, rebuilding the Sparkplug indexes. + /// Wholesale, not additive: a redeploy that drops a tag must stop feeding it. + /// + /// + /// A redeploy that ADDS an edge node does not rebirth it here. Until that deploy the node + /// was unauthored, so its traffic was filtered out and the driver holds no birth for it. It + /// recovers on the node's next DATA message, which lands as data-before-birth and asks for one — + /// self-correcting, and deliberately not a fresh NCMD fan-out on every redeploy, which on a large + /// plant would make a routine config change a group-wide rebirth storm. + /// + /// The authored raw tags delivered by the deploy artifact. + /// How many entries mapped to a usable Sparkplug definition. + public int Register(IEnumerable rawTags) + { + ArgumentNullException.ThrowIfNull(rawTags); + + var byRawPath = new Dictionary(StringComparer.Ordinal); + foreach (var entry in rawTags) + { + if (MqttTagDefinitionFactory.FromSparkplugTagConfig(entry.TagConfig, entry.RawPath, out var def)) + { + byRawPath[entry.RawPath] = def; + } + else + { + _logger?.LogWarning( + "MQTT driver '{DriverId}': tag config did not map to a Sparkplug definition " + + "(groupId/edgeNodeId/metricName are all required); skipping. RawPath={RawPath}", + _driverId, + entry.RawPath); + } + } + + var table = AuthoredTable.Build(byRawPath); + _authored = table; + _warned.Clear(); + + // Drop state belonging to edge nodes this deploy removed. Without this a tracker (and its + // rebirth debounce slot) for a node nobody reads survives every future redeploy. + foreach (var key in _trackers.Keys) + { + if (!table.EdgeNodes.Contains(key)) + { + _trackers.TryRemove(key, out _); + _lastRebirthTicks.TryRemove(key, out _); + } + } + + foreach (var rawPath in _handleByRawPath.Keys) + { + if (!table.ByRawPath.ContainsKey(rawPath)) + { + _handleByRawPath.TryRemove(rawPath, out _); + } + } + + return byRawPath.Count; + } + + /// Resolves a RawPath to its authored definition through the shared v3 resolver. + /// The driver wire reference. + /// The definition, when this returns . + /// when the reference is an authored Sparkplug tag. + public bool TryResolve(string rawPath, out MqttTagDefinition def) => _resolver.TryResolve(rawPath, out def!); + + /// The authored RawPaths, in no particular order — the discovery enumeration source. + public IReadOnlyCollection AuthoredRawPaths => _authored.ByRawPath.Keys; + + /// + /// Establishes the group-wide (and, when configured, STATE) subscription and requests a + /// late-join rebirth from every authored edge node. + /// + /// + /// + /// Unlike plain mode, the subscribe is NOT driven by . A + /// Sparkplug driver must see birth certificates whether or not the OPC UA server has + /// subscribed anything — the birth is where datatypes and aliases come from, and Task 22's + /// discovery depends on having seen one. So the driver establishes the filter once, at + /// connect, and only records which references may be published. + /// + /// + /// Total failure throws. There is exactly one filter and it covers every tag, so a + /// broker that refuses it leaves the whole driver deaf — the one outcome that must be a + /// visible fault (the driver-host resilience layer re-runs initialize under backoff) rather + /// than a healthy-looking driver receiving nothing. This is the same rule + /// applies, stated once. + /// + /// + /// Cancellation for the SUBSCRIBE round-trip. + /// A task that completes when the subscribe has been answered. + /// No group id is configured, or every filter was refused. + public async Task EstablishAsync(CancellationToken cancellationToken) + { + await SubscribeGroupAsync(cancellationToken).ConfigureAwait(false); + RequestLateJoinRebirths("initial connect"); + } + + /// + /// Records a batch of RawPath references against a new handle. Issues no SUBSCRIBE — the + /// group-wide filter is established once by ; see its remarks. + /// + /// The RawPath references to subscribe. + /// Ignored — Sparkplug is push-based. + /// Unused; this call touches no network. + /// The handle every resulting is attributed to. + public Task SubscribeAsync( + IReadOnlyList fullReferences, + TimeSpan publishingInterval, + CancellationToken cancellationToken) + { + ArgumentNullException.ThrowIfNull(fullReferences); + + var defs = new List(fullReferences.Count); + var now = _utcNow(); + foreach (var reference in fullReferences) + { + if (_resolver.TryResolve(reference, out var def)) + { + defs.Add(def); + continue; + } + + // Not an authored Sparkplug tag. Say so rather than leaving it in "waiting for initial + // data" forever — a reference that will never be fed is a different fault from one that + // has simply not been fed yet. + _logger?.LogWarning( + "MQTT driver '{DriverId}': subscribe reference '{RawPath}' is not an authored Sparkplug tag.", + _driverId, + reference); + Values.Update(reference, new DataValueSnapshot(null, StatusBadNodeIdUnknown, null, now)); + } + + var handle = new SparkplugSubscriptionHandle( + $"sparkplug:{Interlocked.Increment(ref _handleSeq)}:[{defs.Count} ref(s)]"); + _refsByHandle[handle] = [.. defs.Select(d => d.Name)]; + foreach (var def in defs) + { + _handleByRawPath[def.Name] = handle; + } + + return Task.FromResult(handle); + } + + /// + /// Stops attributing notifications to . References another live + /// subscription also covers keep publishing; the broker-side group filter is untouched. + /// + /// The handle returned by . + /// Unused; present for the ISubscribable shape. + /// A completed task. + public Task UnsubscribeAsync(ISubscriptionHandle handle, CancellationToken cancellationToken) + { + if (handle is not null && _refsByHandle.TryRemove(handle, out var refs)) + { + foreach (var rawPath in refs) + { + // Only clear the mapping if it is still OURS — a later overlapping subscription may + // have taken the reference over, and it must keep publishing. + _handleByRawPath.TryRemove(new KeyValuePair(rawPath, handle)); + } + } + + return Task.CompletedTask; + } + + /// + /// Recovers the ingest state after a reconnect: drops every birth, resets every sequence + /// tracker, re-establishes the group filter and requests a late-join rebirth from every authored + /// edge node (§3.6 invariant #5). + /// + /// + /// + /// Clearing the birth cache is not housekeeping — it is the correctness step. Across + /// an outage of unknown length an edge node may have restarted and rebound its aliases; a + /// DDATA resolved against the pre-outage table would route a value to whichever tag used to + /// own that alias. Same for the sequence baseline: carried across, a stream could look + /// contiguous purely because the node published the right number of messages while the + /// driver was away. + /// + /// + /// Values are deliberately not staled here. Reconnect is a driver-side event, not a + /// statement about the plant; the last observed values remain the best available answer + /// until the rebirth lands, and the connection's own health surface already reports + /// Reconnecting. Only a death — the source saying it is offline — stales a tag. + /// + /// + /// The connection's lifetime token; cancelled by its dispose. + /// A task that completes when the re-subscribe has been answered. + /// Every filter was refused — the caller tears the session down. + public async Task OnReconnectedAsync(CancellationToken cancellationToken) + { + Births.Clear(); + _trackers.Clear(); + + await SubscribeGroupAsync(cancellationToken).ConfigureAwait(false); + RequestLateJoinRebirths("reconnect"); + } + + /// + /// Routes one inbound MQTT message: parses the topic, decodes the Sparkplug payload and hands it + /// to . Runs on MQTTnet's dispatcher thread and never throws. + /// + /// The concrete topic the message arrived on. + /// The message body; valid only for the duration of this call. + /// + /// The message's retain flag. Not filtered on: a retained NBIRTH/STATE is exactly the + /// late-join metadata this driver wants, and Sparkplug DATA is never published retained. + /// + public void HandleMessage(string topic, ReadOnlySpan payload, bool retained) + { + try + { + if (!SparkplugTopic.TryParse(topic, out var parsed)) + { + // A group-wide '#' subscription legitimately delivers topics this driver has no opinion + // about. Debug, not warn — an operator must not have to read one line per stray message. + _logger?.LogDebug( + "MQTT driver '{DriverId}': '{Topic}' is not a Sparkplug topic; ignored.", + _driverId, + topic); + return; + } + + if (payload.Length > MaxPayloadBytes) + { + // Refused BEFORE any decode: the whole point of the bound is that one publisher cannot + // impose an unbounded protobuf parse on the shared dispatcher thread. Nothing mutates. + WarnOnce( + $"oversize:{topic}", + "MQTT driver '{DriverId}': '{Topic}' carried {Bytes} bytes, over the {Max}-byte ceiling; dropped.", + _driverId, + topic, + payload.Length, + MaxPayloadBytes); + return; + } + + if (parsed.Type == SparkplugMessageType.STATE) + { + HandleState(parsed, payload); + return; + } + + Dispatch(parsed, SparkplugCodec.Decode(payload)); + } + catch (Exception ex) + { + // Defence in depth: every path below is written not to throw, but an escape here surfaces + // inside MQTTnet's own pump and stalls delivery for every subscription on the connection. + _logger?.LogError( + ex, + "MQTT driver '{DriverId}': Sparkplug ingest threw for '{Topic}'; the message was dropped.", + _driverId, + topic); + } + } + + /// + /// Routes one decoded Sparkplug message — the state machine proper, and the seam the + /// §3.6 matrix is driven through without a broker. + /// + /// The parsed topic. + /// The decoded payload. Checked for validity here, not by the caller. + public void Dispatch(SparkplugTopic topic, SparkplugPayload payload) + { + ArgumentNullException.ThrowIfNull(topic); + ArgumentNullException.ThrowIfNull(payload); + + if (topic.Type == SparkplugMessageType.STATE) + { + return; // STATE carries no Sparkplug payload; HandleMessage routes it separately. + } + + // NCMD/DCMD flow host → node. Our own rebirth requests come back on the '#' subscription; + // treating them as ingest would be, at best, noise. + if (topic.Type is SparkplugMessageType.NCMD or SparkplugMessageType.DCMD) + { + return; + } + + if (topic.GroupId is not { } groupId || topic.EdgeNodeId is not { } edgeNodeId) + { + return; // TryParse guarantees both for a non-STATE topic; belt and braces. + } + + var node = new SparkplugNodeKey(groupId, edgeNodeId); + if (!_authored.EdgeNodes.Contains(node)) + { + _logger?.LogDebug( + "MQTT driver '{DriverId}': {Type} for unauthored edge node '{Node}'; ignored.", + _driverId, + topic.Type, + node); + return; + } + + if (!payload.IsValid) + { + // NOT a rebirth trigger and NOT a state mutation — see the type remarks. An undecodable + // body is not evidence that anything about the node's stream has changed. + WarnOnce( + $"invalid:{node}", + "MQTT driver '{DriverId}': {Type} from '{Node}' could not be decoded as a Sparkplug payload; dropped.", + _driverId, + topic.Type, + node); + return; + } + + switch (topic.Type) + { + case SparkplugMessageType.NBIRTH: + OnNodeBirth(node, payload); + break; + + case SparkplugMessageType.DBIRTH when topic.DeviceId is { } dbirthDevice: + OnDeviceBirth(node, dbirthDevice, payload); + break; + + case SparkplugMessageType.NDATA: + OnData(node, new SparkplugScope(groupId, edgeNodeId, null), payload); + break; + + case SparkplugMessageType.DDATA when topic.DeviceId is { } ddataDevice: + OnData(node, new SparkplugScope(groupId, edgeNodeId, ddataDevice), payload); + break; + + case SparkplugMessageType.NDEATH: + OnNodeDeath(node, payload); + break; + + case SparkplugMessageType.DDEATH when topic.DeviceId is { } ddeathDevice: + OnDeviceDeath(node, ddeathDevice, payload); + break; + + default: + break; + } + } + + /// + /// Awaits every rebirth publish this ingestor has dispatched and not yet observed. Test seam and + /// teardown helper — the publishes themselves are fire-and-forget by design, because a rebirth + /// NCMD must never be able to park MQTTnet's dispatcher thread. + /// + /// A task that completes when no dispatched rebirth publish is outstanding. + internal async Task DrainPendingRebirthsAsync() + { + Task[] pending; + lock (_pendingRebirths) + { + pending = [.. _pendingRebirths]; + } + + foreach (var task in pending) + { + try + { + await task.ConfigureAwait(false); + } + catch (Exception) + { + // Already logged where it was dispatched; a failed rebirth is not a caller's problem. + } + } + } + + // ---- births ---- + + /// + /// Applies an NBIRTH: restarts the node's sequence at the birth's seq, adopts its + /// bdSeq session token, rebuilds the node's catalog, stales every device the birth + /// invalidated, and publishes the birth's own values. + /// + private void OnNodeBirth(SparkplugNodeKey node, SparkplugPayload payload) + { + // AcceptNodeBirth, never Accept: an NBIRTH restarts the sequence by definition, so running it + // through the gap detector would flag a gap on the very message that resynchronized everything + // and answer a birth with a demand for another one. + var bdSeq = SequenceTracker.TryReadBdSeq(payload, out var token) ? token : (ulong?)null; + if (!TrackerFor(node).AcceptNodeBirth(payload.Seq, bdSeq)) + { + // Deliberately NOT a rebirth trigger: the metadata just arrived, and commanding another + // birth from a node whose seq we cannot read would loop against the same malformed payload. + WarnOnce( + $"birthseq:{node}", + "MQTT driver '{DriverId}': NBIRTH from '{Node}' carried no usable seq ({Seq}); " + + "gap detection is disabled for this node until its next birth.", + _driverId, + node, + payload.Seq); + } + + var outcome = Births.ApplyNodeBirth(node.GroupId, node.EdgeNodeId, payload.Metrics); + LogBirthAnomaly(node.ToString(), outcome.Result); + + var sourceUtc = ToUtc(payload.TimestampMs); + + // An NBIRTH voids every prior DBIRTH under the node (Sparkplug spec): those devices MUST + // re-DBIRTH, and until they do their aliases are meaningless. Staling them is what stops the + // driver serving values resolved against a table it has just thrown away. + foreach (var invalidated in outcome.InvalidatedDevices) + { + FanStale(invalidated.Scope, sourceUtc, "its edge node re-birthed; awaiting a fresh DBIRTH"); + } + + PublishBirthValues(new SparkplugScope(node.GroupId, node.EdgeNodeId, null), payload, sourceUtc); + RaiseBirthObserved(new SparkplugScope(node.GroupId, node.EdgeNodeId, null), payload); + } + + /// Applies a DBIRTH: rebuilds exactly that device's catalog and publishes its values. + private void OnDeviceBirth(SparkplugNodeKey node, string deviceId, SparkplugPayload payload) + { + // A DBIRTH is a sequenced member of the edge node's stream — only the NBIRTH restarts it. + CheckSequence(node, payload, "DBIRTH"); + + var result = Births.ApplyDeviceBirth(node.GroupId, node.EdgeNodeId, deviceId, payload.Metrics); + var scope = new SparkplugScope(node.GroupId, node.EdgeNodeId, deviceId); + LogBirthAnomaly(scope.ToString(), result); + + var sourceUtc = ToUtc(payload.TimestampMs); + PublishBirthValues(scope, payload, sourceUtc); + RaiseBirthObserved(scope, payload); + } + + // ---- data ---- + + /// + /// Applies an NDATA/DDATA: checks stream continuity, resolves every metric through the scope's + /// birth-built alias table, and publishes under the authored tag's RawPath. + /// + private void OnData(SparkplugNodeKey node, SparkplugScope scope, SparkplugPayload payload) + { + CheckSequence(node, payload, scope.IsDevice ? "DDATA" : "NDATA"); + + // Scope-level filtering happens HERE, after the sequence check, never before it: an unauthored + // device's DDATA is still a member of its edge node's stream, and skipping it earlier would + // manufacture a gap on the very next authored message. Past this point there is nothing this + // scope could publish — so there is also nothing worth demanding a rebirth for. Without the + // guard, a device nobody authored that publishes DDATA without ever DBIRTHing (spec-violating, + // but they exist) would drive a permanent NCMD at the debounce cadence, forever. + if (!_authored.ByScope.ContainsKey(scope)) + { + return; + } + + var table = Births.Find(scope); + if (table is null) + { + // Data before any birth (§3.6 #3): we hold no alias table for this scope, so every metric + // in this payload is unroutable. One rebirth request for the whole message, not one per + // metric — the debounce would collapse them anyway, but not asking is cheaper than asking + // and being suppressed. + WarnOnce( + $"prebirth:{scope}", + "MQTT driver '{DriverId}': data for '{Scope}' arrived before any birth certificate; " + + "requesting a rebirth.", + _driverId, + scope); + RequestRebirth(node, "data before birth"); + return; + } + + var sourceUtc = ToUtc(payload.TimestampMs); + var unknownAlias = false; + + foreach (var metric in payload.Metrics) + { + var binding = ResolveBinding(table, metric); + if (binding is null) + { + // The birth this table was built from never declared this alias/name — our metadata is + // behind the node's. Same remedy as data-before-birth, different log. + unknownAlias = true; + continue; + } + + PublishMetric(scope, binding, metric, sourceUtc); + } + + if (unknownAlias) + { + WarnOnce( + $"alias:{scope}", + "MQTT driver '{DriverId}': data for '{Scope}' carried a metric the current birth does not " + + "declare; requesting a rebirth.", + _driverId, + scope); + RequestRebirth(node, "unknown alias"); + } + } + + /// + /// Resolves a DATA metric to its birth binding: by alias when it carries one (the normal + /// case — a DATA metric after a birth carries an alias and neither name nor datatype), otherwise + /// by name, since aliases are optional in Sparkplug. + /// + private static SparkplugMetricBinding? ResolveBinding(AliasTable table, SparkplugMetric metric) + { + if (metric.Alias is { } alias && table.Resolve(alias) is { } byAlias) + { + return byAlias; + } + + return metric.Name is { } name ? table.ResolveByName(name) : null; + } + + // ---- deaths ---- + + /// + /// Applies an NDEATH: verifies it belongs to the session currently believed alive, then forgets + /// the node and every device under it and stales their authored tags. + /// + private void OnNodeDeath(SparkplugNodeKey node, SparkplugPayload payload) + { + // NOT SequenceTracker.Accept. An NDEATH is the broker-published Last Will: it carries no seq at + // all, so Accept would report a gap on every death and answer a node that has just died with a + // rebirth request. The bdSeq tie is the whole mechanism instead. + var tracker = TrackerFor(node); + var deathBdSeq = SequenceTracker.TryReadBdSeq(payload, out var token) ? token : (ulong?)null; + if (!tracker.IsDeathForCurrentSession(deathBdSeq)) + { + // A previous session's will, delivered after the node already reconnected and re-birthed. + // Acting on it would mark a live node dead with nothing coming to correct it. + _logger?.LogInformation( + "MQTT driver '{DriverId}': ignoring a stale NDEATH for '{Node}' (bdSeq {DeathBdSeq} ≠ current " + + "{BirthBdSeq}); the node re-birthed before its old will was delivered.", + _driverId, + node, + deathBdSeq, + tracker.BirthBdSeq); + return; + } + + Births.ForgetNode(node.GroupId, node.EdgeNodeId); + + var sourceUtc = ToUtc(payload.TimestampMs); + foreach (var scope in _authored.ScopesUnder(node)) + { + FanStale(scope, sourceUtc, "its edge node published NDEATH"); + } + } + + /// Applies a DDEATH: forgets exactly that device and stales its authored tags. + private void OnDeviceDeath(SparkplugNodeKey node, string deviceId, SparkplugPayload payload) + { + // A DDEATH IS published by the edge node and IS sequenced — unlike an NDEATH. + CheckSequence(node, payload, "DDEATH"); + + Births.TryForgetDevice(node.GroupId, node.EdgeNodeId, deviceId, out _); + + FanStale( + new SparkplugScope(node.GroupId, node.EdgeNodeId, deviceId), + ToUtc(payload.TimestampMs), + "the device published DDEATH"); + } + + // ---- STATE ---- + + /// + /// Records an observed primary-host STATE message. + /// + /// + /// + /// Receive only. This driver never publishes STATE, and + /// is therefore inert — deliberately, and + /// loudly. Being a Sparkplug Primary Host is not "publish an ONLINE message": it is a + /// three-part contract — a retained ONLINE STATE published after CONNECT, a matching + /// retained OFFLINE registered as the client's MQTT Last Will at CONNECT time, + /// and an explicit OFFLINE on clean shutdown. Edge nodes subscribe to it and stop publishing + /// while their host is offline. Shipping the ONLINE half without the Will is strictly worse + /// than shipping neither: an OtOpcUa node that crashes would leave a retained ONLINE asserting + /// forever that a dead host is alive, which is precisely the state the mechanism exists to + /// prevent. The Will has to be set inside MqttConnection.BuildClientOptions — a + /// different file, a different task — so this task implements the observation half and + /// warns when the flag is set. See the design doc §7/§8. + /// + /// + /// Both payload forms are accepted on receive: the v3.0 JSON {"online":bool,...} and + /// the pre-3.0 bare ONLINE/OFFLINE text. + /// + /// + private void HandleState(SparkplugTopic topic, ReadOnlySpan payload) + { + if (topic.HostId is not { } hostId) + { + return; + } + + if (!TryReadStateOnline(payload, out var online)) + { + WarnOnce( + $"state:{hostId}", + "MQTT driver '{DriverId}': STATE for host '{HostId}' carried an unrecognised payload; ignored.", + _driverId, + hostId); + return; + } + + var previous = _hostState; + var state = new SparkplugHostState(hostId, online, _utcNow()); + _hostState = state; + + if (previous is null || previous.Online != online || !string.Equals(previous.HostId, hostId, StringComparison.Ordinal)) + { + _logger?.LogInformation( + "MQTT driver '{DriverId}': Sparkplug host '{HostId}' is {State}.", + _driverId, + hostId, + online ? "ONLINE" : "OFFLINE"); + } + + HostStateObserved?.Invoke(this, new SparkplugHostStateEventArgs(state)); + } + + /// Reads the online flag out of a STATE body — v3.0 JSON first, then the legacy text form. + private static bool TryReadStateOnline(ReadOnlySpan payload, out bool online) + { + online = false; + if (payload.IsEmpty) + { + return false; + } + + try + { + var reader = new Utf8JsonReader(payload); + if (JsonDocument.TryParseValue(ref reader, out var doc)) + { + using (doc) + { + if (doc.RootElement.ValueKind == JsonValueKind.Object + && doc.RootElement.TryGetProperty("online", out var flag) + && flag.ValueKind is JsonValueKind.True or JsonValueKind.False) + { + online = flag.GetBoolean(); + return true; + } + } + } + } + catch (JsonException) + { + // Fall through to the legacy text form. + } + + Span text = stackalloc char[16]; + if (payload.Length > text.Length || !Encoding.UTF8.TryGetChars(payload, text, out var written)) + { + return false; + } + + var trimmed = text[..written].Trim(); + if (trimmed.Equals("ONLINE", StringComparison.OrdinalIgnoreCase)) + { + online = true; + return true; + } + + if (trimmed.Equals("OFFLINE", StringComparison.OrdinalIgnoreCase)) + { + online = false; + return true; + } + + return false; + } + + // ---- publish ---- + + /// + /// Publishes every named metric a birth certificate carried. This is what restores Good quality + /// after a death: a birth is a full snapshot, not a delta. + /// + private void PublishBirthValues(SparkplugScope scope, SparkplugPayload payload, DateTime? sourceUtc) + { + var table = Births.Find(scope); + if (table is null) + { + return; + } + + foreach (var metric in payload.Metrics) + { + if (metric.Name is not { } name || table.ResolveByName(name) is not { } binding) + { + continue; + } + + PublishMetric(scope, binding, metric, sourceUtc); + } + } + + /// + /// Publishes one resolved metric to every authored tag bound to + /// (scope, binding.Name) — under the tag's RawPath. + /// + private void PublishMetric( + SparkplugScope scope, + SparkplugMetricBinding binding, + SparkplugMetric metric, + DateTime? sourceUtc) + { + // Bound by NAME. The alias got us here; it is never the key. + var key = new SparkplugMetricKey(scope.GroupId, scope.EdgeNodeId, scope.DeviceId, binding.Name); + if (!_authored.ByMetric.TryGetValue(key, out var defs)) + { + return; // A published metric nobody authored a tag for. Not an error — that is most of them. + } + + var birthType = binding.DataType.ToDriverDataType(); + if (birthType is null) + { + // DataSet / Template / PropertySet / Unknown — warn-and-skip per the datatype map's + // contract, and explicitly NOT a rebirth trigger: the node would answer with the same + // unsupported metric and the pair would loop. + WarnOnce( + $"unsupported-type:{key}", + "MQTT driver '{DriverId}': metric '{Metric}' on '{Scope}' has unsupported Sparkplug type " + + "{DataType}; its tag(s) are not being fed.", + _driverId, + binding.Name, + scope, + binding.DataType); + return; + } + + var serverUtc = _utcNow(); + + foreach (var def in defs) + { + // The authored dataType wins when the operator declared one; otherwise the birth's own + // declaration does. That is what lets a tag be authored with nothing but the metric tuple. + var target = def.DataTypeAuthored ? def.DataType : birthType.Value; + if (BuildSnapshot(def, binding, metric, target, sourceUtc, serverUtc) is not { } snapshot) + { + continue; + } + + Values.Update(def.Name, snapshot); + Raise(def.Name, snapshot); + } + } + + /// + /// Turns one metric's wire value into a snapshot for one authored tag, or + /// when the metric carried no value at all — in which case the tag keeps whatever it last had + /// rather than being re-published unchanged or given an invented quality. + /// + private DataValueSnapshot? BuildSnapshot( + MqttTagDefinition def, + SparkplugMetricBinding binding, + SparkplugMetric metric, + DriverDataType target, + DateTime? sourceUtc, + DateTime serverUtc) + { + // Metric timestamp wins over the payload's — it is the acquisition instant at the source. + var stamp = ToUtc(metric.TimestampMs) ?? sourceUtc; + + switch (metric.ValueKind) + { + case SparkplugValueKind.Scalar: + break; + + case SparkplugValueKind.Null: + // An explicit is_null. The tag has a declared scalar type and null is not a value of + // it — the same verdict the plain-mode path reaches for a JSON null. + return new DataValueSnapshot(null, StatusBadTypeMismatch, stamp, serverUtc); + + default: + // Absent (no value oneof at all) or Unsupported (DataSet/Template). Neither is a value; + // leave the tag holding whatever it last had rather than inventing a quality for it. + WarnOnce( + $"valuekind:{def.Name}", + "MQTT driver '{DriverId}': metric '{Metric}' for '{RawPath}' carried {ValueKind} rather " + + "than a value; skipped.", + _driverId, + binding.Name, + def.Name, + metric.ValueKind); + return null; + } + + // THE call. Sparkplug carries signed integers as two's complement in an unsigned field, and a + // DATA metric carries no datatype of its own — only the birth binding knows which applies. + // Skipping this publishes 4294967254 for a value of -42: plausible, Good, and wrong. + var raw = binding.Reinterpret(metric.Value); + + return SparkplugValueCoercion.TryCoerce(raw, target, out var value) + ? new DataValueSnapshot(value, StatusGood, stamp, serverUtc) + : BadTypeMismatch(def, binding, target, stamp, serverUtc); + } + + /// + /// The metric decoded, but does not fit the tag's effective type. Warned once per tag; the value + /// itself is deliberately never logged — a payload value is plant data, not diagnostics. + /// + private DataValueSnapshot BadTypeMismatch( + MqttTagDefinition def, + SparkplugMetricBinding binding, + DriverDataType target, + DateTime? stamp, + DateTime serverUtc) + { + WarnOnce( + $"coerce:{def.Name}", + "MQTT driver '{DriverId}': metric '{Metric}' ({Sparkplug}) does not fit tag '{RawPath}'s " + + "{Target}; status BadTypeMismatch.", + _driverId, + binding.Name, + binding.DataType, + def.Name, + target); + + return new DataValueSnapshot(null, StatusBadTypeMismatch, stamp, serverUtc); + } + + /// Marks every authored tag in as no longer communicating. + private void FanStale(SparkplugScope scope, DateTime? sourceUtc, string reason) + { + if (!_authored.ByScope.TryGetValue(scope, out var defs)) + { + return; + } + + var snapshot = new DataValueSnapshot(null, StatusBadCommunicationError, sourceUtc, _utcNow()); + foreach (var def in defs) + { + Values.Update(def.Name, snapshot); + Raise(def.Name, snapshot); + } + + _logger?.LogInformation( + "MQTT driver '{DriverId}': staled {Count} tag(s) on '{Scope}' — {Reason}.", + _driverId, + defs.Length, + scope, + reason); + } + + /// + /// Raises for a RawPath, if a live subscription covers it. A throwing + /// subscriber is contained so it cannot starve the rest of the fan-out or escape onto MQTTnet's + /// dispatcher thread. + /// + private void Raise(string rawPath, DataValueSnapshot snapshot) + { + if (!_handleByRawPath.TryGetValue(rawPath, out var handle)) + { + return; // Authored but not subscribed: cached so a read answers, nothing to attribute to. + } + + try + { + OnDataChange?.Invoke(this, new DataChangeEventArgs(handle, rawPath, snapshot)); + } + catch (Exception ex) + { + _logger?.LogError( + ex, + "MQTT driver '{DriverId}': a data-change subscriber threw for '{RawPath}'.", + _driverId, + rawPath); + } + } + + private void RaiseBirthObserved(SparkplugScope scope, SparkplugPayload payload) + { + var subscribers = BirthObserved; + if (subscribers is null) + { + return; + } + + try + { + subscribers(this, new SparkplugBirthObservedEventArgs( + scope, + [.. payload.Metrics.Where(m => m.Name is not null).Select(m => m.Name!)])); + } + catch (Exception ex) + { + _logger?.LogError(ex, "MQTT driver '{DriverId}': a birth-observed subscriber threw.", _driverId); + } + } + + // ---- sequence + rebirth ---- + + private SequenceTracker TrackerFor(SparkplugNodeKey node) => + _trackers.GetOrAdd(node, static _ => new SequenceTracker()); + + /// + /// Offers a sequenced message's seq and requests a rebirth when the stream is not + /// contiguous — or when it is contiguous only because the driver joined mid-stream and has never + /// seen this node's NBIRTH. + /// + /// + /// The second case is the one that is easy to miss: returns + /// for the first message after a reset (there is nothing to measure it + /// against), so a driver that only checked its return value would sit happily behind a stream + /// whose birth certificate — and therefore whose entire alias table — it never received. + /// + private void CheckSequence(SparkplugNodeKey node, SparkplugPayload payload, string kind) + { + var tracker = TrackerFor(node); + if (!tracker.Accept(payload.Seq)) + { + WarnOnce( + $"gap:{node}", + "MQTT driver '{DriverId}': {Kind} from '{Node}' broke the sequence (seq {Seq}, last {Last}); " + + "requesting a rebirth.", + _driverId, + kind, + node, + payload.Seq, + tracker.LastSeq); + RequestRebirth(node, "sequence gap"); + return; + } + + if (!tracker.IsBirthSynchronized) + { + WarnOnce( + $"latejoin:{node}", + "MQTT driver '{DriverId}': {Kind} from '{Node}' arrived with no NBIRTH seen since connect; " + + "requesting a rebirth.", + _driverId, + kind, + node); + RequestRebirth(node, "late join (no NBIRTH observed)"); + } + } + + /// + /// Requests a rebirth from every authored edge node — §3.6 invariant #5. Called on connect and + /// on every reconnect. + /// + /// + /// Bounded by configuration, not by the plant. Sparkplug has no group-wide rebirth + /// command — NCMD addresses exactly one edge node — so this is inherently one publish per node. + /// The set is the authored edge nodes rather than every node in the group, which is what + /// keeps a 200-node group from producing 200 NCMDs for the two nodes this deployment reads. Each + /// one still passes through the per-node debounce, so a connection that flaps cannot multiply + /// them. + /// + private void RequestLateJoinRebirths(string reason) + { + foreach (var node in _authored.EdgeNodes) + { + RequestRebirth(node, reason); + } + } + + /// + /// Publishes a rebirth NCMD to one edge node, subject to the + /// policy and the per-node debounce. + /// Never blocks the caller and never throws. + /// + private void RequestRebirth(SparkplugNodeKey node, string reason) + { + if (_options.Sparkplug is { RequestRebirthOnGap: false }) + { + _logger?.LogDebug( + "MQTT driver '{DriverId}': would request a rebirth from '{Node}' ({Reason}) but " + + "requestRebirthOnGap is off.", + _driverId, + node, + reason); + return; + } + + if (_publishTransport is not { } transport) + { + _logger?.LogDebug( + "MQTT driver '{DriverId}': cannot request a rebirth from '{Node}' ({Reason}) — no publish " + + "transport is attached.", + _driverId, + node, + reason); + return; + } + + if (!TryAcquireRebirthSlot(node)) + { + _logger?.LogDebug( + "MQTT driver '{DriverId}': rebirth request to '{Node}' ({Reason}) suppressed; one was sent " + + "less than {Debounce} ago.", + _driverId, + node, + reason, + _rebirthDebounce); + return; + } + + _logger?.LogInformation( + "MQTT driver '{DriverId}': requesting a Sparkplug rebirth from '{Node}' — {Reason}.", + _driverId, + node, + reason); + + // Off the dispatcher thread: RebirthRequester.RequestAsync serializes a payload and hands it to + // MQTTnet's publish path, and neither belongs on the thread that is delivering every other + // subscription's messages. It is bounded by its own 5s deadline. + TrackRebirth(Task.Run(async () => + { + try + { + await RebirthRequester + .RequestAsync(transport, node.GroupId, node.EdgeNodeId, CancellationToken.None) + .ConfigureAwait(false); + } + catch (Exception ex) + { + _logger?.LogWarning( + ex, + "MQTT driver '{DriverId}': the rebirth NCMD to '{Node}' failed; the next gap will retry.", + _driverId, + node); + } + })); + } + + /// + /// Claims this node's rebirth slot, or reports that one was claimed too recently. Lock-free: this + /// runs on MQTTnet's dispatcher thread. + /// + private bool TryAcquireRebirthSlot(SparkplugNodeKey node) + { + if (_rebirthDebounce <= TimeSpan.Zero) + { + return true; + } + + var now = _utcNow().Ticks; + while (true) + { + if (!_lastRebirthTicks.TryGetValue(node, out var last)) + { + if (_lastRebirthTicks.TryAdd(node, now)) + { + return true; + } + + continue; + } + + if (now - last < _rebirthDebounce.Ticks) + { + return false; + } + + if (_lastRebirthTicks.TryUpdate(node, now, last)) + { + return true; + } + } + } + + private void TrackRebirth(Task task) + { + lock (_pendingRebirths) + { + _pendingRebirths.RemoveAll(static t => t.IsCompleted); + _pendingRebirths.Add(task); + } + } + + // ---- subscribe ---- + + /// Issues the group-wide (and STATE) SUBSCRIBE, throwing when nothing could be established. + private async Task SubscribeGroupAsync(CancellationToken cancellationToken) + { + if (GroupFilter is not { } groupFilter) + { + throw new InvalidOperationException( + $"MQTT driver '{_driverId}': Sparkplug mode requires a groupId — there is no filter to " + + "subscribe, so the driver would receive nothing."); + } + + // QoS 1 for both. Sparkplug publishes NBIRTH/DBIRTH/DATA at QoS 0 and NDEATH/STATE at QoS 1; + // the SUBSCRIBE's QoS is the ceiling the broker may deliver at, so asking for 1 keeps the + // death and state messages at their intended guarantee and costs nothing for the rest. + var filters = new List(2) + { + // SeedRetained: a retained NBIRTH/STATE replayed at subscribe time IS the late-join + // metadata this driver wants — the opposite of the plain-mode tag opt-out. + new(groupFilter, Qos: 1, SeedRetained: true), + }; + + if (StateFilter is { } stateFilter) + { + filters.Add(new MqttTopicSubscription(stateFilter, Qos: 1, SeedRetained: true)); + } + + var transport = _subscribeTransport; + if (transport is null) + { + _logger?.LogWarning( + "MQTT driver '{DriverId}': Sparkplug subscription recorded with no connection attached; " + + "nothing was sent to a broker. Call AttachTo before establishing.", + _driverId); + return; + } + + var outcomes = await transport.SubscribeAsync(filters, cancellationToken).ConfigureAwait(false); + + // The group filter is the one that matters: without it the driver is deaf. A refused STATE + // filter costs an observability signal, not the data plane, so it warns rather than throws. + var groupGranted = outcomes.Any(o => o.Granted && string.Equals(o.Topic, groupFilter, StringComparison.Ordinal)); + if (!groupGranted) + { + throw new InvalidOperationException( + $"MQTT driver '{_driverId}': the broker refused '{groupFilter}'; tearing the session down " + + "rather than serving a Sparkplug driver that receives nothing."); + } + + foreach (var refused in outcomes.Where(o => !o.Granted)) + { + _logger?.LogWarning( + "MQTT driver '{DriverId}': broker refused SUBSCRIBE '{Topic}' ({Reason}).", + _driverId, + refused.Topic, + refused.Reason); + } + } + + // ---- helpers ---- + + private void LogBirthAnomaly(string scope, BirthApplyResult result) + { + if (!result.HasAnomaly) + { + return; + } + + // The only window onto a malformed birth — AliasTable takes no logger by design. + _logger?.LogWarning( + "MQTT driver '{DriverId}': birth for '{Scope}' was malformed — {Accepted} accepted, " + + "{Unnamed} unnamed, {AliasCollisions} alias collision(s), {DuplicateNames} duplicate name(s).", + _driverId, + scope, + result.Accepted, + result.RejectedUnnamed, + result.AliasCollisions, + result.DuplicateNames); + } + + private static DateTime? ToUtc(ulong? epochMs) => + epochMs is { } ms && ms <= (ulong)DateTimeOffset.MaxValue.ToUnixTimeMilliseconds() + ? DateTimeOffset.FromUnixTimeMilliseconds((long)ms).UtcDateTime + : null; + + /// + /// Logs a warning the first time is seen and stays quiet thereafter. A + /// 10 Hz metric with a mis-authored datatype must be diagnosable without drowning the log; the + /// key set is bounded because every path that calls this has already filtered to authored nodes, + /// scopes or RawPaths, and it is cleared on every . + /// + private void WarnOnce(string key, string message, params object?[] args) + { + if (_warned.TryAdd(key, 0)) + { +#pragma warning disable CA2254 // Template is a compile-time constant at every call site. + _logger?.LogWarning(message, args); +#pragma warning restore CA2254 + } + } + + /// Opaque subscription identity for the Sparkplug ingest path. + private sealed record SparkplugSubscriptionHandle(string DiagnosticId) : ISubscriptionHandle; + + /// + /// The authored tag table and its Sparkplug indexes, published as one immutable snapshot so the + /// dispatcher thread reads a consistent view while a redeploy rebuilds it. + /// + /// RawPath → definition; the v3 resolver's backing table. + /// + /// (group, node, device?, metricName) → every definition bound to it. A list, + /// because two raw tags may legitimately project the same Sparkplug metric. + /// + /// Scope → its definitions; the death/invalidation STALE fan-out's input. + /// + /// The distinct (group, edgeNode) pairs any authored tag names — the ingest filter, the + /// late-join rebirth set, and what bounds the per-node tracker map. + /// + private sealed record AuthoredTable( + FrozenDictionary ByRawPath, + FrozenDictionary ByMetric, + FrozenDictionary ByScope, + FrozenSet EdgeNodes) + { + public static readonly AuthoredTable Empty = + Build(new Dictionary(StringComparer.Ordinal)); + + /// Every authored scope belonging to one edge node — the NDEATH fan-out set. + /// The edge node. + /// The scopes, node-own and device alike. + public IEnumerable ScopesUnder(SparkplugNodeKey node) => + ByScope.Keys.Where(s => s.IsUnder(node.GroupId, node.EdgeNodeId)); + + /// Builds the snapshot from the mapped definitions. + /// The mapped definitions, keyed by RawPath. + /// The immutable snapshot. + public static AuthoredTable Build(IReadOnlyDictionary byRawPath) + { + var byMetric = new Dictionary>(); + var byScope = new Dictionary>(); + var nodes = new HashSet(); + + foreach (var def in byRawPath.Values) + { + // The factory guarantees all three; a definition that reached here without them could + // never be routed, so it is skipped rather than indexed under a nonsense key. + if (def.GroupId is not { } group || def.EdgeNodeId is not { } node || def.MetricName is not { } metric) + { + continue; + } + + var scope = new SparkplugScope(group, node, def.DeviceId); + nodes.Add(new SparkplugNodeKey(group, node)); + + if (!byMetric.TryGetValue(new SparkplugMetricKey(group, node, def.DeviceId, metric), out var metricList)) + { + byMetric[new SparkplugMetricKey(group, node, def.DeviceId, metric)] = metricList = []; + } + + metricList.Add(def); + + if (!byScope.TryGetValue(scope, out var scopeList)) + { + byScope[scope] = scopeList = []; + } + + scopeList.Add(def); + } + + return new AuthoredTable( + byRawPath.ToFrozenDictionary(StringComparer.Ordinal), + byMetric.ToFrozenDictionary(kv => kv.Key, kv => kv.Value.ToArray()), + byScope.ToFrozenDictionary(kv => kv.Key, kv => kv.Value.ToArray()), + nodes.ToFrozenSet()); + } + } +} + +/// +/// One edge node's identity — the unit Sparkplug sequences messages by, and the unit a rebirth NCMD +/// is addressed to. +/// +/// The Sparkplug group id. +/// The Sparkplug edge-node id. +public readonly record struct SparkplugNodeKey(string GroupId, string EdgeNodeId) +{ + /// + public override string ToString() => $"{GroupId}/{EdgeNodeId}"; +} + +/// +/// An authored tag's Sparkplug binding key: the stable tuple a tag is resolved by, across every +/// rebirth. The metric name is part of it; the per-birth alias never is. +/// +/// The Sparkplug group id. +/// The Sparkplug edge-node id. +/// The Sparkplug device id, or for a node-level metric. +/// The stable metric name. +public readonly record struct SparkplugMetricKey( + string GroupId, + string EdgeNodeId, + string? DeviceId, + string MetricName) +{ + /// + public override string ToString() => + DeviceId is null + ? $"{GroupId}/{EdgeNodeId}:{MetricName}" + : $"{GroupId}/{EdgeNodeId}/{DeviceId}:{MetricName}"; +} + +/// An observed Sparkplug primary-host STATE. +/// The host id the STATE topic named. +/// Whether the host declared itself online. +/// When this driver observed it. +public sealed record SparkplugHostState(string HostId, bool Online, DateTime ObservedUtc); + +/// Event payload for . +/// The observed state. +public sealed class SparkplugHostStateEventArgs(SparkplugHostState state) : EventArgs +{ + /// The observed state. + public SparkplugHostState State { get; } = state; +} + +/// +/// Event payload for — the Task-22 rediscovery seam. +/// +/// The scope whose birth certificate was applied. +/// The named metrics the birth declared, in wire order. +public sealed class SparkplugBirthObservedEventArgs(SparkplugScope scope, IReadOnlyList metricNames) + : EventArgs +{ + /// The scope whose birth certificate was applied. + public SparkplugScope Scope { get; } = scope; + + /// The named metrics the birth declared. + public IReadOnlyList MetricNames { get; } = metricNames; +} + +/// +/// Coerces a Sparkplug wire value — after +/// has undone the two's-complement encoding — onto +/// an authored tag's . +/// +/// +/// +/// A value that does not fit is refused, never silently converted. Same rule the +/// plain-MQTT path follows: a wrong value delivered at Good quality is worse than no value. +/// The one deliberate latitude is an exactly integral real (5.0) for an integer +/// tag — edge gateways emit those routinely — while a genuinely fractional one is refused +/// rather than rounded. +/// +/// +/// Sparkplug DateTime is epoch milliseconds, carried in the unsigned 64-bit +/// field, not an ISO string. Reading it as a number and calling it a timestamp is the whole +/// conversion. +/// +/// +internal static class SparkplugValueCoercion +{ + /// Coerces onto . + /// The reinterpreted wire value. + /// The tag's effective data type. + /// The coerced value when this returns . + /// when the value fits the target type. + public static bool TryCoerce(object? raw, DriverDataType target, out object? value) + { + value = null; + if (raw is null) + { + return false; + } + + try + { + switch (target) + { + case DriverDataType.String: + case DriverDataType.Reference: + // Bytes/File metrics map to String per the datatype map; base64 is the v1 fallback + // the design names, and it round-trips where a lossy UTF-8 decode would not. + value = raw switch + { + string s => s, + byte[] bytes => Convert.ToBase64String(bytes), + bool b => b ? "true" : "false", + _ => Convert.ToString(raw, CultureInfo.InvariantCulture), + }; + return value is not null; + + case DriverDataType.Boolean: + return TryCoerceBoolean(raw, out value); + + case DriverDataType.Int16: + case DriverDataType.Int32: + case DriverDataType.Int64: + case DriverDataType.UInt16: + case DriverDataType.UInt32: + case DriverDataType.UInt64: + return TryCoerceInteger(raw, target, out value); + + case DriverDataType.Float32: + if (raw is string f32Text) + { + return float.TryParse(f32Text, NumberStyles.Float, CultureInfo.InvariantCulture, out var pf) + && Assign(pf, out value); + } + + value = Convert.ToSingle(raw, CultureInfo.InvariantCulture); + return true; + + case DriverDataType.Float64: + if (raw is string f64Text) + { + return double.TryParse(f64Text, NumberStyles.Float, CultureInfo.InvariantCulture, out var pd) + && Assign(pd, out value); + } + + value = Convert.ToDouble(raw, CultureInfo.InvariantCulture); + return true; + + case DriverDataType.DateTime: + return TryCoerceDateTime(raw, out value); + + default: + return false; + } + } + catch (Exception ex) when (ex is OverflowException or FormatException or InvalidCastException) + { + // Out of range, unparseable, or a wire type that cannot become the target at all. All three + // are "does not fit", which is a Bad status on one tag — never an exception on the + // dispatcher thread. + value = null; + return false; + } + } + + private static bool Assign(T parsed, out object? value) + { + value = parsed; + return true; + } + + private static bool TryCoerceBoolean(object raw, out object? value) + { + switch (raw) + { + case bool b: + value = b; + return true; + case string s when bool.TryParse(s.Trim(), out var parsed): + value = parsed; + return true; + case string s: + value = null; + return long.TryParse(s.Trim(), NumberStyles.Integer, CultureInfo.InvariantCulture, out var n) + && Assign(n != 0L, out value); + case byte[]: + value = null; + return false; + default: + // Any numeric wire type: nonzero is true, matching the plain-mode path's rule. + value = Convert.ToDouble(raw, CultureInfo.InvariantCulture) != 0d; + return true; + } + } + + private static bool TryCoerceInteger(object raw, DriverDataType target, out object? value) + { + value = null; + + // Reals (and textual reals) must be exactly integral: rounding 5.5 to 5 or 6 is a wrong value, + // which is worse than no value. Parsed as decimal so the test and the conversion stay exact + // across the whole Int64/UInt64 range, where double loses precision above 2^53. + if (raw is float or double or decimal or string) + { + var text = raw as string ?? Convert.ToString(raw, CultureInfo.InvariantCulture); + if (text is null + || !decimal.TryParse(text, NumberStyles.Float, CultureInfo.InvariantCulture, out var dec) + || dec != decimal.Truncate(dec)) + { + return false; + } + + raw = dec; + } + + value = target switch + { + DriverDataType.Int16 => Convert.ToInt16(raw, CultureInfo.InvariantCulture), + DriverDataType.Int32 => Convert.ToInt32(raw, CultureInfo.InvariantCulture), + DriverDataType.Int64 => Convert.ToInt64(raw, CultureInfo.InvariantCulture), + DriverDataType.UInt16 => Convert.ToUInt16(raw, CultureInfo.InvariantCulture), + DriverDataType.UInt32 => Convert.ToUInt32(raw, CultureInfo.InvariantCulture), + DriverDataType.UInt64 => Convert.ToUInt64(raw, CultureInfo.InvariantCulture), + _ => null, + }; + + return value is not null; + } + + private static bool TryCoerceDateTime(object raw, out object? value) + { + value = null; + + switch (raw) + { + case DateTime dt: + value = dt; + return true; + + case string s: + return DateTime.TryParse( + s, + CultureInfo.InvariantCulture, + DateTimeStyles.RoundtripKind | DateTimeStyles.AdjustToUniversal | DateTimeStyles.AssumeUniversal, + out var parsed) + && Assign(parsed, out value); + + case bool: + case byte[]: + return false; + + default: + // Sparkplug DateTime rides as epoch milliseconds in the unsigned 64-bit value field. + var ms = Convert.ToInt64(raw, CultureInfo.InvariantCulture); + if (ms < DateTimeOffset.MinValue.ToUnixTimeMilliseconds() + || ms > DateTimeOffset.MaxValue.ToUnixTimeMilliseconds()) + { + return false; + } + + value = DateTimeOffset.FromUnixTimeMilliseconds(ms).UtcDateTime; + return true; + } + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/SparkplugIngestorTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/SparkplugIngestorTests.cs new file mode 100644 index 00000000..4dde20b9 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/SparkplugIngestorTests.cs @@ -0,0 +1,1308 @@ +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; + +/// +/// Covers (Task 21) — the design doc's §3.6 correctness matrix: +/// rebirth, missed birth, seq gap, alias reuse across a rebirth, death → stale → rebirth, plus +/// data-before-birth and unknown-alias. Every test runs without a broker: payloads are built with +/// the generated Tahu types, decoded through the real , and fed to +/// (the method +/// calls on MQTTnet's dispatcher thread). +/// +/// +/// The identity assertion is the point of half of these. The plan's illustrative snippet +/// asserts firedRef.ShouldContain("Filler1:Temperature") — the retired pre-v3 shape. A +/// Sparkplug-composed reference satisfies it and is silently dead against DriverHostActor's +/// RawPath-keyed dual-namespace fan-out, so every publish here is asserted to be the RawPath +/// and to not be the composed form. +/// +public sealed class SparkplugIngestorTests +{ + // OPC UA status codes, verified against Opc.Ua.StatusCodes rather than recalled. + private const uint Good = 0x00000000u; + private const uint BadCommunicationError = 0x80050000u; + private const uint BadTypeMismatch = 0x80740000u; + private const uint BadNodeIdUnknown = 0x80340000u; + + private const string Group = "Plant1"; + private const string Node = "EdgeA"; + private const string Device = "Filler1"; + + private const string TempPath = "Plant/Mqtt/spb/Filler1Temp"; + private const string PressPath = "Plant/Mqtt/spb/Filler1Press"; + private const string CountPath = "Plant/Mqtt/spb/Filler1Count"; + private const string NodeUptimePath = "Plant/Mqtt/spb/EdgeAUptime"; + + /// The retired pre-v3 composed shape — asserted ABSENT, never present. + private const string ComposedRef = "Plant1/EdgeA/Filler1:Temperature"; + + private const ulong TempAlias = 5UL; + private const ulong PressAlias = 6UL; + private const ulong CountAlias = 9UL; + + // --------------------------------------------------------------------------------- + // §3.6 #1/#2 — birth → data, resolved BY ALIAS, published UNDER THE RAWPATH + // --------------------------------------------------------------------------------- + + /// + /// The plan's first snippet, corrected: the published reference is asserted to BE the tag's + /// RawPath, not merely to contain the composed Filler1:Temperature form. + /// + [Fact] + public async Task BirthThenData_ResolvesByAlias_RaisesOnDataChangeUnderRawPath() + { + var ing = await NewSubscribedIngestorAsync(); + + string? firedRef = null; + object? firedValue = null; + ing.OnDataChange += (_, e) => { firedRef = e.FullReference; firedValue = e.Snapshot.Value; }; + + ing.Dispatch(Topic(SparkplugMessageType.NBIRTH), Nbirth(seq: 0, bdSeq: 1)); + ing.Dispatch(Topic(SparkplugMessageType.DBIRTH, Device), Dbirth(seq: 1)); + ing.Dispatch( + Topic(SparkplugMessageType.DDATA, Device), + Ddata(seq: 2, (TempAlias, TahuDataType.Float, 21.5f))); + + firedRef.ShouldBe(TempPath); + firedRef.ShouldNotBe(ComposedRef); + firedValue.ShouldBe(21.5f); + } + + /// + /// Falsifiability control for the identity assertion above: the composed Sparkplug reference — + /// the shape the plan's own snippet would have accepted — must never be published. + /// + [Fact] + public async Task PublishedReference_IsNeverTheComposedSparkplugForm() + { + var ing = await NewSubscribedIngestorAsync(); + var refs = new List(); + ing.OnDataChange += (_, e) => refs.Add(e.FullReference); + + FeedBirths(ing); + ing.Dispatch( + Topic(SparkplugMessageType.DDATA, Device), + Ddata(seq: 2, (TempAlias, TahuDataType.Float, 21.5f))); + + refs.ShouldNotBeEmpty(); + refs.ShouldAllBe(r => r == TempPath || r == PressPath || r == CountPath || r == NodeUptimePath); + refs.ShouldNotContain(ComposedRef); + } + + /// A DATA metric carries an alias and NO name — the shape every real Sparkplug DATA has. + [Fact] + public async Task DataMetric_CarriesNoNameOrDatatype_StillResolves() + { + var ing = await NewSubscribedIngestorAsync(); + FeedBirths(ing); + + var payload = Ddata(seq: 2, (TempAlias, TahuDataType.Float, 33.5f)); + payload.Metrics.ShouldAllBe(m => m.Name == null && m.DataType == null); + + var seen = new Dictionary(StringComparer.Ordinal); + ing.OnDataChange += (_, e) => seen[e.FullReference] = e.Snapshot; + ing.Dispatch(Topic(SparkplugMessageType.DDATA, Device), payload); + + seen[TempPath].Value.ShouldBe(33.5f); + seen[TempPath].StatusCode.ShouldBe(Good); + } + + /// A node-scoped metric (NDATA) binds against the node's own scope, not a device's. + [Fact] + public async Task NodeScopedMetric_BindsToTheNodeScope() + { + var ing = await NewSubscribedIngestorAsync(); + var seen = new Dictionary(StringComparer.Ordinal); + ing.OnDataChange += (_, e) => seen[e.FullReference] = e.Snapshot.Value; + + ing.Dispatch(Topic(SparkplugMessageType.NBIRTH), Nbirth(seq: 0, bdSeq: 1)); + ing.Dispatch(Topic(SparkplugMessageType.NDATA), Ndata(seq: 1, (7UL, TahuDataType.Int64, 99L))); + + seen[NodeUptimePath].ShouldBe(99L); + seen.ShouldNotContainKey(TempPath); + } + + // --------------------------------------------------------------------------------- + // §3.6 #1 — ALIAS REUSE ACROSS A REBIRTH (the mis-route this whole design prevents) + // --------------------------------------------------------------------------------- + + /// + /// The single most important test in the file. An edge node is free to point alias 5 at a + /// different metric after a rebirth. Binding a tag to an alias silently routes Pressure into the + /// Temperature tag — good quality, plausible value, completely wrong. + /// + [Fact] + public async Task AliasReusedAcrossRebirth_RoutesByName_NotByAlias() + { + var ing = await NewSubscribedIngestorAsync(); + var seen = new Dictionary(StringComparer.Ordinal); + ing.OnDataChange += (_, e) => seen[e.FullReference] = e.Snapshot.Value; + + ing.Dispatch(Topic(SparkplugMessageType.NBIRTH), Nbirth(seq: 0, bdSeq: 1)); + ing.Dispatch(Topic(SparkplugMessageType.DBIRTH, Device), Dbirth(seq: 1)); + ing.Dispatch( + Topic(SparkplugMessageType.DDATA, Device), + Ddata(seq: 2, (TempAlias, TahuDataType.Float, 21.5f))); + seen[TempPath].ShouldBe(21.5f); + + // Rebirth: the SAME alias 5 now names Pressure, and Temperature moves to alias 6. + ing.Dispatch(Topic(SparkplugMessageType.NBIRTH), Nbirth(seq: 0, bdSeq: 2)); + ing.Dispatch( + Topic(SparkplugMessageType.DBIRTH, Device), + DbirthCustom( + seq: 1, + ("Pressure", TempAlias, TahuDataType.Float, 1.0f), + ("Temperature", PressAlias, TahuDataType.Float, 2.0f))); + + seen.Clear(); + ing.Dispatch( + Topic(SparkplugMessageType.DDATA, Device), + Ddata(seq: 2, (TempAlias, TahuDataType.Float, 4.25f))); + + // Alias 5 is Pressure NOW. A tag bound by alias would put 4.25 on the Temperature RawPath. + seen.ShouldContainKey(PressPath); + seen[PressPath].ShouldBe(4.25f); + seen.ShouldNotContainKey(TempPath); + } + + /// + /// A rebirth REPLACES the catalog: an alias the new birth does not declare must stop resolving, + /// rather than lingering from the previous birth's table. + /// + [Fact] + public async Task Rebirth_DroppingAMetric_StopsFeedingIt() + { + var ing = await NewSubscribedIngestorAsync(); + var seen = new List(); + ing.OnDataChange += (_, e) => seen.Add(e.FullReference); + + FeedBirths(ing); + ing.Dispatch(Topic(SparkplugMessageType.NBIRTH), Nbirth(seq: 0, bdSeq: 2)); + ing.Dispatch( + Topic(SparkplugMessageType.DBIRTH, Device), + DbirthCustom(seq: 1, ("Pressure", PressAlias, TahuDataType.Float, 0f))); + + seen.Clear(); + ing.Dispatch( + Topic(SparkplugMessageType.DDATA, Device), + Ddata(seq: 2, (TempAlias, TahuDataType.Float, 21.5f))); + + seen.ShouldNotContain(TempPath); + } + + // --------------------------------------------------------------------------------- + // §3.6 #3 — data before birth / unknown alias / seq gap ⇒ rebirth NCMD + // --------------------------------------------------------------------------------- + + /// The plan's second snippet, against the real NCMD publish seam. + [Fact] + public async Task DataBeforeBirth_RequestsRebirth() + { + var publish = new RecordingPublishTransport(); + var ing = await NewSubscribedIngestorAsync(publish); + + ing.Dispatch( + Topic(SparkplugMessageType.DDATA, Device), + Ddata(seq: 0, (CountAlias, TahuDataType.Float, 1f))); + await ing.DrainPendingRebirthsAsync(); + + publish.Topics.ShouldContain("spBv1.0/Plant1/NCMD/EdgeA"); + } + + /// An alias the current birth never declared is our metadata being behind the node's. + [Fact] + public async Task UnknownAlias_RequestsRebirth() + { + var publish = new RecordingPublishTransport(); + var ing = await NewSubscribedIngestorAsync(publish); + FeedBirths(ing); + publish.Topics.Clear(); + + ing.Dispatch( + Topic(SparkplugMessageType.DDATA, Device), + Ddata(seq: 2, (4242UL, TahuDataType.Float, 1f))); + await ing.DrainPendingRebirthsAsync(); + + publish.Topics.ShouldContain("spBv1.0/Plant1/NCMD/EdgeA"); + } + + /// A missed message in the edge node's stream — §3.6 #3. + [Fact] + public async Task SeqGap_RequestsRebirth() + { + var publish = new RecordingPublishTransport(); + var ing = await NewSubscribedIngestorAsync(publish); + FeedBirths(ing); + publish.Topics.Clear(); + + // seq 2 was expected; 4 means two messages were lost. + ing.Dispatch( + Topic(SparkplugMessageType.DDATA, Device), + Ddata(seq: 4, (TempAlias, TahuDataType.Float, 1f))); + await ing.DrainPendingRebirthsAsync(); + + publish.Topics.ShouldContain("spBv1.0/Plant1/NCMD/EdgeA"); + } + + /// + /// Falsifiability control for every rebirth test above: a contiguous, birth-synchronised stream + /// must publish NOTHING. Without this the rebirth assertions could pass off a driver that + /// NCMDs on every single message. + /// + [Fact] + public async Task ContiguousStream_RequestsNoRebirth() + { + var publish = new RecordingPublishTransport(); + var ing = await NewSubscribedIngestorAsync(publish); + + FeedBirths(ing); + ing.Dispatch(Topic(SparkplugMessageType.DDATA, Device), Ddata(seq: 2, (TempAlias, TahuDataType.Float, 1f))); + ing.Dispatch(Topic(SparkplugMessageType.DDATA, Device), Ddata(seq: 3, (TempAlias, TahuDataType.Float, 2f))); + await ing.DrainPendingRebirthsAsync(); + + publish.Topics.ShouldBeEmpty(); + } + + /// + /// 255 → 0 is the sequence continuing, not a gap. Reading the wrap as a gap asks every edge node + /// to rebirth once per 256 messages, so a busy node spends its life republishing births. + /// + [Fact] + public async Task SeqWrapsFrom255To0_IsNotAGap() + { + var publish = new RecordingPublishTransport(); + var ing = await NewSubscribedIngestorAsync(publish); + + ing.Dispatch(Topic(SparkplugMessageType.NBIRTH), Nbirth(seq: 255, bdSeq: 1)); + ing.Dispatch(Topic(SparkplugMessageType.DBIRTH, Device), Dbirth(seq: 0)); + ing.Dispatch(Topic(SparkplugMessageType.DDATA, Device), Ddata(seq: 1, (TempAlias, TahuDataType.Float, 1f))); + await ing.DrainPendingRebirthsAsync(); + + publish.Topics.ShouldBeEmpty(); + } + + /// + /// A data message whose sequence happens to be contiguous but which arrived with no NBIRTH seen + /// since connect — the driver joined mid-stream and holds no alias table. + /// returns for it, so a driver that + /// only read the return value would sit happily behind a stream it cannot decode. + /// + [Fact] + public async Task LateJoin_FirstMessageIsNotABirth_RequestsRebirth() + { + var publish = new RecordingPublishTransport(); + var ing = await NewSubscribedIngestorAsync(publish); + + ing.Dispatch(Topic(SparkplugMessageType.DBIRTH, Device), Dbirth(seq: 17)); + await ing.DrainPendingRebirthsAsync(); + + publish.Topics.ShouldContain("spBv1.0/Plant1/NCMD/EdgeA"); + } + + /// + /// Repeatedly commanding a node that is already republishing IS the storm the design warns + /// about. Three unroutable messages in a row must produce one NCMD, not three. + /// + [Fact] + public async Task RepeatedRebirthTriggers_AreDebouncedPerEdgeNode() + { + var publish = new RecordingPublishTransport(); + var ing = await NewSubscribedIngestorAsync(publish, rebirthDebounce: TimeSpan.FromMinutes(5)); + + for (var i = 0; i < 3; i++) + { + ing.Dispatch( + Topic(SparkplugMessageType.DDATA, Device), + Ddata(seq: (ulong)i, (CountAlias, TahuDataType.Float, 1f))); + } + + await ing.DrainPendingRebirthsAsync(); + + publish.Topics.Count.ShouldBe(1); + } + + /// The debounce floor is per edge node, not global — a second node must not be starved. + [Fact] + public async Task RebirthDebounce_IsPerEdgeNode() + { + var publish = new RecordingPublishTransport(); + var ing = await NewSubscribedIngestorAsync( + publish, + rebirthDebounce: TimeSpan.FromMinutes(5), + extraTags: [Tag("Plant/Mqtt/spb/EdgeBTemp", Blob(Group, "EdgeB", Device, "Temperature", "Float32"))]); + + ing.Dispatch(Topic(SparkplugMessageType.DDATA, Device), Ddata(seq: 0, (CountAlias, TahuDataType.Float, 1f))); + ing.Dispatch( + new SparkplugTopic(SparkplugMessageType.DDATA, Group, "EdgeB", Device, null), + Ddata(seq: 0, (CountAlias, TahuDataType.Float, 1f))); + await ing.DrainPendingRebirthsAsync(); + + publish.Topics.ShouldContain("spBv1.0/Plant1/NCMD/EdgeA"); + publish.Topics.ShouldContain("spBv1.0/Plant1/NCMD/EdgeB"); + } + + /// requestRebirthOnGap = false is the operator's off switch for ALL outbound NCMDs. + [Fact] + public async Task RequestRebirthOnGapDisabled_PublishesNothing() + { + var publish = new RecordingPublishTransport(); + var ing = await NewSubscribedIngestorAsync(publish, requestRebirthOnGap: false); + + ing.Dispatch(Topic(SparkplugMessageType.DDATA, Device), Ddata(seq: 0, (CountAlias, TahuDataType.Float, 1f))); + await ing.DrainPendingRebirthsAsync(); + + publish.Topics.ShouldBeEmpty(); + } + + /// A missing publish transport degrades to a log, never an exception on the dispatcher. + [Fact] + public async Task RebirthWithNoPublishTransport_DoesNotThrow() + { + var ing = await NewSubscribedIngestorAsync(publish: null); + + Should.NotThrow(() => ing.Dispatch( + Topic(SparkplugMessageType.DDATA, Device), + Ddata(seq: 0, (CountAlias, TahuDataType.Float, 1f)))); + } + + // --------------------------------------------------------------------------------- + // §3.6 #4 — death → STALE, next birth restores Good + // --------------------------------------------------------------------------------- + + /// The plan's third snippet. + [Fact] + public async Task Ddeath_EmitsBadForDeviceMetrics() + { + var ing = await NewSubscribedIngestorAsync(); + FeedBirths(ing); + + var quals = new List(); + ing.OnDataChange += (_, e) => quals.Add(e.Snapshot.StatusCode); + + ing.Dispatch(Topic(SparkplugMessageType.DDEATH, Device), Ddeath(seq: 2)); + + quals.ShouldNotBeEmpty(); + quals.ShouldAllBe(q => q == BadCommunicationError); + } + + /// The completing half of §3.6 #4 — a birth is a full snapshot, so it restores Good. + [Fact] + public async Task DdeathThenDbirth_RestoresGood() + { + var ing = await NewSubscribedIngestorAsync(); + FeedBirths(ing); + + var last = new Dictionary(StringComparer.Ordinal); + ing.OnDataChange += (_, e) => last[e.FullReference] = e.Snapshot; + + ing.Dispatch(Topic(SparkplugMessageType.DDEATH, Device), Ddeath(seq: 2)); + last[TempPath].StatusCode.ShouldBe(BadCommunicationError); + + ing.Dispatch(Topic(SparkplugMessageType.DBIRTH, Device), Dbirth(seq: 3)); + + last[TempPath].StatusCode.ShouldBe(Good); + last[TempPath].Value.ShouldBe(0f); + } + + /// A DDEATH stales exactly its own device; the node's own metrics keep flowing. + [Fact] + public async Task Ddeath_DoesNotStaleTheNodesOwnMetrics() + { + var ing = await NewSubscribedIngestorAsync(); + FeedBirths(ing); + + var last = new Dictionary(StringComparer.Ordinal); + ing.OnDataChange += (_, e) => last[e.FullReference] = e.Snapshot; + + ing.Dispatch(Topic(SparkplugMessageType.DDEATH, Device), Ddeath(seq: 2)); + + last.ShouldContainKey(TempPath); + last.ShouldNotContainKey(NodeUptimePath); + } + + /// An NDEATH kills the node AND every device under it. + [Fact] + public async Task Ndeath_StalesTheNodeAndEveryDeviceUnderIt() + { + var ing = await NewSubscribedIngestorAsync(); + FeedBirths(ing); + + var last = new Dictionary(StringComparer.Ordinal); + ing.OnDataChange += (_, e) => last[e.FullReference] = e.Snapshot; + + ing.Dispatch(Topic(SparkplugMessageType.NDEATH), Ndeath(bdSeq: 1)); + + last[TempPath].StatusCode.ShouldBe(BadCommunicationError); + last[NodeUptimePath].StatusCode.ShouldBe(BadCommunicationError); + } + + /// + /// The bdSeq tie of §3.6 #3. A node's connection drops, it reconnects and re-births, and + /// only THEN does the broker deliver the old session's Last Will. Acting on it marks a live node + /// dead with nothing coming to correct it. + /// + [Fact] + public async Task StaleNdeath_FromAPreviousSession_IsIgnored() + { + var ing = await NewSubscribedIngestorAsync(); + + ing.Dispatch(Topic(SparkplugMessageType.NBIRTH), Nbirth(seq: 0, bdSeq: 2)); + ing.Dispatch(Topic(SparkplugMessageType.DBIRTH, Device), Dbirth(seq: 1)); + + var quals = new List(); + ing.OnDataChange += (_, e) => quals.Add(e.Snapshot.StatusCode); + + ing.Dispatch(Topic(SparkplugMessageType.NDEATH), Ndeath(bdSeq: 1)); + + quals.ShouldBeEmpty(); + ing.Births.Find(new SparkplugScope(Group, Node, Device)).ShouldNotBeNull(); + } + + /// + /// NDEATH must never reach . It is the broker-published + /// Last Will and carries no seq at all, so Accept would report a gap on every death + /// and answer a node that has just died with a rebirth request it can never serve. + /// + [Fact] + public async Task Ndeath_DoesNotAdvanceTheSequenceTracker_AndRequestsNoRebirth() + { + var publish = new RecordingPublishTransport(); + var ing = await NewSubscribedIngestorAsync(publish); + + FeedBirths(ing); + publish.Topics.Clear(); + + ing.Dispatch(Topic(SparkplugMessageType.NDEATH), Ndeath(bdSeq: 1)); + await ing.DrainPendingRebirthsAsync(); + + publish.Topics.ShouldBeEmpty(); + } + + /// + /// Per the Sparkplug spec an NBIRTH voids every prior DBIRTH under that node: those devices MUST + /// re-DBIRTH, and until they do their aliases mean nothing. + /// + [Fact] + public async Task Nbirth_InvalidatesPriorDeviceBirths_AndStalesThem() + { + var ing = await NewSubscribedIngestorAsync(); + FeedBirths(ing); + + var last = new Dictionary(StringComparer.Ordinal); + ing.OnDataChange += (_, e) => last[e.FullReference] = e.Snapshot; + + ing.Dispatch(Topic(SparkplugMessageType.NBIRTH), Nbirth(seq: 0, bdSeq: 2)); + + last[TempPath].StatusCode.ShouldBe(BadCommunicationError); + ing.Births.Find(new SparkplugScope(Group, Node, Device)).ShouldBeNull(); + } + + // --------------------------------------------------------------------------------- + // Wire-level value correctness + // --------------------------------------------------------------------------------- + + /// + /// Sparkplug carries Int8/16/32/64 as two's complement in an unsigned proto field, so a + /// metric whose value is -43 arrives as 4294967253. Skipping + /// publishes a plausible, Good-quality, + /// completely wrong number — or, once the tag is Int32, an out-of-range refusal. + /// + [Fact] + public async Task NegativeInteger_IsReinterpretedFromItsTwosComplementWireValue() + { + var ing = await NewSubscribedIngestorAsync(); + FeedBirths(ing); + + var seen = new Dictionary(StringComparer.Ordinal); + ing.OnDataChange += (_, e) => seen[e.FullReference] = e.Snapshot; + + ing.Dispatch( + Topic(SparkplugMessageType.DDATA, Device), + Ddata(seq: 2, (CountAlias, TahuDataType.Int32, -43))); + + seen[CountPath].StatusCode.ShouldBe(Good); + seen[CountPath].Value.ShouldBe(-43); + seen[CountPath].Value.ShouldNotBe(4294967253L); + } + + /// The birth's own values are published too — that is how a rebirth restores Good. + [Fact] + public async Task BirthValues_ArePublished() + { + var ing = await NewSubscribedIngestorAsync(); + var seen = new Dictionary(StringComparer.Ordinal); + ing.OnDataChange += (_, e) => seen[e.FullReference] = e.Snapshot.Value; + + FeedBirths(ing); + + seen[TempPath].ShouldBe(0f); + seen[CountPath].ShouldBe(-42); + } + + /// + /// A tag authored with no dataType takes the type its BIRTH declared — the reason + /// dataType is optional in the Sparkplug tag shape at all. + /// + [Fact] + public async Task TagWithNoAuthoredDataType_TakesTheBirthsType() + { + const string Untyped = "Plant/Mqtt/spb/Untyped"; + var ing = NewIngestor(tags: [Tag(Untyped, """{"groupId":"Plant1","edgeNodeId":"EdgeA","deviceId":"Filler1","metricName":"Temperature"}""")]); + await ing.SubscribeAsync([Untyped], TimeSpan.Zero, TestContext.Current.CancellationToken); + + object? value = null; + ing.OnDataChange += (_, e) => value = e.Snapshot.Value; + + ing.Dispatch(Topic(SparkplugMessageType.NBIRTH), Nbirth(seq: 0, bdSeq: 1)); + ing.Dispatch(Topic(SparkplugMessageType.DBIRTH, Device), Dbirth(seq: 1)); + ing.Dispatch(Topic(SparkplugMessageType.DDATA, Device), Ddata(seq: 2, (TempAlias, TahuDataType.Float, 7.5f))); + + // Float32, from the birth — NOT the record's String default, which would have produced "7.5". + value.ShouldBe(7.5f); + } + + /// An authored dataType wins over the birth's — the operator declared it deliberately. + [Fact] + public async Task AuthoredDataType_WinsOverTheBirths() + { + const string AsText = "Plant/Mqtt/spb/TempAsText"; + var ing = NewIngestor(tags: [Tag(AsText, Blob(Group, Node, Device, "Temperature", "String"))]); + await ing.SubscribeAsync([AsText], TimeSpan.Zero, TestContext.Current.CancellationToken); + + object? value = null; + ing.OnDataChange += (_, e) => value = e.Snapshot.Value; + + ing.Dispatch(Topic(SparkplugMessageType.NBIRTH), Nbirth(seq: 0, bdSeq: 1)); + ing.Dispatch(Topic(SparkplugMessageType.DBIRTH, Device), Dbirth(seq: 1)); + ing.Dispatch(Topic(SparkplugMessageType.DDATA, Device), Ddata(seq: 2, (TempAlias, TahuDataType.Float, 7.5f))); + + value.ShouldBe("7.5"); + } + + /// An explicit Sparkplug is_null is not a value of the tag's declared type. + [Fact] + public async Task ExplicitNullMetric_PublishesBadTypeMismatch() + { + var ing = await NewSubscribedIngestorAsync(); + FeedBirths(ing); + + var seen = new Dictionary(StringComparer.Ordinal); + ing.OnDataChange += (_, e) => seen[e.FullReference] = e.Snapshot; + + var payload = new Payload { Seq = 2UL, Timestamp = 1UL }; + payload.Metrics.Add(new Payload.Types.Metric { Alias = TempAlias, IsNull = true }); + ing.Dispatch(Topic(SparkplugMessageType.DDATA, Device), Decode(payload)); + + seen[TempPath].StatusCode.ShouldBe(BadTypeMismatch); + seen[TempPath].Value.ShouldBeNull(); + } + + /// + /// A DataSet/Template metric is warn-and-skip — NOT a missing metric and NOT a rebirth trigger. + /// A node asked to rebirth would answer with the same unsupported metric and the pair would loop. + /// + [Fact] + public async Task UnsupportedMetricKind_IsSkipped_AndRequestsNoRebirth() + { + var publish = new RecordingPublishTransport(); + var ing = await NewSubscribedIngestorAsync(publish); + FeedBirths(ing); + publish.Topics.Clear(); + + var seen = new List(); + ing.OnDataChange += (_, e) => seen.Add(e.FullReference); + + var payload = new Payload { Seq = 2UL, Timestamp = 1UL }; + payload.Metrics.Add(new Payload.Types.Metric { Alias = TempAlias, DatasetValue = new Payload.Types.DataSet() }); + ing.Dispatch(Topic(SparkplugMessageType.DDATA, Device), Decode(payload)); + await ing.DrainPendingRebirthsAsync(); + + seen.ShouldBeEmpty(); + publish.Topics.ShouldBeEmpty(); + } + + /// A value that does not fit the authored type is refused, never rounded or wrapped. + [Fact] + public async Task FractionalValueForAnIntegerTag_PublishesBadTypeMismatch() + { + const string IntTag = "Plant/Mqtt/spb/TempAsInt"; + var ing = NewIngestor(tags: [Tag(IntTag, Blob(Group, Node, Device, "Temperature", "Int32"))]); + await ing.SubscribeAsync([IntTag], TimeSpan.Zero, TestContext.Current.CancellationToken); + + DataValueSnapshot? snapshot = null; + ing.OnDataChange += (_, e) => snapshot = e.Snapshot; + + ing.Dispatch(Topic(SparkplugMessageType.NBIRTH), Nbirth(seq: 0, bdSeq: 1)); + ing.Dispatch(Topic(SparkplugMessageType.DBIRTH, Device), Dbirth(seq: 1)); + ing.Dispatch(Topic(SparkplugMessageType.DDATA, Device), Ddata(seq: 2, (TempAlias, TahuDataType.Float, 21.5f))); + + snapshot!.StatusCode.ShouldBe(BadTypeMismatch); + } + + // --------------------------------------------------------------------------------- + // Robustness: nothing may throw or mutate state off bad input + // --------------------------------------------------------------------------------- + + /// + /// An undecodable body is not evidence of anything. It must not clear an alias table, advance a + /// sequence tracker, or evict a birth — a hostile publisher on a group-wide # subscription + /// would otherwise be able to blank the address space with garbage bytes. + /// + [Fact] + public async Task InvalidPayload_MutatesNothing() + { + var publish = new RecordingPublishTransport(); + var ing = await NewSubscribedIngestorAsync(publish); + FeedBirths(ing); + publish.Topics.Clear(); + + ing.HandleMessage("spBv1.0/Plant1/DDATA/EdgeA/Filler1", [0xDE, 0xAD, 0xBE, 0xEF], retained: false); + await ing.DrainPendingRebirthsAsync(); + + // The birth survived … + ing.Births.Find(new SparkplugScope(Group, Node, Device)).ShouldNotBeNull(); + + // … the sequence baseline survived (seq 2 is still contiguous, so no rebirth) … + var seen = new Dictionary(StringComparer.Ordinal); + ing.OnDataChange += (_, e) => seen[e.FullReference] = e.Snapshot.Value; + ing.Dispatch(Topic(SparkplugMessageType.DDATA, Device), Ddata(seq: 2, (TempAlias, TahuDataType.Float, 9f))); + await ing.DrainPendingRebirthsAsync(); + + seen[TempPath].ShouldBe(9f); + publish.Topics.ShouldBeEmpty(); + } + + /// An undecodable NDEATH must not kill a live node either. + [Fact] + public async Task InvalidNdeathPayload_DoesNotStale() + { + var ing = await NewSubscribedIngestorAsync(); + FeedBirths(ing); + + var quals = new List(); + ing.OnDataChange += (_, e) => quals.Add(e.Snapshot.StatusCode); + + ing.HandleMessage("spBv1.0/Plant1/NDEATH/EdgeA", [0x01, 0x02, 0x03], retained: false); + + quals.ShouldBeEmpty(); + } + + /// + /// Runs on MQTTnet's dispatcher thread: an escaping exception stalls delivery for every + /// subscription on the connection. + /// + [Theory] + [InlineData("")] + [InlineData("not/a/sparkplug/topic")] + [InlineData("spBv1.0/Plant1/NBIRTH/EdgeA")] + [InlineData("spBv1.0/Plant1/DDATA/EdgeA/Filler1")] + [InlineData("spBv1.0/STATE/otopcua-host-1")] + [InlineData("spBv1.0")] + public async Task HandleMessage_NeverThrows(string topic) + { + var ing = await NewSubscribedIngestorAsync(); + + Should.NotThrow(() => ing.HandleMessage(topic, [], retained: false)); + Should.NotThrow(() => ing.HandleMessage(topic, [0xFF, 0x00, 0x7F], retained: true)); + } + + /// A throwing subscriber is contained rather than escaping onto the dispatcher thread. + [Fact] + public async Task ThrowingSubscriber_IsContained() + { + var ing = await NewSubscribedIngestorAsync(); + ing.OnDataChange += (_, _) => throw new InvalidOperationException("boom"); + + Should.NotThrow(() => FeedBirths(ing)); + } + + /// An oversize body is refused BEFORE any protobuf parse, and mutates nothing. + [Fact] + public async Task OversizePayload_IsDroppedBeforeDecoding() + { + var ing = NewIngestor(maxPayloadBytes: 16); + await ing.SubscribeAsync([TempPath], TimeSpan.Zero, TestContext.Current.CancellationToken); + + var bytes = Nbirth(seq: 0, bdSeq: 1); + bytes.Metrics.Count.ShouldBeGreaterThan(0); + + ing.HandleMessage("spBv1.0/Plant1/NBIRTH/EdgeA", new byte[64], retained: false); + + ing.Births.Count.ShouldBe(0); + } + + /// + /// The driver subscribes spBv1.0/{group}/#, so a large plant delivers traffic for every + /// node in the group. Only authored edge nodes are processed — that is what bounds the tracker + /// and birth maps by configuration rather than by whatever the plant publishes. + /// + [Fact] + public async Task UnauthoredEdgeNode_IsIgnoredEntirely() + { + var publish = new RecordingPublishTransport(); + var ing = await NewSubscribedIngestorAsync(publish); + + var fired = false; + ing.OnDataChange += (_, _) => fired = true; + + ing.Dispatch(new SparkplugTopic(SparkplugMessageType.NBIRTH, Group, "StrangerNode", null, null), Nbirth(seq: 0, bdSeq: 1)); + ing.Dispatch( + new SparkplugTopic(SparkplugMessageType.DDATA, Group, "StrangerNode", "Dev", null), + Ddata(seq: 1, (TempAlias, TahuDataType.Float, 1f))); + await ing.DrainPendingRebirthsAsync(); + + fired.ShouldBeFalse(); + ing.Births.Count.ShouldBe(0); + publish.Topics.ShouldBeEmpty(); + } + + /// + /// An unauthored DEVICE under an authored node still advances its edge node's sequence — dropping + /// it early would manufacture a gap on the very next authored message — and, because nothing in + /// that scope could ever be published, it must not demand a rebirth either. A spec-violating + /// device that publishes DDATA without ever DBIRTHing would otherwise drive a permanent NCMD at + /// the debounce cadence, forever. + /// + [Fact] + public async Task UnauthoredDeviceUnderAnAuthoredNode_StillAdvancesTheSequence() + { + var publish = new RecordingPublishTransport(); + var ing = await NewSubscribedIngestorAsync(publish); + FeedBirths(ing); + publish.Topics.Clear(); + + // seq 2 belongs to a device nobody authored … + ing.Dispatch( + new SparkplugTopic(SparkplugMessageType.DDATA, Group, Node, "OtherDevice", null), + Ddata(seq: 2, (TempAlias, TahuDataType.Float, 1f))); + + // … and seq 3 must therefore still be contiguous. + ing.Dispatch(Topic(SparkplugMessageType.DDATA, Device), Ddata(seq: 3, (TempAlias, TahuDataType.Float, 2f))); + await ing.DrainPendingRebirthsAsync(); + + publish.Topics.ShouldBeEmpty(); + } + + // --------------------------------------------------------------------------------- + // Registration / resolution / read + // --------------------------------------------------------------------------------- + + [Fact] + public void Register_MapsOnlySparkplugShapedBlobs() + { + var ing = NewIngestor(tags: []); + + var mapped = ing.Register([ + Tag(TempPath, Blob(Group, Node, Device, "Temperature", "Float32")), + Tag("Plant/Mqtt/spb/Plainish", """{"topic":"f/oven/temp","payloadFormat":"Scalar","dataType":"Float64"}"""), + Tag("Plant/Mqtt/spb/NoMetric", """{"groupId":"Plant1","edgeNodeId":"EdgeA"}"""), + ]); + + mapped.ShouldBe(1); + ing.TryResolve(TempPath, out var def).ShouldBeTrue(); + def.MetricName.ShouldBe("Temperature"); + def.Name.ShouldBe(TempPath); + } + + [Fact] + public void Register_IsWholesale_ADroppedTagStopsResolving() + { + var ing = NewIngestor(); + ing.TryResolve(TempPath, out _).ShouldBeTrue(); + + ing.Register([]); + + ing.TryResolve(TempPath, out _).ShouldBeFalse(); + } + + [Fact] + public async Task SubscribeAsync_UnauthoredReference_ReportsBadNodeIdUnknown() + { + var ing = NewIngestor(); + + await ing.SubscribeAsync(["Plant/Mqtt/spb/NotAThing"], TimeSpan.Zero, TestContext.Current.CancellationToken); + + ing.Values.Read("Plant/Mqtt/spb/NotAThing").StatusCode.ShouldBe(BadNodeIdUnknown); + } + + [Fact] + public async Task UnsubscribedButAuthoredTag_IsCachedButNotRaised() + { + var ing = NewIngestor(); + var fired = false; + ing.OnDataChange += (_, _) => fired = true; + + FeedBirths(ing); + + fired.ShouldBeFalse(); + ing.Values.Read(TempPath).Value.ShouldBe(0f); + } + + /// Two raw tags may legitimately project the same Sparkplug metric; both must be fed. + [Fact] + public async Task TwoTagsBoundToOneMetric_BothReceive() + { + const string Mirror = "Plant/Mqtt/spb/Filler1TempMirror"; + var ing = NewIngestor(extraTags: [Tag(Mirror, Blob(Group, Node, Device, "Temperature", "Float32"))]); + await ing.SubscribeAsync([TempPath, Mirror], TimeSpan.Zero, TestContext.Current.CancellationToken); + + var seen = new List(); + ing.OnDataChange += (_, e) => seen.Add(e.FullReference); + + FeedBirths(ing); + ing.Dispatch(Topic(SparkplugMessageType.DDATA, Device), Ddata(seq: 2, (TempAlias, TahuDataType.Float, 1f))); + + seen.ShouldContain(TempPath); + seen.ShouldContain(Mirror); + } + + // --------------------------------------------------------------------------------- + // Subscribe / reconnect + // --------------------------------------------------------------------------------- + + [Fact] + public async Task EstablishAsync_SubscribesTheGroupAndStateFilters_AndRequestsLateJoinRebirth() + { + var subscribe = new FakeSubscribeTransport(); + var publish = new RecordingPublishTransport(); + var ing = NewIngestor(publish, subscribe, hostId: "otopcua-host-1"); + + await ing.EstablishAsync(TestContext.Current.CancellationToken); + await ing.DrainPendingRebirthsAsync(); + + subscribe.Filters.Select(f => f.Topic).ShouldBe(["spBv1.0/Plant1/#", "spBv1.0/STATE/otopcua-host-1"]); + publish.Topics.ShouldContain("spBv1.0/Plant1/NCMD/EdgeA"); + } + + /// No host id ⇒ no STATE filter. This driver has no use for a stranger's host state. + [Fact] + public async Task EstablishAsync_WithNoHostId_SubscribesTheGroupFilterOnly() + { + var subscribe = new FakeSubscribeTransport(); + var ing = NewIngestor(subscribe: subscribe, hostId: ""); + + await ing.EstablishAsync(TestContext.Current.CancellationToken); + + subscribe.Filters.Select(f => f.Topic).ShouldBe(["spBv1.0/Plant1/#"]); + } + + /// + /// There is exactly one filter and it covers every tag, so a broker that refuses it leaves the + /// whole driver deaf — the one outcome that must be a visible fault rather than a healthy-looking + /// driver receiving nothing. + /// + [Fact] + public async Task EstablishAsync_GroupFilterRefused_Throws() + { + var subscribe = new FakeSubscribeTransport { Reject = _ => true }; + var ing = NewIngestor(subscribe: subscribe); + + await Should.ThrowAsync( + () => ing.EstablishAsync(TestContext.Current.CancellationToken)); + } + + /// A refused STATE filter costs an observability signal, not the data plane. + [Fact] + public async Task EstablishAsync_StateFilterRefused_DoesNotThrow() + { + var subscribe = new FakeSubscribeTransport { Reject = t => t.Contains("STATE", StringComparison.Ordinal) }; + var ing = NewIngestor(subscribe: subscribe, hostId: "otopcua-host-1"); + + await ing.EstablishAsync(TestContext.Current.CancellationToken); + } + + /// + /// Across an outage of unknown length an edge node may have restarted and rebound its aliases; a + /// DDATA resolved against the pre-outage table would route a value to whichever tag used to own + /// that alias. + /// + [Fact] + public async Task OnReconnected_ClearsBirths_ResubscribesAndRequestsRebirth() + { + var subscribe = new FakeSubscribeTransport(); + var publish = new RecordingPublishTransport(); + var ing = NewIngestor(publish, subscribe); + await ing.SubscribeAsync([TempPath], TimeSpan.Zero, TestContext.Current.CancellationToken); + + FeedBirths(ing); + ing.Births.Count.ShouldBeGreaterThan(0); + publish.Topics.Clear(); + subscribe.Filters.Clear(); + + await ing.OnReconnectedAsync(TestContext.Current.CancellationToken); + await ing.DrainPendingRebirthsAsync(); + + ing.Births.Count.ShouldBe(0); + subscribe.Filters.Select(f => f.Topic).ShouldContain("spBv1.0/Plant1/#"); + publish.Topics.ShouldContain("spBv1.0/Plant1/NCMD/EdgeA"); + } + + /// + /// Reconnect is a driver-side event, not a statement about the plant: last-known values remain + /// the best available answer until the rebirth lands. Only a death stales a tag. + /// + [Fact] + public async Task OnReconnected_DoesNotStaleValues() + { + var ing = await NewSubscribedIngestorAsync(new RecordingPublishTransport(), new FakeSubscribeTransport()); + FeedBirths(ing); + + await ing.OnReconnectedAsync(TestContext.Current.CancellationToken); + + ing.Values.Read(TempPath).StatusCode.ShouldBe(Good); + } + + /// After a reconnect the driver has no birth, so the first DDATA must ask for one. + [Fact] + public async Task AfterReconnect_DataResolvesAgainstNothing_AndRequestsRebirth() + { + var publish = new RecordingPublishTransport(); + var ing = await NewSubscribedIngestorAsync(publish, new FakeSubscribeTransport(), rebirthDebounce: TimeSpan.Zero); + FeedBirths(ing); + + await ing.OnReconnectedAsync(TestContext.Current.CancellationToken); + publish.Topics.Clear(); + + var fired = false; + ing.OnDataChange += (_, _) => fired = true; + ing.Dispatch(Topic(SparkplugMessageType.DDATA, Device), Ddata(seq: 2, (TempAlias, TahuDataType.Float, 1f))); + await ing.DrainPendingRebirthsAsync(); + + fired.ShouldBeFalse(); + publish.Topics.ShouldContain("spBv1.0/Plant1/NCMD/EdgeA"); + } + + // --------------------------------------------------------------------------------- + // STATE (receive only — this driver never publishes one; see the ingestor's remarks) + // --------------------------------------------------------------------------------- + + [Fact] + public async Task State_V30JsonPayload_IsObserved() + { + var ing = await NewSubscribedIngestorAsync(); + + ing.HandleMessage("spBv1.0/STATE/otopcua-host-1", """{"online":true,"timestamp":1721822400000}"""u8, retained: true); + + ing.HostState!.HostId.ShouldBe("otopcua-host-1"); + ing.HostState!.Online.ShouldBeTrue(); + } + + [Fact] + public async Task State_LegacyTextPayload_IsObserved() + { + var ing = await NewSubscribedIngestorAsync(); + + ing.HandleMessage("STATE/otopcua-host-1", "OFFLINE"u8, retained: true); + + ing.HostState!.Online.ShouldBeFalse(); + } + + [Fact] + public async Task State_UnrecognisedPayload_IsIgnored() + { + var ing = await NewSubscribedIngestorAsync(); + + ing.HandleMessage("spBv1.0/STATE/otopcua-host-1", "maybe?"u8, retained: true); + + ing.HostState.ShouldBeNull(); + } + + // --------------------------------------------------------------------------------- + // Task 22 seam + // --------------------------------------------------------------------------------- + + /// + /// The rediscovery seam Task 22 wires to IRediscoverable.OnRediscoveryNeeded. Deliberately + /// unwired in MqttDriver today — this asserts the event exists and carries the scope + + /// metric set that Task 22 needs to decide whether the tree changed. + /// + [Fact] + public async Task BirthObserved_FiresWithTheScopeAndMetricNames() + { + var ing = await NewSubscribedIngestorAsync(); + var observed = new List(); + ing.BirthObserved += (_, e) => observed.Add(e); + + FeedBirths(ing); + + observed.Count.ShouldBe(2); + observed[0].Scope.ShouldBe(new SparkplugScope(Group, Node, null)); + observed[1].Scope.ShouldBe(new SparkplugScope(Group, Node, Device)); + observed[1].MetricNames.ShouldContain("Temperature"); + } + + // ================================================================================= + // Fixtures + // ================================================================================= + + private static RawTagEntry Tag(string rawPath, string tagConfig) => new(rawPath, tagConfig, WriteIdempotent: false); + + private static string Blob(string group, string node, string? device, string metric, string? dataType) + { + var deviceKey = device is null ? "" : $"""{'"'}deviceId{'"'}:"{device}","""; + var typeKey = dataType is null ? "" : $""","dataType":"{dataType}" """; + return $$"""{"groupId":"{{group}}","edgeNodeId":"{{node}}",{{deviceKey}}"metricName":"{{metric}}"{{typeKey}}}"""; + } + + /// The standard authored set: three device metrics plus one node-level metric. + private static RawTagEntry[] DefaultTags() => + [ + Tag(TempPath, Blob(Group, Node, Device, "Temperature", "Float32")), + Tag(PressPath, Blob(Group, Node, Device, "Pressure", "Float32")), + Tag(CountPath, Blob(Group, Node, Device, "Count", "Int32")), + Tag(NodeUptimePath, Blob(Group, Node, null, "Uptime", "Int64")), + ]; + + private static SparkplugIngestor NewIngestor( + RecordingPublishTransport? publish = null, + FakeSubscribeTransport? subscribe = null, + string hostId = "otopcua-host-1", + bool requestRebirthOnGap = true, + TimeSpan? rebirthDebounce = null, + int maxPayloadBytes = MqttSubscriptionManager.DefaultMaxPayloadBytes, + RawTagEntry[]? tags = null, + RawTagEntry[]? extraTags = null) + { + var options = new MqttDriverOptions + { + Mode = MqttMode.SparkplugB, + Sparkplug = new MqttSparkplugOptions + { + GroupId = Group, + HostId = hostId, + RequestRebirthOnGap = requestRebirthOnGap, + }, + }; + + var ingestor = new SparkplugIngestor( + options, + "mqtt-spb-1", + subscribe, + publish, + logger: null, + cache: null, + maxPayloadBytes: maxPayloadBytes, + // Zero by default so a test asserting "these two triggers each produce an NCMD" is not + // silently satisfied by the production debounce; the debounce itself has its own tests. + rebirthDebounce: rebirthDebounce ?? TimeSpan.Zero); + + ingestor.Register([.. tags ?? DefaultTags(), .. extraTags ?? []]); + return ingestor; + } + + /// An ingestor with every default tag registered AND subscribed, so notifications fire. + private static async Task NewSubscribedIngestorAsync( + RecordingPublishTransport? publish = null, + FakeSubscribeTransport? subscribe = null, + bool requestRebirthOnGap = true, + TimeSpan? rebirthDebounce = null, + RawTagEntry[]? extraTags = null) + { + var ingestor = NewIngestor( + publish, + subscribe, + requestRebirthOnGap: requestRebirthOnGap, + rebirthDebounce: rebirthDebounce, + extraTags: extraTags); + + await ingestor.SubscribeAsync( + [.. ingestor.AuthoredRawPaths], + TimeSpan.Zero, + TestContext.Current.CancellationToken); + + return ingestor; + } + + /// Feeds the standard NBIRTH (seq 0, bdSeq 1) + DBIRTH (seq 1) pair. + private static void FeedBirths(SparkplugIngestor ing) + { + ing.Dispatch(Topic(SparkplugMessageType.NBIRTH), Nbirth(seq: 0, bdSeq: 1)); + ing.Dispatch(Topic(SparkplugMessageType.DBIRTH, Device), Dbirth(seq: 1)); + } + + private static SparkplugTopic Topic(SparkplugMessageType type, string? device = null) => + new(type, Group, Node, device, null); + + private static SparkplugPayload Decode(Payload payload) => SparkplugCodec.Decode(payload.ToByteArray()); + + /// An NBIRTH: the bdSeq session token plus the node's own metric catalog. + private static SparkplugPayload Nbirth(ulong seq, ulong bdSeq) + { + 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(BirthMetric("Uptime", 7UL, TahuDataType.Int64, 0L)); + return Decode(payload); + } + + /// The standard DBIRTH: Temperature@5 (Float), Pressure@6 (Float), Count@9 (Int32, -42). + private static SparkplugPayload Dbirth(ulong seq) => DbirthCustom( + seq, + ("Temperature", TempAlias, TahuDataType.Float, 0f), + ("Pressure", PressAlias, TahuDataType.Float, 0f), + ("Count", CountAlias, TahuDataType.Int32, -42)); + + private static SparkplugPayload DbirthCustom( + ulong seq, + params (string Name, ulong Alias, TahuDataType Type, object Value)[] metrics) + { + var payload = new Payload { Seq = seq, Timestamp = 1721822400000UL }; + foreach (var (name, alias, type, value) in metrics) + { + payload.Metrics.Add(BirthMetric(name, alias, type, value)); + } + + return Decode(payload); + } + + /// A DDATA: metrics carry an alias and NOTHING else — the real post-birth wire shape. + private static SparkplugPayload Ddata(ulong seq, params (ulong Alias, TahuDataType Type, object Value)[] metrics) + => DataPayload(seq, metrics); + + /// An NDATA — identical shape to ; the topic is what makes it node-scoped. + private static SparkplugPayload Ndata(ulong seq, params (ulong Alias, TahuDataType Type, object Value)[] metrics) + => DataPayload(seq, metrics); + + private static SparkplugPayload DataPayload( + ulong seq, + (ulong Alias, TahuDataType Type, object Value)[] metrics) + { + var payload = new Payload { Seq = seq, Timestamp = 1721822405000UL }; + foreach (var (alias, type, value) in metrics) + { + // Deliberately NO Name and NO Datatype: that is what a Sparkplug DATA metric looks like + // after a birth, and it is the case an "absent means empty string" consumer gets wrong. + var metric = new Payload.Types.Metric { Alias = alias }; + SetWireValue(metric, type, value); + payload.Metrics.Add(metric); + } + + return Decode(payload); + } + + /// A DDEATH — published by the edge node, and therefore sequenced. + private static SparkplugPayload Ddeath(ulong seq) => + Decode(new Payload { Seq = seq, Timestamp = 1721822405000UL }); + + /// An NDEATH — the broker-published Last Will: a bdSeq and NO seq at all. + private static SparkplugPayload Ndeath(ulong bdSeq) + { + var payload = new Payload { Timestamp = 1721822405000UL }; + payload.Metrics.Add(new Payload.Types.Metric + { + Name = "bdSeq", + Datatype = (uint)TahuDataType.Uint64, + LongValue = bdSeq, + }); + return Decode(payload); + } + + private static Payload.Types.Metric BirthMetric(string name, ulong alias, TahuDataType type, object value) + { + var metric = new Payload.Types.Metric { Name = name, Alias = alias, Datatype = (uint)type }; + SetWireValue(metric, type, value); + return metric; + } + + /// + /// Writes a value into the oneof field Sparkplug actually carries it in — including the + /// two's-complement-in-an-unsigned-field encoding for the signed integer types, which is the + /// whole reason exists. + /// + private static void SetWireValue(Payload.Types.Metric metric, TahuDataType type, object value) + { + switch (type) + { + case TahuDataType.Boolean: + metric.BooleanValue = (bool)value; + break; + case TahuDataType.Float: + metric.FloatValue = (float)value; + break; + case TahuDataType.Double: + metric.DoubleValue = (double)value; + break; + case TahuDataType.Int8: + case TahuDataType.Int16: + case TahuDataType.Int32: + metric.IntValue = unchecked((uint)Convert.ToInt32(value)); + break; + case TahuDataType.Uint8: + case TahuDataType.Uint16: + case TahuDataType.Uint32: + metric.IntValue = Convert.ToUInt32(value); + break; + case TahuDataType.Int64: + metric.LongValue = unchecked((ulong)Convert.ToInt64(value)); + break; + case TahuDataType.Uint64: + case TahuDataType.DateTime: + metric.LongValue = Convert.ToUInt64(value); + break; + case TahuDataType.String: + case TahuDataType.Text: + case TahuDataType.Uuid: + metric.StringValue = (string)value; + break; + default: + throw new NotSupportedException($"Test helper does not encode {type}."); + } + } + + /// Records every NCMD so the rebirth policy is assertable without a broker. + private sealed class RecordingPublishTransport : IMqttPublishTransport + { + public List Topics { get; } = []; + + public List Payloads { get; } = []; + + public Task PublishAsync(string topic, byte[] payload, int qos, bool retain, CancellationToken cancellationToken) + { + lock (Topics) + { + Topics.Add(topic); + Payloads.Add(payload); + } + + // A retained NCMD replays to the edge node on every future subscribe, driving a permanent + // rebirth loop from one stale command. Assert it here so every rebirth path is covered. + retain.ShouldBeFalse(); + return Task.CompletedTask; + } + } + + /// Records every SUBSCRIBE so the group/STATE filters are assertable without a broker. + private sealed class FakeSubscribeTransport : IMqttSubscribeTransport + { + public List Filters { get; } = []; + + /// Topics for which the fake SUBACK reports a rejection. + public Func Reject { get; set; } = _ => false; + + public Task> SubscribeAsync( + IReadOnlyList filters, + CancellationToken cancellationToken) + { + Filters.AddRange(filters); + IReadOnlyList outcomes = + [ + .. filters.Select(f => new MqttSubscribeOutcome( + f.Topic, + !Reject(f.Topic), + Reject(f.Topic) ? "NotAuthorized" : "GrantedQoS1")), + ]; + return Task.FromResult(outcomes); + } + } +}