diff --git a/docs/requirements/Component-DataConnectionLayer.md b/docs/requirements/Component-DataConnectionLayer.md index e78f590f..036d064c 100644 --- a/docs/requirements/Component-DataConnectionLayer.md +++ b/docs/requirements/Component-DataConnectionLayer.md @@ -235,7 +235,7 @@ Three subscribe paths, deliberately paced differently: | **Reconnect re-subscribe** (`ReSubscribeAll`) | **Sequential** chunks of `SubscribeBatchSize` with `SubscribeBatchDelay` between them, all inside ONE background task; each chunk reports its per-tag results back to the actor as its own message. Replaces a tight loop that fired one fire-and-forget task per tag. | | **Tag-resolution probe** | The whole not-in-flight unresolved set, chunked at `SubscribeBatchSize`, as one probe round per timer tick. | -Seeding (the initial-value read that keeps a STATIC tag from staying Uncertain) issues chunked bulk reads — `SeedReadBatchSize` per chunk, `SeedReadMaxParallelism` chunks in flight, `SeedReadTimeout` per chunk — under a single `SeedOverallTimeout` deadline covering every chunk and retry round. +Seeding (the initial-value read that keeps a STATIC tag from staying Uncertain) issues chunked bulk reads — `SeedReadBatchSize` per chunk, `SeedReadMaxParallelism` chunks in flight, `SeedReadTimeout` per chunk — under a single `SeedOverallTimeout` deadline covering every chunk and retry round. The reconnect re-seed is additionally bound to its session: entering Reconnecting, an adapter swap or actor stop cancels it, so a flapping link no longer stacks one full-tag-set seed read per flap (review 2026-10-21). **OPC UA subscription sharding.** `RealOpcUaClient` places monitored items on the first subscription shard with free capacity (`MaxMonitoredItemsPerSubscription`), creating a shard when all are full and deleting one once it empties. A batch is N × `AddItem` plus **one** `ApplyChanges` per touched shard. Alarms & Conditions event items are pinned to a **dedicated event shard**, so `ConditionRefresh` has a single subscription id to target and `QueueSize:1000` event items never consume data-shard capacity. Every structural mutation (AddItem / RemoveItem / ApplyChanges / Create / Delete) and every shard-list mutation serializes behind ONE per-client gate — the SDK `Subscription` is not safe under concurrent structural mutation, and the subscribe task, resolution probe and alarm subscribe are otherwise unordered. @@ -446,7 +446,7 @@ The DCL reports the following metrics to the Health Monitoring component via the - **Tag resolution counts**: Per connection, the number of total subscribed tags vs. successfully resolved tags. This gives operators visibility into misconfigured templates without needing to open the debug view for individual instances. Both numbers are **derived from the connection actor's authoritative per-tag state**, never accumulated in counters: the total is the number of distinct tag paths at least one instance currently subscribes to (the per-tag counted set that also drives value fan-out and the last-subscriber release decision), and the resolved count is the number of tags for which the adapter currently holds a subscription handle. A tag counts toward the total from the moment an instance registers it, whatever its resolution outcome — resolved, awaiting a resolution retry, or failed at connection level — so resolved can never exceed total. Deriving rather than accumulating is deliberate: increment/decrement counters drifted whenever an unsubscribe landed inside a reconnect window (the reconnect clears the per-tag maps the decrements keyed off), leaking a phantom tag into the reported total on every churn cycle. The pair is **pushed to the health collector, not pulled from it**, so it is a cached value that is only as fresh as its last push — and the collector holds it until the next one. Every site that mutates either underlying set must therefore push: entering the Connected state (a legitimate 0/0 — no instance has subscribed yet), the first-subscribe apply, the batch apply (reconnect re-subscribe and resolution probe), and unsubscribe. Missing the first-subscribe push is what made a *perfectly healthy* connection report 0/0 indefinitely — the connect-time 0/0 was never superseded, because a connection whose tags all resolve on the first attempt never reaches the batch path and has its resolution-retry timer cancelled outright for having nothing unresolved. That is the reverse of the usual staleness risk, and it blinded the diagnostic exactly when it should have read clean (deferred-work register row 55; the actor's own `GetHealthReport` reply, which really is derived on demand, was correct throughout). -- **Tag quality counters** are pushed on a genuine quality **transition** only, coalesced onto a `QualityFlushInterval` (1s) single-shot timer. Counter arithmetic still runs per message; only the collector push is deferred, and a value whose quality is unchanged moves no counter at all. Health reports poll at 30s, so the coalescing loses nothing. Three paths flush **synchronously** because their correctness depends on it: the bad-quality push on disconnect, unsubscribe, and the reconnect counter reset. +- **Tag quality counters** are pushed on a genuine quality **transition** only, coalesced onto a `QualityFlushInterval` (1s) single-shot timer. Counter arithmetic still runs per message; only the collector push is deferred, and a value whose quality is unchanged moves no counter at all. Health reports poll at 30s, so the coalescing loses nothing. Three paths flush **synchronously** because their correctness depends on it: the bad-quality push on disconnect, unsubscribe, and the reconnect counter reset. A value for a tag no instance subscribes to any more (a callback queued before the device unadvise landed) is dropped before the quality bookkeeping, so it can neither re-insert the tag's last-quality entry nor count a phantom bucket above the subscribed total (review 2026-10-21). ## Dependencies diff --git a/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionActor.cs b/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionActor.cs index d0a732c1..0537c948 100644 --- a/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionActor.cs @@ -277,6 +277,15 @@ public class DataConnectionActor : UntypedActor, IWithTimers /// private bool _disconnectDuringConnectAttempt; + /// + /// Cancels the reconnect re-seed started by 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 + /// against a device that was already + /// struggling — work whose values the stale-generation guard then dropped anyway. + /// + private CancellationTokenSource? _sessionWorkCts; + /// /// Initializes the data connection actor with its adapter and configuration. /// @@ -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 /// private void SwapAdapter(IDictionary newConfig) { + CancelSessionWork(); UnhookAdapterDisconnected(); _ = _adapter.DisposeAsync().AsTask(); @@ -710,6 +723,16 @@ public class DataConnectionActor : UntypedActor, IWithTimers HookAdapterDisconnected(); } + /// + /// 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. + /// + private void CancelSessionWork() + { + _sessionWorkCts?.Cancel(); + _sessionWorkCts = null; + } + /// /// Advances for a new SESSION on the SAME adapter — a plain /// drop + reconnect, where 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 /// and returns the seeds — all actor-state mutation/delivery stays on the actor thread. /// - private async Task> SeedTagsAsync(IDataConnection adapter, IReadOnlyCollection tags) + private async Task> SeedTagsAsync( + IDataConnection adapter, IReadOnlyCollection tags, CancellationToken sessionToken = default) { var seedValues = new List(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)); } } diff --git a/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Actors/DataConnectionHousekeepingTests.cs b/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Actors/DataConnectionHousekeepingTests.cs new file mode 100644 index 00000000..8b862cb9 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Actors/DataConnectionHousekeepingTests.cs @@ -0,0 +1,107 @@ +using Akka.Actor; +using Akka.TestKit.Xunit2; +using NSubstitute; +using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Protocol; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.DataConnection; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; +using ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Actors; +using ZB.MOM.WW.ScadaBridge.HealthMonitoring; + +namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests.Actors; + +/// +/// Low-severity housekeeping findings from review 2026-10-21 (02/L2, 02/L4): bookkeeping that +/// outlived the subscription or the session it belonged to. +/// +[Collection("DataConnectionManagerActor")] +public class DataConnectionHousekeepingTests : TestKit +{ + private const string Tag = "sensor/temp"; + + private readonly ISiteHealthCollector _health = Substitute.For(); + private readonly IDataConnectionFactory _factory = Substitute.For(); + + private static DataConnectionOptions Options() => new() + { + ReconnectInterval = TimeSpan.FromMilliseconds(100), + TagResolutionRetryInterval = TimeSpan.FromSeconds(30), + WriteTimeout = TimeSpan.FromSeconds(5), + SeedReadMaxAttempts = 1, + SeedReadRetryDelay = TimeSpan.FromMilliseconds(10), + SeedOverallTimeout = TimeSpan.FromSeconds(60), + SeedReadTimeout = TimeSpan.FromSeconds(60), + QualityFlushInterval = TimeSpan.FromMilliseconds(100) + }; + + private IActorRef CreateActor(FakeBatchDataConnection adapter, string name) => + Sys.ActorOf(Props.Create(() => new DataConnectionActor( + name, adapter, Options(), _health, _factory, "OpcUa")), name); + + private int[]? LastQualityPush() => + _health.ReceivedCalls() + .Where(c => c.GetMethodInfo().Name == nameof(ISiteHealthCollector.UpdateTagQuality)) + .Select(c => c.GetArguments()) + .Select(a => new[] { (int)a[1]!, (int)a[2]!, (int)a[3]! }) + .LastOrDefault(); + + private int ConnectedTransitions(string name) => + _health.ReceivedCalls().Count(c => + c.GetMethodInfo().Name == nameof(ISiteHealthCollector.UpdateConnectionHealth) && + c.GetArguments() is [string n, ConnectionHealth h, ..] && n == name && h == ConnectionHealth.Connected); + + /// + /// 02/L2: a value for a tag nobody subscribes to any more (delivered after the last + /// unsubscribe, before the device unadvise lands) must not re-enter the quality bookkeeping — + /// that leaked a _lastTagQuality entry and a phantom Good count above the subscribed + /// total until the next reconnect. + /// + [Fact] + public void ValueForUnsubscribedTag_DoesNotReEnterQualityBookkeeping() + { + const string name = "late-value-after-unsubscribe"; + var adapter = new FakeBatchDataConnection(); + var actor = CreateActor(adapter, name); + + actor.Tell(new SubscribeTagsRequest("c1", "inst1", name, [Tag], DateTimeOffset.UtcNow)); + ExpectMsg(m => m.Success, TimeSpan.FromSeconds(5)); + AwaitCondition(() => LastQualityPush() is [1, 0, 0], TimeSpan.FromSeconds(5)); + + actor.Tell(new UnsubscribeTagsRequest("c2", "inst1", name, DateTimeOffset.UtcNow)); + AwaitCondition(() => !adapter.UnsubscribeBatches.IsEmpty, TimeSpan.FromSeconds(5)); + AwaitCondition(() => LastQualityPush() is [0, 0, 0], TimeSpan.FromSeconds(5)); + + // The late callback. + adapter.ValueCallback!(Tag, new TagValue(7, QualityCode.Good, DateTimeOffset.UtcNow)); + Thread.Sleep(500); // > QualityFlushInterval: a dirty counter would have been pushed + + Assert.Equal([0, 0, 0], LastQualityPush()!); + } + + /// + /// 02/L4: the reconnect re-seed is bound to its session. A drop while it is still reading must + /// cancel it instead of leaving a full-tag-set read running against a struggling device for + /// up to SeedOverallTimeout — one more per flap. + /// + [Fact] + public void ReconnectReseed_IsCancelled_WhenTheSessionDropsAgain() + { + const string name = "reseed-cancelled-on-drop"; + var adapter = new FakeBatchDataConnection(); + var actor = CreateActor(adapter, name); + + actor.Tell(new SubscribeTagsRequest("c1", "inst1", name, [Tag], DateTimeOffset.UtcNow)); + ExpectMsg(m => m.Success, TimeSpan.FromSeconds(5)); + + adapter.HangReads = true; + var readsBefore = adapter.ReadTokens.Count; + adapter.RaiseDisconnected(); + AwaitCondition(() => ConnectedTransitions(name) >= 2, TimeSpan.FromSeconds(5)); + AwaitCondition(() => adapter.ReadTokens.Count > readsBefore, TimeSpan.FromSeconds(5)); + var reseedToken = adapter.ReadTokens.ElementAt(readsBefore); + Assert.False(reseedToken.IsCancellationRequested); + + adapter.RaiseDisconnected(); + + AwaitCondition(() => reseedToken.IsCancellationRequested, TimeSpan.FromSeconds(5)); + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Actors/FakeBatchDataConnection.cs b/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Actors/FakeBatchDataConnection.cs index 8de09b1e..504699d2 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Actors/FakeBatchDataConnection.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Actors/FakeBatchDataConnection.cs @@ -22,6 +22,8 @@ public sealed class FakeBatchDataConnection public readonly ConcurrentQueue> UnsubscribeBatches = new(); /// Tag lists handed to , one entry per call. public readonly ConcurrentQueue> ReadBatches = new(); + /// The cancellation token handed to each call, in order. + public readonly ConcurrentQueue ReadTokens = new(); /// Count of SINGLE-tag subscribe calls; must stay 0 on a batch-capable adapter. public int SingleSubscribeCalls; /// Count of SINGLE-tag read calls; must stay 0 on a batch-capable adapter. @@ -143,6 +145,7 @@ public sealed class FakeBatchDataConnection { var tags = tagPaths.ToList(); ReadBatches.Enqueue(tags); + ReadTokens.Enqueue(cancellationToken); if (HangReads) await Task.Delay(Timeout.Infinite, cancellationToken);