feat(mqtt): AliasTable/BirthCache — bind-by-name, rebuild-per-birth
Design §3.6 invariants #1/#2. A Sparkplug alias is scoped to ONE birth: after a rebirth an edge node may point alias 5 at a different metric. So authored tags bind by the stable metric NAME and the alias is a per-birth cache only — RebuildFromBirth builds both indexes fresh and publishes them in one assignment, never merging or upserting, including for an empty birth (which legitimately clears the scope). BirthCache keys scopes by the full (group, edgeNode, device?) triple and applies the spec's node→device rules: an NBIRTH invalidates every DBIRTH under that node (returning their metric sets for the STALE fan-out), a death evicts rather than blanks so data-before-rebirth stays detectable. Reads are lock-free on MQTTnet's dispatcher thread — an immutable snapshot swapped atomically per scope, the same discipline MqttSubscriptionManager.AuthoredTable uses, with rebuilds landing in place so a held table reference always observes the newest birth. Running values are deliberately NOT stored here — LastValueCache owns those, keyed by RawPath. SparkplugMetricBinding.Reinterpret exposes the birth datatype to the codec's two's-complement fix, since a DATA metric carries no datatype. Falsifiability: mutating RebuildFromBirth to merge reddened 5/19 tests (both plan invariants); making the alias carry metric identity across a birth reddened 3/19, including the headline alias-reuse case. Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW
This commit is contained in:
Joseph Doherty
@@ -0,0 +1,263 @@
|
||||
using System.Collections.Frozen;
|
||||
using System.Diagnostics.CodeAnalysis;
|
||||
using TahuDataType = Org.Eclipse.Tahu.Protobuf.DataType;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Sparkplug;
|
||||
|
||||
/// <summary>
|
||||
/// One <c>(edgeNode, device)</c> scope's metric catalog, as declared by its most recent
|
||||
/// NBIRTH/DBIRTH: the alias→metric cache <b>and</b> the stable name→metric index, rebuilt wholesale
|
||||
/// on every birth. Design doc §3.6 invariants #1 and #2.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// <b>The alias is a per-birth cache; the NAME is the binding key.</b> Sparkplug's alias
|
||||
/// mechanism exists to shrink DATA payloads — a BIRTH declares <c>name ↔ alias</c> and every
|
||||
/// subsequent DATA metric carries only the alias — and the alias is scoped to <i>that one
|
||||
/// birth</i>. After a rebirth an edge node is entirely free to point alias 5 at a different
|
||||
/// metric. So an authored tag binds by <see cref="SparkplugMetricBinding.Name"/>; the alias is
|
||||
/// only ever the lookup that turns an incoming DATA metric back into a name. Binding a tag to
|
||||
/// an alias instead means a rebirth silently starts routing Pressure into a Temperature tag —
|
||||
/// <b>good quality, plausible values, completely wrong</b>, and invisible unless something
|
||||
/// specifically reuses an alias across a rebirth.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b><see cref="RebuildFromBirth"/> REPLACES; it never merges or upserts.</b> The whole
|
||||
/// snapshot — both indexes — is built fresh and published in one assignment. An alias the new
|
||||
/// birth did not declare is gone, including when the new birth declares no metrics at all. This
|
||||
/// is the single behaviour the type exists to guarantee: a merge (or an "optimisation" that
|
||||
/// skips an empty birth) reintroduces exactly the stale-alias mis-route above.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Both indexes are replaced together, from the same birth.</b> They are deliberately held
|
||||
/// in one type rather than split across an "alias table" and a separate "birth cache": they are
|
||||
/// two views of one birth, and any seam between them is somewhere the two can be replaced
|
||||
/// independently and skew — an alias resolving through the new birth into a name catalog still
|
||||
/// holding the old one.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Thread-safety: an immutable snapshot swapped atomically</b>, the same discipline
|
||||
/// <c>MqttSubscriptionManager.AuthoredTable</c> uses and for the same reason — this is read on
|
||||
/// MQTTnet's shared dispatcher thread as messages arrive, and a reader must never see a
|
||||
/// half-rebuilt map. No lock is taken on any path.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Last values do not live here.</b> The running last-observed value per tag is
|
||||
/// <see cref="LastValueCache"/>'s, keyed by RawPath — the identity every publish in this driver
|
||||
/// is made under. A second copy keyed by metric name would be a duplicate source of truth with
|
||||
/// a different key, and it would grow with every metric the plant births whether or not any tag
|
||||
/// was ever authored for it. A birth's own values flow straight through the ingest path to that
|
||||
/// cache; nothing is retained here.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
public sealed class AliasTable
|
||||
{
|
||||
private volatile Snapshot _snapshot = Snapshot.Empty;
|
||||
|
||||
/// <summary>How many distinct metrics the most recent birth declared.</summary>
|
||||
public int Count => _snapshot.Metrics.Length;
|
||||
|
||||
/// <summary>
|
||||
/// Every metric the most recent birth declared. The consumer's STALE fan-out on a death, and
|
||||
/// the discovery path's "which metrics does this scope have" question, both read this.
|
||||
/// </summary>
|
||||
public IReadOnlyList<SparkplugMetricBinding> Metrics => _snapshot.Metrics;
|
||||
|
||||
/// <summary>
|
||||
/// Replaces this scope's whole catalog with <paramref name="birthMetrics"/> — the metrics of one
|
||||
/// NBIRTH/DBIRTH payload, exactly as <see cref="SparkplugCodec"/> projected them.
|
||||
/// </summary>
|
||||
/// <param name="birthMetrics">The birth payload's metrics.</param>
|
||||
/// <returns>
|
||||
/// What was installed, and what was refused — the caller's only window onto a malformed birth,
|
||||
/// since this type takes no logger.
|
||||
/// </returns>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// <b>Only an unusable NAME is a rejection.</b> A metric with a blank/absent name cannot be
|
||||
/// bound to an authored tag by any route, and storing it alias-only would create a binding
|
||||
/// that can never resolve (or, worse, one that matches a tag whose authored metric name
|
||||
/// happens to be blank), so it is dropped and counted.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>A missing datatype is NOT a rejection</b> — it becomes
|
||||
/// <see cref="TahuDataType.Unknown"/>. Dropping such a metric would turn every DATA message
|
||||
/// for its alias into an unknown-alias event, which the ingest state machine answers with a
|
||||
/// rebirth request; the edge node would answer with the same malformed birth, and the pair
|
||||
/// would loop. Kept as Unknown, the typed layer refuses the value once
|
||||
/// (<c>ToDriverDataType()</c> returns null for it) and nothing storms.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>An alias claimed by two metrics in one birth resolves to neither.</b> Any choice
|
||||
/// between them is a coin flip between two different signals, which is the mis-route this
|
||||
/// type exists to prevent; the alias is dropped (the consumer sees an unknown alias and can
|
||||
/// request a rebirth) while both metrics stay bindable by name. A duplicate <i>name</i>, by
|
||||
/// contrast, is last-wins: the name is the binding key, so evicting it would take the
|
||||
/// authored tag dark entirely.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
public BirthApplyResult RebuildFromBirth(IEnumerable<SparkplugMetric> birthMetrics)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(birthMetrics);
|
||||
|
||||
var byName = new Dictionary<string, SparkplugMetricBinding>(StringComparer.Ordinal);
|
||||
var byAlias = new Dictionary<ulong, SparkplugMetricBinding>();
|
||||
HashSet<ulong>? collidedAliases = null;
|
||||
var rejectedUnnamed = 0;
|
||||
var duplicateNames = 0;
|
||||
var aliasCollisions = 0;
|
||||
|
||||
foreach (var metric in birthMetrics)
|
||||
{
|
||||
if (string.IsNullOrWhiteSpace(metric.Name))
|
||||
{
|
||||
rejectedUnnamed++;
|
||||
continue;
|
||||
}
|
||||
|
||||
var binding = new SparkplugMetricBinding(metric.Name, metric.Alias, metric.DataType ?? TahuDataType.Unknown);
|
||||
|
||||
if (!byName.TryAdd(binding.Name, binding))
|
||||
{
|
||||
duplicateNames++;
|
||||
byName[binding.Name] = binding;
|
||||
}
|
||||
|
||||
if (binding.Alias is { } alias && !byAlias.TryAdd(alias, binding))
|
||||
{
|
||||
aliasCollisions++;
|
||||
(collidedAliases ??= []).Add(alias);
|
||||
}
|
||||
}
|
||||
|
||||
if (collidedAliases is not null)
|
||||
{
|
||||
foreach (var alias in collidedAliases)
|
||||
{
|
||||
byAlias.Remove(alias);
|
||||
}
|
||||
}
|
||||
|
||||
// A duplicate name is last-wins in `byName`, so the alias index can still hold the losing
|
||||
// binding — drop those too rather than let an alias resolve to a name that resolves elsewhere.
|
||||
if (duplicateNames > 0)
|
||||
{
|
||||
foreach (var alias in byAlias.Where(kv => !ReferenceEquals(byName[kv.Value.Name], kv.Value))
|
||||
.Select(kv => kv.Key).ToList())
|
||||
{
|
||||
byAlias.Remove(alias);
|
||||
}
|
||||
}
|
||||
|
||||
// THE replace. One assignment of a fully-built snapshot — no merge, no upsert, and no
|
||||
// short-circuit for an empty birth (an empty birth legitimately clears the scope).
|
||||
_snapshot = new Snapshot(
|
||||
byAlias.ToFrozenDictionary(),
|
||||
byName.ToFrozenDictionary(StringComparer.Ordinal),
|
||||
[.. byName.Values]);
|
||||
|
||||
return new BirthApplyResult(byName.Count, rejectedUnnamed, aliasCollisions, duplicateNames);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Resolves a DATA metric's alias to the metric the <b>most recent</b> birth bound it to, or
|
||||
/// <see langword="null"/> when this birth declared no such alias.
|
||||
/// </summary>
|
||||
/// <param name="alias">The alias carried by an inbound DATA metric.</param>
|
||||
/// <returns>The binding, or <see langword="null"/> — which the consumer treats as an unknown alias.</returns>
|
||||
public SparkplugMetricBinding? Resolve(ulong alias) =>
|
||||
_snapshot.ByAlias.TryGetValue(alias, out var binding) ? binding : null;
|
||||
|
||||
/// <summary>The <see cref="Resolve(ulong)"/> try-shape.</summary>
|
||||
/// <param name="alias">The alias carried by an inbound DATA metric.</param>
|
||||
/// <param name="binding">The binding when this returns <see langword="true"/>.</param>
|
||||
/// <returns><see langword="true"/> when the most recent birth declared this alias.</returns>
|
||||
public bool TryResolve(ulong alias, [NotNullWhen(true)] out SparkplugMetricBinding? binding)
|
||||
{
|
||||
binding = Resolve(alias);
|
||||
return binding is not null;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Resolves by the stable metric name — the binding key. Used for a DATA metric that carried its
|
||||
/// name rather than an alias (aliases are optional in Sparkplug), and to answer "does this scope
|
||||
/// publish the metric this tag was authored against".
|
||||
/// </summary>
|
||||
/// <param name="name">The stable metric name. Matched ordinally; Sparkplug names are case-sensitive.</param>
|
||||
/// <returns>The binding, or <see langword="null"/>.</returns>
|
||||
public SparkplugMetricBinding? ResolveByName(string name) =>
|
||||
!string.IsNullOrEmpty(name) && _snapshot.ByName.TryGetValue(name, out var binding) ? binding : null;
|
||||
|
||||
/// <summary>The <see cref="ResolveByName"/> try-shape.</summary>
|
||||
/// <param name="name">The stable metric name.</param>
|
||||
/// <param name="binding">The binding when this returns <see langword="true"/>.</param>
|
||||
/// <returns><see langword="true"/> when the most recent birth declared this metric.</returns>
|
||||
public bool TryResolveByName(string name, [NotNullWhen(true)] out SparkplugMetricBinding? binding)
|
||||
{
|
||||
binding = ResolveByName(name);
|
||||
return binding is not null;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// One birth's catalog, immutable and published in a single assignment so the dispatcher thread
|
||||
/// always reads one whole birth's worth of state.
|
||||
/// </summary>
|
||||
/// <param name="ByAlias">Alias → binding, for the birth that declared it. Collided aliases are absent.</param>
|
||||
/// <param name="ByName">Stable metric name → binding — the index authored tags bind through.</param>
|
||||
/// <param name="Metrics">Every declared metric, for the STALE fan-out and discovery.</param>
|
||||
private sealed record Snapshot(
|
||||
FrozenDictionary<ulong, SparkplugMetricBinding> ByAlias,
|
||||
FrozenDictionary<string, SparkplugMetricBinding> ByName,
|
||||
SparkplugMetricBinding[] Metrics)
|
||||
{
|
||||
public static readonly Snapshot Empty = new(
|
||||
FrozenDictionary<ulong, SparkplugMetricBinding>.Empty,
|
||||
FrozenDictionary<string, SparkplugMetricBinding>.Empty,
|
||||
[]);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>One metric as its birth declared it: the stable name, the per-birth alias, the datatype.</summary>
|
||||
/// <param name="Name">
|
||||
/// The stable metric name — <b>the binding key</b>. An authored tag's <c>metricName</c> matches
|
||||
/// against this, never against <paramref name="Alias"/>.
|
||||
/// </param>
|
||||
/// <param name="Alias">
|
||||
/// The alias this birth assigned, or <see langword="null"/> when the birth declared none (legal —
|
||||
/// such a publisher sends names in DATA too). <b>Valid only until the next birth for this scope.</b>
|
||||
/// </param>
|
||||
/// <param name="DataType">
|
||||
/// The datatype the birth declared, or <see cref="TahuDataType.Unknown"/> when it declared none.
|
||||
/// This is the <i>only</i> place a DATA metric's datatype can come from — DATA carries none.
|
||||
/// </param>
|
||||
public sealed record SparkplugMetricBinding(string Name, ulong? Alias, TahuDataType DataType)
|
||||
{
|
||||
/// <summary>
|
||||
/// Applies <see cref="SparkplugCodec.ReinterpretSigned"/> to a raw DATA value using
|
||||
/// <b>this birth's</b> datatype.
|
||||
/// </summary>
|
||||
/// <param name="rawValue">The raw <see cref="SparkplugMetric.Value"/> from a DATA metric.</param>
|
||||
/// <returns>The signed value for a signed-integer datatype; <paramref name="rawValue"/> otherwise.</returns>
|
||||
/// <remarks>
|
||||
/// Exposed here because this is the one object that holds the datatype at the moment a DATA
|
||||
/// value is being resolved, and forgetting the call is silent: Sparkplug carries Int8/16/32/64
|
||||
/// as two's complement in an <i>unsigned</i> proto field, so a metric whose value is -42
|
||||
/// publishes as 4294967254 — plausible, Good quality, and invisible until someone reads a gauge.
|
||||
/// </remarks>
|
||||
public object? Reinterpret(object? rawValue) => SparkplugCodec.ReinterpretSigned(rawValue, DataType);
|
||||
}
|
||||
|
||||
/// <summary>What one <see cref="AliasTable.RebuildFromBirth"/> installed, and what it refused.</summary>
|
||||
/// <param name="Accepted">Distinct named metrics installed.</param>
|
||||
/// <param name="RejectedUnnamed">Metrics dropped for carrying no usable name.</param>
|
||||
/// <param name="AliasCollisions">Aliases claimed by more than one metric, and therefore left unresolvable.</param>
|
||||
/// <param name="DuplicateNames">Metrics whose name a later metric in the same birth overwrote.</param>
|
||||
public readonly record struct BirthApplyResult(
|
||||
int Accepted,
|
||||
int RejectedUnnamed,
|
||||
int AliasCollisions,
|
||||
int DuplicateNames)
|
||||
{
|
||||
/// <summary>Whether the birth was malformed in any way worth logging.</summary>
|
||||
public bool HasAnomaly => RejectedUnnamed > 0 || AliasCollisions > 0 || DuplicateNames > 0;
|
||||
}
|
||||
@@ -0,0 +1,234 @@
|
||||
using System.Collections.Concurrent;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Sparkplug;
|
||||
|
||||
/// <summary>
|
||||
/// The driver's live Sparkplug birth state: one <see cref="AliasTable"/> per
|
||||
/// <see cref="SparkplugScope"/>, with the spec's node→device invalidation rules applied on birth
|
||||
/// and on death. Design doc §3.6 invariants #2 and #4.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// <b>Scoping.</b> A scope is the full <c>(group, edgeNode, device?)</c> triple — never the
|
||||
/// device id alone and never <c>node/device</c> without the group, or one plant's
|
||||
/// <c>Filler1</c> would answer for another's. A node-scoped birth (NBIRTH) owns the metrics the
|
||||
/// edge node publishes itself; each DBIRTH owns its device's. They are separate scopes because
|
||||
/// Sparkplug gives them separate alias spaces: alias 5 on <c>EdgeA</c> and alias 5 on
|
||||
/// <c>EdgeA/Filler1</c> are unrelated metrics.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>An NBIRTH invalidates every device under that node.</b> Per the Sparkplug spec an NBIRTH
|
||||
/// means the edge node (re)started and must re-publish a DBIRTH for each of its devices, so
|
||||
/// every previously published DBIRTH is void. Keeping a device's alias table across its node's
|
||||
/// rebirth is the same stale-alias mis-route as
|
||||
/// <see cref="AliasTable.RebuildFromBirth"/> guards against, one scope up — and it is easier to
|
||||
/// miss, because nothing about the DBIRTH itself changed. The invalidated devices are returned
|
||||
/// (<see cref="NodeBirthOutcome.InvalidatedDevices"/>) rather than merely dropped, so the
|
||||
/// consumer can fan STALE out over their metrics without a lookup against a table that has
|
||||
/// already been evicted.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>A death evicts, it does not blank.</b> After an NDEATH/DDEATH the scope is <i>absent</i>,
|
||||
/// which is what makes a DATA message arriving before the next birth a data-before-birth event
|
||||
/// (⇒ request a rebirth, §3.6 invariant #3) rather than a silent unknown-alias drop against a
|
||||
/// table that still looks alive.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Thread-safety.</b> A <see cref="ConcurrentDictionary{TKey,TValue}"/> of scopes, each
|
||||
/// holding an <see cref="AliasTable"/> whose own state is an atomically swapped immutable
|
||||
/// snapshot. A rebirth <i>rebuilds the existing table in place</i> rather than replacing the
|
||||
/// table object, so a consumer holding a reference to a scope's table always observes the newest
|
||||
/// birth instead of quietly reading a detached older one. No lock is taken on any path — this is
|
||||
/// read (and written) on MQTTnet's shared dispatcher thread.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Not a value store.</b> See the remarks on <see cref="AliasTable"/>: running values belong
|
||||
/// to <see cref="LastValueCache"/>, keyed by RawPath.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
public sealed class BirthCache
|
||||
{
|
||||
private readonly ConcurrentDictionary<SparkplugScope, AliasTable> _scopes = new();
|
||||
|
||||
/// <summary>How many scopes currently hold a live birth.</summary>
|
||||
public int Count => _scopes.Count;
|
||||
|
||||
/// <summary>A snapshot of the scopes currently holding a live birth.</summary>
|
||||
public IReadOnlyList<SparkplugScope> Scopes => [.. _scopes.Keys];
|
||||
|
||||
/// <summary>
|
||||
/// Applies an NBIRTH: rebuilds the edge node's own catalog and <b>invalidates every device</b>
|
||||
/// under it, per the Sparkplug spec.
|
||||
/// </summary>
|
||||
/// <param name="groupId">The Sparkplug group id.</param>
|
||||
/// <param name="edgeNodeId">The Sparkplug edge-node id.</param>
|
||||
/// <param name="birthMetrics">The NBIRTH payload's metrics.</param>
|
||||
/// <returns>The node's apply result plus the devices this birth invalidated, with their metrics.</returns>
|
||||
public NodeBirthOutcome ApplyNodeBirth(string groupId, string edgeNodeId, IEnumerable<SparkplugMetric> birthMetrics)
|
||||
{
|
||||
ArgumentException.ThrowIfNullOrEmpty(groupId);
|
||||
ArgumentException.ThrowIfNullOrEmpty(edgeNodeId);
|
||||
ArgumentNullException.ThrowIfNull(birthMetrics);
|
||||
|
||||
// Devices first: a device table must never outlive the node birth that voided it, whatever the
|
||||
// node's own rebuild does.
|
||||
var invalidated = Evict(scope => scope.DeviceId is not null && scope.IsUnder(groupId, edgeNodeId));
|
||||
|
||||
var result = TableFor(new SparkplugScope(groupId, edgeNodeId, null)).RebuildFromBirth(birthMetrics);
|
||||
return new NodeBirthOutcome(result, invalidated);
|
||||
}
|
||||
|
||||
/// <summary>Applies a DBIRTH: rebuilds exactly that device's catalog, wholesale.</summary>
|
||||
/// <param name="groupId">The Sparkplug group id.</param>
|
||||
/// <param name="edgeNodeId">The Sparkplug edge-node id.</param>
|
||||
/// <param name="deviceId">The Sparkplug device id.</param>
|
||||
/// <param name="birthMetrics">The DBIRTH payload's metrics.</param>
|
||||
/// <returns>What the birth installed, and what it refused.</returns>
|
||||
public BirthApplyResult ApplyDeviceBirth(
|
||||
string groupId,
|
||||
string edgeNodeId,
|
||||
string deviceId,
|
||||
IEnumerable<SparkplugMetric> birthMetrics)
|
||||
{
|
||||
ArgumentException.ThrowIfNullOrEmpty(groupId);
|
||||
ArgumentException.ThrowIfNullOrEmpty(edgeNodeId);
|
||||
ArgumentException.ThrowIfNullOrEmpty(deviceId);
|
||||
ArgumentNullException.ThrowIfNull(birthMetrics);
|
||||
|
||||
return TableFor(new SparkplugScope(groupId, edgeNodeId, deviceId)).RebuildFromBirth(birthMetrics);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The scope's catalog, or <see langword="null"/> when no live birth has been seen for it —
|
||||
/// which is the consumer's data-before-birth signal.
|
||||
/// </summary>
|
||||
/// <param name="scope">The scope to look up.</param>
|
||||
/// <returns>The catalog, or <see langword="null"/>.</returns>
|
||||
public AliasTable? Find(SparkplugScope scope) => _scopes.TryGetValue(scope, out var table) ? table : null;
|
||||
|
||||
/// <summary>
|
||||
/// Applies an NDEATH: forgets the edge node and every device under it, returning what was
|
||||
/// forgotten so the consumer can emit STALE for those metrics.
|
||||
/// </summary>
|
||||
/// <param name="groupId">The Sparkplug group id.</param>
|
||||
/// <param name="edgeNodeId">The Sparkplug edge-node id.</param>
|
||||
/// <returns>The evicted scopes and their metrics; empty when nothing was live.</returns>
|
||||
public IReadOnlyList<SparkplugScopeMetrics> ForgetNode(string groupId, string edgeNodeId)
|
||||
{
|
||||
ArgumentException.ThrowIfNullOrEmpty(groupId);
|
||||
ArgumentException.ThrowIfNullOrEmpty(edgeNodeId);
|
||||
|
||||
return Evict(scope => scope.IsUnder(groupId, edgeNodeId));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Applies a DDEATH: forgets exactly one device, returning its metrics for the STALE fan-out.
|
||||
/// The node's own catalog is untouched.
|
||||
/// </summary>
|
||||
/// <param name="groupId">The Sparkplug group id.</param>
|
||||
/// <param name="edgeNodeId">The Sparkplug edge-node id.</param>
|
||||
/// <param name="deviceId">The Sparkplug device id.</param>
|
||||
/// <param name="removed">The evicted scope and its metrics when this returns <see langword="true"/>.</param>
|
||||
/// <returns><see langword="true"/> when the device had a live birth to forget.</returns>
|
||||
public bool TryForgetDevice(
|
||||
string groupId,
|
||||
string edgeNodeId,
|
||||
string deviceId,
|
||||
out SparkplugScopeMetrics removed)
|
||||
{
|
||||
ArgumentException.ThrowIfNullOrEmpty(groupId);
|
||||
ArgumentException.ThrowIfNullOrEmpty(edgeNodeId);
|
||||
ArgumentException.ThrowIfNullOrEmpty(deviceId);
|
||||
|
||||
var scope = new SparkplugScope(groupId, edgeNodeId, deviceId);
|
||||
if (_scopes.TryRemove(scope, out var table))
|
||||
{
|
||||
removed = new SparkplugScopeMetrics(scope, table.Metrics);
|
||||
return true;
|
||||
}
|
||||
|
||||
removed = default;
|
||||
return false;
|
||||
}
|
||||
|
||||
/// <summary>Forgets every scope — the disconnect/reinitialise reset.</summary>
|
||||
public void Clear() => _scopes.Clear();
|
||||
|
||||
/// <summary>
|
||||
/// The scope's catalog, created empty on first use. Rebuilds happen <b>in place</b> on the
|
||||
/// returned instance, so a consumer holding it observes the newest birth rather than a detached
|
||||
/// older one.
|
||||
/// </summary>
|
||||
/// <param name="scope">The scope.</param>
|
||||
/// <returns>The scope's catalog.</returns>
|
||||
private AliasTable TableFor(SparkplugScope scope) => _scopes.GetOrAdd(scope, static _ => new AliasTable());
|
||||
|
||||
/// <summary>Removes every scope matching <paramref name="predicate"/>, capturing their metrics first.</summary>
|
||||
/// <param name="predicate">Which scopes to evict.</param>
|
||||
/// <returns>The evicted scopes and the metrics they held.</returns>
|
||||
private List<SparkplugScopeMetrics> Evict(Func<SparkplugScope, bool> predicate)
|
||||
{
|
||||
List<SparkplugScopeMetrics>? evicted = null;
|
||||
|
||||
// ConcurrentDictionary enumeration is weakly consistent, which is exactly right here: this runs
|
||||
// on the dispatcher thread and only needs to see the scopes that existed when the birth/death
|
||||
// arrived.
|
||||
foreach (var scope in _scopes.Keys)
|
||||
{
|
||||
if (!predicate(scope) || !_scopes.TryRemove(scope, out var table))
|
||||
{
|
||||
continue;
|
||||
}
|
||||
|
||||
(evicted ??= []).Add(new SparkplugScopeMetrics(scope, table.Metrics));
|
||||
}
|
||||
|
||||
return evicted ?? [];
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A Sparkplug metric-namespace scope: an edge node, or one device under it. The identity every
|
||||
/// birth, death and alias resolution is keyed by.
|
||||
/// </summary>
|
||||
/// <param name="GroupId">The Sparkplug group id.</param>
|
||||
/// <param name="EdgeNodeId">The Sparkplug edge-node id.</param>
|
||||
/// <param name="DeviceId">The Sparkplug device id, or <see langword="null"/> for the node's own scope.</param>
|
||||
public readonly record struct SparkplugScope(string GroupId, string EdgeNodeId, string? DeviceId)
|
||||
{
|
||||
/// <summary>Whether this is a device scope (a DBIRTH's) rather than an edge node's own.</summary>
|
||||
public bool IsDevice => DeviceId is not null;
|
||||
|
||||
/// <summary>The owning edge node's scope — this instance itself when it is already node-scoped.</summary>
|
||||
public SparkplugScope NodeScope => DeviceId is null ? this : new SparkplugScope(GroupId, EdgeNodeId, null);
|
||||
|
||||
/// <summary>Whether this scope belongs to the given edge node (the node's own scope included).</summary>
|
||||
/// <param name="groupId">The Sparkplug group id.</param>
|
||||
/// <param name="edgeNodeId">The Sparkplug edge-node id.</param>
|
||||
/// <returns><see langword="true"/> when both segments match ordinally.</returns>
|
||||
public bool IsUnder(string groupId, string edgeNodeId) =>
|
||||
string.Equals(GroupId, groupId, StringComparison.Ordinal)
|
||||
&& string.Equals(EdgeNodeId, edgeNodeId, StringComparison.Ordinal);
|
||||
|
||||
/// <inheritdoc/>
|
||||
public override string ToString() =>
|
||||
DeviceId is null ? $"{GroupId}/{EdgeNodeId}" : $"{GroupId}/{EdgeNodeId}/{DeviceId}";
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A scope's metrics, captured at the moment it was evicted — the STALE fan-out's input, taken
|
||||
/// before the eviction so it cannot race a re-birth of the same scope.
|
||||
/// </summary>
|
||||
/// <param name="Scope">The scope that was evicted.</param>
|
||||
/// <param name="Metrics">The metrics its last birth declared.</param>
|
||||
public readonly record struct SparkplugScopeMetrics(SparkplugScope Scope, IReadOnlyList<SparkplugMetricBinding> Metrics);
|
||||
|
||||
/// <summary>The result of an NBIRTH: the node's own apply, plus the devices it invalidated.</summary>
|
||||
/// <param name="Result">What the node's catalog rebuild installed and refused.</param>
|
||||
/// <param name="InvalidatedDevices">
|
||||
/// The device scopes this NBIRTH voided, with the metrics they held — the STALE fan-out's input
|
||||
/// until each device re-DBIRTHs.
|
||||
/// </param>
|
||||
public readonly record struct NodeBirthOutcome(
|
||||
BirthApplyResult Result,
|
||||
IReadOnlyList<SparkplugScopeMetrics> InvalidatedDevices);
|
||||
Reference in New Issue
Block a user