dohertj2/lmxopcua
1
0
Fork 0
Code Issues 3 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

fix(historian-gateway): historize under the historian name, not the mux ref, when HistorianTagname overrides (FU-3)

Browse Source 
The continuous-historization recorder conflated two identifiers into one string:
the dependency mux fans DependencyValueChanged keyed by the driver FullName
(the mux ref), but a value must be historized under the resolved historian name
(HistorianTagname override, else FullName). In the common no-override case the
two are equal, so it worked; with an override they diverge and the recorder
registered mux interest under a key the mux never fans — that tag's values were
never captured (and, had they been, would have been written under the mux ref).

Carry BOTH identifiers through the seam: a new HistorizedTagRef(MuxRef,
HistorianName) record on IHistorizedTagSubscriptionSink. The applier resolves
MuxRef = FullName and HistorianName = override-or-FullName. The recorder now
keeps a muxRef->historianName map: it registers/filters mux interest by MuxRef
but writes the outbox entry (and drains) under HistorianName. The convergence
handler re-registers the mux only when the registered key-set changes, so an
override-only rename (same FullName) updates the write target without mux churn.

Tests: a divergent-override recorder test (interest by mux ref, value written
under the override name, never the mux ref) + an override-rename no-churn test;
the applier feed tests now assert the full (mux ref, historian name) pairs.
Runtime 348/0, OpcUaServer 327/0; 0 warnings. Closes FU-3.

Claude-Session: https://claude.ai/code/session_012SDSQ3AcaXqPcBtDESBRii
This commit is contained in:
Joseph Doherty
2026-06-27 00:43:28 -04:00
parent b2276b5b04
commit 111adc92b6
6 changed files with 203 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/AddressSpaceApplier.cs
+22 -16
View File
@@ -309,37 +309,38 @@ public sealed class AddressSpaceApplier
{
try
{
List<string>? added = null;
List<string>? removed = null;
List<HistorizedTagRef>? added = null;
List<HistorizedTagRef>? removed = null;
// Added historized value variables → new interest.
foreach (var tag in plan.AddedEquipmentTags)
{
if (HistorizedRef(tag) is { } r) (added ??= new List<string>()).Add(r);
if (HistorizedRef(tag) is { } r) (added ??= new List<HistorizedTagRef>()).Add(r);
}
// Removed historized value variables → drop interest.
foreach (var tag in plan.RemovedEquipmentTags)
{
if (HistorizedRef(tag) is { } r) (removed ??= new List<string>()).Add(r);
if (HistorizedRef(tag) is { } r) (removed ??= new List<HistorizedTagRef>()).Add(r);
}
// Changed tags: the historized ref may have flipped on/off or been renamed (override/FullName
// change). Compare previous-vs-current resolved refs — an unchanged ref is a no-op.
// change). Compare previous-vs-current resolved ref PAIRS (record equality compares both the
// mux ref and the historian name) — an unchanged pair is a no-op.
foreach (var d in plan.ChangedEquipmentTags)
{
var prev = HistorizedRef(d.Previous);
var cur = HistorizedRef(d.Current);
if (string.Equals(prev, cur, StringComparison.Ordinal)) continue;
if (prev is not null) (removed ??= new List<string>()).Add(prev);
if (cur is not null) (added ??= new List<string>()).Add(cur);
if (prev == cur) continue;
if (prev is not null) (removed ??= new List<HistorizedTagRef>()).Add(prev);
if (cur is not null) (added ??= new List<HistorizedTagRef>()).Add(cur);
}
if (added is null && removed is null) return;
_historizedSubscriptions.UpdateHistorizedRefs(
added ?? (IReadOnlyList<string>)Array.Empty<string>(),
removed ?? (IReadOnlyList<string>)Array.Empty<string>());
added ?? (IReadOnlyList<HistorizedTagRef>)Array.Empty<HistorizedTagRef>(),
removed ?? (IReadOnlyList<HistorizedTagRef>)Array.Empty<HistorizedTagRef>());
}
catch (Exception ex)
{
@@ -350,16 +351,21 @@ public sealed class AddressSpaceApplier
}
/// <summary>
/// Resolve the historized tag ref for <paramref name="tag"/> EXACTLY as the provisioning hook /
/// materialiser do: a non-alarm historized value variable's <c>HistorianTagname</c> override,
/// else its driver-side <c>FullName</c>. Returns <c>null</c> when the tag is not a historized
/// Resolve the historized tag ref for <paramref name="tag"/> as a
/// <see cref="HistorizedTagRef"/> carrying BOTH identifiers the recorder needs: the
/// <c>MuxRef</c> = the driver-side <c>FullName</c> the dependency mux fans values by, and the
/// <c>HistorianName</c> = the value the EnsureTags hook / materialiser write under (a non-alarm
/// historized value variable's <c>HistorianTagname</c> override, else its <c>FullName</c>). The
/// two diverge ONLY when an override is set. Returns <c>null</c> when the tag is not a historized
/// value variable (not historized, or a native-alarm condition node).
/// </summary>
/// <param name="tag">The equipment tag to resolve a historized ref for.</param>
/// <returns>The resolved historian ref, or <c>null</c> when the tag is not a historized value variable.</returns>
private static string? HistorizedRef(EquipmentTagPlan tag) =>
/// <returns>The resolved historized ref pair, or <c>null</c> when the tag is not a historized value variable.</returns>
private static HistorizedTagRef? HistorizedRef(EquipmentTagPlan tag) =>
tag.IsHistorized && tag.Alarm is null
? (string.IsNullOrWhiteSpace(tag.HistorianTagname) ? tag.FullName : tag.HistorianTagname)
? new HistorizedTagRef(
tag.FullName,
string.IsNullOrWhiteSpace(tag.HistorianTagname) ? tag.FullName : tag.HistorianTagname)
: null;
private void SafeRebuild()
src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ActorHistorizedTagSubscriptionSink.cs
+1 -1
View File
@@ -24,7 +24,7 @@ public sealed class ActorHistorizedTagSubscriptionSink : IHistorizedTagSubscript
}
/// <inheritdoc />
public void UpdateHistorizedRefs(IReadOnlyList<string> added, IReadOnlyList<string> removed)
public void UpdateHistorizedRefs(IReadOnlyList<HistorizedTagRef> added, IReadOnlyList<HistorizedTagRef> removed)
{
ArgumentNullException.ThrowIfNull(added);
ArgumentNullException.ThrowIfNull(removed);
src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationRecorder.cs
+57 -24
View File
@@ -1,5 +1,6 @@
using Akka.Actor;
using Akka.Event;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions.Historian;
using ZB.MOM.WW.OtOpcUa.Runtime.VirtualTags;
@@ -67,12 +68,14 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers
/// Converge the recorder's historized-ref interest by an add/remove DELTA — sent by the
/// address-space applier (via <see cref="ZB.MOM.WW.OtOpcUa.Core.Abstractions.IHistorizedTagSubscriptionSink"/>)
/// after every deploy. The applier only ever sees a plan diff, so it feeds a delta; the recorder
/// holds the full set and re-registers it (see <see cref="OnUpdateHistorizedRefs"/>). The refs are
/// the same ones the EnsureTags provisioning hook resolves (override-or-FullName).
/// holds the full set and re-registers it (see <see cref="OnUpdateHistorizedRefs"/>). Each ref
/// carries both its mux key (<see cref="HistorizedTagRef.MuxRef"/>, the driver ref the mux fans by)
/// and the resolved historian name (<see cref="HistorizedTagRef.HistorianName"/>,
/// override-or-FullName — the same name the EnsureTags provisioning hook ensures).
/// </summary>
/// <param name="Added">Refs newly historized by this deploy.</param>
/// <param name="Removed">Refs no longer historized by this deploy.</param>
public sealed record UpdateHistorizedRefs(IReadOnlyList<string> Added, IReadOnlyList<string> Removed);
public sealed record UpdateHistorizedRefs(IReadOnlyList<HistorizedTagRef> Added, IReadOnlyList<HistorizedTagRef> Removed);
/// <summary>A point-in-time snapshot of the recorder's counters.</summary>
/// <param name="QueuedDepth">Un-acked entries currently held in the durable outbox.</param>
@@ -93,8 +96,12 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers
private readonly IActorRef _dependencyMux;
private readonly IHistorianValueWriter _writer;
private readonly IHistorizationOutbox _outbox;
private readonly IReadOnlyList<string> _historizedRefs;
private readonly HashSet<string> _historizedSet;
/// <summary>The tracked historized tags, keyed by mux ref (<see cref="HistorizedTagRef.MuxRef"/> — the
/// driver ref the mux fans by, and the key mux interest is registered under) → the historian name the
/// value is written under (<see cref="HistorizedTagRef.HistorianName"/>). A <c>HistorianTagname</c>
/// override is the only case the two diverge; in the common case they are equal.</summary>
private readonly Dictionary<string, string> _refMap;
private readonly int _drainBatchSize;
private readonly TimeSpan _drainInterval;
private readonly TimeSpan _minBackoff;
@@ -157,8 +164,15 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers
_dependencyMux = dependencyMux ?? throw new ArgumentNullException(nameof(dependencyMux));
_writer = writer ?? throw new ArgumentNullException(nameof(writer));
_outbox = outbox ?? throw new ArgumentNullException(nameof(outbox));
_historizedRefs = historizedRefs ?? throw new ArgumentNullException(nameof(historizedRefs));
_historizedSet = new HashSet<string>(_historizedRefs, StringComparer.Ordinal);
ArgumentNullException.ThrowIfNull(historizedRefs);
// The ctor seed is the no-override identity (mux ref == historian name); production seeds an EMPTY
// set and converges via UpdateHistorizedRefs on each deploy (which carries diverging override pairs).
_refMap = new Dictionary<string, string>(StringComparer.Ordinal);
foreach (string r in historizedRefs)
{
_refMap[r] = r;
}
_drainBatchSize = drainBatchSize > 0 ? drainBatchSize : 64;
_drainInterval = drainInterval is { } di && di > TimeSpan.Zero ? di : TimeSpan.FromSeconds(2);
_minBackoff = minBackoff is { } mb && mb > TimeSpan.Zero ? mb : TimeSpan.FromSeconds(1);
@@ -182,8 +196,9 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers
/// <inheritdoc />
protected override void PreStart()
{
// Register interest for the historized refs so the mux fans their DependencyValueChanged to us.
_dependencyMux.Tell(new DependencyMuxActor.RegisterInterest(_historizedRefs, Self));
// Register interest by mux ref (the key the mux fans DependencyValueChanged by) so it fans those
// tags' values to us. Historian-name overrides are tracked separately for the write side.
_dependencyMux.Tell(new DependencyMuxActor.RegisterInterest(_refMap.Keys.ToList(), Self));
// Seed the steady drain cadence; appends also nudge a prompt drain (see OnValueChangedAsync).
Timers.StartSingleTimer(DrainTimerKey, DrainTick.Instance, _drainInterval);
base.PreStart();
@@ -202,8 +217,11 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers
private async Task OnValueChangedAsync(VirtualTagActor.DependencyValueChanged msg)
{
// Defensive: only historize refs we registered interest for (the mux already scopes to these).
if (!_historizedSet.Contains(msg.TagId))
// The mux fans values keyed by the driver ref (msg.TagId == the mux ref). Only historize refs we
// registered interest for, and resolve the HISTORIAN NAME to write under — which differs from the
// mux ref when a HistorianTagname override is set. (The mux already scopes to our registered refs;
// this is also the defensive filter.)
if (!_refMap.TryGetValue(msg.TagId, out string? historianName))
{
return;
}
@@ -217,9 +235,11 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers
return;
}
// Record under the historian name (override-or-FullName), NOT the mux ref — the outbox entry +
// the drain's WriteLiveValues target the historian tag the EnsureTags hook provisioned.
var entry = new HistorizationOutboxEntry(
Guid.NewGuid(),
msg.TagId,
historianName,
numeric,
GoodQuality,
DateTime.SpecifyKind(msg.TimestampUtc, DateTimeKind.Utc));
@@ -277,31 +297,44 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers
/// </summary>
private void OnUpdateHistorizedRefs(UpdateHistorizedRefs msg)
{
var next = new HashSet<string>(_historizedSet, StringComparer.Ordinal);
next.ExceptWith(msg.Removed);
next.UnionWith(msg.Added);
if (next.SetEquals(_historizedSet))
// Snapshot the registered mux key-set BEFORE applying the delta, so we re-register only when the
// set the mux fans by actually changes (an override-only rename updates the WRITE target but not
// which refs the mux fans — no mux churn for those).
var beforeKeys = new HashSet<string>(_refMap.Keys, StringComparer.Ordinal);
// Apply removed-then-added so a ref present in BOTH (a HistorianTagname override changed while the
// mux ref / FullName stayed the same) ends mapped to its NEW historian name.
foreach (HistorizedTagRef r in msg.Removed)
{
// No net change — stay idempotent (no mux churn).
_refMap.Remove(r.MuxRef);
}
foreach (HistorizedTagRef r in msg.Added)
{
_refMap[r.MuxRef] = r.HistorianName;
}
if (_refMap.Keys.ToHashSet(StringComparer.Ordinal).SetEquals(beforeKeys))
{
// Mux key-set unchanged (no-op delta, or an override-only rename) — the map (write targets) is
// already updated; stay idempotent at the mux (no register/unregister churn).
return;
}
_historizedSet.Clear();
_historizedSet.UnionWith(next);
if (_historizedSet.Count == 0)
if (_refMap.Count == 0)
{
// The mux has no per-ref unregister; drop ALL interest in one message.
_dependencyMux.Tell(new DependencyMuxActor.UnregisterInterest(Self));
}
else
{
// RegisterInterest REPLACES the prior set at the mux, so one message converges it exactly.
_dependencyMux.Tell(new DependencyMuxActor.RegisterInterest(_historizedSet.ToList(), Self));
// RegisterInterest REPLACES the prior set at the mux, so one message (carrying the mux keys)
// converges it exactly.
_dependencyMux.Tell(new DependencyMuxActor.RegisterInterest(_refMap.Keys.ToList(), Self));
}
_log.Debug("ContinuousHistorization: historized-ref interest converged to {Count} ref(s).",
_historizedSet.Count);
_refMap.Count);
}
private void OnDrainTick()