fix(GWC-26): attach the alarm monitor's lease before SubscribeAlarms
RunMonitorAsync issued SubscribeAlarms and the first reconcile before the internal distributor subscriber was attached (via ISessionManager .ReadAlarmEventsAsync). The pump has been running since MarkReady started the dashboard mirror and only fans to subscribers registered at fan-out time, so every transition raised in that two-round-trip window bypassed the alarm feed — and a missed Acknowledge was never repaired, because ApplyReconcile broadcast presence deltas only. - The monitor now takes the internal lease directly from its session BEFORE SubscribeAlarms and drains it after the first reconcile; window transitions buffer in the lease's bounded channel. Processing them after ApplyReconcile is order-safe (ApplyTransition handles alarms the snapshot already placed). - ISessionManager.ReadAlarmEventsAsync removed — zero remaining callers. - ApplyReconcile broadcasts an Acknowledge feed transition when a both-present alarm's state advanced to ActiveAcked. This is a feed-level repair on the AlarmFeedMessage/StreamAlarms surface rebuilt from the worker's own snapshot, not MxEvent emission, so the "never synthesize events" rule is untouched; the reasoning is recorded on ApplyReconcile. The alarm-monitor test fakes now hand the monitor a real Ready GatewaySession with a dashboard mirror, which is what makes the window reproducible. Docs: docs/Sessions.md and gateway.md alarm-monitor ordering notes. Refs: archreview/2026-07-12/remediation/10-gateway-core.md GWC-26
This commit is contained in:
Joseph Doherty
@@ -210,6 +210,17 @@ public sealed class GatewayAlarmMonitor : BackgroundService, IGatewayAlarmServic
|
||||
|
||||
try
|
||||
{
|
||||
// Attach the internal distributor subscriber BEFORE subscribing (GWC-26). The pump
|
||||
// has been running since MarkReady started the dashboard mirror, and the distributor
|
||||
// only fans to subscribers registered at the time of the fan-out, so a subscriber
|
||||
// taken after SubscribeAlarms + the first reconcile would silently lose every
|
||||
// transition raised inside that two-round-trip window — and a missed Acknowledge is
|
||||
// never repaired by the presence-only reconcile deltas. Transitions arriving while we
|
||||
// subscribe and reconcile simply buffer in this lease's bounded channel; if it ever
|
||||
// overflowed, the internal subscriber is disconnected (it never faults the session),
|
||||
// the enumeration below ends, and the supervisor loop restarts the lifecycle.
|
||||
using IEventSubscriberLease alarmLease = session.AttachInternalEventSubscriber();
|
||||
|
||||
await SubscribeAlarmsAsync(session.SessionId, subscription, stoppingToken).ConfigureAwait(false);
|
||||
await ReconcileAsync(session.SessionId, stoppingToken).ConfigureAwait(false);
|
||||
|
||||
@@ -228,9 +239,11 @@ public sealed class GatewayAlarmMonitor : BackgroundService, IGatewayAlarmServic
|
||||
// Consume mapped MxEvents through the session's single distributor pump (as an
|
||||
// internal, non-counted subscriber) rather than opening a second raw drain of the
|
||||
// worker event channel — a second drain would split events with the dashboard
|
||||
// mirror pump and silently lose Acknowledge/mode-change transitions.
|
||||
await foreach (MxEvent mxEvent in _sessionManager
|
||||
.ReadAlarmEventsAsync(session.SessionId, linked.Token)
|
||||
// mirror pump and silently lose Acknowledge/mode-change transitions. The lease was
|
||||
// taken above, before SubscribeAlarms; draining it only now is order-safe because
|
||||
// ApplyTransition handles alarms the reconcile already placed in the cache.
|
||||
await foreach (MxEvent mxEvent in alarmLease.Reader
|
||||
.ReadAllAsync(linked.Token)
|
||||
.ConfigureAwait(false))
|
||||
{
|
||||
if (mxEvent is { BodyCase: MxEvent.BodyOneofCase.OnAlarmTransition }
|
||||
@@ -511,6 +524,12 @@ public sealed class GatewayAlarmMonitor : BackgroundService, IGatewayAlarmServic
|
||||
|
||||
// Replaces the cache with the worker's authoritative snapshot, broadcasting
|
||||
// a synthetic transition for any alarm the live stream missed.
|
||||
//
|
||||
// Repair scope (GWC-26): presence deltas (Clear/Raise) plus acked-state deltas. These are
|
||||
// ALARM FEED transitions (AlarmFeedMessage on the StreamAlarms/dashboard surface), rebuilt
|
||||
// from the worker's own authoritative snapshot to repair what the live feed missed. They are
|
||||
// not MxEvents and never reach the gRPC StreamEvents path, so this feed-level repair does not
|
||||
// breach the "never synthesize events" rule, which governs MxEvent emission.
|
||||
private void ApplyReconcile(IEnumerable<ActiveAlarmSnapshot> snapshots)
|
||||
{
|
||||
Dictionary<string, ActiveAlarmSnapshot> next = new(StringComparer.Ordinal);
|
||||
@@ -536,12 +555,23 @@ public sealed class GatewayAlarmMonitor : BackgroundService, IGatewayAlarmServic
|
||||
|
||||
foreach (KeyValuePair<string, ActiveAlarmSnapshot> incoming in next)
|
||||
{
|
||||
if (!_alarms.ContainsKey(incoming.Key))
|
||||
if (!_alarms.TryGetValue(incoming.Key, out ActiveAlarmSnapshot? existing))
|
||||
{
|
||||
Broadcast(
|
||||
new AlarmFeedMessage { Transition = TransitionFromSnapshot(incoming.Value, AlarmTransitionKind.Raise) },
|
||||
incoming.Key);
|
||||
}
|
||||
else if (existing.CurrentState != incoming.Value.CurrentState
|
||||
&& incoming.Value.CurrentState == AlarmConditionState.ActiveAcked)
|
||||
{
|
||||
// The alarm was already known but the worker now reports it acknowledged: the
|
||||
// live Acknowledge transition never reached the feed. Without this the acked
|
||||
// state is absorbed silently by the snapshot replace below and subscribers show
|
||||
// the alarm unacked until it clears.
|
||||
Broadcast(
|
||||
new AlarmFeedMessage { Transition = TransitionFromSnapshot(incoming.Value, AlarmTransitionKind.Acknowledge) },
|
||||
incoming.Key);
|
||||
}
|
||||
}
|
||||
|
||||
_alarms.Clear();
|
||||
|
||||
@@ -43,18 +43,6 @@ public interface ISessionManager
|
||||
string sessionId,
|
||||
CancellationToken cancellationToken);
|
||||
|
||||
/// <summary>
|
||||
/// Reads mapped events for the central alarm monitor by attaching an internal
|
||||
/// (non-counted) distributor subscriber, so the alarm feed shares the one worker-event
|
||||
/// pump instead of opening a second raw drain of the single worker event channel.
|
||||
/// </summary>
|
||||
/// <param name="sessionId">Identifier of the session.</param>
|
||||
/// <param name="cancellationToken">Token to cancel the asynchronous operation.</param>
|
||||
/// <returns>The mapped <see cref="MxEvent"/>s fanned by the session's distributor.</returns>
|
||||
IAsyncEnumerable<MxEvent> ReadAlarmEventsAsync(
|
||||
string sessionId,
|
||||
CancellationToken cancellationToken);
|
||||
|
||||
/// <summary>Closes a session and terminates its worker process.</summary>
|
||||
/// <param name="sessionId">Identifier of the session to close.</param>
|
||||
/// <param name="cancellationToken">Token to cancel the asynchronous operation.</param>
|
||||
|
||||
@@ -1,5 +1,4 @@
|
||||
using System.Diagnostics.CodeAnalysis;
|
||||
using System.Runtime.CompilerServices;
|
||||
using System.Security.Cryptography;
|
||||
using Google.Protobuf.WellKnownTypes;
|
||||
using Microsoft.Extensions.Logging;
|
||||
@@ -191,22 +190,6 @@ public sealed class SessionManager : ISessionManager
|
||||
return session.ReadEventsAsync(cancellationToken);
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async IAsyncEnumerable<MxEvent> ReadAlarmEventsAsync(
|
||||
string sessionId,
|
||||
[EnumeratorCancellation] CancellationToken cancellationToken)
|
||||
{
|
||||
GatewaySession session = GetRequiredSession(sessionId);
|
||||
using IEventSubscriberLease lease = session.AttachInternalEventSubscriber();
|
||||
|
||||
await foreach (MxEvent mxEvent in lease.Reader
|
||||
.ReadAllAsync(cancellationToken)
|
||||
.ConfigureAwait(false))
|
||||
{
|
||||
yield return mxEvent;
|
||||
}
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async Task<SessionCloseResult> CloseSessionAsync(
|
||||
string sessionId,
|
||||
|
||||
@@ -8,6 +8,7 @@ using ZB.MOM.WW.MxGateway.Server.Alarms;
|
||||
using ZB.MOM.WW.MxGateway.Server.Configuration;
|
||||
using ZB.MOM.WW.MxGateway.Server.Metrics;
|
||||
using ZB.MOM.WW.MxGateway.Server.Sessions;
|
||||
using ZB.MOM.WW.MxGateway.Server.Workers;
|
||||
|
||||
namespace ZB.MOM.WW.MxGateway.Tests.Alarms;
|
||||
|
||||
@@ -420,6 +421,9 @@ public sealed class AlarmFailoverEndToEndTests
|
||||
string? ownerKeyId,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
// The monitor attaches its internal subscriber directly on this session, so the
|
||||
// session has to be a genuinely Ready one with a worker client feeding the
|
||||
// distributor pump — EmitEvent writes into that worker's event stream.
|
||||
GatewaySession session = new(
|
||||
Guid.NewGuid().ToString("N"),
|
||||
"Galaxy",
|
||||
@@ -432,6 +436,8 @@ public sealed class AlarmFailoverEndToEndTests
|
||||
TimeSpan.FromSeconds(30),
|
||||
TimeSpan.FromSeconds(30),
|
||||
DateTimeOffset.UtcNow);
|
||||
session.AttachWorkerClient(new ChannelWorkerClient(session.SessionId, _events.Reader));
|
||||
session.MarkReady();
|
||||
return Task.FromResult(session);
|
||||
}
|
||||
|
||||
@@ -460,29 +466,9 @@ public sealed class AlarmFailoverEndToEndTests
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async IAsyncEnumerable<WorkerEvent> ReadEventsAsync(
|
||||
public IAsyncEnumerable<WorkerEvent> ReadEventsAsync(
|
||||
string sessionId,
|
||||
[EnumeratorCancellation] CancellationToken cancellationToken)
|
||||
{
|
||||
await foreach (WorkerEvent workerEvent in _events.Reader.ReadAllAsync(cancellationToken))
|
||||
{
|
||||
yield return workerEvent;
|
||||
}
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async IAsyncEnumerable<MxEvent> ReadAlarmEventsAsync(
|
||||
string sessionId,
|
||||
[EnumeratorCancellation] CancellationToken cancellationToken)
|
||||
{
|
||||
await foreach (WorkerEvent workerEvent in _events.Reader.ReadAllAsync(cancellationToken))
|
||||
{
|
||||
if (workerEvent.Event is not null)
|
||||
{
|
||||
yield return workerEvent.Event;
|
||||
}
|
||||
}
|
||||
}
|
||||
CancellationToken cancellationToken) => throw new NotSupportedException();
|
||||
|
||||
/// <inheritdoc />
|
||||
public bool TryGetSession(string sessionId, [MaybeNullWhen(false)] out GatewaySession session)
|
||||
@@ -509,4 +495,50 @@ public sealed class AlarmFailoverEndToEndTests
|
||||
/// <inheritdoc />
|
||||
public Task ShutdownAsync(CancellationToken cancellationToken) => Task.CompletedTask;
|
||||
}
|
||||
|
||||
/// <summary>Ready worker client whose event stream is the fake session manager's channel.</summary>
|
||||
private sealed class ChannelWorkerClient(string sessionId, ChannelReader<WorkerEvent> events) : IWorkerClient
|
||||
{
|
||||
/// <inheritdoc />
|
||||
public string SessionId { get; } = sessionId;
|
||||
|
||||
/// <inheritdoc />
|
||||
public int? ProcessId { get; } = 4321;
|
||||
|
||||
/// <inheritdoc />
|
||||
public WorkerClientState State { get; } = WorkerClientState.Ready;
|
||||
|
||||
/// <inheritdoc />
|
||||
public DateTimeOffset LastHeartbeatAt { get; } = DateTimeOffset.UtcNow;
|
||||
|
||||
/// <inheritdoc />
|
||||
public Task StartAsync(CancellationToken cancellationToken) => Task.CompletedTask;
|
||||
|
||||
/// <inheritdoc />
|
||||
public Task<WorkerCommandReply> InvokeAsync(
|
||||
WorkerCommand command,
|
||||
TimeSpan timeout,
|
||||
CancellationToken cancellationToken) => Task.FromResult(new WorkerCommandReply());
|
||||
|
||||
/// <inheritdoc />
|
||||
public async IAsyncEnumerable<WorkerEvent> ReadEventsAsync(
|
||||
[EnumeratorCancellation] CancellationToken cancellationToken)
|
||||
{
|
||||
await foreach (WorkerEvent workerEvent in events.ReadAllAsync(cancellationToken).ConfigureAwait(false))
|
||||
{
|
||||
yield return workerEvent;
|
||||
}
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public Task ShutdownAsync(TimeSpan timeout, CancellationToken cancellationToken) => Task.CompletedTask;
|
||||
|
||||
/// <inheritdoc />
|
||||
public void Kill(string reason)
|
||||
{
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public ValueTask DisposeAsync() => ValueTask.CompletedTask;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,508 @@
|
||||
using System.Diagnostics.CodeAnalysis;
|
||||
using System.Runtime.CompilerServices;
|
||||
using System.Threading.Channels;
|
||||
using Google.Protobuf.WellKnownTypes;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using ZB.MOM.WW.MxGateway.Contracts.Proto;
|
||||
using ZB.MOM.WW.MxGateway.Server.Alarms;
|
||||
using ZB.MOM.WW.MxGateway.Server.Configuration;
|
||||
using ZB.MOM.WW.MxGateway.Server.Grpc;
|
||||
using ZB.MOM.WW.MxGateway.Server.Metrics;
|
||||
using ZB.MOM.WW.MxGateway.Server.Sessions;
|
||||
using ZB.MOM.WW.MxGateway.Server.Workers;
|
||||
using ZB.MOM.WW.MxGateway.Tests.TestSupport;
|
||||
|
||||
namespace ZB.MOM.WW.MxGateway.Tests.Alarms;
|
||||
|
||||
/// <summary>
|
||||
/// GWC-26 regression tests for the alarm monitor's startup ordering. Unlike the
|
||||
/// sibling alarm-monitor tests, the session manager here hands the monitor a REAL
|
||||
/// <see cref="GatewaySession"/> that is driven to Ready with a dashboard mirror, so the
|
||||
/// distributor pump is already running when the monitor attaches — the production
|
||||
/// condition under which a late internal subscriber silently misses everything the pump
|
||||
/// already fanned.
|
||||
/// </summary>
|
||||
public sealed class GatewayAlarmMonitorAttachOrderTests
|
||||
{
|
||||
private const string AlarmReference = "Galaxy!Area.Tank01.Hi";
|
||||
private static readonly TimeSpan WaitTimeout = TimeSpan.FromSeconds(30);
|
||||
|
||||
/// <summary>
|
||||
/// The monitor must take its internal distributor lease BEFORE issuing
|
||||
/// <c>SubscribeAlarms</c>, so transitions the worker emits during the
|
||||
/// subscribe + first-reconcile window buffer in the lease instead of being fanned to a
|
||||
/// subscriber set the monitor has not joined yet. The test parks the monitor inside its
|
||||
/// <c>SubscribeAlarms</c> round trip, emits a Raise and an Acknowledge, waits until the
|
||||
/// dashboard mirror proves the pump has already fanned both, and only then releases the
|
||||
/// monitor: with a late attach both transitions are lost to the alarm feed forever.
|
||||
/// </summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[Fact]
|
||||
public async Task TransitionsDuringSubscribeWindow_StillReachTheAlarmFeed()
|
||||
{
|
||||
using GatewayMetrics metrics = new();
|
||||
await using FakeSessionManager sessions = new();
|
||||
sessions.HoldSubscribeUntilReleased();
|
||||
using GatewayAlarmMonitor monitor = CreateMonitor(sessions, metrics);
|
||||
|
||||
using CancellationTokenSource cts = new();
|
||||
await monitor.StartAsync(cts.Token);
|
||||
await sessions.WaitForSubscribeStartAsync(WaitTimeout);
|
||||
|
||||
// A live feed subscriber, drained past its baseline ProviderStatus so it is registered
|
||||
// before any window transition is broadcast.
|
||||
List<AlarmFeedMessage> received = [];
|
||||
TaskCompletionSource baselineReceived = new(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
using CancellationTokenSource streamCts = new();
|
||||
Task reader = ReadFeedAsync(monitor, received, baselineReceived, streamCts.Token);
|
||||
await baselineReceived.Task.WaitAsync(WaitTimeout);
|
||||
|
||||
// The window: the worker reports a raise and an acknowledge while the monitor is still
|
||||
// waiting for its SubscribeAlarms reply.
|
||||
sessions.EmitEvent(Transition(1, AlarmTransitionKind.Raise));
|
||||
sessions.EmitEvent(Transition(2, AlarmTransitionKind.Acknowledge));
|
||||
|
||||
// The dashboard mirror is an independent distributor subscriber: once it has both
|
||||
// events, the pump has provably already fanned them, so a subscriber that registers
|
||||
// after this point can never receive them.
|
||||
await WaitUntilAsync(() => sessions.Broadcaster.Captures.Count == 2, WaitTimeout);
|
||||
|
||||
sessions.ReleaseSubscribe();
|
||||
|
||||
AlarmFeedMessage raise = await WaitForAsync(
|
||||
received,
|
||||
m => m.PayloadCase == AlarmFeedMessage.PayloadOneofCase.Transition
|
||||
&& m.Transition.TransitionKind == AlarmTransitionKind.Raise,
|
||||
WaitTimeout);
|
||||
AlarmFeedMessage acknowledge = await WaitForAsync(
|
||||
received,
|
||||
m => m.PayloadCase == AlarmFeedMessage.PayloadOneofCase.Transition
|
||||
&& m.Transition.TransitionKind == AlarmTransitionKind.Acknowledge,
|
||||
WaitTimeout);
|
||||
|
||||
Assert.Equal(AlarmReference, raise.Transition.AlarmFullReference);
|
||||
Assert.Equal(AlarmReference, acknowledge.Transition.AlarmFullReference);
|
||||
|
||||
await streamCts.CancelAsync();
|
||||
await reader;
|
||||
await cts.CancelAsync();
|
||||
await monitor.StopAsync(CancellationToken.None);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Defense in depth for any window the attach reorder cannot cover (worker restart,
|
||||
/// internal-subscriber overflow disconnect): when a reconcile snapshot reports an alarm
|
||||
/// the cache already holds but with the state advanced to
|
||||
/// <see cref="AlarmConditionState.ActiveAcked"/>, the monitor broadcasts an
|
||||
/// <see cref="AlarmTransitionKind.Acknowledge"/> feed transition. Before the fix the
|
||||
/// acked state was absorbed silently by the snapshot replace, leaving live subscribers
|
||||
/// showing the alarm unacked until it cleared.
|
||||
/// </summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[Fact]
|
||||
public async Task ApplyReconcileBroadcastsAcknowledgeDelta()
|
||||
{
|
||||
using GatewayMetrics metrics = new();
|
||||
await using FakeSessionManager sessions = new();
|
||||
using GatewayAlarmMonitor monitor = CreateMonitor(sessions, metrics);
|
||||
|
||||
using CancellationTokenSource cts = new();
|
||||
await monitor.StartAsync(cts.Token);
|
||||
await sessions.WaitForSubscribeStartAsync(WaitTimeout);
|
||||
|
||||
// Seed the cache with an active (unacked) alarm and keep the reconcile snapshot in
|
||||
// agreement, so a periodic reconcile pass cannot clear it out from under the test.
|
||||
sessions.SetReconcileSnapshot(Snapshot(AlarmConditionState.Active));
|
||||
sessions.EmitEvent(Transition(1, AlarmTransitionKind.Raise));
|
||||
await WaitUntilAsync(
|
||||
() => monitor.CurrentAlarms.Any(alarm => alarm.AlarmFullReference == AlarmReference),
|
||||
WaitTimeout);
|
||||
|
||||
// Subscribe AFTER the raise: this reader's snapshot carries the alarm, so every
|
||||
// transition it observes from here on is a reconcile-derived broadcast.
|
||||
List<AlarmFeedMessage> received = [];
|
||||
TaskCompletionSource snapshotComplete = new(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
using CancellationTokenSource streamCts = new();
|
||||
Task reader = ReadFeedAsync(monitor, received, snapshotComplete, streamCts.Token, untilSnapshotComplete: true);
|
||||
await snapshotComplete.Task.WaitAsync(WaitTimeout);
|
||||
|
||||
// The worker now reports the same alarm acknowledged. A provider-mode event forces an
|
||||
// immediate reconcile pass so the test does not wait on the periodic timer.
|
||||
sessions.SetReconcileSnapshot(Snapshot(AlarmConditionState.ActiveAcked));
|
||||
sessions.EmitEvent(new MxEvent
|
||||
{
|
||||
Family = MxEventFamily.OnAlarmProviderModeChanged,
|
||||
WorkerSequence = 2,
|
||||
OnAlarmProviderModeChanged = new OnAlarmProviderModeChangedEvent
|
||||
{
|
||||
Mode = AlarmProviderMode.Alarmmgr,
|
||||
Reason = "probe",
|
||||
At = Timestamp.FromDateTimeOffset(DateTimeOffset.UtcNow),
|
||||
},
|
||||
});
|
||||
|
||||
AlarmFeedMessage acknowledge = await WaitForAsync(
|
||||
received,
|
||||
m => m.PayloadCase == AlarmFeedMessage.PayloadOneofCase.Transition
|
||||
&& m.Transition.TransitionKind == AlarmTransitionKind.Acknowledge,
|
||||
WaitTimeout);
|
||||
Assert.Equal(AlarmReference, acknowledge.Transition.AlarmFullReference);
|
||||
|
||||
lock (received)
|
||||
{
|
||||
AlarmFeedMessage[] transitions = received
|
||||
.Where(m => m.PayloadCase == AlarmFeedMessage.PayloadOneofCase.Transition)
|
||||
.ToArray();
|
||||
AlarmFeedMessage single = Assert.Single(transitions);
|
||||
Assert.Equal(AlarmTransitionKind.Acknowledge, single.Transition.TransitionKind);
|
||||
}
|
||||
|
||||
await streamCts.CancelAsync();
|
||||
await reader;
|
||||
await cts.CancelAsync();
|
||||
await monitor.StopAsync(CancellationToken.None);
|
||||
}
|
||||
|
||||
private static GatewayAlarmMonitor CreateMonitor(FakeSessionManager sessions, GatewayMetrics metrics)
|
||||
{
|
||||
AlarmsOptions options = new()
|
||||
{
|
||||
Enabled = true,
|
||||
SubscriptionExpression = @"\\NODE\Galaxy!Area",
|
||||
};
|
||||
return new GatewayAlarmMonitor(
|
||||
sessions,
|
||||
new StubWatchListResolver(),
|
||||
metrics,
|
||||
Microsoft.Extensions.Options.Options.Create(new GatewayOptions { Alarms = options }),
|
||||
NullLogger<GatewayAlarmMonitor>.Instance);
|
||||
}
|
||||
|
||||
// Drains the monitor's feed into received, signalling gate on the first message (the
|
||||
// baseline ProviderStatus) or, when untilSnapshotComplete is set, on SnapshotComplete.
|
||||
private static Task ReadFeedAsync(
|
||||
GatewayAlarmMonitor monitor,
|
||||
List<AlarmFeedMessage> received,
|
||||
TaskCompletionSource gate,
|
||||
CancellationToken cancellationToken,
|
||||
bool untilSnapshotComplete = false)
|
||||
{
|
||||
return Task.Run(
|
||||
async () =>
|
||||
{
|
||||
try
|
||||
{
|
||||
await foreach (AlarmFeedMessage message in monitor.StreamAsync(null, cancellationToken))
|
||||
{
|
||||
bool opensGate = !untilSnapshotComplete
|
||||
|| message.PayloadCase == AlarmFeedMessage.PayloadOneofCase.SnapshotComplete;
|
||||
|
||||
// Record only what arrives AFTER the gate opened: everything up to and
|
||||
// including the gate message is this subscriber's snapshot preamble, not
|
||||
// a live broadcast, so assertions stay about broadcasts alone.
|
||||
lock (received)
|
||||
{
|
||||
if (gate.Task.IsCompleted)
|
||||
{
|
||||
received.Add(message);
|
||||
}
|
||||
}
|
||||
|
||||
if (opensGate)
|
||||
{
|
||||
gate.TrySetResult();
|
||||
}
|
||||
}
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
// Expected when the test cancels the stream.
|
||||
}
|
||||
},
|
||||
CancellationToken.None);
|
||||
}
|
||||
|
||||
private static MxEvent Transition(ulong sequence, AlarmTransitionKind kind) => new()
|
||||
{
|
||||
Family = MxEventFamily.OnAlarmTransition,
|
||||
WorkerSequence = sequence,
|
||||
OnAlarmTransition = new OnAlarmTransitionEvent
|
||||
{
|
||||
AlarmFullReference = AlarmReference,
|
||||
SourceObjectReference = "Tank01",
|
||||
AlarmTypeName = "AnalogLimitAlarm.Hi",
|
||||
TransitionKind = kind,
|
||||
Severity = 500,
|
||||
SourceProvider = AlarmProviderMode.Alarmmgr,
|
||||
TransitionTimestamp = Timestamp.FromDateTimeOffset(DateTimeOffset.UtcNow),
|
||||
},
|
||||
};
|
||||
|
||||
private static ActiveAlarmSnapshot Snapshot(AlarmConditionState state) => new()
|
||||
{
|
||||
AlarmFullReference = AlarmReference,
|
||||
SourceObjectReference = "Tank01",
|
||||
AlarmTypeName = "AnalogLimitAlarm.Hi",
|
||||
CurrentState = state,
|
||||
Severity = 500,
|
||||
SourceProvider = AlarmProviderMode.Alarmmgr,
|
||||
};
|
||||
|
||||
private static async Task<AlarmFeedMessage> WaitForAsync(
|
||||
List<AlarmFeedMessage> received,
|
||||
Func<AlarmFeedMessage, bool> predicate,
|
||||
TimeSpan timeout)
|
||||
{
|
||||
DateTime deadline = DateTime.UtcNow + timeout;
|
||||
while (DateTime.UtcNow < deadline)
|
||||
{
|
||||
lock (received)
|
||||
{
|
||||
AlarmFeedMessage? match = received.FirstOrDefault(predicate);
|
||||
if (match is not null)
|
||||
{
|
||||
return match;
|
||||
}
|
||||
}
|
||||
|
||||
await Task.Delay(25);
|
||||
}
|
||||
|
||||
throw new TimeoutException("No matching AlarmFeedMessage was received in time.");
|
||||
}
|
||||
|
||||
private static async Task WaitUntilAsync(Func<bool> condition, TimeSpan timeout)
|
||||
{
|
||||
DateTime deadline = DateTime.UtcNow + timeout;
|
||||
while (DateTime.UtcNow < deadline)
|
||||
{
|
||||
if (condition())
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
await Task.Delay(25);
|
||||
}
|
||||
|
||||
throw new TimeoutException("Condition was not met in time.");
|
||||
}
|
||||
|
||||
/// <summary><see cref="IAlarmWatchListResolver"/> that resolves an empty watch-list.</summary>
|
||||
private sealed class StubWatchListResolver : IAlarmWatchListResolver
|
||||
{
|
||||
/// <inheritdoc />
|
||||
public Task<IReadOnlyList<AlarmSubtagTarget>> ResolveAsync(
|
||||
AlarmsOptions options,
|
||||
CancellationToken cancellationToken = default) =>
|
||||
Task.FromResult<IReadOnlyList<AlarmSubtagTarget>>([]);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Session manager that hands the monitor a real <see cref="GatewaySession"/> driven to
|
||||
/// Ready with a dashboard mirror, so the distributor pump is running before the monitor
|
||||
/// attaches. <see cref="EmitEvent"/> pushes worker events through the fake worker client
|
||||
/// into that pump, exactly as a live worker would.
|
||||
/// </summary>
|
||||
private sealed class FakeSessionManager : ISessionManager, IAsyncDisposable
|
||||
{
|
||||
private readonly Channel<WorkerEvent> _events = Channel.CreateUnbounded<WorkerEvent>();
|
||||
private readonly TaskCompletionSource _subscribeStarted =
|
||||
new(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
private readonly object _sync = new();
|
||||
private TaskCompletionSource _subscribeGate = CreateReleasedGate();
|
||||
private ActiveAlarmSnapshot[] _reconcileSnapshot = [];
|
||||
private GatewaySession? _session;
|
||||
|
||||
/// <summary>Dashboard mirror attached to the session; proves what the pump has fanned.</summary>
|
||||
public RecordingDashboardEventBroadcaster Broadcaster { get; } = new();
|
||||
|
||||
/// <summary>Re-arms the gate so <c>SubscribeAlarms</c> parks until <see cref="ReleaseSubscribe"/>.</summary>
|
||||
public void HoldSubscribeUntilReleased() =>
|
||||
_subscribeGate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
|
||||
/// <summary>Releases a gate armed by <see cref="HoldSubscribeUntilReleased"/>.</summary>
|
||||
public void ReleaseSubscribe() => _subscribeGate.TrySetResult();
|
||||
|
||||
/// <summary>Completes once the monitor's <c>SubscribeAlarms</c> command has arrived.</summary>
|
||||
/// <param name="timeout">The maximum time to wait.</param>
|
||||
/// <returns>A task that completes when the command arrives.</returns>
|
||||
public Task WaitForSubscribeStartAsync(TimeSpan timeout) => _subscribeStarted.Task.WaitAsync(timeout);
|
||||
|
||||
/// <summary>Sets the active-alarm snapshot every <c>QueryActiveAlarms</c> reconcile returns.</summary>
|
||||
/// <param name="snapshots">The snapshots to report.</param>
|
||||
public void SetReconcileSnapshot(params ActiveAlarmSnapshot[] snapshots)
|
||||
{
|
||||
lock (_sync)
|
||||
{
|
||||
_reconcileSnapshot = snapshots;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Pushes a worker event into the session's distributor pump.</summary>
|
||||
/// <param name="mxEvent">The event to push.</param>
|
||||
public void EmitEvent(MxEvent mxEvent) =>
|
||||
_events.Writer.TryWrite(new WorkerEvent { Event = mxEvent });
|
||||
|
||||
/// <inheritdoc />
|
||||
public Task<GatewaySession> OpenSessionAsync(
|
||||
SessionOpenRequest request,
|
||||
string? clientIdentity,
|
||||
string? ownerKeyId,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
GatewaySession session = new(
|
||||
sessionId: "session-alarm-attach-order",
|
||||
backendName: "Galaxy",
|
||||
pipeName: "mxaccess-gateway-1-session-alarm-attach-order",
|
||||
nonce: "nonce",
|
||||
clientIdentity: clientIdentity,
|
||||
ownerKeyId: ownerKeyId,
|
||||
clientSessionName: request.ClientSessionName,
|
||||
clientCorrelationId: request.ClientCorrelationId,
|
||||
commandTimeout: TimeSpan.FromSeconds(30),
|
||||
startupTimeout: TimeSpan.FromSeconds(30),
|
||||
shutdownTimeout: TimeSpan.FromSeconds(30),
|
||||
leaseDuration: TimeSpan.FromMinutes(30),
|
||||
openedAt: DateTimeOffset.UtcNow,
|
||||
eventStreaming: new SessionEventStreaming(
|
||||
new MxAccessGrpcMapper(),
|
||||
new EventOptions { QueueCapacity = 64 },
|
||||
NullLogger<SessionEventDistributor>.Instance,
|
||||
TimeProvider.System,
|
||||
new GatewayMetrics(),
|
||||
Broadcaster));
|
||||
session.AttachWorkerClient(new ChannelWorkerClient(session.SessionId, _events.Reader));
|
||||
|
||||
// MarkReady starts the dashboard mirror, and with it the distributor pump — the
|
||||
// production precondition this regression depends on.
|
||||
session.MarkReady();
|
||||
_session = session;
|
||||
return Task.FromResult(session);
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async Task<WorkerCommandReply> InvokeAsync(
|
||||
string sessionId,
|
||||
WorkerCommand command,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
MxCommandReply reply = new()
|
||||
{
|
||||
ProtocolStatus = new ProtocolStatus { Code = ProtocolStatusCode.Ok },
|
||||
};
|
||||
|
||||
switch (command.Command?.Kind)
|
||||
{
|
||||
case MxCommandKind.SubscribeAlarms:
|
||||
_subscribeStarted.TrySetResult();
|
||||
await _subscribeGate.Task.WaitAsync(cancellationToken).ConfigureAwait(false);
|
||||
break;
|
||||
case MxCommandKind.QueryActiveAlarms:
|
||||
QueryActiveAlarmsReplyPayload payload = new();
|
||||
lock (_sync)
|
||||
{
|
||||
payload.Snapshots.AddRange(_reconcileSnapshot.Select(snapshot => snapshot.Clone()));
|
||||
}
|
||||
|
||||
reply.QueryActiveAlarms = payload;
|
||||
break;
|
||||
}
|
||||
|
||||
return new WorkerCommandReply { Reply = reply };
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public IAsyncEnumerable<WorkerEvent> ReadEventsAsync(
|
||||
string sessionId,
|
||||
CancellationToken cancellationToken) => throw new NotSupportedException();
|
||||
|
||||
/// <inheritdoc />
|
||||
public bool TryGetSession(string sessionId, [MaybeNullWhen(false)] out GatewaySession session)
|
||||
{
|
||||
session = _session;
|
||||
return session is not null;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public Task<SessionCloseResult> CloseSessionAsync(string sessionId, CancellationToken cancellationToken)
|
||||
{
|
||||
_events.Writer.TryComplete();
|
||||
return Task.FromResult(new SessionCloseResult(sessionId, SessionState.Closed, AlreadyClosed: false));
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public Task<SessionCloseResult> KillWorkerAsync(string sessionId, string reason, CancellationToken cancellationToken) =>
|
||||
Task.FromResult(new SessionCloseResult(sessionId, SessionState.Closed, AlreadyClosed: false));
|
||||
|
||||
/// <inheritdoc />
|
||||
public Task<int> CloseExpiredLeasesAsync(DateTimeOffset now, CancellationToken cancellationToken) =>
|
||||
Task.FromResult(0);
|
||||
|
||||
/// <inheritdoc />
|
||||
public Task ShutdownAsync(CancellationToken cancellationToken) => Task.CompletedTask;
|
||||
|
||||
/// <summary>Disposes the session the fake handed out.</summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
public async ValueTask DisposeAsync()
|
||||
{
|
||||
_events.Writer.TryComplete();
|
||||
if (_session is not null)
|
||||
{
|
||||
await _session.DisposeAsync().ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
|
||||
private static TaskCompletionSource CreateReleasedGate()
|
||||
{
|
||||
TaskCompletionSource gate = new(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
gate.SetResult();
|
||||
return gate;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Worker client whose event stream is a test-driven channel.</summary>
|
||||
private sealed class ChannelWorkerClient(string sessionId, ChannelReader<WorkerEvent> events) : IWorkerClient
|
||||
{
|
||||
/// <inheritdoc />
|
||||
public string SessionId { get; } = sessionId;
|
||||
|
||||
/// <inheritdoc />
|
||||
public int? ProcessId { get; } = 4321;
|
||||
|
||||
/// <inheritdoc />
|
||||
public WorkerClientState State { get; } = WorkerClientState.Ready;
|
||||
|
||||
/// <inheritdoc />
|
||||
public DateTimeOffset LastHeartbeatAt { get; } = DateTimeOffset.UtcNow;
|
||||
|
||||
/// <inheritdoc />
|
||||
public Task StartAsync(CancellationToken cancellationToken) => Task.CompletedTask;
|
||||
|
||||
/// <inheritdoc />
|
||||
public Task<WorkerCommandReply> InvokeAsync(
|
||||
WorkerCommand command,
|
||||
TimeSpan timeout,
|
||||
CancellationToken cancellationToken) => Task.FromResult(new WorkerCommandReply());
|
||||
|
||||
/// <inheritdoc />
|
||||
public async IAsyncEnumerable<WorkerEvent> ReadEventsAsync(
|
||||
[EnumeratorCancellation] CancellationToken cancellationToken)
|
||||
{
|
||||
await foreach (WorkerEvent workerEvent in events.ReadAllAsync(cancellationToken).ConfigureAwait(false))
|
||||
{
|
||||
yield return workerEvent;
|
||||
}
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public Task ShutdownAsync(TimeSpan timeout, CancellationToken cancellationToken) => Task.CompletedTask;
|
||||
|
||||
/// <inheritdoc />
|
||||
public void Kill(string reason)
|
||||
{
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public ValueTask DisposeAsync() => ValueTask.CompletedTask;
|
||||
}
|
||||
}
|
||||
@@ -10,6 +10,7 @@ using ZB.MOM.WW.MxGateway.Server.Alarms;
|
||||
using ZB.MOM.WW.MxGateway.Server.Configuration;
|
||||
using ZB.MOM.WW.MxGateway.Server.Metrics;
|
||||
using ZB.MOM.WW.MxGateway.Server.Sessions;
|
||||
using ZB.MOM.WW.MxGateway.Server.Workers;
|
||||
|
||||
namespace ZB.MOM.WW.MxGateway.Tests.Alarms;
|
||||
|
||||
@@ -733,6 +734,9 @@ public sealed class GatewayAlarmMonitorProviderModeTests
|
||||
string? ownerKeyId,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
// The monitor attaches its internal subscriber directly on this session, so the
|
||||
// session has to be a genuinely Ready one with a worker client feeding the
|
||||
// distributor pump — EmitEvent writes into that worker's event stream.
|
||||
GatewaySession session = new(
|
||||
Guid.NewGuid().ToString("N"),
|
||||
"Galaxy",
|
||||
@@ -745,6 +749,8 @@ public sealed class GatewayAlarmMonitorProviderModeTests
|
||||
TimeSpan.FromSeconds(30),
|
||||
TimeSpan.FromSeconds(30),
|
||||
DateTimeOffset.UtcNow);
|
||||
session.AttachWorkerClient(new ChannelWorkerClient(session.SessionId, _events.Reader));
|
||||
session.MarkReady();
|
||||
return Task.FromResult(session);
|
||||
}
|
||||
|
||||
@@ -773,29 +779,9 @@ public sealed class GatewayAlarmMonitorProviderModeTests
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async IAsyncEnumerable<WorkerEvent> ReadEventsAsync(
|
||||
public IAsyncEnumerable<WorkerEvent> ReadEventsAsync(
|
||||
string sessionId,
|
||||
[EnumeratorCancellation] CancellationToken cancellationToken)
|
||||
{
|
||||
await foreach (WorkerEvent workerEvent in _events.Reader.ReadAllAsync(cancellationToken))
|
||||
{
|
||||
yield return workerEvent;
|
||||
}
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async IAsyncEnumerable<MxEvent> ReadAlarmEventsAsync(
|
||||
string sessionId,
|
||||
[EnumeratorCancellation] CancellationToken cancellationToken)
|
||||
{
|
||||
await foreach (WorkerEvent workerEvent in _events.Reader.ReadAllAsync(cancellationToken))
|
||||
{
|
||||
if (workerEvent.Event is not null)
|
||||
{
|
||||
yield return workerEvent.Event;
|
||||
}
|
||||
}
|
||||
}
|
||||
CancellationToken cancellationToken) => throw new NotSupportedException();
|
||||
|
||||
/// <inheritdoc />
|
||||
public bool TryGetSession(string sessionId, [MaybeNullWhen(false)] out GatewaySession session)
|
||||
@@ -822,4 +808,50 @@ public sealed class GatewayAlarmMonitorProviderModeTests
|
||||
/// <inheritdoc />
|
||||
public Task ShutdownAsync(CancellationToken cancellationToken) => Task.CompletedTask;
|
||||
}
|
||||
|
||||
/// <summary>Ready worker client whose event stream is the fake session manager's channel.</summary>
|
||||
private sealed class ChannelWorkerClient(string sessionId, ChannelReader<WorkerEvent> events) : IWorkerClient
|
||||
{
|
||||
/// <inheritdoc />
|
||||
public string SessionId { get; } = sessionId;
|
||||
|
||||
/// <inheritdoc />
|
||||
public int? ProcessId { get; } = 4321;
|
||||
|
||||
/// <inheritdoc />
|
||||
public WorkerClientState State { get; } = WorkerClientState.Ready;
|
||||
|
||||
/// <inheritdoc />
|
||||
public DateTimeOffset LastHeartbeatAt { get; } = DateTimeOffset.UtcNow;
|
||||
|
||||
/// <inheritdoc />
|
||||
public Task StartAsync(CancellationToken cancellationToken) => Task.CompletedTask;
|
||||
|
||||
/// <inheritdoc />
|
||||
public Task<WorkerCommandReply> InvokeAsync(
|
||||
WorkerCommand command,
|
||||
TimeSpan timeout,
|
||||
CancellationToken cancellationToken) => Task.FromResult(new WorkerCommandReply());
|
||||
|
||||
/// <inheritdoc />
|
||||
public async IAsyncEnumerable<WorkerEvent> ReadEventsAsync(
|
||||
[EnumeratorCancellation] CancellationToken cancellationToken)
|
||||
{
|
||||
await foreach (WorkerEvent workerEvent in events.ReadAllAsync(cancellationToken).ConfigureAwait(false))
|
||||
{
|
||||
yield return workerEvent;
|
||||
}
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public Task ShutdownAsync(TimeSpan timeout, CancellationToken cancellationToken) => Task.CompletedTask;
|
||||
|
||||
/// <inheritdoc />
|
||||
public void Kill(string reason)
|
||||
{
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public ValueTask DisposeAsync() => ValueTask.CompletedTask;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -378,14 +378,6 @@ public sealed class DashboardSessionAdminServiceTests
|
||||
throw new NotSupportedException();
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public IAsyncEnumerable<MxEvent> ReadAlarmEventsAsync(
|
||||
string sessionId,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
throw new NotSupportedException();
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public Task<SessionCloseResult> CloseSessionAsync(
|
||||
string sessionId,
|
||||
|
||||
@@ -755,14 +755,6 @@ public sealed class EventStreamServiceTests
|
||||
return _sessions[sessionId].ReadEventsAsync(cancellationToken);
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public IAsyncEnumerable<MxEvent> ReadAlarmEventsAsync(
|
||||
string sessionId,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
throw new NotSupportedException();
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public Task<SessionCloseResult> CloseSessionAsync(
|
||||
string sessionId,
|
||||
|
||||
@@ -948,14 +948,6 @@ public sealed class MxAccessGatewayServiceConstraintTests
|
||||
}
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public IAsyncEnumerable<MxEvent> ReadAlarmEventsAsync(
|
||||
string sessionId,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
throw new NotSupportedException();
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public Task<SessionCloseResult> CloseSessionAsync(
|
||||
string sessionId,
|
||||
|
||||
@@ -616,14 +616,6 @@ public sealed class MxAccessGatewayServiceTests
|
||||
}
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public IAsyncEnumerable<MxEvent> ReadAlarmEventsAsync(
|
||||
string sessionId,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
throw new NotSupportedException();
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public Task<SessionCloseResult> CloseSessionAsync(
|
||||
string sessionId,
|
||||
|
||||
@@ -352,11 +352,6 @@ public sealed class GatewaySessionDashboardMirrorTests
|
||||
string sessionId,
|
||||
CancellationToken cancellationToken) => session.ReadEventsAsync(cancellationToken);
|
||||
|
||||
/// <inheritdoc />
|
||||
public IAsyncEnumerable<MxEvent> ReadAlarmEventsAsync(
|
||||
string sessionId,
|
||||
CancellationToken cancellationToken) => throw new NotSupportedException();
|
||||
|
||||
/// <inheritdoc />
|
||||
public Task<SessionCloseResult> CloseSessionAsync(
|
||||
string sessionId,
|
||||
|
||||
-8
@@ -579,14 +579,6 @@ public sealed class GatewayGrpcAuthorizationInterceptorTests
|
||||
return AsyncEnumerable.Empty<WorkerEvent>();
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public IAsyncEnumerable<MxEvent> ReadAlarmEventsAsync(
|
||||
string sessionId,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
return AsyncEnumerable.Empty<MxEvent>();
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public Task<SessionCloseResult> CloseSessionAsync(
|
||||
string sessionId,
|
||||
|
||||
Reference in New Issue
Block a user