dohertj2/mxaccessgw
1
0
Fork 0
Code Issues 14 Pull Requests 2 Actions Packages Projects Releases Wiki Activity

feat(alarms): structural degraded-status signal for truncated alarm snapshots

Browse Source 
The truncation-cliff fix made alarm transitions truncation-safe but silent:
when GetXmlCurrentAlarms2 returns exactly maxAlmCnt records the worker
suppresses absence-implies-Clear inference and says so only in a rate-limited
stderr warning. No client and no operator could tell a complete active set
from a capped one.

Two additive proto3 booleans carry the verdict out:

- QueryActiveAlarmsReplyPayload.snapshot_truncated = 2 (worker IPC reply)
- ActiveAlarmSnapshot.from_truncated_snapshot = 16 (per record)

The per-record field is not an aesthetic choice. QueryActiveAlarms returns a
bare `stream ActiveAlarmSnapshot` with no envelope, header, or trailer, so a
per-record boolean is the only carrier that stays wire-compatible; an envelope
message would change every existing client's stream element type. The reply
payload states it too because a prefix filter can leave zero records and a
truncated fetch with nothing to report still has to say so. The flag means
"this set may be incomplete", never "this record is unreliable" — it is
independent of the subtag-fallback `degraded` field.

Detection is deliberately UNCHANGED: IsTruncatedFetch remains
`fetchedRecordCount >= maxAlarmsPerFetch`. The live probe (docs/AlarmProbeFindings.md,
ce5d8ae) could not verify whether ALARM_RECORDS/@COUNT reports the total active
count or only the records in the reply, so @COUNT is not parsed for detection;
switching to it stays blocked on probe evidence. The probe's comment
annotations in WnWrapAlarmConsumer.cs are preserved.

Reset semantics: not latched. WnWrapAlarmConsumer.FoldFetch replaces the
verdict on every poll under the same lock as the snapshot merge, so the first
sub-cap fetch clears it; GatewayAlarmMonitor.ClearCache drops it with the cache
generation it describes. A caveat that never turns off is one operators learn
to ignore.

Flow: WnWrapAlarmConsumer.LastSnapshotTruncated -> AlarmDispatcher (stamps every
record) / IAlarmCommandHandler (payload) -> MxAccessCommandExecutor reply ->
GatewayAlarmMonitor._snapshotTruncated -> IGatewayAlarmService.SnapshotTruncated
-> DashboardAlarmQueryResult -> AlarmsPage warning banner (render-side only; the
poll loop and DisposeAsync drain are untouched). The public QueryActiveAlarms
RPC forwards worker snapshots unmodified, so the per-record flag needed no
mapper change — a test pins that.

Parity: this describes OUR fetch mechanics — additive gateway metadata — not
MXAccess provider behavior. No event is synthesized and no MXAccess-observable
semantics change, so it is not a parity deviation.

Tests: worker LastSnapshotTruncated set/reset/consecutive-burst (windev-run);
gateway end-to-end truncated reply -> monitor -> public stream, with the
complete-reply control as the load-bearing assertion; AlarmsPage banner
present/absent. Docs: gateway.md alarm surface, docs/DesignDecisions.md entry.
This commit is contained in:
Joseph Doherty
2026-08-17 04:18:34 -04:00
parent b8b7b69ba0
commit 693a78db7d
41 changed files with 2217 additions and 309 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ZB.MOM.WW.MxGateway.Tests/Alarms/AlarmTruncationSignalTests.cs
+329
View File
@@ -0,0 +1,329 @@
using System.Diagnostics.CodeAnalysis;
using System.Runtime.CompilerServices;
using System.Threading.Channels;
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.Security.Authorization;
using ZB.MOM.WW.MxGateway.Server.Sessions;
using ZB.MOM.WW.MxGateway.Tests.TestSupport;
namespace ZB.MOM.WW.MxGateway.Tests.Alarms;
/// <summary>
/// Carries the worker's truncated-fetch verdict across the gateway: worker
/// reply payload → <see cref="GatewayAlarmMonitor"/> → the public
/// <c>QueryActiveAlarms</c> stream.
/// </summary>
/// <remarks>
/// <para>
/// The truncation guard itself (the worker merging rather than replacing
/// a capped snapshot) is already covered in the worker suite. What was
/// missing is that the guard is <em>silent</em>: a capped fetch suppresses
/// absence-implies-Clear inference and says so only in a rate-limited
/// stderr warning, so a consumer of the alarm surface could not tell a
/// complete active set from a capped one. These tests pin the structural
/// signal that replaces the guesswork.
/// </para>
/// <para>
/// The load-bearing assertion is the <em>false</em> one
/// (<see cref="QueryActiveAlarms_WithCompleteWorkerReply_LeavesFlagUnset"/>).
/// "Truncated reply sets the flag" would also pass against a field
/// hard-wired to true; only the complete-reply case proves the flag is
/// actually derived from the worker's verdict.
/// </para>
/// </remarks>
public sealed class AlarmTruncationSignalTests
{
private static readonly TimeSpan WaitTimeout = TimeSpan.FromSeconds(15);
/// <summary>
/// A capped worker reply sets the monitor's completeness caveat and
/// stamps every cached snapshot, so both the dashboard (which reads the
/// service flag) and the RPC (which reads the records) can surface it.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task Reconcile_WithTruncatedWorkerReply_SurfacesTheFlagOnMonitorAndSnapshots()
{
using GatewayMetrics metrics = new();
StubSessionManager sessions = new()
{
SnapshotTruncated = true,
Snapshots = [NewSnapshot("Galaxy!Area.Tank01.Level.HiHi", fromTruncatedSnapshot: true)],
};
using GatewayAlarmMonitor monitor = CreateMonitor(sessions, metrics);
using CancellationTokenSource cts = new();
await monitor.StartAsync(cts.Token);
await sessions.WaitForReconcileAsync(WaitTimeout);
await WaitUntilAsync(() => monitor.CurrentAlarms.Count == 1, WaitTimeout);
Assert.True(monitor.SnapshotTruncated);
ActiveAlarmSnapshot cached = Assert.Single(monitor.CurrentAlarms);
Assert.True(cached.FromTruncatedSnapshot);
// Truncation is about the completeness of the SET, not the fidelity of
// the record — the subtag-fallback flag must stay independent of it.
Assert.False(cached.Degraded);
await cts.CancelAsync();
await monitor.StopAsync(CancellationToken.None);
}
/// <summary>
/// The public <c>QueryActiveAlarms</c> stream carries the per-record flag
/// through untouched. That RPC returns a bare
/// <c>stream ActiveAlarmSnapshot</c> with no envelope message, so the
/// per-record boolean is the only place set-level degraded status can
/// ride — if the service ever starts re-projecting records instead of
/// forwarding them, this is what catches the dropped field.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task QueryActiveAlarms_WithTruncatedSnapshot_StreamsTheFlagToTheClient()
{
FakeGatewayAlarmService alarms = new()
{
SnapshotTruncated = true,
CurrentAlarms = [NewSnapshot("Galaxy!Area.Tank01.Level.HiHi", fromTruncatedSnapshot: true)],
};
MxAccessGatewayService service = CreateService(alarms);
RecordingServerStreamWriter<ActiveAlarmSnapshot> sink = new();
await service.QueryActiveAlarms(
new QueryActiveAlarmsRequest(),
sink,
new TestServerCallContext());
ActiveAlarmSnapshot streamed = Assert.Single(sink.Messages);
Assert.True(streamed.FromTruncatedSnapshot);
Assert.Equal("Galaxy!Area.Tank01.Level.HiHi", streamed.AlarmFullReference);
}
/// <summary>
/// The control. A complete worker reply must leave both the monitor flag
/// and the streamed records unset — otherwise every snapshot would read
/// as possibly-incomplete and the signal would carry no information.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task QueryActiveAlarms_WithCompleteWorkerReply_LeavesFlagUnset()
{
using GatewayMetrics metrics = new();
StubSessionManager sessions = new()
{
SnapshotTruncated = false,
Snapshots = [NewSnapshot("Galaxy!Area.Tank01.Level.HiHi", fromTruncatedSnapshot: false)],
};
using GatewayAlarmMonitor monitor = CreateMonitor(sessions, metrics);
using CancellationTokenSource cts = new();
await monitor.StartAsync(cts.Token);
await sessions.WaitForReconcileAsync(WaitTimeout);
await WaitUntilAsync(() => monitor.CurrentAlarms.Count == 1, WaitTimeout);
Assert.False(monitor.SnapshotTruncated);
MxAccessGatewayService service = CreateService(new FakeGatewayAlarmService
{
SnapshotTruncated = monitor.SnapshotTruncated,
CurrentAlarms = monitor.CurrentAlarms,
});
RecordingServerStreamWriter<ActiveAlarmSnapshot> sink = new();
await service.QueryActiveAlarms(
new QueryActiveAlarmsRequest(),
sink,
new TestServerCallContext());
Assert.False(Assert.Single(sink.Messages).FromTruncatedSnapshot);
await cts.CancelAsync();
await monitor.StopAsync(CancellationToken.None);
}
private static ActiveAlarmSnapshot NewSnapshot(string reference, bool fromTruncatedSnapshot)
{
return new ActiveAlarmSnapshot
{
AlarmFullReference = reference,
SourceObjectReference = "Tank01.Level",
AlarmTypeName = "HiHi",
Category = "Area",
Severity = 500,
CurrentState = AlarmConditionState.Active,
SourceProvider = AlarmProviderMode.Alarmmgr,
FromTruncatedSnapshot = fromTruncatedSnapshot,
};
}
private static GatewayAlarmMonitor CreateMonitor(StubSessionManager 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);
}
private static MxAccessGatewayService CreateService(FakeGatewayAlarmService alarms)
{
StubSessionManager sessions = new();
return new MxAccessGatewayService(
sessions,
new GatewayRequestIdentityAccessor(),
new AllowAllConstraintEnforcer(),
new MxAccessGrpcRequestValidator(),
new MxAccessGrpcMapper(),
new StubEventStreamService(),
new GatewayMetrics(),
NullLogger<MxAccessGatewayService>.Instance,
alarms);
}
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>
/// Minimal <see cref="ISessionManager"/> that answers the monitor's
/// QueryActiveAlarms with a scripted reply payload — the seam this suite
/// drives the truncation verdict through.
/// </summary>
private sealed class StubSessionManager : ISessionManager
{
private readonly Channel<WorkerEvent> _events = Channel.CreateUnbounded<WorkerEvent>();
private readonly TaskCompletionSource _reconciled =
new(TaskCreationOptions.RunContinuationsAsynchronously);
/// <summary>Gets or sets the truncation verdict the scripted reply carries.</summary>
public bool SnapshotTruncated { get; init; }
/// <summary>Gets or sets the snapshots the scripted reply carries.</summary>
public IReadOnlyList<ActiveAlarmSnapshot> Snapshots { get; init; } = [];
/// <summary>Completes once the monitor has issued its first QueryActiveAlarms.</summary>
/// <param name="timeout">The maximum time to wait.</param>
/// <returns>A task that represents the asynchronous operation.</returns>
public Task WaitForReconcileAsync(TimeSpan timeout) => _reconciled.Task.WaitAsync(timeout);
/// <inheritdoc />
public Task<GatewaySession> OpenSessionAsync(
SessionOpenRequest request,
string? clientIdentity,
string? ownerKeyId,
CancellationToken cancellationToken)
{
GatewaySession session = new(
Guid.NewGuid().ToString("N"),
"Galaxy",
"pipe-test",
"nonce-test",
clientIdentity,
null,
null,
TimeSpan.FromSeconds(30),
TimeSpan.FromSeconds(30),
TimeSpan.FromSeconds(30),
DateTimeOffset.UtcNow);
session.AttachWorkerClient(new ChannelWorkerClient(session.SessionId, _events.Reader));
session.MarkReady();
return Task.FromResult(session);
}
/// <inheritdoc />
public Task<WorkerCommandReply> InvokeAsync(
string sessionId,
WorkerCommand command,
CancellationToken cancellationToken)
{
MxCommandReply reply = new()
{
ProtocolStatus = new ProtocolStatus { Code = ProtocolStatusCode.Ok },
};
if (command.Command?.Kind == MxCommandKind.QueryActiveAlarms)
{
QueryActiveAlarmsReplyPayload payload = new() { SnapshotTruncated = SnapshotTruncated };
payload.Snapshots.AddRange(Snapshots.Select(snapshot => snapshot.Clone()));
reply.QueryActiveAlarms = payload;
_reconciled.TrySetResult();
}
return Task.FromResult(new WorkerCommandReply { Reply = reply });
}
/// <inheritdoc />
public bool TryGetSession(string sessionId, [MaybeNullWhen(false)] out GatewaySession session)
{
session = null;
return false;
}
/// <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>
/// <see cref="IEventStreamService"/> stub — QueryActiveAlarms never
/// touches the event path, but the service constructor requires one.
/// </summary>
private sealed class StubEventStreamService : IEventStreamService
{
/// <inheritdoc />
public async IAsyncEnumerable<MxEvent> StreamEventsAsync(
StreamEventsRequest request,
string? callerKeyId,
[EnumeratorCancellation] CancellationToken cancellationToken)
{
await Task.CompletedTask.ConfigureAwait(false);
yield break;
}
}
}