diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/RebirthRequester.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/RebirthRequester.cs new file mode 100644 index 00000000..8f4451cb --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/RebirthRequester.cs @@ -0,0 +1,146 @@ +using Google.Protobuf; +using Org.Eclipse.Tahu.Protobuf; +using TahuDataType = Org.Eclipse.Tahu.Protobuf.DataType; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Sparkplug; + +/// +/// The one seam through which a Sparkplug NCMD is published — is the +/// production implementation once it grows a publish leg (Task 21); tests substitute a fake so the +/// encode/topic/QoS/retain contract is exercisable without a broker. Deliberately narrower than +/// "publish anything": a Sparkplug driver's only outbound traffic is NCMD/DCMD, so this is not a +/// general MQTT client abstraction in waiting. +/// +public interface IMqttPublishTransport +{ + /// + /// Publishes one application message. Implementations must be bounded — a broker that accepts + /// PUBLISH and never completes it (QoS > 0 awaiting PUBACK) must fail at a deadline, not + /// hang; supplies one, but an implementation used + /// directly by a future caller must not rely on that. + /// + /// The concrete topic to publish on. + /// The message body. + /// Requested QoS, 0–2. + /// The MQTT retain flag. + /// Caller cancellation. + Task PublishAsync(string topic, byte[] payload, int qos, bool retain, CancellationToken cancellationToken); +} + +/// +/// Encodes and publishes a Sparkplug B rebirth request — an NCMD carrying the well-known +/// Node Control/Rebirth boolean metric, addressed to one edge node's command topic. See +/// Sparkplug B v3.0 spec §6/§7 and design doc §3.6. +/// +/// +/// +/// is pure. It does no I/O and touches no transport, so it is +/// trivially unit-testable and safe to call from anywhere (the ingest state machine on a +/// detected gap, the browser on an operator click) without pulling in a live connection. The +/// bounded, transport-facing publish is the separate overload. +/// +/// +/// No seq. Sparkplug's sequence-number stream (§6.4.1) belongs to the messages an +/// edge node itself publishes (NBIRTH/NDATA/DBIRTH/DDATA/NDEATH/DDEATH) — it is how a host +/// detects a gap in that node's outbound stream. An NCMD flows the other direction, host +/// → node, and is not a member of the node's sequence at all: stamping one on here would not be +/// "wrong per spec" so much as meaningless, and a strict edge-node implementation that validated +/// it against its own counter would have grounds to reject the command outright. +/// +/// +/// QoS 0, retain — always. NCMD is ordinary Sparkplug command +/// traffic, published at the same QoS 0 the spec uses for NDATA/DDATA/NBIRTH/DBIRTH (only +/// NDEATH — the broker-issued Will — and STATE use QoS 1). is the one +/// that matters operationally: a retained NCMD would be replayed by the broker to the edge node +/// on every future subscribe — including the node's own reconnect — driving it into a +/// permanent rebirth loop from a single stale command. See the carry-forward landmine on +/// MqttDriverBrowser/NDEATH-as-Will for the sibling failure shape this type must not +/// introduce on the publish side. +/// +/// +/// Datatype and both timestamps are set. Node Control/Rebirth is Boolean; +/// setting explicitly (rather than leaving it for the +/// node to infer) and stamping both the metric and payload timestamp matches what a +/// well-behaved command payload carries per spec §6.4.13 and what the Task-25 simulator (and any +/// real Cirrus-Link-derived edge node) expects to validate against. An edge node that treats a +/// missing/ambiguous field as malformed and silently drops the command would otherwise leave the +/// driver retrying forever with no visible error — the whole point of a rebirth request is to +/// recover from exactly that kind of silent desync, so the request itself must not risk being +/// the next cause of one. +/// +/// +public static class RebirthRequester +{ + /// The well-known Sparkplug B node-control metric name that triggers a rebirth. + public const string RebirthMetricName = "Node Control/Rebirth"; + + /// + /// Default bounded deadline for when the caller does not supply one. + /// Deliberately this type's own value rather than MqttDriverOptions.ConnectTimeoutSeconds + /// — that knob already governs every MQTTnet connect/subscribe operation, and repurposing it + /// would couple the rebirth-publish budget to the connect budget in a way neither side could + /// tune independently (the same reasoning MqttConnection.SubscribeAsync documents for its + /// own deadline). + /// + public static readonly TimeSpan DefaultRequestTimeout = TimeSpan.FromSeconds(5); + + /// + /// Builds the wire bytes and target topic for a rebirth-request NCMD, without publishing it. + /// + /// The Sparkplug group id. + /// The Sparkplug edge-node id to address the command to. + /// The NCMD topic (spBv1.0/{groupId}/NCMD/{edgeNodeId}) and its encoded payload. + /// / is null or empty. + public static (string Topic, byte[] Bytes) Build(string groupId, string edgeNodeId) + { + // Validates both ids and does the topic assembly — see the "don't hand-concatenate" remark on + // SparkplugTopic.Format itself. + var topic = SparkplugTopic.Format(groupId, SparkplugMessageType.NCMD, edgeNodeId); + + var timestampMs = (ulong)DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(); + + var payload = new Payload { Timestamp = timestampMs }; + payload.Metrics.Add(new Payload.Types.Metric + { + Name = RebirthMetricName, + Datatype = (uint)TahuDataType.Boolean, + Timestamp = timestampMs, + BooleanValue = true, + }); + + return (topic, payload.ToByteArray()); + } + + /// + /// Builds and publishes a rebirth-request NCMD under a bounded deadline. QoS 0, retain + /// — see the type remarks for why both are fixed rather than + /// configurable. + /// + /// The publish seam — the live connection, or a test fake. + /// The Sparkplug group id. + /// The Sparkplug edge-node id to address the command to. + /// Caller cancellation; linked with the publish deadline. + /// + /// The publish deadline. Defaults to when omitted or + /// non-positive. + /// + /// / is null or empty. + /// is null. + public static async Task RequestAsync( + IMqttPublishTransport transport, + string groupId, + string edgeNodeId, + CancellationToken cancellationToken, + TimeSpan? timeout = null) + { + ArgumentNullException.ThrowIfNull(transport); + + var (topic, bytes) = Build(groupId, edgeNodeId); + + var deadlineSpan = timeout is { } t && t > TimeSpan.Zero ? t : DefaultRequestTimeout; + using var deadline = new CancellationTokenSource(deadlineSpan); + using var linked = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, deadline.Token); + + await transport.PublishAsync(topic, bytes, qos: 0, retain: false, linked.Token).ConfigureAwait(false); + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/RebirthRequesterTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/RebirthRequesterTests.cs new file mode 100644 index 00000000..79e7fc5b --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/RebirthRequesterTests.cs @@ -0,0 +1,208 @@ +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; + +/// +/// Covers (Task 20) — the NCMD encode + bounded publish seam that +/// Task 21 (ingest gap/unknown-alias/data-before-birth detection) and Task 23 (the browser's +/// operator-triggered scoped rebirth) both call into. +/// +public sealed class RebirthRequesterTests +{ + // ----------------------------------------------------------------------------------------- + // Build — pure encode. The plan's own test, verbatim. + // ----------------------------------------------------------------------------------------- + + [Fact] + public void BuildRebirthNcmd_EncodesControlMetric_AndTopic() + { + var (topic, bytes) = RebirthRequester.Build("Plant1", "EdgeA"); + + topic.ShouldBe("spBv1.0/Plant1/NCMD/EdgeA"); + var p = SparkplugCodec.Decode(bytes); + p.Metrics.ShouldContain(m => m.Name == "Node Control/Rebirth" && Equals(m.Value, true)); + } + + /// + /// Falsifiability control for the test above: a metric named anything else must NOT satisfy the + /// assertion — proves the ShouldContain predicate is actually discriminating on the name, + /// not vacuously true because always returns something. + /// + [Fact] + public void BuildRebirthNcmd_DoesNotContainAWrongMetricName() + { + var (_, bytes) = RebirthRequester.Build("Plant1", "EdgeA"); + var p = SparkplugCodec.Decode(bytes); + + p.Metrics.ShouldNotContain(m => m.Name == "Node Control/NotRebirth" && Equals(m.Value, true)); + } + + /// + /// Falsifiability control: a boolean false must not satisfy the plan's assertion either — + /// proves the test is checking the metric's value, not merely its presence. + /// + [Fact] + public void BuildRebirthNcmd_ValueIsTrue_NotFalse() + { + var (_, bytes) = RebirthRequester.Build("Plant1", "EdgeA"); + var p = SparkplugCodec.Decode(bytes); + + p.Metrics.ShouldNotContain(m => m.Name == "Node Control/Rebirth" && Equals(m.Value, false)); + } + + [Fact] + public void Build_DifferentGroupAndNode_ProducesMatchingTopic() + { + var (topic, _) = RebirthRequester.Build("Line7", "Filler42"); + topic.ShouldBe("spBv1.0/Line7/NCMD/Filler42"); + } + + [Theory] + [InlineData(null, "EdgeA")] + [InlineData("", "EdgeA")] + [InlineData("Plant1", null)] + [InlineData("Plant1", "")] + public void Build_MissingGroupOrNode_Throws(string? groupId, string? edgeNodeId) + => Should.Throw(() => RebirthRequester.Build(groupId!, edgeNodeId!)); + + // ----------------------------------------------------------------------------------------- + // Encoded metric shape — datatype + no bogus seq + // ----------------------------------------------------------------------------------------- + + [Fact] + public void Build_MetricCarriesBooleanDatatype() + { + var (_, bytes) = RebirthRequester.Build("Plant1", "EdgeA"); + var p = SparkplugCodec.Decode(bytes); + + var metric = p.Metrics.Single(m => m.Name == RebirthRequester.RebirthMetricName); + metric.DataType.ShouldBe(TahuDataType.Boolean); + } + + /// + /// A host-issued NCMD is not part of the edge node's own outbound sequence stream (Sparkplug + /// §6.4.1) — stamping a seq here would be meaningless at best. See the design rationale + /// on . + /// + [Fact] + public void Build_PayloadCarriesNoSeq() + { + var (_, bytes) = RebirthRequester.Build("Plant1", "EdgeA"); + var p = SparkplugCodec.Decode(bytes); + + p.Seq.ShouldBeNull(); + } + + [Fact] + public void Build_PayloadAndMetricCarryATimestamp() + { + var (_, bytes) = RebirthRequester.Build("Plant1", "EdgeA"); + var p = SparkplugCodec.Decode(bytes); + + p.TimestampMs.ShouldNotBeNull(); + var metric = p.Metrics.Single(m => m.Name == RebirthRequester.RebirthMetricName); + metric.TimestampMs.ShouldNotBeNull(); + } + + // ----------------------------------------------------------------------------------------- + // RequestAsync — the bounded publish seam + // ----------------------------------------------------------------------------------------- + + private sealed class RecordingTransport : IMqttPublishTransport + { + public int PublishCount { get; private set; } + public string? LastTopic { get; private set; } + public byte[]? LastPayload { get; private set; } + public int LastQos { get; private set; } = -1; + public bool LastRetain { get; private set; } + + public Task PublishAsync(string topic, byte[] payload, int qos, bool retain, CancellationToken cancellationToken) + { + PublishCount++; + LastTopic = topic; + LastPayload = payload; + LastQos = qos; + LastRetain = retain; + return Task.CompletedTask; + } + } + + [Fact] + public async Task RequestAsync_PublishesExactlyOnce_WithBuiltTopicAndBytes() + { + var transport = new RecordingTransport(); + + await RebirthRequester.RequestAsync(transport, "Plant1", "EdgeA", TestContext.Current.CancellationToken); + + transport.PublishCount.ShouldBe(1); + transport.LastTopic.ShouldBe("spBv1.0/Plant1/NCMD/EdgeA"); + var decoded = SparkplugCodec.Decode(transport.LastPayload!); + decoded.Metrics.ShouldContain(m => m.Name == "Node Control/Rebirth" && Equals(m.Value, true)); + } + + /// + /// Retain MUST be false — a retained NCMD replays to the edge node on every future + /// subscribe/reconnect, driving a permanent rebirth loop from one stale command (see the + /// carry-forward landmine remarks on ). + /// + [Fact] + public async Task RequestAsync_NeverRetains() + { + var transport = new RecordingTransport(); + await RebirthRequester.RequestAsync(transport, "Plant1", "EdgeA", TestContext.Current.CancellationToken); + + transport.LastRetain.ShouldBeFalse(); + } + + [Fact] + public async Task RequestAsync_UsesQos0() + { + var transport = new RecordingTransport(); + await RebirthRequester.RequestAsync(transport, "Plant1", "EdgeA", TestContext.Current.CancellationToken); + + transport.LastQos.ShouldBe(0); + } + + [Fact] + public async Task RequestAsync_NullTransport_Throws() + => await Should.ThrowAsync( + () => RebirthRequester.RequestAsync(null!, "Plant1", "EdgeA", TestContext.Current.CancellationToken)); + + /// + /// A transport that hangs forever must not be able to park the caller — the cross-cutting rule + /// every network call in this driver follows (mirrors MqttConnection.SubscribeAsync). + /// + private sealed class HangingTransport : IMqttPublishTransport + { + public Task PublishAsync(string topic, byte[] payload, int qos, bool retain, CancellationToken cancellationToken) + => Task.Delay(Timeout.InfiniteTimeSpan, cancellationToken); + } + + [Fact] + public async Task RequestAsync_BoundedByDeadline_ThrowsRatherThanHangs() + { + var transport = new HangingTransport(); + + await Should.ThrowAsync(() => RebirthRequester.RequestAsync( + transport, + "Plant1", + "EdgeA", + TestContext.Current.CancellationToken, + timeout: TimeSpan.FromMilliseconds(50))); + } + + [Fact] + public async Task RequestAsync_CallerCancellation_Propagates() + { + var transport = new HangingTransport(); + using var cts = new CancellationTokenSource(); + await cts.CancelAsync(); + + await Should.ThrowAsync( + () => RebirthRequester.RequestAsync(transport, "Plant1", "EdgeA", cts.Token)); + } +}