7f2108dcf2
`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
232 lines
11 KiB
C#
232 lines
11 KiB
C#
using System.ComponentModel.DataAnnotations;
|
|
using System.Text;
|
|
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
|
|
|
namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt;
|
|
|
|
/// <summary>
|
|
/// MQTT / Sparkplug B driver configuration. Bound from <c>DriverConfig</c> JSON at
|
|
/// driver-host registration time. Models the settings documented in
|
|
/// <c>docs/plans/2026-07-15-mqtt-sparkplug-driver-design.md</c> §5.1.
|
|
/// </summary>
|
|
/// <remarks>
|
|
/// A record (not a plain class) so a future secret-resolution seam can produce a
|
|
/// credential-resolved copy with a <c>with</c> expression — mirrors
|
|
/// <c>OpcUaClientDriverOptions</c>. <see cref="Mode"/> selects the ingest shape;
|
|
/// <see cref="Sparkplug"/> and <see cref="Plain"/> are nullable sub-objects — only the one
|
|
/// matching <see cref="Mode"/> is populated.
|
|
/// </remarks>
|
|
public sealed record MqttDriverOptions
|
|
{
|
|
/// <summary>Broker hostname or IP address.</summary>
|
|
public string Host { get; init; } = "localhost";
|
|
|
|
/// <summary>Broker TCP port.</summary>
|
|
[Range(1, 65535)]
|
|
public int Port { get; init; } = 8883;
|
|
|
|
/// <summary>
|
|
/// MQTT client identifier sent at CONNECT. Leave unset to let the driver generate one
|
|
/// (a stable per-instance id is recommended so broker-side ACLs / session state persist
|
|
/// across reconnects).
|
|
/// </summary>
|
|
public string? ClientId { get; init; }
|
|
|
|
/// <summary>
|
|
/// When <c>true</c>, connect over TLS. Default <c>true</c> — this driver never ships an
|
|
/// anonymous/plaintext-by-default posture; a deployment must opt into <c>false</c> for an
|
|
/// on-prem/dev broker with no TLS listener.
|
|
/// </summary>
|
|
public bool UseTls { get; init; } = true;
|
|
|
|
/// <summary>
|
|
/// When <c>true</c>, accept any self-signed / untrusted broker certificate. Dev/on-prem
|
|
/// escape hatch only — mirrors the <c>ServerHistorian</c> TLS knobs. Must stay
|
|
/// <c>false</c> in production so MITM attacks against the broker connection fail closed.
|
|
/// </summary>
|
|
public bool AllowUntrustedServerCertificate { get; init; } = false;
|
|
|
|
/// <summary>
|
|
/// PEM CA file pinning the broker's TLS chain. <c>null</c>/empty uses the OS trust
|
|
/// store.
|
|
/// </summary>
|
|
public string? CaCertificatePath { get; init; }
|
|
|
|
/// <summary>Username for broker authentication. <c>null</c> connects without a username.</summary>
|
|
public string? Username { get; init; }
|
|
|
|
/// <summary>
|
|
/// Password for broker authentication. Blank in committed JSON — supply via env, never
|
|
/// commit or log.
|
|
/// </summary>
|
|
public string? Password { get; init; } = "";
|
|
|
|
/// <summary>MQTT protocol version to negotiate at CONNECT.</summary>
|
|
public MqttProtocolVersion ProtocolVersion { get; init; } = MqttProtocolVersion.V500;
|
|
|
|
/// <summary>Whether to request a clean session (v3.1.1) / clean start (v5.0) at CONNECT.</summary>
|
|
public bool CleanSession { get; init; } = true;
|
|
|
|
/// <summary>Keep-alive interval sent at CONNECT.</summary>
|
|
[Range(1, int.MaxValue)]
|
|
public int KeepAliveSeconds { get; init; } = 30;
|
|
|
|
/// <summary>Bounded connect deadline (see design §8) — the driver never hangs past this.</summary>
|
|
[Range(1, int.MaxValue)]
|
|
public int ConnectTimeoutSeconds { get; init; } = 15;
|
|
|
|
/// <summary>Initial reconnect backoff after a connection drop.</summary>
|
|
[Range(1, int.MaxValue)]
|
|
public int ReconnectMinBackoffSeconds { get; init; } = 1;
|
|
|
|
/// <summary>Cap on the exponential reconnect backoff.</summary>
|
|
[Range(1, int.MaxValue)]
|
|
public int ReconnectMaxBackoffSeconds { get; init; } = 30;
|
|
|
|
/// <summary>
|
|
/// Ceiling on an inbound message body, in bytes. A larger message is refused before any
|
|
/// decode or parse and degrades its own tags to <c>BadDecodingError</c>. Default 1 MiB.
|
|
/// </summary>
|
|
/// <remarks>
|
|
/// Decode and parse both run synchronously on MQTTnet's shared dispatcher thread, so their
|
|
/// cost is paid by <i>every</i> subscription, not just the offending topic — without a bound,
|
|
/// one publisher shipping a multi-megabyte body stalls the whole driver's delivery. "Unbounded"
|
|
/// is deliberately not offered; a non-positive value falls back to the driver's own 1 MiB
|
|
/// default. The literal is duplicated from <c>MqttSubscriptionManager.DefaultMaxPayloadBytes</c>
|
|
/// because this <c>.Contracts</c> assembly is <i>referenced by</i> the driver assembly and
|
|
/// cannot reference it back; the manager's constructor is the single enforcement point.
|
|
/// </remarks>
|
|
[Range(1, int.MaxValue)]
|
|
public int MaxPayloadBytes { get; init; } = 1024 * 1024;
|
|
|
|
/// <summary>Selects the ingest shape — Plain topics or Sparkplug B.</summary>
|
|
public MqttMode Mode { get; init; } = MqttMode.Plain;
|
|
|
|
/// <summary>
|
|
/// The cluster's authored raw MQTT tags, as delivered by the deploy artifact: each carries the
|
|
/// tag's <b>RawPath</b> (its v3 identity and driver wire reference) plus its <c>TagConfig</c>
|
|
/// blob. <see cref="MqttDriver"/> maps each through <see cref="MqttTagDefinitionFactory"/> at
|
|
/// initialize and re-registers the set <b>wholesale</b> on every reinitialize, so a redeploy
|
|
/// that drops a tag stops feeding it.
|
|
/// </summary>
|
|
/// <remarks>
|
|
/// This is also the <b>only</b> source of the driver's discoverable node set — plain MQTT has
|
|
/// no browsable address space, and a tag that is not authored here is never materialized no
|
|
/// matter how much traffic its topic carries. Mirrors <c>ModbusDriverOptions.RawTags</c> /
|
|
/// <c>FocasDriverOptions.RawTags</c>.
|
|
/// </remarks>
|
|
public IReadOnlyList<RawTagEntry> RawTags { get; init; } = [];
|
|
|
|
/// <summary>
|
|
/// Sparkplug-only settings. Populated when <see cref="Mode"/> is
|
|
/// <see cref="MqttMode.SparkplugB"/>; <c>null</c> in Plain mode.
|
|
/// </summary>
|
|
public MqttSparkplugOptions? Sparkplug { get; init; }
|
|
|
|
/// <summary>
|
|
/// Plain-mode-only settings. Populated when <see cref="Mode"/> is
|
|
/// <see cref="MqttMode.Plain"/>; <c>null</c> in Sparkplug B mode.
|
|
/// </summary>
|
|
public MqttPlainOptions? Plain { get; init; }
|
|
|
|
/// <summary>
|
|
/// Record-generated <c>ToString()</c> member printer, overridden so <see cref="Password"/>
|
|
/// never renders in plaintext — this DTO routinely lands in logs / exception messages /
|
|
/// debugger watches, and the plan's cross-cutting rule is explicit: never commit or log
|
|
/// creds. A set password prints as <c>***</c>; unset (<c>null</c>/empty) renders
|
|
/// distinguishably so the redaction can't be mistaken for a real secret.
|
|
/// </summary>
|
|
private bool PrintMembers(StringBuilder builder)
|
|
{
|
|
var passwordDisplay = Password switch
|
|
{
|
|
null => "<null>",
|
|
"" => "<empty>",
|
|
_ => "***",
|
|
};
|
|
|
|
builder.Append("Host = ").Append(Host);
|
|
builder.Append(", Port = ").Append(Port);
|
|
builder.Append(", ClientId = ").Append(ClientId);
|
|
builder.Append(", UseTls = ").Append(UseTls);
|
|
builder.Append(", AllowUntrustedServerCertificate = ").Append(AllowUntrustedServerCertificate);
|
|
builder.Append(", CaCertificatePath = ").Append(CaCertificatePath);
|
|
builder.Append(", Username = ").Append(Username);
|
|
builder.Append(", Password = ").Append(passwordDisplay);
|
|
builder.Append(", ProtocolVersion = ").Append(ProtocolVersion);
|
|
builder.Append(", CleanSession = ").Append(CleanSession);
|
|
builder.Append(", KeepAliveSeconds = ").Append(KeepAliveSeconds);
|
|
builder.Append(", ConnectTimeoutSeconds = ").Append(ConnectTimeoutSeconds);
|
|
builder.Append(", ReconnectMinBackoffSeconds = ").Append(ReconnectMinBackoffSeconds);
|
|
builder.Append(", ReconnectMaxBackoffSeconds = ").Append(ReconnectMaxBackoffSeconds);
|
|
builder.Append(", MaxPayloadBytes = ").Append(MaxPayloadBytes);
|
|
builder.Append(", Mode = ").Append(Mode);
|
|
builder.Append(", Sparkplug = ").Append(Sparkplug);
|
|
builder.Append(", Plain = ").Append(Plain);
|
|
// Count only: a deployment routinely authors thousands of raw tags and each carries a full
|
|
// TagConfig blob, so printing the list would turn any log of this DTO into a config dump.
|
|
builder.Append(", RawTags = ").Append(RawTags.Count).Append(" tag(s)");
|
|
return true;
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Sparkplug B settings for an <see cref="MqttDriverOptions"/> instance in
|
|
/// <see cref="MqttMode.SparkplugB"/> mode. See
|
|
/// <c>docs/plans/2026-07-15-mqtt-sparkplug-driver-design.md</c> §5.1.
|
|
/// </summary>
|
|
public sealed record MqttSparkplugOptions
|
|
{
|
|
/// <summary>Sparkplug group id — the driver subscribes <c>spBv1.0/{GroupId}/#</c>.</summary>
|
|
public string GroupId { get; init; } = "";
|
|
|
|
/// <summary>
|
|
/// Sparkplug Host Application identity — published as <c>spBv1.0/STATE/{HostId}</c>
|
|
/// (Sparkplug v3.0).
|
|
/// </summary>
|
|
public string HostId { get; init; } = "";
|
|
|
|
/// <summary>
|
|
/// When <c>true</c>, this driver instance acts as the Sparkplug Primary Host
|
|
/// Application. At most one primary host may exist per host-id per broker.
|
|
/// </summary>
|
|
/// <remarks>
|
|
/// Implemented by <c>SparkplugPrimaryHost</c> (Gitea #511) on a dedicated broker connection
|
|
/// whose lifetime <i>is</i> the claim, and gated on the redundancy Primary role so the two nodes
|
|
/// of a pair never both claim. Requires <see cref="HostId"/>; set without one, the driver warns
|
|
/// and declines rather than publishing to a malformed topic.
|
|
/// </remarks>
|
|
public bool ActAsPrimaryHost { get; init; } = false;
|
|
|
|
/// <summary>
|
|
/// When <c>true</c>, a detected sequence-number gap triggers a Sparkplug rebirth
|
|
/// request (NCMD) rather than silently continuing with stale metric state.
|
|
/// </summary>
|
|
public bool RequestRebirthOnGap { get; init; } = true;
|
|
|
|
/// <summary>
|
|
/// How long, in seconds, browse/discovery collects NBIRTH/DBIRTH traffic before
|
|
/// considering the observed metric set stable.
|
|
/// </summary>
|
|
[Range(1, int.MaxValue)]
|
|
public int BirthObservationWindowSeconds { get; init; } = 15;
|
|
}
|
|
|
|
/// <summary>
|
|
/// Plain-MQTT-mode settings for an <see cref="MqttDriverOptions"/> instance in
|
|
/// <see cref="MqttMode.Plain"/> mode. See
|
|
/// <c>docs/plans/2026-07-15-mqtt-sparkplug-driver-design.md</c> §5.1.
|
|
/// </summary>
|
|
public sealed record MqttPlainOptions
|
|
{
|
|
/// <summary>
|
|
/// Optional prefix prepended when the driver needs to compose a topic (e.g. discovery
|
|
/// scoping); per-tag topics are authored explicitly and are unaffected.
|
|
/// </summary>
|
|
public string TopicPrefix { get; init; } = "";
|
|
|
|
/// <summary>Default QoS used when a tag's own <c>qos</c> is unset.</summary>
|
|
[Range(0, 2)]
|
|
public int DefaultQos { get; init; } = 1;
|
|
}
|