feat(GWC-37,WRK-32): refresh live alarm watch lists
ci / nightly-windev (push) Has been skipped
ci / windows-x86 (push) Failing after 51s
ci / java (push) Successful in 2m8s
ci / portable (push) Failing after 1h1m54s

This commit is contained in:
Joseph Doherty
2026-08-25 11:46:07 -04:00
parent 1b11e89f29
commit 484028e721
22 changed files with 3961 additions and 1858 deletions
File diff suppressed because it is too large Load Diff
@@ -130,6 +130,7 @@ message MxCommand {
WriteSecuredBulkCommand write_secured_bulk = 41;
WriteSecured2BulkCommand write_secured2_bulk = 42;
ReadBulkCommand read_bulk = 43;
UpdateAlarmWatchListCommand update_alarm_watch_list = 44;
PingCommand ping = 100;
GetSessionStateCommand get_session_state = 101;
GetWorkerInfoCommand get_worker_info = 102;
@@ -174,6 +175,7 @@ enum MxCommandKind {
MX_COMMAND_KIND_WRITE_SECURED_BULK = 32;
MX_COMMAND_KIND_WRITE_SECURED2_BULK = 33;
MX_COMMAND_KIND_READ_BULK = 34;
MX_COMMAND_KIND_UPDATE_ALARM_WATCH_LIST = 35;
MX_COMMAND_KIND_PING = 100;
MX_COMMAND_KIND_GET_SESSION_STATE = 101;
MX_COMMAND_KIND_GET_WORKER_INFO = 102;
@@ -344,8 +346,8 @@ enum AlarmProviderMode {
// Subscription expression follows the canonical
// `\\<machine>\Galaxy!<area>` format (literal "Galaxy" provider). The
// worker spins up a wnwrapConsumer-backed subscription on its STA on
// first call; subsequent calls are an error (use UnsubscribeAlarms then
// SubscribeAlarms to reconfigure).
// first call; subsequent calls are an error. Use UpdateAlarmWatchList to
// replace the standby targets without restarting the primary subscription.
message SubscribeAlarmsCommand {
string subscription_expression = 1;
// UNSPECIFIED = auto-failover/failback. ALARMMGR/SUBTAG force one provider.
@@ -354,6 +356,17 @@ message SubscribeAlarmsCommand {
// alarmmgr mode; in subtag mode it bounds what the consumer can observe.
repeated AlarmSubtagTarget watch_list = 3;
AlarmFailoverConfig failover = 4;
// True when auto mode should retain subtag-fallback capability even if the
// gateway's initial discovery produced an empty watch list. False leaves an
// empty list as an intentional alarm-manager-only subscription.
bool fallback_enabled = 5;
}
// Replace the live alarm subscription's subtag standby targets. The worker
// constructs and attaches the replacement before swapping it into service, so
// a setup failure leaves the existing standby and active provider unchanged.
message UpdateAlarmWatchListCommand {
repeated AlarmSubtagTarget watch_list = 1;
}
// Tear down the worker's alarm consumer. No-op if no subscription is
@@ -537,11 +550,14 @@ message MxCommandReply {
MxValue return_value = 6;
// Correlated per-item outcome rows. For WRITE / WRITE2 / WRITE_SECURED /
// WRITE_SECURED2 replies the worker holds the reply for a bounded window
// (default 1.5 s, MXGATEWAY_WORKER_WRITE_COMPLETION_WAIT_MS) waiting for
// (default 1.5 s, MxGateway:Worker:WriteCompletionWaitMilliseconds) waiting for
// the matching MXAccess OnWriteComplete callback and copies its status rows
// here, so statuses[0] carries the real MXAccess commit outcome (success OR
// failure) while protocol_status/hresult still describe command acceptance
// only. Empty statuses on a write reply means the completion did not arrive
// only. Callback correlation is recency-based per item handle: after the
// bounded wait, a sequential write's callback supersedes a timed-out
// predecessor, and a later stale callback for that predecessor is ignored.
// Empty statuses on a write reply means the completion did not arrive
// within the window — the write is unconfirmed, not failed. Correlation is
// best-effort per (server_handle, item_handle): MXAccess's callback carries
// no transaction id, so concurrent writes to the same item within the
@@ -66,9 +66,10 @@ public sealed class GatewayAlarmMonitor : BackgroundService, IGatewayAlarmServic
private bool _subtagCoverageMissing;
private string _subtagCoverageReason = string.Empty;
// The watch-list references actually sent to the worker in SubscribeAlarms, guarded by _sync.
// The worker holds that list for the life of the subscription, so a later re-resolution is
// compared against this to report drift rather than silently diverging.
// The full canonical watch-list fingerprint acknowledged by the worker and its alarm
// references, guarded by _sync. The fingerprint advances only after an OK subscribe/update
// reply, so a rejected replacement remains material and is retried on the next refresh edge.
private string _acknowledgedWatchListFingerprint = string.Empty;
private IReadOnlyCollection<string> _sentWatchListReferences = [];
// Whether the worker's most recent reconcile fetch was capped, guarded by _sync.
@@ -356,9 +357,6 @@ public sealed class GatewayAlarmMonitor : BackgroundService, IGatewayAlarmServic
// feed subscribers outlive the monitor's worker session and may already be attached.
lock (_sync)
{
_sentWatchListReferences = watchList
.Select(target => target.AlarmFullReference)
.ToHashSet(StringComparer.OrdinalIgnoreCase);
SetSubtagCoverage(GradeSubtagCoverage(forcedMode, resolution, staleAgainstSent: false), broadcast: true);
}
@@ -374,6 +372,7 @@ public sealed class GatewayAlarmMonitor : BackgroundService, IGatewayAlarmServic
{
SubscriptionExpression = subscription,
ForcedMode = forcedMode,
FallbackEnabled = CanUseSubtagFallback(forcedMode),
Failover = new AlarmFailoverConfig
{
ConsecutiveFailureThreshold = _options.Fallback.ConsecutiveFailureThreshold,
@@ -404,6 +403,11 @@ public sealed class GatewayAlarmMonitor : BackgroundService, IGatewayAlarmServic
?? $"status {code}";
throw new InvalidOperationException($"Worker rejected SubscribeAlarms: {diagnostic}");
}
lock (_sync)
{
AcknowledgeWatchList(watchList);
}
}
private async Task ReconcileLoopAsync(string sessionId, CancellationToken cancellationToken)
@@ -416,6 +420,8 @@ public sealed class GatewayAlarmMonitor : BackgroundService, IGatewayAlarmServic
{
try
{
await RefreshAlarmWatchListAsync(sessionId, broadcast: true, cancellationToken)
.ConfigureAwait(false);
await ReconcileAsync(sessionId, cancellationToken).ConfigureAwait(false);
}
catch (OperationCanceledException)
@@ -545,18 +551,14 @@ public sealed class GatewayAlarmMonitor : BackgroundService, IGatewayAlarmServic
AlarmProviderMode toMode = change.Mode;
string reason = change.Reason ?? string.Empty;
// GWC-37: this is the moment the subtag watch-list stops being theoretical — the provider
// that is about to serve is the one built from it. Re-resolve BEFORE the status is
// published so the frame the operator sees on the failover carries the coverage verdict
// rather than a second frame later, and so a watch-list that has drifted since
// SubscribeAlarms is reported instead of silently diverging. The re-resolution is NOT
// re-sent: the worker rejects a second SubscribeAlarms while subscribed, and the only way
// to replace the list — Unsubscribe then Subscribe — rebuilds the FailoverAlarmConsumer,
// which restarts on the primary that just failed and discards the warm standby state, so
// it would flap the feed rather than repair it. A monitor restart re-resolves and re-sends.
// The failover edge is the moment the subtag watch-list stops being theoretical. Refresh
// it before publishing the status frame so recovered discovery is applied to the already
// active standby and the operator sees the acknowledged coverage verdict atomically with
// the provider switch.
if (toMode == AlarmProviderMode.Subtag)
{
await RefreshSubtagCoverageAsync(cancellationToken).ConfigureAwait(false);
await RefreshAlarmWatchListAsync(sessionId, broadcast: false, cancellationToken)
.ConfigureAwait(false);
}
AlarmProviderStatus status;
@@ -606,14 +608,17 @@ public sealed class GatewayAlarmMonitor : BackgroundService, IGatewayAlarmServic
}
}
// GWC-37: re-resolves the subtag watch-list and re-grades the coverage verdict without
// touching the worker's subscription. Best-effort: a resolver that throws (only cancellation
// and programming errors do — discovery failure is reported in the resolution) leaves the
// verdict from the last subscribe in place, because a stale verdict is better than none.
// Does not broadcast: the callers publish a status frame of their own immediately after.
private async Task RefreshSubtagCoverageAsync(CancellationToken cancellationToken)
private async Task RefreshAlarmWatchListAsync(
string sessionId,
bool broadcast,
CancellationToken cancellationToken)
{
AlarmProviderMode forcedMode = MapForcedMode(_options.Fallback.Mode);
if (!CanUseSubtagFallback(forcedMode))
{
return;
}
AlarmWatchListResolution resolution;
try
{
@@ -627,18 +632,96 @@ public sealed class GatewayAlarmMonitor : BackgroundService, IGatewayAlarmServic
{
_logger.LogWarning(
exception,
"Re-resolving the alarm subtag watch-list failed; keeping the coverage verdict from the last subscribe.");
"Re-resolving the alarm subtag watch-list failed; keeping the last acknowledged watch list.");
return;
}
string resolvedFingerprint = CanonicalWatchListFingerprint(resolution.Targets);
string acknowledgedFingerprint;
lock (_sync)
{
bool stale = !resolution.Targets
.Select(target => target.AlarmFullReference)
.ToHashSet(StringComparer.OrdinalIgnoreCase)
.SetEquals(_sentWatchListReferences);
SetSubtagCoverage(GradeSubtagCoverage(forcedMode, resolution, stale), broadcast: false);
acknowledgedFingerprint = _acknowledgedWatchListFingerprint;
}
if (string.Equals(resolvedFingerprint, acknowledgedFingerprint, StringComparison.Ordinal))
{
lock (_sync)
{
SetSubtagCoverage(
GradeSubtagCoverage(forcedMode, resolution, staleAgainstSent: false),
broadcast);
}
return;
}
UpdateAlarmWatchListCommand update = new();
update.WatchList.AddRange(resolution.Targets);
WorkerCommandReply reply = await _sessionManager.InvokeAsync(
sessionId,
new WorkerCommand
{
Command = new MxCommand
{
Kind = MxCommandKind.UpdateAlarmWatchList,
UpdateAlarmWatchList = update,
},
},
cancellationToken)
.ConfigureAwait(false);
ProtocolStatusCode? code = reply.Reply?.ProtocolStatus?.Code;
lock (_sync)
{
if (code == ProtocolStatusCode.Ok)
{
AcknowledgeWatchList(resolution.Targets);
SetSubtagCoverage(
GradeSubtagCoverage(forcedMode, resolution, staleAgainstSent: false),
broadcast);
return;
}
string diagnostic = reply.Reply?.DiagnosticMessage
?? reply.Reply?.ProtocolStatus?.Message
?? $"status {code}";
string reason = GradeSubtagCoverage(forcedMode, resolution, staleAgainstSent: true);
SetSubtagCoverage($"{reason}; worker rejected watch-list update ({diagnostic})", broadcast);
}
}
private bool CanUseSubtagFallback(AlarmProviderMode forcedMode) =>
forcedMode != AlarmProviderMode.Alarmmgr
&& (forcedMode == AlarmProviderMode.Subtag
|| _options.Fallback.Discovery.UseGalaxyRepository
|| _options.Fallback.Discovery.IncludeAttributes.Length > 0);
// Caller holds _sync.
private void AcknowledgeWatchList(IReadOnlyList<AlarmSubtagTarget> watchList)
{
_acknowledgedWatchListFingerprint = CanonicalWatchListFingerprint(watchList);
_sentWatchListReferences = watchList
.Select(target => target.AlarmFullReference)
.ToHashSet(StringComparer.OrdinalIgnoreCase);
}
private static string CanonicalWatchListFingerprint(IEnumerable<AlarmSubtagTarget> watchList) =>
string.Join(
"\n",
watchList
.Select(target => string.Join(
"|",
FingerprintPart(target.AlarmFullReference),
FingerprintPart(target.SourceObjectReference),
FingerprintPart(target.ActiveSubtag),
FingerprintPart(target.AckedSubtag),
FingerprintPart(target.AckCommentSubtag),
FingerprintPart(target.PrioritySubtag)))
.Order(StringComparer.Ordinal));
private static string FingerprintPart(string? value)
{
string normalized = value ?? string.Empty;
return $"{normalized.Length}:{normalized}";
}
// Grades one resolution against the effective mode. Returns the reason the subtag coverage is
@@ -684,8 +767,7 @@ public sealed class GatewayAlarmMonitor : BackgroundService, IGatewayAlarmServic
{
reasons.Add(
$"{AlarmProviderReasons.SubtagWatchListStale} ({resolution.Targets.Count} target(s) now "
+ $"resolve against {_sentWatchListReferences.Count} sent at subscribe); restart the alarm "
+ "monitor to re-send it");
+ $"resolve against {_sentWatchListReferences.Count} acknowledged by the worker)");
}
return string.Join("; ", reasons);
@@ -646,6 +646,7 @@ public sealed class GatewayAlarmMonitorProviderModeTests
// The empty list still went to the worker — the verdict is a report, not a behaviour change.
Assert.Empty(sessions.LastSubscribeCommand!.WatchList);
Assert.True(sessions.LastSubscribeCommand.FallbackEnabled);
AlarmProviderStatus status = await ReadFirstProviderStatusAsync(monitor);
@@ -661,16 +662,14 @@ public sealed class GatewayAlarmMonitorProviderModeTests
/// <summary>
/// GWC-37: the failover edge is the moment the watch-list stops being theoretical, so the
/// monitor re-resolves it there and publishes the fresh verdict on the same status frame
/// the failover produces. Discovery that has recovered since <c>SubscribeAlarms</c> is
/// reported as drift rather than applied — the worker rejects a second <c>SubscribeAlarms</c>
/// while subscribed, and the only way to replace the list rebuilds the failover consumer on
/// the primary that just failed. The status names both the fresh count and the sent count so
/// an operator knows a monitor restart will pick the new list up.
/// monitor re-resolves it there, hot-swaps the recovered list through
/// <c>UpdateAlarmWatchList</c>, and publishes the acknowledged verdict on the same status
/// frame the failover produces. The update must precede the status frame so the frame never
/// claims recovered coverage before the worker has accepted it.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task SubtagCoverage_WhenDiscoveryRecoversByFailover_ReResolvesAndReportsDrift()
public async Task SubtagCoverage_WhenDiscoveryRecoversByFailover_UpdatesWorkerBeforePublishingStatus()
{
using GatewayMetrics metrics = new();
FakeSessionManager sessions = new();
@@ -734,13 +733,11 @@ public sealed class GatewayAlarmMonitorProviderModeTests
// The re-resolve happened on the failover edge, before the status was published.
Assert.Equal(2, resolver.Calls);
Assert.Single(sessions.UpdateCommands);
Assert.Equal(2, sessions.UpdateCommands[0].WatchList.Count);
string reason = degraded.ProviderStatus.Reason;
Assert.Contains("alarmmgr failed", reason, StringComparison.Ordinal);
Assert.Contains(AlarmProviderReasons.SubtagWatchListStale, reason, StringComparison.Ordinal);
Assert.Contains("2 target(s) now resolve against 0 sent", reason, StringComparison.Ordinal);
// The stale-list warning replaced the empty/failed pair: discovery works again, so the only
// remaining problem is that the worker is still using the list it was handed.
Assert.DoesNotContain(AlarmProviderReasons.SubtagWatchListStale, reason, StringComparison.Ordinal);
Assert.DoesNotContain(AlarmProviderReasons.SubtagWatchListEmpty, reason, StringComparison.Ordinal);
Assert.DoesNotContain(AlarmProviderReasons.SubtagDiscoveryFailed, reason, StringComparison.Ordinal);
@@ -750,6 +747,69 @@ public sealed class GatewayAlarmMonitorProviderModeTests
await monitor.StopAsync(CancellationToken.None);
}
/// <summary>Periodic reconciliation applies a recovered watch list once and suppresses an unchanged resolution on the next cadence.</summary>
[Fact]
public async Task WatchListRefresh_PeriodicRecovery_SendsExactlyOneMaterialUpdate()
{
using GatewayMetrics metrics = new();
FakeSessionManager sessions = new();
AlarmWatchListResolution recovered = new(
[Target("Galaxy!Area.Tank01.Level.HiHi")],
DiscoveryAttempted: true,
DiscoveryFailed: false,
DiscoveryFailureMessage: null);
ScriptedWatchListResolver resolver = new(DiscoveryFailedResolution(), recovered, recovered);
using GatewayAlarmMonitor monitor = CreateMonitor(sessions, metrics, resolver, reconcileIntervalSeconds: 5);
using CancellationTokenSource cts = new();
await monitor.StartAsync(cts.Token);
await sessions.WaitForUpdateCountAsync(1, WaitTimeout);
await WaitUntilAsync(() => resolver.Calls >= 3, WaitTimeout);
Assert.Single(sessions.UpdateCommands);
Assert.Single(sessions.UpdateCommands[0].WatchList);
await cts.CancelAsync();
await monitor.StopAsync(CancellationToken.None);
}
/// <summary>A rejected refresh keeps the old fingerprint and degraded verdict so the same material change is retried.</summary>
[Fact]
public async Task WatchListRefresh_WorkerRejects_RetainsFingerprintAndRetriesSameTargets()
{
using GatewayMetrics metrics = new();
FakeSessionManager sessions = new();
sessions.UpdateReplyCodes.Enqueue(ProtocolStatusCode.MxaccessFailure);
sessions.UpdateReplyCodes.Enqueue(ProtocolStatusCode.Ok);
AlarmWatchListResolution recovered = new(
[Target("Galaxy!Area.Tank01.Level.HiHi")],
DiscoveryAttempted: true,
DiscoveryFailed: false,
DiscoveryFailureMessage: null);
ScriptedWatchListResolver resolver = new(DiscoveryFailedResolution(), recovered, recovered);
using GatewayAlarmMonitor monitor = CreateMonitor(sessions, metrics, resolver);
using CancellationTokenSource cts = new();
await monitor.StartAsync(cts.Token);
await sessions.WaitForSubscribeAsync(WaitTimeout);
sessions.EmitModeChange(AlarmProviderMode.Subtag, "alarmmgr failed");
await sessions.WaitForUpdateCountAsync(1, WaitTimeout);
AlarmProviderStatus rejected = await ReadFirstProviderStatusAsync(monitor);
Assert.True(rejected.Degraded);
Assert.Contains(AlarmProviderReasons.SubtagWatchListStale, rejected.Reason, StringComparison.Ordinal);
sessions.EmitModeChange(AlarmProviderMode.Subtag, "alarmmgr still failed");
await sessions.WaitForUpdateCountAsync(2, WaitTimeout);
Assert.Equal(
sessions.UpdateCommands[0].WatchList[0].AlarmFullReference,
sessions.UpdateCommands[1].WatchList[0].AlarmFullReference);
await cts.CancelAsync();
await monitor.StopAsync(CancellationToken.None);
}
/// <summary>Reads the provider-status frame every stream opens with, then closes the stream.</summary>
/// <param name="monitor">The monitor to read from.</param>
/// <returns>The provider status carried by the first frame.</returns>
@@ -786,12 +846,14 @@ public sealed class GatewayAlarmMonitorProviderModeTests
private static GatewayAlarmMonitor CreateMonitor(
FakeSessionManager sessions,
GatewayMetrics metrics,
IAlarmWatchListResolver resolver)
IAlarmWatchListResolver resolver,
int reconcileIntervalSeconds = 30)
{
AlarmsOptions options = new()
{
Enabled = true,
SubscriptionExpression = @"\\NODE\Galaxy!Area",
ReconcileIntervalSeconds = reconcileIntervalSeconds,
};
return new GatewayAlarmMonitor(
sessions,
@@ -923,11 +985,34 @@ public sealed class GatewayAlarmMonitorProviderModeTests
/// <summary>The most recent SubscribeAlarms command the monitor sent.</summary>
public SubscribeAlarmsCommand? LastSubscribeCommand { get; private set; }
/// <summary>Gets each live replacement command in invocation order.</summary>
public List<UpdateAlarmWatchListCommand> UpdateCommands { get; } = [];
/// <summary>Scripted protocol statuses for replacement commands.</summary>
public Queue<ProtocolStatusCode> UpdateReplyCodes { get; } = new();
/// <summary>Pushes a worker event onto the monitor's event stream.</summary>
/// <param name="mxEvent">The worker event to push.</param>
public void EmitEvent(MxEvent mxEvent) =>
_events.Writer.TryWrite(new WorkerEvent { Event = mxEvent });
/// <summary>Pushes a provider-mode change onto the worker event stream.</summary>
public void EmitModeChange(AlarmProviderMode mode, string reason) => EmitEvent(new MxEvent
{
OnAlarmProviderModeChanged = new OnAlarmProviderModeChangedEvent
{
Mode = mode,
Reason = reason,
At = Timestamp.FromDateTimeOffset(DateTimeOffset.UtcNow),
},
});
/// <summary>Waits until at least the requested number of update commands have arrived.</summary>
public async Task WaitForUpdateCountAsync(int count, TimeSpan timeout)
{
await WaitUntilAsync(() => UpdateCommands.Count >= count, timeout);
}
/// <summary>Completes once the monitor has issued its SubscribeAlarms command.</summary>
/// <param name="timeout">The maximum time to wait.</param>
/// <returns>A task that represents the asynchronous operation.</returns>
@@ -972,9 +1057,20 @@ public sealed class GatewayAlarmMonitorProviderModeTests
_subscribed.TrySetResult();
}
if (command.Command?.Kind == MxCommandKind.UpdateAlarmWatchList)
{
UpdateCommands.Add(command.Command.UpdateAlarmWatchList.Clone());
}
MxCommandReply reply = new()
{
ProtocolStatus = new ProtocolStatus { Code = ProtocolStatusCode.Ok },
ProtocolStatus = new ProtocolStatus
{
Code = command.Command?.Kind == MxCommandKind.UpdateAlarmWatchList
&& UpdateReplyCodes.TryDequeue(out ProtocolStatusCode updateCode)
? updateCode
: ProtocolStatusCode.Ok,
},
};
if (command.Command?.Kind == MxCommandKind.QueryActiveAlarms)
{
@@ -1507,6 +1507,52 @@ public sealed class ProtobufContractRoundTripTests
Assert.Equal(5, parsed.Failover.FailbackStableProbes);
}
/// <summary>
/// Verifies the live alarm watch-list replacement command and the
/// subscription's explicit empty-list fallback intent retain their
/// command kind, oneof arm, and payload across protobuf serialization.
/// </summary>
[Fact]
public void AlarmWatchListCommands_RoundTripFallbackIntentAndReplacementPayload()
{
var subscribe = new SubscribeAlarmsCommand
{
SubscriptionExpression = @"\\node\Galaxy!Area",
FallbackEnabled = true,
};
var parsedSubscribe = SubscribeAlarmsCommand.Parser.ParseFrom(subscribe.ToByteArray());
Assert.True(parsedSubscribe.FallbackEnabled);
Assert.Empty(parsedSubscribe.WatchList);
var original = new MxCommand
{
Kind = MxCommandKind.UpdateAlarmWatchList,
UpdateAlarmWatchList = new UpdateAlarmWatchListCommand
{
WatchList =
{
new AlarmSubtagTarget
{
AlarmFullReference = "Galaxy!Area.Tank01.Level.HiHi",
SourceObjectReference = "Tank01",
ActiveSubtag = "Tank01.Level.HiHi.InAlarm",
AckedSubtag = "Tank01.Level.HiHi.Acked",
AckCommentSubtag = "Tank01.Level.HiHi.AckMsg",
PrioritySubtag = "Tank01.Level.HiHi.Priority",
},
},
},
};
var parsed = MxCommand.Parser.ParseFrom(original.ToByteArray());
Assert.Equal(original, parsed);
Assert.Equal(MxCommandKind.UpdateAlarmWatchList, parsed.Kind);
Assert.Equal(MxCommand.PayloadOneofCase.UpdateAlarmWatchList, parsed.PayloadCase);
Assert.Single(parsed.UpdateAlarmWatchList.WatchList);
}
/// <summary>
/// Verifies that an <see cref="MxEvent"/> carrying an
/// <see cref="OnAlarmProviderModeChangedEvent"/> body (the
@@ -91,6 +91,27 @@ public sealed class AlarmCommandExecutorTests
Assert.Equal(ProtocolStatusCode.InvalidRequest, reply.ProtocolStatus.Code);
}
/// <summary>Verifies that a live watch-list replacement routes to the alarm handler on the executor thread.</summary>
[Fact]
public void UpdateAlarmWatchList_WithHandler_RoutesToHandlerAndReturnsOk()
{
FakeAlarmHandler handler = new FakeAlarmHandler();
MxAccessCommandExecutor executor = NewExecutor(handler);
var target = new AlarmSubtagTarget { AlarmFullReference = "Galaxy!Area.Tank01.Alarm" };
MxCommandReply reply = executor.Execute(new StaCommand(
SessionId,
CorrelationId,
new MxCommand
{
Kind = MxCommandKind.UpdateAlarmWatchList,
UpdateAlarmWatchList = new UpdateAlarmWatchListCommand { WatchList = { target } },
}));
Assert.Equal(ProtocolStatusCode.Ok, reply.ProtocolStatus.Code);
Assert.Equal(target, Assert.Single(handler.LastWatchList!));
}
/// <summary>Verifies that acknowledge routes native status into hresult and payload.</summary>
[Fact]
public void AcknowledgeAlarm_WithHandler_RoutesNativeStatusIntoHresultAndPayload()
@@ -447,6 +468,15 @@ public sealed class AlarmCommandExecutorTests
LastSessionId = sessionId;
}
/// <summary>Gets the most recent replacement watch list.</summary>
public IReadOnlyList<AlarmSubtagTarget>? LastWatchList { get; private set; }
/// <inheritdoc />
public void UpdateWatchList(IReadOnlyList<AlarmSubtagTarget> watchList)
{
LastWatchList = watchList;
}
/// <inheritdoc />
public void Unsubscribe()
{
@@ -287,8 +287,12 @@ public sealed class AlarmCommandHandlerTests
handler.PollOnce();
Assert.Equal(5, guardInvocations);
handler.Unsubscribe();
Assert.Throws<InvalidOperationException>(
() => handler.UpdateWatchList(Array.Empty<AlarmSubtagTarget>()));
Assert.Equal(6, guardInvocations);
handler.Unsubscribe();
Assert.Equal(7, guardInvocations);
}
/// <summary>
@@ -324,6 +328,8 @@ public sealed class AlarmCommandHandlerTests
() => handler.AcknowledgeByName("", "", "", "", "", "", "", ""));
Assert.Throws<InvalidOperationException>(() => handler.QueryActive(null, out _));
Assert.Throws<InvalidOperationException>(() => handler.PollOnce());
Assert.Throws<InvalidOperationException>(
() => handler.UpdateWatchList(Array.Empty<AlarmSubtagTarget>()));
Assert.Throws<InvalidOperationException>(() => handler.Unsubscribe());
}
@@ -412,6 +418,114 @@ public sealed class AlarmCommandHandlerTests
Assert.Equal(AlarmProviderMode.Subtag, evt.OnAlarmProviderModeChanged.Mode);
}
/// <summary>An explicitly fallback-capable auto subscription builds a replaceable standby even when discovery is initially empty.</summary>
[Fact]
public void Subscribe_AutoModeWithEmptyWatchListAndFallbackEnabled_BuildsReplaceableStandby()
{
FakeConsumer primary = new FakeConsumer { AcknowledgeReturn = 11 };
FakeConsumer initialStandby = new FakeConsumer { AcknowledgeReturn = 22 };
FakeConsumer replacement = new FakeConsumer { AcknowledgeReturn = 33 };
int buildCount = 0;
AlarmCommandHandler handler = new AlarmCommandHandler(
new MxAccessEventQueue(),
() => primary,
threadAffinityCheck: null,
comFactory: null,
standbyFactory: _ => ++buildCount == 1 ? initialStandby : replacement);
handler.Subscribe(new SubscribeAlarmsCommand
{
SubscriptionExpression = @"\\HOST\Galaxy!Area",
FallbackEnabled = true,
}, "s1");
handler.UpdateWatchList(new[] { NewTarget("Galaxy!Area.Tank01.Alarm") });
Assert.Equal(2, buildCount);
Assert.Equal(@"\\HOST\Galaxy!Area", replacement.LastSubscription);
Assert.True(initialStandby.Disposed);
Assert.Equal(11, handler.Acknowledge(Guid.Empty, "", "", "", "", ""));
}
/// <summary>A replacement is rejected when no fallback-capable subscription is active.</summary>
[Fact]
public void UpdateWatchList_WithoutFallbackCapableSubscription_Throws()
{
AlarmCommandHandler handler = new AlarmCommandHandler(
new MxAccessEventQueue(),
() => new FakeConsumer());
Assert.Throws<InvalidOperationException>(
() => handler.UpdateWatchList(new[] { NewTarget("Galaxy!Area.Tank01.Alarm") }));
handler.Subscribe(new SubscribeAlarmsCommand
{
SubscriptionExpression = @"\\HOST\Galaxy!Area",
ForcedMode = AlarmProviderMode.Alarmmgr,
}, "s1");
Assert.Throws<InvalidOperationException>(
() => handler.UpdateWatchList(new[] { NewTarget("Galaxy!Area.Tank02.Alarm") }));
}
/// <summary>A replacement setup failure leaves the old standby and primary mode intact.</summary>
[Fact]
public void UpdateWatchList_WhenReplacementSubscribeFails_PreservesOldStandbyAndPrimaryMode()
{
FakeConsumer primary = new FakeConsumer { AcknowledgeReturn = 11 };
FakeConsumer oldStandby = new FakeConsumer { AcknowledgeReturn = 22 };
FakeConsumer failedReplacement = new FakeConsumer { ThrowOnSubscribe = true, AcknowledgeReturn = 33 };
int buildCount = 0;
AlarmCommandHandler handler = new AlarmCommandHandler(
new MxAccessEventQueue(),
() => primary,
null,
null,
_ => ++buildCount == 1 ? oldStandby : failedReplacement);
SubscribeAlarmsCommand subscribe = new SubscribeAlarmsCommand
{
SubscriptionExpression = @"\\HOST\Galaxy!Area",
FallbackEnabled = true,
};
subscribe.WatchList.Add(NewTarget("Galaxy!Area.Tank01.Alarm"));
handler.Subscribe(subscribe, "s1");
Assert.Throws<InvalidOperationException>(
() => handler.UpdateWatchList(new[] { NewTarget("Galaxy!Area.Tank02.Alarm") }));
Assert.False(oldStandby.Disposed);
Assert.True(failedReplacement.Disposed);
Assert.Equal(11, handler.Acknowledge(Guid.Empty, "", "", "", "", ""));
}
/// <summary>A successful replacement while failed over keeps subtag mode active and delegates to the new standby.</summary>
[Fact]
public void UpdateWatchList_WhileSubtagActive_PreservesModeAndUsesReplacement()
{
FakeConsumer primary = new FakeConsumer { ThrowOnSubscribe = true, AcknowledgeReturn = 11 };
FakeConsumer oldStandby = new FakeConsumer { AcknowledgeReturn = 22 };
FakeConsumer replacement = new FakeConsumer { AcknowledgeReturn = 33 };
int buildCount = 0;
AlarmCommandHandler handler = new AlarmCommandHandler(
new MxAccessEventQueue(),
() => primary,
null,
null,
_ => ++buildCount == 1 ? oldStandby : replacement);
SubscribeAlarmsCommand subscribe = new SubscribeAlarmsCommand
{
SubscriptionExpression = @"\\HOST\Galaxy!Area",
FallbackEnabled = true,
Failover = new AlarmFailoverConfig { ConsecutiveFailureThreshold = 1 },
};
subscribe.WatchList.Add(NewTarget("Galaxy!Area.Tank01.Alarm"));
handler.Subscribe(subscribe, "s1");
handler.UpdateWatchList(new[] { NewTarget("Galaxy!Area.Tank02.Alarm") });
Assert.True(oldStandby.Disposed);
Assert.Equal(33, handler.Acknowledge(Guid.Empty, "", "", "", "", ""));
}
/// <summary>
/// A non-failover subscribe (alarmmgr-only) never enqueues a
/// provider-mode-changed event, and a subsequent Unsubscribe detaches
@@ -467,6 +581,14 @@ public sealed class AlarmCommandHandlerTests
};
}
private static AlarmSubtagTarget NewTarget(string alarmFullReference) => new AlarmSubtagTarget
{
AlarmFullReference = alarmFullReference,
ActiveSubtag = alarmFullReference + ".InAlarm",
AckedSubtag = alarmFullReference + ".Acked",
AckCommentSubtag = alarmFullReference + ".AckMsg",
};
private sealed class FakeConsumer : IMxAccessAlarmConsumer
{
#pragma warning disable CS0067 // Event never invoked — fake; AlarmCommandHandler tests don't drive transitions.
@@ -651,6 +651,11 @@ public sealed class MxAccessStaSessionTests
LastSubscription = command.SubscriptionExpression;
}
/// <inheritdoc />
public void UpdateWatchList(IReadOnlyList<AlarmSubtagTarget> watchList)
{
}
/// <inheritdoc />
public void Unsubscribe()
{
@@ -45,6 +45,7 @@ public sealed class AlarmCommandHandler : IAlarmCommandHandler
private FailoverAlarmConsumer? failoverConsumer;
private EventHandler<AlarmProviderModeChange>? providerModeChangedHandler;
private string subscribeSessionId = string.Empty;
private string subscriptionExpression = string.Empty;
private bool disposed;
/// <summary>Initializes a new alarm command handler with the given event queue.</summary>
@@ -153,6 +154,7 @@ public sealed class AlarmCommandHandler : IAlarmCommandHandler
}
subscribeSessionId = sessionId ?? string.Empty;
subscriptionExpression = command.SubscriptionExpression ?? string.Empty;
IMxAccessAlarmConsumer consumer = BuildConsumer(command)
?? throw new InvalidOperationException("Alarm consumer factory returned null.");
@@ -171,7 +173,7 @@ public sealed class AlarmCommandHandler : IAlarmCommandHandler
dispatcher = new AlarmDispatcher(consumer, sink, subscribeSessionId);
try
{
dispatcher.Subscribe(command.SubscriptionExpression ?? string.Empty);
dispatcher.Subscribe(subscriptionExpression);
}
catch
{
@@ -212,7 +214,8 @@ public sealed class AlarmCommandHandler : IAlarmCommandHandler
return BuildStandby(watchList);
}
if (command.ForcedMode == AlarmProviderMode.Unspecified && watchList.Count > 0)
if (command.ForcedMode == AlarmProviderMode.Unspecified
&& (command.FallbackEnabled || watchList.Count > 0))
{
IMxAccessAlarmConsumer primary = consumerFactory()
?? throw new InvalidOperationException("Alarm consumer factory returned null.");
@@ -230,6 +233,26 @@ public sealed class AlarmCommandHandler : IAlarmCommandHandler
?? throw new InvalidOperationException("Alarm consumer factory returned null.");
}
/// <inheritdoc />
public void UpdateWatchList(IReadOnlyList<AlarmSubtagTarget> watchList)
{
if (disposed) throw new ObjectDisposedException(nameof(AlarmCommandHandler));
if (watchList is null) throw new ArgumentNullException(nameof(watchList));
threadAffinityCheck?.Invoke();
lock (syncRoot)
{
if (dispatcher is null || failoverConsumer is null)
{
throw new InvalidOperationException(
"UpdateAlarmWatchList requires an active fallback-capable alarm subscription.");
}
IMxAccessAlarmConsumer replacement = BuildStandby(watchList);
failoverConsumer.ReplaceStandby(replacement, subscriptionExpression);
}
}
private IMxAccessAlarmConsumer BuildStandby(IReadOnlyList<AlarmSubtagTarget> watchList)
{
if (standbyFactory is not null)
@@ -276,6 +299,7 @@ public sealed class AlarmCommandHandler : IAlarmCommandHandler
DetachProviderModeChanged();
toDispose = dispatcher;
dispatcher = null;
subscriptionExpression = string.Empty;
}
toDispose?.Dispose();
}
@@ -55,7 +55,7 @@ public sealed class FailoverAlarmConsumer : IMxAccessAlarmConsumer
}
private readonly IMxAccessAlarmConsumer primary;
private readonly IMxAccessAlarmConsumer standby;
private IMxAccessAlarmConsumer standby;
private readonly FailoverSettings settings;
private Active active = Active.Primary;
@@ -103,6 +103,36 @@ public sealed class FailoverAlarmConsumer : IMxAccessAlarmConsumer
/// </summary>
public AlarmProviderMode Mode => mode;
/// <summary>
/// Arms and attaches a replacement standby before atomically swapping
/// it into the current provider mode. Setup failure disposes only the
/// replacement and leaves the existing standby untouched.
/// </summary>
/// <param name="replacement">The newly built standby consumer.</param>
/// <param name="subscription">The retained subscription expression.</param>
public void ReplaceStandby(IMxAccessAlarmConsumer replacement, string subscription)
{
if (disposed) throw new ObjectDisposedException(nameof(FailoverAlarmConsumer));
if (replacement is null) throw new ArgumentNullException(nameof(replacement));
try
{
replacement.Subscribe(subscription ?? string.Empty);
replacement.AlarmTransitionEmitted += OnChildTransition;
}
catch
{
try { replacement.Dispose(); } catch { /* preserve setup failure */ }
throw;
}
IMxAccessAlarmConsumer previous = standby;
standby = replacement;
try { previous.AlarmTransitionEmitted -= OnChildTransition; } catch { /* best effort */ }
try { previous.Dispose(); } catch { /* replacement is already active */ }
}
/// <inheritdoc />
public void Subscribe(string subscription)
{
@@ -7,7 +7,8 @@ namespace ZB.MOM.WW.MxGateway.Worker.MxAccess;
/// <summary>
/// Per-session interface routing the worker's alarm IPC commands —
/// <c>SubscribeAlarmsCommand</c>, <c>AcknowledgeAlarmCommand</c>,
/// <c>QueryActiveAlarmsCommand</c>, <c>UnsubscribeAlarmsCommand</c> —
/// <c>UpdateAlarmWatchListCommand</c>, <c>QueryActiveAlarmsCommand</c>,
/// <c>UnsubscribeAlarmsCommand</c> —
/// to the underlying <see cref="AlarmDispatcher"/>. Production binding
/// is <see cref="AlarmCommandHandler"/>; tests substitute a fake.
/// </summary>
@@ -24,6 +25,13 @@ public interface IAlarmCommandHandler : IDisposable
/// <param name="sessionId">The session identifier.</param>
void Subscribe(SubscribeAlarmsCommand command, string sessionId);
/// <summary>
/// Replace the active auto-failover subscription's subtag standby
/// targets without restarting its primary provider.
/// </summary>
/// <param name="watchList">The complete replacement watch list.</param>
void UpdateWatchList(IReadOnlyList<AlarmSubtagTarget> watchList);
/// <summary>Tear down the active subscription. No-op if not subscribed.</summary>
void Unsubscribe();
@@ -158,6 +158,7 @@ public sealed class MxAccessCommandExecutor : IStaCommandExecutor
MxCommandKind.WriteSecured2Bulk => ExecuteWriteSecured2Bulk(command),
MxCommandKind.ReadBulk => ExecuteReadBulk(command),
MxCommandKind.SubscribeAlarms => ExecuteSubscribeAlarms(command),
MxCommandKind.UpdateAlarmWatchList => ExecuteUpdateAlarmWatchList(command),
MxCommandKind.UnsubscribeAlarms => ExecuteUnsubscribeAlarms(command),
MxCommandKind.AcknowledgeAlarm => ExecuteAcknowledgeAlarm(command),
MxCommandKind.AcknowledgeAlarmByName => ExecuteAcknowledgeAlarmByName(command),
@@ -866,6 +867,30 @@ public sealed class MxAccessCommandExecutor : IStaCommandExecutor
}
}
private MxCommandReply ExecuteUpdateAlarmWatchList(StaCommand command)
{
if (command.Command.PayloadCase != MxCommand.PayloadOneofCase.UpdateAlarmWatchList)
{
return CreateInvalidRequestReply(command, "UpdateAlarmWatchList command payload is required.");
}
if (alarmCommandHandler is null)
{
return CreateInvalidRequestReply(
command,
"UpdateAlarmWatchList requires an alarm command handler; the worker was constructed without one.");
}
try
{
alarmCommandHandler.UpdateWatchList(command.Command.UpdateAlarmWatchList.WatchList);
return CreateOkReply(command);
}
catch (Exception ex)
{
return CreateAlarmFailureReply(command, ex);
}
}
private MxCommandReply ExecuteAcknowledgeAlarm(StaCommand command)
{
if (command.Command.PayloadCase != MxCommand.PayloadOneofCase.AcknowledgeAlarmCommand)