perf(dcl): tag->instances reverse index replaces per-update all-instances scan (P3)

src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionActor.cs

@@ -61,6 +61,18 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
     /// </summary>
     private readonly Dictionary<string, int> _tagSubscriberCount = new();
 
+    /// <summary>
+    /// Reverse index of which instances subscribe to each tag path: tagPath → set of
+    /// instanceUniqueName. Mirrors <see cref="_subscriptionsByInstance"/> inverted so the
+    /// <see cref="HandleTagValueReceived"/> 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 <see cref="IndexTag"/>/<see cref="UnindexTag"/>
+    /// at every <see cref="_subscriptionsByInstance"/> tag-set mutation; a tag key is dropped
+    /// when its instance set empties. Derived purely from <see cref="_subscriptionsByInstance"/>,
+    /// so it is preserved across reconnect exactly as that map is (see <see cref="ReSubscribeAll"/>).
+    /// </summary>
+    private readonly Dictionary<string, HashSet<string>> _instancesByTag = new();
+
     /// <summary>
     /// Tags whose path resolution failed and are awaiting retry.
     /// </summary>
@@ -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
         }
     }
 
+    /// <summary>
+    /// Adds <paramref name="instance"/> to the reverse index for <paramref name="tag"/>.
+    /// Call at every site that adds a tag to an instance's <see cref="_subscriptionsByInstance"/> set.
+    /// </summary>
+    private void IndexTag(string tag, string instance)
+    {
+        if (!_instancesByTag.TryGetValue(tag, out var insts))
+        {
+            insts = new HashSet<string>();
+            _instancesByTag[tag] = insts;
+        }
+        insts.Add(instance);
+    }
+
+    /// <summary>
+    /// Removes <paramref name="instance"/> from the reverse index for <paramref name="tag"/>,
+    /// dropping the tag key entirely when its instance set empties. Call at every site that
+    /// removes a tag from an instance's <see cref="_subscriptionsByInstance"/> set.
+    /// </summary>
+    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
+        // Fan out to exactly the instances subscribed to this tag via the reverse index —
-        foreach (var (instanceName, tags) in _subscriptionsByInstance)
+        // 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))
+            foreach (var instanceName in interestedInstances)
-                continue;
-
-            if (_subscribers.TryGetValue(instanceName, out var subscriber))
             {
-                subscriber.Tell(new TagValueUpdate(
+                if (_subscribers.TryGetValue(instanceName, out var subscriber))
-                    _connectionName, msg.TagPath, msg.Value.Value, msg.Value.Quality, msg.Value.Timestamp));
+                {
+                    subscriber.Tell(new TagValueUpdate(
+                        _connectionName, msg.TagPath, msg.Value.Value, msg.Value.Quality, msg.Value.Timestamp));
+                }
             }
         }
 

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<IDictionary<string, string>>(), Arg.Any<CancellationToken>())
+            .Returns(Task.CompletedTask);
+        _mockAdapter.Status.Returns(ConnectionHealth.Connected);
+        _mockAdapter.SubscribeAsync(Arg.Any<string>(), Arg.Any<SubscriptionCallback>(), Arg.Any<CancellationToken>())
+            .Returns(callInfo => Task.FromResult("sub-" + callInfo.ArgAt<string>(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<string>(), Arg.Any<CancellationToken>())
+            .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<SubscribeTagsResponse>(TimeSpan.FromSeconds(5));
+        probeB.ExpectMsg<SubscribeTagsResponse>(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<TagValueUpdate>(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<SubscribeTagsResponse>(TimeSpan.FromSeconds(5));
+        probeB.ExpectMsg<SubscribeTagsResponse>(TimeSpan.FromSeconds(5));
+
+        actor.Tell(new DataConnectionActor.TagValueReceived(
+            "tag-A", new TagValue(1.0, QualityCode.Good, DateTimeOffset.UtcNow), 0));
+
+        var toA = probeA.ExpectMsg<TagValueUpdate>(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<string> DelayedSubscribeAsync()