Files
lmxopcua/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriverFactoryExtensions.cs
T
Joseph Doherty 7f2108dcf2 feat(mqtt): Sparkplug primary-host STATE publishing, Primary-gated (#511)
`ActAsPrimaryHost` was accepted by config and by the AdminUI form and did nothing but
log a warning. It now delivers the whole three-part contract or none of it: a retained
`ONLINE` on `spBv1.0/STATE/{hostId}`, a matching retained `OFFLINE` armed as the MQTT
Last Will, and an explicit `OFFLINE` on a clean stand-down.

**The claim is a dedicated 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 a
claim that is acquired and released at runtime cannot ride the long-lived ingest session.
`SparkplugPrimaryHost` opens its own connection (client id `{clientId}-spbhost`, forced
clean-session) when the node gains Primary and closes it when the node loses Primary; the
connection's lifetime *is* the claim, so every leg follows from its lifecycle. Ingest is
untouched by a role change — no dropped subscriptions, no lost birth cache, no data gap at
the Secondary->Primary transition.

**Primary-gated** (owner decision, recorded on the issue). Both pair nodes subscribe
independently, so an ungated claim means both claim the same host id — and the sharp
failure is not the two fighting over the retained STATE, it is that the SECONDARY's death
fires its Will and publishes OFFLINE while the Primary is healthy, stopping every edge node
in the group. 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.

New seam: `IDriverPrimaryRoleSource` (`Core.Abstractions`) carrying the *data-plane* verdict
(`PrimaryGatePolicy.ShouldServiceAsPrimary`), not the alarm-drain verdict — an unknown role
with a driver peer present must deny. Registered unconditionally by `AddOtOpcUaRuntime`,
deliberately NOT alongside `IRedundancyRoleView`, which `AddAlarmHistorian` registers only
when that section is enabled: piggybacking would leave a pair with no alarm historian
falling back to "claim unconditionally". Polled on a 1s timer rather than evented — a
process singleton raising events at per-instance drivers outlives every driver it holds.

Accepted trade: a brief edge-visible OFFLINE window during failover. That is a genuine
state transition, not a glitch to hide.

Live-gated 5/5 against the Mosquitto fixture, including the Will leg — `AbandonForTests`
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
stand-down OFFLINE and pass against an implementation that never armed a Will. Falsified:
removing the Will turns that leg and only that leg red.

Also corrects two stale claims found en route: the AdminUI form still said MQTT tags are
read-only (#508 shipped), and the driver doc still said nothing consumes `IRediscoverable`.

Claude-Session: https://claude.ai/code/session_015p7wGqy3YpZNCpDzTpGMKo
2026-07-31 07:17:17 -04:00

101 lines
5.7 KiB
C#

using System.Text.Json;
using Microsoft.Extensions.Logging;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Core.Hosting;
namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt;
/// <summary>
/// Registers the MQTT / Sparkplug B driver with the <see cref="DriverFactoryRegistry"/>. The
/// Host's <c>DriverFactoryBootstrap</c> calls <see cref="Register"/> once at startup; the
/// driver-instance bootstrapper then materialises <c>DriverInstance</c> rows of type
/// <see cref="DriverTypeNames.Mqtt"/> into live <see cref="MqttDriver"/> instances.
/// </summary>
/// <remarks>
/// <para>
/// <b>Direct deserialization, no intermediate DTO.</b> The <c>DriverConfig</c> blob binds
/// straight onto <see cref="MqttDriverOptions"/> — the same shape
/// <see cref="MqttDriverProbe"/>, <c>MqttDriverBrowser</c> and
/// <see cref="MqttDriver.ReinitializeAsync"/> already bind, and the shape the options record
/// was designed for (every knob carries its own default, and <c>rawTags</c> is a plain
/// <see cref="RawTagEntry"/> list needing no translation). Mirrors
/// <c>OpcUaClientDriverFactoryExtensions</c>. <c>ModbusDriverFactoryExtensions</c>' separate
/// DTO exists to service a legacy nullable-everything blob plus string→enum parsing this
/// driver does with a converter instead; copying it here would add a <i>fourth</i> parse
/// shape for one config, which is precisely the divergence
/// <see cref="MqttJson.Options"/> exists to prevent.
/// </para>
/// <para>
/// <b>Connection-free.</b> <see cref="CreateInstance"/> parses and constructs only — the
/// <see cref="MqttDriver"/> constructor touches no network, and the registry contract
/// forbids the factory from calling <c>InitializeAsync</c> itself (the driver host owns the
/// retry semantics).
/// </para>
/// </remarks>
public static class MqttDriverFactoryExtensions
{
/// <summary>
/// Driver type name — matches <c>DriverInstance.DriverType</c> values. Sourced from
/// <see cref="DriverTypeNames.Mqtt"/> so the registration key can never drift from the
/// constant the dispatch maps, the probe and the browser reference.
/// </summary>
public const string DriverTypeName = DriverTypeNames.Mqtt;
/// <summary>
/// Register the MQTT factory with the driver registry. The optional
/// <paramref name="loggerFactory"/> is captured at registration time and used to construct an
/// <see cref="ILogger{MqttDriver}"/> per driver instance — without it the driver runs with no
/// logger (standalone/test callers stay unchanged).
/// </summary>
/// <param name="registry">The driver factory registry to register with.</param>
/// <param name="loggerFactory">Optional logger factory used to create per-instance loggers.</param>
/// <param name="primaryRoleSource">
/// Gitea #511: the Primary data-plane verdict the Sparkplug primary-host claim is gated on,
/// captured at registration time like <paramref name="loggerFactory"/>. Null falls back to
/// <see cref="UnmanagedDriverPrimaryRoleSource"/> (always claims) — correct for a standalone
/// host or a test, and the reason <c>DriverFactoryBootstrap</c> resolves it with
/// <c>GetRequiredService</c> rather than letting a redundant deployment reach that fallback.
/// </param>
public static void Register(
DriverFactoryRegistry registry,
ILoggerFactory? loggerFactory = null,
IDriverPrimaryRoleSource? primaryRoleSource = null)
{
ArgumentNullException.ThrowIfNull(registry);
registry.Register(DriverTypeName, (id, json) => CreateInstance(id, json, loggerFactory, primaryRoleSource));
}
/// <summary>Public for the Server-side bootstrapper + test consumers.</summary>
/// <param name="driverInstanceId">Stable logical id of the driver instance.</param>
/// <param name="driverConfigJson">The <c>DriverConfig</c> JSON blob for the instance.</param>
/// <param name="loggerFactory">Optional logger factory for the per-instance logger.</param>
/// <param name="primaryRoleSource">
/// Gitea #511: the Primary data-plane verdict gating the Sparkplug primary-host claim; null
/// falls back to <see cref="UnmanagedDriverPrimaryRoleSource"/>.
/// </param>
/// <returns>A configured, not-yet-connected <see cref="MqttDriver"/>.</returns>
/// <exception cref="ArgumentException">
/// <paramref name="driverInstanceId"/> or <paramref name="driverConfigJson"/> is blank.
/// </exception>
/// <exception cref="InvalidOperationException">The config blob deserialised to <c>null</c>.</exception>
/// <exception cref="JsonException">The config blob is not valid JSON for the options shape.</exception>
public static MqttDriver CreateInstance(
string driverInstanceId,
string driverConfigJson,
ILoggerFactory? loggerFactory = null,
IDriverPrimaryRoleSource? primaryRoleSource = null)
{
ArgumentException.ThrowIfNullOrWhiteSpace(driverInstanceId);
ArgumentException.ThrowIfNullOrWhiteSpace(driverConfigJson);
// MqttJson.Options — the one shared instance (see its remarks). A local copy here is the
// repo's documented systemic enum bug in the making.
var options = JsonSerializer.Deserialize<MqttDriverOptions>(driverConfigJson, MqttJson.Options)
?? throw new InvalidOperationException(
$"MQTT driver config for '{driverInstanceId}' deserialised to null");
return new MqttDriver(
options, driverInstanceId, loggerFactory?.CreateLogger<MqttDriver>(), primaryRoleSource);
}
}