feat(mqtt): wire sparkplug.birthObservationWindowSeconds into the browse surface (R06:M-4)
The authored window was persisted and consumed by nothing. It now (clamped 1-120 s) extends the Sparkplug OpenBudget, bounds a passive UntilStable-style settle on open (return after a 1 s quiet spell in NBIRTH/DBIRTH traffic), and bounds how long an operator rebirth request waits for the births it asked for before returning, so the re-rendered tree carries them. A caller deadline during the wait still reports the publish. Refs #744 Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
This commit is contained in:
@@ -117,7 +117,7 @@ driver are pure organisational grouping in the RawPath.
|
||||
"HostId": "", // Host Application identity — spBv1.0/STATE/{HostId}
|
||||
"ActAsPrimaryHost": false, // claim the primary host id (Primary-gated; see below)
|
||||
"RequestRebirthOnGap": true, // a seq gap asks the edge node to re-announce (NCMD)
|
||||
"BirthObservationWindowSeconds": 15 // how long discovery collects births before calling it stable
|
||||
"BirthObservationWindowSeconds": 15 // how long the browse picker collects births before calling them stable
|
||||
}
|
||||
}
|
||||
```
|
||||
@@ -128,7 +128,7 @@ driver are pure organisational grouping in the RawPath.
|
||||
| `HostId` | `""` | Sparkplug Host Application identity — the `spBv1.0/STATE/{HostId}` topic. Optional for observation; **required** to claim. Set without `ActAsPrimaryHost`, the driver only observes |
|
||||
| `ActAsPrimaryHost` | `false` | Claim the Sparkplug Primary Host Application identity. **Primary-gated** — only the node holding the redundancy Primary role claims it. See [Primary host](#primary-host-actasprimaryhost) |
|
||||
| `RequestRebirthOnGap` | `true` | On a detected sequence-number gap, publish a rebirth-request NCMD instead of running on stale metric state |
|
||||
| `BirthObservationWindowSeconds` | `15` | Discovery's birth-collection window. Must be ≥ 1 |
|
||||
| `BirthObservationWindowSeconds` | `15` | The Sparkplug browse picker's birth-collection window, clamped to 1–120 s (R06:M-4). The open budget grows by it; the open lingers while NBIRTH/DBIRTH traffic is still arriving (returning after a 1 s quiet spell — an idle plant costs 1 s); and an operator **Request rebirth** waits, bounded by it, for the births it asked for to arrive and go quiet for 500 ms, so the tree re-rendered afterwards carries them. A caller deadline during that wait still reports the publish. The runtime driver does not read it (its `DiscoverAsync` has no production caller — `SupportsOnlineDiscovery` is false) |
|
||||
|
||||
`MqttDriverForm` authors all five in Sparkplug mode. The sub-object is **merged, never replaced** (a
|
||||
key a newer driver adds inside it survives an older AdminUI), and in `Plain` mode it is not touched at
|
||||
|
||||
@@ -149,6 +149,9 @@ internal sealed class MqttBrowseSession : IRebirthCapableBrowseSession
|
||||
/// </summary>
|
||||
private static readonly UTF8Encoding Utf8Strict = new(encoderShouldEmitUTF8Identifier: false, throwOnInvalidBytes: true);
|
||||
|
||||
/// <summary>The quiet spell that ends a rebirth request's wait once its births have started arriving.</summary>
|
||||
private static readonly TimeSpan RebirthQuietInterval = TimeSpan.FromMilliseconds(500);
|
||||
|
||||
/// <summary>Bound on the courtesy DISCONNECT issued at dispose — teardown never hangs on a dead broker.</summary>
|
||||
private static readonly TimeSpan DisconnectTimeout = TimeSpan.FromSeconds(5);
|
||||
|
||||
@@ -174,6 +177,7 @@ internal sealed class MqttBrowseSession : IRebirthCapableBrowseSession
|
||||
private readonly ConcurrentQueue<string> _publishedTopics = new();
|
||||
|
||||
private int _disposed;
|
||||
private long _birthCount;
|
||||
private long _lastUsedTicksUtc = DateTime.UtcNow.Ticks;
|
||||
private int _nodeCount;
|
||||
private int _oversized;
|
||||
@@ -195,9 +199,16 @@ internal sealed class MqttBrowseSession : IRebirthCapableBrowseSession
|
||||
/// <param name="client">The connected MQTT client; disconnected and disposed by <see cref="DisposeAsync"/>.</param>
|
||||
/// <param name="logger">Optional logger; never receives credentials or payload bodies.</param>
|
||||
/// <param name="nodeCap">Cap on recorded nodes; defaults to <see cref="DefaultNodeCap"/>.</param>
|
||||
internal MqttBrowseSession(MqttMode mode, IMqttClient? client, ILogger? logger, int nodeCap = DefaultNodeCap)
|
||||
/// <param name="birthWindow">
|
||||
/// The Sparkplug birth-observation window: how long <see cref="RequestRebirthAsync"/> waits for the
|
||||
/// births it asked for. <see cref="TimeSpan.Zero"/> (the test default) returns as soon as the NCMDs
|
||||
/// are published.
|
||||
/// </param>
|
||||
internal MqttBrowseSession(
|
||||
MqttMode mode, IMqttClient? client, ILogger? logger, int nodeCap = DefaultNodeCap, TimeSpan birthWindow = default)
|
||||
{
|
||||
Mode = mode;
|
||||
BirthWindow = birthWindow;
|
||||
_client = client;
|
||||
_logger = logger;
|
||||
_nodeCap = nodeCap;
|
||||
@@ -211,6 +222,56 @@ internal sealed class MqttBrowseSession : IRebirthCapableBrowseSession
|
||||
/// </summary>
|
||||
internal int PublishCountForTest => Volatile.Read(ref _publishCount);
|
||||
|
||||
/// <summary>The Sparkplug birth-observation window this session was opened with (R06:M-4).</summary>
|
||||
internal TimeSpan BirthWindow { get; }
|
||||
|
||||
/// <summary>How many NBIRTH/DBIRTH messages this session has observed — the settle loop's signal.</summary>
|
||||
internal long BirthCount => Interlocked.Read(ref _birthCount);
|
||||
|
||||
/// <summary>
|
||||
/// Waits until birth traffic has gone quiet for one <paramref name="quiet"/> interval, bounded by
|
||||
/// <paramref name="window"/> — the <c>UntilStable</c> cadence of the birth-observation window.
|
||||
/// </summary>
|
||||
/// <param name="birthCount">Reads the running birth count.</param>
|
||||
/// <param name="baseline">The count before the wait began.</param>
|
||||
/// <param name="awaitFirstBirth">
|
||||
/// When true, quiet only counts once at least one birth has arrived past <paramref name="baseline"/>
|
||||
/// — the shape a rebirth request needs, since its answer takes a broker round-trip to arrive.
|
||||
/// </param>
|
||||
/// <param name="quiet">The quiet spell that ends the wait.</param>
|
||||
/// <param name="window">The hard ceiling; non-positive returns at once.</param>
|
||||
/// <param name="cancellationToken">Cancels the wait.</param>
|
||||
/// <returns>A task that completes when the births have settled or the window has elapsed.</returns>
|
||||
internal static async Task SettleBirthsAsync(
|
||||
Func<long> birthCount,
|
||||
long baseline,
|
||||
bool awaitFirstBirth,
|
||||
TimeSpan quiet,
|
||||
TimeSpan window,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
if (window <= TimeSpan.Zero) return;
|
||||
|
||||
var clock = System.Diagnostics.Stopwatch.StartNew();
|
||||
var last = baseline;
|
||||
var sawBirth = !awaitFirstBirth;
|
||||
while (clock.Elapsed < window)
|
||||
{
|
||||
var step = window - clock.Elapsed;
|
||||
await Task.Delay(step < quiet ? step : quiet, cancellationToken).ConfigureAwait(false);
|
||||
|
||||
var now = birthCount();
|
||||
if (now != last)
|
||||
{
|
||||
last = now;
|
||||
sawBirth = true;
|
||||
continue;
|
||||
}
|
||||
|
||||
if (sawBirth) return;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>The topics this session published, in order — the discriminating view of the above.</summary>
|
||||
internal IReadOnlyList<string> PublishedTopicsForTest => [.. _publishedTopics];
|
||||
|
||||
@@ -557,6 +618,8 @@ internal sealed class MqttBrowseSession : IRebirthCapableBrowseSession
|
||||
if (!SparkplugTopic.TryParse(topic, out var parsed)) return;
|
||||
if (parsed.Type is not (SparkplugMessageType.NBIRTH or SparkplugMessageType.DBIRTH)) return;
|
||||
|
||||
Interlocked.Increment(ref _birthCount);
|
||||
|
||||
// The bound has to be on the whole message: unlike a plain payload, a birth must be parsed
|
||||
// entire to yield its metric list, and the parse runs on the shared dispatcher thread.
|
||||
if (payload.Length > MaxBirthPayloadBytes)
|
||||
@@ -898,6 +961,7 @@ internal sealed class MqttBrowseSession : IRebirthCapableBrowseSession
|
||||
ArgumentException.ThrowIfNullOrWhiteSpace(scope);
|
||||
|
||||
var targets = ResolveRebirthTargets(scope);
|
||||
var baseline = BirthCount;
|
||||
|
||||
foreach (var (groupId, edgeNodeId) in targets)
|
||||
{
|
||||
@@ -908,6 +972,20 @@ internal sealed class MqttBrowseSession : IRebirthCapableBrowseSession
|
||||
.ConfigureAwait(false);
|
||||
}
|
||||
|
||||
// The birth-observation window (R06:M-4): wait — bounded — for the births just asked for to arrive
|
||||
// and settle, so the tree the picker re-renders after this call already carries them. A courtesy,
|
||||
// never a failure: the NCMDs are out, so a caller deadline here still reports the publish.
|
||||
try
|
||||
{
|
||||
await SettleBirthsAsync(
|
||||
() => BirthCount, baseline, awaitFirstBirth: true, RebirthQuietInterval, BirthWindow, cancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
// Published; the caller simply stopped waiting for the answer.
|
||||
}
|
||||
|
||||
Touch();
|
||||
return targets.Count;
|
||||
}
|
||||
|
||||
@@ -39,6 +39,15 @@ public sealed class MqttDriverBrowser : IDriverBrowser
|
||||
/// <summary>Ceiling on the open budget — the picker must never hang on an unreachable broker.</summary>
|
||||
internal const int MaxOpenBudgetSeconds = 30;
|
||||
|
||||
/// <summary>Ceiling on the Sparkplug birth-observation window — an authored hour must not park a picker.</summary>
|
||||
internal const int MaxBirthWindowSeconds = 120;
|
||||
|
||||
/// <summary>
|
||||
/// How long the open waits for a quiet spell in birth traffic before handing the session back. An
|
||||
/// idle plant (births are never retained) costs exactly this, once.
|
||||
/// </summary>
|
||||
internal static readonly TimeSpan BirthQuietInterval = TimeSpan.FromSeconds(1);
|
||||
|
||||
/// <summary>Marks the transient browse identity in the broker's client-id/session logs.</summary>
|
||||
internal const string BrowseClientIdPrefix = "-browse-";
|
||||
|
||||
@@ -77,7 +86,8 @@ public sealed class MqttDriverBrowser : IDriverBrowser
|
||||
openCts.CancelAfter(OpenBudget(opts));
|
||||
|
||||
var client = new MqttClientFactory().CreateMqttClient();
|
||||
var session = new MqttBrowseSession(opts.Mode, client, _logger);
|
||||
var birthWindow = opts.Mode == MqttMode.SparkplugB ? BirthWindow(opts) : TimeSpan.Zero;
|
||||
var session = new MqttBrowseSession(opts.Mode, client, _logger, birthWindow: birthWindow);
|
||||
try
|
||||
{
|
||||
// Runs on MQTTnet's dispatcher: record and return, nothing else.
|
||||
@@ -100,6 +110,29 @@ public sealed class MqttDriverBrowser : IDriverBrowser
|
||||
.Build();
|
||||
await client.SubscribeAsync(subscribeOptions, openCts.Token).ConfigureAwait(false);
|
||||
|
||||
if (opts.Mode == MqttMode.SparkplugB)
|
||||
{
|
||||
// The birth-observation window (R06:M-4): births already in flight — an edge node mid-rebirth,
|
||||
// a primary-host flap — are collected until the traffic goes quiet, bounded by the window, so
|
||||
// the first RootAsync is not a snapshot taken in the middle of a burst. Passive: nothing is
|
||||
// published to provoke a birth. An idle plant costs one quiet interval.
|
||||
try
|
||||
{
|
||||
await MqttBrowseSession.SettleBirthsAsync(
|
||||
() => session.BirthCount,
|
||||
baseline: session.BirthCount,
|
||||
awaitFirstBirth: false,
|
||||
BirthQuietInterval,
|
||||
birthWindow,
|
||||
openCts.Token)
|
||||
.ConfigureAwait(false);
|
||||
}
|
||||
catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested)
|
||||
{
|
||||
// The open budget ran out mid-settle: what has been observed is still a usable session.
|
||||
}
|
||||
}
|
||||
|
||||
_logger.LogInformation(
|
||||
"AdminUI MQTT browse session opened against {Host}:{Port} in {Mode} mode observing "
|
||||
+ "'{Filter}' (read-only).",
|
||||
@@ -179,12 +212,30 @@ public sealed class MqttDriverBrowser : IDriverBrowser
|
||||
options with { CleanSession = true };
|
||||
|
||||
/// <summary>
|
||||
/// Whole-open bound (connect + subscribe), clamped to
|
||||
/// <see cref="MinOpenBudgetSeconds"/>–<see cref="MaxOpenBudgetSeconds"/> around the config's
|
||||
/// own connect deadline.
|
||||
/// Whole-open bound: connect + subscribe, clamped to
|
||||
/// <see cref="MinOpenBudgetSeconds"/>–<see cref="MaxOpenBudgetSeconds"/> around the config's own
|
||||
/// connect deadline — plus, in Sparkplug mode, the <see cref="BirthWindow"/> the open may spend
|
||||
/// collecting births.
|
||||
/// </summary>
|
||||
/// <param name="options">The browse configuration.</param>
|
||||
/// <returns>The clamped open budget.</returns>
|
||||
internal static TimeSpan OpenBudget(MqttDriverOptions options) =>
|
||||
TimeSpan.FromSeconds(Math.Clamp(options.ConnectTimeoutSeconds, MinOpenBudgetSeconds, MaxOpenBudgetSeconds));
|
||||
/// <returns>The open budget.</returns>
|
||||
internal static TimeSpan OpenBudget(MqttDriverOptions options)
|
||||
{
|
||||
var connect = TimeSpan.FromSeconds(
|
||||
Math.Clamp(options.ConnectTimeoutSeconds, MinOpenBudgetSeconds, MaxOpenBudgetSeconds));
|
||||
return options.Mode == MqttMode.SparkplugB ? connect + BirthWindow(options) : connect;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The authored <c>sparkplug.birthObservationWindowSeconds</c>, clamped to 1–<see cref="MaxBirthWindowSeconds"/>
|
||||
/// s: how long the browse surface collects NBIRTH/DBIRTH traffic before calling the observed metric set
|
||||
/// stable (on open, and after an operator rebirth request).
|
||||
/// </summary>
|
||||
/// <param name="options">The browse configuration.</param>
|
||||
/// <returns>The clamped window.</returns>
|
||||
internal static TimeSpan BirthWindow(MqttDriverOptions options) =>
|
||||
TimeSpan.FromSeconds(Math.Clamp(
|
||||
options.Sparkplug?.BirthObservationWindowSeconds ?? new MqttSparkplugOptions().BirthObservationWindowSeconds,
|
||||
1,
|
||||
MaxBirthWindowSeconds));
|
||||
}
|
||||
|
||||
@@ -0,0 +1,145 @@
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser;
|
||||
using SparkplugDataType = Org.Eclipse.Tahu.Protobuf.DataType;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests;
|
||||
|
||||
/// <summary>
|
||||
/// <b>R06:M-4.</b> <c>sparkplug.birthObservationWindowSeconds</c> was authored, range-checked and
|
||||
/// persisted by the AdminUI and consumed by nothing. It now bounds how long the Sparkplug browse surface
|
||||
/// collects NBIRTH/DBIRTH traffic before calling the observed set stable: the open budget grows by it,
|
||||
/// the open lingers while births are still arriving, and an operator rebirth request waits (bounded) for
|
||||
/// the births it asked for, so the tree the picker re-renders already carries them.
|
||||
/// </summary>
|
||||
public sealed class MqttBirthObservationWindowTests
|
||||
{
|
||||
private static MqttDriverOptions Sparkplug(int windowSeconds, int connectTimeoutSeconds = 10) => new()
|
||||
{
|
||||
Mode = MqttMode.SparkplugB,
|
||||
ConnectTimeoutSeconds = connectTimeoutSeconds,
|
||||
Sparkplug = new MqttSparkplugOptions { GroupId = "Plant1", BirthObservationWindowSeconds = windowSeconds },
|
||||
};
|
||||
|
||||
[Fact]
|
||||
public void OpenBudget_Sparkplug_AddsTheBirthObservationWindow()
|
||||
{
|
||||
MqttDriverBrowser.OpenBudget(Sparkplug(windowSeconds: 15)).ShouldBe(TimeSpan.FromSeconds(10 + 15));
|
||||
MqttDriverBrowser.OpenBudget(Sparkplug(windowSeconds: 40)).ShouldBe(TimeSpan.FromSeconds(10 + 40));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void OpenBudget_Plain_IgnoresTheSparkplugWindow()
|
||||
{
|
||||
var plain = Sparkplug(windowSeconds: 40) with { Mode = MqttMode.Plain };
|
||||
MqttDriverBrowser.OpenBudget(plain).ShouldBe(TimeSpan.FromSeconds(10));
|
||||
}
|
||||
|
||||
[Theory]
|
||||
[InlineData(0, 1)]
|
||||
[InlineData(-5, 1)]
|
||||
[InlineData(15, 15)]
|
||||
[InlineData(100_000, 120)]
|
||||
public void BirthWindow_IsClamped(int configured, int expectedSeconds)
|
||||
=> MqttDriverBrowser.BirthWindow(Sparkplug(configured)).ShouldBe(TimeSpan.FromSeconds(expectedSeconds));
|
||||
|
||||
/// <summary>An idle plant costs one quiet interval, not the whole window.</summary>
|
||||
[Fact]
|
||||
public async Task Settle_WithNoBirths_ReturnsAfterOneQuietInterval()
|
||||
{
|
||||
var reads = 0;
|
||||
await MqttBrowseSession.SettleBirthsAsync(
|
||||
() => { reads++; return 0; },
|
||||
baseline: 0,
|
||||
awaitFirstBirth: false,
|
||||
quiet: TimeSpan.FromMilliseconds(20),
|
||||
window: TimeSpan.FromMinutes(5),
|
||||
TestContext.Current.CancellationToken)
|
||||
.WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken);
|
||||
|
||||
reads.ShouldBe(1);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// While births keep arriving the settle keeps collecting; it returns on the first quiet interval
|
||||
/// after the last one. Counted in reads, not wall-clock, so it cannot flake.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task Settle_KeepsCollectingWhileBirthsArrive_ThenReturnsOnQuiet()
|
||||
{
|
||||
var reads = 0;
|
||||
long births = 0;
|
||||
await MqttBrowseSession.SettleBirthsAsync(
|
||||
() =>
|
||||
{
|
||||
reads++;
|
||||
if (reads <= 5) births++; // five intervals with fresh births, then silence
|
||||
return births;
|
||||
},
|
||||
baseline: 0,
|
||||
awaitFirstBirth: false,
|
||||
quiet: TimeSpan.FromMilliseconds(20),
|
||||
window: TimeSpan.FromMinutes(5),
|
||||
TestContext.Current.CancellationToken)
|
||||
.WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken);
|
||||
|
||||
reads.ShouldBe(6);
|
||||
}
|
||||
|
||||
/// <summary>The window is a hard ceiling: a birth flood that never goes quiet still ends.</summary>
|
||||
[Trait("Category", "SlowUnit")]
|
||||
[Fact]
|
||||
public async Task Settle_NeverQuiet_EndsAtTheWindow()
|
||||
{
|
||||
long births = 0;
|
||||
await MqttBrowseSession.SettleBirthsAsync(
|
||||
() => ++births,
|
||||
baseline: 0,
|
||||
awaitFirstBirth: false,
|
||||
quiet: TimeSpan.FromMilliseconds(20),
|
||||
window: TimeSpan.FromSeconds(1),
|
||||
TestContext.Current.CancellationToken)
|
||||
.WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken);
|
||||
|
||||
births.ShouldBeGreaterThan(1);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A rebirth request waits for the birth it asked for. The fixed version cannot complete
|
||||
/// synchronously (it awaits the settle), the unfixed one completes before the call returns — so the
|
||||
/// <c>IsCompleted</c> check is deterministic, not a timing race. The birth then releases it.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task RequestRebirth_WaitsForTheRequestedBirths_ThenReturns()
|
||||
{
|
||||
await using var s = new MqttBrowseSession(
|
||||
MqttMode.SparkplugB, client: null, logger: null, birthWindow: TimeSpan.FromMinutes(5));
|
||||
|
||||
var request = s.RequestRebirthAsync("Plant1/EdgeA", TestContext.Current.CancellationToken);
|
||||
request.IsCompleted.ShouldBeFalse("the request must wait for the births it asked for");
|
||||
|
||||
s.ObserveBirthForTest("Plant1", "EdgeA", null, "Uptime", SparkplugDataType.Int64);
|
||||
|
||||
(await request.WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken)).ShouldBe(1);
|
||||
s.PublishCountForTest.ShouldBe(1);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The wait is a courtesy, never a failure: the NCMD was already published, so a caller deadline
|
||||
/// during the settle returns the published count instead of reporting an error for a request that
|
||||
/// succeeded.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task RequestRebirth_CallerCancelsDuringTheSettle_StillReportsThePublish()
|
||||
{
|
||||
await using var s = new MqttBrowseSession(
|
||||
MqttMode.SparkplugB, client: null, logger: null, birthWindow: TimeSpan.FromMinutes(5));
|
||||
using var cts = CancellationTokenSource.CreateLinkedTokenSource(TestContext.Current.CancellationToken);
|
||||
|
||||
var request = s.RequestRebirthAsync("Plant1/EdgeA", cts.Token);
|
||||
request.IsCompleted.ShouldBeFalse();
|
||||
await cts.CancelAsync();
|
||||
|
||||
(await request.WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken)).ShouldBe(1);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user