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

fix(runtime): bound DependencyMux last-value retention to replay-opting refs (fix round 1 I2)

Browse Source 
DependencyMuxActor retained the last value of every ref ever registered,
and never removed the entries. The continuous-historization recorder
registers every historized ref, so on a historizing node the map grew
with the historized address space and never shrank. The class remark
claimed a bound that did not hold.

- Retain only refs that at least one subscriber registered with
  ReplayRetained: true names (VT actors, the Calculation mux adapter, the
  scripted-alarm host). A non-opting subscriber is still delivered to but
  retains nothing.
- When a ref's last replay-opting subscriber leaves, its value is kept
  for a grace (default 2 min, Props(retentionGrace)) so a restart
  (old adapter gone, new one not yet registered) still replays. A
  periodic sweep then drops it along with the entry. Entries with no
  subscriber and nothing retained are removed.
- New SourceRemoved(DriverInstanceId) evicts the values a removed driver
  produced, so its last Good is never replayed. The mux records the
  publishing driver per retained ref. No production sender exists yet:
  DriverHostActor is outside this lane (follow-up).
- Remark rewritten. Tests: retention bounded to replay-opting refs,
  release after grace (with an in-grace replay as the positive control),
  SourceRemoved eviction, and the restart-replay scenario still green.

Refs #734

Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
This commit is contained in:
Joseph Doherty
2026-10-22 13:31:41 -04:00
parent 6aa5edaf30
commit ad6be56850
2 changed files with 286 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/DependencyMuxActor.cs
+197 -54
View File
@@ -28,53 +28,132 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.VirtualTags;
/// would over-emit if it spanned the cluster.
///
/// <para>
/// <b>Last-value retention + replay (R07:H6).</b> For every ref that has EVER been registered the
/// mux retains the last <c>(value, status, timestamp)</c> it forwarded — through unregisters too,
/// because the case that matters is a subscriber that went away and came back: a Calculation
/// driver Restart is a new driver instance + a new mux adapter, and before this every calc tag sat
/// <c>BadWaitingForInitialData</c> until each of its inputs happened to change again (a
/// scripted-alarm host restart has the same shape). A <see cref="RegisterInterest"/> with
/// <see cref="RegisterInterest.ReplayRetained"/> receives one batch of the retained values for its
/// refs, in the same shape a driver burst would arrive in. Opt-in per registration: the
/// continuous-historization recorder re-registers on every deploy and a replay there would
/// record a stale sample per tag per deploy. Memory is bounded by the number of refs scripts have
/// ever named on this node, not by the address space.
/// <b>Last-value retention + replay (R07:H6, bounded in review fix round 1 / I2).</b> The mux
/// retains the last <c>(value, status, timestamp)</c> of a ref <b>only while at least one subscriber
/// registered with <see cref="RegisterInterest.ReplayRetained"/> names it</b> (the virtual-tag
/// actors, the Calculation mux adapter, the scripted-alarm host). A registration that asks receives
/// one batch of the retained values for its refs, in the shape a driver burst arrives in. The case
/// this serves is a subscriber that goes away and comes back: a Calculation driver Restart is a new
/// driver instance + a new mux adapter, and before this every calc tag sat
/// <c>BadWaitingForInitialData</c> until each of its inputs happened to change again. So when a
/// ref's last replay-opting subscriber leaves, its value is kept for <see cref="RetentionGrace"/>
/// (default <see cref="DefaultRetentionGrace"/>; a returning replay registration inside it cancels
/// the release), then dropped together with the ref's entry.
/// </para>
/// <para>
/// A subscriber that does NOT opt in retains nothing: the continuous-historization recorder
/// registers every historized ref on the node (a replay there would record a stale sample per tag
/// per deploy), and retaining on its behalf made the map grow with the historized address space and
/// never shrink. Memory is therefore bounded by the refs replay-opting subscribers currently name,
/// plus those released within the last grace window. A ref with no subscriber and nothing retained
/// has no entry at all.
/// </para>
/// <para>
/// <b>Source removal.</b> The mux records which driver instance last published each retained ref;
/// <see cref="SourceRemoved"/> evicts every value that driver produced, so a removed driver's last
/// Good is never replayed as current. A comms-loss fan (a Bad batch) already overwrites the retained
/// value with Bad. ⚠️ As of fix round 1 no production actor sends <see cref="SourceRemoved"/> yet —
/// <c>DriverHostActor</c> (outside the scripting lane) must Tell it when it stops / removes a driver
/// child. Until then a removed driver's last value can still be replayed to a replay-opting
/// subscriber that re-registers for that ref while another replay-opting subscriber holds it, or
/// within the grace window.
/// </para>
/// </summary>
public sealed class DependencyMuxActor : ReceiveActor
public sealed class DependencyMuxActor : ReceiveActor, IWithTimers
{
public const string ActorName = "dependency-mux";
/// <summary>How long a ref's retained value outlives its last replay-opting subscriber.</summary>
public static readonly TimeSpan DefaultRetentionGrace = TimeSpan.FromMinutes(2);
/// <summary>Register a subscriber's interest in a set of tag refs. Idempotent on re-register —
/// the second call replaces the prior interest set for that subscriber.</summary>
/// <param name="TagRefs">The refs the subscriber wants.</param>
/// <param name="Subscriber">The subscriber.</param>
/// <param name="ReplayRetained">When true, the mux immediately sends the subscriber one
/// <see cref="VirtualTagActor.DependencyValueBatch"/> holding the last retained value of every ref
/// in <paramref name="TagRefs"/> it has one for (nothing when it has none).</param>
/// <param name="ReplayRetained">When true, the mux retains the last value of every ref in
/// <paramref name="TagRefs"/> while this subscriber is registered, and immediately sends the subscriber one
/// <see cref="VirtualTagActor.DependencyValueBatch"/> holding the last retained value of every ref it has
/// one for (nothing when it has none).</param>
public sealed record RegisterInterest(IReadOnlyList<string> TagRefs, IActorRef Subscriber, bool ReplayRetained = false);
/// <summary>Unregister every interest held by <see cref="Subscriber"/>. Sent on PostStop by
/// the subscriber, or by Terminated handling when the mux watches.</summary>
public sealed record UnregisterInterest(IActorRef Subscriber);
// A ref stays in _byRef once registered — with an EMPTY set after its last subscriber leaves — so
// the hot path's single lookup per item also decides retention. _byRef.Count == 0 keeps meaning
// "nothing was ever registered", which the batch path early-outs on.
private readonly Dictionary<string, HashSet<IActorRef>> _byRef = new(StringComparer.Ordinal);
private readonly Dictionary<IActorRef, HashSet<string>> _bySubscriber = new();
private readonly Dictionary<string, VirtualTagActor.DependencyValueChanged> _last = new(StringComparer.Ordinal);
/// <summary>A driver instance was removed / stopped: evict every retained value it produced so it is
/// never replayed. Subscriptions are untouched.</summary>
/// <param name="DriverInstanceId">The removed driver instance.</param>
public sealed record SourceRemoved(string DriverInstanceId);
private sealed class SweepReleased
{
public static readonly SweepReleased Instance = new();
}
private const string SweepTimerKey = "sweep-released";
/// <summary>Per-ref state. One lookup on the hot path answers both "who wants it" and "retain it?".</summary>
private sealed class RefEntry
{
public readonly HashSet<IActorRef> Subscribers = new();
/// <summary>Registered subscribers of this ref that opted into replay.</summary>
public int Replayers;
/// <summary>When the last replay-opting subscriber left (retention in grace), or null.</summary>
public DateTime? ReleasedAtUtc;
public VirtualTagActor.DependencyValueChanged? Last;
public string? LastSource;
public bool Retains => Replayers > 0 || ReleasedAtUtc is not null;
public bool IsDead => Subscribers.Count == 0 && Replayers == 0 && Last is null;
public void Forget()
{
Last = null;
LastSource = null;
ReleasedAtUtc = null;
}
}
private sealed record Registration(HashSet<string> Refs, bool Replay);
private readonly Dictionary<string, RefEntry> _byRef = new(StringComparer.Ordinal);
private readonly Dictionary<IActorRef, Registration> _bySubscriber = new();
private readonly ILoggingAdapter _log = Context.GetLogger();
/// <summary>How long a released ref's value is kept.</summary>
public TimeSpan RetentionGrace { get; }
/// <inheritdoc />
public ITimerScheduler Timers { get; set; } = null!;
/// <summary>Test hook — refs that currently hold a retained value.</summary>
internal int RetainedRefCount => _byRef.Values.Count(e => e.Last is not null);
/// <summary>Test hook — refs the mux holds any state for (subscribers or a retained value).</summary>
internal int TrackedRefCount => _byRef.Count;
/// <summary>Creates props for the DependencyMuxActor.</summary>
/// <param name="retentionGrace">How long a ref's value outlives its last replay-opting subscriber;
/// null ⇒ <see cref="DefaultRetentionGrace"/>.</param>
/// <returns>The props for creating this actor.</returns>
public static Props Props() => Akka.Actor.Props.Create<DependencyMuxActor>();
public static Props Props(TimeSpan? retentionGrace = null)
{
var grace = retentionGrace ?? DefaultRetentionGrace;
return Akka.Actor.Props.Create(() => new DependencyMuxActor(grace));
}
/// <summary>Initializes a new instance of the <see cref="DependencyMuxActor"/> class.</summary>
public DependencyMuxActor()
/// <param name="retentionGrace">See <see cref="Props"/>; non-positive ⇒ <see cref="DefaultRetentionGrace"/>.</param>
public DependencyMuxActor(TimeSpan retentionGrace)
{
RetentionGrace = retentionGrace > TimeSpan.Zero ? retentionGrace : DefaultRetentionGrace;
Receive<RegisterInterest>(OnRegister);
Receive<UnregisterInterest>(msg => RemoveSubscriber(msg.Subscriber));
Receive<SourceRemoved>(OnSourceRemoved);
Receive<SweepReleased>(_ => OnSweep());
Receive<DriverInstanceActor.AttributeValuePublished>(OnAttributeValuePublished);
Receive<DriverInstanceActor.AttributeValueBatchPublished>(OnAttributeValueBatchPublished);
Receive<Terminated>(msg => RemoveSubscriber(msg.ActorRef));
@@ -82,31 +161,30 @@ public sealed class DependencyMuxActor : ReceiveActor
private void OnRegister(RegisterInterest msg)
{
// Replace any prior interest set so re-registers on actor restart don't leak old refs.
if (_bySubscriber.TryGetValue(msg.Subscriber, out var priorRefs))
{
foreach (var r in priorRefs)
{
if (_byRef.TryGetValue(r, out var set)) set.Remove(msg.Subscriber);
}
}
// Replace any prior interest set so re-registers on actor restart don't leak old refs. Detaching first
// may put a ref into grace; the re-attach below takes it straight back out.
Detach(msg.Subscriber);
var refs = new HashSet<string>(msg.TagRefs, StringComparer.Ordinal);
_bySubscriber[msg.Subscriber] = refs;
_bySubscriber[msg.Subscriber] = new Registration(refs, msg.ReplayRetained);
List<VirtualTagActor.DependencyValueChanged>? replay = null;
foreach (var r in refs)
{
if (!_byRef.TryGetValue(r, out var set))
if (!_byRef.TryGetValue(r, out var entry))
{
set = new HashSet<IActorRef>();
_byRef[r] = set;
entry = new RefEntry();
_byRef[r] = entry;
}
set.Add(msg.Subscriber);
if (msg.ReplayRetained && _last.TryGetValue(r, out var retained))
entry.Subscribers.Add(msg.Subscriber);
if (!msg.ReplayRetained) continue;
entry.Replayers++;
entry.ReleasedAtUtc = null;
if (entry.Last is { } retained)
(replay ??= []).Add(retained);
}
Context.Watch(msg.Subscriber);
_log.Debug("DependencyMux: subscriber {Sub} registered for {Count} refs", msg.Subscriber, refs.Count);
_log.Debug("DependencyMux: subscriber {Sub} registered for {Count} refs (replay {Replay})",
msg.Subscriber, refs.Count, msg.ReplayRetained);
// One batch, after the registration is complete, in exactly the shape a driver burst arrives in —
// so the subscriber's hot path handles it with no special case.
@@ -119,29 +197,89 @@ public sealed class DependencyMuxActor : ReceiveActor
private void RemoveSubscriber(IActorRef subscriber)
{
if (!_bySubscriber.TryGetValue(subscriber, out var refs)) return;
foreach (var r in refs)
{
// The ref's entry stays (possibly empty) so its last value keeps being retained for a
// subscriber that comes back — see the class remarks.
if (_byRef.TryGetValue(r, out var set)) set.Remove(subscriber);
}
_bySubscriber.Remove(subscriber);
if (!Detach(subscriber)) return;
Context.Unwatch(subscriber);
_log.Debug("DependencyMux: subscriber {Sub} removed", subscriber);
}
/// <summary>Drop <paramref name="subscriber"/>'s registration. A ref losing its last replay-opting
/// subscriber goes into grace; a ref left with no state at all is removed.</summary>
private bool Detach(IActorRef subscriber)
{
if (!_bySubscriber.Remove(subscriber, out var registration)) return false;
var now = DateTime.UtcNow;
var released = false;
foreach (var r in registration.Refs)
{
if (!_byRef.TryGetValue(r, out var entry)) continue;
entry.Subscribers.Remove(subscriber);
if (registration.Replay && --entry.Replayers == 0 && entry.Last is not null)
{
entry.ReleasedAtUtc = now;
released = true;
}
if (entry.IsDead) _byRef.Remove(r);
}
if (released && !Timers.IsTimerActive(SweepTimerKey))
Timers.StartPeriodicTimer(SweepTimerKey, SweepReleased.Instance, RetentionGrace);
return true;
}
/// <summary>Drop retained values whose grace has run out — with a periodic sweep at the grace interval a
/// released value lives between one and two grace periods — and stop the sweep once nothing is in grace.</summary>
private void OnSweep()
{
var cutoff = DateTime.UtcNow - RetentionGrace;
var stillInGrace = false;
List<string>? dead = null;
foreach (var (r, entry) in _byRef)
{
if (entry.ReleasedAtUtc is not { } releasedAt) continue;
if (releasedAt > cutoff)
{
stillInGrace = true;
continue;
}
entry.Forget();
if (entry.IsDead) (dead ??= []).Add(r);
}
if (dead is not null)
foreach (var r in dead) _byRef.Remove(r);
if (!stillInGrace) Timers.Cancel(SweepTimerKey);
}
private void OnSourceRemoved(SourceRemoved msg)
{
List<string>? dead = null;
var evicted = 0;
foreach (var (r, entry) in _byRef)
{
if (entry.Last is null || !string.Equals(entry.LastSource, msg.DriverInstanceId, StringComparison.Ordinal))
continue;
entry.Forget();
evicted++;
if (entry.IsDead) (dead ??= []).Add(r);
}
if (dead is not null)
foreach (var r in dead) _byRef.Remove(r);
_log.Debug("DependencyMux: source {Driver} removed — evicted {Count} retained value(s)", msg.DriverInstanceId, evicted);
}
private void OnAttributeValuePublished(DriverInstanceActor.AttributeValuePublished msg)
{
if (!_byRef.TryGetValue(msg.FullReference, out var subscribers))
if (!_byRef.TryGetValue(msg.FullReference, out var entry))
{
// No virtual tag ever cared about this ref — drop. Common in normal operation; the address
// Nothing is registered for (or retaining) this ref — drop. Common in normal operation; the address
// space carries thousands of tags and only a fraction feed virtual-tag expressions.
return;
}
var dep = new VirtualTagActor.DependencyValueChanged(msg.FullReference, msg.Value, msg.TimestampUtc, msg.StatusCode);
_last[msg.FullReference] = dep;
foreach (var sub in subscribers)
if (entry.Retains)
{
entry.Last = dep;
entry.LastSource = msg.DriverInstanceId;
}
foreach (var sub in entry.Subscribers)
{
// A single-item batch, so the mux emits exactly ONE message shape and every subscriber needs
// only the batch receive on its hot path. One list per subscriber: batch ownership transfers
@@ -153,7 +291,7 @@ public sealed class DependencyMuxActor : ReceiveActor
private void OnAttributeValueBatchPublished(DriverInstanceActor.AttributeValueBatchPublished msg)
{
// Nobody has registered anything — skip the whole burst rather than probing an empty map per item.
// Nothing registered and nothing retained — skip the whole burst rather than probing an empty map per item.
if (_byRef.Count == 0) return;
// Allocated lazily on the first matched item: a burst in which nothing is registered for — still
@@ -166,10 +304,15 @@ public sealed class DependencyMuxActor : ReceiveActor
for (var i = 0; i < count; i++)
{
var item = msg.Items[i];
if (!_byRef.TryGetValue(item.FullReference, out var subscribers)) continue;
if (!_byRef.TryGetValue(item.FullReference, out var entry)) continue;
var dep = new VirtualTagActor.DependencyValueChanged(item.FullReference, item.Value, item.TimestampUtc, item.StatusCode);
_last[item.FullReference] = dep;
if (subscribers.Count == 0) continue; // retained for a returning subscriber; nobody to forward to now
if (entry.Retains)
{
entry.Last = dep;
entry.LastSource = msg.DriverInstanceId;
}
var subscribers = entry.Subscribers;
if (subscribers.Count == 0) continue; // retained (in grace) for a returning subscriber; nobody to forward to now
perSubscriber ??= new Dictionary<IActorRef, List<VirtualTagActor.DependencyValueChanged>>();
foreach (var sub in subscribers)
{