diff --git a/docs/requirements/Component-DataConnectionLayer.md b/docs/requirements/Component-DataConnectionLayer.md index 9750fd15..e78f590f 100644 --- a/docs/requirements/Component-DataConnectionLayer.md +++ b/docs/requirements/Component-DataConnectionLayer.md @@ -401,6 +401,8 @@ The DCL manages connection lifecycle automatically: 3. **Connection state transitions**: The DCL tracks each connection's state as `connected`, `disconnected`, or `reconnecting`. All transitions are logged to Site Event Logging. 4. **Transparent re-subscribe**: On successful reconnection, the DCL automatically re-establishes all previously active subscriptions for that connection. Instance Actors require no action — they simply see quality return to `good` as fresh values arrive from restored subscriptions. 5. **Clean client replacement on (re)connect**: On every (re)connect the adapter detaches and disposes the previous protocol client (session, keep-alive timers, event loop) before creating a new one — a stale client can neither leak nor signal a disconnect against the new session. +6. **Session generation on same-adapter reconnect** (review 2026-10-21, finding C1): a plain drop + reconnect reuses the adapter object, so `SwapAdapter`'s generation bump never runs. On a successful reconnect the actor therefore advances `_adapterGeneration` itself — **before** `ReSubscribeAll` — and re-hooks the adapter's `Disconnected` handler for the new generation (new handler hooked before the old one is removed, so a disconnect in between is never lost). Every subscribe/alarm completion, value callback and seed started against the dead session then takes the existing stale path. That path releases a stale completion's handles against the adapter that created them, **except** when that adapter is still the current one: the handles died with the dead session, and releasing by handle against the adapter's NEW session could tear down a live item. Before the fix a dead-session initial-subscribe completion (in flight up to `SeedOverallTimeout`) was applied against the new session — parking a dead handle that made the live re-subscribe row look redundant and get released (the tag read resolved and stayed silent), or releasing a handle number the new MxGateway session had reused, and a connection-level failure row in it knocked the fresh link back into Reconnecting. +7. **Session-scoped MxGateway subscription ids** (findings C1 + L1): gateway item handles are plain ints that restart per session, so `MxGatewayDataConnection` hands out ids of the form `"{session}:{handle}"` and only ever sends a handle to the gateway when its session is the live one — a dead session's id is dropped silently instead of unadvising whatever the new session gave that number. The adapter's routing maps are cleared on every `ConnectAsync`, `DisconnectAsync` and `DisposeAsync` (they previously kept every dead session's entries, growing per flap), and a subscribe completing after a reconnect does not repopulate them. An alarm unsubscribe for an id the adapter does not know no longer decrements the shared alarm stream's reference count. ### Disconnect Detection Pattern diff --git a/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionActor.cs b/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionActor.cs index 2d32ab3f..d0a732c1 100644 --- a/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionActor.cs @@ -710,6 +710,38 @@ public class DataConnectionActor : UntypedActor, IWithTimers HookAdapterDisconnected(); } + /// + /// Advances for a new SESSION on the SAME adapter — a plain + /// drop + reconnect, where never runs (review 2026-10-21 02/C1). + /// + /// + /// Without it, an initial subscribe issued against the dead session (in flight for up to + /// ) carried the current generation and + /// was applied against the new session: landing before the re-subscribe row for its tag it + /// parked a dead handle under the tag's key, so the LIVE row looked redundant and was + /// released (the tag reads resolved and is silent until the next reconnect); landing after, + /// its dead handle was released against the adapter whose new session may have reused that + /// handle number for another tag (MxGateway); and a connection-level failure row in it + /// knocked the freshly reconnected link back into Reconnecting. + /// + /// + /// + /// The Disconnected handler stamps the generation it was hooked with, so it is + /// re-hooked for the new generation. The new handler is hooked BEFORE the old one is + /// removed: a disconnect raised in between is then delivered twice (old stamp dropped as + /// stale, new stamp acted on) rather than lost, which would wedge the actor Connected on a + /// dead session. + /// + /// + private void AdvanceSessionGeneration() + { + var previousHandler = _disconnectedHandler; + _adapterGeneration++; + HookAdapterDisconnected(); + if (previousHandler is not null) + _adapter.Disconnected -= previousHandler; + } + /// /// Applies a new configuration to this LIVE connection in place (deferred register /// row 97) — the non-destructive alternative to stopping and re-creating the connection @@ -1076,6 +1108,12 @@ public class DataConnectionActor : UntypedActor, IWithTimers $"Connection restored on {_activeEndpoint} endpoint", null); } + // The adapter object survived the drop but its SESSION did not. Advance the + // generation BEFORE re-subscribing so every completion, value and alarm callback + // started against the dead session takes the existing stale path (review + // 2026-10-21 02/C1) — exactly as it would after a swap. + AdvanceSessionGeneration(); + // Transparent re-subscribe — re-establish all active subscriptions ReSubscribeAll(); // Re-establish native alarm feeds (source replays a snapshot). @@ -1405,8 +1443,19 @@ public class DataConnectionActor : UntypedActor, IWithTimers /// The outgoing adapter is disposed fire-and-forget by and may /// already be gone, so a failure here is expected and logged at Debug rather than surfaced. /// - private void ReleaseStaleAgainstOwnAdapter(string description, Func release) + private void ReleaseStaleAgainstOwnAdapter(object owner, string description, Func release) { + // Same-adapter reconnect (review 2026-10-21 02/C1): the generation advanced because the + // SESSION changed, not the adapter. The dead session's handles died with it, and the + // adapter now speaks for the NEW session — where a session-scoped handle number (an + // MxGateway int) may already belong to a live item. Releasing would tear that down. + if (ReferenceEquals(owner, _adapter)) + { + _log.Debug("[{0}] Not releasing {1}: it belongs to a session of the current adapter that " + + "has since ended, so it died with that session", _connectionName, description); + return; + } + try { release().ContinueWith( @@ -1618,6 +1667,7 @@ public class DataConnectionActor : UntypedActor, IWithTimers // MxGateway's are plain ints that collide across sessions, so releasing by handle // against the replacement adapter can tear down a live subscription. ReleaseStaleAgainstOwnAdapter( + msg.Adapter, $"{staleIds.Count} tag subscription handle(s) from adapter generation {msg.AdapterGeneration}", () => UnsubscribeIdsAsync(msg.Adapter, staleIds)); @@ -3078,6 +3128,7 @@ public class DataConnectionActor : UntypedActor, IWithTimers // one (see ReleaseStaleAgainstOwnAdapter). if (msg is { Success: true, SubscriptionId: not null }) ReleaseStaleAgainstOwnAdapter( + msg.Adapter, $"the alarm feed for {msg.SourceReference} from adapter generation {msg.AdapterGeneration}", () => msg.Adapter.UnsubscribeAlarmsAsync(msg.SubscriptionId)); diff --git a/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Adapters/MxGatewayDataConnection.cs b/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Adapters/MxGatewayDataConnection.cs index 3365166c..76811576 100644 --- a/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Adapters/MxGatewayDataConnection.cs +++ b/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Adapters/MxGatewayDataConnection.cs @@ -52,9 +52,16 @@ public class MxGatewayDataConnection // subscriptionId → (tagPath, callback) so the event loop can route updates by tag, // plus tagPath → subscriptionId for reverse lookup. Concurrent because the event // loop reads from a background thread while Subscribe/Unsubscribe mutate. + // Keys are SESSION-SCOPED ids ("{session}:{handle}") — see ScopeId. private readonly ConcurrentDictionary _subs = new(); private readonly ConcurrentDictionary _tagToSub = new(); + // Incremented by every ConnectAsync (each opens a NEW gateway session). Subscription ids + // handed out carry the session they were minted in, because gateway item handles are plain + // ints that restart — i.e. collide — per session (review 2026-10-21 02/C1): a dead + // session's id must never parse to a live handle and unadvise someone else's item. + private int _session; + // DataConnectionLayer mirror of OpcUaDataConnection's once-only guard: an int toggled // with Interlocked.Exchange so only the first caller raises Disconnected. // 0 = not fired, 1 = fired. Reset on (re)connect. @@ -158,6 +165,12 @@ public class MxGatewayDataConnection _alarmStreamPrefix = string.Empty; } + // New session: the previous session's items died with it. Drop their routing entries + // (review 2026-10-21 02/L1 — they were never removed and grew per flap) and advance the + // session so their ids are recognised as dead from now on. + Interlocked.Increment(ref _session); + ClearSubscriptionMaps(); + Interlocked.Exchange(ref _disconnectFired, 0); // reset guard on (re)connect, like OPC UA _client = _clientFactory.Create(); @@ -240,10 +253,46 @@ public class MxGatewayDataConnection } } + /// Drops every subscription routing entry (the session they belong to is gone). + private void ClearSubscriptionMaps() + { + _subs.Clear(); + _tagToSub.Clear(); + } + + /// Formats a gateway item handle as a subscription id scoped to . + private static string ScopeId(int session, string handle) => $"{session}:{handle}"; + + /// + /// Extracts the raw gateway handle from a session-scoped id, or returns null when the + /// id was minted by a session other than the live one (or is not one of ours) — such a handle + /// must never be sent to the gateway. + /// + private string? CurrentSessionHandle(string subscriptionId) + { + var sep = subscriptionId.IndexOf(':'); + if (sep <= 0 + || !int.TryParse(subscriptionId.AsSpan(0, sep), System.Globalization.NumberStyles.Integer, + System.Globalization.CultureInfo.InvariantCulture, out var session) + || session != Volatile.Read(ref _session)) + return null; + return subscriptionId[(sep + 1)..]; + } + + /// Removes an id's routing entries; returns true if it was a known subscription. + private bool ForgetSubscription(string subscriptionId) + { + if (!_subs.TryRemove(subscriptionId, out var s)) + return false; + ((ICollection>)_tagToSub).Remove(new(s.TagPath, subscriptionId)); + return true; + } + /// public async Task DisconnectAsync(CancellationToken cancellationToken = default) { _eventLoopCts?.Cancel(); + ClearSubscriptionMaps(); lock (_alarmLock) { _alarmCts?.Cancel(); @@ -261,18 +310,26 @@ public class MxGatewayDataConnection /// public async Task SubscribeAsync(string tagPath, SubscriptionCallback callback, CancellationToken cancellationToken = default) { - var subId = await _client!.SubscribeAsync(tagPath, cancellationToken); - _subs[subId] = (tagPath, callback); - _tagToSub[tagPath] = subId; + // Capture the session with the client it belongs to: a reconnect landing while this + // call is in flight must not route the dead session's handle into the new maps. + var session = Volatile.Read(ref _session); + var handle = await _client!.SubscribeAsync(tagPath, cancellationToken); + var subId = ScopeId(session, handle); + if (session == Volatile.Read(ref _session)) + { + _subs[subId] = (tagPath, callback); + _tagToSub[tagPath] = subId; + } return subId; } /// public async Task UnsubscribeAsync(string subscriptionId, CancellationToken cancellationToken = default) { - if (_subs.TryRemove(subscriptionId, out var s)) - ((ICollection>)_tagToSub).Remove(new(s.TagPath, subscriptionId)); - await _client!.UnsubscribeAsync(subscriptionId, cancellationToken); + ForgetSubscription(subscriptionId); + // A dead session's handle is never sent: the gateway may have reused the number. + if (CurrentSessionHandle(subscriptionId) is { } handle) + await _client!.UnsubscribeAsync(handle, cancellationToken); } /// @@ -281,18 +338,25 @@ public class MxGatewayDataConnection { // ONE gateway round trip per batch (SubscribeBulk = AddItem + Advise per tag in a // single worker pass) instead of the historical 2 RPCs per tag. + var session = Volatile.Read(ref _session); var outcomes = await _client!.SubscribeBulkAsync(tagPaths, cancellationToken); + var current = session == Volatile.Read(ref _session); var results = new List(outcomes.Count); foreach (var outcome in outcomes) { + string? subId = null; if (outcome is { Success: true, SubscriptionId: not null }) { - _subs[outcome.SubscriptionId] = (outcome.TagPath, callback); - _tagToSub[outcome.TagPath] = outcome.SubscriptionId; + subId = ScopeId(session, outcome.SubscriptionId); + if (current) + { + _subs[subId] = (outcome.TagPath, callback); + _tagToSub[outcome.TagPath] = subId; + } } results.Add(new TagSubscribeResult( - outcome.TagPath, outcome.Success, outcome.SubscriptionId, outcome.Error)); + outcome.TagPath, outcome.Success, subId, outcome.Error)); } return results; @@ -302,13 +366,17 @@ public class MxGatewayDataConnection public async Task UnsubscribeBatchAsync( IReadOnlyList subscriptionIds, CancellationToken cancellationToken = default) { + var handles = new List(subscriptionIds.Count); foreach (var id in subscriptionIds) { - if (_subs.TryRemove(id, out var s)) - ((ICollection>)_tagToSub).Remove(new(s.TagPath, id)); + ForgetSubscription(id); + // A dead session's handle is never sent: the gateway may have reused the number. + if (CurrentSessionHandle(id) is { } handle) + handles.Add(handle); } - await _client!.UnsubscribeBulkAsync(subscriptionIds, cancellationToken); + if (handles.Count > 0) + await _client!.UnsubscribeBulkAsync(handles, cancellationToken); } /// @@ -371,7 +439,11 @@ public class MxGatewayDataConnection { lock (_alarmLock) { - _alarmSubSources.Remove(subscriptionId); + // An id this adapter does not know (a dead session's — ConnectAsync resets the + // registry) must not decrement the live count: reaching zero tears down the shared + // stream every live subscriber depends on. + if (!_alarmSubSources.Remove(subscriptionId)) + return Task.CompletedTask; if (_alarmSubCount > 0) _alarmSubCount--; // Deliberately no stream restart here: dropping a source can only leave the @@ -444,6 +516,7 @@ public class MxGatewayDataConnection public async ValueTask DisposeAsync() { _eventLoopCts?.Cancel(); + ClearSubscriptionMaps(); // The DataConnectionActor disposes adapters // fire-and-forget on failover/stop without necessarily calling // DisconnectAsync first, so tear down the alarm stream here too — otherwise diff --git a/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Actors/DataConnectionSameAdapterReconnectTests.cs b/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Actors/DataConnectionSameAdapterReconnectTests.cs new file mode 100644 index 00000000..84808ada --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Actors/DataConnectionSameAdapterReconnectTests.cs @@ -0,0 +1,296 @@ +using System.Collections.Concurrent; +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; + +/// +/// The SAME-adapter twin of (review 2026-10-21 +/// 02/C1). A plain drop + reconnect keeps the adapter object — only the session underneath it +/// changes — so before the fix the adapter generation did not advance and an initial-subscribe +/// completion issued against the DEAD session passed every stale guard and was applied against the +/// new one: +/// +/// (a) landing before the reconnect re-subscribe row for its tag, it parked the dead handle +/// under the tag's key, so the LIVE row then looked redundant and its handle was released — the +/// tag reads resolved and never delivers again; +/// (b) landing after it, its dead handle was released against the same adapter — and an +/// MxGateway handle is a session-scoped int the new session may already have handed to a +/// different tag; +/// a connection-level failure row in it knocked the freshly reconnected link back into +/// Reconnecting. +/// +/// +[Collection("DataConnectionManagerActor")] +public class DataConnectionSameAdapterReconnectTests : TestKit +{ + private const string Tag = "sensor/temp"; + private const string OtherTag = "sensor/pressure"; + + private readonly ISiteHealthCollector _health = Substitute.For(); + private readonly IDataConnectionFactory _factory = Substitute.For(); + + private readonly DataConnectionOptions _options = new() + { + ReconnectInterval = TimeSpan.FromMilliseconds(100), + TagResolutionRetryInterval = TimeSpan.FromSeconds(30), + WriteTimeout = TimeSpan.FromSeconds(5), + SeedReadMaxAttempts = 1, + SeedReadRetryDelay = TimeSpan.FromMilliseconds(10), + SubscribeBatchDelay = TimeSpan.Zero + }; + + private IActorRef CreateConnectionActor(string name, IDataConnection adapter) + { + var config = new Dictionary { ["Endpoint"] = "opc.tcp://plc:4840" }; + return Sys.ActorOf(Props.Create(() => + new DataConnectionActor( + name, adapter, _options, _health, _factory, "OpcUa", config, null, 3)), name); + } + + private int HealthTransitions(string connectionName, ConnectionHealth health) => + _health.ReceivedCalls().Count(c => + c.GetMethodInfo().Name == nameof(ISiteHealthCollector.UpdateConnectionHealth) && + c.GetArguments() is [string name, ConnectionHealth reported, ..] && + name == connectionName && reported == health); + + private void AwaitConnectedCount(string name, int count) => + AwaitCondition(() => HealthTransitions(name, ConnectionHealth.Connected) >= count, + TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(10)); + + private void AwaitResolved(IActorRef actor, int resolved) => + AwaitCondition(() => actor.Ask( + new DataConnectionActor.GetHealthReport(), TimeSpan.FromSeconds(1)).Result.ResolvedTags == resolved, + TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(20)); + + /// Drops the session and waits for the same adapter to reconnect. + private void DropAndReconnect(string name, GatedAdapter adapter, int connectedCountAfter) + { + adapter.RaiseDisconnected(); + AwaitConnectedCount(name, connectedCountAfter); + } + + /// + /// (a) The stale completion lands BEFORE the re-subscribe row for its tag. The live handle + /// must never be released, and the registry must still hold it afterwards. + /// + [Fact] + public void StaleCompletion_BeforeResubscribeRow_DoesNotReleaseTheLiveHandle() + { + const string name = "same-adapter-a"; + var adapter = new GatedAdapter(); + var actor = CreateConnectionActor(name, adapter); + AwaitConnectedCount(name, 1); + + var probe = CreateTestProbe(); + probe.Send(actor, new SubscribeTagsRequest("c1", "inst1", name, [Tag], DateTimeOffset.UtcNow)); + var initial = adapter.AwaitCall(0); + + DropAndReconnect(name, adapter, 2); + var resubscribe = adapter.AwaitCall(1); + var disconnectsAfterReconnect = HealthTransitions(name, ConnectionHealth.Disconnected); + + // Dead-session completion first … + initial.Complete(Tag, "dead-1"); + probe.FishForMessage(m => m.CorrelationId == "c1", TimeSpan.FromSeconds(5)); + // … then the live re-subscribe row. + resubscribe.Complete(Tag, "live-2"); + + // Let the live completion (and any release it would trigger) be applied. + Thread.Sleep(500); + + Assert.DoesNotContain("live-2", adapter.ReleasedIds); + Assert.Equal(disconnectsAfterReconnect, HealthTransitions(name, ConnectionHealth.Disconnected)); + + // The registry holds the LIVE handle: tearing the instance down releases live-2. + probe.Send(actor, new UnsubscribeTagsRequest("c2", "inst1", name, DateTimeOffset.UtcNow)); + AwaitCondition(() => adapter.ReleasedIds.Contains("live-2"), TimeSpan.FromSeconds(5)); + } + + /// + /// (b) The stale completion lands AFTER the re-subscribe row. Its handle belongs to the dead + /// session and must not be released against the same adapter: the new session may have reused + /// that handle for a different, live tag (MxGateway handles are session-scoped ints). + /// + [Fact] + public void StaleCompletion_AfterResubscribeRow_NeverReleasesAHandleTheNewSessionReused() + { + const string name = "same-adapter-b"; + var adapter = new GatedAdapter(); + var actor = CreateConnectionActor(name, adapter); + AwaitConnectedCount(name, 1); + + var probe = CreateTestProbe(); + // inst2 owns OtherTag, fully established in the first session. + probe.Send(actor, new SubscribeTagsRequest("o1", "inst2", name, [OtherTag], DateTimeOffset.UtcNow)); + adapter.AwaitCall(0).Complete(OtherTag, "h0"); + probe.FishForMessage(m => m.CorrelationId == "o1" && m.Success, TimeSpan.FromSeconds(5)); + + // inst1's subscribe is still in flight when the session drops. + probe.Send(actor, new SubscribeTagsRequest("c1", "inst1", name, [Tag], DateTimeOffset.UtcNow)); + var initial = adapter.AwaitCall(1); + + DropAndReconnect(name, adapter, 2); + var resubscribe = adapter.AwaitCall(2); + var disconnectsAfterReconnect = HealthTransitions(name, ConnectionHealth.Disconnected); + + // The new session hands OtherTag the handle number "h1" … + resubscribe.CompleteMany((OtherTag, "h1"), (Tag, "h2")); + AwaitResolved(actor, 2); + + // … which is the very number the dead session gave inst1's in-flight subscribe. + initial.Complete(Tag, "h1"); + probe.FishForMessage(m => m.CorrelationId == "c1", TimeSpan.FromSeconds(5)); + actor.Tell(new DataConnectionActor.GetHealthReport(), probe.Ref); + probe.FishForMessage(_ => true, TimeSpan.FromSeconds(5)); + + Assert.DoesNotContain("h1", adapter.ReleasedIds); + Assert.DoesNotContain("h2", adapter.ReleasedIds); + Assert.Equal(disconnectsAfterReconnect, HealthTransitions(name, ConnectionHealth.Disconnected)); + } + + /// + /// A connection-level failure row in a dead-session completion must not knock the freshly + /// reconnected link back into Reconnecting. + /// + [Fact] + public void StaleConnectionLevelFailure_DoesNotDriveTheReconnectedLinkBackToReconnecting() + { + const string name = "same-adapter-connfail"; + var adapter = new GatedAdapter(); + var actor = CreateConnectionActor(name, adapter); + AwaitConnectedCount(name, 1); + + var probe = CreateTestProbe(); + probe.Send(actor, new SubscribeTagsRequest("c1", "inst1", name, [Tag], DateTimeOffset.UtcNow)); + var initial = adapter.AwaitCall(0); + + DropAndReconnect(name, adapter, 2); + adapter.AwaitCall(1).Complete(Tag, "live-2"); + var disconnectsAfterReconnect = HealthTransitions(name, ConnectionHealth.Disconnected); + + // InvalidOperationException is classified as connection-level (transport down). + initial.Fail(new InvalidOperationException("adapter not connected")); + probe.FishForMessage(m => m.CorrelationId == "c1", TimeSpan.FromSeconds(5)); + actor.Tell(new DataConnectionActor.GetHealthReport(), probe.Ref); + probe.FishForMessage(_ => true, TimeSpan.FromSeconds(5)); + ExpectNoMsg(TimeSpan.FromMilliseconds(300)); + + Assert.Equal(disconnectsAfterReconnect, HealthTransitions(name, ConnectionHealth.Disconnected)); + Assert.Equal(2, HealthTransitions(name, ConnectionHealth.Connected)); + Assert.DoesNotContain("live-2", adapter.ReleasedIds); + } + + /// One gated SubscribeBatchAsync call. + internal sealed class PendingBatch + { + private readonly TaskCompletionSource> _tcs = + new(TaskCreationOptions.RunContinuationsAsynchronously); + + public PendingBatch(IReadOnlyList tags) => Tags = tags; + + public IReadOnlyList Tags { get; } + + public Task> Task => _tcs.Task; + + public void Complete(string tag, string id) => CompleteMany((tag, id)); + + public void CompleteMany(params (string Tag, string Id)[] rows) => + _tcs.SetResult(rows.Select(r => new TagSubscribeResult(r.Tag, true, r.Id, null)).ToList()); + + public void Fail(Exception ex) => _tcs.SetException(ex); + } + + /// + /// Batch-capable adapter whose every subscribe batch is held until the test completes it, so + /// the arrival order of dead-session and live-session completions is fully controlled. The + /// same instance is reused across reconnects, exactly like the shipped adapters. + /// + internal sealed class GatedAdapter : IDataConnection, IBatchSubscribableConnection + { + private readonly ConcurrentDictionary _calls = new(); + private int _callCount; + + public readonly ConcurrentQueue ReleasedIds = new(); + + public ConnectionHealth Status { get; private set; } = ConnectionHealth.Disconnected; + + public event Action? Disconnected; + + public void RaiseDisconnected() + { + Status = ConnectionHealth.Disconnected; + Disconnected?.Invoke(); + } + + public PendingBatch AwaitCall(int index) + { + var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(5); + while (DateTime.UtcNow < deadline) + { + if (_calls.TryGetValue(index, out var call)) + return call; + Thread.Sleep(10); + } + + throw new TimeoutException($"Subscribe batch #{index} was never issued."); + } + + public Task ConnectAsync(IDictionary connectionDetails, CancellationToken cancellationToken = default) + { + Status = ConnectionHealth.Connected; + return System.Threading.Tasks.Task.CompletedTask; + } + + public Task DisconnectAsync(CancellationToken cancellationToken = default) => + System.Threading.Tasks.Task.CompletedTask; + + public Task> SubscribeBatchAsync( + IReadOnlyList tagPaths, SubscriptionCallback callback, CancellationToken cancellationToken = default) + { + var call = new PendingBatch(tagPaths.ToList()); + _calls[Interlocked.Increment(ref _callCount) - 1] = call; + return call.Task; + } + + public Task UnsubscribeBatchAsync(IReadOnlyList subscriptionIds, CancellationToken cancellationToken = default) + { + foreach (var id in subscriptionIds) + ReleasedIds.Enqueue(id); + return System.Threading.Tasks.Task.CompletedTask; + } + + public Task SubscribeAsync(string tagPath, SubscriptionCallback callback, CancellationToken cancellationToken = default) => + throw new NotSupportedException("batch-capable adapter"); + + public Task UnsubscribeAsync(string subscriptionId, CancellationToken cancellationToken = default) + { + ReleasedIds.Enqueue(subscriptionId); + return System.Threading.Tasks.Task.CompletedTask; + } + + public Task ReadAsync(string tagPath, CancellationToken cancellationToken = default) => + System.Threading.Tasks.Task.FromResult(new ReadResult(false, null, "no value")); + + public Task> ReadBatchAsync( + IEnumerable tagPaths, CancellationToken cancellationToken = default) => + System.Threading.Tasks.Task.FromResult>( + new Dictionary()); + + public Task WriteAsync(string tagPath, object? value, CancellationToken cancellationToken = default) => + System.Threading.Tasks.Task.FromResult(new WriteResult(true, null)); + + public Task> WriteBatchAsync( + IDictionary values, CancellationToken cancellationToken = default) => + System.Threading.Tasks.Task.FromResult>( + values.ToDictionary(kv => kv.Key, _ => new WriteResult(true, null))); + + public ValueTask DisposeAsync() => ValueTask.CompletedTask; + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Adapters/FakeMxGatewayClient.cs b/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Adapters/FakeMxGatewayClient.cs index 70f71281..105d9098 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Adapters/FakeMxGatewayClient.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Adapters/FakeMxGatewayClient.cs @@ -24,6 +24,13 @@ public sealed class FakeMxGatewayClient : IMxGatewayClient, IMxGatewayClientFact public Func, IReadOnlyList>? ReadHandler; public Func, IReadOnlyList>? WriteHandler; public Func, bool)>? BrowseHandler; + /// + /// When true, every restarts handle numbering at 1 — the + /// gateway's per-session item handles, which collide across sessions. + /// + public bool RestartHandlesOnConnect; + /// The transition callback of the most recently opened alarm stream. + public Action? AlarmCallback; private int _nextHandle; /// @@ -33,6 +40,8 @@ public sealed class FakeMxGatewayClient : IMxGatewayClient, IMxGatewayClientFact public Task ConnectAsync(MxGatewayConnectionOptions o, CancellationToken ct = default) { ConnectedWith = o; + if (RestartHandlesOnConnect) + _nextHandle = 0; return Task.CompletedTask; } @@ -115,6 +124,7 @@ public sealed class FakeMxGatewayClient : IMxGatewayClient, IMxGatewayClientFact // recorded so the union-filter (longest-common-prefix) behaviour is testable. lock (AlarmStreamPrefixes) AlarmStreamPrefixes.Add(alarmFilterPrefix); + AlarmCallback = onTransition; return Task.CompletedTask; } diff --git a/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Adapters/MxGatewayDataConnectionTests.cs b/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Adapters/MxGatewayDataConnectionTests.cs index fd91290c..e1a8f040 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Adapters/MxGatewayDataConnectionTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Adapters/MxGatewayDataConnectionTests.cs @@ -122,7 +122,8 @@ public class MxGatewayDataConnectionTests fake.OnUpdate!(new MxValueUpdate("T", 1, QualityCode.Good, DateTimeOffset.UtcNow)); Assert.Equal(0, hits); - Assert.Contains(subId, fake.Unsubscribed); + // Ids are session-scoped ("{session}:{handle}"); the gateway receives the raw handle. + Assert.Equal(subId[(subId.IndexOf(':') + 1)..], Assert.Single(fake.Unsubscribed)); } // ── read / write + error classification ── diff --git a/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Adapters/MxGatewaySessionScopedHandleTests.cs b/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Adapters/MxGatewaySessionScopedHandleTests.cs new file mode 100644 index 00000000..01ff1939 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Adapters/MxGatewaySessionScopedHandleTests.cs @@ -0,0 +1,136 @@ +using Microsoft.Extensions.Logging.Abstractions; +using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Protocol; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Alarms; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; +using ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Adapters; + +namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests.Adapters; + +/// +/// The adapter half of review 2026-10-21 02/C1 + 02/L1. The adapter instance is reused across +/// reconnects, but every reconnect opens a NEW gateway session whose item handles are plain ints +/// that restart (collide) per session. A subscription id handed out by a dead session must +/// therefore never reach the gateway again — it would unadvise whatever the live session gave that +/// number — and the routing maps must not keep dead-session entries alive. +/// +[Collection("DataConnectionManagerActor")] +public class MxGatewaySessionScopedHandleTests +{ + private static Dictionary Details() => new() + { + ["Endpoint"] = "http://gw:5000", + ["ApiKey"] = "key", + ["ReadTimeoutMs"] = "2000", + }; + + private static MxGatewayDataConnection NewAdapter(FakeMxGatewayClient fake) => + new(fake, NullLogger.Instance); + + /// + /// A batch release of a dead-session id never reaches the gateway, even when the new session + /// has handed the same handle number to a live tag — and that live tag keeps routing. + /// + [Fact] + public async Task DeadSessionId_IsNeverReleasedAtTheGateway_EvenWhenTheHandleWasReused() + { + var fake = new FakeMxGatewayClient { RestartHandlesOnConnect = true }; + var adapter = NewAdapter(fake); + await adapter.ConnectAsync(Details()); + var dead = (await adapter.SubscribeBatchAsync(["Old.Tag"], (_, _) => { }))[0].SubscriptionId!; + + await adapter.ConnectAsync(Details()); // reconnect: new session, handles restart at 1 + var liveHits = 0; + var live = (await adapter.SubscribeBatchAsync(["Live.Tag"], (_, _) => liveHits++))[0].SubscriptionId!; + + await adapter.UnsubscribeBatchAsync([dead]); + await adapter.UnsubscribeAsync(dead); + + Assert.Empty(fake.Unsubscribed); + Assert.NotEqual(dead, live); + + await WaitUntil(() => fake.OnUpdate is not null); + fake.OnUpdate!(new MxValueUpdate("Live.Tag", 2, QualityCode.Good, DateTimeOffset.UtcNow)); + Assert.True(liveHits >= 1); + + // A current-session id still releases normally. + await adapter.UnsubscribeBatchAsync([live]); + Assert.Single(fake.Unsubscribed); + } + + /// + /// A reconnect drops the previous session's routing entries (L1): an update for a tag the + /// new session has not (re-)subscribed is not routed to the dead session's callback. + /// + [Fact] + public async Task Reconnect_ClearsTheDeadSessionsRoutingEntries() + { + var fake = new FakeMxGatewayClient(); + var adapter = NewAdapter(fake); + await adapter.ConnectAsync(Details()); + var hits = 0; + await adapter.SubscribeBatchAsync(["A.Tag"], (_, _) => hits++); + + await adapter.ConnectAsync(Details()); + await WaitUntil(() => fake.OnUpdate is not null); + fake.OnUpdate!(new MxValueUpdate("A.Tag", 1, QualityCode.Good, DateTimeOffset.UtcNow)); + + Assert.Equal(0, hits); + } + + /// + /// A disconnect drops the routing entries too, so a later update is not delivered to a + /// subscription that no longer exists at the gateway. + /// + [Fact] + public async Task Disconnect_ClearsRoutingEntries() + { + var fake = new FakeMxGatewayClient(); + var adapter = NewAdapter(fake); + await adapter.ConnectAsync(Details()); + var hits = 0; + await adapter.SubscribeBatchAsync(["A.Tag"], (_, _) => hits++); + await WaitUntil(() => fake.OnUpdate is not null); + + await adapter.DisconnectAsync(); + fake.OnUpdate!(new MxValueUpdate("A.Tag", 1, QualityCode.Good, DateTimeOffset.UtcNow)); + + Assert.Equal(0, hits); + } + + /// + /// Releasing an alarm subscription id the adapter does not know (a dead session's) must not + /// decrement the live feed's reference count — at zero the shared stream is torn down while + /// live subscribers still depend on it. + /// + [Fact] + public async Task UnknownAlarmSubscriptionId_DoesNotTearDownTheLiveStream() + { + var fake = new FakeMxGatewayClient(); + var adapter = NewAdapter(fake); + await adapter.ConnectAsync(Details()); + + var delivered = 0; + await adapter.SubscribeAlarmsAsync("Tank01", null, t => + { + if (t.Kind == AlarmTransitionKind.SnapshotComplete) + Interlocked.Increment(ref delivered); + }); + await WaitUntil(() => fake.AlarmCallback is not null); + + await adapter.UnsubscribeAlarmsAsync(Guid.NewGuid().ToString()); + + fake.AlarmCallback!(MxGatewayAlarmMapper.SnapshotComplete()); + Assert.Equal(1, Volatile.Read(ref delivered)); + } + + private static async Task WaitUntil(Func condition, int timeoutMs = 30_000) + { + var deadline = DateTime.UtcNow.AddMilliseconds(timeoutMs); + while (!condition()) + { + if (DateTime.UtcNow > deadline) + throw new TimeoutException("Condition not met in time."); + await Task.Delay(10); + } + } +}