using System.Collections.Concurrent;
using System.Collections.Frozen;
using System.Text.Json;
using Microsoft.Extensions.Logging;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Sparkplug;
namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt;
///
/// The MQTT / Sparkplug B driver shell: composes (broker session +
/// reconnect supervisor), (authored table, topic routing,
/// value extraction) and the manager's (the push→poll bridge) into
/// the driver capability set the Core consumes.
///
///
///
/// Authored-only discovery. Plain MQTT has no browsable address space — a broker will
/// happily carry thousands of topics this deployment has nothing to do with.
/// therefore replays exactly the raw tags the deploy artifact
/// authored () and nothing else, and
/// stays so the
/// universal browser never treats broker traffic as a tag catalogue. In
/// the policy is :
/// the authored set — and every authored tag's datatype — is fully known synchronously, so
/// re-running discovery on a timer can only produce the same tree.
///
///
/// Two ingest paths, one driver. routes through
/// (authored topics);
/// routes through (one group-wide filter, birth certificates,
/// alias resolution, seq gaps, rebirth NCMDs). Exactly one is live, selected by
/// and rebuilt whenever
/// changes anything either captures. They share the
/// driver-owned , so IReadable reads one store either way.
///
///
/// fires in Sparkplug B only, and only on a real change.
/// A Sparkplug tag's dataType is optional — the birth certificate declares it — so the
/// discovered tree fills in asynchronously and the policy is
/// .
/// is wired to through
/// , which fires only when the birth's metric-name SET
/// differs from the last one seen for that scope (and only for scopes an authored tag actually
/// binds). That gate is the anti-storm mechanism, not an optimisation: rediscovery triggers an
/// address-space rebuild, edge nodes re-birth freely — on their own timer, on every reconnect
/// via the late-join rebirth, and once per NCMD the ingestor's gap policy sends — and firing on
/// each of those would turn a healthy plant into a permanent rebuild loop. Plain mode never
/// raises it at all; its authored set changes only by redeploy.
///
///
/// Composition order is load-bearing.
/// must run before any subscribe and before the connect completes — it is what wires
/// message delivery and, critically, the reconnect re-subscribe. The connection's
/// Reconnected handler is passed through unwrapped: the manager throws when it can
/// re-establish nothing, and that throw is precisely what tears the session down and retries.
/// Swallowing it would leave a driver that reports Connected and receives nothing.
///
///
/// Constructor is connection-free. It maps the authored TagConfig blobs (pure, no I/O)
/// so discovery and reads work off a configured-but-not-yet-connected driver; every network
/// operation happens in .
///
///
public sealed class MqttDriver
: IDriver, ITagDiscovery, ISubscribable, IReadable, IHostConnectivityProbe, IRediscoverable, IAsyncDisposable, IDisposable
{
///
/// Rough per-tag driver-attributable cost: the parsed (five
/// strings — RawPath, topic, JSONPath and the record header) plus its cached
/// and the two dictionary entries that index it. An estimate
/// by construction — GetMemoryFootprint exists to drive a cache-budget decision, not
/// to be exact.
///
private const long ApproxBytesPerAuthoredTag = 512;
///
/// The single folder streams every variable under — and therefore
/// the only subtree path a rediscovery ScopeHint can name truthfully. See
/// .
///
private const string DiscoveryFolderName = "Mqtt";
/// How often the host-connectivity poll samples .
private static readonly TimeSpan HostProbeInterval = TimeSpan.FromSeconds(1);
private readonly string _driverInstanceId;
private readonly ILogger? _logger;
///
/// Owned by the driver, not by the manager, so a manager rebuilt for changed ingest settings
/// inherits the observed values instead of silently regressing every node to
/// BadWaitingForInitialData.
///
private readonly LastValueCache _values = new();
private readonly object _hostLock = new();
///
/// Per authored Sparkplug scope, the signature of the metric-name set its most recent birth
/// declared — the rediscovery change gate's memory. Written on MQTTnet's dispatcher thread and
/// read nowhere else, but concurrent because a reconnect's birth flood and the dispatcher are
/// not guaranteed to be the same thread forever.
///
///
/// Deliberately NOT cleared on a death, a reconnect or an ingest rebuild. The ingestor
/// drops its whole cache on every reconnect (an alias may
/// have been rebound while the driver was away) — but "the driver forgot" is not "the address
/// space changed". Clearing this alongside it would make every reconnect's late-join rebirth
/// flood look like a tree full of brand-new scopes and fire one rediscovery per authored scope
/// on every single flap, which is the storm this map exists to prevent. It survives so that an
/// identical rebirth stays silent.
///
private readonly ConcurrentDictionary _birthSignatures = new();
/// Guards / / against each other.
private readonly SemaphoreSlim _lifecycleGate = new(1, 1);
// _options / _subscriptions / _authoredRawPaths are written only under _lifecycleGate (or in the
// ctor) and read freely elsewhere, WITHOUT the Volatile discipline _health / _hostState get.
// That is deliberate, not an oversight: all three are reference/interface fields, so a reader
// sees either the whole old object or the whole new one — never a torn one. The looser fields
// are the ones a reader may legitimately observe one publish stale (a read served from the
// previous authored table is a correct read of a value that was correct a microsecond ago);
// _health and _hostState feed ServiceLevel and the status dashboard, where a stale per-core copy
// is an operator-visible lie about whether the driver is up.
private MqttDriverOptions _options;
private MqttSubscriptionManager _subscriptions;
///
/// The Sparkplug B ingest path, or in .
/// Exactly one of this and is fed authored tags and attached to the
/// connection; the other exists but is never registered against, so it can neither route a
/// message nor answer a resolve.
///
private SparkplugIngestor? _sparkplug;
///
/// RawPaths of the currently registered authored tags, in authoring order — the discovery
/// enumeration set. The manager resolves a RawPath to its definition but does not enumerate,
/// and enumerating from here is what makes "authored only" structural rather than incidental.
///
private IReadOnlyList _authoredRawPaths = [];
///
/// The Sparkplug scopes at least one authored tag binds — empty outside
/// . Rebuilt wholesale by
/// alongside the authored table it is derived from.
///
///
/// The rediscovery gate's first question. The ingestor applies a birth for any device
/// under an authored edge node (it has to — a DBIRTH is a sequenced member of that node's
/// stream), so on a large plant it observes births for devices this deployment never reads. Such
/// a birth cannot change 's output by construction, because discovery
/// replays the authored set — so firing rediscovery for it would rebuild the address space in
/// response to traffic the configuration has no opinion about. Gating on this set also bounds
/// by configuration rather than by whatever the plant
/// happens to publish — the same rule the ingestor's own authored-edge-node filter follows.
///
private FrozenSet _authoredScopes = FrozenSet.Empty;
private MqttConnection? _connection;
private CancellationTokenSource? _hostProbeCts;
private DriverHealth _health = new(DriverState.Unknown, null, null);
private HostState _hostState = HostState.Unknown;
private DateTime _hostStateChangedUtc = DateTime.UtcNow;
private int _disposed;
/// Initializes a new driver. Stores + maps configuration only — no network, no client.
///
/// Driver configuration, including the authored .
/// Task 9's factory deserializes it from the DriverConfig JSON; a later
/// may replace it from a fresh config blob.
///
/// Stable logical id of this driver instance.
/// Optional logger; never receives credentials or payload bodies.
public MqttDriver(MqttDriverOptions options, string driverInstanceId, ILogger? logger = null)
{
ArgumentNullException.ThrowIfNull(options);
ArgumentException.ThrowIfNullOrWhiteSpace(driverInstanceId);
_options = options;
_driverInstanceId = driverInstanceId;
_logger = logger;
_subscriptions = CreateSubscriptionManager(options);
_sparkplug = CreateSparkplugIngestor(options);
RegisterAuthoredTags(options);
}
///
public event EventHandler? OnDataChange;
///
public event EventHandler? OnHostStatusChanged;
///
public event EventHandler? OnRediscoveryNeeded;
///
public string DriverInstanceId => _driverInstanceId;
///
public string DriverType => DriverTypeNames.Mqtt;
///
/// The ingest path this driver composes. Internal: the P2 Sparkplug handler feeds the same
/// OnDataChange + sinks through it, and the shell tests
/// drive to simulate broker traffic
/// without a broker.
///
internal MqttSubscriptionManager Subscriptions => _subscriptions;
///
/// The Sparkplug B ingest path, or outside
/// . Internal: the shell tests drive
/// to simulate an edge node without a broker, and
/// Tasks 22/23 read its .
///
internal SparkplugIngestor? Sparkplug => _sparkplug;
/// Whether this driver instance ingests Sparkplug B rather than plain topics.
private bool IsSparkplug => _options.Mode == MqttMode.SparkplugB;
///
/// Test seam: awaited inside while the lifecycle gate is
/// held, immediately before the broker connect. Lets a test park a lifecycle operation
/// mid-flight and prove serializes behind it rather than racing
/// past a half-built session. Mirrors 's own
/// AfterConnectHookForTests, which exists for the same class of race. Always
/// in production.
///
internal Func? BeforeConnectHookForTests { get; set; }
// ---- IDriver: lifecycle ----
///
/// Adopts (falling back to the constructor's options when
/// it is blank or unusable), registers the authored tags, and opens the broker session.
///
///
/// Ordering is the contract: register → attach → connect. Attaching before the connect
/// completes means the reconnect re-subscribe is wired for the very first drop, and the
/// manager's SUBSCRIBE transport is live before the OPC UA server's first subscribe arrives.
///
/// The driver configuration as JSON.
/// Cancellation for the connect.
/// A task that represents the asynchronous operation.
public async Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken)
{
await _lifecycleGate.WaitAsync(cancellationToken).ConfigureAwait(false);
try
{
await InitializeCoreAsync(driverConfigJson, cancellationToken).ConfigureAwait(false);
}
finally
{
_lifecycleGate.Release();
}
}
///
/// Applies a config delta in place. A tag-only delta re-registers the authored table without
/// touching the broker session; a delta that changes how the driver connects or ingests
/// rebuilds the session.
///
///
/// A bad delta never faults the driver. Unparseable or unusable config is logged and
/// discarded, leaving the running configuration exactly as it was — a redeploy carrying one
/// malformed driver blob must not take a healthy driver's whole address space Bad. A genuine
/// connect failure against a changed endpoint is a different thing and does fault,
/// exactly as would.
///
/// The driver configuration as JSON.
/// Cancellation for any reconnect the delta forces.
/// A task that represents the asynchronous operation.
public async Task ReinitializeAsync(string driverConfigJson, CancellationToken cancellationToken)
{
MqttDriverOptions? next;
try
{
next = ParseOptions(driverConfigJson);
}
catch (Exception ex)
{
// Defence in depth: ParseOptions already swallows the JSON failure modes.
_logger?.LogWarning(
ex,
"MQTT driver '{DriverId}': reinitialize config could not be read; keeping the running configuration.",
_driverInstanceId);
return;
}
if (next is null)
{
_logger?.LogWarning(
"MQTT driver '{DriverId}': reinitialize config was blank or unusable; keeping the running configuration.",
_driverInstanceId);
return;
}
await _lifecycleGate.WaitAsync(cancellationToken).ConfigureAwait(false);
try
{
if (!SameSession(next, _options))
{
// Broker endpoint / credentials / ingest shape changed — the live session cannot be
// reused. Full rebuild; a failure here is a real fault, not a bad delta.
_logger?.LogInformation(
"MQTT driver '{DriverId}': reinitialize changes the broker session; rebuilding.",
_driverInstanceId);
await TeardownAsync().ConfigureAwait(false);
await InitializeCoreAsync(driverConfigJson, cancellationToken).ConfigureAwait(false);
return;
}
_options = next;
RegisterAuthoredTags(next);
_logger?.LogInformation(
"MQTT driver '{DriverId}': reinitialize applied in place; {TagCount} authored tag(s).",
_driverInstanceId,
_authoredRawPaths.Count);
}
finally
{
_lifecycleGate.Release();
}
}
///
public async Task ShutdownAsync(CancellationToken cancellationToken)
{
var lastMessage = ReadHealth().LastSuccessfulRead;
if (await _lifecycleGate.WaitAsync(TimeSpan.FromSeconds(_options.ConnectTimeoutSeconds), cancellationToken)
.ConfigureAwait(false))
{
try
{
await TeardownAsync().ConfigureAwait(false);
}
finally
{
_lifecycleGate.Release();
}
}
else
{
// A wedged initialize must not turn shutdown into a hang; tear the session down anyway.
_logger?.LogWarning(
"MQTT driver '{DriverId}': lifecycle gate still held at shutdown; tearing the session down anyway.",
_driverInstanceId);
await TeardownAsync().ConfigureAwait(false);
}
WriteHealth(new DriverHealth(DriverState.Unknown, lastMessage, null));
TransitionHost(HostState.Unknown);
}
///
/// Current health. Derived live from the connection's own state where one exists, so a
/// background reconnect is visible without the driver having to mirror every transition.
/// LastSuccessfulRead carries the last inbound message instant — MQTT never polls, so
/// "last message age" is the only evidence the session is actually delivering.
///
/// The driver's current health snapshot.
public DriverHealth GetHealth()
{
var stored = ReadHealth();
var connection = _connection;
if (connection is null)
{
return stored;
}
var lastMessage = connection.LastMessageUtc ?? stored.LastSuccessfulRead;
return connection.State switch
{
MqttConnectionState.Connected => new DriverHealth(DriverState.Healthy, lastMessage, stored.LastError),
MqttConnectionState.Reconnecting => new DriverHealth(DriverState.Reconnecting, lastMessage, stored.LastError),
MqttConnectionState.Faulted => new DriverHealth(DriverState.Faulted, lastMessage, stored.LastError),
// Disconnected before the first successful connect, or Disposed after teardown: the
// stored snapshot (Initializing / Faulted / Unknown) is the more informative answer.
_ => stored with { LastSuccessfulRead = lastMessage },
};
}
///
/// Approximate driver-attributable footprint: the authored definition table and the
/// last-value cache that indexes it. Both are sized by the authored tag count, which is the
/// only thing a deployment can grow.
///
/// The approximate memory footprint in bytes.
public long GetMemoryFootprint() => _authoredRawPaths.Count * ApproxBytesPerAuthoredTag;
///
/// No-op, deliberately. This driver holds no optional cache: the authored table is the
/// address space itself, and the last-value cache is — MQTT is
/// subscribe-first, so a dropped value is not re-fetchable, and flushing it would turn every
/// node Bad until its topic next published, which for a slow-changing tag can be never.
/// Birth/alias state (Sparkplug, P2) is correctness state for the same reason.
///
/// Unused; present for the shape.
/// A completed task.
public Task FlushOptionalCachesAsync(CancellationToken cancellationToken) => Task.CompletedTask;
// ---- ITagDiscovery ----
///
///
///
/// Plain mode discovers synchronously from the authored set — the RawPaths and their
/// datatypes are both in the deploy artifact — so one pass is all there is.
///
///
/// Sparkplug B is because half of what
/// it discovers arrives on the wire: dataType is optional in the Sparkplug tag shape
/// (the birth certificate declares it), so a tag authored with nothing but its
/// (group, node, device, metric) tuple can only be streamed with the record's default
/// type until its NBIRTH/DBIRTH lands. The host's UntilStable contract — re-run
/// discovery until the captured set is non-empty and its signature repeats — is exactly the
/// retry that lets those passes catch a birth that had not yet arrived at connect.
///
///
/// The retry is a floor, not the mechanism. The host's stability signature is built
/// from FullReferences alone, so a pass whose only change is a filled-in datatype still
/// looks "stable" and the loop stops. That is why the same fill-in also fires
/// (see ): a birth arriving
/// after the loop settled is the case the timer cannot cover.
///
///
public DiscoveryRediscoverPolicy RediscoverPolicy =>
IsSparkplug ? DiscoveryRediscoverPolicy.UntilStable : DiscoveryRediscoverPolicy.Once;
///
///
/// Always : replays authored tags, it does
/// not enumerate a backend. A broker's live topic set is not a tag catalogue — the
/// Driver.Mqtt.Browser project is the surface for observing traffic during authoring.
///
public bool SupportsOnlineDiscovery => false;
///
/// Streams the authored tag set — and only that — into .
///
///
///
/// The enumeration source is the driver's own list of registered RawPaths, never anything
/// observed on the wire, so no amount of broker traffic can add a node. A RawPath whose
/// TagConfig failed to map is absent from the manager and is skipped here too — it surfaces
/// as BadNodeIdUnknown on read rather than as a wrongly-typed variable.
///
///
/// Sparkplug B: the datatype is resolved per pass, the tag set never is. An authored
/// tag is streamed on every pass, birth or no birth — it is part of the declared
/// configuration, not something the plant grants by publishing, and hiding it until a birth
/// arrives would make the node's existence depend on wire traffic (and, on a deployment
/// whose other tags do carry an authored type, would hide it behind a "non-empty and stable"
/// verdict the host reaches without it). What a birth changes is the type: see
/// .
///
///
/// The address space builder to stream discovered nodes into.
/// Cancellation for the discovery operation.
/// A task that represents the asynchronous operation.
public Task DiscoverAsync(IAddressSpaceBuilder builder, CancellationToken cancellationToken)
{
ArgumentNullException.ThrowIfNull(builder);
cancellationToken.ThrowIfCancellationRequested();
var folder = builder.Folder(DiscoveryFolderName, DiscoveryFolderName);
foreach (var rawPath in _authoredRawPaths)
{
if (!TryResolve(rawPath, out var def))
{
continue;
}
folder.Variable(def.Name, def.Name, new DriverAttributeInfo(
FullName: def.Name,
DriverDataType: ResolveDiscoveredDataType(def),
IsArray: false,
ArrayDim: null,
// MQTT ingest is one-way in P1 — this driver implements no IWritable leg, so every
// node is ViewOnly rather than an Operate node whose writes would silently vanish.
SecurityClass: SecurityClassification.ViewOnly,
IsHistorized: false,
IsAlarm: false,
WriteIdempotent: false));
}
return Task.CompletedTask;
}
///
/// The datatype reports for one authored tag: the authored type when
/// the operator declared one, otherwise the type the metric's live birth certificate declared,
/// otherwise the record's default.
///
///
///
/// The precedence is the ingest path's, restated. SparkplugIngestor.PublishMetric
/// coerces every value with DataTypeAuthored ? authored : birth; a discovery surface
/// that disagreed with it would advertise a type no value ever arrives as. The two must be
/// changed together.
///
///
/// An unsupported Sparkplug type (DataSet / Template / PropertySet / Unknown) maps to
/// and falls back rather than blanking the tag — the ingest path
/// already refuses those metrics once, loudly, and a discovery pass is not the place to
/// relitigate it.
///
///
/// The authored tag definition.
/// The effective data type for this pass.
private DriverDataType ResolveDiscoveredDataType(MqttTagDefinition def)
{
if (def.DataTypeAuthored || _sparkplug is not { } sparkplug)
{
return def.DataType;
}
if (def.GroupId is not { } groupId || def.EdgeNodeId is not { } edgeNodeId || def.MetricName is not { } metric)
{
return def.DataType;
}
var table = sparkplug.Births.Find(new SparkplugScope(groupId, edgeNodeId, def.DeviceId));
return table?.ResolveByName(metric)?.DataType.ToDriverDataType() ?? def.DataType;
}
// ---- ISubscribable (straight through to the manager) ----
///
public Task SubscribeAsync(
IReadOnlyList fullReferences,
TimeSpan publishingInterval,
CancellationToken cancellationToken)
=> _sparkplug is { } sparkplug
? sparkplug.SubscribeAsync(fullReferences, publishingInterval, cancellationToken)
: _subscriptions.SubscribeAsync(fullReferences, publishingInterval, cancellationToken);
///
public Task UnsubscribeAsync(ISubscriptionHandle handle, CancellationToken cancellationToken)
=> _sparkplug is { } sparkplug
? sparkplug.UnsubscribeAsync(handle, cancellationToken)
: _subscriptions.UnsubscribeAsync(handle, cancellationToken);
/// Resolves a RawPath through whichever ingest path this driver's mode selected.
/// The driver wire reference.
/// The definition when this returns .
/// when the reference is an authored tag of the live ingest path.
private bool TryResolve(string rawPath, out MqttTagDefinition def)
=> _sparkplug is { } sparkplug
? sparkplug.TryResolve(rawPath, out def)
: _subscriptions.TryResolve(rawPath, out def);
// ---- IReadable ----
///
/// Serves the batch from the last-value cache. MQTT is push-based: there is nothing to poll,
/// so a read is always "what was most recently published".
///
///
/// Never throws. A reference that has not been observed — including one that is not an
/// authored tag at all — returns BadWaitingForInitialData in its own slot. Failing the
/// whole call would let one stale reference blank every other node in the same OPC UA read.
///
/// The RawPath references to read.
/// Unused; the read touches no network.
/// One snapshot per requested reference, in request order.
public Task> ReadAsync(
IReadOnlyList fullReferences,
CancellationToken cancellationToken)
{
ArgumentNullException.ThrowIfNull(fullReferences);
var results = new DataValueSnapshot[fullReferences.Count];
for (var i = 0; i < fullReferences.Count; i++)
{
// The driver-owned cache, not the live ingest path's: both are handed this same instance,
// and reading it directly means a mid-flight ingest rebuild can never blank a read.
results[i] = _values.Read(fullReferences[i]);
}
return Task.FromResult>(results);
}
// ---- IHostConnectivityProbe ----
/// The single broker this driver instance talks to, as host:port.
public string HostName => $"{_options.Host}:{_options.Port}";
///
public IReadOnlyList GetHostStatuses()
{
lock (_hostLock)
{
return [new HostConnectivityStatus(HostName, _hostState, _hostStateChangedUtc)];
}
}
// ---- IRediscoverable ----
///
/// Raises . Nothing in calls it
/// — the authored set only changes by redeploy. It exists for the Sparkplug B path, where a
/// DBIRTH can introduce metrics the previous birth certificate did not carry.
///
/// Driver-supplied reason for the diagnostic log.
/// Optional subtree hint; means the whole tree.
internal void RaiseRediscoveryNeeded(string reason, string? scopeHint = null)
=> OnRediscoveryNeeded?.Invoke(this, new RediscoveryEventArgs(reason, scopeHint));
///
/// The rediscovery decision: a birth changed what can produce only
/// when it is for an authored scope and its metric-name set differs from the last
/// one seen for that scope. Anything else is dropped.
///
///
///
/// Both conditions are load-bearing, and both are anti-storm. Rediscovery triggers an
/// address-space rebuild — expensive and disruptive on a running server. An edge node
/// re-births freely: on its own schedule, on every reconnect (the ingestor's late-join
/// rebirth fan-out), and once per rebirth NCMD its gap policy sends. Firing on each of those
/// is a permanent rebuild loop against a plant that is behaving perfectly. Firing on
/// changed metric set only is the primary defence; the authored-scope filter is the
/// second, and removes the whole class of churn sourced from devices this deployment does
/// not read (see ).
///
///
/// The signature is order-insensitive — ordered-distinct, ordinal. Nothing obliges an
/// edge node to keep its birth's metric ordering stable, and a reordered birth describes the
/// same address space; comparing the raw sequence would leak the storm back in.
///
///
/// No additional debounce or coalescing. One was considered and rejected: the change
/// gate already collapses the unbounded sources (rebirth timers, reconnect floods, NCMD
/// answers) to zero, and what survives it is a genuine metric-set change — of which a given
/// scope has a handful in its lifetime, each one a real edit to the served tree. A timer on
/// top would only delay a rebuild that must happen, and would need its own trailing-edge
/// flush to avoid dropping the last change entirely. If a consumer ever needs coalescing
/// across many scopes birthing at once (a whole-plant restart), that belongs in the consumer,
/// which alone knows what a rebuild costs it — today there is no such consumer at all.
///
///
/// Runs on MQTTnet's dispatcher thread (via the ingestor's own contained raise), so it
/// does no I/O and never throws — catches a throwing
/// subscriber, but relying on that would still have cost the rest of the birth's fan-out.
///
///
private void OnBirthObserved(object? sender, SparkplugBirthObservedEventArgs e)
{
if (!_authoredScopes.Contains(e.Scope))
{
// A birth for a scope no authored tag binds cannot change the authored-only discovery tree.
return;
}
var signature = BirthSignature(e.MetricNames);
if (_birthSignatures.TryGetValue(e.Scope, out var previous)
&& string.Equals(previous, signature, StringComparison.Ordinal))
{
return; // Same metric set as last time — nothing to rediscover.
}
_birthSignatures[e.Scope] = signature;
var reason = previous is null
? $"Sparkplug birth observed for '{e.Scope}' ({e.MetricNames.Count} metric(s))"
: $"Sparkplug rebirth changed the metric set for '{e.Scope}' ({e.MetricNames.Count} metric(s))";
_logger?.LogInformation(
"MQTT driver '{DriverId}': {Reason}; requesting rediscovery.",
_driverInstanceId,
reason);
// ScopeHint names the folder DiscoverAsync actually streams under — deliberately NOT the
// Sparkplug scope path. The discovered tree is flat (one 'Mqtt' folder of RawPath-named
// variables); it has no edge-node/device folders, so a hint of 'Plant1/EdgeA/Filler1' would
// name a subtree that does not exist and a consumer scoping a rebuild on it would rebuild
// nothing. The scope belongs in the Reason, which is the documented diagnostic field.
RaiseRediscoveryNeeded(reason, DiscoveryFolderName);
}
/// An order-insensitive, ordinal signature of a birth's metric-name set.
/// The names the birth declared.
/// The comparable signature.
private static string BirthSignature(IReadOnlyList metricNames)
=> string.Join(
'\u0001', // A separator no Sparkplug metric name can contain, so two sets cannot alias.
metricNames.Distinct(StringComparer.Ordinal).Order(StringComparer.Ordinal));
// ---- disposal ----
///
public void Dispose() => DisposeAsync().AsTask().GetAwaiter().GetResult();
///
/// Performs the same teardown as , under the same lifecycle
/// gate, so a caller that uses await using without an explicit shutdown does not
/// leak a live broker session.
///
///
///
/// The gate wait is the whole point, not ceremony. An ungated dispose that ran
/// while 's rebuild branch held the gate would observe
/// _connection == null — the rebuild has torn the old session down but not yet
/// assigned the new one — do nothing, and set _disposed. The rebuild would then
/// complete, assign a live, connected connection and start a host-probe loop that
/// no later dispose can ever reach, because the disposed guard short-circuits them all.
/// That is the orphaned-connection class 's own remarks
/// describe having already fixed once. Waiting for the gate makes this dispose tear down
/// whatever the in-flight operation ends up assigning.
///
///
/// _disposed is set before the wait, so an in-flight
/// sees it at its post-connect re-check and refuses to
/// publish the connection at all — which is what closes the residual window on the
/// bounded-timeout fallback path below.
///
///
/// Separately (and for an unrelated reason): the gate object itself is deliberately never
/// Dispose()d — same call as 's semaphores. A caller
/// that disposes before its last would otherwise get an
/// naming , which says
/// nothing useful; the gate holds no timer and no surviving registration, so leaving it
/// undisposed costs nothing.
///
///
/// A task that represents the asynchronous dispose.
public async ValueTask DisposeAsync()
{
if (Interlocked.Exchange(ref _disposed, 1) != 0)
{
return;
}
if (await _lifecycleGate.WaitAsync(TimeSpan.FromSeconds(_options.ConnectTimeoutSeconds))
.ConfigureAwait(false))
{
try
{
await TeardownAsync().ConfigureAwait(false);
}
finally
{
_lifecycleGate.Release();
}
return;
}
// Bounded even here: a wedged initialize must not turn dispose into a hang. The post-connect
// disposed re-check in InitializeCoreAsync is what stops that operation from publishing a
// connection behind this teardown's back.
_logger?.LogWarning(
"MQTT driver '{DriverId}': lifecycle gate still held at dispose; tearing the session down anyway.",
_driverInstanceId);
await TeardownAsync().ConfigureAwait(false);
}
// ---- internals ----
/// The gate-held body of .
private async Task InitializeCoreAsync(string driverConfigJson, CancellationToken cancellationToken)
{
WriteHealth(new DriverHealth(DriverState.Initializing, ReadHealth().LastSuccessfulRead, null));
try
{
// A blank / unusable blob keeps the constructor's options rather than degrading the
// driver to defaults pointing at localhost.
var parsed = ParseOptions(driverConfigJson);
if (parsed is not null)
{
AdoptOptions(parsed);
}
RegisterAuthoredTags(_options);
var connection = new MqttConnection(_options, _driverInstanceId, _logger);
// MUST precede the connect: this wires message delivery AND the reconnect re-subscribe.
// The handler is passed through unwrapped on purpose — its throw-on-total-failure is what
// tears a deaf session down and retries it.
if (_sparkplug is { } attaching)
{
attaching.AttachTo(connection);
}
else
{
_subscriptions.AttachTo(connection);
}
if (BeforeConnectHookForTests is { } hook)
{
await hook(cancellationToken).ConfigureAwait(false);
}
await connection.ConnectAsync(cancellationToken).ConfigureAwait(false);
// Re-check disposal AFTER the connect, before publishing the connection. DisposeAsync
// normally waits for this gate, but its wait is bounded — on the wedged-gate fallback
// path it can run concurrently with this method, and assigning a live connection behind
// a completed teardown is exactly the orphaned-session bug. Losing the race means
// disposing what we just built, never leaking it.
if (Volatile.Read(ref _disposed) != 0)
{
await connection.DisposeAsync().ConfigureAwait(false);
throw new ObjectDisposedException(nameof(MqttDriver));
}
// Sparkplug subscribes ONCE, here, rather than on the OPC UA server's first subscribe: the
// driver must see birth certificates whether or not anything is subscribed yet, because a
// birth is where every alias and datatype comes from. This also issues the late-join
// rebirth (§3.6 #5) — the initial connect does not fire Reconnected.
//
// It runs BEFORE `_connection` is published, and disposes the session it was given if it
// fails. Publishing first would leave a live, supervised connection behind a throw: the
// driver-host resilience layer re-runs this method, the retry overwrites `_connection`, and
// the first session becomes an orphan no later dispose can reach — the exact leak
// MqttConnection's own remarks describe having already fixed once.
if (_sparkplug is { } sparkplug)
{
try
{
await sparkplug.EstablishAsync(cancellationToken).ConfigureAwait(false);
}
catch
{
await connection.DisposeAsync().ConfigureAwait(false);
throw;
}
}
_connection = connection;
WriteHealth(new DriverHealth(DriverState.Healthy, connection.LastMessageUtc, null));
TransitionHost(HostState.Running);
StartHostProbe();
}
catch (Exception ex)
{
WriteHealth(new DriverHealth(DriverState.Faulted, ReadHealth().LastSuccessfulRead, ex.Message));
TransitionHost(HostState.Faulted);
throw;
}
}
///
/// Deserializes a driver-config blob. Returns — never throws — when the
/// blob is blank or unusable, so both the initialize fallback and the reinitialize
/// keep-running-config rule read off one answer.
///
private MqttDriverOptions? ParseOptions(string driverConfigJson)
{
if (string.IsNullOrWhiteSpace(driverConfigJson))
{
return null;
}
try
{
// MqttJson.Options — the ONE shared instance (factory / probe / this / browser),
// deliberately: enums must round-trip by NAME. A second, divergent
// JsonSerializerOptions is this repo's documented systemic enum bug, where an
// AdminUI-authored config with a string enum field faults the driver.
return JsonSerializer.Deserialize(driverConfigJson, MqttJson.Options);
}
catch (JsonException ex)
{
_logger?.LogWarning(
ex,
"MQTT driver '{DriverId}': driver config JSON is invalid.",
_driverInstanceId);
return null;
}
catch (NotSupportedException ex)
{
_logger?.LogWarning(
ex,
"MQTT driver '{DriverId}': driver config JSON could not be bound to the options shape.",
_driverInstanceId);
return null;
}
}
///
/// Adopts new options, rebuilding the subscription manager only when a setting it captures at
/// construction actually changed. The rebuilt manager inherits the driver-owned
/// , so adopting new options never blanks observed values.
///
private void AdoptOptions(MqttDriverOptions next)
{
if (!SameIngest(next, _options))
{
_subscriptions = CreateSubscriptionManager(next);
_sparkplug = CreateSparkplugIngestor(next);
}
_options = next;
}
/// Builds a manager over the driver-owned cache and bridges its notifications to .
private MqttSubscriptionManager CreateSubscriptionManager(MqttDriverOptions options)
{
var manager = new MqttSubscriptionManager(
options,
_driverInstanceId,
transport: null,
logger: _logger,
cache: _values,
maxPayloadBytes: options.MaxPayloadBytes);
// The published reference is the RawPath the manager already stamped on the args; the driver
// only re-raises with itself as sender.
manager.OnDataChange += (_, args) => OnDataChange?.Invoke(this, args);
return manager;
}
///
/// Builds the Sparkplug B ingest path for , or
/// outside . Shares the driver-owned
/// for the same reason the plain manager does: a rebuild for
/// changed settings must not blank every observed value.
///
private SparkplugIngestor? CreateSparkplugIngestor(MqttDriverOptions options)
{
if (options.Mode != MqttMode.SparkplugB)
{
return null;
}
if (options.Sparkplug is { ActAsPrimaryHost: true })
{
// 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
// MqttConnection.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);
}
var ingestor = new SparkplugIngestor(
options,
_driverInstanceId,
subscribeTransport: null,
publishTransport: null,
logger: _logger,
cache: _values,
maxPayloadBytes: options.MaxPayloadBytes);
// The published reference is the RawPath the ingestor already stamped on the args; the driver
// only re-raises with itself as sender.
ingestor.OnDataChange += (_, args) => OnDataChange?.Invoke(this, args);
// The ingestor detects a birth; the driver decides whether it changed the address space. Wired
// here rather than in InitializeAsync so a driver rebuilt for a config delta re-arms it — an
// ingest rebuild that silently dropped this subscription would leave a Sparkplug driver whose
// datatypes fill in and whose rediscovery never fires again, with nothing to show for it.
ingestor.BirthObserved += OnBirthObserved;
return ingestor;
}
/// Registers the authored raw tags wholesale and refreshes the discovery enumeration set.
private void RegisterAuthoredTags(MqttDriverOptions options)
{
// Wholesale into the LIVE path only. Feeding Sparkplug tags to the plain parser would log a
// "did not map" warning for every one of them (they carry no topic) and vice versa.
var mapped = _sparkplug is { } sparkplug
? sparkplug.Register(options.RawTags)
: _subscriptions.Register(options.RawTags);
// Enumerate in authoring order; the manager owns which of these actually mapped, and
// DiscoverAsync skips the rest.
_authoredRawPaths = [.. options.RawTags.Select(t => t.RawPath)];
_authoredScopes = BuildAuthoredScopes();
// Drop change-gate memory for scopes this deploy stopped authoring — the same housekeeping
// SparkplugIngestor.Register does for its trackers, and for the same reason: without it a scope
// nobody reads keeps a slot across every future redeploy. Re-authoring such a scope later means
// its next birth reads as new and fires once, which is correct — a newly authored scope IS a
// change to the discovered tree.
foreach (var scope in _birthSignatures.Keys)
{
if (!_authoredScopes.Contains(scope))
{
_birthSignatures.TryRemove(scope, out _);
}
}
if (mapped != options.RawTags.Count)
{
_logger?.LogWarning(
"MQTT driver '{DriverId}': {Mapped} of {Authored} authored tag(s) mapped to a definition; "
+ "the rest will report BadNodeIdUnknown.",
_driverInstanceId,
mapped,
options.RawTags.Count);
}
}
///
/// The Sparkplug scopes the currently registered authored tags bind, derived from the same
/// definitions enumerates — empty outside
/// , and empty for any tag whose TagConfig did not map.
///
/// The authored scope set.
private FrozenSet BuildAuthoredScopes()
{
if (_sparkplug is null)
{
return FrozenSet.Empty;
}
var scopes = new HashSet();
foreach (var rawPath in _authoredRawPaths)
{
if (TryResolve(rawPath, out var def)
&& def.GroupId is { } groupId
&& def.EdgeNodeId is { } edgeNodeId)
{
scopes.Add(new SparkplugScope(groupId, edgeNodeId, def.DeviceId));
}
}
return scopes.ToFrozenSet();
}
///
/// Whether two option sets describe the same broker session. Note the explicit
/// : the identity projections are object-typed
/// anonymous instances, so != would compare references and report "changed"
/// every single time — which would tear down and rebuild a perfectly good session on every
/// redeploy, and (via ) swap in a subscription manager that is not
/// attached to the live connection.
///
private static bool SameSession(MqttDriverOptions a, MqttDriverOptions b)
=> SessionIdentity(a).Equals(SessionIdentity(b));
///
/// Whether two option sets produce an identical ingest path —
/// in Plain mode, in Sparkplug B.
///
///
/// ⚠️ Every setting an ingest path captures at construction MUST be named by
/// . A setting that is captured but unnamed makes a delta
/// changing it compare "same ingest", get applied in place, and silently never reach the
/// component that needed rebuilding — the driver would keep subscribing to the OLD Sparkplug
/// group id and report perfect health. Task 21 closed the Sparkplug half of that gap by naming
/// the whole record (it has value equality, so a member
/// added there is covered automatically — which is the point of naming the record rather than
/// enumerating its fields).
///
private static bool SameIngest(MqttDriverOptions a, MqttDriverOptions b)
=> IngestIdentity(a).Equals(IngestIdentity(b));
///
/// The settings whose change invalidates the live broker session. Projected onto an anonymous
/// type (structural Equals) rather than comparing the whole options record, whose
/// RawTags list would make every comparison unequal by reference.
///
private static object SessionIdentity(MqttDriverOptions o) => new
{
o.Host,
o.Port,
o.ClientId,
o.UseTls,
o.AllowUntrustedServerCertificate,
o.CaCertificatePath,
o.Username,
o.Password,
o.ProtocolVersion,
o.CleanSession,
o.KeepAliveSeconds,
o.ConnectTimeoutSeconds,
o.ReconnectMinBackoffSeconds,
o.ReconnectMaxBackoffSeconds,
Ingest = IngestIdentity(o),
};
/// The settings the live ingest path captures at construction.
///
/// is named as a whole record rather than field-by-field: it
/// has structural value equality, so a member added to it in a later task is covered here with
/// no edit — and the failure mode of forgetting one is silent (see ).
///
private static object IngestIdentity(MqttDriverOptions o) => new
{
o.Mode,
DefaultQos = o.Plain?.DefaultQos,
o.MaxPayloadBytes,
o.Sparkplug,
};
/// Shared teardown for , and the session rebuild.
private async Task TeardownAsync()
{
var probe = Interlocked.Exchange(ref _hostProbeCts, null);
if (probe is not null)
{
await probe.CancelAsync().ConfigureAwait(false);
probe.Dispose();
}
var connection = Interlocked.Exchange(ref _connection, null);
if (connection is not null)
{
await connection.DisposeAsync().ConfigureAwait(false);
}
}
///
/// Starts the host-connectivity poll. exposes state but no
/// state-changed event, and an whose event never fires is
/// a capability that looks wired and is inert — so the transition signal is sampled here.
///
private void StartHostProbe()
{
var cts = new CancellationTokenSource();
var previous = Interlocked.Exchange(ref _hostProbeCts, cts);
if (previous is not null)
{
// Cancel before disposing: disposing a live source leaves its loop running against a
// token that will never be cancelled, and a second probe would fight the first.
previous.Cancel();
previous.Dispose();
}
_ = Task.Run(() => HostProbeLoopAsync(cts.Token), cts.Token);
}
private async Task HostProbeLoopAsync(CancellationToken ct)
{
try
{
while (!ct.IsCancellationRequested)
{
await Task.Delay(HostProbeInterval, ct).ConfigureAwait(false);
var connection = _connection;
if (connection is null)
{
continue;
}
TransitionHost(connection.State switch
{
MqttConnectionState.Connected => HostState.Running,
MqttConnectionState.Reconnecting or MqttConnectionState.Disconnected => HostState.Stopped,
MqttConnectionState.Faulted => HostState.Faulted,
_ => HostState.Unknown,
});
}
}
catch (OperationCanceledException)
{
// Shutdown.
}
catch (Exception ex)
{
_logger?.LogWarning(ex, "MQTT driver '{DriverId}': host connectivity probe stopped.", _driverInstanceId);
}
}
/// Publishes a host state transition once; repeats of the current state are ignored.
private void TransitionHost(HostState next)
{
HostState previous;
lock (_hostLock)
{
if (_hostState == next)
{
return;
}
previous = _hostState;
_hostState = next;
_hostStateChangedUtc = DateTime.UtcNow;
}
OnHostStatusChanged?.Invoke(this, new HostStatusChangedEventArgs(HostName, previous, next));
}
/// Barrier-protected read of the multi-threaded health field.
private DriverHealth ReadHealth() => Volatile.Read(ref _health);
/// Barrier-protected publish of a new health snapshot.
private void WriteHealth(DriverHealth value) => Volatile.Write(ref _health, value);
}