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

feat(mqtt): RebirthRequester NCMD encode

Browse Source 
Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW
This commit is contained in:
Joseph Doherty
2026-07-24 21:40:43 -04:00
parent 3fdf24659f
commit 31a98a1bf5
2 changed files with 354 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/RebirthRequester.cs
+146
View File
@@ -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;
/// <summary>
/// The one seam through which a Sparkplug NCMD is published — <see cref="MqttConnection"/> 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.
/// </summary>
public interface IMqttPublishTransport
{
/// <summary>
/// Publishes one application message. Implementations must be bounded — a broker that accepts
/// PUBLISH and never completes it (QoS &gt; 0 awaiting PUBACK) must fail at a deadline, not
/// hang; <see cref="RebirthRequester.RequestAsync"/> supplies one, but an implementation used
/// directly by a future caller must not rely on that.
/// </summary>
/// <param name="topic">The concrete topic to publish on.</param>
/// <param name="payload">The message body.</param>
/// <param name="qos">Requested QoS, 0–2.</param>
/// <param name="retain">The MQTT <c>retain</c> flag.</param>
/// <param name="cancellationToken">Caller cancellation.</param>
Task PublishAsync(string topic, byte[] payload, int qos, bool retain, CancellationToken cancellationToken);
}
/// <summary>
/// Encodes and publishes a Sparkplug B rebirth request — an NCMD carrying the well-known
/// <c>Node Control/Rebirth</c> boolean metric, addressed to one edge node's command topic. See
/// Sparkplug B v3.0 spec §6/§7 and design doc §3.6.
/// </summary>
/// <remarks>
/// <para>
/// <b><see cref="Build"/> is pure.</b> 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 <see cref="RequestAsync"/> overload.
/// </para>
/// <para>
/// <b>No <c>seq</c>.</b> 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 <i>that node's</i> 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.
/// </para>
/// <para>
/// <b>QoS 0, retain <see langword="false"/> — always.</b> 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). <see langword="retain"/> is the one
/// that matters operationally: a retained NCMD would be replayed by the broker to the edge node
/// on <i>every</i> 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
/// <c>MqttDriverBrowser</c>/NDEATH-as-Will for the sibling failure shape this type must not
/// introduce on the publish side.
/// </para>
/// <para>
/// <b><c>Datatype</c> and both timestamps are set.</b> <c>Node Control/Rebirth</c> is Boolean;
/// setting <see cref="Payload.Types.Metric.Datatype"/> explicitly (rather than leaving it for the
/// node to infer) and stamping both the metric and payload <c>timestamp</c> 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.
/// </para>
/// </remarks>
public static class RebirthRequester
{
/// <summary>The well-known Sparkplug B node-control metric name that triggers a rebirth.</summary>
public const string RebirthMetricName = "Node Control/Rebirth";
/// <summary>
/// Default bounded deadline for <see cref="RequestAsync"/> when the caller does not supply one.
/// Deliberately this type's own value rather than <c>MqttDriverOptions.ConnectTimeoutSeconds</c>
/// — 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 <c>MqttConnection.SubscribeAsync</c> documents for its
/// own deadline).
/// </summary>
public static readonly TimeSpan DefaultRequestTimeout = TimeSpan.FromSeconds(5);
/// <summary>
/// Builds the wire bytes and target topic for a rebirth-request NCMD, without publishing it.
/// </summary>
/// <param name="groupId">The Sparkplug group id.</param>
/// <param name="edgeNodeId">The Sparkplug edge-node id to address the command to.</param>
/// <returns>The NCMD topic (<c>spBv1.0/{groupId}/NCMD/{edgeNodeId}</c>) and its encoded payload.</returns>
/// <exception cref="ArgumentException"><paramref name="groupId"/>/<paramref name="edgeNodeId"/> is null or empty.</exception>
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());
}
/// <summary>
/// Builds and publishes a rebirth-request NCMD under a bounded deadline. QoS 0, retain
/// <see langword="false"/> — see the type remarks for why both are fixed rather than
/// configurable.
/// </summary>
/// <param name="transport">The publish seam — the live connection, or a test fake.</param>
/// <param name="groupId">The Sparkplug group id.</param>
/// <param name="edgeNodeId">The Sparkplug edge-node id to address the command to.</param>
/// <param name="cancellationToken">Caller cancellation; linked with the publish deadline.</param>
/// <param name="timeout">
/// The publish deadline. Defaults to <see cref="DefaultRequestTimeout"/> when omitted or
/// non-positive.
/// </param>
/// <exception cref="ArgumentException"><paramref name="groupId"/>/<paramref name="edgeNodeId"/> is null or empty.</exception>
/// <exception cref="ArgumentNullException"><paramref name="transport"/> is null.</exception>
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);
}
}
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/RebirthRequesterTests.cs
+208
View File
@@ -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;
/// <summary>
/// Covers <see cref="RebirthRequester"/> (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.
/// </summary>
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));
}
/// <summary>
/// Falsifiability control for the test above: a metric named anything else must NOT satisfy the
/// assertion — proves the <c>ShouldContain</c> predicate is actually discriminating on the name,
/// not vacuously true because <see cref="SparkplugCodec.Decode"/> always returns something.
/// </summary>
[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));
}
/// <summary>
/// Falsifiability control: a boolean <c>false</c> must not satisfy the plan's assertion either —
/// proves the test is checking the metric's <i>value</i>, not merely its presence.
/// </summary>
[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<ArgumentException>(() => 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);
}
/// <summary>
/// A host-issued NCMD is not part of the edge node's own outbound sequence stream (Sparkplug
/// §6.4.1) — stamping a <c>seq</c> here would be meaningless at best. See the design rationale
/// on <see cref="RebirthRequester"/>.
/// </summary>
[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));
}
/// <summary>
/// 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 <see cref="RebirthRequester"/>).
/// </summary>
[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<ArgumentNullException>(
() => RebirthRequester.RequestAsync(null!, "Plant1", "EdgeA", TestContext.Current.CancellationToken));
/// <summary>
/// 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 <c>MqttConnection.SubscribeAsync</c>).
/// </summary>
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<OperationCanceledException>(() => 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<OperationCanceledException>(
() => RebirthRequester.RequestAsync(transport, "Plant1", "EdgeA", cts.Token));
}
}