diff --git a/docs/drivers/Mqtt.md b/docs/drivers/Mqtt.md index 6d1b921e..76be2763 100644 --- a/docs/drivers/Mqtt.md +++ b/docs/drivers/Mqtt.md @@ -2,7 +2,8 @@ In-process MQTT **broker-subscribe** driver. It holds one MQTTnet 5 client against one broker, subscribes to the topics its authored tags name, and forwards each PUBLISH straight to -`ISubscribable.OnDataChange`. Ingest is **push, not poll**, and **one-way** — there is no `IWritable`. +`ISubscribable.OnDataChange`. Ingest is **push, not poll**; writes go back out as ordinary publishes +(see [Write-through](#write-through)). **Two ingest shapes, selected by the driver's `Mode`:** `Plain` binds a tag to a concrete MQTT topic; `SparkplugB` binds it to a `group / edge node / device? / metric` tuple and decodes Eclipse-Tahu @@ -12,18 +13,18 @@ See [Mqtt-Test-Fixture.md](Mqtt-Test-Fixture.md) for the harness and [`../plans/2026-07-15-mqtt-sparkplug-driver-design.md`](../plans/2026-07-15-mqtt-sparkplug-driver-design.md) for the design. -> **Status.** P1 (plain MQTT) and P2 (Sparkplug B ingest) are both shipped and live-gated against the -> Mosquitto + Sparkplug-simulator fixture. Write-through (`IWritable` — NCMD/DCMD/plain publish) is -> **deferred** ([#508](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/508)); every MQTT node -> materializes read-only. See [Known gaps](#known-gaps) — every entry there carries a tracking issue -> (#507–#515). +> **Status.** P1 (plain MQTT), P2 (Sparkplug B ingest), write-through +> ([#508](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/508)) and the Sparkplug primary-host +> claim ([#511](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/511)) are all shipped and +> live-gated against the Mosquitto + Sparkplug-simulator fixture. See +> [Known gaps](#known-gaps) — every entry there carries a tracking issue. ## Project Layout | Project | Holds | |---|---| | `Driver.Mqtt.Contracts` | Config records, enums, `MqttTagDefinition` + the pure `MqttTagDefinitionFactory`, and the one shared `MqttJson.Options`. **Deliberately transport-free** — no MQTTnet reference | -| `Driver.Mqtt` | Runtime: `MqttDriver`, `MqttConnection` (connect + reconnect supervisor), `MqttSubscriptionManager`, `LastValueCache`, factory registration, `MqttDriverProbe` | +| `Driver.Mqtt` | Runtime: `MqttDriver`, `MqttConnection` (connect + reconnect supervisor), `MqttSubscriptionManager`, `LastValueCache`, `SparkplugIngestor`, `SparkplugPrimaryHost`, factory registration, `MqttDriverProbe` | | `Driver.Mqtt.Transport` | `MqttClientOptionsFactory` — TLS posture, PEM CA-pin chain validator, credentials, protocol mapping. Connection **policy** only; nothing here connects | | `Driver.Mqtt.Browser` | `MqttDriverBrowser` + `MqttBrowseSession` — the AdminUI address picker's `#`-observation session | @@ -113,8 +114,8 @@ driver are pure organisational grouping in the RawPath. "Mode": "SparkplugB", "Sparkplug": { "GroupId": "OtOpcUaSim", // REQUIRED — the driver's whole subscription scope - "HostId": "", // Host Application identity; receive-only (see below) - "ActAsPrimaryHost": false, // NOT IMPLEMENTED — logs a warning when true + "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 } @@ -124,8 +125,8 @@ driver are pure organisational grouping in the RawPath. | Key | Default | Notes | |---|---|---| | `GroupId` | `""` | **Required.** The driver subscribes exactly `spBv1.0/{GroupId}/#`. Blank ⇒ it connects, reports `Healthy`, and ingests **nothing**. It is a literal topic segment, so `/`, `+`, `#` are refused by the form | -| `HostId` | `""` | Sparkplug Host Application identity. **Receive-only** — the driver observes `STATE` and never publishes one | -| `ActAsPrimaryHost` | `false` | **Not implemented.** Setting it logs a startup **warning** rather than being silently inert — being a primary host is a three-part contract (retained `ONLINE`, a matching `OFFLINE` registered as the MQTT Will *at connect time*, an explicit `OFFLINE` on clean shutdown), and shipping the `ONLINE` half without the Will is strictly worse than shipping neither: a crashed node would leave a retained `ONLINE` asserting forever that a dead host is alive | +| `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 | @@ -252,12 +253,80 @@ One subscription (`spBv1.0/{GroupId}/#`) feeds a per-edge-node state machine anti-storm, not optimisations: rediscovery triggers an address-space rebuild, and an edge node re-births freely — on its own schedule, on every reconnect, and once per rebirth NCMD. -> ⚠️ **Nothing consumes the signal today.** No runtime component subscribes to -> `IRediscoverable.OnRediscoveryNeeded`, and `DriverHostActor.HandleDiscoveredNodes` hard-returns — -> discovered-node injection is dormant in v3. The served address space comes from the deploy artifact, -> so **a DBIRTH introducing a new metric does not change the OPC UA tree**; you must redeploy. The -> driver's decision is observable only as an Information log line -> (`… requesting rediscovery.`). This is a v3 platform gap, not an MQTT one. +> ⚠️ **The signal is an operator prompt, not a rebuild.** `DriverInstanceActor` subscribes and the +> raise rides the driver-health snapshot to the AdminUI `/hosts` page as a **re-browse chip** — but the +> served address space still comes from the deploy artifact, so **a DBIRTH introducing a new metric does +> not change the OPC UA tree**. Re-browse the device and commit. That is deliberate: 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. + +## Primary host (`actAsPrimaryHost`) + +A Sparkplug **Primary Host Application** is the thing edge nodes watch to decide whether anyone is +listening — many stop publishing while their host is offline. Claiming it is a three-part contract, +and this driver ships all three or none of it +([#511](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/511)): + +1. a **retained** `{"online":true,"timestamp":…}` on `spBv1.0/STATE/{hostId}`, QoS 1, published after CONNACK; +2. a matching **retained `OFFLINE` armed as the MQTT Last Will**, so a crash still reports the truth; +3. an explicit `OFFLINE` on a clean stand-down, so edge nodes do not wait out a keep-alive timeout. + +Set `Sparkplug.HostId` and `Sparkplug.ActAsPrimaryHost: true`. A blank host id is **refused** with a +warning — the driver does not publish to `spBv1.0/STATE/`, which is a legal topic reaching nobody. + +### The claim is a whole connection, not a message + +A Will can only be registered in the **CONNECT packet**. It cannot be added to a live session, changed, +or withdrawn. So the claim gets its **own** broker connection (client id `{clientId}-spbhost`, forced +clean-session) whose lifetime *is* the claim: opened on gaining Primary, closed on losing it. Every leg +then follows from the connection's lifecycle rather than from bookkeeping that could drift. + +The ingest session is untouched by a role change — no dropped subscriptions, no lost birth cache, and +no data gap at the Secondary→Primary transition, which is the moment a plant most needs values flowing. + +### Primary-gated — and why the obvious alternatives are wrong + +Both nodes of a redundant pair subscribe independently (MQTT fan-out is per-connection, and the Primary +gate applies downstream of ingest), so an ungated claim means both claim the same host id. The sharp +failure is not the two fighting over the retained `STATE`. It is that the **Secondary's** death fires +*its* Will, publishing `OFFLINE` while the Primary is perfectly healthy — every edge node in the group +stops publishing, and redundancy has caused the outage the mechanism exists to prevent. + +Registering the Will only on whichever node is Primary *at connect time* fails the other way: a node +that later gains Primary would publish `ONLINE` with no Will behind it, and a crash would leave a +retained `ONLINE` asserting forever that a dead host is alive. + +Hence: bind the claim to a connection, and rebuild the connection on role change. + +| Node state | STATE connection | Retained `STATE` | +|---|---|---| +| Primary | open, Will armed | `ONLINE` | +| Secondary / Detached | **never opened** | — (whatever the Primary published) | +| Role unknown, driver peer present | not opened (fail closed) | — | +| Role unknown, no driver peer (single node) | open | `ONLINE` | + +The verdict comes from `PrimaryGatePolicy.ShouldServiceAsPrimary` via `IDriverPrimaryRoleSource` +(`Core.Abstractions`), published by `DriverHostActor` on every redundancy snapshot and **sampled on a +1-second timer** rather than evented — a process-singleton raising events at per-instance drivers +outlives every driver it holds a reference to. + +> **Accepted trade: a brief edge-visible `OFFLINE` during failover**, while the old Primary stands down +> and the new one connects and republishes. That is a genuine state transition — for that window there +> really is no host — not a glitch worth hiding. + +### Operational notes + +- **At most one primary host per host id per broker.** Two separate OtOpcUa deployments pointing at one + broker with the same `hostId` will fight, and nothing in either can detect the other. The gate covers + a redundant *pair*, not two independent deployments. +- **A reconnect republishes the birth.** A broker restart drops retained messages; without the + republish the host would be connected and believed offline by everyone, silently. +- **A failed claim never faults the driver.** A broker whose ACL denies `spBv1.0/STATE/#` costs a + warning and a plant-visible missing host, not every tag's value. Ingest is the data plane; the claim + is an assertion to other applications. +- **The driver still observes its own `STATE`.** The ingestor subscribes the configured host id + regardless, so a claiming node sees its own retained `ONLINE` come back. Nothing gates on it, so the + echo costs one log line. ## Connection + Failure Semantics @@ -446,9 +515,14 @@ 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` (**15 tests**) against the +- **Live (env-gated)** — `tests/Drivers/…Driver.Mqtt.IntegrationTests` (**26 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 + reach: `AbandonClaimForTests` drops the socket with **no** `DISCONNECT`, because a clean disconnect + makes the broker *discard* the Will by protocol rule — so a test driven through `StopAsync` would + observe the explicit stand-down `OFFLINE` and pass against an implementation that never armed a Will + at all. - **AdminUI** — `MqttDriverFormModelTests` / `MqttTagConfigModelTests` cover both editors' round-trip, mode inference and validation. AdminUI has no bUnit, so the razor shells themselves are only ever proven by a live `/run` gate. @@ -466,7 +540,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), the slashed-metric browse-commit +**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 ([#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)). @@ -474,5 +549,4 @@ 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 | -| **Primary-host STATE publishing** | [#511](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/511) | `ActAsPrimaryHost` does nothing | Logs a startup warning rather than being silently inert — see the `Sparkplug` sub-object table | | **`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/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/IDriverPrimaryRoleSource.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/IDriverPrimaryRoleSource.cs new file mode 100644 index 00000000..b2780389 --- /dev/null +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/IDriverPrimaryRoleSource.cs @@ -0,0 +1,90 @@ +namespace ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +/// +/// A driver-facing snapshot of "should this node be the one asserting an exclusive identity on a +/// shared external system right now?". +/// +/// +/// +/// Why drivers need this at all. The driver layer has no notion of redundancy, by +/// design — a driver talks to a device and the Primary gate is applied above it, in +/// DriverHostActor, on the way in. That works because every gated operation is +/// inbound: a write, an alarm ack, an alerts publish. A Sparkplug primary-host claim is +/// not inbound. It is a standing assertion the driver makes to the broker, on its own +/// connection, for as long as it holds one — there is no message for the host actor to +/// intercept. So the decision has to reach the driver instead (Gitea #511). +/// +/// +/// The decision, not the role. Same discipline as the alarm-drain view in +/// Runtime: this publishes the verdict a single policy produced, so the policy stays the +/// one place that decides and a driver can never re-derive a subtly different rule from a raw +/// role. The verdict fed here is the data-plane one (an unknown role on a cluster that +/// has a driver peer denies), because two nodes claiming one Sparkplug host id is the harm the +/// gate exists to prevent — not a duplicate row, but a Secondary's Last Will publishing +/// OFFLINE over a live Primary's ONLINE and stopping every edge node in the +/// group. +/// +/// +/// Polled, not evented, and that is deliberate. A process-wide singleton raising an +/// event onto per-instance drivers outlives every driver it hands a reference to — the exact +/// missing--= leak shape the rediscovery signal had to be careful about. Consumers +/// sample this instead, so a driver that is torn down simply stops asking. The cost is up to +/// one poll interval of latency on a failover, which is inside the edge-visible +/// OFFLINE window a failover already implies. +/// +/// +public interface IDriverPrimaryRoleSource +{ + /// + /// Whether this node currently holds the Primary data-plane role — and may therefore claim an + /// exclusive external identity such as a Sparkplug host id. + /// + bool ShouldActAsPrimary { get; } + + /// + /// Records a fresh decision. Called by DriverHostActor on every redundancy snapshot. + /// + /// + /// The verdict PrimaryGatePolicy.ShouldServiceAsPrimary produced. + /// + void PublishPrimaryRole(bool shouldActAsPrimary); +} + +/// +/// The a driver uses when nothing publishes one — a +/// standalone host, a unit test, or any deployment with no redundancy manager wired. +/// +/// +/// +/// It answers , and the alternative is worse. Defaulting closed +/// would be the reflexive "fail-safe" choice, but there is no safety to gain: with nothing +/// publishing a role there is by definition no peer to collide with, and a closed default +/// would make an exclusive-identity claim silently never happen on the single-node deployment +/// that is the common case — an inert feature with no error, which is precisely the failure +/// mode Gitea #511 was filed about. +/// +/// +/// The risk this shifts onto the wiring is real and is guarded there instead: a redundant +/// deployment whose composition root forgets to supply the live source would fall back to this +/// and let both nodes claim. DriverFactoryBootstrapPrimaryRoleSourceTests pins that the +/// Host resolves a real one. +/// +/// +public sealed class UnmanagedDriverPrimaryRoleSource : IDriverPrimaryRoleSource +{ + /// The shared instance — it holds no state. + public static readonly UnmanagedDriverPrimaryRoleSource Instance = new(); + + private UnmanagedDriverPrimaryRoleSource() + { + } + + /// + public bool ShouldActAsPrimary => true; + + /// + /// No-op: there is no redundancy manager to publish anything here. + public void PublishPrimaryRole(bool shouldActAsPrimary) + { + } +} diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttDriverOptions.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttDriverOptions.cs index 8acb9148..c40bae79 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttDriverOptions.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttDriverOptions.cs @@ -190,6 +190,12 @@ public sealed record MqttSparkplugOptions /// When true, this driver instance acts as the Sparkplug Primary Host /// Application. At most one primary host may exist per host-id per broker. /// + /// + /// Implemented by SparkplugPrimaryHost (Gitea #511) on a dedicated broker connection + /// whose lifetime is the claim, and gated on the redundancy Primary role so the two nodes + /// of a pair never both claim. Requires ; set without one, the driver warns + /// and declines rather than publishing to a malformed topic. + /// public bool ActAsPrimaryHost { get; init; } = false; /// diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttLastWill.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttLastWill.cs new file mode 100644 index 00000000..a65a905d --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttLastWill.cs @@ -0,0 +1,37 @@ +namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt; + +/// +/// An MQTT Last Will and Testament, as the client options factory wants it (Gitea #511). +/// +/// +/// +/// A Will can only be registered at CONNECT, and that single fact drives the whole design +/// of the Sparkplug primary-host claim. It rides in the CONNECT packet, so it cannot be +/// added, changed or withdrawn on a live session. A node that gains the Primary role after +/// connecting therefore cannot simply start publishing ONLINE — it would have no Will, +/// and a crash would leave a retained ONLINE asserting forever that a dead host is +/// alive. Equally, a node that registers a Will and later loses the role cannot withdraw it, +/// so its eventual disconnect would publish OFFLINE over the new Primary's live claim +/// and stop every edge node in the group. +/// +/// +/// The consequence is that the claim must be bound to the lifetime of a connection: +/// SparkplugPrimaryHost opens a dedicated connection when the node gains Primary and +/// closes it when the node loses Primary, rather than trying to mutate the ingest session it +/// shares with the rest of the driver. +/// +/// +/// Clean session matters too. A broker that resumes an existing session for the same +/// client id may discard the Will registered by the new CONNECT. The primary-host connection +/// forces clean-session on for that reason, regardless of the driver's configured value. +/// +/// +/// The topic the broker publishes the Will to. +/// The Will body. +/// Requested QoS, 0–2. +/// +/// Whether the broker retains the Will. For a Sparkplug host death certificate this is +/// : an edge node that connects after the host died must still learn +/// the host is offline, and only a retained message reaches it. +/// +public sealed record MqttLastWill(string Topic, byte[] Payload, int Qos, bool Retain); diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/Sparkplug/SparkplugStatePayload.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/Sparkplug/SparkplugStatePayload.cs new file mode 100644 index 00000000..1065f109 --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/Sparkplug/SparkplugStatePayload.cs @@ -0,0 +1,141 @@ +using System.Text; +using System.Text.Json; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Sparkplug; + +/// +/// Builds the Sparkplug v3.0 primary-host STATE bodies — the retained ONLINE birth +/// and the matching OFFLINE death that is registered as the MQTT Last Will (Gitea #511). +/// +/// +/// +/// Encode only. The receive side lives in SparkplugIngestor.TryReadStateOnline, +/// which additionally tolerates the pre-3.0 bare ONLINE/OFFLINE text form. This +/// type only ever produces the v3.0 JSON shape, matching +/// 's refusal to produce the legacy topic. +/// +/// +/// The two bodies share one timestamp, and that is a spec requirement rather than a +/// convenience. Sparkplug 3.0 pairs a host's birth and death certificates by their +/// timestamp field: an edge node that receives an OFFLINE whose timestamp is +/// older than the ONLINE it already holds is entitled to ignore it as a stale death +/// from a previous session. That rule is what makes the mechanism safe across a reconnect — +/// and it means the death body has to be built before CONNECT (a Will is registered in +/// the CONNECT packet) while the birth is published after CONNACK. Deriving each from +/// its own UtcNow would give them different timestamps and quietly break the pairing, +/// so mints both at once and there is no single-body overload to +/// reach for by accident. +/// +/// +public static class SparkplugStatePayload +{ + /// + /// QoS for both STATE bodies. Sparkplug 3.0 requires QoS 1 for the host birth and death — + /// unlike the QoS 0 that carries NBIRTH/DBIRTH/DATA. A host claim delivered at most once is a + /// claim that may simply never arrive. + /// + public const int StateQos = 1; + + /// + /// Mints the matched ONLINE/OFFLINE pair for one host session, both carrying + /// . + /// + /// The Sparkplug host id; used to build the topic. + /// + /// Sparkplug epoch milliseconds for this session. Supplied by the caller rather than read from + /// the clock here so the value is visible at the call site and testable. + /// + /// The topic plus both bodies. + /// is blank. + public static SparkplugStateSession ForSession(string hostId, long timestampMs) + { + ArgumentException.ThrowIfNullOrWhiteSpace(hostId); + + return new SparkplugStateSession( + Topic: SparkplugTopic.FormatState(hostId), + TimestampMs: timestampMs, + OnlineBody: Encode(online: true, timestampMs), + OfflineBody: Encode(online: false, timestampMs)); + } + + /// Encodes one STATE body: {"online":<bool>,"timestamp":<ms>}. + /// The online flag. + /// Sparkplug epoch milliseconds. + /// The UTF-8 body bytes. + /// + /// Hand-written rather than serialized from a DTO. The body is two fields, and the wire form + /// is fixed by the spec — a serializer would put naming policy, casing options and a shared + /// JsonSerializerOptions between this repo and a wire contract it does not control. + /// timestamp is emitted as a bare JSON number, never a string: a quoted timestamp + /// parses as a valid JSON object and is silently wrong to every conformant host. + /// + public static byte[] Encode(bool online, long timestampMs) => + Encoding.UTF8.GetBytes( + $"{{\"online\":{(online ? "true" : "false")},\"timestamp\":{timestampMs.ToString(System.Globalization.CultureInfo.InvariantCulture)}}}"); + + /// Reads the online flag back out of a v3.0 STATE body, for tests and diagnostics. + /// The STATE body bytes. + /// The decoded flag on success. + /// The decoded timestamp on success. + /// when the body is a well-formed v3.0 STATE object. + public static bool TryDecode(ReadOnlySpan body, out bool online, out long timestampMs) + { + online = false; + timestampMs = 0; + + if (body.IsEmpty) + { + return false; + } + + try + { + var reader = new Utf8JsonReader(body); + if (!JsonDocument.TryParseValue(ref reader, out var doc)) + { + return false; + } + + using (doc) + { + if (doc.RootElement.ValueKind != JsonValueKind.Object + || !doc.RootElement.TryGetProperty("online", out var flag) + || flag.ValueKind is not (JsonValueKind.True or JsonValueKind.False)) + { + return false; + } + + online = flag.GetBoolean(); + timestampMs = doc.RootElement.TryGetProperty("timestamp", out var ts) + && ts.ValueKind == JsonValueKind.Number + && ts.TryGetInt64(out var parsed) + ? parsed + : 0; + return true; + } + } + catch (JsonException) + { + return false; + } + } +} + +/// +/// The STATE topic plus the matched pair of bodies for one primary-host session (Gitea #511). +/// +/// The v3.0 STATE topic, spBv1.0/STATE/{hostId}. +/// The session timestamp carried by BOTH bodies. +/// +/// The birth certificate — published retained, at QoS 1, immediately after CONNACK. +/// +/// +/// The death certificate — registered as the MQTT Last Will in the CONNECT packet, and also +/// published explicitly on a clean stand-down so the broker holds it without waiting for a +/// keep-alive timeout. +/// +public sealed record SparkplugStateSession( + string Topic, + long TimestampMs, + byte[] OnlineBody, + byte[] OfflineBody); diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Transport/MqttClientOptionsFactory.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Transport/MqttClientOptionsFactory.cs index 4e4e9278..8b0dcd69 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Transport/MqttClientOptionsFactory.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Transport/MqttClientOptionsFactory.cs @@ -41,20 +41,51 @@ public static class MqttClientOptionsFactory /// empty MQTTnet generates a random client id. /// /// Optional logger for the deferred CA-pin load; never receives credentials. + /// + /// Optional MQTT Last Will registered in the CONNECT packet (Gitea #511 — the Sparkplug + /// primary-host death certificate). See for why this can only be + /// supplied here. + /// + /// + /// Forces clean-session on or off, ignoring . Used + /// by the primary-host connection, which must connect clean so the broker cannot resume a + /// previous session and discard the Will — see . + /// public static MqttClientOptions BuildClientOptions( MqttDriverOptions options, string? clientIdSuffix, - ILogger? logger = null) + ILogger? logger = null, + MqttLastWill? lastWill = null, + bool? cleanSessionOverride = null) { ArgumentNullException.ThrowIfNull(options); var builder = new MqttClientOptionsBuilder() .WithTcpServer(options.Host, options.Port) .WithProtocolVersion(MapProtocolVersion(options.ProtocolVersion)) - .WithCleanSession(options.CleanSession) + .WithCleanSession(cleanSessionOverride ?? options.CleanSession) .WithKeepAlivePeriod(TimeSpan.FromSeconds(options.KeepAliveSeconds)) .WithTimeout(TimeSpan.FromSeconds(options.ConnectTimeoutSeconds)); + if (lastWill is { } will) + { + builder = builder + .WithWillTopic(will.Topic) + .WithWillPayload(will.Payload) + .WithWillQualityOfServiceLevel((MQTTnet.Protocol.MqttQualityOfServiceLevel)will.Qos) + .WithWillRetain(will.Retain); + + // MQTT 5.0 only. Left at its default the broker MAY hold the Will back for up to the + // session-expiry interval after an ungraceful disconnect, which for a host death + // certificate is precisely the wrong behaviour: the whole point is that edge nodes learn + // promptly. Zero means "publish it the moment the network connection is lost". + // The setter is harmless on 3.1.1 — the property simply has no wire representation there. + if (options.ProtocolVersion == MqttProtocolVersion.V500) + { + builder = builder.WithWillDelayInterval(0); + } + } + var clientId = string.Concat(options.ClientId ?? string.Empty, clientIdSuffix ?? string.Empty); if (!string.IsNullOrWhiteSpace(clientId)) { diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttConnection.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttConnection.cs index 6582de2e..57219b83 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttConnection.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttConnection.cs @@ -286,11 +286,40 @@ public sealed class MqttConnection : IAsyncDisposable, IMqttSubscribeTransport, private Task? _reconnectWorker; private int _state = (int)MqttConnectionState.Disconnected; + /// Client-id suffix distinguishing this connection from the driver's other one; see the ctor. + private readonly string? _clientIdSuffix; + + /// The Will to register on every CONNECT, or null. See . + private readonly MqttLastWill? _lastWill; + + /// Clean-session override, or null to use the configured value. + private readonly bool? _cleanSessionOverride; + /// Stores configuration only — no network, no client instantiation, no supervisor. /// Broker connection settings. /// Driver instance id, used only for log/diagnostic correlation. /// Optional logger; never receives credentials. - public MqttConnection(MqttDriverOptions options, string driverId, ILogger? logger = null) + /// + /// Appended to the configured client id so a second connection from the same driver — the + /// Sparkplug primary-host claim (Gitea #511) — does not collide with the ingest session. Two + /// MQTT clients presenting one client id is not an error the broker reports: it silently + /// disconnects the older one, so the two would take turns evicting each other forever. + /// + /// + /// Optional Last Will registered in the CONNECT packet, and therefore rebuilt on every + /// reconnect attempt along with the rest of the options. See . + /// + /// + /// Forces clean-session on or off regardless of . + /// Null keeps the configured value. + /// + public MqttConnection( + MqttDriverOptions options, + string driverId, + ILogger? logger = null, + string? clientIdSuffix = null, + MqttLastWill? lastWill = null, + bool? cleanSessionOverride = null) { ArgumentNullException.ThrowIfNull(options); ArgumentException.ThrowIfNullOrWhiteSpace(driverId); @@ -298,6 +327,9 @@ public sealed class MqttConnection : IAsyncDisposable, IMqttSubscribeTransport, _options = options; _driverId = driverId; _logger = logger; + _clientIdSuffix = clientIdSuffix; + _lastWill = lastWill; + _cleanSessionOverride = cleanSessionOverride; } /// @@ -370,6 +402,47 @@ public sealed class MqttConnection : IAsyncDisposable, IMqttSubscribeTransport, /// internal Func? AfterConnectHookForTests { get; set; } + /// + /// Test seam: drops the session the way a crashed process does — no MQTT DISCONNECT, so + /// the broker treats the closed socket as an abnormal termination and publishes the Last Will. + /// + /// A task completing once the client is gone and the supervisor has stopped. + /// + /// + /// Exists because the Sparkplug primary-host contract's most important leg is the one a + /// normal teardown cannot reach (Gitea #511): DisposeAsync sends a clean + /// DISCONNECT, which by MQTT rule means the broker discards the Will. So an + /// in-suite test driven through the ordinary lifecycle can only ever observe the explicit + /// stand-down OFFLINE, and would pass identically against an implementation whose + /// Will was never armed at all. + /// + /// + /// Not a full substitute for the kill -9 the live gate runs by hand — that also + /// proves the process dies without any of this code running — but it is the closest an + /// in-process test can get, and it is what makes the Will's content observable. + /// + /// + internal async Task AbandonForTests() + { + // Mark disposed first, so the supervisor cannot reconnect into the gap and publish a fresh + // birth on top of the Will the broker is about to deliver. + Interlocked.Exchange(ref _disposed, 1); + SetState(MqttConnectionState.Disposed); + + try + { + await _lifetimeCts.CancelAsync().ConfigureAwait(false); + } + catch (ObjectDisposedException) + { + // Nothing to cancel. + } + + // Dispose WITHOUT DisconnectAsync: MQTTnet closes the channel, the broker sees a dead socket + // rather than a graceful goodbye, and the Will fires. This is the whole point of the seam. + Interlocked.Exchange(ref _client, null)?.Dispose(); + } + private bool Disposed => Volatile.Read(ref _disposed) != 0; /// @@ -568,7 +641,8 @@ public sealed class MqttConnection : IAsyncDisposable, IMqttSubscribeTransport, MqttClientOptions clientOptions; try { - clientOptions = MqttClientOptionsFactory.BuildClientOptions(_options, clientIdSuffix: null, _logger); + clientOptions = MqttClientOptionsFactory.BuildClientOptions( + _options, _clientIdSuffix, _logger, _lastWill, _cleanSessionOverride); } catch (Exception ex) { diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriver.cs index 4f1e4d54..e8826724 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriver.cs @@ -185,6 +185,16 @@ public sealed class MqttDriver private MqttConnection? _connection; private CancellationTokenSource? _hostProbeCts; + /// The Primary data-plane verdict the Sparkplug host claim is gated on (Gitea #511). + private readonly IDriverPrimaryRoleSource _primaryRoleSource; + + /// + /// The Sparkplug primary-host claim, or when this configuration does not + /// ask for one. Owns its own broker connection — see for why + /// it cannot share the ingest session. + /// + private SparkplugPrimaryHost? _primaryHost; + private DriverHealth _health = new(DriverState.Unknown, null, null); private HostState _hostState = HostState.Unknown; private DateTime _hostStateChangedUtc = DateTime.UtcNow; @@ -198,7 +208,18 @@ public sealed class MqttDriver /// /// Stable logical id of this driver instance. /// Optional logger; never receives credentials or payload bodies. - public MqttDriver(MqttDriverOptions options, string driverInstanceId, ILogger? logger = null) + /// + /// Gitea #511: the Primary data-plane verdict the Sparkplug primary-host claim is gated on. + /// falls back to — which + /// always claims — for standalone hosts and tests. On a redundant pair that fallback is the + /// dual-claim this gate exists to prevent, so the Host resolves a real one and + /// DriverFactoryBootstrap does so with GetRequiredService. + /// + public MqttDriver( + MqttDriverOptions options, + string driverInstanceId, + ILogger? logger = null, + IDriverPrimaryRoleSource? primaryRoleSource = null) { ArgumentNullException.ThrowIfNull(options); ArgumentException.ThrowIfNullOrWhiteSpace(driverInstanceId); @@ -206,6 +227,7 @@ public sealed class MqttDriver _options = options; _driverInstanceId = driverInstanceId; _logger = logger; + _primaryRoleSource = primaryRoleSource ?? UnmanagedDriverPrimaryRoleSource.Instance; _subscriptions = CreateSubscriptionManager(options); _sparkplug = CreateSparkplugIngestor(options); RegisterAuthoredTags(options); @@ -1001,6 +1023,7 @@ public sealed class MqttDriver WriteHealth(new DriverHealth(DriverState.Healthy, connection.LastMessageUtc, null)); TransitionHost(HostState.Running); StartHostProbe(); + await StartPrimaryHostAsync(cancellationToken).ConfigureAwait(false); } catch (Exception ex) { @@ -1094,21 +1117,15 @@ public sealed class MqttDriver return null; } - if (options.Sparkplug is { ActAsPrimaryHost: true }) + // The flag no longer reads as if it does something it does not (Gitea #511) — but a request it + // CANNOT honour still warns loudly rather than being dropped, because the operator's plant is + // waiting on a STATE message that will never arrive. + if (SparkplugPrimaryHost.DescribeUnusable(options) is { } unusable) { - // Loud, because the flag reads as if it does something. Being a Sparkplug Primary Host is a - // three-part contract — a retained ONLINE after CONNECT, a matching retained OFFLINE - // registered as the MQTT Last Will AT CONNECT TIME, and an explicit OFFLINE on clean - // shutdown — and edge nodes stop publishing while their host is offline. Shipping the - // ONLINE half without the Will is strictly worse than shipping neither: a crashed node - // would leave a retained ONLINE asserting forever that a dead host is alive, which is the - // exact state the mechanism exists to prevent. The Will belongs in - // MqttClientOptionsFactory.BuildClientOptions; until it is there this flag stays inert and said so. _logger?.LogWarning( - "MQTT driver '{DriverId}': Sparkplug actAsPrimaryHost is set but primary-host STATE " - + "publishing is not implemented — this driver observes STATE and never publishes one. " - + "Edge nodes gated on this host id will not see it come online.", - _driverInstanceId); + "MQTT driver '{DriverId}': {Reason} Edge nodes gated on this host id will not see it come online.", + _driverInstanceId, + unusable); } var ingestor = new SparkplugIngestor( @@ -1263,9 +1280,55 @@ public sealed class MqttDriver o.Sparkplug, }; + /// + /// Opens the Sparkplug primary-host claim when this configuration asks for one (Gitea #511). + /// + /// + /// + /// Runs last, and can never fail the initialize. The claim is an advisory assertion + /// to other applications on the broker; the ingest path is this driver's data plane. A + /// broker that accepts the ingest session but rejects a publish on the STATE topic — an ACL + /// that does not grant spBv1.0/STATE/# is the ordinary way this happens — must cost + /// the operator a warning and a plant-visible missing host, not every tag's value. + /// already swallows and retries; the catch + /// here covers the rest so no future edit can turn this into a fault path by accident. + /// + /// + private async Task StartPrimaryHostAsync(CancellationToken cancellationToken) + { + if (!SparkplugPrimaryHost.CanClaim(_options)) + { + return; + } + + try + { + var host = new SparkplugPrimaryHost(_options, _driverInstanceId, _primaryRoleSource, _logger); + _primaryHost = host; + await host.StartAsync(cancellationToken).ConfigureAwait(false); + } + catch (Exception ex) + { + _logger?.LogWarning( + ex, + "MQTT driver '{DriverId}': the Sparkplug primary-host claim could not be started; ingest is unaffected.", + _driverInstanceId); + } + } + /// Shared teardown for , and the session rebuild. private async Task TeardownAsync() { + // Stand the host claim down FIRST, while the broker is still reachable and before the ingest + // session goes. Its explicit retained OFFLINE is what makes the transition immediate for edge + // nodes; leaving it to the Will costs them up to 1.5 keep-alive periods of believing a host + // that has already stopped is still online. + var primaryHost = Interlocked.Exchange(ref _primaryHost, null); + if (primaryHost is not null) + { + await primaryHost.DisposeAsync().ConfigureAwait(false); + } + var probe = Interlocked.Exchange(ref _hostProbeCts, null); if (probe is not null) { diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriverFactoryExtensions.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriverFactoryExtensions.cs index dc6d9bea..cf6dc094 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriverFactoryExtensions.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriverFactoryExtensions.cs @@ -49,16 +49,30 @@ public static class MqttDriverFactoryExtensions /// /// The driver factory registry to register with. /// Optional logger factory used to create per-instance loggers. - public static void Register(DriverFactoryRegistry registry, ILoggerFactory? loggerFactory = null) + /// + /// Gitea #511: the Primary data-plane verdict the Sparkplug primary-host claim is gated on, + /// captured at registration time like . Null falls back to + /// (always claims) — correct for a standalone + /// host or a test, and the reason DriverFactoryBootstrap resolves it with + /// GetRequiredService rather than letting a redundant deployment reach that fallback. + /// + public static void Register( + DriverFactoryRegistry registry, + ILoggerFactory? loggerFactory = null, + IDriverPrimaryRoleSource? primaryRoleSource = null) { ArgumentNullException.ThrowIfNull(registry); - registry.Register(DriverTypeName, (id, json) => CreateInstance(id, json, loggerFactory)); + registry.Register(DriverTypeName, (id, json) => CreateInstance(id, json, loggerFactory, primaryRoleSource)); } /// Public for the Server-side bootstrapper + test consumers. /// Stable logical id of the driver instance. /// The DriverConfig JSON blob for the instance. /// Optional logger factory for the per-instance logger. + /// + /// Gitea #511: the Primary data-plane verdict gating the Sparkplug primary-host claim; null + /// falls back to . + /// /// A configured, not-yet-connected . /// /// or is blank. @@ -68,7 +82,8 @@ public static class MqttDriverFactoryExtensions public static MqttDriver CreateInstance( string driverInstanceId, string driverConfigJson, - ILoggerFactory? loggerFactory = null) + ILoggerFactory? loggerFactory = null, + IDriverPrimaryRoleSource? primaryRoleSource = null) { ArgumentException.ThrowIfNullOrWhiteSpace(driverInstanceId); ArgumentException.ThrowIfNullOrWhiteSpace(driverConfigJson); @@ -79,6 +94,7 @@ public static class MqttDriverFactoryExtensions ?? throw new InvalidOperationException( $"MQTT driver config for '{driverInstanceId}' deserialised to null"); - return new MqttDriver(options, driverInstanceId, loggerFactory?.CreateLogger()); + return new MqttDriver( + options, driverInstanceId, loggerFactory?.CreateLogger(), primaryRoleSource); } } 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 a5c294ef..91b0836d 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 @@ -73,9 +73,11 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Sparkplug; /// subscriber is contained, and the rebirth publish is dispatched off-thread. /// /// -/// Primary-host STATE publishing is deliberately NOT implemented — see -/// and the remarks on -/// . This type observes STATE; it never publishes one. +/// This type only ever observes STATE. Primary-host STATE publishing ships +/// (Gitea #511) but deliberately does not live here: the claim is bound to the lifetime of its +/// own dedicated connection, because an MQTT Will can only be registered at CONNECT. See +/// , and the remarks on for what +/// this side does with an observed one. /// /// public sealed class SparkplugIngestor @@ -1029,18 +1031,17 @@ public sealed class SparkplugIngestor /// /// /// - /// Receive only. This driver never publishes STATE, and - /// is therefore inert — deliberately, and - /// loudly. Being a Sparkplug Primary Host is not "publish an ONLINE message": it is a - /// three-part contract — a retained ONLINE STATE published after CONNECT, a matching - /// retained OFFLINE registered as the client's MQTT Last Will at CONNECT time, - /// and an explicit OFFLINE on clean shutdown. Edge nodes subscribe to it and stop publishing - /// while their host is offline. Shipping the ONLINE half without the Will is strictly worse - /// than shipping neither: an OtOpcUa node that crashes would leave a retained ONLINE asserting - /// forever that a dead host is alive, which is precisely the state the mechanism exists to - /// prevent. The Will has to be set inside MqttClientOptionsFactory.BuildClientOptions — a - /// different file, a different task — so this task implements the observation half and - /// warns when the flag is set. See the design doc §7/§8. + /// Receive only, on this connection. The publish half ships as + /// (Gitea #511), on a connection of its own — an MQTT Will + /// can only be registered in the CONNECT packet, so a claim that is acquired and released at + /// runtime cannot ride the long-lived ingest session this type is attached to. + /// + /// + /// A host claimed by THIS node is observed here too, and that is correct rather than an + /// oversight. The ingestor subscribes the configured host id whether or not this node + /// publishes it, so a node acting as primary host sees its own retained ONLINE come + /// back. Nothing here gates on it — is informational — so the + /// echo costs one log line at the transition and cannot feed back into the claim. /// /// /// Both payload forms are accepted on receive: the v3.0 JSON {"online":bool,...} and diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/SparkplugPrimaryHost.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/SparkplugPrimaryHost.cs new file mode 100644 index 00000000..7911c872 --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/SparkplugPrimaryHost.cs @@ -0,0 +1,470 @@ +using Microsoft.Extensions.Logging; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Sparkplug; + +/// +/// Claims the Sparkplug Primary Host Application identity for this node — the three-part contract +/// promises and, until Gitea #511, did not +/// deliver: a retained ONLINE while the claim is held, a matching retained OFFLINE +/// registered as the MQTT Last Will, and an explicit OFFLINE on a clean stand-down. +/// +/// +/// +/// The claim is a whole dedicated connection, not a message. A Will can only be +/// registered in the CONNECT packet (see ), so a claim that must be +/// acquired and released at runtime cannot live on the driver's shared ingest session — that +/// session connects once and stays up, and its Will could be neither added on gaining the role +/// nor withdrawn on losing it. This type therefore opens its own connection when the +/// node gains Primary and closes it when the node loses Primary. The connection's lifetime +/// is the claim, which is what makes every leg of the contract fall out for free: +/// +/// the Will is registered because the connection was opened to hold the claim; +/// a kill -9 drops the connection, so the broker publishes the Will; +/// a clean stand-down publishes OFFLINE and then disconnects, so the retained +/// message is correct before the Will could ever fire. +/// +/// +/// +/// Why not rebuild the ingest connection instead. That was the shape first sketched on +/// the issue, and it works, but it drops every subscription and the whole birth cache on each +/// role change — producing a data gap on the Secondary→Primary transition, which is precisely +/// the moment a plant most needs values flowing. A second connection costs one extra client id +/// and leaves ingest untouched. +/// +/// +/// Primary-gated, by owner decision (2026-07-31). Both nodes of a redundant pair +/// subscribe independently — MQTT fan-out is per-connection and the Primary gate applies +/// downstream of ingest — so without gating, both would claim the same host id. The sharp +/// failure is not the two fighting over the retained STATE: it is that the +/// Secondary's death would publish OFFLINE, edge nodes would stop publishing, +/// and the surviving Primary would be perfectly healthy throughout. Redundancy would cause the +/// outage the mechanism exists to prevent. The gate is +/// , sampled on a timer. +/// +/// +/// Accepted trade: a brief edge-visible OFFLINE window during failover, while the +/// old Primary stands down and the new one connects and republishes. That is a genuine state +/// transition — for a window there really is no host — not a glitch to paper over. +/// +/// +/// Reconnect re-asserts the claim. The dedicated connection carries the same supervisor +/// as the ingest one, and Reconnected republishes the birth. Without that, a broker +/// restart would leave the host connected but its retained ONLINE gone — a live host +/// invisible to every edge node, with nothing in any log to say so. +/// +/// +public sealed class SparkplugPrimaryHost : IAsyncDisposable +{ + /// + /// Appended to the driver's configured client id for the STATE connection. Two MQTT clients + /// presenting one client id is not reported as an error — the broker silently evicts the older + /// one — so ingest and the host claim would take turns killing each other forever. + /// + internal const string ClientIdSuffix = "-spbhost"; + + /// How often the role gate is sampled. Bounds the failover republish latency. + private static readonly TimeSpan RoleSampleInterval = TimeSpan.FromSeconds(1); + + private readonly MqttDriverOptions _options; + private readonly string _hostId; + private readonly string _driverId; + private readonly IDriverPrimaryRoleSource _roleSource; + private readonly ILogger? _logger; + private readonly TimeProvider _timeProvider; + + /// Serializes claim/release against each other and against . + private readonly SemaphoreSlim _gate = new(1, 1); + + private CancellationTokenSource? _loopCts; + + /// The live claim, or null when this node is not currently the primary host. + private MqttConnection? _connection; + + /// The session bodies backing ; non-null exactly when it is. + private SparkplugStateSession? _session; + + private int _disposed; + + /// Creates a primary-host claim. Opens nothing — does that. + /// Broker connection settings; the host id comes from its Sparkplug section. + /// Driver instance id, for log correlation. + /// The Primary data-plane verdict this claim is gated on. + /// Optional logger. + /// Clock seam; defaults to . + /// + /// The options carry no Sparkplug host id — exists so a caller never has + /// to hit this. + /// + public SparkplugPrimaryHost( + MqttDriverOptions options, + string driverId, + IDriverPrimaryRoleSource roleSource, + ILogger? logger = null, + TimeProvider? timeProvider = null) + { + ArgumentNullException.ThrowIfNull(options); + ArgumentException.ThrowIfNullOrWhiteSpace(driverId); + ArgumentNullException.ThrowIfNull(roleSource); + + var hostId = options.Sparkplug?.HostId; + if (string.IsNullOrWhiteSpace(hostId)) + { + throw new ArgumentException( + "A Sparkplug primary-host claim needs a hostId; check CanClaim first.", nameof(options)); + } + + _options = options; + _hostId = hostId.Trim(); + _driverId = driverId; + _roleSource = roleSource; + _logger = logger; + _timeProvider = timeProvider ?? TimeProvider.System; + } + + /// + /// Whether asks for — and can support — a primary-host claim: + /// Sparkplug mode, actAsPrimaryHost set, and a non-blank hostId. + /// + /// The driver options to test. + /// when a claim should be made. + /// + /// The blank-host-id case is a configuration error the AdminUI form already refuses, but the + /// driver must not fault on it: the config could have been authored before that validation + /// existed. It is reported by and the flag is treated as off. + /// + public static bool CanClaim(MqttDriverOptions options) => + options is { Mode: MqttMode.SparkplugB, Sparkplug: { ActAsPrimaryHost: true } spb } + && !string.IsNullOrWhiteSpace(spb.HostId); + + /// + /// Describes why an actAsPrimaryHost request cannot be honoured, or null when it can be + /// (or was never made). + /// + /// The driver options to test. + /// An operator-facing reason, or . + public static string? DescribeUnusable(MqttDriverOptions options) + { + if (options?.Sparkplug is not { ActAsPrimaryHost: true } spb) + { + return null; + } + + if (options.Mode != MqttMode.SparkplugB) + { + return "actAsPrimaryHost is set but the driver is not in SparkplugB mode; there is no host state to publish."; + } + + return string.IsNullOrWhiteSpace(spb.HostId) + ? "actAsPrimaryHost is set but hostId is blank; a host claim has no topic to publish on." + : null; + } + + /// The STATE topic this claim publishes on. + public string StateTopic => SparkplugTopic.FormatState(_hostId); + + /// + /// Whether the claim is currently held — i.e. a retained ONLINE has been published and a + /// Will is armed behind it. + /// + public bool IsClaimed => Volatile.Read(ref _connection) is not null; + + /// + /// Test seam: how many times a claim connect has been attempted, successful or not. + /// Always ignored in production. + /// + /// + /// Without this, "a Secondary does not claim" is a vacuous assertion. The obvious test — + /// start a Secondary and assert is false — passes identically whether + /// the gate works or whether the gate is deleted and the connect simply fails against a broker + /// the test cannot reach. Counting attempts separates "did not try" from "tried and failed", + /// which is the whole distinction the gate is. + /// + internal int ClaimAttemptsForTests => Volatile.Read(ref _claimAttempts); + + private int _claimAttempts; + + /// + /// Test seam: kills the claim connection the way a crashed process does — no clean + /// DISCONNECT and no explicit stand-down publish — so the broker delivers the Last Will. + /// See for why this cannot be reached through the + /// ordinary lifecycle. + /// + /// A task completing once the connection is gone. + internal async Task AbandonClaimForTests() + { + // Stop the reconcile loop first, or the next tick sees no connection, decides the claim is + // lost, and re-claims — republishing ONLINE over the Will this is meant to observe. + var loop = Interlocked.Exchange(ref _loopCts, null); + if (loop is not null) + { + await loop.CancelAsync().ConfigureAwait(false); + loop.Dispose(); + } + + var connection = Interlocked.Exchange(ref _connection, null); + _session = null; + if (connection is not null) + { + await connection.AbandonForTests().ConfigureAwait(false); + } + } + + /// + /// Starts the role-reconcile loop. Returns once the first reconcile has run, so a caller that + /// starts as Primary has published its ONLINE before this completes. + /// + /// Cancels the initial reconcile. + /// A task completing after the first reconcile. + public async Task StartAsync(CancellationToken cancellationToken) + { + ObjectDisposedException.ThrowIf(Volatile.Read(ref _disposed) != 0, this); + + var cts = new CancellationTokenSource(); + var previous = Interlocked.Exchange(ref _loopCts, cts); + if (previous is not null) + { + await previous.CancelAsync().ConfigureAwait(false); + previous.Dispose(); + } + + // Reconcile once inline. A failure here is logged and swallowed, never rethrown: the driver's + // ingest path is the data plane, and refusing to start it because a broker would not accept a + // host claim would trade every tag's value for an advisory message. The loop retries. + await ReconcileAsync(cancellationToken).ConfigureAwait(false); + + _ = Task.Run(() => ReconcileLoopAsync(cts.Token), CancellationToken.None); + } + + /// + /// Stands the claim down cleanly: publishes an explicit retained OFFLINE, then + /// disconnects. Idempotent, and safe to call when no claim is held. + /// + /// Bounds the stand-down publish. + /// A task completing when the claim is released. + public async Task StopAsync(CancellationToken cancellationToken) + { + var loop = Interlocked.Exchange(ref _loopCts, null); + if (loop is not null) + { + await loop.CancelAsync().ConfigureAwait(false); + loop.Dispose(); + } + + await ReleaseAsync("shutdown", cancellationToken).ConfigureAwait(false); + } + + /// + public async ValueTask DisposeAsync() + { + if (Interlocked.Exchange(ref _disposed, 1) != 0) + { + return; + } + + // Bounded: dispose must not hang on an unreachable broker. A stand-down we cannot publish is + // covered by the Will the moment the socket closes, which is the whole point of having one. + using var deadline = new CancellationTokenSource(TimeSpan.FromSeconds(_options.ConnectTimeoutSeconds)); + try + { + await StopAsync(deadline.Token).ConfigureAwait(false); + } + catch (Exception ex) + { + _logger?.LogDebug( + ex, "MQTT driver '{DriverId}': Sparkplug primary-host stand-down failed during dispose.", _driverId); + } + + _gate.Dispose(); + } + + private async Task ReconcileLoopAsync(CancellationToken ct) + { + try + { + while (!ct.IsCancellationRequested) + { + await Task.Delay(RoleSampleInterval, _timeProvider, ct).ConfigureAwait(false); + await ReconcileAsync(ct).ConfigureAwait(false); + } + } + catch (OperationCanceledException) + { + // Shutdown. + } + catch (Exception ex) + { + // The loop must not die silently — a dead loop means the claim freezes in whatever state + // it was last in, including "claimed by a node that is no longer Primary". + _logger?.LogError( + ex, "MQTT driver '{DriverId}': Sparkplug primary-host reconcile loop stopped.", _driverId); + } + } + + /// + /// Brings the held claim into line with the current role verdict. Never throws — a transient + /// broker failure leaves the claim as it was and the next tick tries again. + /// + private async Task ReconcileAsync(CancellationToken cancellationToken) + { + try + { + if (_roleSource.ShouldActAsPrimary) + { + await ClaimAsync(cancellationToken).ConfigureAwait(false); + } + else + { + await ReleaseAsync("this node is not the redundancy Primary", cancellationToken).ConfigureAwait(false); + } + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + throw; + } + catch (Exception ex) + { + _logger?.LogWarning( + ex, + "MQTT driver '{DriverId}': could not reconcile the Sparkplug primary-host claim for '{HostId}'; will retry.", + _driverId, + _hostId); + } + } + + private async Task ClaimAsync(CancellationToken cancellationToken) + { + if (Volatile.Read(ref _connection) is not null) + { + return; // Already held; the connection's own supervisor keeps it that way. + } + + await _gate.WaitAsync(cancellationToken).ConfigureAwait(false); + try + { + if (_connection is not null || Volatile.Read(ref _disposed) != 0) + { + return; + } + + Interlocked.Increment(ref _claimAttempts); + + // ONE timestamp for both bodies — the Will is built here, before CONNECT, and the birth + // published after CONNACK. See SparkplugStatePayload's remarks on why they must match. + var session = SparkplugStatePayload.ForSession( + _hostId, _timeProvider.GetUtcNow().ToUnixTimeMilliseconds()); + + var connection = new MqttConnection( + _options, + _driverId, + _logger, + clientIdSuffix: ClientIdSuffix, + lastWill: new MqttLastWill( + session.Topic, session.OfflineBody, SparkplugStatePayload.StateQos, Retain: true), + // Forced clean, ignoring the driver's configured value: a broker that resumes a prior + // session for this client id may keep the OLD session's Will and discard the one this + // CONNECT carries — leaving a claim with no death certificate behind it, which is the + // single state this whole mechanism exists to make impossible. + cleanSessionOverride: true); + + // Re-assert on every reconnect. A broker restart drops retained messages; without this the + // host would be connected and believed offline by every edge node, silently. + connection.Reconnected += async ct => + { + await connection.PublishAsync( + session.Topic, session.OnlineBody, SparkplugStatePayload.StateQos, retain: true, ct) + .ConfigureAwait(false); + + _logger?.LogInformation( + "MQTT driver '{DriverId}': re-published Sparkplug primary-host ONLINE for '{HostId}' after reconnect.", + _driverId, + _hostId); + }; + + await connection.ConnectAsync(cancellationToken).ConfigureAwait(false); + + try + { + await connection.PublishAsync( + session.Topic, session.OnlineBody, SparkplugStatePayload.StateQos, retain: true, cancellationToken) + .ConfigureAwait(false); + } + catch + { + // A connection whose birth never landed is worse than no connection: it holds a Will + // that will publish OFFLINE for a claim that was never asserted, and it looks claimed + // to IsClaimed. Tear it down and let the next tick retry from scratch. + await connection.DisposeAsync().ConfigureAwait(false); + throw; + } + + _session = session; + Volatile.Write(ref _connection, connection); + + _logger?.LogInformation( + "MQTT driver '{DriverId}': claimed Sparkplug primary host '{HostId}' — retained ONLINE published on '{Topic}', " + + "OFFLINE armed as the Last Will.", + _driverId, + _hostId, + session.Topic); + } + finally + { + _gate.Release(); + } + } + + private async Task ReleaseAsync(string reason, CancellationToken cancellationToken) + { + if (Volatile.Read(ref _connection) is null) + { + return; + } + + await _gate.WaitAsync(CancellationToken.None).ConfigureAwait(false); + try + { + var connection = Interlocked.Exchange(ref _connection, null); + var session = _session; + _session = null; + if (connection is null || session is null) + { + return; + } + + try + { + // Explicit OFFLINE before disconnecting. The Will would eventually say the same thing, + // but only after the broker's keep-alive timeout notices the socket is gone — up to + // 1.5 keep-alive periods of edge nodes believing a host that has already stood down is + // still online. Publishing it makes the transition immediate. + await connection.PublishAsync( + session.Topic, session.OfflineBody, SparkplugStatePayload.StateQos, retain: true, cancellationToken) + .ConfigureAwait(false); + + _logger?.LogInformation( + "MQTT driver '{DriverId}': released Sparkplug primary host '{HostId}' ({Reason}) — retained OFFLINE published.", + _driverId, + _hostId, + reason); + } + catch (Exception ex) + { + // Not fatal, and not silent. The Will still covers this: dropping the connection below + // makes the broker publish the identical OFFLINE body. The only thing lost is promptness. + _logger?.LogWarning( + ex, + "MQTT driver '{DriverId}': could not publish the Sparkplug primary-host OFFLINE for '{HostId}'; " + + "the Last Will will deliver it when the session drops.", + _driverId, + _hostId); + } + + await connection.DisposeAsync().ConfigureAwait(false); + } + finally + { + _gate.Release(); + } + } +} diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.csproj b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.csproj index 79edc634..2a88db4e 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.csproj +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.csproj @@ -25,6 +25,10 @@ + + diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Shared/Drivers/Forms/MqttDriverForm.razor b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Shared/Drivers/Forms/MqttDriverForm.razor index cb18f6bc..bef3d0d3 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Shared/Drivers/Forms/MqttDriverForm.razor +++ b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Shared/Drivers/Forms/MqttDriverForm.razor @@ -225,7 +225,7 @@
-
Sparkplug Host Application identity. Receive-only in this version.
+
Sparkplug Host Application identity — the spBv1.0/STATE/{hostId} topic. Required to act as Primary Host.
@@ -237,25 +237,42 @@
@if (_form.SparkplugActAsPrimaryHost) { - @* Shown rather than the flag being hidden: an operator who needs a primary host must - learn it is absent HERE, not from a plant that never sees a STATE message. The - driver logs the same warning at startup so the two cannot drift. *@ -
- + } + else + { +
+ +
+ } }
@@ -263,7 +280,6 @@ Sparkplug tags bind by the group / edge node / device / metric tuple, not by topic — author them with Browse device… on a device under this driver, which builds its tree from observed birth certificates. - MQTT tags are read-only (write-through is not implemented).
} diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Shared/Drivers/Forms/MqttDriverFormModel.cs b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Shared/Drivers/Forms/MqttDriverFormModel.cs index a003bc63..ebc3c40f 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Shared/Drivers/Forms/MqttDriverFormModel.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Shared/Drivers/Forms/MqttDriverFormModel.cs @@ -142,17 +142,23 @@ public sealed class MqttDriverFormModel public string SparkplugGroupId { get; set; } = ""; /// - /// Sparkplug mode: the Host Application identity. Optional, and receive-only in this - /// version — see . + /// Sparkplug mode: the Host Application identity — the spBv1.0/STATE/{hostId} topic the + /// driver observes, and publishes on when is set. + /// Optional for observation; required to claim. /// public string SparkplugHostId { get; set; } = ""; /// - /// Sparkplug mode: claim the Primary Host Application role. Not implemented — STATE - /// publishing is out of scope, and the driver logs a warning rather than being silently inert - /// when this is set. Surfaced (rather than hidden) so an operator who needs it learns that from - /// the form instead of from a quiet plant. + /// Sparkplug mode: claim the Primary Host Application role (Gitea #511) — a retained + /// ONLINE, a matching OFFLINE armed as the MQTT Last Will, and an explicit + /// OFFLINE on clean shutdown. /// + /// + /// Primary-gated: only the node holding the redundancy Primary role claims it, so the two + /// nodes of a pair cannot fight over one host id. At most one primary host may exist per host id + /// per broker, which the form states — it is not enforceable from here, since a second + /// deployment pointing at the same broker is invisible to this one. + /// public bool SparkplugActAsPrimaryHost { get; set; } /// Sparkplug mode: answer a detected sequence-number gap with a rebirth request (NCMD). diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Shared/Drivers/Forms/SparkplugPrimaryHostAdvice.cs b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Shared/Drivers/Forms/SparkplugPrimaryHostAdvice.cs new file mode 100644 index 00000000..1601d9aa --- /dev/null +++ b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Shared/Drivers/Forms/SparkplugPrimaryHostAdvice.cs @@ -0,0 +1,35 @@ +namespace ZB.MOM.WW.OtOpcUa.AdminUI.Components.Shared.Drivers.Forms; + +/// +/// The rule deciding which primary-host advisory MqttDriverForm shows when +/// Act as Primary Host is on (Gitea #511). +/// +/// +/// +/// A plain static rather than markup logic, because this repo has no bUnit. A condition +/// written inline in a .razor file is unreachable from any test, and the failure mode +/// here is specific: an operator who ticks the box with a blank host id must see a +/// refusal, not the ordinary "here is what will happen" note. Getting that backwards +/// tells them a claim is being made when the driver is in fact declining to make one. +/// +/// +/// Deliberately mirrors SparkplugPrimaryHost.DescribeUnusable's blank-host-id arm. The +/// mode arm is not duplicated: this panel only renders inside the SparkplugB branch of the +/// form, so mode cannot be wrong here. +/// +/// +public static class SparkplugPrimaryHostAdvice +{ + /// + /// Describes why a primary-host claim cannot be made for , or + /// when it can. + /// + /// The Sparkplug host id as typed into the form. + /// An operator-facing reason, or . + public static string? DescribeBlockingProblem(string? hostId) => + string.IsNullOrWhiteSpace(hostId) + ? "A Sparkplug host claim is published on spBv1.0/STATE/{hostId}, so it needs a Host ID. " + + "Set one above, or turn this off — the driver will otherwise log a warning at startup " + + "and publish nothing." + : null; +} 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 3b743031..29037bfe 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Drivers/DriverFactoryBootstrap.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Drivers/DriverFactoryBootstrap.cs @@ -55,7 +55,16 @@ public static class DriverFactoryBootstrap // The Galaxy driver resolves its Gateway.ApiKeySecretRef secret: arm through the // shared ISecretResolver (registered unconditionally by AddZbSecrets on every node). var secretResolver = sp.GetRequiredService(); - Register(registry, loggerFactory, secretResolver, scriptRoot); + // Gitea #511: the MQTT driver's Sparkplug primary-host claim is Primary-gated, and unlike + // every other gated operation it is a STANDING assertion on the driver's own connection — + // there is no inbound message for DriverHostActor to intercept — so the verdict has to + // reach the driver itself. Registered unconditionally on driver-role nodes by + // AddOtOpcUaRuntime. GetRequiredService, not GetService: a null here silently falls the + // driver back to "claim unconditionally", which on a redundant pair is the dual-claim this + // whole mechanism exists to prevent, and it would fail as a plant outage rather than as a + // startup error. + var primaryRoleSource = sp.GetRequiredService(); + Register(registry, loggerFactory, secretResolver, primaryRoleSource, scriptRoot); return registry; }); services.AddSingleton(sp => @@ -142,6 +151,7 @@ public static class DriverFactoryBootstrap DriverFactoryRegistry registry, ILoggerFactory? loggerFactory, ISecretResolver secretResolver, + IDriverPrimaryRoleSource primaryRoleSource, ZB.MOM.WW.OtOpcUa.Core.Scripting.ScriptRootLogger? scriptRoot = null) { Driver.AbCip.AbCipDriverFactoryExtensions.Register(registry); @@ -150,7 +160,7 @@ public static class DriverFactoryBootstrap Driver.FOCAS.FocasDriverFactoryExtensions.Register(registry); Driver.Galaxy.GalaxyDriverFactoryExtensions.Register(registry, secretResolver, loggerFactory); Driver.Modbus.ModbusDriverFactoryExtensions.Register(registry, loggerFactory); - Driver.Mqtt.MqttDriverFactoryExtensions.Register(registry, loggerFactory); + Driver.Mqtt.MqttDriverFactoryExtensions.Register(registry, loggerFactory, primaryRoleSource); // Tier A (the Register default): fully managed — HttpClient + System.Xml.Linq, no native SDK, // no COM. Tier C is the only tier that arms process-level recycle (MemoryRecycle hard-breach / // ScheduledRecycleScheduler), which would be wrong here: the driver is in-process and killing diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs index 5359cfa5..6d257dbd 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs @@ -273,6 +273,19 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers /// today, the alarm store-and-forward drain. Null on nodes that do not wire one. private readonly IRedundancyRoleView? _redundancyRoleView; + /// + /// Where the Primary data-plane verdict is published for drivers that must know it + /// directly (Gitea #511). Null on nodes that wire no redundancy manager. + /// + /// + /// Note the different verdict from : that one carries + /// , which OPENS on an unknown role. A + /// Sparkplug host-id claim is an exclusive assertion on a shared broker, so it takes the + /// data-plane verdict, which closes. Publishing the drain verdict here would let both nodes of + /// a pair claim the host id during every boot window. + /// + private readonly IDriverPrimaryRoleSource? _primaryRoleSource; + /// /// Host of this node's LocalDb replication partner — the only node that holds a copy of this /// node's alarm queue, and therefore the only node it may ever stand down in favour of. @@ -408,7 +421,8 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers bool fetchAndCacheMode = false, IDeploymentArtifactFetcher? artifactFetcher = null, IAlarmStateStore? alarmStateStore = null, - ITelemetryLocalHub? telemetryHub = null) => + ITelemetryLocalHub? telemetryHub = null, + IDriverPrimaryRoleSource? primaryRoleSource = null) => // WARNING: this forwarding list is POSITIONAL, and Props.Create compiles it into an // expression tree. Six IActorRef? parameters and several interface-typed ones mean a // mis-ordered argument is usually type-compatible and therefore compiles clean, then binds @@ -419,7 +433,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers healthPublisher, virtualTagEvaluator, historyWriter, virtualTagHostOverride, loggerFactory, scriptRootLogger, scriptedAlarmHostOverride, invokerFactory, driverMemberCountProvider, deploymentArtifactCache, redundancyRoleView, replicationPeerHost, fetchAndCacheMode, artifactFetcher, - alarmStateStore, telemetryHub)); + alarmStateStore, telemetryHub, primaryRoleSource)); /// Initializes a new DriverHostActor with the specified dependencies. /// Database context factory for configuration database access. @@ -460,6 +474,12 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers /// Per-cluster mesh Phase 5: optional node-local live-telemetry hub each /// Primary-gated native-alarm transition is also emitted into, and which is threaded into the spawned /// VirtualTag + ScriptedAlarm hosts. Null (admin-only nodes / tests) skips the emit and passes null down. + /// Gitea #511: where the Primary data-plane verdict is published + /// for drivers that assert a standing exclusive identity on a shared external system — today the MQTT + /// driver's Sparkplug primary-host STATE claim. Deliberately a separate parameter from + /// even though the production object implements both: a test that + /// supplies only a role-view double must not silently also satisfy this, or its primary-role assertions + /// pass without the publish ever happening. Null on nodes that wire no redundancy manager. public DriverHostActor( IDbContextFactory? dbFactory, CommonsNodeId localNode, @@ -483,11 +503,13 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers bool fetchAndCacheMode = false, IDeploymentArtifactFetcher? artifactFetcher = null, IAlarmStateStore? alarmStateStore = null, - ITelemetryLocalHub? telemetryHub = null) + ITelemetryLocalHub? telemetryHub = null, + IDriverPrimaryRoleSource? primaryRoleSource = null) { _telemetryHub = telemetryHub; _deploymentArtifactCache = deploymentArtifactCache; _redundancyRoleView = redundancyRoleView; + _primaryRoleSource = primaryRoleSource; _replicationPeerHost = replicationPeerHost; _fetchAndCacheMode = fetchAndCacheMode; _artifactFetcher = artifactFetcher; @@ -1331,6 +1353,12 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers _redundancyRoleView?.Publish( PrimaryGatePolicy.ShouldDrainAlarmHistory(_localRole, queueSharingPeerIsPrimary: peerIsPrimary)); + + // The DATA-PLANE verdict, deliberately — the same one HandleRouteNodeWrite gates on, not the + // drain verdict published immediately above. A driver reading this claims an exclusive identity + // on a shared external system (Gitea #511: the Sparkplug host id), where two claimants is the + // harm, so an unknown role with a driver peer present must deny. + _primaryRoleSource?.PublishPrimaryRole(ShouldServiceAsPrimary()); } private void Stale() diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Redundancy/IRedundancyRoleView.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Redundancy/IRedundancyRoleView.cs index a9c41dff..b450e9e2 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Redundancy/IRedundancyRoleView.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Redundancy/IRedundancyRoleView.cs @@ -1,3 +1,4 @@ +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; using ZB.MOM.WW.OtOpcUa.Runtime.Drivers; namespace ZB.MOM.WW.OtOpcUa.Runtime.Redundancy; @@ -39,8 +40,22 @@ public interface IRedundancyRoleView void Publish(bool shouldDrainAlarmHistory); } -/// Thread-safe backed by a volatile field. -public sealed class RedundancyRoleView : IRedundancyRoleView +/// +/// Thread-safe + , each +/// backed by its own volatile field. +/// +/// +/// +/// Two verdicts, one object, on purpose. Both are published from the same +/// DriverHostActor snapshot handler and both are read off-actor, so a second class +/// would duplicate the publish plumbing to hold a second bool. They are kept as distinct +/// properties rather than collapsed into one because they are genuinely different +/// decisions from different policies — see 's remarks on why +/// the drain gate must open on an unknown role where the data-plane gate closes. Collapsing +/// them is the bug that live gate already found once. +/// +/// +public sealed class RedundancyRoleView : IRedundancyRoleView, IDriverPrimaryRoleSource { // Seeded with the decision for "role unknown" rather than a bare false. An unpublished view and a // node whose role nothing ever reports are genuinely the same situation, and they must behave the @@ -49,9 +64,22 @@ public sealed class RedundancyRoleView : IRedundancyRoleView private volatile bool _shouldDrainAlarmHistory = PrimaryGatePolicy.ShouldDrainAlarmHistory(localRole: null, queueSharingPeerIsPrimary: false); + // Seeded from the same policy that will later publish here, for the unknown-role/no-peer case: + // a node nothing has reported a role for, on a cluster with no driver peer, is a single-node + // deployment and may claim. The moment a snapshot arrives this is replaced by the real verdict, + // and a multi-driver cluster's unknown role denies. + private volatile bool _shouldActAsPrimary = + PrimaryGatePolicy.ShouldServiceAsPrimary(localRole: null, driverMemberCount: 1); + /// public bool ShouldDrainAlarmHistory => _shouldDrainAlarmHistory; + /// + public bool ShouldActAsPrimary => _shouldActAsPrimary; + /// public void Publish(bool shouldDrainAlarmHistory) => _shouldDrainAlarmHistory = shouldDrainAlarmHistory; + + /// + public void PublishPrimaryRole(bool shouldActAsPrimary) => _shouldActAsPrimary = shouldActAsPrimary; } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs index 12648380..793088f0 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs @@ -75,6 +75,22 @@ public static class ServiceCollectionExtensions // telemetry (never DPS) to the Phase-5 gRPC streaming service central dials in on. AddOtOpcUaRuntime // runs only inside Program.cs's hasDriver block, so this lands on driver-role nodes only. services.TryAddSingleton(); + + // Gitea #511: the Primary data-plane verdict, published for drivers that assert a standing + // exclusive identity on a shared external system (today the MQTT driver's Sparkplug host id). + // + // Registered HERE, unconditionally on every driver-role node, and NOT alongside the alarm + // drain's IRedundancyRoleView — which AddAlarmHistorian registers only when that section is + // enabled. Piggybacking on it would leave a redundant pair with no alarm historian configured + // with no primary-role source at all, and the driver's documented no-source fallback is + // "claim" — so both nodes of that pair would claim the same host id and their Wills would + // fight over the retained STATE. Exactly the failure #511 exists to prevent, reached by a + // registration accident in an unrelated section. + // + // Both interfaces are forwarded to ONE RedundancyRoleView instance, so a node that has both + // publishes two verdicts into the same object rather than into two that disagree. + services.TryAddSingleton(); + services.TryAddSingleton(sp => sp.GetRequiredService()); return services; } @@ -117,8 +133,10 @@ public static class ServiceCollectionExtensions // The view is a plain singleton so it exists even on nodes whose DriverHostActor never // publishes to it — an unpublished view reads as "no peer, drain", which is the correct - // posture for a deployment that runs no redundancy at all. - services.TryAddSingleton(); + // posture for a deployment that runs no redundancy at all. Forwarded to the same instance + // AddOtOpcUaRuntime registered for IDriverPrimaryRoleSource: one object, two verdicts. + services.TryAddSingleton(); + services.TryAddSingleton(sp => sp.GetRequiredService()); // THE GATE ONLY APPLIES TO A REPLICATED QUEUE. Standing down is only ever safe because some // other node holds the same rows and will send them instead — and the only node that does is @@ -499,7 +517,11 @@ public static class ServiceCollectionExtensions // Phase 5: the node-local live-telemetry hub (registered in AddOtOpcUaRuntime). The host // emits its Primary-gated native-alarm transitions into it and threads it into the // VirtualTag + ScriptedAlarm hosts it spawns. - telemetryHub: resolver.GetService()), + telemetryHub: resolver.GetService(), + // Gitea #511: where this actor publishes its Primary DATA-PLANE verdict for drivers + // that claim an exclusive external identity. Registered unconditionally by + // AddOtOpcUaRuntime, unlike redundancyRoleView above — see the comment there. + primaryRoleSource: resolver.GetService()), DriverHostActorName); registry.Register(driverHost); diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/MqttPrimaryHostLiveTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/MqttPrimaryHostLiveTests.cs new file mode 100644 index 00000000..fc9f6825 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/MqttPrimaryHostLiveTests.cs @@ -0,0 +1,379 @@ +using System.Buffers; +using System.Collections.Concurrent; +using MQTTnet; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Sparkplug; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests; + +/// +/// Live gate for the Sparkplug primary-host claim (Gitea #511), against the Mosquitto fixture. +/// +/// +/// +/// The issue's acceptance criteria, leg by leg: the broker holds a retained ONLINE while +/// the claim runs; an abrupt death leaves OFFLINE (the Will fires); a clean shutdown +/// leaves OFFLINE; and the pair-node question is answered rather than left to chance — +/// a node that is not Primary claims nothing, and a role change moves the claim. +/// +/// +/// Each test uses its own host id. A Sparkplug host id is a broker-wide singleton by +/// definition, and the retained STATE outlives the test that wrote it — two tests sharing an id +/// would see each other's leftovers, and would do so intermittently depending on ordering. +/// +/// +/// Retained messages are cleaned up. A retained ONLINE for a host that no longer +/// exists is exactly the lie this feature prevents, so leaving one on a shared fixture would be +/// poor manners at best and a confusing false positive for the next run at worst. +/// +/// +[Trait("Category", "Integration")] +[Collection("MqttFixture")] +public sealed class MqttPrimaryHostLiveTests(MqttFixture fixture) +{ + private static readonly TimeSpan Timeout = TimeSpan.FromSeconds(15); + + private void SkipIfUnconfigured() => Assert.SkipWhen(fixture.SkipReason is not null, fixture.SkipReason ?? ""); + + [Fact] + public async Task A_primary_publishes_a_retained_ONLINE_that_a_late_subscriber_still_sees() + { + // Retention is the point, not an optimisation: an edge node that connects AFTER the host came + // up must still learn the host is alive. A non-retained ONLINE reaches only whoever happened to + // be subscribed at that instant, and every later joiner sees an absent host forever. + SkipIfUnconfigured(); + + var hostId = NewHostId(); + var ct = TestContext.Current.CancellationToken; + await using var host = NewHost(hostId, isPrimary: true); + + await host.StartAsync(ct); + host.IsClaimed.ShouldBeTrue("the fixture broker accepted the claim connection"); + + // Subscribe only NOW — after the birth was published — so only a retained message can satisfy it. + await using var watcher = await StateWatcher.StartAsync(fixture, hostId, ct); + var body = await watcher.WaitForOneAsync(Timeout, ct); + + SparkplugStatePayload.TryDecode(body, out var online, out var timestamp).ShouldBeTrue(); + online.ShouldBeTrue(); + timestamp.ShouldBeGreaterThan(0); + + await host.StopAsync(ct); + await ClearRetainedAsync(hostId, ct); + } + + [Fact] + public async Task An_abrupt_death_leaves_a_retained_OFFLINE_because_the_will_fires() + { + // THE leg the whole design exists for, and the one no unit test can reach: proof that the + // broker is holding our death certificate and delivers it when the socket dies without a + // goodbye. Without it a crashed node would leave a retained ONLINE asserting forever that a + // dead host is alive — strictly worse than never publishing at all. + SkipIfUnconfigured(); + + var hostId = NewHostId(); + var ct = TestContext.Current.CancellationToken; + var host = NewHost(hostId, isPrimary: true); + + await host.StartAsync(ct); + host.IsClaimed.ShouldBeTrue(); + + await using var watcher = await StateWatcher.StartAsync(fixture, hostId, ct); + var birth = await watcher.WaitForOneAsync(Timeout, ct); + SparkplugStatePayload.TryDecode(birth, out var wasOnline, out var birthTimestamp).ShouldBeTrue(); + wasOnline.ShouldBeTrue(); + + // Kill the socket with no DISCONNECT. A clean teardown would make the broker DISCARD the Will + // by protocol rule, so this test would then be observing the explicit stand-down publish and + // would pass against an implementation that never armed a Will. + await host.AbandonClaimForTests(); + + var death = await watcher.WaitForMatchAsync( + p => SparkplugStatePayload.TryDecode(p, out var o, out _) && !o, Timeout, ct); + + SparkplugStatePayload.TryDecode(death, out _, out var deathTimestamp).ShouldBeTrue(); + + // Sparkplug 3.0 pairs a host's birth and death by timestamp — an edge node may ignore an + // OFFLINE older than the ONLINE it holds. Mismatched timestamps produce two well-formed + // messages and a death certificate conformant hosts are entitled to throw away. + deathTimestamp.ShouldBe(birthTimestamp); + + await host.DisposeAsync(); + await ClearRetainedAsync(hostId, ct); + } + + [Fact] + public async Task A_clean_shutdown_leaves_a_retained_OFFLINE_without_waiting_for_the_will() + { + SkipIfUnconfigured(); + + var hostId = NewHostId(); + var ct = TestContext.Current.CancellationToken; + var host = NewHost(hostId, isPrimary: true); + + await host.StartAsync(ct); + await using var watcher = await StateWatcher.StartAsync(fixture, hostId, ct); + await watcher.WaitForMatchAsync( + p => SparkplugStatePayload.TryDecode(p, out var o, out _) && o, Timeout, ct); + + await host.StopAsync(ct); + + // Promptly — the assertion is that this arrives well inside the keep-alive window it would take + // the broker to notice a dead socket, which is why the explicit publish exists at all. + await watcher.WaitForMatchAsync( + p => SparkplugStatePayload.TryDecode(p, out var o, out _) && !o, Timeout, ct); + + await host.DisposeAsync(); + await ClearRetainedAsync(hostId, ct); + } + + [Fact] + public async Task A_node_that_is_not_primary_publishes_nothing_at_all() + { + // The pair-node answer. Not merely "does not win the race" — a Secondary must never open a + // claim connection, because the Will it would register is the hazard: its own death would then + // publish OFFLINE over a perfectly healthy Primary's ONLINE and stop the whole group. + SkipIfUnconfigured(); + + var hostId = NewHostId(); + var ct = TestContext.Current.CancellationToken; + await using var watcher = await StateWatcher.StartAsync(fixture, hostId, ct); + await using var host = NewHost(hostId, isPrimary: false); + + await host.StartAsync(ct); + host.ClaimAttemptsForTests.ShouldBe(0); + + // Give the reconcile loop several ticks to misbehave in. + await Task.Delay(TimeSpan.FromSeconds(3), ct); + + host.IsClaimed.ShouldBeFalse(); + host.ClaimAttemptsForTests.ShouldBe(0); + watcher.Count.ShouldBe(0, "a Secondary must not touch the STATE topic"); + } + + [Fact] + public async Task A_failover_moves_the_claim_offline_then_online_again() + { + // Both halves of the accepted trade, observed: losing Primary publishes OFFLINE (a genuine + // state transition — for a window there really is no host), and regaining it republishes + // ONLINE. The republish is the half that would be easy to omit; without it a node that failed + // back would run as Primary while every edge node believed it was gone. + SkipIfUnconfigured(); + + var hostId = NewHostId(); + var ct = TestContext.Current.CancellationToken; + var role = new MutableRole(isPrimary: true); + await using var host = new SparkplugPrimaryHost(Options(hostId), "live-failover", role); + + await using var watcher = await StateWatcher.StartAsync(fixture, hostId, ct); + + await host.StartAsync(ct); + await watcher.WaitForMatchAsync( + p => SparkplugStatePayload.TryDecode(p, out var o, out _) && o, Timeout, ct); + + role.IsPrimary = false; + await watcher.WaitForMatchAsync( + p => SparkplugStatePayload.TryDecode(p, out var o, out _) && !o, Timeout, ct); + await WaitUntilAsync(() => !host.IsClaimed, ct); + + var offlineSeen = watcher.CountMatching( + p => SparkplugStatePayload.TryDecode(p, out var o, out _) && !o); + + role.IsPrimary = true; + await WaitUntilAsync(() => host.IsClaimed, ct); + + // A SECOND online, after the offline — not merely "an online exists", which the birth already + // satisfied before the failover even started. + await WaitUntilAsync( + () => watcher.CountMatching(p => SparkplugStatePayload.TryDecode(p, out var o, out _) && o) >= 2, + ct); + + offlineSeen.ShouldBeGreaterThanOrEqualTo(1); + + await host.StopAsync(ct); + await ClearRetainedAsync(hostId, ct); + } + + // ---- helpers ----------------------------------------------------------------------------------- + + /// A per-test host id: the retained STATE outlives the test that wrote it. + private static string NewHostId() => $"otopcua-live-{Guid.NewGuid():N}"[..28]; + + private MqttDriverOptions Options(string hostId) => new() + { + Host = fixture.PlainHost, + Port = fixture.PlainPort, + Username = fixture.Username, + Password = fixture.Password, + UseTls = false, + ClientId = $"otopcua-live-{Guid.NewGuid():N}"[..20], + ConnectTimeoutSeconds = 10, + Mode = MqttMode.SparkplugB, + Sparkplug = new MqttSparkplugOptions + { + GroupId = "OtOpcUaSim", + HostId = hostId, + ActAsPrimaryHost = true, + }, + }; + + private SparkplugPrimaryHost NewHost(string hostId, bool isPrimary) => + new(Options(hostId), $"live-{hostId}", new MutableRole(isPrimary)); + + /// + /// Clears the retained STATE this test left behind — a zero-length retained publish is MQTT's + /// "forget this". Leaving a retained ONLINE for a host that no longer exists is precisely the + /// lie the feature prevents. + /// + private async Task ClearRetainedAsync(string hostId, CancellationToken ct) + { + var client = new MqttClientFactory().CreateMqttClient(); + try + { + await client.ConnectAsync( + new MqttClientOptionsBuilder() + .WithTcpServer(fixture.PlainHost, fixture.PlainPort) + .WithCredentials(fixture.Username, fixture.Password) + .WithClientId($"otopcua-live-clean-{Guid.NewGuid():N}"[..23]) + .WithTimeout(TimeSpan.FromSeconds(10)) + .Build(), + ct); + + await client.PublishAsync( + new MqttApplicationMessageBuilder() + .WithTopic(SparkplugTopic.FormatState(hostId)) + .WithPayload(Array.Empty()) + .WithRetainFlag(true) + .Build(), + ct); + + await client.DisconnectAsync(cancellationToken: ct); + } + finally + { + client.Dispose(); + } + } + + private static async Task WaitUntilAsync(Func condition, CancellationToken ct) + { + var deadline = DateTime.UtcNow + Timeout; + while (!condition()) + { + if (DateTime.UtcNow >= deadline) + { + throw new TimeoutException($"Condition was not satisfied within {Timeout}."); + } + + await Task.Delay(100, ct); + } + } + + /// A role source a test can flip, standing in for a redundancy failover. + private sealed class MutableRole(bool isPrimary) : IDriverPrimaryRoleSource + { + private volatile bool _isPrimary = isPrimary; + + public bool IsPrimary + { + get => _isPrimary; + set => _isPrimary = value; + } + + public bool ShouldActAsPrimary => _isPrimary; + + public void PublishPrimaryRole(bool shouldActAsPrimary) => _isPrimary = shouldActAsPrimary; + } + + /// Subscribes one host's STATE topic and records every body seen, in order. + private sealed class StateWatcher : IAsyncDisposable + { + private readonly IMqttClient _client; + private readonly ConcurrentQueue _bodies = new(); + + private StateWatcher(IMqttClient client) => _client = client; + + public int Count => _bodies.Count; + + public static async Task StartAsync(MqttFixture fixture, string hostId, CancellationToken ct) + { + var client = new MqttClientFactory().CreateMqttClient(); + var watcher = new StateWatcher(client); + + client.ApplicationMessageReceivedAsync += args => + { + watcher._bodies.Enqueue(args.ApplicationMessage.Payload.ToArray()); + return Task.CompletedTask; + }; + + await client.ConnectAsync( + new MqttClientOptionsBuilder() + .WithTcpServer(fixture.PlainHost, fixture.PlainPort) + .WithCredentials(fixture.Username, fixture.Password) + .WithClientId($"otopcua-live-state-{Guid.NewGuid():N}"[..23]) + .WithTimeout(TimeSpan.FromSeconds(10)) + .Build(), + ct); + + await client.SubscribeAsync( + new MqttClientSubscribeOptionsBuilder() + .WithTopicFilter(f => f.WithTopic(SparkplugTopic.FormatState(hostId))) + .Build(), + ct); + + return watcher; + } + + public int CountMatching(Func predicate) => _bodies.Count(b => Safe(predicate, b)); + + public async Task WaitForOneAsync(TimeSpan timeout, CancellationToken ct) + => await WaitForMatchAsync(_ => true, timeout, ct); + + public async Task WaitForMatchAsync(Func predicate, TimeSpan timeout, CancellationToken ct) + { + var deadline = DateTime.UtcNow + timeout; + while (true) + { + if (_bodies.FirstOrDefault(b => Safe(predicate, b)) is { } match) + { + return match; + } + + if (DateTime.UtcNow >= deadline) + { + throw new TimeoutException( + $"No matching STATE message within {timeout} ({_bodies.Count} seen, none matching)."); + } + + await Task.Delay(100, ct); + } + } + + private static bool Safe(Func predicate, byte[] body) + { + try { return predicate(body); } + catch { return false; } + } + + public async ValueTask DisposeAsync() + { + try + { + if (_client.IsConnected) + { + await _client.DisconnectAsync(); + } + } + catch + { + // Teardown only. + } + finally + { + _client.Dispose(); + } + } + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/SparkplugPrimaryHostTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/SparkplugPrimaryHostTests.cs new file mode 100644 index 00000000..451a060c --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/SparkplugPrimaryHostTests.cs @@ -0,0 +1,263 @@ +using System.Text; +using MQTTnet; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Sparkplug; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests; + +/// +/// Unit tests for the Sparkplug primary-host claim (Gitea #511) — the parts that are decidable +/// without a broker: the STATE bodies, the Will registration, and the gate that decides whether a +/// claim is attempted at all. +/// +/// +/// The claim/release state machine itself needs a broker and is covered by +/// MqttPrimaryHostLiveTests in the integration suite. What is pinned here is everything a +/// live test could not tell apart from a working implementation by accident — above all that the +/// birth and the Will carry the SAME timestamp, which a live test would see as two well-formed +/// messages. +/// +[Trait("Category", "Unit")] +public sealed class SparkplugPrimaryHostTests +{ + private static MqttDriverOptions Options( + bool actAsPrimaryHost = true, + string hostId = "otopcua-primary", + MqttMode mode = MqttMode.SparkplugB) + => new() + { + Host = "broker.example", + Port = 1883, + ClientId = "otopcua", + Mode = mode, + Sparkplug = new MqttSparkplugOptions + { + GroupId = "OtOpcUaSim", + HostId = hostId, + ActAsPrimaryHost = actAsPrimaryHost, + }, + }; + + // ---- The matched-timestamp invariant -------------------------------------------------------- + + [Fact] + public void The_birth_and_the_will_carry_the_same_timestamp() + { + // Sparkplug 3.0 pairs a host's birth and death certificates by their timestamp, so an edge node + // may ignore an OFFLINE older than the ONLINE it holds. The two bodies are necessarily built at + // different moments — the Will goes in the CONNECT packet, the birth is published after CONNACK + // — so reading the clock twice is the natural mistake, and it produces two individually + // well-formed messages that no live test would flag. + var session = SparkplugStatePayload.ForSession("otopcua-primary", 1_770_000_000_123); + + SparkplugStatePayload.TryDecode(session.OnlineBody, out var online, out var birthTs).ShouldBeTrue(); + SparkplugStatePayload.TryDecode(session.OfflineBody, out var offline, out var deathTs).ShouldBeTrue(); + + online.ShouldBeTrue(); + offline.ShouldBeFalse(); + birthTs.ShouldBe(1_770_000_000_123); + deathTs.ShouldBe(birthTs); + } + + [Fact] + public void The_state_topic_is_the_v3_form() + => SparkplugStatePayload.ForSession("scada-primary", 1).Topic.ShouldBe("spBv1.0/STATE/scada-primary"); + + [Fact] + public void The_timestamp_is_a_bare_json_number() + { + // A quoted timestamp still parses as a valid JSON object, so a serializer choosing string + // formatting for a long would be silently wrong to every conformant host. + var body = Encoding.UTF8.GetString(SparkplugStatePayload.Encode(online: true, 1_770_000_000_123)); + + body.ShouldBe("{\"online\":true,\"timestamp\":1770000000123}"); + } + + [Fact] + public void The_body_survives_a_non_invariant_culture() + { + var previous = Thread.CurrentThread.CurrentCulture; + Thread.CurrentThread.CurrentCulture = new System.Globalization.CultureInfo("de-DE"); + try + { + Encoding.UTF8.GetString(SparkplugStatePayload.Encode(online: false, 1_770_000_000_123)) + .ShouldBe("{\"online\":false,\"timestamp\":1770000000123}"); + } + finally + { + Thread.CurrentThread.CurrentCulture = previous; + } + } + + // ---- The Will actually reaches the CONNECT packet -------------------------------------------- + + [Fact] + public void The_will_is_registered_on_the_client_options_retained_at_qos1() + { + // The single most important assertion in this file. A retained ONLINE with no Will behind it is + // strictly worse than publishing nothing: a crashed node then asserts forever that a dead host + // is alive. This is the check that the death certificate is armed at CONNECT rather than merely + // constructed. + var session = SparkplugStatePayload.ForSession("otopcua-primary", 42); + var will = new MqttLastWill(session.Topic, session.OfflineBody, SparkplugStatePayload.StateQos, Retain: true); + + var built = MqttClientOptionsFactory.BuildClientOptions( + Options(), SparkplugPrimaryHost.ClientIdSuffix, logger: null, lastWill: will, cleanSessionOverride: true); + + built.WillTopic.ShouldBe("spBv1.0/STATE/otopcua-primary"); + built.WillRetain.ShouldBeTrue(); + built.WillQualityOfServiceLevel.ShouldBe(MQTTnet.Protocol.MqttQualityOfServiceLevel.AtLeastOnce); + SparkplugStatePayload.TryDecode(built.WillPayload, out var online, out _).ShouldBeTrue(); + online.ShouldBeFalse(); + } + + [Fact] + public void No_will_is_registered_when_none_is_supplied() + // Every other connection this repo opens — ingest, browse, probe — must stay Will-free. A Will + // leaking onto the ingest session would publish a host death when ingest merely reconnects. + => MqttClientOptionsFactory.BuildClientOptions(Options(), clientIdSuffix: null).WillTopic.ShouldBeNull(); + + [Fact] + public void The_claim_connection_forces_clean_session_over_the_configured_value() + { + // A broker resuming a previous session for this client id may keep the OLD session's Will and + // discard the one this CONNECT carries — a claim with no death certificate behind it. + var persistent = Options() with { CleanSession = false }; + + MqttClientOptionsFactory.BuildClientOptions(persistent, "-spbhost", cleanSessionOverride: true) + .CleanSession.ShouldBeTrue(); + + // ...and the override is genuinely an override, not a constant. + MqttClientOptionsFactory.BuildClientOptions(persistent, clientIdSuffix: null) + .CleanSession.ShouldBeFalse(); + } + + [Fact] + public void The_claim_connection_uses_a_distinct_client_id() + { + // Two MQTT clients presenting one client id is not an error the broker reports: it evicts the + // older one. Ingest and the claim would take turns killing each other, forever, looking like a + // flapping broker. + var ingest = MqttClientOptionsFactory.BuildClientOptions(Options(), clientIdSuffix: null).ClientId; + var claim = MqttClientOptionsFactory + .BuildClientOptions(Options(), SparkplugPrimaryHost.ClientIdSuffix).ClientId; + + claim.ShouldNotBe(ingest); + claim.ShouldBe("otopcua-spbhost"); + } + + // ---- Whether a claim is attempted at all ----------------------------------------------------- + + [Fact] + public void A_configured_claim_is_claimable() + => SparkplugPrimaryHost.CanClaim(Options()).ShouldBeTrue(); + + [Theory] + [InlineData("")] + [InlineData(" ")] + public void A_blank_host_id_is_refused_rather_than_publishing_to_a_malformed_topic(string hostId) + { + // "spBv1.0/STATE/" is a legal MQTT topic, so a publish would succeed and reach nobody. + SparkplugPrimaryHost.CanClaim(Options(hostId: hostId)).ShouldBeFalse(); + SparkplugPrimaryHost.DescribeUnusable(Options(hostId: hostId)).ShouldNotBeNullOrWhiteSpace(); + } + + [Fact] + public void The_flag_is_ignored_outside_sparkplug_mode_and_says_so() + { + var plain = Options(mode: MqttMode.Plain); + + SparkplugPrimaryHost.CanClaim(plain).ShouldBeFalse(); + SparkplugPrimaryHost.DescribeUnusable(plain).ShouldNotBeNullOrWhiteSpace(); + } + + [Fact] + public void An_unset_flag_produces_no_complaint() + { + // DescribeUnusable must stay silent about a configuration that never asked for anything, or the + // driver warns on every ordinary Sparkplug deployment. + SparkplugPrimaryHost.CanClaim(Options(actAsPrimaryHost: false)).ShouldBeFalse(); + SparkplugPrimaryHost.DescribeUnusable(Options(actAsPrimaryHost: false)).ShouldBeNull(); + SparkplugPrimaryHost.DescribeUnusable(Options(actAsPrimaryHost: false, hostId: "")).ShouldBeNull(); + } + + [Fact] + public void Constructing_a_claim_without_a_host_id_throws_rather_than_half_working() + => Should.Throw(() => + new SparkplugPrimaryHost(Options(hostId: " "), "drv-1", UnmanagedDriverPrimaryRoleSource.Instance)); + + // ---- The gate --------------------------------------------------------------------------------- + + [Fact] + public void A_driver_with_no_role_source_claims() + { + // The documented fallback for a standalone host or a test: with nothing publishing a role there + // is no peer to collide with, and defaulting closed would make the feature silently inert on the + // single-node deployment that is its commonest use. + UnmanagedDriverPrimaryRoleSource.Instance.ShouldActAsPrimary.ShouldBeTrue(); + + // Publishing into it is a no-op rather than a throw, so a caller need not special-case it. + UnmanagedDriverPrimaryRoleSource.Instance.PublishPrimaryRole(false); + UnmanagedDriverPrimaryRoleSource.Instance.ShouldActAsPrimary.ShouldBeTrue(); + } + + [Fact] + public async Task A_secondary_never_even_attempts_a_claim_connection() + { + // The whole point of gating. Asserted on ATTEMPTS, not on IsClaimed: both nodes point at the + // same unreachable broker here, so IsClaimed is false either way and an IsClaimed-only + // assertion would pass just as happily with the gate deleted. See ClaimAttemptsForTests. + var host = new SparkplugPrimaryHost( + Unreachable(), "drv-1", new FixedRole(shouldActAsPrimary: false)); + + await host.StartAsync(TestContext.Current.CancellationToken); + + host.ClaimAttemptsForTests.ShouldBe(0); + host.IsClaimed.ShouldBeFalse(); + await host.DisposeAsync(); + } + + [Fact] + public async Task A_primary_attempts_the_claim_and_survives_a_broker_it_cannot_reach() + { + // The falsifying control for the test above — same options, same broker, opposite verdict — so + // "0 attempts" there is a fact about the gate rather than about the fixture. + // + // It also pins the swallow: the claim is advisory and ingest is the data plane, so a broker that + // refuses it must cost a warning, not the driver. And IsClaimed must stay honest — a connection + // whose ONLINE never landed is not a claim. + var host = new SparkplugPrimaryHost(Unreachable(), "drv-1", new FixedRole(shouldActAsPrimary: true)); + + await Should.NotThrowAsync(() => host.StartAsync(TestContext.Current.CancellationToken)); + + host.ClaimAttemptsForTests.ShouldBe(1); + host.IsClaimed.ShouldBeFalse(); + await host.DisposeAsync(); + } + + /// A broker that refuses instantly, so a connect attempt is observable but not slow. + private static MqttDriverOptions Unreachable() => + Options() with { Host = "127.0.0.1", Port = 1, ConnectTimeoutSeconds = 1 }; + + [Fact] + public async Task Stopping_an_unclaimed_host_is_a_no_op() + { + var host = new SparkplugPrimaryHost(Options(), "drv-1", new FixedRole(shouldActAsPrimary: false)); + + await Should.NotThrowAsync(() => host.StopAsync(TestContext.Current.CancellationToken)); + + // Twice: teardown runs from both ShutdownAsync and DisposeAsync, so a second release that threw + // — or that published a second OFFLINE — would surface on every ordinary driver shutdown. + await Should.NotThrowAsync(async () => await host.DisposeAsync()); + await Should.NotThrowAsync(async () => await host.DisposeAsync()); + } + + private sealed class FixedRole(bool shouldActAsPrimary) : IDriverPrimaryRoleSource + { + public bool ShouldActAsPrimary { get; private set; } = shouldActAsPrimary; + + public void PublishPrimaryRole(bool value) => ShouldActAsPrimary = value; + } +}