diff --git a/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionActor.cs b/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionActor.cs index 32ab75f7..d3940ff3 100644 --- a/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionActor.cs @@ -61,6 +61,18 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers /// private readonly Dictionary _tagSubscriberCount = new(); + /// + /// Reverse index of which instances subscribe to each tag path: tagPath → set of + /// instanceUniqueName. Mirrors inverted so the + /// hot path fans a value out to exactly the + /// interested instances in O(subscribers) instead of scanning every instance's tag set + /// on every tag update. Maintained via / + /// at every tag-set mutation; a tag key is dropped + /// when its instance set empties. Derived purely from , + /// so it is preserved across reconnect exactly as that map is (see ). + /// + private readonly Dictionary> _instancesByTag = new(); + /// /// Tags whose path resolution failed and are awaiting retry. /// @@ -907,8 +919,11 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers // increments the reference count, so the count stays an accurate "number // of distinct instances subscribed to this tag". if (instanceTags.Add(result.TagPath)) + { _tagSubscriberCount[result.TagPath] = _tagSubscriberCount.GetValueOrDefault(result.TagPath) + 1; + IndexTag(result.TagPath, instanceName); + } // Re-check against current state: another subscribe may have resolved the // same tag while this request's I/O was in flight. @@ -1053,6 +1068,10 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers // Cleanup on Instance Actor stop foreach (var tagPath in tags) { + // Drop this instance from the fan-out reverse index (mirrors the + // _subscriptionsByInstance removal at the end of this method). + UnindexTag(tagPath, request.InstanceUniqueName); + // Drop this instance's reference; the tag is only // released at the adapter when no other instance still subscribes to it. // The reference count makes this O(1) instead of an O(instances) scan. @@ -1585,6 +1604,10 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers _log.Info("[{0}] Re-subscribing {1} tags after reconnect", _connectionName, allTags.Count); var self = Self; + // NOTE: do NOT clear _instancesByTag here. It is the inverse of + // _subscriptionsByInstance, which this method deliberately preserves as the durable + // source of truth across reconnect — clearing the index would silently stop the + // post-reconnect value fan-out for every already-subscribed tag. _subscriptionIds.Clear(); _unresolvedTags.Clear(); _resolutionInFlight.Clear(); @@ -1710,6 +1733,31 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers } } + /// + /// Adds to the reverse index for . + /// Call at every site that adds a tag to an instance's set. + /// + private void IndexTag(string tag, string instance) + { + if (!_instancesByTag.TryGetValue(tag, out var insts)) + { + insts = new HashSet(); + _instancesByTag[tag] = insts; + } + insts.Add(instance); + } + + /// + /// Removes from the reverse index for , + /// dropping the tag key entirely when its instance set empties. Call at every site that + /// removes a tag from an instance's set. + /// + private void UnindexTag(string tag, string instance) + { + if (_instancesByTag.TryGetValue(tag, out var insts) && insts.Remove(instance) && insts.Count == 0) + _instancesByTag.Remove(tag); + } + private void HandleTagValueReceived(TagValueReceived msg) { // Drop values delivered by a disposed adapter. After a @@ -1722,16 +1770,17 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers return; } - // Fan out to all subscribed instances - foreach (var (instanceName, tags) in _subscriptionsByInstance) + // Fan out to exactly the instances subscribed to this tag via the reverse index — + // O(subscribers of this tag) instead of scanning every instance's tag set per update. + if (_instancesByTag.TryGetValue(msg.TagPath, out var interestedInstances)) { - if (!tags.Contains(msg.TagPath)) - continue; - - if (_subscribers.TryGetValue(instanceName, out var subscriber)) + foreach (var instanceName in interestedInstances) { - subscriber.Tell(new TagValueUpdate( - _connectionName, msg.TagPath, msg.Value.Value, msg.Value.Quality, msg.Value.Timestamp)); + if (_subscribers.TryGetValue(instanceName, out var subscriber)) + { + subscriber.Tell(new TagValueUpdate( + _connectionName, msg.TagPath, msg.Value.Value, msg.Value.Quality, msg.Value.Timestamp)); + } } } diff --git a/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/DataConnectionActorTests.cs b/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/DataConnectionActorTests.cs index 0914ed78..6bd888e5 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/DataConnectionActorTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/DataConnectionActorTests.cs @@ -534,6 +534,75 @@ public class DataConnectionActorTests : TestKit c => c.GetMethodInfo().Name == "SubscribeAsync"); } + // ── PLAN-03 Task 17 (P3): tag→instances reverse index for the value fan-out hot path ── + + private IActorRef CreateInstantSubscribeActor(string name) + { + _mockAdapter.ConnectAsync(Arg.Any>(), Arg.Any()) + .Returns(Task.CompletedTask); + _mockAdapter.Status.Returns(ConnectionHealth.Connected); + _mockAdapter.SubscribeAsync(Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(callInfo => Task.FromResult("sub-" + callInfo.ArgAt(0))); + // Failed reads → no seed values are delivered, so the probes see only the ack plus + // whatever TagValueReceived we explicitly deliver. + _mockAdapter.ReadAsync(Arg.Any(), Arg.Any()) + .Returns(new ReadResult(false, null, null)); + return CreateConnectionActor(name); + } + + [Fact] + public async Task Task17_TagFanOut_AfterOneInstanceUnsubscribes_OnlyRemainingSubscriberReceives() + { + // Pins the fan-out contract that the reverse index must preserve: after one of two + // instances sharing a tag unsubscribes, a subsequent value for that tag reaches only + // the remaining subscriber. Passes with the O(N) scan AND the index — the index's + // bookkeeping on unsubscribe is exactly what this guards. + var actor = CreateInstantSubscribeActor("task17-unsub"); + await Task.Delay(300); // reach Connected + var probeA = CreateTestProbe(); + var probeB = CreateTestProbe(); + + actor.Tell(new SubscribeTagsRequest("cA", "inst-A", "task17-unsub", ["shared-tag"], DateTimeOffset.UtcNow), probeA.Ref); + actor.Tell(new SubscribeTagsRequest("cB", "inst-B", "task17-unsub", ["shared-tag"], DateTimeOffset.UtcNow), probeB.Ref); + probeA.ExpectMsg(TimeSpan.FromSeconds(5)); + probeB.ExpectMsg(TimeSpan.FromSeconds(5)); + + // inst-A leaves; the shared tag now has exactly one remaining subscriber. + actor.Tell(new UnsubscribeTagsRequest("cA2", "inst-A", "task17-unsub", DateTimeOffset.UtcNow), probeA.Ref); + await Task.Delay(200); // let the unsubscribe be processed before delivering the value + + actor.Tell(new DataConnectionActor.TagValueReceived( + "shared-tag", new TagValue(42.0, QualityCode.Good, DateTimeOffset.UtcNow), 0)); + + var update = probeB.ExpectMsg(TimeSpan.FromSeconds(5)); + Assert.Equal("shared-tag", update.TagPath); + Assert.Equal(42.0, update.Value); + probeA.ExpectNoMsg(TimeSpan.FromMilliseconds(300)); + } + + [Fact] + public async Task Task17_TagFanOut_TwoInstancesDisjointTags_EachReceivesOnlyItsOwn() + { + // Pins that the index keys values to the right instances: two instances on disjoint + // tags each receive only their own tag's value, never the other's. + var actor = CreateInstantSubscribeActor("task17-disjoint"); + await Task.Delay(300); + var probeA = CreateTestProbe(); + var probeB = CreateTestProbe(); + + actor.Tell(new SubscribeTagsRequest("cA", "inst-A", "task17-disjoint", ["tag-A"], DateTimeOffset.UtcNow), probeA.Ref); + actor.Tell(new SubscribeTagsRequest("cB", "inst-B", "task17-disjoint", ["tag-B"], DateTimeOffset.UtcNow), probeB.Ref); + probeA.ExpectMsg(TimeSpan.FromSeconds(5)); + probeB.ExpectMsg(TimeSpan.FromSeconds(5)); + + actor.Tell(new DataConnectionActor.TagValueReceived( + "tag-A", new TagValue(1.0, QualityCode.Good, DateTimeOffset.UtcNow), 0)); + + var toA = probeA.ExpectMsg(TimeSpan.FromSeconds(5)); + Assert.Equal("tag-A", toA.TagPath); + probeB.ExpectNoMsg(TimeSpan.FromMilliseconds(300)); + } + // ── DataConnectionLayer-001: subscribe must not mutate actor state off-thread ── private static async Task DelayedSubscribeAsync()