fix(dcl): reset MxGateway alarm-stream state on reconnect so the native-alarm feed resumes (arch-review 2026-08-19 P2-2 High)
ConnectAsync reset the event loop and client on reconnect but left _alarmCts/_alarmSubCount/_alarmSubSources/_alarmStreamPrefix untouched. The orphaned old RunAlarmStreamAsync loop kept retrying the disposed client every 5s forever, and SubscribeAlarmsAsync's coverage check saw the stale state and short-circuited the actor's post-reconnect transparent re-subscribe (ReSubscribeAllAlarms) without ever opening a stream on the new client — freezing the native-alarm mirror until failover/redeploy. Reset the alarm state in ConnectAsync, mirroring the existing lock-guarded block in DisconnectAsync/DisposeAsync.
This commit is contained in:
Joseph Doherty
@@ -133,6 +133,23 @@ public class MxGatewayDataConnection
|
||||
TaskContinuationOptions.OnlyOnFaulted);
|
||||
}
|
||||
|
||||
// Same reasoning as the event loop above, but for the native-alarm feed: without
|
||||
// this reset, the OLD RunAlarmStreamAsync loop keeps retrying against the now-
|
||||
// disposed previous client every 5s forever (it only exits on its own token), and
|
||||
// SubscribeAlarmsAsync's coverage check sees the stale _alarmCts/_alarmStreamPrefix
|
||||
// and short-circuits — so the transparent re-subscribe the actor issues after a
|
||||
// successful reconnect never opens a fresh stream on the NEW client. Mirror the
|
||||
// lock-guarded block already in DisconnectAsync/DisposeAsync.
|
||||
lock (_alarmLock)
|
||||
{
|
||||
_alarmCts?.Cancel();
|
||||
_alarmCts?.Dispose();
|
||||
_alarmCts = null;
|
||||
_alarmSubCount = 0;
|
||||
_alarmSubSources.Clear();
|
||||
_alarmStreamPrefix = string.Empty;
|
||||
}
|
||||
|
||||
Interlocked.Exchange(ref _disconnectFired, 0); // reset guard on (re)connect, like OPC UA
|
||||
_client = _clientFactory.Create();
|
||||
|
||||
|
||||
+186
@@ -0,0 +1,186 @@
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
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;
|
||||
|
||||
/// <summary>
|
||||
/// P2-2 (arch-review 2026-08-19, High): before the fix, <c>ConnectAsync</c> reset the
|
||||
/// event loop and the client but left the alarm-stream state (<c>_alarmCts</c>,
|
||||
/// <c>_alarmSubCount</c>, <c>_alarmSubSources</c>, <c>_alarmStreamPrefix</c>) untouched.
|
||||
/// After a reconnect, the DataConnectionActor's transparent re-subscribe
|
||||
/// (<c>ReSubscribeAllAlarms</c>) called <c>SubscribeAlarmsAsync</c> again, but the stale
|
||||
/// non-null <c>_alarmCts</c> plus the coverage check made it short-circuit and return
|
||||
/// success WITHOUT ever opening a new stream on the new client — leaving the orphaned
|
||||
/// old stream loop retrying the disposed old client forever and the native-alarm mirror
|
||||
/// frozen until failover/redeploy.
|
||||
/// </summary>
|
||||
[Collection("DataConnectionManagerActor")]
|
||||
public class MxGatewayDataConnectionAlarmReconnectTests
|
||||
{
|
||||
private static Dictionary<string, string> Details() => new()
|
||||
{
|
||||
["Endpoint"] = "http://gw:5000",
|
||||
["ApiKey"] = "key",
|
||||
};
|
||||
|
||||
[Fact]
|
||||
public async Task Reconnect_ThenSubscribeAlarms_OpensNewStreamOnNewClient()
|
||||
{
|
||||
var clients = new List<PerInstanceFakeMxGatewayClient>();
|
||||
var factory = new RecordingFactory(clients);
|
||||
var adapter = new MxGatewayDataConnection(factory, NullLogger<MxGatewayDataConnection>.Instance);
|
||||
|
||||
await adapter.ConnectAsync(Details());
|
||||
await adapter.SubscribeAlarmsAsync("Area1.Tank1", null, _ => { });
|
||||
await WaitUntilAsync(() => clients[0].AlarmStreamOpens.Count == 1);
|
||||
|
||||
// Reconnect (same source set as before — the coverage check would previously
|
||||
// treat this as already-covered and never touch the new client at all).
|
||||
await adapter.ConnectAsync(Details());
|
||||
Assert.Equal(2, clients.Count);
|
||||
|
||||
await adapter.SubscribeAlarmsAsync("Area1.Tank1", null, _ => { });
|
||||
|
||||
await WaitUntilAsync(() => clients[1].AlarmStreamOpens.Count == 1);
|
||||
Assert.Equal("Area1.Tank1", clients[1].AlarmStreamOpens[0]);
|
||||
// The OLD client must never see a second alarm-stream open after reconnect.
|
||||
Assert.Single(clients[0].AlarmStreamOpens);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Reconnect_CancelsTheOldAlarmStreamToken()
|
||||
{
|
||||
var clients = new List<PerInstanceFakeMxGatewayClient>();
|
||||
var factory = new RecordingFactory(clients);
|
||||
var adapter = new MxGatewayDataConnection(factory, NullLogger<MxGatewayDataConnection>.Instance);
|
||||
|
||||
await adapter.ConnectAsync(Details());
|
||||
await adapter.SubscribeAlarmsAsync("Area1.Tank1", null, _ => { });
|
||||
await WaitUntilAsync(() => clients[0].AlarmStreamTokens.Count == 1);
|
||||
var oldStreamToken = clients[0].AlarmStreamTokens[0];
|
||||
Assert.False(oldStreamToken.IsCancellationRequested);
|
||||
|
||||
await adapter.ConnectAsync(Details()); // reconnect
|
||||
|
||||
// The token the OLD RunAlarmStreamAsync loop was given must now be cancelled —
|
||||
// that is what stops it retrying against the disposed old client every 5s.
|
||||
Assert.True(oldStreamToken.IsCancellationRequested);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Reconnect_AlarmEventsFlowAgainOnTheNewClient()
|
||||
{
|
||||
var clients = new List<PerInstanceFakeMxGatewayClient>();
|
||||
var factory = new RecordingFactory(clients);
|
||||
var adapter = new MxGatewayDataConnection(factory, NullLogger<MxGatewayDataConnection>.Instance);
|
||||
|
||||
var received = new List<NativeAlarmTransition>();
|
||||
await adapter.ConnectAsync(Details());
|
||||
await adapter.SubscribeAlarmsAsync("Area1.Tank1", null, t => received.Add(t));
|
||||
await WaitUntilAsync(() => clients[0].AlarmStreamOpens.Count == 1);
|
||||
|
||||
// Prove the feed works pre-reconnect.
|
||||
clients[0].EmitAlarm(MakeTransition(AlarmTransitionKind.Snapshot));
|
||||
await WaitUntilAsync(() => received.Count == 1);
|
||||
|
||||
await adapter.ConnectAsync(Details()); // reconnect
|
||||
await adapter.SubscribeAlarmsAsync("Area1.Tank1", null, t => received.Add(t));
|
||||
await WaitUntilAsync(() => clients[1].AlarmStreamOpens.Count == 1);
|
||||
|
||||
// A transition on the OLD (disposed, orphaned) client must never reach the
|
||||
// callback again — only the NEW client's feed is live post-reconnect.
|
||||
clients[0].EmitAlarm(MakeTransition(AlarmTransitionKind.Raise));
|
||||
await Task.Delay(100);
|
||||
Assert.Single(received);
|
||||
|
||||
clients[1].EmitAlarm(MakeTransition(AlarmTransitionKind.Snapshot));
|
||||
await WaitUntilAsync(() => received.Count == 2);
|
||||
}
|
||||
|
||||
private static NativeAlarmTransition MakeTransition(AlarmTransitionKind kind) => new(
|
||||
"Area1.Tank1.HiAlarm", "Area1.Tank1", "HiAlarm", kind,
|
||||
new AlarmConditionState(true, false, null, AlarmShelveState.Unshelved, false, 400),
|
||||
"Area", "", "", "", "", null, DateTimeOffset.UtcNow, "", "");
|
||||
|
||||
private static async Task WaitUntilAsync(Func<bool> condition)
|
||||
{
|
||||
for (var i = 0; i < 100 && !condition(); i++)
|
||||
await Task.Delay(50);
|
||||
Assert.True(condition(), "condition not met within timeout");
|
||||
}
|
||||
|
||||
/// <summary>Builds a NEW client instance per <c>Create()</c> call, mirroring the real
|
||||
/// factory and <see cref="MxGatewayDataConnectionReconnectTests"/>'s NSubstitute-based
|
||||
/// setup — needed here (unlike <c>FakeMxGatewayClient</c>) because the test must tell
|
||||
/// the pre- and post-reconnect clients apart.</summary>
|
||||
private sealed class RecordingFactory : IMxGatewayClientFactory
|
||||
{
|
||||
private readonly List<PerInstanceFakeMxGatewayClient> _clients;
|
||||
public RecordingFactory(List<PerInstanceFakeMxGatewayClient> clients) => _clients = clients;
|
||||
|
||||
public IMxGatewayClient Create()
|
||||
{
|
||||
var c = new PerInstanceFakeMxGatewayClient();
|
||||
_clients.Add(c);
|
||||
return c;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Minimal per-instance fake — only the members this test exercises are
|
||||
/// implemented meaningfully; the rest are unused no-ops for the connect/alarm-only
|
||||
/// path under test.</summary>
|
||||
private sealed class PerInstanceFakeMxGatewayClient : IMxGatewayClient
|
||||
{
|
||||
public readonly List<string?> AlarmStreamOpens = new();
|
||||
public readonly List<CancellationToken> AlarmStreamTokens = new();
|
||||
private Action<NativeAlarmTransition>? _onTransition;
|
||||
private CancellationToken _streamToken;
|
||||
|
||||
public Task ConnectAsync(MxGatewayConnectionOptions o, CancellationToken ct = default) => Task.CompletedTask;
|
||||
public Task DisconnectAsync(CancellationToken ct = default) => Task.CompletedTask;
|
||||
public Task<string> SubscribeAsync(string tagPath, CancellationToken ct = default) => Task.FromResult(Guid.NewGuid().ToString());
|
||||
public Task UnsubscribeAsync(string subscriptionId, CancellationToken ct = default) => Task.CompletedTask;
|
||||
public Task<IReadOnlyList<MxSubscribeOutcome>> SubscribeBulkAsync(IReadOnlyList<string> tagPaths, CancellationToken ct = default)
|
||||
=> Task.FromResult<IReadOnlyList<MxSubscribeOutcome>>(Array.Empty<MxSubscribeOutcome>());
|
||||
public Task UnsubscribeBulkAsync(IReadOnlyList<string> subscriptionIds, CancellationToken ct = default) => Task.CompletedTask;
|
||||
public Task<IReadOnlyList<MxReadOutcome>> ReadAsync(IReadOnlyList<string> tagPaths, CancellationToken ct = default)
|
||||
=> Task.FromResult<IReadOnlyList<MxReadOutcome>>(Array.Empty<MxReadOutcome>());
|
||||
public Task<IReadOnlyList<MxWriteOutcome>> WriteAsync(IReadOnlyList<(string TagPath, object? Value)> writes, CancellationToken ct = default)
|
||||
=> Task.FromResult<IReadOnlyList<MxWriteOutcome>>(Array.Empty<MxWriteOutcome>());
|
||||
public Task<(IReadOnlyList<MxBrowseChild> Children, bool Truncated)> BrowseChildrenAsync(string? parentNodeId, CancellationToken ct = default)
|
||||
=> Task.FromResult<(IReadOnlyList<MxBrowseChild>, bool)>((Array.Empty<MxBrowseChild>(), false));
|
||||
public Task RunEventLoopAsync(Action<MxValueUpdate> onUpdate, CancellationToken ct = default)
|
||||
=> Task.Delay(Timeout.Infinite, ct).ContinueWith(_ => { }, TaskScheduler.Default);
|
||||
|
||||
public Task RunAlarmStreamAsync(string? alarmFilterPrefix, Action<NativeAlarmTransition> onTransition, CancellationToken ct = default)
|
||||
{
|
||||
// Mirrors the real RunAlarmStreamAsync seam: records the open, then just
|
||||
// parks on the token like the live gateway stream would between events —
|
||||
// the adapter-level bug under test is about which CLIENT/TOKEN the loop was
|
||||
// started on, not the internal 5s retry loop inside RealMxGatewayClient.
|
||||
lock (AlarmStreamOpens)
|
||||
{
|
||||
AlarmStreamOpens.Add(alarmFilterPrefix);
|
||||
AlarmStreamTokens.Add(ct);
|
||||
}
|
||||
_onTransition = onTransition;
|
||||
_streamToken = ct;
|
||||
return Task.Delay(Timeout.Infinite, ct).ContinueWith(_ => { }, TaskScheduler.Default);
|
||||
}
|
||||
|
||||
public ValueTask DisposeAsync() => ValueTask.CompletedTask;
|
||||
|
||||
/// <summary>Simulates a transition arriving on this client's live stream. Gated on
|
||||
/// the stream token exactly like the real gateway stream would be — once cancelled,
|
||||
/// no more events are delivered — so a test can prove the OLD client's feed is
|
||||
/// genuinely dead post-reconnect rather than just unobserved.</summary>
|
||||
public void EmitAlarm(NativeAlarmTransition transition)
|
||||
{
|
||||
if (_streamToken.IsCancellationRequested)
|
||||
return;
|
||||
_onTransition?.Invoke(transition);
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user