fix(dcl): drop values for unsubscribed tags before quality bookkeeping; cancel the reconnect re-seed when its session ends (review 2026-10-21 02/L2, 02/L4)

2026-10-22 06:26:06 -04:00
parent 8c1bd2c3fc
commit d51ea5136a
4 changed files with 163 additions and 13 deletions
@@ -277,6 +277,15 @@ public class DataConnectionActor : UntypedActor, IWithTimers
    /// </summary>
    private bool _disconnectDuringConnectAttempt;

+    /// <summary>
+    /// Cancels the reconnect re-seed started by <see cref="ReSubscribeAll"/> when the session it
+    /// belongs to ends (review 2026-10-21 02/L4). Without it a flapping link stacked one
+    /// full-tag-set seed read per flap, each alive for up to
+    /// <see cref="DataConnectionOptions.SeedOverallTimeout"/> against a device that was already
+    /// struggling — work whose values the stale-generation guard then dropped anyway.
+    /// </summary>
+    private CancellationTokenSource? _sessionWorkCts;
+
    /// <summary>
    /// Initializes the data connection actor with its adapter and configuration.
    /// </summary>
@@ -362,6 +371,7 @@ public class DataConnectionActor : UntypedActor, IWithTimers
    protected override void PostStop()
    {
        _log.Info("DataConnectionActor [{0}] stopping — disposing adapter", _connectionName);
+        CancelSessionWork();
        UnhookAdapterDisconnected();
        // S9: fire-and-forget, but observe a faulted dispose so a failing adapter
        // teardown is logged rather than swallowed into an unobserved task exception.
@@ -606,6 +616,8 @@ public class DataConnectionActor : UntypedActor, IWithTimers
    {
        _log.Warning("[{0}] Entering Reconnecting state{1}", _connectionName,
            afterReconfigure ? " (reconfigured in place)" : string.Empty);
+        // The session is over: abandon its re-seed (02/L4).
+        CancelSessionWork();
        _healthCollector.UpdateConnectionHealth(_connectionName, ConnectionHealth.Disconnected);
        _healthCollector.UpdateConnectionEndpoint(_connectionName, "Disconnected");
        // Register row 54 — drops this connection out of the connected gauge while the
@@ -697,6 +709,7 @@ public class DataConnectionActor : UntypedActor, IWithTimers
    /// </summary>
    private void SwapAdapter(IDictionary<string, string> newConfig)
    {
+        CancelSessionWork();
        UnhookAdapterDisconnected();
        _ = _adapter.DisposeAsync().AsTask();

@@ -710,6 +723,16 @@ public class DataConnectionActor : UntypedActor, IWithTimers
        HookAdapterDisconnected();
    }

+    /// <summary>
+    /// Cancels background work bound to the current session (the reconnect re-seed). Cancel
+    /// only — never Dispose — because the background task may still be linking to the token.
+    /// </summary>
+    private void CancelSessionWork()
+    {
+        _sessionWorkCts?.Cancel();
+        _sessionWorkCts = null;
+    }
+
    /// <summary>
    /// Advances <see cref="_adapterGeneration"/> for a new SESSION on the SAME adapter — a plain
    /// drop + reconnect, where <see cref="SwapAdapter"/> never runs (review 2026-10-21 02/C1).
@@ -1497,7 +1520,8 @@ public class DataConnectionActor : UntypedActor, IWithTimers
    /// Runs on a background task: it reads only the supplied <paramref name="adapter"/>
    /// and returns the seeds — all actor-state mutation/delivery stays on the actor thread.
    /// </summary>
-    private async Task<List<SeededValue>> SeedTagsAsync(IDataConnection adapter, IReadOnlyCollection<string> tags)
+    private async Task<List<SeededValue>> SeedTagsAsync(
+        IDataConnection adapter, IReadOnlyCollection<string> tags, CancellationToken sessionToken = default)
    {
        var seedValues = new List<SeededValue>(tags.Count);
        if (tags.Count == 0)
@@ -1509,8 +1533,10 @@ public class DataConnectionActor : UntypedActor, IWithTimers
        var parallelism = Math.Max(1, _options.SeedReadMaxParallelism);
        var useBatch = adapter is IBatchSubscribableConnection;

-        // ONE deadline across every chunk and every retry round.
-        using var overall = new CancellationTokenSource(_options.SeedOverallTimeout);
+        // ONE deadline across every chunk and every retry round, also cut short when the
+        // session this seed belongs to ends (02/L4).
+        using var overall = CancellationTokenSource.CreateLinkedTokenSource(sessionToken);
+        overall.CancelAfter(_options.SeedOverallTimeout);
        var deadlineHit = false;

        for (var attempt = 1; attempt <= attempts && pending.Count > 0 && !overall.IsCancellationRequested; attempt++)
@@ -1561,6 +1587,12 @@ public class DataConnectionActor : UntypedActor, IWithTimers
            }
        }

+        if (sessionToken.IsCancellationRequested)
+        {
+            _log.Debug("[{0}] Seed read abandoned — the session it belonged to has ended", _connectionName);
+            return seedValues;
+        }
+
        if (pending.Count > 0)
        {
            const int maxNamed = 20;
@@ -2834,11 +2866,14 @@ public class DataConnectionActor : UntypedActor, IWithTimers
        // repopulated for the new adapter generation. Seeds delivered after registration is
        // already established, so the ordering hazard does not apply here.
        var reseedAdapter = _adapter;
+        CancelSessionWork();
+        _sessionWorkCts = new CancellationTokenSource();
+        var reseedToken = _sessionWorkCts.Token;
        Task.Run(async () =>
        {
            try
            {
-                var seeds = await SeedTagsAsync(reseedAdapter, allTags);
+                var seeds = await SeedTagsAsync(reseedAdapter, allTags, reseedToken);
                foreach (var seed in seeds)
                    self.Tell(new TagValueReceived(seed.TagPath, seed.Value, generation));
            }
@@ -2903,17 +2938,22 @@ public class DataConnectionActor : UntypedActor, IWithTimers
            return;
        }

+        // A tag nobody subscribes to any more (a callback already queued, or raised before
+        // the device unadvise landed, after the last unsubscribe) is dropped WHOLE — including
+        // the quality bookkeeping below, which would otherwise re-insert the tag into
+        // _lastTagQuality and count a phantom bucket above the subscribed total until the
+        // next reconnect (review 2026-10-21 02/L2).
+        if (!_instancesByTag.TryGetValue(msg.TagPath, out var interestedInstances))
+            return;
+
        // 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))
+        foreach (var instanceName in interestedInstances)
        {
-            foreach (var instanceName in interestedInstances)
+            if (_subscribers.TryGetValue(instanceName, out var subscriber))
            {
-                if (_subscribers.TryGetValue(instanceName, out var subscriber))
-                {
-                    subscriber.Tell(new TagValueUpdate(
-                        _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));
            }
        }