diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/DependencyMuxActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/DependencyMuxActor.cs
index 3af36d6a..40e9bf19 100644
--- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/DependencyMuxActor.cs
+++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/DependencyMuxActor.cs
@@ -28,53 +28,132 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.VirtualTags;
/// would over-emit if it spanned the cluster.
///
///
-/// Last-value retention + replay (R07:H6). For every ref that has EVER been registered the
-/// mux retains the last (value, status, timestamp) 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
-/// BadWaitingForInitialData until each of its inputs happened to change again (a
-/// scripted-alarm host restart has the same shape). A with
-/// 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.
+/// Last-value retention + replay (R07:H6, bounded in review fix round 1 / I2). The mux
+/// retains the last (value, status, timestamp) of a ref only while at least one subscriber
+/// registered with names it (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
+/// BadWaitingForInitialData until each of its inputs happened to change again. So when a
+/// ref's last replay-opting subscriber leaves, its value is kept for
+/// (default ; a returning replay registration inside it cancels
+/// the release), then dropped together with the ref's entry.
+///
+///
+/// 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.
+///
+///
+/// Source removal. The mux records which driver instance last published each retained ref;
+/// 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 yet —
+/// DriverHostActor (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.
///
///
-public sealed class DependencyMuxActor : ReceiveActor
+public sealed class DependencyMuxActor : ReceiveActor, IWithTimers
{
public const string ActorName = "dependency-mux";
+ /// How long a ref's retained value outlives its last replay-opting subscriber.
+ public static readonly TimeSpan DefaultRetentionGrace = TimeSpan.FromMinutes(2);
+
/// 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.
/// The refs the subscriber wants.
/// The subscriber.
- /// When true, the mux immediately sends the subscriber one
- /// holding the last retained value of every ref
- /// in it has one for (nothing when it has none).
+ /// When true, the mux retains the last value of every ref in
+ /// while this subscriber is registered, and immediately sends the subscriber one
+ /// holding the last retained value of every ref it has
+ /// one for (nothing when it has none).
public sealed record RegisterInterest(IReadOnlyList TagRefs, IActorRef Subscriber, bool ReplayRetained = false);
/// Unregister every interest held by . Sent on PostStop by
/// the subscriber, or by Terminated handling when the mux watches.
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> _byRef = new(StringComparer.Ordinal);
- private readonly Dictionary> _bySubscriber = new();
- private readonly Dictionary _last = new(StringComparer.Ordinal);
+ /// A driver instance was removed / stopped: evict every retained value it produced so it is
+ /// never replayed. Subscriptions are untouched.
+ /// The removed driver instance.
+ public sealed record SourceRemoved(string DriverInstanceId);
+
+ private sealed class SweepReleased
+ {
+ public static readonly SweepReleased Instance = new();
+ }
+
+ private const string SweepTimerKey = "sweep-released";
+
+ /// Per-ref state. One lookup on the hot path answers both "who wants it" and "retain it?".
+ private sealed class RefEntry
+ {
+ public readonly HashSet Subscribers = new();
+
+ /// Registered subscribers of this ref that opted into replay.
+ public int Replayers;
+
+ /// When the last replay-opting subscriber left (retention in grace), or null.
+ 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 Refs, bool Replay);
+
+ private readonly Dictionary _byRef = new(StringComparer.Ordinal);
+ private readonly Dictionary _bySubscriber = new();
private readonly ILoggingAdapter _log = Context.GetLogger();
+ /// How long a released ref's value is kept.
+ public TimeSpan RetentionGrace { get; }
+
+ ///
+ public ITimerScheduler Timers { get; set; } = null!;
+
+ /// Test hook — refs that currently hold a retained value.
+ internal int RetainedRefCount => _byRef.Values.Count(e => e.Last is not null);
+
+ /// Test hook — refs the mux holds any state for (subscribers or a retained value).
+ internal int TrackedRefCount => _byRef.Count;
+
/// Creates props for the DependencyMuxActor.
+ /// How long a ref's value outlives its last replay-opting subscriber;
+ /// null ⇒ .
/// The props for creating this actor.
- public static Props Props() => Akka.Actor.Props.Create();
+ public static Props Props(TimeSpan? retentionGrace = null)
+ {
+ var grace = retentionGrace ?? DefaultRetentionGrace;
+ return Akka.Actor.Props.Create(() => new DependencyMuxActor(grace));
+ }
/// Initializes a new instance of the class.
- public DependencyMuxActor()
+ /// See ; non-positive ⇒ .
+ public DependencyMuxActor(TimeSpan retentionGrace)
{
+ RetentionGrace = retentionGrace > TimeSpan.Zero ? retentionGrace : DefaultRetentionGrace;
Receive(OnRegister);
Receive(msg => RemoveSubscriber(msg.Subscriber));
+ Receive(OnSourceRemoved);
+ Receive(_ => OnSweep());
Receive(OnAttributeValuePublished);
Receive(OnAttributeValueBatchPublished);
Receive(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(msg.TagRefs, StringComparer.Ordinal);
- _bySubscriber[msg.Subscriber] = refs;
+ _bySubscriber[msg.Subscriber] = new Registration(refs, msg.ReplayRetained);
List? replay = null;
foreach (var r in refs)
{
- if (!_byRef.TryGetValue(r, out var set))
+ if (!_byRef.TryGetValue(r, out var entry))
{
- set = new HashSet();
- _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);
}
+ /// Drop 's registration. A ref losing its last replay-opting
+ /// subscriber goes into grace; a ref left with no state at all is removed.
+ 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;
+ }
+
+ /// 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.
+ private void OnSweep()
+ {
+ var cutoff = DateTime.UtcNow - RetentionGrace;
+ var stillInGrace = false;
+ List? 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? 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>();
foreach (var sub in subscribers)
{
diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/VirtualTags/DependencyMuxReplayTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/VirtualTags/DependencyMuxReplayTests.cs
index ae42edb9..f761edd9 100644
--- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/VirtualTags/DependencyMuxReplayTests.cs
+++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/VirtualTags/DependencyMuxReplayTests.cs
@@ -12,10 +12,13 @@ using ZB.MOM.WW.OtOpcUa.Runtime.VirtualTags;
namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.VirtualTags;
///
-/// R07:H6 — the mux retains the last value per ever-registered ref and replays it to a registration
-/// that asks (). A Calculation driver
-/// Restart (new driver + new adapter) therefore republishes from current inputs instead of sitting
-/// BadWaitingForInitialData until every input happens to change.
+/// R07:H6 — the mux retains the last value of every ref a REPLAY-OPTING subscriber
+/// () is registered for, keeps it for a
+/// short grace after the last such subscriber leaves, and replays it to a registration that asks. A
+/// Calculation driver Restart (new driver + new adapter) therefore republishes from current inputs instead
+/// of sitting BadWaitingForInitialData until every input happens to change. Fix round 1 (I2): a
+/// non-opting subscriber (the historization recorder) retains nothing, retention is released after the
+/// grace, and a removed source's values are evicted so its last Good is never replayed.
///
public sealed class DependencyMuxReplayTests : RuntimeActorTestBase
{
@@ -27,7 +30,7 @@ public sealed class DependencyMuxReplayTests : RuntimeActorTestBase
{
var mux = Sys.ActorOf(DependencyMuxActor.Props());
var first = CreateTestProbe();
- mux.Tell(new DependencyMuxActor.RegisterInterest(new[] { "a", "b" }, first.Ref));
+ mux.Tell(new DependencyMuxActor.RegisterInterest(new[] { "a", "b" }, first.Ref, ReplayRetained: true));
mux.Tell(Burst(("a", 1), ("b", 2), ("c", 3)));
first.ExpectMsg(PresenceBudget).Items.Count.ShouldBe(2);
@@ -38,7 +41,7 @@ public sealed class DependencyMuxReplayTests : RuntimeActorTestBase
mux.Tell(new DependencyMuxActor.RegisterInterest(new[] { "a", "c" }, second.Ref, ReplayRetained: true));
var replay = second.ExpectMsg(PresenceBudget);
- replay.Items.Select(i => i.TagId).ShouldBe(new[] { "a" }, "'c' was never registered by anyone, so it is not retained; 'b' is not in this subscriber's refs");
+ replay.Items.Select(i => i.TagId).ShouldBe(new[] { "a" }, "'c' was never registered by a replay-opting subscriber, so it is not retained; 'b' is not in this subscriber's refs");
replay.Items[0].Value.ShouldBe(1);
}
@@ -60,6 +63,86 @@ public sealed class DependencyMuxReplayTests : RuntimeActorTestBase
fresh.ExpectNoMsg(TimeSpan.FromMilliseconds(300));
}
+ /// I2 — a subscriber that did NOT opt into replay (the continuous-historization recorder registers
+ /// every historized ref) retains nothing; retention is bounded to the refs a replay-opting subscriber named.
+ [Fact]
+ public void Only_refs_named_by_a_replay_opting_subscriber_are_retained()
+ {
+ var mux = ActorOfAsTestActorRef(DependencyMuxActor.Props());
+ var recorder = CreateTestProbe();
+ var calc = CreateTestProbe();
+ mux.Tell(new DependencyMuxActor.RegisterInterest(new[] { "h1", "h2", "h3", "shared" }, recorder.Ref));
+ mux.Tell(new DependencyMuxActor.RegisterInterest(new[] { "r", "shared" }, calc.Ref, ReplayRetained: true));
+ mux.Tell(Burst(("h1", 1), ("h2", 2), ("h3", 3), ("r", 4), ("shared", 5)));
+ recorder.ExpectMsg(PresenceBudget).Items.Count.ShouldBe(4, "the recorder is still DELIVERED to");
+ calc.ExpectMsg(PresenceBudget);
+
+ mux.UnderlyingActor.RetainedRefCount.ShouldBe(2, "only 'r' and 'shared' — the recorder-only refs retain nothing");
+
+ var late = CreateTestProbe();
+ mux.Tell(new DependencyMuxActor.RegisterInterest(new[] { "h1", "r" }, late.Ref, ReplayRetained: true));
+ late.ExpectMsg(PresenceBudget)
+ .Items.Select(i => i.TagId).ShouldBe(new[] { "r" }, "positive control: the opted-in ref replays; 'h1' was never retained");
+
+ // Unregistering everyone ends up with no entries at all once the grace is over — tested below; here
+ // the recorder leaving changes nothing about retention.
+ mux.Tell(new DependencyMuxActor.UnregisterInterest(recorder.Ref));
+ mux.UnderlyingActor.RetainedRefCount.ShouldBe(2);
+ }
+
+ /// I2 — when the last replay-opting subscriber of a ref leaves, its retained value survives a short
+ /// grace (the restart window: old adapter gone, new one not yet registered) and is then dropped.
+ [Fact]
+ public void Retention_is_released_after_the_grace_once_no_replay_subscriber_remains()
+ {
+ var mux = ActorOfAsTestActorRef(DependencyMuxActor.Props(TimeSpan.FromMilliseconds(200)));
+ var gen1 = CreateTestProbe();
+ mux.Tell(new DependencyMuxActor.RegisterInterest(new[] { "a" }, gen1.Ref, ReplayRetained: true));
+ mux.Tell(Burst(("a", 1)));
+ gen1.ExpectMsg(PresenceBudget);
+ mux.Tell(new DependencyMuxActor.UnregisterInterest(gen1.Ref));
+
+ // Positive control: inside the grace a returning subscriber still gets the value.
+ var gen2 = CreateTestProbe();
+ mux.Tell(new DependencyMuxActor.RegisterInterest(new[] { "a" }, gen2.Ref, ReplayRetained: true));
+ gen2.ExpectMsg(PresenceBudget).Items.Single().Value.ShouldBe(1);
+ mux.Tell(new DependencyMuxActor.UnregisterInterest(gen2.Ref));
+
+ // Past the grace: nothing retained, no entry left behind, and a returning subscriber replays nothing.
+ AwaitAssert(() =>
+ {
+ mux.UnderlyingActor.RetainedRefCount.ShouldBe(0);
+ mux.UnderlyingActor.TrackedRefCount.ShouldBe(0);
+ }, PresenceBudget);
+ var gen3 = CreateTestProbe();
+ mux.Tell(new DependencyMuxActor.RegisterInterest(new[] { "a" }, gen3.Ref, ReplayRetained: true));
+ gen3.ExpectNoMsg(TimeSpan.FromMilliseconds(300));
+ }
+
+ /// I2 — a removed source's retained values are evicted, so its last Good is never replayed to a
+ /// subscriber that (re)registers after the removal. Values from other sources are untouched.
+ [Fact]
+ public void SourceRemoved_evicts_that_sources_retained_values_only()
+ {
+ var mux = ActorOfAsTestActorRef(DependencyMuxActor.Props());
+ var sub = CreateTestProbe();
+ mux.Tell(new DependencyMuxActor.RegisterInterest(new[] { "a", "b" }, sub.Ref, ReplayRetained: true));
+ mux.Tell(new DriverInstanceActor.AttributeValueBatchPublished("driver-1",
+ [new DriverInstanceActor.AttributeValueItem("a", 1, OpcUaStatus.Good, DateTime.UtcNow)]));
+ mux.Tell(new DriverInstanceActor.AttributeValueBatchPublished("driver-2",
+ [new DriverInstanceActor.AttributeValueItem("b", 2, OpcUaStatus.Good, DateTime.UtcNow)]));
+ sub.ExpectMsg(PresenceBudget);
+ sub.ExpectMsg(PresenceBudget);
+
+ mux.Tell(new DependencyMuxActor.SourceRemoved("driver-1"));
+ mux.UnderlyingActor.RetainedRefCount.ShouldBe(1);
+
+ var restarted = CreateTestProbe();
+ mux.Tell(new DependencyMuxActor.RegisterInterest(new[] { "a", "b" }, restarted.Ref, ReplayRetained: true));
+ restarted.ExpectMsg(PresenceBudget)
+ .Items.Select(i => i.TagId).ShouldBe(new[] { "b" }, "driver-1 was removed: its last Good must not be replayed");
+ }
+
/// The user-facing scenario: a calc tag computes, the driver is restarted (new instance + new
/// adapter) with NO input changing afterwards, and the new instance republishes the current value.
[Fact]