From c74ccc55b5f3d9cfb3d32d89dff4370f9ad9f889 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Fri, 31 Jul 2026 08:14:04 -0400 Subject: [PATCH] feat(mqtt): flatten Sparkplug Template/DataSet metrics to sibling tags (#515) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `SparkplugCodec` handled scalars; `DataSet`, `Template` and `PropertySet` were skipped with a warning. They now flatten to addressable tags (owner decision, recorded on the issue): Template → sibling metrics, `/`-joined, recursively: Motor → Motor/Speed, Motor/Stator/Temp DataSet → one metric per cell, {column}/{rowIndex}: Recipe → Recipe/Step/0, Recipe/Target/1 PropertySet → node metadata, never tags — it describes a metric rather than being one Flattening was chosen over OPC UA Structures because flattened scalars are consumable by any client with no type dictionary, and the result mirrors how the `/raw` tree works. **The row index is always present, including on row 0.** Emitting a bare column name for a single-row table and an indexed one otherwise would make a tag's name depend on how many rows the plant published — a table growing from one row to two would rename every tag already authored against it. **A DATA metric carries the parent's alias and no name**, so member names cannot be composed until the alias table has resolved the parent. Expansion therefore happens in `SparkplugIngestor.PublishStructuredMembers` — after resolution, at the single funnel both the DATA and birth-value paths already pass through — not at decode time. The alias table installs a binding per member so browse enumerates them and a DDATA member that omits its datatype falls back to the birth's (guessing from the CLR type would read a signed Int32 of -42 out of its unsigned field as 4294967254). **Skip-and-warn survives, and refusal is whole** — the explicit requirement on the issue. A template definition, an unnamed member, an extension value, a column/type mismatch, a short row, or a payload past MaxDepth(5)/MaxMembers(512) refuses the ENTIRE metric. A silently half-flattened UDT looks exactly like a complete one in the picker. The browse picker offers the members and deliberately NOT the structured parent, which has no value and no servable datatype. The rediscovery signature likewise tracks members rather than the parent, or a UDT gaining a member would leave the signature unmoved. Member paths contain `/`, which #509 already solved: the slashed name travels intact as `metricName` while the derived tag Name is coerced to a legal RawPath segment. Live-gated 3/3 against the Mosquitto fixture with the project-owned simulator extended to publish a nested Template and a 2x2 DataSet. Falsified at three points: dropping the member bindings, not expanding the parent, and refusing to flatten templates each turn exactly the expected tests red. Also fixes a regression from the #511 commit in the same branch: `AddOtOpcUaDriverFactories` resolved `IDriverPrimaryRoleSource` with `GetRequiredService` but only `AddOtOpcUaRuntime` registered it, so a graph wiring driver factories without the full runtime failed at resolution. Two Host integration suites do exactly that and caught it; both extensions now TryAdd the same concrete `RedundancyRoleView`, so the registration is order-independent and the fail-loud resolve is kept. Claude-Session: https://claude.ai/code/session_015p7wGqy3YpZNCpDzTpGMKo --- docs/drivers/Mqtt.md | 68 +++- .../MqttBrowseSession.cs | 105 ++++- .../Sparkplug/SparkplugCodec.cs | 114 ++++-- .../Sparkplug/SparkplugFlattener.cs | 368 ++++++++++++++++++ .../Sparkplug/AliasTable.cs | 30 ++ .../Sparkplug/SparkplugIngestor.cs | 101 ++++- .../Drivers/DriverFactoryBootstrap.cs | 14 + .../SparkplugSimulator/Program.cs | 61 ++- .../SparkplugSimulator/SparkplugEdgeNode.cs | 86 ++++ .../SparkplugStructuredLiveTests.cs | 332 ++++++++++++++++ .../MqttBrowseSessionTests.cs | 25 ++ .../SparkplugCodecTests.cs | 67 +++- .../SparkplugFlattenerTests.cs | 294 ++++++++++++++ .../SparkplugGoldenPayloads.cs | 5 +- .../SparkplugStructuredIngestTests.cs | 283 ++++++++++++++ 15 files changed, 1900 insertions(+), 53 deletions(-) create mode 100644 src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/Sparkplug/SparkplugFlattener.cs create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/SparkplugStructuredLiveTests.cs create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/SparkplugFlattenerTests.cs create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/SparkplugStructuredIngestTests.cs diff --git a/docs/drivers/Mqtt.md b/docs/drivers/Mqtt.md index 76be2763..c01aeb83 100644 --- a/docs/drivers/Mqtt.md +++ b/docs/drivers/Mqtt.md @@ -204,7 +204,7 @@ survive a reconnect; a message fans out to *every* matching tag, and unauthored | `groupId` | string | — | **Required.** Must match the driver's `Sparkplug.GroupId` to ever receive a message | | `edgeNodeId` | string | — | **Required.** | | `deviceId` | string | *absent* | **Omit** for a metric published by the edge node itself (NBIRTH/NDATA). Absent is meaningfully different from `""` | -| `metricName` | string | — | **Required.** The tag's stable binding key across rebirths. **May contain `/`** — `Node Control/Rebirth` and `Properties/Serial Number` are canonical Sparkplug names, and the whole name is one value, never split | +| `metricName` | string | — | **Required.** The tag's stable binding key across rebirths. **May contain `/`** — `Node Control/Rebirth` and `Properties/Serial Number` are canonical Sparkplug names, and the whole name is one value, never split. Also how a flattened structured member is addressed: `Motor/Speed`, `Recipe/Target/0` (see [Structured metrics](#structured-metrics-template--dataset--propertyset)) | | `dataType` | `DriverDataType` | *from the birth* | **Optional override.** Absent ⇒ take whatever type the birth certificate declared. The AdminUI's "(from birth certificate)" option removes the key entirely | | `isHistorized` / `historianTagname` | — | — | Server-side historization, unchanged from Plain | @@ -222,6 +222,11 @@ bind. `metricName` is deliberately **exempt**: it is not a topic segment. // A node-level metric — note deviceId is ABSENT, not blank. {"groupId":"OtOpcUaSim","edgeNodeId":"EdgeA","metricName":"Temperature"} + +// A flattened Template member, and a flattened DataSet cell. `Motor` and `Recipe` are what the wire +// carries; nothing on the wire is ever called by these names. +{"groupId":"OtOpcUaSim","edgeNodeId":"EdgeA","deviceId":"Filler1","metricName":"Motor/Stator/Temp"} +{"groupId":"OtOpcUaSim","edgeNodeId":"EdgeA","deviceId":"Filler1","metricName":"Recipe/Target/1"} ``` ## Sparkplug B Ingest @@ -240,9 +245,11 @@ One subscription (`spBv1.0/{GroupId}/#`) feeds a per-edge-node state machine - **Death → STALE.** An NDEATH whose `bdSeq` matches the live session (the bdSeq tie is what stops a stale death from killing a fresh session) marks every metric under that edge node `Bad` / `BadNoCommunication`. A DDEATH does the same for one device. The next birth restores them. -- **Unsupported datatypes are skipped with a warning, never guessed**: `DataSet`, `Template`, - `PropertySet`, `PropertySetList`, `Unknown`. `Int8`→`Int16` and `UInt8`→`UInt16` widen; `*Array` - variants map to the element type with the array bit set. +- **Structured metrics flatten to sibling tags** — `Template` and `DataSet`; see + [Structured metrics](#structured-metrics-template--dataset--propertyset). +- **Unsupported datatypes are skipped with a warning, never guessed**: `PropertySetList`, `Unknown`, + extension values, and any `Template`/`DataSet` that cannot be flattened honestly. `Int8`→`Int16` and + `UInt8`→`UInt16` widen; `*Array` variants map to the element type with the array bit set. - **Ingest state is reset at every session/authoring seam** — a reconnect or a `ReinitializeAsync` must not carry a previous session's alias table into a new one. @@ -260,6 +267,53 @@ re-births freely — on its own schedule, on every reconnect, and once per rebir > through the `/raw` browse-commit flow, and grafting nodes at runtime would materialise ones nobody > approved and no deployment artifact records. +## Structured metrics (`Template` / `DataSet` / `PropertySet`) + +Sparkplug metrics are not all scalars. `SparkplugFlattener` +([#515](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/515)) turns the structured ones into +addressable tags: + +| Metric kind | Mapping | Example | +|---|---|---| +| `Template` (UDT instance) | members become **sibling metrics**, `/`-joined, recursively | `Motor` → `Motor/Speed`, `Motor/Stator/Temp` | +| `DataSet` | **one metric per cell**, `{column}/{rowIndex}` | `Recipe` → `Recipe/Step/0`, `Recipe/Target/1` | +| `PropertySet` (a metric's `properties`) | node **metadata**, never tags | `engUnit`, `Range/Low` | + +Flattened members are authored exactly like any other Sparkplug metric — the composed name goes in +`metricName`. The browse picker offers them directly (the structured **parent** is deliberately *not* +offered: it has no value and no servable datatype, so a tag committed against it could never read). + +**Why flattening rather than OPC UA Structures.** Owner decision, 2026-07-31. Structures are truer to +Sparkplug's model but much harder to consume — flattened scalars are readable by any OPC UA client +with no type dictionary, and the result mirrors how the `/raw` tree already works. + +### Things worth knowing before you touch this + +- **The row index is always present, including on row 0.** `Recipe/Target/0`, never `Recipe/Target`. + Emitting a bare column name for a single-row table and an indexed one otherwise would make a tag's + name depend on how many rows the plant published — so a table growing from one row to two would + rename every tag already authored against it. +- **A DATA metric carries the parent's alias and no name**, so member names cannot be composed until + the alias table has resolved the parent. Flattening at decode time alone is not enough; the + expansion happens in `SparkplugIngestor.PublishStructuredMembers`, after resolution. +- **Member datatype falls back to the birth's.** A DDATA template may omit its members' datatypes. + Guessing from the CLR type of the wire value would read a signed `Int32` of −42 out of its unsigned + field as 4294967254 — plausible, `Good`, and wrong. +- **Skip-and-warn survives, and refusal is whole.** A template **definition** (`is_definition`), an + unnamed member, an extension value, a column/type count mismatch, a short row, or a payload past + `MaxDepth` (5) / `MaxMembers` (512) refuses the **entire** metric rather than publishing part of it. + A silently half-flattened UDT looks exactly like a complete one in the picker. +- **Member paths contain `/`**, which `RawBrowseCommitMapper.ToRawSegment` + ([#509](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/509)) already handles: the slashed name + travels intact as `metricName` while the derived tag **Name** is coerced to a legal RawPath segment, + with a collision guard for two members that would sanitise onto one name. +- **The rediscovery signature tracks members, not the parent.** A UDT gaining or losing a member leaves + the parent metric's name identical, so a gate built on raw wire names would never move. + +`PropertySet` is projected as metadata rather than as sibling tags because it *describes* a metric +(units, ranges, quality hints) — surfacing it as tags would invent plant signals out of documentation. +`PropertySetList` and property extension values are dropped rather than stringified. + ## Primary host (`actAsPrimaryHost`) A Sparkplug **Primary Host Application** is the thing edge nodes watch to decide whether anyone is @@ -515,7 +569,7 @@ identical words. See [TestConnectProbes.md](TestConnectProbes.md). reconnect/rebuild branches, CONNACK classification, and the Sparkplug half — golden-payload decode, topic parse/format, datatype map, alias/birth-cache rebuild, sequence + bdSeq-death ties, the rediscovery change gate, and the browse session's "publishes nothing" assertion. -- **Live (env-gated)** — `tests/Drivers/…Driver.Mqtt.IntegrationTests` (**26 tests**) against the +- **Live (env-gated)** — `tests/Drivers/…Driver.Mqtt.IntegrationTests` (**29 tests**) against the Mosquitto broker + the C# Sparkplug edge-node simulator; skips cleanly when `MQTT_FIXTURE_ENDPOINT` is unset. See [Mqtt-Test-Fixture.md](Mqtt-Test-Fixture.md) and `infra/README.md` §3.
The primary-host legs (`MqttPrimaryHostLiveTests`) include the one the ordinary lifecycle cannot @@ -541,7 +595,8 @@ identical words. See [TestConnectProbes.md](TestConnectProbes.md). Every row is tracked. None is a silent failure — each either refuses in words or logs. **Closed 2026-07-31:** write-through ([#508](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/508), live-gated), primary-host STATE publishing -([#511](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/511), live-gated), the slashed-metric browse-commit +([#511](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/511), live-gated), structured +`Template`/`DataSet`/`PropertySet` metrics ([#515](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/515), live-gated), the slashed-metric browse-commit ([#509](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/509)), the stale rebirth affordance ([#510](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/510)), the `.Browser` layering exception ([#512](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/512)), the stray Sparkplug `payloadFormat` ([#513](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/513)) and rebirth on a not-yet-birthed plant ([#514](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/514)). @@ -549,4 +604,3 @@ not-yet-birthed plant ([#514](https://gitea.dohertylan.com/dohertj2/lmxopcua/iss | Gap | Issue | Effect | Notes | |---|:---:|---|---| | **Rediscovery is advisory, not a rebuild** | [#507](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/507) (closed, superseded) | A DBIRTH introducing a metric does not change the served OPC UA tree; an operator must re-browse the device and commit | **No longer "inert".** `DriverInstanceActor` now subscribes to `OnRediscoveryNeeded` and the raise rides the driver-health snapshot to the AdminUI `/hosts` page as a **re-browse chip**. It deliberately does not inject nodes: v3 authors raw tags through the `/raw` browse-commit flow, and grafting nodes at runtime would materialise ones nobody approved and no deployment artifact records. #507's injection path was deleted rather than revived | -| **`DataSet` / `Template` / `PropertySet` metrics** | [#515](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/515) | Skipped with a warning | v1-unsupported; never coerced into a guessed type | diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser/MqttBrowseSession.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser/MqttBrowseSession.cs index cb240b42..2e37d579 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser/MqttBrowseSession.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser/MqttBrowseSession.cs @@ -585,20 +585,77 @@ internal sealed class MqttBrowseSession : IRebirthCapableBrowseSession continue; } - var metricPath = ownerPath + MetricSeparator + name; - if (metricPath.Length > MaxTopicChars) + // Gitea #515: a Template/DataSet metric contributes its flattened MEMBERS as leaves, never + // itself. The parent has no value and no servable datatype, so a leaf for it would offer + // the operator an address that can only ever commit a tag which never reads. + if (metric.ValueKind == SparkplugValueKind.Structured) { - Volatile.Write(ref _oversized, 1); + foreach (var member in metric.MemberList) + { + var memberName = $"{name}{SparkplugFlattener.MemberSeparator}{member.Path}"; + if (!IsUsableLabel(memberName)) + { + Volatile.Write(ref _oversized, 1); + continue; + } + + if (!AddMetricLeaf(owner, ownerPath, ownerScope, memberName, member.DataType, parsed.Type, alias: null)) + { + return; // node cap tripped + } + } + continue; } - var node = GetOrCreate(owner.Children, metricPath, name, BrowseNodeKind.Leaf, ownerScope); - if (node is null) return; // node cap tripped - - node.RecordMetric(new SparkplugMetricInfo(name, metric.DataType, parsed.Type, metric.Alias)); + if (!AddMetricLeaf(owner, ownerPath, ownerScope, name, metric.DataType, parsed.Type, metric.Alias)) + { + return; // node cap tripped + } } } + /// + /// Places one metric leaf under . + /// + /// The owning edge-node / device folder. + /// That folder's path. + /// The scope stamped on the leaf. + /// + /// The metric's Sparkplug name — for a flattened structured member, the composed + /// {parent}/{memberPath}. It travels intact: it is the binding key, and only the + /// display name is later coerced to a legal RawPath segment (Gitea #509). + /// + /// The declared datatype, or null. + /// The birth message type this leaf was seen in. + /// The metric's alias, or null — flattened members carry none of their own. + /// when the node cap tripped and the caller must stop. + private bool AddMetricLeaf( + TopicNode owner, + string ownerPath, + SparkplugScope ownerScope, + string metricName, + Org.Eclipse.Tahu.Protobuf.DataType? dataType, + SparkplugMessageType messageType, + ulong? alias) + { + var metricPath = ownerPath + MetricSeparator + metricName; + if (metricPath.Length > MaxTopicChars) + { + Volatile.Write(ref _oversized, 1); + return true; + } + + var node = GetOrCreate(owner.Children, metricPath, metricName, BrowseNodeKind.Leaf, ownerScope); + if (node is null) + { + return false; + } + + node.RecordMetric(new SparkplugMetricInfo(metricName, dataType, messageType, alias)); + return true; + } + /// /// Finds or creates one tree node under , honouring the node cap. /// @@ -698,6 +755,40 @@ internal sealed class MqttBrowseSession : IRebirthCapableBrowseSession ObserveBirth(groupId, edgeNodeId, device, metric); } + /// + /// Test seam feeding a birth whose metric is a structured Template (Gitea #515). + /// + /// The Sparkplug group id. + /// The Sparkplug edge-node id. + /// The device id, or for an NBIRTH. + /// The template metric's own name. + /// The template's members, as (name, datatype). + internal void ObserveTemplateBirthForTest( + string groupId, + string edgeNodeId, + string? device, + string metricName, + params (string Name, SparkplugDataType DataType)[] members) + { + var template = new Payload.Types.Template { Version = "1", IsDefinition = false }; + foreach (var (name, dataType) in members) + { + var member = new Payload.Types.Metric { Name = name, Datatype = (uint)dataType }; + + // A value is required: a member carrying only a name and a datatype is indistinguishable + // from a template DEFINITION's member, which the flattener deliberately refuses. + member.DoubleValue = 0; + template.Metrics.Add(member); + } + + ObserveBirth(groupId, edgeNodeId, device, new Payload.Types.Metric + { + Name = metricName, + Datatype = (uint)SparkplugDataType.Template, + TemplateValue = template, + }); + } + /// Test seam feeding a birth certificate that declares no metrics at all. /// The Sparkplug group id. /// The Sparkplug edge-node id. diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/Sparkplug/SparkplugCodec.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/Sparkplug/SparkplugCodec.cs index b428ee19..309aa3c1 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/Sparkplug/SparkplugCodec.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/Sparkplug/SparkplugCodec.cs @@ -45,10 +45,19 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Sparkplug; /// wire, not of the tag. /// /// -/// DataSet and Template metrics are out of scope for v1. They decode to +/// Template and DataSet metrics decode to +/// with their members flattened into +/// (Gitea #515) — a template member Speed of metric +/// Motor becomes the addressable Motor/Speed. See +/// for the mapping and its bounds. Anything that cannot be flattened honestly stays /// with a null value — deliberately visible, so a -/// consumer can warn about a metric it cannot serve instead of silently treating it as one -/// that was never published. +/// consumer warns about a metric it cannot serve instead of silently treating it as one that was +/// never published, and never coerces it into a guessed scalar. +/// +/// +/// A metric's PropertySet is projected to +/// as node metadata, never as sibling values: a PropertySet describes a metric (units, ranges, +/// quality hints); it is not something the metric is. /// /// public static class SparkplugCodec @@ -161,7 +170,7 @@ public static class SparkplugCodec /// The projection. private static SparkplugMetric ProjectMetric(Payload.Types.Metric metric) { - var (kind, value) = ProjectValue(metric); + var (kind, value, members) = ProjectValue(metric); return new SparkplugMetric( Name: metric.HasName ? metric.Name : null, @@ -173,7 +182,22 @@ public static class SparkplugCodec DataType: metric.HasDatatype ? (TahuDataType)metric.Datatype : null, TimestampMs: metric.HasTimestamp ? metric.Timestamp : null, ValueKind: kind, - Value: value); + Value: value, + Members: members, + Properties: SparkplugFlattener.ProjectProperties(metric.Properties)); + } + + /// + /// Resolves a template member's value oneof. Internal to the assembly: shared with + /// , which walks nested templates and needs the identical + /// projection so a member and a top-level metric of the same type never decode differently. + /// + /// The template member, itself a generated Metric. + /// The value kind and the projected raw value. + internal static (SparkplugValueKind Kind, object? Value) ProjectMemberValue(Payload.Types.Metric metric) + { + var (kind, value, _) = ProjectValue(metric); + return (kind, value); } /// Resolves a metric's value oneof to a kind plus a raw CLR value. @@ -184,34 +208,52 @@ public static class SparkplugCodec /// that some datatypes have no spare sentinel, so a publisher that sets it means "null" even if /// a value field is also populated. /// - private static (SparkplugValueKind Kind, object? Value) ProjectValue(Payload.Types.Metric metric) + private static (SparkplugValueKind Kind, object? Value, IReadOnlyList Members) ProjectValue( + Payload.Types.Metric metric) { if (metric.HasIsNull && metric.IsNull) { - return (SparkplugValueKind.Null, null); + return (SparkplugValueKind.Null, null, []); + } + + // Gitea #515: structured values flatten to addressable members. A refusal to flatten falls + // through to Unsupported — the pre-#515 behaviour — rather than to a guessed scalar. + switch (metric.ValueCase) + { + case Payload.Types.Metric.ValueOneofCase.TemplateValue: + return SparkplugFlattener.TryFlattenTemplate(metric.TemplateValue, out var templateMembers) + ? (SparkplugValueKind.Structured, null, templateMembers) + : (SparkplugValueKind.Unsupported, null, []); + + case Payload.Types.Metric.ValueOneofCase.DatasetValue: + return SparkplugFlattener.TryFlattenDataSet(metric.DatasetValue, out var cells) + ? (SparkplugValueKind.Structured, null, cells) + : (SparkplugValueKind.Unsupported, null, []); } return metric.ValueCase switch { - Payload.Types.Metric.ValueOneofCase.IntValue => (SparkplugValueKind.Scalar, metric.IntValue), - Payload.Types.Metric.ValueOneofCase.LongValue => (SparkplugValueKind.Scalar, metric.LongValue), - Payload.Types.Metric.ValueOneofCase.FloatValue => (SparkplugValueKind.Scalar, metric.FloatValue), - Payload.Types.Metric.ValueOneofCase.DoubleValue => (SparkplugValueKind.Scalar, metric.DoubleValue), - Payload.Types.Metric.ValueOneofCase.BooleanValue => (SparkplugValueKind.Scalar, metric.BooleanValue), - Payload.Types.Metric.ValueOneofCase.StringValue => (SparkplugValueKind.Scalar, metric.StringValue), + Payload.Types.Metric.ValueOneofCase.IntValue => (SparkplugValueKind.Scalar, metric.IntValue, Empty), + Payload.Types.Metric.ValueOneofCase.LongValue => (SparkplugValueKind.Scalar, metric.LongValue, Empty), + Payload.Types.Metric.ValueOneofCase.FloatValue => (SparkplugValueKind.Scalar, metric.FloatValue, Empty), + Payload.Types.Metric.ValueOneofCase.DoubleValue => (SparkplugValueKind.Scalar, metric.DoubleValue, Empty), + Payload.Types.Metric.ValueOneofCase.BooleanValue => (SparkplugValueKind.Scalar, metric.BooleanValue, Empty), + Payload.Types.Metric.ValueOneofCase.StringValue => (SparkplugValueKind.Scalar, metric.StringValue, Empty), // Copied out of the ByteString: the projection must outlive the parsed message. Payload.Types.Metric.ValueOneofCase.BytesValue => - (SparkplugValueKind.Scalar, metric.BytesValue.ToByteArray()), + (SparkplugValueKind.Scalar, metric.BytesValue.ToByteArray(), Empty), - // v1 scope boundary — decoded as a visible refusal, not as an exception or a silent drop. - Payload.Types.Metric.ValueOneofCase.DatasetValue => (SparkplugValueKind.Unsupported, null), - Payload.Types.Metric.ValueOneofCase.TemplateValue => (SparkplugValueKind.Unsupported, null), - Payload.Types.Metric.ValueOneofCase.ExtensionValue => (SparkplugValueKind.Unsupported, null), + // An extension value is a third-party escape hatch with no defined semantics — there is + // nothing honest to flatten it into. + Payload.Types.Metric.ValueOneofCase.ExtensionValue => (SparkplugValueKind.Unsupported, null, Empty), - _ => (SparkplugValueKind.Absent, null), + _ => (SparkplugValueKind.Absent, null, Empty), }; } + + /// The shared empty member list — every scalar metric would otherwise allocate one. + private static readonly IReadOnlyList Empty = []; } /// @@ -279,13 +321,31 @@ public sealed record SparkplugPayload( /// switch over a dozen arms. /// /// +/// +/// The flattened members of a metric — a +/// Template or DataSet broken into addressable scalars (Gitea #515). Empty for every +/// other kind. Paths are relative to this metric; see . +/// +/// +/// The metric's PropertySet, projected as node metadata (Gitea #515). Empty when the metric +/// carried none. Never a value — a PropertySet describes the metric rather than being one. +/// public readonly record struct SparkplugMetric( string? Name, ulong? Alias, TahuDataType? DataType, ulong? TimestampMs, SparkplugValueKind ValueKind, - object? Value); + object? Value, + IReadOnlyList? Members = null, + IReadOnlyList? Properties = null) +{ + /// The flattened members, never null. + public IReadOnlyList MemberList => Members ?? []; + + /// The projected properties, never null. + public IReadOnlyList PropertyList => Properties ?? []; +} /// What a decoded metric's represents. /// @@ -305,9 +365,17 @@ public enum SparkplugValueKind Scalar, /// - /// A DataSet, Template or extension value — decoded and reported, but not - /// supported in v1. The consumer should warn and skip the metric rather than treat it as - /// missing. + /// An extension value, or a Template/DataSet that could not be flattened + /// honestly (see ). Decoded and reported, but not servable: + /// the consumer warns and skips the metric rather than treating it as missing — and never + /// coerces it into a guessed scalar. /// Unsupported, + + /// + /// A Template or DataSet whose members were flattened into + /// (Gitea #515). is + /// null — the parent has no scalar value of its own; its members carry the values. + /// + Structured, } diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/Sparkplug/SparkplugFlattener.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/Sparkplug/SparkplugFlattener.cs new file mode 100644 index 00000000..7cb81bb8 --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/Sparkplug/SparkplugFlattener.cs @@ -0,0 +1,368 @@ +using Org.Eclipse.Tahu.Protobuf; +using TahuDataType = Org.Eclipse.Tahu.Protobuf.DataType; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Sparkplug; + +/// +/// Flattens Sparkplug's structured metric values — Template (UDT instances) and +/// DataSet — into addressable scalar members, and projects a metric's PropertySet +/// into node metadata (Gitea #515). +/// +/// +/// +/// Flatten to siblings, by owner decision (2026-07-31). The alternative was mapping a +/// Template to an OPC UA Structure/UDT, which is truer to Sparkplug's model but far harder to +/// consume: flattened scalars are readable by any OPC UA client (ScadaBridge included) with no +/// type dictionary, and the result mirrors how the /raw tree already works. A member of +/// template metric Motor called Speed becomes the metric Motor/Speed. +/// +/// +/// Member paths contain /, which is already solved. That interaction is what +/// #515 flagged against #509, and #509 shipped first: +/// RawBrowseCommitMapper.ToRawSegment coerces the tag Name to a legal RawPath +/// segment while carrying the original slashed string through as metricName, plus a +/// collision guard that refuses two members which would sanitise onto one name. So the +/// flattened name travels intact as the binding key and only the display name is coerced. +/// +/// +/// Skip-and-warn survives, and that is a requirement rather than a leftover. Anything +/// this cannot flatten honestly — an extension value, a nested member with no name, a payload +/// past the bounds below — stays and is skipped +/// with a warning. A structured value coerced into a guessed scalar would bind a tag to data +/// that is not what its name says, which is worse than not binding it at all. +/// +/// +/// Bounded, because this walks an attacker-shaped object graph. A Template may contain +/// Templates, so the natural recursion is unbounded in a format anyone on the broker can +/// publish. and cap it; exceeding either +/// abandons the whole metric to Unsupported rather than publishing a truncated member +/// set, because a silently half-flattened UDT looks exactly like a complete one. +/// +/// +/// Nothing here throws — same contract as , for the same +/// reason: this runs on MQTTnet's shared dispatcher thread behind an unauthenticated firehose. +/// +/// +public static class SparkplugFlattener +{ + /// + /// How deeply nested a Template may be before the whole metric is refused. Five is far past + /// anything a real UDT library produces and still bounds the walk. + /// + public const int MaxDepth = 5; + + /// + /// Ceiling on the flattened members one metric may produce. A DataSet is columns × rows, so a + /// modest-looking table is the realistic way to reach this. + /// + public const int MaxMembers = 512; + + /// + /// Separator between a structured metric's name and its member path. The same character + /// Sparkplug itself uses for metric-name hierarchy, so Motor/Speed is what an edge node + /// would have called a flat metric of the same meaning. + /// + public const char MemberSeparator = '/'; + + /// + /// Flattens a Template into member paths relative to the owning metric. + /// + /// The template value. + /// The flattened members on success; empty otherwise. + /// + /// when the whole template flattened. means the + /// caller must treat the metric as . + /// + public static bool TryFlattenTemplate(Payload.Types.Template template, out IReadOnlyList members) + { + var collected = new List(); + var ok = template is not null && AppendTemplate(template, prefix: "", depth: 1, collected); + members = ok ? collected : []; + return ok; + } + + /// + /// Flattens a DataSet into one member per cell, named {column}/{rowIndex}. + /// + /// The dataset value. + /// The flattened members on success; empty otherwise. + /// + /// when the whole dataset flattened. means the + /// caller must treat the metric as . + /// + /// + /// + /// Per cell, not per column — and the row index is always present. The recorded + /// decision says "per-column tags, not one array-valued variable", and a column of N rows + /// is not a scalar, so per-column can only mean per-cell once N > 1. Emitting a bare + /// {column} for the single-row case and {column}/{row} otherwise would make + /// a tag's own name depend on how many rows the plant happened to publish — so a table + /// that grew from one row to two would rename every existing tag out from under its + /// authored binding. Uniform naming costs a /0 on the common single-row "current + /// values" shape and buys a stable address. + /// + /// + /// A row that disappears simply stops publishing, and its tag holds its last value — the + /// same as any metric an edge node stops sending. + /// + /// + public static bool TryFlattenDataSet(Payload.Types.DataSet dataSet, out IReadOnlyList members) + { + members = []; + if (dataSet is null) + { + return false; + } + + // Columns and types are parallel arrays. A publisher that disagrees with itself has produced a + // dataset whose cells cannot be typed, and guessing which array to trust would type a column + // by its neighbour — refuse instead. + if (dataSet.Columns.Count == 0 || dataSet.Types_.Count != dataSet.Columns.Count) + { + return false; + } + + if ((long)dataSet.Columns.Count * dataSet.Rows.Count > MaxMembers) + { + return false; + } + + var collected = new List(); + for (var row = 0; row < dataSet.Rows.Count; row++) + { + var elements = dataSet.Rows[row].Elements; + + // A short row is a malformed dataset, not a row of nulls: emitting the cells that ARE + // present would publish a table silently missing its right-hand columns. + if (elements.Count != dataSet.Columns.Count) + { + return false; + } + + for (var col = 0; col < dataSet.Columns.Count; col++) + { + var column = dataSet.Columns[col]; + if (string.IsNullOrEmpty(column)) + { + return false; // An unnamed column has no address. + } + + var (kind, value) = ProjectDataSetValue(elements[col]); + collected.Add(new SparkplugMember( + Path: $"{column}{MemberSeparator}{row.ToString(System.Globalization.CultureInfo.InvariantCulture)}", + DataType: (TahuDataType)dataSet.Types_[col], + ValueKind: kind, + Value: value)); + } + } + + members = collected; + return true; + } + + /// + /// Projects a metric's PropertySet into flat name/value metadata. + /// + /// The property set, or . + /// The projected properties; empty when there are none or the set is malformed. + /// + /// + /// Metadata, never tags — the recorded decision, and the honest reading of the + /// spec: a PropertySet describes a metric (engineering units, ranges, quality hints), it is + /// not a value the metric has. Surfacing it as sibling tags would invent plant signals out + /// of documentation. + /// + /// + /// Nested PropertySet/PropertySetList values are flattened one level with a + /// /-joined key, matching the metric-member convention; anything deeper, or an + /// extension value, is dropped rather than stringified. + /// + /// + public static IReadOnlyList ProjectProperties(Payload.Types.PropertySet? properties) + { + if (properties is null || properties.Keys.Count == 0 || properties.Keys.Count != properties.Values.Count) + { + return []; + } + + var collected = new List(); + AppendProperties(properties, prefix: "", depth: 1, collected); + return collected; + } + + private static void AppendProperties( + Payload.Types.PropertySet set, string prefix, int depth, List into) + { + if (set.Keys.Count != set.Values.Count || depth > 2 || into.Count >= MaxMembers) + { + return; + } + + for (var i = 0; i < set.Keys.Count && into.Count < MaxMembers; i++) + { + var key = set.Keys[i]; + if (string.IsNullOrEmpty(key)) + { + continue; + } + + var name = prefix.Length == 0 ? key : $"{prefix}{MemberSeparator}{key}"; + var value = set.Values[i]; + + if (value.HasIsNull && value.IsNull) + { + into.Add(new SparkplugProperty(name, null)); + continue; + } + + switch (value.ValueCase) + { + case Payload.Types.PropertyValue.ValueOneofCase.IntValue: + into.Add(new SparkplugProperty(name, value.IntValue)); + break; + case Payload.Types.PropertyValue.ValueOneofCase.LongValue: + into.Add(new SparkplugProperty(name, value.LongValue)); + break; + case Payload.Types.PropertyValue.ValueOneofCase.FloatValue: + into.Add(new SparkplugProperty(name, value.FloatValue)); + break; + case Payload.Types.PropertyValue.ValueOneofCase.DoubleValue: + into.Add(new SparkplugProperty(name, value.DoubleValue)); + break; + case Payload.Types.PropertyValue.ValueOneofCase.BooleanValue: + into.Add(new SparkplugProperty(name, value.BooleanValue)); + break; + case Payload.Types.PropertyValue.ValueOneofCase.StringValue: + into.Add(new SparkplugProperty(name, value.StringValue)); + break; + case Payload.Types.PropertyValue.ValueOneofCase.PropertysetValue: + AppendProperties(value.PropertysetValue, name, depth + 1, into); + break; + default: + // PropertySetList and extension values are dropped rather than stringified — a + // property whose displayed value is not what its type says is worse than an absent + // one, and these are documentation, not signal. + break; + } + } + } + + /// Walks one template level, appending members under . + /// as soon as anything cannot be flattened honestly. + private static bool AppendTemplate( + Payload.Types.Template template, string prefix, int depth, List into) + { + if (depth > MaxDepth) + { + return false; + } + + // A template DEFINITION carries the UDT's shape with no instance values. It is metadata an + // edge node publishes so consumers can understand instances — flattening it would materialise + // a parallel set of tags that never carry a value. + if (template.HasIsDefinition && template.IsDefinition) + { + return false; + } + + foreach (var member in template.Metrics) + { + if (into.Count >= MaxMembers) + { + return false; + } + + if (!member.HasName || string.IsNullOrEmpty(member.Name)) + { + // A template member is addressed by name; there is no alias fallback inside a template. + return false; + } + + var path = prefix.Length == 0 ? member.Name : $"{prefix}{MemberSeparator}{member.Name}"; + + if (member.ValueCase == Payload.Types.Metric.ValueOneofCase.TemplateValue) + { + if (!AppendTemplate(member.TemplateValue, path, depth + 1, into)) + { + return false; + } + + continue; + } + + if (member.ValueCase == Payload.Types.Metric.ValueOneofCase.DatasetValue) + { + if (!TryFlattenDataSet(member.DatasetValue, out var cells) + || into.Count + cells.Count > MaxMembers) + { + return false; + } + + foreach (var cell in cells) + { + into.Add(cell with { Path = $"{path}{MemberSeparator}{cell.Path}" }); + } + + continue; + } + + var (kind, value) = SparkplugCodec.ProjectMemberValue(member); + if (kind == SparkplugValueKind.Unsupported) + { + return false; // An extension value inside a template: refuse the whole metric. + } + + into.Add(new SparkplugMember( + Path: path, + DataType: member.HasDatatype ? (TahuDataType)member.Datatype : null, + ValueKind: kind, + Value: value)); + } + + return true; + } + + private static (SparkplugValueKind Kind, object? Value) ProjectDataSetValue( + Payload.Types.DataSet.Types.DataSetValue value) => value.ValueCase switch + { + Payload.Types.DataSet.Types.DataSetValue.ValueOneofCase.IntValue => (SparkplugValueKind.Scalar, value.IntValue), + Payload.Types.DataSet.Types.DataSetValue.ValueOneofCase.LongValue => (SparkplugValueKind.Scalar, value.LongValue), + Payload.Types.DataSet.Types.DataSetValue.ValueOneofCase.FloatValue => (SparkplugValueKind.Scalar, value.FloatValue), + Payload.Types.DataSet.Types.DataSetValue.ValueOneofCase.DoubleValue => (SparkplugValueKind.Scalar, value.DoubleValue), + Payload.Types.DataSet.Types.DataSetValue.ValueOneofCase.BooleanValue => (SparkplugValueKind.Scalar, value.BooleanValue), + Payload.Types.DataSet.Types.DataSetValue.ValueOneofCase.StringValue => (SparkplugValueKind.Scalar, value.StringValue), + + // A DataSetValue has no is_null field, so an empty oneof IS the null cell — unlike a Metric, + // where Absent and explicit-null are genuinely different states. + _ => (SparkplugValueKind.Null, null), + }; +} + +/// +/// One flattened member of a structured Sparkplug metric (Gitea #515). +/// +/// +/// The member path relative to the owning metric, /-joined — Speed for a +/// top-level template member, Motor/Speed for a nested one, Temp/0 for a dataset +/// cell. The consumer composes the full metric name as {parentName}/{Path}; it is kept +/// relative here because a DATA metric carries only the parent's alias, so the parent's name is +/// not known until the alias table has resolved it. +/// +/// +/// The member's declared datatype, or when it carried none — in which case +/// the consumer falls back to the datatype the birth installed for this member path. +/// +/// What means; never Unsupported here. +/// The raw wire value, with the same signed-integer caveat as a scalar metric. +public readonly record struct SparkplugMember( + string Path, + TahuDataType? DataType, + SparkplugValueKind ValueKind, + object? Value); + +/// One entry of a metric's PropertySet, projected as node metadata (Gitea #515). +/// The property key, /-joined for a nested set. +/// +/// The boxed scalar value, or for an explicitly-null property. +/// +public readonly record struct SparkplugProperty(string Name, object? Value); diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/AliasTable.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/AliasTable.cs index 45d66b0b..e5dc9030 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/AliasTable.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/AliasTable.cs @@ -128,6 +128,36 @@ public sealed class AliasTable aliasCollisions++; (collidedAliases ??= []).Add(alias); } + + // Gitea #515: a Template/DataSet metric also installs one binding per flattened member, + // named `{parent}/{memberPath}`. The parent's own binding above is kept — a DATA metric + // carries only the parent's ALIAS, so resolving that alias to the parent's name is the only + // way the member names can be composed at all. But the parent is not itself servable (its + // Sparkplug datatype maps to no DriverDataType), so it is the member bindings that browse + // enumerates and that authored tags bind to. + // + // Members carry NO alias, deliberately: aliases are per-metric on the wire, so inventing + // one here would create a number the edge node never assigned, which a later birth could + // legitimately hand to a different metric. + foreach (var member in metric.MemberList) + { + if (string.IsNullOrWhiteSpace(member.Path)) + { + rejectedUnnamed++; + continue; + } + + var memberBinding = new SparkplugMetricBinding( + $"{binding.Name}{SparkplugFlattener.MemberSeparator}{member.Path}", + Alias: null, + member.DataType ?? TahuDataType.Unknown); + + if (!byName.TryAdd(memberBinding.Name, memberBinding)) + { + duplicateNames++; + byName[memberBinding.Name] = memberBinding; + } + } } if (collidedAliases is not null) 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 index 91b0836d..156d356e 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/SparkplugIngestor.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/SparkplugIngestor.cs @@ -1177,6 +1177,15 @@ public sealed class SparkplugIngestor SparkplugMetric metric, DateTime? sourceUtc) { + // Gitea #515: a Template/DataSet has no value of its own — its flattened members do. Expand + // here, at the single funnel both the DATA and the birth-value paths already pass through, so + // neither can end up supporting structured metrics while the other silently does not. + if (metric.ValueKind == SparkplugValueKind.Structured) + { + PublishStructuredMembers(scope, binding, metric, sourceUtc); + return; + } + // 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)) @@ -1287,6 +1296,62 @@ public sealed class SparkplugIngestor : BadTypeMismatch(def, binding, target, stamp, serverUtc); } + /// + /// Publishes each flattened member of a structured (Template / DataSet) metric as + /// the sibling metric {parent}/{memberPath} (Gitea #515). + /// + /// + /// + /// Each member is re-entered through as a scalar, rather + /// than duplicating the coercion, array-gate and authored-tag fan-out here. A second copy of + /// that logic would be free to drift — and the drift would be silent, since both paths + /// publish values that look plausible. + /// + /// + /// The member's datatype falls back to the BIRTH's. A DDATA template usually + /// re-declares its members' datatypes, but it is not obliged to; the binding installed at + /// birth is the authority when it does not. Guessing from the CLR type of the wire value + /// would type a signed Int32 of -42 as a UInt32 of 4294967254 — the exact + /// plausible-and-wrong failure ReinterpretSigned exists to prevent. + /// + /// + private void PublishStructuredMembers( + SparkplugScope scope, + SparkplugMetricBinding parent, + SparkplugMetric metric, + DateTime? sourceUtc) + { + var table = Births.Find(scope); + + foreach (var member in metric.MemberList) + { + if (string.IsNullOrWhiteSpace(member.Path)) + { + continue; + } + + var name = $"{parent.Name}{SparkplugFlattener.MemberSeparator}{member.Path}"; + + // Wire datatype first, birth binding second. Unknown last — which the typed layer refuses + // once, loudly, rather than coercing. + var dataType = member.DataType + ?? table?.ResolveByName(name)?.DataType + ?? Org.Eclipse.Tahu.Protobuf.DataType.Unknown; + + PublishMetric( + scope, + new SparkplugMetricBinding(name, Alias: null, dataType), + new SparkplugMetric( + Name: name, + Alias: null, + DataType: dataType, + TimestampMs: metric.TimestampMs, + ValueKind: member.ValueKind, + Value: member.Value), + sourceUtc); + } + } + /// /// 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. @@ -1370,9 +1435,7 @@ public sealed class SparkplugIngestor try { - subscribers(this, new SparkplugBirthObservedEventArgs( - scope, - [.. payload.Metrics.Where(m => m.Name is not null).Select(m => m.Name!)])); + subscribers(this, new SparkplugBirthObservedEventArgs(scope, [.. DiscoverableNames(payload)])); } catch (Exception ex) { @@ -1380,6 +1443,38 @@ public sealed class SparkplugIngestor } } + /// + /// The metric names a birth contributes to the discoverable address space — which is + /// what the rediscovery change gate must compare, not the raw wire names. + /// + /// + /// Gitea #515: a structured metric contributes its flattened MEMBERS and not itself. Comparing + /// raw wire names instead would make the gate blind to the change that matters most for a UDT: + /// a template gaining or losing a member leaves the parent metric's name identical, so the + /// signature would not move and the operator would never be told the tree changed. + /// + private static IEnumerable DiscoverableNames(SparkplugPayload payload) + { + foreach (var metric in payload.Metrics) + { + if (metric.Name is not { } name) + { + continue; + } + + if (metric.ValueKind != SparkplugValueKind.Structured) + { + yield return name; + continue; + } + + foreach (var member in metric.MemberList) + { + yield return $"{name}{SparkplugFlattener.MemberSeparator}{member.Path}"; + } + } + } + // ---- sequence + rebirth ---- private SequenceTracker TrackerFor(SparkplugNodeKey node) => diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Drivers/DriverFactoryBootstrap.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Drivers/DriverFactoryBootstrap.cs index 29037bfe..4943dee6 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Drivers/DriverFactoryBootstrap.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Drivers/DriverFactoryBootstrap.cs @@ -5,6 +5,7 @@ using ZB.MOM.WW.OtOpcUa.Core.Abstractions; using ZB.MOM.WW.OtOpcUa.Core.Hosting; using ZB.MOM.WW.OtOpcUa.Core.Resilience; using ZB.MOM.WW.Secrets.Abstractions; +using ZB.MOM.WW.OtOpcUa.Runtime.Redundancy; namespace ZB.MOM.WW.OtOpcUa.Host.Drivers; @@ -45,6 +46,19 @@ public static class DriverFactoryBootstrap /// The same service collection, for chaining. public static IServiceCollection AddOtOpcUaDriverFactories(this IServiceCollection services) { + // Gitea #511: the Primary data-plane verdict the MQTT driver's Sparkplug host claim is gated on. + // + // Registered HERE as well as in AddOtOpcUaRuntime, and both sides TryAdd the same concrete + // RedundancyRoleView, so whichever extension runs first creates it and the other resolves the + // SAME instance. That is deliberately not a defensive duplicate: the resolve below is + // GetRequiredService (a null would silently fall the driver back to "claim unconditionally", + // which on a redundant pair is the dual-claim this mechanism exists to prevent), so without + // this line the registration becomes order-dependent on two extensions a caller is free to + // invoke in either order — and any graph that wires driver factories without the full runtime + // fails at resolution. Two Host integration suites do exactly that, and caught it. + services.TryAddSingleton(); + services.TryAddSingleton(sp => sp.GetRequiredService()); + services.AddSingleton(sp => { var registry = new DriverFactoryRegistry(); diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/SparkplugSimulator/Program.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/SparkplugSimulator/Program.cs index db045291..87b81759 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/SparkplugSimulator/Program.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/SparkplugSimulator/Program.cs @@ -81,7 +81,14 @@ foreach (var node in nodes) "Filler1", new SparkplugMetricSeed("Temperature", 10, DataType.Float, 55.5f), new SparkplugMetricSeed("FillCount", 11, DataType.Int64, 1000L), - new SparkplugMetricSeed("Jammed", 12, DataType.Boolean, false)); + new SparkplugMetricSeed("Jammed", 12, DataType.Boolean, false), + + // Structured metrics for the Gitea #515 gate. They live on the DEVICE rather than the node + // deliberately: a DDATA template is the case a naive implementation gets wrong, because its + // metric carries the parent's ALIAS and no name, so the member names cannot be composed until + // the alias table has resolved the parent. + new SparkplugMetricSeed("Motor", 13, DataType.Template, BuildMotorTemplate()), + new SparkplugMetricSeed("Recipe", 14, DataType.DataSet, BuildRecipeDataSet())); edges.Add(edge); } @@ -112,3 +119,55 @@ return 0; static string Env(string name, string fallback) => Environment.GetEnvironmentVariable(name) is { Length: > 0 } value ? value : fallback; + +// A UDT instance: `Motor` with a `Speed` double, a `Running` bool and a nested `Stator` template. +// Flattens to `Motor/Speed`, `Motor/Running` and `Motor/Stator/Temp` (Gitea #515). +static Payload.Types.Template BuildMotorTemplate() +{ + var stator = new Payload.Types.Template { Version = "1", IsDefinition = false }; + stator.Metrics.Add(new Payload.Types.Metric + { + Name = "Temp", Datatype = (uint)DataType.Double, DoubleValue = 71.5, + }); + + var motor = new Payload.Types.Template { Version = "1", IsDefinition = false }; + motor.Metrics.Add(new Payload.Types.Metric + { + Name = "Speed", Datatype = (uint)DataType.Double, DoubleValue = 1500, + }); + motor.Metrics.Add(new Payload.Types.Metric + { + Name = "Running", Datatype = (uint)DataType.Boolean, BooleanValue = true, + }); + + // Nested one level: proves the recursion, and proves the `/`-joined path composes correctly rather + // than only working for flat templates. + motor.Metrics.Add(new Payload.Types.Metric + { + Name = "Stator", Datatype = (uint)DataType.Template, TemplateValue = stator, + }); + + return motor; +} + +// A two-column, two-row table. Flattens to `Recipe/Step/0`, `Recipe/Target/0`, `Recipe/Step/1`, +// `Recipe/Target/1` — the row index is always present, so a table that grows a row never renames the +// tags already authored against it. +static Payload.Types.DataSet BuildRecipeDataSet() +{ + var dataSet = new Payload.Types.DataSet { NumOfColumns = 2UL }; + dataSet.Columns.Add("Step"); + dataSet.Columns.Add("Target"); + dataSet.Types_.Add((uint)DataType.Int32); + dataSet.Types_.Add((uint)DataType.Double); + + for (var row = 0; row < 2; row++) + { + var r = new Payload.Types.DataSet.Types.Row(); + r.Elements.Add(new Payload.Types.DataSet.Types.DataSetValue { IntValue = (uint)row }); + r.Elements.Add(new Payload.Types.DataSet.Types.DataSetValue { DoubleValue = 20d + row }); + dataSet.Rows.Add(r); + } + + return dataSet; +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/SparkplugSimulator/SparkplugEdgeNode.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/SparkplugSimulator/SparkplugEdgeNode.cs index 0140af0d..425cafdc 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/SparkplugSimulator/SparkplugEdgeNode.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/SparkplugSimulator/SparkplugEdgeNode.cs @@ -756,6 +756,18 @@ public sealed class SparkplugEdgeNode : IAsyncDisposable case DataType.File: metric.BytesValue = ByteString.CopyFrom((byte[])value); break; + + // Structured values (Gitea #515). The seed carries the already-built protobuf message + // rather than a scalar, because a Template's shape IS its value — there is nothing simpler + // to convert from, and inventing a mini-DSL here would put a second, divergent notion of + // "what a template is" between the simulator and the wire it is supposed to reproduce. + case DataType.Template: + metric.TemplateValue = (Payload.Types.Template)value; + break; + case DataType.DataSet: + metric.DatasetValue = (Payload.Types.DataSet)value; + break; + default: throw new NotSupportedException( $"The simulator does not encode Sparkplug datatype {dataType}; v1 of the driver does " @@ -824,9 +836,83 @@ public sealed class SparkplugEdgeNode : IAsyncDisposable DataType.Uint8 or DataType.Uint16 or DataType.Uint32 or DataType.Uint64 => (uint)tick, DataType.Boolean => tick % 2 == 0, DataType.String or DataType.Text or DataType.Uuid => $"tick-{tick}", + + // Structured values move too, or the live gate could not tell a flattened member that is being + // fed from one that was merely published once at birth and then went quiet. + DataType.Template => AdvanceTemplate((Payload.Types.Template)seed.Value!, tick), + DataType.DataSet => AdvanceDataSet((Payload.Types.DataSet)seed.Value!, tick), + _ => seed.Value, }; + /// + /// Rebuilds a template with its numeric members nudged, preserving names, datatypes and order. + /// + private static Payload.Types.Template AdvanceTemplate(Payload.Types.Template seed, int tick) + { + var next = new Payload.Types.Template { Version = seed.Version, IsDefinition = false }; + foreach (var member in seed.Metrics) + { + var clone = member.Clone(); + switch ((DataType)clone.Datatype) + { + case DataType.Double: + clone.DoubleValue = 1500d + (tick % 100); + break; + case DataType.Float: + clone.FloatValue = 40f + (tick % 20) / 10f; + break; + case DataType.Int32: + clone.IntValue = unchecked((uint)(-tick)); + break; + case DataType.Boolean: + clone.BooleanValue = tick % 2 == 0; + break; + default: + break; // leave anything else at its seeded value + } + + clone.Timestamp = NowMs(); + next.Metrics.Add(clone); + } + + return next; + } + + /// Rebuilds a dataset with its numeric cells nudged, preserving columns, types and shape. + private static Payload.Types.DataSet AdvanceDataSet(Payload.Types.DataSet seed, int tick) + { + var next = new Payload.Types.DataSet { NumOfColumns = seed.NumOfColumns }; + next.Columns.AddRange(seed.Columns); + next.Types_.AddRange(seed.Types_); + + for (var row = 0; row < seed.Rows.Count; row++) + { + var cloned = new Payload.Types.DataSet.Types.Row(); + for (var col = 0; col < seed.Rows[row].Elements.Count; col++) + { + var cell = seed.Rows[row].Elements[col].Clone(); + switch ((DataType)seed.Types_[col]) + { + case DataType.Double: + cell.DoubleValue = 20d + row + (tick % 10) / 10d; + break; + case DataType.Int32: + cell.IntValue = (uint)(tick + row); + break; + default: + break; + } + + cloned.Elements.Add(cell); + } + + next.Rows.Add(cloned); + } + + return next; + } + private void ConfigureTls(MqttClientTlsOptionsBuilder tls) { if (!_options.UseTls) diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/SparkplugStructuredLiveTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/SparkplugStructuredLiveTests.cs new file mode 100644 index 00000000..1ccb00c5 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/SparkplugStructuredLiveTests.cs @@ -0,0 +1,332 @@ +using Org.Eclipse.Tahu.Protobuf; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Driver.Mqtt.SparkplugSimulator; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests; + +/// +/// Live gate for structured Sparkplug metrics (Gitea #515): a Template and a +/// DataSet published by a real edge node over the Mosquitto fixture, flattened into +/// addressable OPC UA tags. +/// +/// +/// +/// What this adds over the offline suite. The offline tests build decoded payloads by +/// hand, so they prove the mapping but not the wire: that a nested template really does +/// round-trip through protobuf with its member names intact, and — the case a naive +/// implementation gets wrong — that a DDATA structured metric arrives carrying the parent's +/// alias and nothing else, so the member names cannot be composed until the alias table +/// has resolved the parent. +/// +/// +/// The issue's acceptance criterion, verbatim: "a simulator publishing a Template and a +/// DataSet metric produces addressable, correctly-typed OPC UA nodes, live-verified; +/// every still-unsupported kind continues to skip with a warning." +/// +/// +[Trait("Category", "Integration")] +[Collection("MqttFixture")] +public sealed class SparkplugStructuredLiveTests(MqttFixture fixture) +{ + private const uint Good = 0x00000000u; + private static readonly TimeSpan Timeout = TimeSpan.FromSeconds(20); + + private void SkipIfUnconfigured() => Assert.SkipWhen(fixture.SkipReason is not null, fixture.SkipReason ?? ""); + + [Fact] + public async Task A_template_metric_flattens_to_addressable_tags_across_birth_and_data() + { + SkipIfUnconfigured(); + + var ct = TestContext.Current.CancellationToken; + var group = NewGroup("tmpl"); + const string node = "EdgeA"; + const string device = "Filler1"; + + const string speedPath = "Fixture/Sparkplug/Motor/Speed"; + const string runningPath = "Fixture/Sparkplug/Motor/Running"; + const string statorPath = "Fixture/Sparkplug/Motor/Stator/Temp"; + + await using var sim = new SparkplugEdgeNode(SimOptions(group, node)); + sim.SetDeviceMetrics(device, new SparkplugMetricSeed("Motor", 13, DataType.Template, MotorTemplate(speed: 1500, running: true, statorTemp: 71.5))); + + await sim.ConnectAsync(ct); + await sim.BirthAsync(ct); + + await using var driver = new MqttDriver( + SparkplugOptions( + group, + Tag(speedPath, group, node, device, "Motor/Speed"), + Tag(runningPath, group, node, device, "Motor/Running"), + Tag(statorPath, group, node, device, "Motor/Stator/Temp")), + "mqtt-live-sp-template"); + var recorder = new Recorder(driver); + + await driver.InitializeAsync(string.Empty, ct); + await driver.SubscribeAsync([speedPath, runningPath, statorPath], TimeSpan.FromSeconds(1), ct); + + // The DBIRTH is a full snapshot, so these are the birth's own member values — including the + // NESTED template's, which is what proves the `/`-joined path composes past one level. + var birth = await recorder.WaitAsync(speedPath, v => Equals(v.Value, 1500d), Timeout, ct); + birth.Snapshot.StatusCode.ShouldBe(Good); + await recorder.WaitAsync(runningPath, v => Equals(v.Value, true), Timeout, ct); + await recorder.WaitAsync(statorPath, v => Equals(v.Value, 71.5d), Timeout, ct); + + // …and now DDATA. The metric on the wire carries the PARENT's alias and no name at all, so a + // driver that flattened only at decode time — without resolving the parent alias to its name + // first — would have nothing to compose `Motor/Speed` from and would feed none of these. + await sim.PublishDeviceDataAsync( + device, [("Motor", MotorTemplate(speed: 1750, running: false, statorTemp: 68.25))], ct); + + var data = await recorder.WaitAsync(speedPath, v => Equals(v.Value, 1750d), Timeout, ct); + data.Snapshot.StatusCode.ShouldBe(Good); + data.FullReference.ShouldBe(speedPath, "values publish under the tag's RawPath, never the metric name"); + + await recorder.WaitAsync(runningPath, v => Equals(v.Value, false), Timeout, ct); + await recorder.WaitAsync(statorPath, v => Equals(v.Value, 68.25d), Timeout, ct); + + // The read path serves the same last values under the same keys. + var read = await driver.ReadAsync([speedPath, runningPath, statorPath], ct); + read.Select(r => r.StatusCode).ShouldAllBe(s => s == Good); + read[0].Value.ShouldBe(1750d); + } + + [Fact] + public async Task A_dataset_metric_flattens_to_one_tag_per_cell() + { + SkipIfUnconfigured(); + + var ct = TestContext.Current.CancellationToken; + var group = NewGroup("dset"); + const string node = "EdgeA"; + const string device = "Filler1"; + + const string step0Path = "Fixture/Sparkplug/Recipe/Step0"; + const string target1Path = "Fixture/Sparkplug/Recipe/Target1"; + + await using var sim = new SparkplugEdgeNode(SimOptions(group, node)); + sim.SetDeviceMetrics(device, new SparkplugMetricSeed("Recipe", 14, DataType.DataSet, RecipeDataSet(baseTarget: 20d))); + + await sim.ConnectAsync(ct); + await sim.BirthAsync(ct); + + await using var driver = new MqttDriver( + SparkplugOptions( + group, + // The row index is always present, including on row 0 — the naming is uniform so a + // table that grows a row never renames the tags already authored against it. + Tag(step0Path, group, node, device, "Recipe/Step/0"), + Tag(target1Path, group, node, device, "Recipe/Target/1")), + "mqtt-live-sp-dataset"); + var recorder = new Recorder(driver); + + await driver.InitializeAsync(string.Empty, ct); + await driver.SubscribeAsync([step0Path, target1Path], TimeSpan.FromSeconds(1), ct); + + // Cells are typed per COLUMN: Step is Int32, Target is Double. A dataset whose columns and + // types disagree is refused outright, so a wrong type here is not a near-miss — it is the + // difference between a typed tag and no tag. + await recorder.WaitAsync(step0Path, v => Equals(v.Value, 0), Timeout, ct); + await recorder.WaitAsync(target1Path, v => Equals(v.Value, 21d), Timeout, ct); + + await sim.PublishDeviceDataAsync(device, [("Recipe", RecipeDataSet(baseTarget: 30d))], ct); + + var target = await recorder.WaitAsync(target1Path, v => Equals(v.Value, 31d), Timeout, ct); + target.Snapshot.StatusCode.ShouldBe(Good); + } + + [Fact] + public async Task A_template_definition_is_still_skipped_rather_than_flattened() + { + // Skip-and-warn survives — the requirement the issue is explicit about. A definition names its + // members and carries no values, so flattening it would materialise tags that can never read, + // indistinguishable in the picker from an instance's. + SkipIfUnconfigured(); + + var ct = TestContext.Current.CancellationToken; + var group = NewGroup("tdef"); + const string node = "EdgeA"; + const string device = "Filler1"; + + const string speedPath = "Fixture/Sparkplug/Proto/Speed"; + const string livePath = "Fixture/Sparkplug/Proto/Live"; + + var definition = new Payload.Types.Template { Version = "1", IsDefinition = true }; + definition.Metrics.Add(new Payload.Types.Metric { Name = "Speed", Datatype = (uint)DataType.Double }); + + await using var sim = new SparkplugEdgeNode(SimOptions(group, node)); + sim.SetDeviceMetrics( + device, + new SparkplugMetricSeed("MotorType", 15, DataType.Template, definition), + + // The positive control, and it is load-bearing: without it "Speed never arrived" is + // satisfied by any failure to connect, birth or subscribe, and the test would pass against + // a driver that ingests nothing at all. + new SparkplugMetricSeed("Live", 16, DataType.Double, 42.5d)); + + await sim.ConnectAsync(ct); + await sim.BirthAsync(ct); + + await using var driver = new MqttDriver( + SparkplugOptions( + group, + Tag(speedPath, group, node, device, "MotorType/Speed"), + Tag(livePath, group, node, device, "Live")), + "mqtt-live-sp-tdef"); + var recorder = new Recorder(driver); + + await driver.InitializeAsync(string.Empty, ct); + await driver.SubscribeAsync([speedPath, livePath], TimeSpan.FromSeconds(1), ct); + + // Wait for the control FIRST, so the absence assertion below is made at a moment the birth is + // known to have been processed. + await recorder.WaitAsync(livePath, v => Equals(v.Value, 42.5d), Timeout, ct); + + recorder.Count(speedPath).ShouldBe(0, "a template definition carries no values to publish"); + + var read = await driver.ReadAsync([speedPath], ct); + read[0].StatusCode.ShouldNotBe(Good, "a tag bound to an unflattened member has never had a value"); + } + + // ---- helpers ----------------------------------------------------------------------------------- + + private static string NewGroup(string label) => $"OtOpcUaLive-{label}-{Guid.NewGuid():N}"[..40]; + + private static Payload.Types.Template MotorTemplate(double speed, bool running, double statorTemp) + { + var stator = new Payload.Types.Template { Version = "1", IsDefinition = false }; + stator.Metrics.Add(new Payload.Types.Metric + { + Name = "Temp", Datatype = (uint)DataType.Double, DoubleValue = statorTemp, + }); + + var motor = new Payload.Types.Template { Version = "1", IsDefinition = false }; + motor.Metrics.Add(new Payload.Types.Metric + { + Name = "Speed", Datatype = (uint)DataType.Double, DoubleValue = speed, + }); + motor.Metrics.Add(new Payload.Types.Metric + { + Name = "Running", Datatype = (uint)DataType.Boolean, BooleanValue = running, + }); + motor.Metrics.Add(new Payload.Types.Metric + { + Name = "Stator", Datatype = (uint)DataType.Template, TemplateValue = stator, + }); + return motor; + } + + private static Payload.Types.DataSet RecipeDataSet(double baseTarget) + { + var dataSet = new Payload.Types.DataSet { NumOfColumns = 2UL }; + dataSet.Columns.Add("Step"); + dataSet.Columns.Add("Target"); + dataSet.Types_.Add((uint)DataType.Int32); + dataSet.Types_.Add((uint)DataType.Double); + + for (var row = 0; row < 2; row++) + { + var r = new Payload.Types.DataSet.Types.Row(); + r.Elements.Add(new Payload.Types.DataSet.Types.DataSetValue { IntValue = (uint)row }); + r.Elements.Add(new Payload.Types.DataSet.Types.DataSetValue { DoubleValue = baseTarget + row }); + dataSet.Rows.Add(r); + } + + return dataSet; + } + + private MqttDriverOptions SparkplugOptions(string groupId, params RawTagEntry[] rawTags) => new() + { + Host = fixture.TlsHost, + Port = fixture.TlsPort, + UseTls = true, + CaCertificatePath = fixture.CaCertificatePath, + AllowUntrustedServerCertificate = fixture.CaCertificatePath is null, + Username = fixture.Username, + Password = fixture.Password, + ClientId = $"otopcua-live-spx-{Guid.NewGuid():N}", + ConnectTimeoutSeconds = 15, + Mode = MqttMode.SparkplugB, + Sparkplug = new MqttSparkplugOptions { GroupId = groupId, RequestRebirthOnGap = true }, + RawTags = rawTags, + }; + + private SparkplugEdgeNodeOptions SimOptions(string groupId, string edgeNodeId) => new() + { + Host = fixture.TlsHost, + Port = fixture.TlsPort, + UseTls = true, + CaCertificatePath = fixture.CaCertificatePath, + AllowUntrustedServerCertificate = fixture.CaCertificatePath is null, + Username = fixture.Username, + Password = fixture.Password, + GroupId = groupId, + EdgeNodeId = edgeNodeId, + }; + + private static RawTagEntry Tag( + string rawPath, string groupId, string edgeNodeId, string? deviceId, string metricName) + { + var device = deviceId is null ? "" : $"\"deviceId\":\"{deviceId}\","; + return new RawTagEntry( + rawPath, + $$"""{"groupId":"{{groupId}}","edgeNodeId":"{{edgeNodeId}}",{{device}}"metricName":"{{metricName}}"}""", + WriteIdempotent: false); + } + + /// Records every published data change, so a test can wait on one and count others. + private sealed class Recorder + { + private readonly List _events = []; + private readonly Lock _gate = new(); + + public Recorder(MqttDriver driver) => driver.OnDataChange += (_, e) => + { + lock (_gate) + { + _events.Add(e); + } + }; + + public int Count(string rawPath) + { + lock (_gate) + { + return _events.Count(e => e.FullReference == rawPath); + } + } + + public async Task WaitAsync( + string rawPath, Func predicate, TimeSpan timeout, CancellationToken ct) + { + var deadline = DateTime.UtcNow + timeout; + while (true) + { + lock (_gate) + { + foreach (var candidate in _events) + { + if (candidate.FullReference == rawPath && predicate(candidate.Snapshot)) + { + return candidate; + } + } + } + + if (DateTime.UtcNow >= deadline) + { + lock (_gate) + { + throw new TimeoutException( + $"No matching value for '{rawPath}' within {timeout}. Seen: " + + string.Join(", ", _events.Select(e => $"{e.FullReference}={e.Snapshot.Value}"))); + } + } + + await Task.Delay(100, ct); + } + } + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttBrowseSessionTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttBrowseSessionTests.cs index 39cffaa9..5427e985 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttBrowseSessionTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttBrowseSessionTests.cs @@ -486,6 +486,31 @@ public sealed class MqttBrowseSessionTests s.PublishCountForTest.ShouldBe(0); // passive — browsing a live plant broker publishes nothing } + [Fact] + public async Task SparkplugTree_TemplateMetric_OffersItsMembersAndNotItself() + { + // Gitea #515. The parent has no value and no servable datatype, so a leaf for `Motor` would + // offer the operator an address that can only ever commit a tag which never reads — and the + // picker gives no hint that one leaf is real and the other is not. + await using var s = new MqttBrowseSession(MqttMode.SparkplugB); + s.ObserveTemplateBirthForTest( + "Plant1", "EdgeA", "Filler1", "Motor", + ("Speed", SparkplugDataType.Double), + ("Torque", SparkplugDataType.Double)); + + var metrics = await s.ExpandAsync("Plant1/EdgeA/Filler1", TestContext.Current.CancellationToken); + + metrics.Select(n => n.DisplayName).ShouldBe(["Motor/Speed", "Motor/Torque"]); + metrics.ShouldNotContain(n => n.DisplayName == "Motor"); + metrics[0].Kind.ShouldBe(BrowseNodeKind.Leaf); + + // The metric name travels INTACT as the node id — it is the binding key. Only the derived tag + // Name is later coerced to a legal RawPath segment, by RawBrowseCommitMapper (Gitea #509). + metrics[0].NodeId.ShouldBe("Plant1/EdgeA/Filler1::Motor/Speed"); + + s.PublishCountForTest.ShouldBe(0); + } + [Fact] public async Task SparkplugTree_NodeLevelMetrics_HangDirectlyUnderTheEdgeNode() { diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/SparkplugCodecTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/SparkplugCodecTests.cs index 77cbff89..aefaca6a 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/SparkplugCodecTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/SparkplugCodecTests.cs @@ -186,35 +186,82 @@ public sealed class SparkplugCodecTests } /// - /// DataSet and Template metrics are out of scope for v1. They must decode to an - /// explicit "unsupported" kind — never an exception on the MQTT dispatcher thread, and never a - /// silently-dropped metric that looks identical to a metric that was never published. + /// A DataSet metric decodes to with its cells + /// flattened (Gitea #515) — never an exception on the MQTT dispatcher thread, and never a + /// silently-dropped metric that looks identical to one that was never published. /// [Fact] - public void Decode_GoldenNbirth_DataSetMetric_IsUnsupported_NotAThrow() + public void Decode_GoldenNbirth_DataSetMetric_FlattensToCells() { var payload = SparkplugCodec.Decode(SparkplugGoldenPayloads.Read(SparkplugGoldenPayloads.NbirthFile)); var config = payload.Metrics.Single(m => m.Name == "Config"); - config.ValueKind.ShouldBe(SparkplugValueKind.Unsupported); + config.ValueKind.ShouldBe(SparkplugValueKind.Structured); + + // The PARENT carries no value — its members do. A parent that reported a value would be a + // guess about which cell "the" value is. config.Value.ShouldBeNull(); config.DataType.ShouldBe(DataType.DataSet); + + var cell = config.MemberList.Single(); + cell.Path.ShouldBe("col0/0"); + cell.DataType.ShouldBe(DataType.Int32); + cell.Value.ShouldBe(1U); } - /// A Template metric takes the same unsupported path as DataSet. + /// A Template metric flattens its members to sibling paths. [Fact] - public void Decode_TemplateMetric_IsUnsupported() + public void Decode_TemplateMetric_FlattensMembersToSiblingPaths() { + var template = new Payload.Types.Template { Version = "1", IsDefinition = false }; + template.Metrics.Add(new Payload.Types.Metric + { + Name = "Speed", Datatype = (uint)DataType.Double, DoubleValue = 12.5, + }); + template.Metrics.Add(new Payload.Types.Metric + { + Name = "Running", Datatype = (uint)DataType.Boolean, BooleanValue = true, + }); + var payload = new Payload(); payload.Metrics.Add(new Payload.Types.Metric { Name = "Motor", Datatype = (uint)DataType.Template, - TemplateValue = new Payload.Types.Template { Version = "1", IsDefinition = false }, + TemplateValue = template, }); - SparkplugCodec.Decode(payload.ToByteArray()).Metrics.Single().ValueKind - .ShouldBe(SparkplugValueKind.Unsupported); + var metric = SparkplugCodec.Decode(payload.ToByteArray()).Metrics.Single(); + + metric.ValueKind.ShouldBe(SparkplugValueKind.Structured); + metric.MemberList.Select(m => m.Path).ShouldBe(["Speed", "Running"]); + metric.MemberList[0].Value.ShouldBe(12.5); + metric.MemberList[1].Value.ShouldBe(true); + } + + /// + /// A template definition is refused rather than flattened, and refusal is still the + /// visible kind. + /// + [Fact] + public void Decode_TemplateDefinition_StaysUnsupported() + { + // A definition is the UDT's SHAPE, published so consumers can understand instances — its member + // metrics carry no values. Flattening it would materialise a parallel set of tags that can + // never read, indistinguishable in the picker from the real instance's. + var definition = new Payload.Types.Template { Version = "1", IsDefinition = true }; + definition.Metrics.Add(new Payload.Types.Metric { Name = "Speed", Datatype = (uint)DataType.Double }); + + var payload = new Payload(); + payload.Metrics.Add(new Payload.Types.Metric + { + Name = "MotorType", Datatype = (uint)DataType.Template, TemplateValue = definition, + }); + + var metric = SparkplugCodec.Decode(payload.ToByteArray()).Metrics.Single(); + + metric.ValueKind.ShouldBe(SparkplugValueKind.Unsupported); + metric.MemberList.ShouldBeEmpty(); } /// Payload and per-metric timestamps survive as raw Sparkplug epoch milliseconds. diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/SparkplugFlattenerTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/SparkplugFlattenerTests.cs new file mode 100644 index 00000000..5b63c931 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/SparkplugFlattenerTests.cs @@ -0,0 +1,294 @@ +using Google.Protobuf; +using Org.Eclipse.Tahu.Protobuf; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Sparkplug; +using TahuDataType = Org.Eclipse.Tahu.Protobuf.DataType; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests; + +/// +/// Unit tests for — the Template/DataSet/PropertySet mapping +/// (Gitea #515). +/// +/// +/// Two properties dominate this file. First, what is flattened is addressable and correctly +/// typed. Second — and the one that would be easy to lose in a later edit — +/// skip-and-warn survives: anything that cannot be flattened honestly stays +/// rather than becoming a guessed scalar, because a +/// tag bound to data that is not what its name says is worse than a tag that never binds. +/// +[Trait("Category", "Unit")] +public sealed class SparkplugFlattenerTests +{ + // ---- Template ------------------------------------------------------------------------------- + + [Fact] + public void A_nested_template_flattens_to_slash_joined_paths() + { + // `/` is the separator Sparkplug itself uses for metric-name hierarchy, so `Motor/Stator/Temp` + // is exactly what an edge node would have called a flat metric of the same meaning. + var stator = Template( + Member("Temp", TahuDataType.Double, m => m.DoubleValue = 71.5), + Member("Current", TahuDataType.Float, m => m.FloatValue = 3.25f)); + + var motor = Template( + Member("Speed", TahuDataType.Double, m => m.DoubleValue = 1500), + MemberTemplate("Stator", stator)); + + SparkplugFlattener.TryFlattenTemplate(motor, out var members).ShouldBeTrue(); + + members.Select(m => m.Path).ShouldBe(["Speed", "Stator/Temp", "Stator/Current"]); + members[1].DataType.ShouldBe(TahuDataType.Double); + members[2].Value.ShouldBe(3.25f); + } + + [Fact] + public void A_template_member_with_no_name_refuses_the_whole_metric() + { + // A template member is addressed by name; there is no alias fallback inside a template. Emitting + // the members that DO have names would publish a UDT silently missing part of itself, which + // reads in the picker exactly like a complete one. + var template = Template( + Member("Speed", TahuDataType.Double, m => m.DoubleValue = 1), + new Payload.Types.Metric { Datatype = (uint)TahuDataType.Double, DoubleValue = 2 }); + + SparkplugFlattener.TryFlattenTemplate(template, out var members).ShouldBeFalse(); + members.ShouldBeEmpty(); + } + + [Fact] + public void An_extension_value_inside_a_template_refuses_the_whole_metric() + { + // An extension value is a third-party escape hatch with no defined semantics. There is nothing + // honest to flatten it into, and the sibling members cannot be trusted to be the whole story. + var template = Template( + Member("Speed", TahuDataType.Double, m => m.DoubleValue = 1), + Member("Odd", TahuDataType.Unknown, m => m.ExtensionValue = new Payload.Types.Metric.Types.MetricValueExtension())); + + SparkplugFlattener.TryFlattenTemplate(template, out _).ShouldBeFalse(); + } + + [Fact] + public void A_template_definition_is_refused() + { + // A definition carries the UDT's shape with no instance values — metadata for consumers, not + // signal. Flattening it would materialise tags that can never read. + var definition = Template(Member("Speed", TahuDataType.Double, _ => { })); + definition.IsDefinition = true; + + SparkplugFlattener.TryFlattenTemplate(definition, out _).ShouldBeFalse(); + } + + [Fact] + public void Nesting_past_the_depth_bound_refuses_the_whole_metric_rather_than_truncating() + { + // This walks an object graph anyone on the broker can publish, so the recursion must be bounded. + // Refusing whole beats truncating: a silently half-flattened UDT looks exactly like a complete + // one, and the operator would author tags against a subtree that quietly stops partway down. + var deepest = Template(Member("Leaf", TahuDataType.Double, m => m.DoubleValue = 1)); + var current = deepest; + for (var i = 0; i < SparkplugFlattener.MaxDepth + 1; i++) + { + current = Template(MemberTemplate($"L{i}", current)); + } + + SparkplugFlattener.TryFlattenTemplate(current, out var members).ShouldBeFalse(); + members.ShouldBeEmpty(); + + // Falsifying control: the same shape one level inside the bound flattens fine, so the refusal + // above is the bound doing its job rather than nesting being broken outright. + var shallow = Template(MemberTemplate("L0", deepest)); + SparkplugFlattener.TryFlattenTemplate(shallow, out var ok).ShouldBeTrue(); + ok.Single().Path.ShouldBe("L0/Leaf"); + } + + // ---- DataSet --------------------------------------------------------------------------------- + + [Fact] + public void A_dataset_flattens_to_one_member_per_cell_with_the_row_index_always_present() + { + // Uniform naming, deliberately. Emitting a bare `{column}` for a single row and + // `{column}/{row}` otherwise would make a tag's own name depend on how many rows the plant + // happened to publish, so a table growing from one row to two would rename every existing tag + // out from under its authored binding. + var dataSet = DataSet( + columns: ["Temp", "Pressure"], + types: [TahuDataType.Double, TahuDataType.Int32], + rows: + [ + [Cell(v => v.DoubleValue = 20.5), Cell(v => v.IntValue = 3)], + [Cell(v => v.DoubleValue = 21.0), Cell(v => v.IntValue = 4)], + ]); + + SparkplugFlattener.TryFlattenDataSet(dataSet, out var members).ShouldBeTrue(); + + members.Select(m => m.Path).ShouldBe(["Temp/0", "Pressure/0", "Temp/1", "Pressure/1"]); + members[0].DataType.ShouldBe(TahuDataType.Double); + members[1].DataType.ShouldBe(TahuDataType.Int32); + members[3].Value.ShouldBe(4U); + } + + [Fact] + public void A_single_row_dataset_still_carries_the_row_index() + => Flatten(DataSet(["Temp"], [TahuDataType.Double], [[Cell(v => v.DoubleValue = 1)]])) + .Single().Path.ShouldBe("Temp/0"); + + [Fact] + public void An_empty_cell_is_a_null_member_not_a_zero() + { + // A DataSetValue has no is_null field, so an empty oneof IS the null cell. Reading it as the + // proto default would publish 0 for a cell the plant left blank — plausible, Good, and wrong. + var members = Flatten(DataSet(["Temp"], [TahuDataType.Double], [[new Payload.Types.DataSet.Types.DataSetValue()]])); + + members.Single().ValueKind.ShouldBe(SparkplugValueKind.Null); + members.Single().Value.ShouldBeNull(); + } + + [Fact] + public void A_column_and_type_count_mismatch_refuses_the_dataset() + // Columns and types are parallel arrays. Guessing which to trust would type a column by its + // neighbour, and every cell in it would publish at Good quality as the wrong type. + => SparkplugFlattener.TryFlattenDataSet( + DataSet(["Temp", "Pressure"], [TahuDataType.Double], [] ), out _) + .ShouldBeFalse(); + + [Fact] + public void A_short_row_refuses_the_dataset() + // Not a row of nulls: emitting the cells that ARE present publishes a table silently missing + // its right-hand columns. + => SparkplugFlattener.TryFlattenDataSet( + DataSet(["Temp", "Pressure"], [TahuDataType.Double, TahuDataType.Int32], + [[Cell(v => v.DoubleValue = 1)]]), + out _) + .ShouldBeFalse(); + + [Fact] + public void An_unnamed_column_refuses_the_dataset() + => SparkplugFlattener.TryFlattenDataSet( + DataSet(["", "Pressure"], [TahuDataType.Double, TahuDataType.Int32], + [[Cell(v => v.DoubleValue = 1), Cell(v => v.IntValue = 2)]]), + out _) + .ShouldBeFalse(); + + [Fact] + public void A_dataset_past_the_member_bound_refuses_rather_than_truncating() + { + var columns = Enumerable.Range(0, 8).Select(i => $"c{i}").ToArray(); + var types = Enumerable.Repeat(TahuDataType.Int32, 8).ToArray(); + var rows = Enumerable.Range(0, 100) + .Select(_ => Enumerable.Range(0, 8).Select(_ => Cell(v => v.IntValue = 1)).ToArray()) + .ToArray(); + + // 800 cells, past MaxMembers (512). + SparkplugFlattener.TryFlattenDataSet(DataSet(columns, types, rows), out var members).ShouldBeFalse(); + members.ShouldBeEmpty(); + } + + // ---- PropertySet ------------------------------------------------------------------------------ + + [Fact] + public void Properties_project_to_flat_metadata_including_one_nested_level() + { + var nested = new Payload.Types.PropertySet(); + nested.Keys.Add("Low"); + nested.Values.Add(new Payload.Types.PropertyValue { DoubleValue = 0 }); + + var set = new Payload.Types.PropertySet(); + set.Keys.Add("engUnit"); + set.Values.Add(new Payload.Types.PropertyValue { StringValue = "degC" }); + set.Keys.Add("Range"); + set.Values.Add(new Payload.Types.PropertyValue { PropertysetValue = nested }); + + var projected = SparkplugFlattener.ProjectProperties(set); + + projected.Select(p => p.Name).ShouldBe(["engUnit", "Range/Low"]); + projected[0].Value.ShouldBe("degC"); + } + + [Fact] + public void Properties_never_become_members() + { + // The recorded decision, and the honest reading of the spec: a PropertySet describes a metric + // (units, ranges, quality hints) — it is not a value the metric HAS. Surfacing it as sibling + // tags would invent plant signals out of documentation. + var set = new Payload.Types.PropertySet(); + set.Keys.Add("engUnit"); + set.Values.Add(new Payload.Types.PropertyValue { StringValue = "degC" }); + + var payload = new Payload(); + var metric = new Payload.Types.Metric + { + Name = "Temperature", Datatype = (uint)TahuDataType.Double, DoubleValue = 20.5, Properties = set, + }; + payload.Metrics.Add(metric); + + var decoded = SparkplugCodec.Decode(payload.ToByteArray()).Metrics.Single(); + + decoded.ValueKind.ShouldBe(SparkplugValueKind.Scalar); + decoded.MemberList.ShouldBeEmpty(); + decoded.PropertyList.Single().Name.ShouldBe("engUnit"); + } + + [Fact] + public void A_malformed_property_set_projects_to_nothing_rather_than_throwing() + { + var set = new Payload.Types.PropertySet(); + set.Keys.Add("orphan"); // no matching value + + SparkplugFlattener.ProjectProperties(set).ShouldBeEmpty(); + SparkplugFlattener.ProjectProperties(null).ShouldBeEmpty(); + } + + // ---- helpers ---------------------------------------------------------------------------------- + + private static IReadOnlyList Flatten(Payload.Types.DataSet dataSet) + { + SparkplugFlattener.TryFlattenDataSet(dataSet, out var members).ShouldBeTrue(); + return members; + } + + private static Payload.Types.Template Template(params Payload.Types.Metric[] members) + { + var template = new Payload.Types.Template { Version = "1", IsDefinition = false }; + template.Metrics.AddRange(members); + return template; + } + + private static Payload.Types.Metric Member( + string name, TahuDataType dataType, Action setValue) + { + var metric = new Payload.Types.Metric { Name = name, Datatype = (uint)dataType }; + setValue(metric); + return metric; + } + + private static Payload.Types.Metric MemberTemplate(string name, Payload.Types.Template value) => + new() { Name = name, Datatype = (uint)TahuDataType.Template, TemplateValue = value }; + + private static Payload.Types.DataSet.Types.DataSetValue Cell( + Action setValue) + { + var cell = new Payload.Types.DataSet.Types.DataSetValue(); + setValue(cell); + return cell; + } + + private static Payload.Types.DataSet DataSet( + string[] columns, + TahuDataType[] types, + Payload.Types.DataSet.Types.DataSetValue[][] rows) + { + var dataSet = new Payload.Types.DataSet { NumOfColumns = (ulong)columns.Length }; + dataSet.Columns.AddRange(columns); + dataSet.Types_.AddRange(types.Select(t => (uint)t)); + foreach (var cells in rows) + { + var row = new Payload.Types.DataSet.Types.Row(); + row.Elements.AddRange(cells); + dataSet.Rows.Add(row); + } + + return dataSet; + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/SparkplugGoldenPayloads.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/SparkplugGoldenPayloads.cs index 29b0544f..e052f7e4 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/SparkplugGoldenPayloads.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/SparkplugGoldenPayloads.cs @@ -132,8 +132,9 @@ internal static class SparkplugGoldenPayloads IntValue = unchecked((uint)CountBirthValue), }); - // DataSet is explicitly out of scope for v1 — present here so "unsupported" is a decoded, - // testable outcome rather than an exception on the MQTT dispatcher thread. + // A single-column, single-row DataSet. Gitea #515 flattens it to the addressable member + // `Config/col0/0`; before that it decoded as an explicit "unsupported" kind. Kept in the + // golden vector either way — it is the only structured metric on the wire-format path. var dataSet = new Payload.Types.DataSet { NumOfColumns = 1UL }; dataSet.Columns.Add("col0"); dataSet.Types_.Add((uint)DataType.Int32); diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/SparkplugStructuredIngestTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/SparkplugStructuredIngestTests.cs new file mode 100644 index 00000000..d99be5df --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/SparkplugStructuredIngestTests.cs @@ -0,0 +1,283 @@ +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; + +/// +/// End-to-end ingest of flattened structured metrics (Gitea #515): a tag authored against +/// Motor/Speed is fed by a DBIRTH/DDATA carrying a Template metric named +/// Motor. +/// +/// +/// pins the mapping in isolation; this file pins that the +/// chain actually connects — flatten → alias-table binding → authored-tag lookup → publish. Those +/// are four separate places a member name is composed, and a mismatch in any one of them produces +/// a driver that decodes structured metrics perfectly and feeds nothing. +/// +[Trait("Category", "Unit")] +public sealed class SparkplugStructuredIngestTests +{ + private const string Group = "Plant1"; + private const string Node = "EdgeA"; + private const string Device = "Filler1"; + private const ulong MotorAlias = 20UL; + private const ulong TableAlias = 21UL; + private const uint StatusGood = 0x00000000u; + + private const string SpeedPath = "Plant/Mqtt/spb/MotorSpeed"; + private const string TempCellPath = "Plant/Mqtt/spb/TempCell0"; + + [Fact] + public async Task A_template_member_feeds_the_tag_authored_against_its_flattened_name() + { + // The whole point of #515, asserted end to end. `Motor` is what the wire carries; `Motor/Speed` + // is what an operator authors, and nothing on the wire is ever called that. + var (ing, fired) = await NewIngestorAsync(Tag(SpeedPath, Blob("Motor/Speed", "Float64"))); + + ing.Dispatch(Topic(SparkplugMessageType.NBIRTH), Nbirth()); + ing.Dispatch(Topic(SparkplugMessageType.DBIRTH, Device), Birth(seq: 1, MotorTemplateMetric(speed: 1500, running: true))); + + fired.ShouldContainKey(SpeedPath); + fired[SpeedPath].Value.ShouldBe(1500d); + fired[SpeedPath].StatusCode.ShouldBe(StatusGood); + } + + [Fact] + public async Task A_ddata_template_carrying_only_the_parents_alias_still_reaches_the_member_tag() + { + // The load-bearing case, and the one a naive implementation gets wrong. A DDATA metric carries + // an ALIAS and no name — so the member names cannot be composed until the alias table has + // resolved the parent's name. Flattening at decode time alone is not enough. + var (ing, fired) = await NewIngestorAsync(Tag(SpeedPath, Blob("Motor/Speed", "Float64"))); + + ing.Dispatch(Topic(SparkplugMessageType.NBIRTH), Nbirth()); + ing.Dispatch(Topic(SparkplugMessageType.DBIRTH, Device), Birth(seq: 1, MotorTemplateMetric(speed: 1500, running: true))); + fired.Clear(); + + // seq 2, alias only — no Name on the metric at all. + var data = new Payload { Seq = 2UL, Timestamp = 1721822405000UL }; + data.Metrics.Add(new Payload.Types.Metric + { + Alias = MotorAlias, + TemplateValue = TemplateOf(speed: 1750, running: false), + }); + + ing.Dispatch(Topic(SparkplugMessageType.DDATA, Device), Decode(data)); + + fired.ShouldContainKey(SpeedPath); + fired[SpeedPath].Value.ShouldBe(1750d); + } + + [Fact] + public async Task A_dataset_cell_feeds_the_tag_authored_against_its_indexed_name() + { + var (ing, fired) = await NewIngestorAsync(Tag(TempCellPath, Blob("Table/Temp/0", "Float64"))); + + ing.Dispatch(Topic(SparkplugMessageType.NBIRTH), Nbirth()); + ing.Dispatch(Topic(SparkplugMessageType.DBIRTH, Device), Birth(seq: 1, TableDataSetMetric(20.5, 21.5))); + + fired.ShouldContainKey(TempCellPath); + fired[TempCellPath].Value.ShouldBe(20.5d); + } + + [Fact] + public async Task A_member_that_carries_no_datatype_falls_back_to_the_birth_binding() + { + // A DDATA template is not obliged to re-declare its members' datatypes. Guessing from the CLR + // type of the wire value would read a signed Int32 of -42 out of its unsigned field as + // 4294967254 — plausible, Good quality, and wrong. + var (ing, fired) = await NewIngestorAsync(Tag(SpeedPath, Blob("Motor/Offset", "Int32"))); + + var birthTemplate = new Payload.Types.Template { Version = "1", IsDefinition = false }; + birthTemplate.Metrics.Add(new Payload.Types.Metric + { + Name = "Offset", Datatype = (uint)TahuDataType.Int32, IntValue = unchecked((uint)-42), + }); + + ing.Dispatch(Topic(SparkplugMessageType.NBIRTH), Nbirth()); + ing.Dispatch(Topic(SparkplugMessageType.DBIRTH, Device), Birth(seq: 1, new Payload.Types.Metric + { + Name = "Motor", Alias = MotorAlias, Datatype = (uint)TahuDataType.Template, TemplateValue = birthTemplate, + })); + + fired[SpeedPath].Value.ShouldBe(-42); + fired.Clear(); + + // DDATA member with NO datatype of its own. + var dataTemplate = new Payload.Types.Template { Version = "1", IsDefinition = false }; + dataTemplate.Metrics.Add(new Payload.Types.Metric { Name = "Offset", IntValue = unchecked((uint)-7) }); + + var data = new Payload { Seq = 2UL, Timestamp = 1721822405000UL }; + data.Metrics.Add(new Payload.Types.Metric { Alias = MotorAlias, TemplateValue = dataTemplate }); + ing.Dispatch(Topic(SparkplugMessageType.DDATA, Device), Decode(data)); + + // -7, not 4294967289: the birth's Int32 binding drove ReinterpretSigned. + fired[SpeedPath].Value.ShouldBe(-7); + } + + [Fact] + public async Task A_tag_authored_against_the_structured_parent_itself_is_never_fed() + { + // The parent has no value — its members do. Feeding it anything would require choosing which + // member "the" value is, which is exactly the guess #515 refuses to make. + var (ing, fired) = await NewIngestorAsync(Tag(SpeedPath, Blob("Motor", "Float64"))); + + ing.Dispatch(Topic(SparkplugMessageType.NBIRTH), Nbirth()); + ing.Dispatch(Topic(SparkplugMessageType.DBIRTH, Device), Birth(seq: 1, MotorTemplateMetric(speed: 1500, running: true))); + + fired.ShouldNotContainKey(SpeedPath); + } + + [Fact] + public async Task An_unflattenable_structured_metric_still_feeds_nothing_rather_than_a_guess() + { + // Skip-and-warn survives — the explicit requirement on the issue. A template DEFINITION is the + // realistic instance of this: it names members and carries no values. + var definition = new Payload.Types.Template { Version = "1", IsDefinition = true }; + definition.Metrics.Add(new Payload.Types.Metric { Name = "Speed", Datatype = (uint)TahuDataType.Double }); + + var (ing, fired) = await NewIngestorAsync(Tag(SpeedPath, Blob("Motor/Speed", "Float64"))); + + ing.Dispatch(Topic(SparkplugMessageType.NBIRTH), Nbirth()); + ing.Dispatch(Topic(SparkplugMessageType.DBIRTH, Device), Birth(seq: 1, new Payload.Types.Metric + { + Name = "Motor", Alias = MotorAlias, Datatype = (uint)TahuDataType.Template, TemplateValue = definition, + })); + + fired.ShouldNotContainKey(SpeedPath); + } + + [Fact] + public async Task The_rediscovery_signature_tracks_members_not_the_parent_metric_name() + { + // A UDT gaining or losing a member leaves the PARENT metric's name identical, so a gate built + // on raw wire names would never move — and the operator would never be told the tree changed, + // which for a UDT is the change that matters most. + var (ing, _) = await NewIngestorAsync(Tag(SpeedPath, Blob("Motor/Speed", "Float64"))); + + var observed = new List>(); + ing.BirthObserved += (_, e) => observed.Add(e.MetricNames); + + ing.Dispatch(Topic(SparkplugMessageType.NBIRTH), Nbirth()); + ing.Dispatch(Topic(SparkplugMessageType.DBIRTH, Device), Birth(seq: 1, MotorTemplateMetric(speed: 1, running: true))); + + var deviceBirth = observed[^1]; + deviceBirth.ShouldBe(["Motor/Speed", "Motor/Running"], ignoreOrder: true); + deviceBirth.ShouldNotContain("Motor"); + } + + // ---- helpers --------------------------------------------------------------------------------- + + /// + /// A registered and subscribed ingestor. The subscribe step is not optional scaffolding: + /// SparkplugIngestor.Raise only fires OnDataChange for a RawPath that holds a + /// subscription handle, so an unsubscribed ingestor caches values silently and every assertion + /// in this file would read as "no value" whether or not the flattening worked. + /// + private static async Task<(SparkplugIngestor Ingestor, Dictionary Fired)> NewIngestorAsync( + params RawTagEntry[] tags) + { + var options = new MqttDriverOptions + { + Mode = MqttMode.SparkplugB, + Sparkplug = new MqttSparkplugOptions { GroupId = Group, RequestRebirthOnGap = false }, + RawTags = tags, + }; + + var ingestor = new SparkplugIngestor( + options, + "drv-1", + subscribeTransport: new GrantingSubscribeTransport(), + publishTransport: null, + logger: null, + cache: new LastValueCache()); + ingestor.Register(tags); + await ingestor.SubscribeAsync( + [.. ingestor.AuthoredRawPaths], TimeSpan.Zero, TestContext.Current.CancellationToken); + + var fired = new Dictionary(StringComparer.Ordinal); + ingestor.OnDataChange += (_, e) => fired[e.FullReference] = e.Snapshot; + return (ingestor, fired); + } + + /// A SUBACK that grants everything; the tests here are about values, not subscribe policy. + private sealed class GrantingSubscribeTransport : IMqttSubscribeTransport + { + public Task> SubscribeAsync( + IReadOnlyList filters, CancellationToken cancellationToken) + => Task.FromResult>( + [.. filters.Select(f => new MqttSubscribeOutcome(f.Topic, true, "GrantedQoS1"))]); + } + + private static RawTagEntry Tag(string rawPath, string tagConfig) => new(rawPath, tagConfig, WriteIdempotent: false); + + private static string Blob(string metric, string dataType) => + $$"""{"groupId":"{{Group}}","edgeNodeId":"{{Node}}","deviceId":"{{Device}}","metricName":"{{metric}}","dataType":"{{dataType}}"}"""; + + 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()); + + private static SparkplugPayload Nbirth() + { + var payload = new Payload { Seq = 0UL, Timestamp = 1721822400000UL }; + payload.Metrics.Add(new Payload.Types.Metric + { + Name = "bdSeq", Datatype = (uint)TahuDataType.Uint64, LongValue = 1UL, + }); + return Decode(payload); + } + + private static SparkplugPayload Birth(ulong seq, params Payload.Types.Metric[] metrics) + { + var payload = new Payload { Seq = seq, Timestamp = 1721822400000UL }; + payload.Metrics.AddRange(metrics); + return Decode(payload); + } + + private static Payload.Types.Template TemplateOf(double speed, bool running) + { + var template = new Payload.Types.Template { Version = "1", IsDefinition = false }; + template.Metrics.Add(new Payload.Types.Metric + { + Name = "Speed", Datatype = (uint)TahuDataType.Double, DoubleValue = speed, + }); + template.Metrics.Add(new Payload.Types.Metric + { + Name = "Running", Datatype = (uint)TahuDataType.Boolean, BooleanValue = running, + }); + return template; + } + + private static Payload.Types.Metric MotorTemplateMetric(double speed, bool running) => new() + { + Name = "Motor", + Alias = MotorAlias, + Datatype = (uint)TahuDataType.Template, + TemplateValue = TemplateOf(speed, running), + }; + + private static Payload.Types.Metric TableDataSetMetric(params double[] temps) + { + var dataSet = new Payload.Types.DataSet { NumOfColumns = 1UL }; + dataSet.Columns.Add("Temp"); + dataSet.Types_.Add((uint)TahuDataType.Double); + foreach (var temp in temps) + { + var row = new Payload.Types.DataSet.Types.Row(); + row.Elements.Add(new Payload.Types.DataSet.Types.DataSetValue { DoubleValue = temp }); + dataSet.Rows.Add(row); + } + + return new Payload.Types.Metric + { + Name = "Table", Alias = TableAlias, Datatype = (uint)TahuDataType.DataSet, DatasetValue = dataSet, + }; + } +}