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

feat(dashboard): alarms page consumes snapshot_status feed frame for the truncation banner

Browse Source 
The truncated-snapshot caveat moved from poll-only to push-driven. The page
already held an in-process alarm-feed subscription for the provider badge; it
now also handles the feed's snapshot_status frame, so a capped provider fetch is
caveated when the monitor decides it rather than up to three seconds later.

The poll's assignment stays as the reconcile baseline — both sources read the
same monitor verdict, and the frame is consumed, never synthesized page-side.
StreamAsync primes every subscriber with a snapshot_status frame at open, so a
page attaching mid-truncation needs no priming logic of its own; the loop is
renamed StatusFeedLoopAsync because it now feeds two indicators, not one.
This commit is contained in:
Joseph Doherty
2026-08-18 06:43:23 -04:00
parent 2b1efb5e50
commit 7b6dfba654
3 changed files with 266 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ZB.MOM.WW.MxGateway.Server/Dashboard/Components/Pages/AlarmsPage.razor
+39 -15
View File
@@ -191,13 +191,13 @@
private Task? _pollTask;
private DashboardAlarmProviderStatus _providerStatus = DashboardAlarmProviderStatus.Healthy;
private Task? _providerStatusTask;
private Task? _statusFeedTask;
/// <inheritdoc />
protected override void OnInitialized()
{
_pollTask = PollLoopAsync();
_providerStatusTask = ProviderStatusLoopAsync();
_statusFeedTask = StatusFeedLoopAsync();
}
private string? ProviderStatusTitle()
@@ -210,8 +210,13 @@
// The badge tracks the central monitor directly rather than looping back through
// /hubs/alarms: the alarm service is an in-process multi-subscriber fan-out, so a
// server-rendered page needs no SignalR client, no loopback socket and no auth token.
// Alarm rows still come from the 3-second poll below — this loop only feeds the badge.
private async Task ProviderStatusLoopAsync()
// This loop feeds the two gateway-status indicators — the provider badge and the
// truncation banner — from the feed's own status frames, so both move as soon as the
// monitor's verdict changes instead of on the next 3-second tick. Alarm rows still come
// from the poll below, which also re-asserts the truncation verdict as its reconcile
// baseline: both sources read the same monitor verdict, so they cannot disagree for
// longer than one tick, and neither one is synthesized here.
private async Task StatusFeedLoopAsync()
{
while (!_cts.IsCancellationRequested)
{
@@ -221,16 +226,30 @@
.StreamAsync(alarmFilterPrefix: null, _cts.Token)
.ConfigureAwait(false))
{
if (message.PayloadCase != AlarmFeedMessage.PayloadOneofCase.ProviderStatus)
switch (message.PayloadCase)
{
continue;
}
case AlarmFeedMessage.PayloadOneofCase.ProviderStatus:
await InvokeAsync(() =>
{
_providerStatus = DashboardAlarmProviderStatus.FromFeed(message);
StateHasChanged();
}).ConfigureAwait(false);
break;
await InvokeAsync(() =>
{
_providerStatus = DashboardAlarmProviderStatus.FromFeed(message);
StateHasChanged();
}).ConfigureAwait(false);
// Every subscriber is primed with this frame at open, so a page that
// attaches mid-truncation gets the caveat without waiting for an edge —
// no page-side priming needed.
case AlarmFeedMessage.PayloadOneofCase.SnapshotStatus:
await InvokeAsync(() =>
{
_snapshotTruncated = message.SnapshotStatus.Truncated;
StateHasChanged();
}).ConfigureAwait(false);
break;
default:
break;
}
}
}
catch (OperationCanceledException)
@@ -241,7 +260,8 @@
{
// The monitor completes a subscriber's stream when it falls behind, and
// again when the monitor restarts. Both are recoverable by resubscribing;
// the badge holds its last value in the meantime.
// the badge and banner hold their last values in the meantime, and the
// resubscribe is primed with the current ones.
}
try
@@ -321,7 +341,7 @@
};
}
// Fault handling sits inside the loop, matching ProviderStatusLoopAsync: a query or render
// Fault handling sits inside the loop, matching StatusFeedLoopAsync: a query or render
// fault on one tick is transient (a provider blip, a momentarily unavailable session), so it
// is surfaced on the page and retried on the next tick rather than ending polling for the
// life of the page. Cancellation is the only exit. The loop method itself therefore cannot
@@ -398,6 +418,10 @@
{
DashboardAlarmQueryResult result = await LiveData.QueryAlarmsAsync(_cts.Token);
_queryError = result.Error;
// Kept alongside the feed's snapshot_status frame rather than replaced by it: this is
// the reconcile baseline. Both read the same monitor verdict, so the poll can only
// confirm what the frame already showed — but it also re-establishes the banner for a
// page whose feed subscription is mid-resubscribe after the monitor dropped it.
_snapshotTruncated = result.SnapshotTruncated;
_workerPid = result.WorkerProcessId;
_lastRefresh = DateTimeOffset.UtcNow;
@@ -416,7 +440,7 @@
// Drained together, not one after the other: the wedged dispatcher this bound exists
// for blocks both loops at once, so sequential drains would time out twice and make
// the real bound 10 seconds. DrainAsync tolerates a null task.
await Task.WhenAll(DrainAsync(_pollTask), DrainAsync(_providerStatusTask))
await Task.WhenAll(DrainAsync(_pollTask), DrainAsync(_statusFeedTask))
.ConfigureAwait(false);
_cts.Dispose();
src/ZB.MOM.WW.MxGateway.Tests/Dashboard/AlarmsPageTruncationBannerTests.cs
+209
View File
@@ -1,7 +1,11 @@
using System.Diagnostics;
using System.Runtime.CompilerServices;
using System.Threading.Channels;
using Microsoft.AspNetCore.Components.Web.HtmlRendering;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
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.Dashboard;
@@ -36,6 +40,8 @@ public sealed class AlarmsPageTruncationBannerTests
{
private const string BannerMarker = "Alarm snapshot may be incomplete";
private static readonly TimeSpan RenderWaitTimeout = TimeSpan.FromSeconds(10);
/// <summary>A capped provider fetch puts the completeness caveat on the page.</summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
@@ -61,6 +67,120 @@ public sealed class AlarmsPageTruncationBannerTests
Assert.Contains("Active Alarms", html, StringComparison.Ordinal);
}
/// <summary>
/// The push path. A <c>snapshot_status</c> frame arriving on the page's
/// in-process alarm-feed subscription raises the caveat on its own, with
/// no poll tick behind it — the poll is a 3-second reconcile baseline, and
/// an operator should not stare at an un-caveated alarm list for up to
/// three seconds after the gateway has already decided the set is capped.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task AlarmsPage_WhenFeedPushesTruncated_RaisesTheBannerWithoutAPollTick()
{
ScriptedAlarmFeed feed = new();
await using ServiceProvider provider = BuildPushServices(feed);
await using HtmlRenderer renderer = new(
provider,
provider.GetRequiredService<ILoggerFactory>());
HtmlRootComponent page = await renderer.Dispatcher.InvokeAsync(
() => renderer.RenderComponentAsync<AlarmsPage>());
// The one poll answer this page will ever get said "complete", so everything
// the banner does from here is the feed's doing.
Assert.DoesNotContain(BannerMarker, await HtmlAsync(renderer, page), StringComparison.Ordinal);
await feed.PushAsync(SnapshotStatusFrame(truncated: true));
await WaitForHtmlAsync(
renderer,
page,
html => html.Contains(BannerMarker, StringComparison.Ordinal),
"banner to appear after a truncated snapshot_status frame");
}
/// <summary>
/// The clearing edge, page-side. Absence authority comes back when the
/// gateway says so; a banner that only ever went up would caveat the alarm
/// list for the life of the circuit.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task AlarmsPage_WhenFeedPushesComplete_ClearsTheBannerWithoutAPollTick()
{
ScriptedAlarmFeed feed = new();
await using ServiceProvider provider = BuildPushServices(feed);
await using HtmlRenderer renderer = new(
provider,
provider.GetRequiredService<ILoggerFactory>());
HtmlRootComponent page = await renderer.Dispatcher.InvokeAsync(
() => renderer.RenderComponentAsync<AlarmsPage>());
await feed.PushAsync(SnapshotStatusFrame(truncated: true));
await WaitForHtmlAsync(
renderer,
page,
html => html.Contains(BannerMarker, StringComparison.Ordinal),
"banner to appear before the clearing frame is pushed");
await feed.PushAsync(SnapshotStatusFrame(truncated: false));
await WaitForHtmlAsync(
renderer,
page,
html => !html.Contains(BannerMarker, StringComparison.Ordinal),
"banner to clear after a complete snapshot_status frame");
}
private static AlarmFeedMessage SnapshotStatusFrame(bool truncated)
{
return new AlarmFeedMessage
{
SnapshotStatus = new AlarmSnapshotStatus { Truncated = truncated },
};
}
private static ServiceProvider BuildPushServices(ScriptedAlarmFeed feed)
{
ServiceCollection services = new();
services.AddLogging();
services.AddSingleton<IDashboardLiveDataService, SinglePollLiveDataService>();
services.AddSingleton<IGatewayAlarmService>(feed);
services.AddSingleton<IOptions<GatewayOptions>>(
Options.Create(new GatewayOptions { Alarms = new AlarmsOptions { Enabled = true } }));
return services.BuildServiceProvider();
}
private static Task<string> HtmlAsync(HtmlRenderer renderer, HtmlRootComponent page)
{
// Serialization has to happen on the renderer's dispatcher, and it reads the
// component's current render tree — so it reflects renders the page's feed
// loop queued after the initial quiescent render.
return renderer.Dispatcher.InvokeAsync(page.ToHtmlString);
}
private static async Task WaitForHtmlAsync(
HtmlRenderer renderer,
HtmlRootComponent page,
Func<string, bool> predicate,
string expectation)
{
Stopwatch elapsed = Stopwatch.StartNew();
while (elapsed.Elapsed < RenderWaitTimeout)
{
if (predicate(await HtmlAsync(renderer, page)))
{
return;
}
await Task.Delay(TimeSpan.FromMilliseconds(20));
}
Assert.Fail($"Timed out after {RenderWaitTimeout.TotalSeconds:N0}s waiting for the {expectation}.");
}
private static async Task<string> RenderAsync(bool snapshotTruncated)
{
ServiceCollection services = new();
@@ -102,4 +222,93 @@ public sealed class AlarmsPageTruncationBannerTests
WorkerProcessId: null,
SnapshotTruncated: snapshotTruncated));
}
// Answers exactly one poll — the inline first pass — and parks every later tick
// until the page's disposal cancels it. The parking is what makes the push tests
// measure the push: a second tick would re-assert the poll's own verdict, and
// could either mask a banner the feed raised or raise one the feed did not.
private sealed class SinglePollLiveDataService : IDashboardLiveDataService
{
private int _polls;
/// <inheritdoc />
public Task<DashboardLiveReadResult> ReadAsync(
IReadOnlyCollection<string> tagAddresses,
CancellationToken cancellationToken) =>
Task.FromResult(DashboardLiveReadResult.Empty);
/// <inheritdoc />
public async Task<DashboardAlarmQueryResult> QueryAlarmsAsync(CancellationToken cancellationToken)
{
if (Interlocked.Increment(ref _polls) > 1)
{
await Task.Delay(Timeout.Infinite, cancellationToken).ConfigureAwait(false);
}
return new DashboardAlarmQueryResult(
Alarms: [],
Error: null,
WorkerProcessId: null,
SnapshotTruncated: false);
}
}
// A hand-driven stand-in for the alarm feed: the test writes the frames the real
// monitor would push. Unbounded and never completed, so a frame written before the
// page's loop attaches is still delivered, and the loop never has to resubscribe.
private sealed class ScriptedAlarmFeed : IGatewayAlarmService
{
private readonly Channel<AlarmFeedMessage> _frames =
Channel.CreateUnbounded<AlarmFeedMessage>(new UnboundedChannelOptions
{
SingleReader = false,
SingleWriter = false,
});
/// <inheritdoc />
public GatewayAlarmMonitorState State => GatewayAlarmMonitorState.Monitoring;
/// <inheritdoc />
public string? LastError => null;
/// <inheritdoc />
public int? WorkerProcessId => null;
/// <inheritdoc />
public IReadOnlyList<ActiveAlarmSnapshot> CurrentAlarms => [];
/// <inheritdoc />
public bool SnapshotTruncated { get; set; }
/// <summary>Pushes one frame onto the feed the page is subscribed to.</summary>
/// <param name="message">The feed frame to deliver.</param>
/// <returns>A task that represents the asynchronous operation.</returns>
public ValueTask PushAsync(AlarmFeedMessage message) => _frames.Writer.WriteAsync(message);
/// <inheritdoc />
public async IAsyncEnumerable<AlarmFeedMessage> StreamAsync(
string? alarmFilterPrefix,
[EnumeratorCancellation] CancellationToken cancellationToken)
{
await foreach (AlarmFeedMessage message in _frames.Reader
.ReadAllAsync(cancellationToken)
.ConfigureAwait(false))
{
yield return message;
}
}
/// <inheritdoc />
public Task<AcknowledgeAlarmReply> AcknowledgeAsync(
AcknowledgeAlarmRequest request,
CancellationToken cancellationToken)
{
return Task.FromResult(new AcknowledgeAlarmReply
{
CorrelationId = request.ClientCorrelationId,
ProtocolStatus = new ProtocolStatus { Code = ProtocolStatusCode.Ok },
DiagnosticMessage = string.Empty,
});
}
}
}