From d67396cc6d7eaa6ecb8c15554509d0503a414b4d Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 22 Oct 2026 15:33:47 -0400 Subject: [PATCH] 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 --- docs/drivers/Mqtt.md | 4 +- .../MqttBrowseSession.cs | 80 +++++++++- .../MqttDriverBrowser.cs | 65 +++++++- .../MqttBirthObservationWindowTests.cs | 145 ++++++++++++++++++ 4 files changed, 284 insertions(+), 10 deletions(-) create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttBirthObservationWindowTests.cs diff --git a/docs/drivers/Mqtt.md b/docs/drivers/Mqtt.md index 8400ab42..d8620d65 100644 --- a/docs/drivers/Mqtt.md +++ b/docs/drivers/Mqtt.md @@ -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 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 a231124b..9d4f6772 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 @@ -149,6 +149,9 @@ internal sealed class MqttBrowseSession : IRebirthCapableBrowseSession /// private static readonly UTF8Encoding Utf8Strict = new(encoderShouldEmitUTF8Identifier: false, throwOnInvalidBytes: true); + /// The quiet spell that ends a rebirth request's wait once its births have started arriving. + private static readonly TimeSpan RebirthQuietInterval = TimeSpan.FromMilliseconds(500); + /// Bound on the courtesy DISCONNECT issued at dispose — teardown never hangs on a dead broker. private static readonly TimeSpan DisconnectTimeout = TimeSpan.FromSeconds(5); @@ -174,6 +177,7 @@ internal sealed class MqttBrowseSession : IRebirthCapableBrowseSession private readonly ConcurrentQueue _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 /// The connected MQTT client; disconnected and disposed by . /// Optional logger; never receives credentials or payload bodies. /// Cap on recorded nodes; defaults to . - internal MqttBrowseSession(MqttMode mode, IMqttClient? client, ILogger? logger, int nodeCap = DefaultNodeCap) + /// + /// The Sparkplug birth-observation window: how long waits for the + /// births it asked for. (the test default) returns as soon as the NCMDs + /// are published. + /// + 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 /// internal int PublishCountForTest => Volatile.Read(ref _publishCount); + /// The Sparkplug birth-observation window this session was opened with (R06:M-4). + internal TimeSpan BirthWindow { get; } + + /// How many NBIRTH/DBIRTH messages this session has observed — the settle loop's signal. + internal long BirthCount => Interlocked.Read(ref _birthCount); + + /// + /// Waits until birth traffic has gone quiet for one interval, bounded by + /// — the UntilStable cadence of the birth-observation window. + /// + /// Reads the running birth count. + /// The count before the wait began. + /// + /// When true, quiet only counts once at least one birth has arrived past + /// — the shape a rebirth request needs, since its answer takes a broker round-trip to arrive. + /// + /// The quiet spell that ends the wait. + /// The hard ceiling; non-positive returns at once. + /// Cancels the wait. + /// A task that completes when the births have settled or the window has elapsed. + internal static async Task SettleBirthsAsync( + Func 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; + } + } + /// The topics this session published, in order — the discriminating view of the above. internal IReadOnlyList 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; } diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser/MqttDriverBrowser.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser/MqttDriverBrowser.cs index 3ef6b659..8be8affc 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser/MqttDriverBrowser.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser/MqttDriverBrowser.cs @@ -39,6 +39,15 @@ public sealed class MqttDriverBrowser : IDriverBrowser /// Ceiling on the open budget — the picker must never hang on an unreachable broker. internal const int MaxOpenBudgetSeconds = 30; + /// Ceiling on the Sparkplug birth-observation window — an authored hour must not park a picker. + internal const int MaxBirthWindowSeconds = 120; + + /// + /// 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. + /// + internal static readonly TimeSpan BirthQuietInterval = TimeSpan.FromSeconds(1); + /// Marks the transient browse identity in the broker's client-id/session logs. 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 }; /// - /// Whole-open bound (connect + subscribe), clamped to - /// – around the config's - /// own connect deadline. + /// Whole-open bound: connect + subscribe, clamped to + /// – around the config's own + /// connect deadline — plus, in Sparkplug mode, the the open may spend + /// collecting births. /// /// The browse configuration. - /// The clamped open budget. - internal static TimeSpan OpenBudget(MqttDriverOptions options) => - TimeSpan.FromSeconds(Math.Clamp(options.ConnectTimeoutSeconds, MinOpenBudgetSeconds, MaxOpenBudgetSeconds)); + /// The open budget. + 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; + } + + /// + /// The authored sparkplug.birthObservationWindowSeconds, clamped to 1– + /// 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). + /// + /// The browse configuration. + /// The clamped window. + internal static TimeSpan BirthWindow(MqttDriverOptions options) => + TimeSpan.FromSeconds(Math.Clamp( + options.Sparkplug?.BirthObservationWindowSeconds ?? new MqttSparkplugOptions().BirthObservationWindowSeconds, + 1, + MaxBirthWindowSeconds)); } diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttBirthObservationWindowTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttBirthObservationWindowTests.cs new file mode 100644 index 00000000..b7f920af --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttBirthObservationWindowTests.cs @@ -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; + +/// +/// R06:M-4. sparkplug.birthObservationWindowSeconds 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. +/// +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)); + + /// An idle plant costs one quiet interval, not the whole window. + [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); + } + + /// + /// 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. + /// + [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); + } + + /// The window is a hard ceiling: a birth flood that never goes quiet still ends. + [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); + } + + /// + /// 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 + /// IsCompleted check is deterministic, not a timing race. The birth then releases it. + /// + [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); + } + + /// + /// 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. + /// + [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); + } +}