Merge stream D (dashboard: bounded hub fan-out, alarm reset marker, browse debounce, snapshot config once, health status, per-viewer LRU, circuit retention, doc rewrite)
This commit is contained in:
@@ -58,6 +58,8 @@
|
||||
<MetricCard Label="Active (unacked)" Value="@_unackedCount.ToString("N0")" />
|
||||
<MetricCard Label="Acknowledged" Value="@_ackedCount.ToString("N0")" />
|
||||
<MetricCard Label="Total Active" Value="@_alarms.Count.ToString("N0")" />
|
||||
@* FilteredAlarms() is memoized on (alarm generation, filter inputs), so this card and the
|
||||
table below share one computation per change instead of filtering twice per render. *@
|
||||
<MetricCard Label="Showing" Value="@FilteredAlarms().Count.ToString("N0")" />
|
||||
</section>
|
||||
|
||||
@@ -189,6 +191,10 @@
|
||||
private int _maxSeverity = 1000;
|
||||
private string _search = string.Empty;
|
||||
|
||||
// Bumped whenever _alarms is replaced, so the memoized view recomputes exactly then.
|
||||
private readonly DashboardAlarmView _view = new();
|
||||
private long _alarmGeneration;
|
||||
|
||||
// Upper bound on waiting for either background loop while disposing, mirroring
|
||||
// DashboardPageBase's snapshot-watch drain: both loops marshal renders through the
|
||||
// renderer's dispatcher and disposal can run on that same dispatcher, so the wait is
|
||||
@@ -295,42 +301,13 @@
|
||||
return _workerPid is int pid ? $"{refreshed} · worker pid {pid}" : refreshed;
|
||||
}
|
||||
|
||||
private IReadOnlyList<string> Areas()
|
||||
{
|
||||
return _alarms
|
||||
.Select(alarm => alarm.Area)
|
||||
.Where(area => !string.IsNullOrWhiteSpace(area))
|
||||
.Distinct(StringComparer.OrdinalIgnoreCase)
|
||||
.OrderBy(area => area, StringComparer.OrdinalIgnoreCase)
|
||||
.ToArray();
|
||||
}
|
||||
private IReadOnlyList<string> Areas() => _view.Areas(_alarmGeneration, _alarms);
|
||||
|
||||
private IReadOnlyList<DashboardActiveAlarm> FilteredAlarms()
|
||||
{
|
||||
string query = _search.Trim();
|
||||
return _alarms
|
||||
.Where(MatchesState)
|
||||
.Where(alarm => _areaFilter.Length == 0
|
||||
|| string.Equals(alarm.Area, _areaFilter, StringComparison.OrdinalIgnoreCase))
|
||||
.Where(alarm => alarm.Severity >= _minSeverity && alarm.Severity <= _maxSeverity)
|
||||
.Where(alarm => query.Length == 0
|
||||
|| alarm.Reference.Contains(query, StringComparison.OrdinalIgnoreCase)
|
||||
|| alarm.Source.Contains(query, StringComparison.OrdinalIgnoreCase)
|
||||
|| alarm.Description.Contains(query, StringComparison.OrdinalIgnoreCase))
|
||||
.OrderByDescending(alarm => alarm.Severity)
|
||||
.ThenByDescending(alarm => alarm.LastTransition ?? DateTimeOffset.MinValue)
|
||||
.ToArray();
|
||||
}
|
||||
|
||||
private bool MatchesState(DashboardActiveAlarm alarm)
|
||||
{
|
||||
return alarm.State switch
|
||||
{
|
||||
AlarmConditionState.Active => _showActive,
|
||||
AlarmConditionState.ActiveAcked => _showAcked,
|
||||
_ => true,
|
||||
};
|
||||
}
|
||||
private IReadOnlyList<DashboardActiveAlarm> FilteredAlarms() =>
|
||||
_view.Filtered(
|
||||
_alarmGeneration,
|
||||
new DashboardAlarmFilterCriteria(_showActive, _showAcked, _areaFilter, _minSeverity, _maxSeverity, _search),
|
||||
_alarms);
|
||||
|
||||
private static string StateText(AlarmConditionState state)
|
||||
{
|
||||
@@ -439,6 +416,7 @@
|
||||
_lastRefresh = DateTimeOffset.UtcNow;
|
||||
_alarms.Clear();
|
||||
_alarms.AddRange(result.Alarms);
|
||||
_alarmGeneration++;
|
||||
_unackedCount = _alarms.Count(alarm => alarm.State == AlarmConditionState.Active);
|
||||
_ackedCount = _alarms.Count(alarm => alarm.State == AlarmConditionState.ActiveAcked);
|
||||
StateHasChanged();
|
||||
|
||||
@@ -4,6 +4,8 @@
|
||||
@inject IDashboardLiveDataService LiveData
|
||||
@inject IDashboardBrowseService BrowseService
|
||||
@inject IGalaxyDeployNotifier DeployNotifier
|
||||
@inject DashboardBrowseSearchIndex SearchIndex
|
||||
@inject DashboardCircuitConnectionTracker Circuit
|
||||
@using ZB.MOM.WW.GalaxyRepository
|
||||
@using ZB.MOM.WW.GalaxyRepository.Grpc
|
||||
|
||||
@@ -36,7 +38,15 @@
|
||||
}
|
||||
else if (!string.IsNullOrWhiteSpace(Search))
|
||||
{
|
||||
@if (_searchMatches.Count == 0)
|
||||
@if (!string.IsNullOrWhiteSpace(_searchError))
|
||||
{
|
||||
<div class="alert alert-danger">Search failed: @_searchError</div>
|
||||
}
|
||||
else if (_searchPending && _searchMatches.Count == 0)
|
||||
{
|
||||
<div class="empty-state">Searching…</div>
|
||||
}
|
||||
else if (_searchMatches.Count == 0)
|
||||
{
|
||||
<div class="empty-state">No attributes match “@Search”.</div>
|
||||
}
|
||||
@@ -115,6 +125,13 @@
|
||||
<div class="alert alert-danger">Live read failed: @_readError</div>
|
||||
}
|
||||
|
||||
@if (!string.IsNullOrWhiteSpace(_deployError))
|
||||
{
|
||||
<div class="alert alert-warning">
|
||||
Galaxy redeploy notifications failed (retrying): @_deployError
|
||||
</div>
|
||||
}
|
||||
|
||||
@if (_subscribed.Count == 0)
|
||||
{
|
||||
<div class="empty-state">
|
||||
@@ -194,13 +211,35 @@
|
||||
@code {
|
||||
private const int SearchResultLimit = 300;
|
||||
|
||||
// Keystrokes inside this window coalesce into one search: typing a tag name no
|
||||
// longer rescans the Galaxy once per character.
|
||||
private static readonly TimeSpan SearchDebounce = TimeSpan.FromMilliseconds(150);
|
||||
|
||||
// Delay before resubscribing to deploy notifications after the stream ends or faults.
|
||||
private static readonly TimeSpan DeployResubscribeDelay = TimeSpan.FromSeconds(5);
|
||||
|
||||
// Upper bound on waiting for the page's background loops while disposing, the same
|
||||
// bound DashboardPageBase and AlarmsPage use: the loops marshal renders through the
|
||||
// renderer's dispatcher and disposal can run on that same dispatcher, so the wait is
|
||||
// bounded rather than unconditional.
|
||||
private static readonly TimeSpan LoopDrainTimeout = TimeSpan.FromSeconds(5);
|
||||
|
||||
// Identifies this page to the shared live-data service, which budgets the advise set
|
||||
// per viewer so another circuit's reads cannot evict this page's whole panel.
|
||||
private readonly string _viewerId = "browse-" + Guid.NewGuid().ToString("N");
|
||||
|
||||
private List<DashboardBrowseNode> _roots = [];
|
||||
private ulong _cacheSequence;
|
||||
private string? _staleBanner;
|
||||
private CancellationTokenSource _deployCts = new();
|
||||
private readonly CancellationTokenSource _deployCts = new();
|
||||
private Task? _deployTask;
|
||||
private string? _deployError;
|
||||
private string _search = string.Empty;
|
||||
private IReadOnlyList<GalaxyAttribute> _searchMatches = [];
|
||||
private bool _searchPending;
|
||||
private string? _searchError;
|
||||
private CancellationTokenSource? _searchCts;
|
||||
private Task? _searchTask;
|
||||
private readonly List<string> _subscribed = [];
|
||||
private readonly Dictionary<string, DashboardTagValue> _values = new(StringComparer.Ordinal);
|
||||
// Per-tag bookkeeping for the Updated column: the signature of the value
|
||||
@@ -225,8 +264,8 @@
|
||||
BrowseLevelResult roots = BrowseService.GetRoots(new BrowseFilterArgs());
|
||||
_roots = [.. roots.Nodes];
|
||||
_cacheSequence = roots.CacheSequence;
|
||||
_pollTask = PollLoopAsync();
|
||||
_deployTask = SubscribeToDeployEventsAsync();
|
||||
_pollTask = PollLoopAsync(_cts.Token);
|
||||
_deployTask = SubscribeToDeployEventsAsync(_deployCts.Token);
|
||||
}
|
||||
|
||||
private async Task LoadChildrenAsync(DashboardBrowseNode node)
|
||||
@@ -249,30 +288,55 @@
|
||||
await InvokeAsync(StateHasChanged);
|
||||
}
|
||||
|
||||
private async Task SubscribeToDeployEventsAsync()
|
||||
// Faults are handled inside the loop: a notifier stream that faults or ends is
|
||||
// surfaced on the page and resubscribed after a delay, so the tree does not silently
|
||||
// stop tracking redeploys. Only cancellation ends the loop, so DrainAsync never
|
||||
// observes a faulted task.
|
||||
private async Task SubscribeToDeployEventsAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
try
|
||||
while (!cancellationToken.IsCancellationRequested)
|
||||
{
|
||||
await foreach (GalaxyDeployEventInfo info in DeployNotifier
|
||||
.SubscribeAsync(_deployCts.Token)
|
||||
.ConfigureAwait(false))
|
||||
try
|
||||
{
|
||||
// First Latest replay echoes the sequence we already projected
|
||||
// from — skip those to avoid a spurious "redeployed" banner.
|
||||
if (info.Sequence == 0 || (ulong)info.Sequence == _cacheSequence)
|
||||
await foreach (GalaxyDeployEventInfo info in DeployNotifier
|
||||
.SubscribeAsync(cancellationToken)
|
||||
.ConfigureAwait(false))
|
||||
{
|
||||
continue;
|
||||
}
|
||||
// First Latest replay echoes the sequence we already projected
|
||||
// from — skip those to avoid a spurious "redeployed" banner.
|
||||
if (info.Sequence == 0 || (ulong)info.Sequence == _cacheSequence)
|
||||
{
|
||||
continue;
|
||||
}
|
||||
|
||||
BrowseLevelResult roots = BrowseService.GetRoots(new BrowseFilterArgs());
|
||||
_roots = [.. roots.Nodes];
|
||||
_cacheSequence = roots.CacheSequence;
|
||||
_staleBanner = "Galaxy redeployed — tree refreshed.";
|
||||
await InvokeAsync(StateHasChanged);
|
||||
await InvokeAsync(() =>
|
||||
{
|
||||
BrowseLevelResult roots = BrowseService.GetRoots(new BrowseFilterArgs());
|
||||
_roots = [.. roots.Nodes];
|
||||
_cacheSequence = roots.CacheSequence;
|
||||
_staleBanner = "Galaxy redeployed — tree refreshed.";
|
||||
_deployError = null;
|
||||
StateHasChanged();
|
||||
}).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
|
||||
{
|
||||
return;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
await ReportAsync(() => _deployError = ex.Message).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
await Task.Delay(DeployResubscribeDelay, cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
return;
|
||||
}
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
}
|
||||
}
|
||||
|
||||
@@ -288,42 +352,101 @@
|
||||
+ $"{entry.AlarmAttributeCount:N0} alarm attributes";
|
||||
}
|
||||
|
||||
// Bound to the filter box with @bind:event="oninput", so the setter runs once per
|
||||
// keystroke on the render dispatcher. It only schedules the search; the scan itself
|
||||
// runs after the debounce, off the dispatcher, and a newer keystroke cancels it.
|
||||
private string Search
|
||||
{
|
||||
get => _search;
|
||||
set
|
||||
{
|
||||
_search = value ?? string.Empty;
|
||||
_searchMatches = ComputeSearch(_search);
|
||||
ScheduleSearch(_search);
|
||||
}
|
||||
}
|
||||
|
||||
private IReadOnlyList<GalaxyAttribute> ComputeSearch(string rawQuery)
|
||||
private void ScheduleSearch(string query)
|
||||
{
|
||||
string query = rawQuery.Trim();
|
||||
if (query.Length == 0)
|
||||
CancelQuietly(_searchCts);
|
||||
_searchCts = null;
|
||||
_searchError = null;
|
||||
|
||||
if (query.Trim().Length == 0)
|
||||
{
|
||||
return [];
|
||||
_searchMatches = [];
|
||||
_searchPending = false;
|
||||
return;
|
||||
}
|
||||
|
||||
List<GalaxyAttribute> matches = [];
|
||||
foreach (GalaxyObject galaxyObject in GalaxyCache.Current.Objects)
|
||||
// Linked to the page token, so disposal cancels an in-flight search too. The
|
||||
// search task owns the source and disposes it when it finishes.
|
||||
CancellationTokenSource searchCts = CancellationTokenSource.CreateLinkedTokenSource(_cts.Token);
|
||||
_searchCts = searchCts;
|
||||
_searchPending = true;
|
||||
_searchTask = RunSearchAsync(query, searchCts);
|
||||
}
|
||||
|
||||
private async Task RunSearchAsync(string query, CancellationTokenSource searchCts)
|
||||
{
|
||||
CancellationToken cancellationToken = searchCts.Token;
|
||||
try
|
||||
{
|
||||
foreach (GalaxyAttribute attr in galaxyObject.Attributes)
|
||||
await Task.Delay(SearchDebounce, cancellationToken).ConfigureAwait(false);
|
||||
GalaxyHierarchyCacheEntry entry = GalaxyCache.Current;
|
||||
IReadOnlyList<GalaxyAttribute> matches = await Task.Run(
|
||||
() => SearchIndex.Search(entry, query, SearchResultLimit, cancellationToken),
|
||||
cancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
|
||||
await InvokeAsync(() =>
|
||||
{
|
||||
if (attr.FullTagReference.Contains(query, StringComparison.OrdinalIgnoreCase)
|
||||
|| attr.AttributeName.Contains(query, StringComparison.OrdinalIgnoreCase))
|
||||
// Only the latest query may publish: a slower, superseded scan that
|
||||
// finishes late must not overwrite newer results.
|
||||
if (ReferenceEquals(_searchCts, searchCts) && !cancellationToken.IsCancellationRequested)
|
||||
{
|
||||
matches.Add(attr);
|
||||
if (matches.Count >= SearchResultLimit)
|
||||
{
|
||||
return matches;
|
||||
}
|
||||
_searchMatches = matches;
|
||||
_searchPending = false;
|
||||
StateHasChanged();
|
||||
}
|
||||
}
|
||||
}).ConfigureAwait(false);
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
// Superseded by a newer keystroke, or the page is going away.
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
await ReportAsync(() =>
|
||||
{
|
||||
if (ReferenceEquals(_searchCts, searchCts))
|
||||
{
|
||||
_searchPending = false;
|
||||
_searchError = ex.Message;
|
||||
}
|
||||
}).ConfigureAwait(false);
|
||||
}
|
||||
finally
|
||||
{
|
||||
searchCts.Dispose();
|
||||
}
|
||||
}
|
||||
|
||||
// The search task disposes its own source, so a cancel can race that disposal.
|
||||
private static void CancelQuietly(CancellationTokenSource? source)
|
||||
{
|
||||
if (source is null)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
return matches;
|
||||
try
|
||||
{
|
||||
source.Cancel();
|
||||
}
|
||||
catch (ObjectDisposedException)
|
||||
{
|
||||
// The search already finished; there is nothing left to cancel.
|
||||
}
|
||||
}
|
||||
|
||||
private static string FormatType(GalaxyAttribute attr)
|
||||
@@ -422,17 +545,66 @@
|
||||
: "When the dashboard first observed this value — MXAccess did not supply a source timestamp for this tag.";
|
||||
}
|
||||
|
||||
private async Task PollLoopAsync()
|
||||
// Fault handling sits inside the loop, matching AlarmsPage: a fault on one tick is
|
||||
// surfaced in the "Live read failed" banner and retried on the next tick rather than
|
||||
// ending polling for the life of the page. Cancellation is the only exit, so the loop
|
||||
// task never faults — which is what DrainAsync in DisposeAsync relies on.
|
||||
//
|
||||
// A tick is skipped while the circuit's browser is disconnected: a closed tab keeps
|
||||
// its circuit for the retention period, and polling ReadBulk on the shared dashboard
|
||||
// worker for a page nobody can see is pure cost. The first tick after a reconnect
|
||||
// refreshes as normal.
|
||||
private async Task PollLoopAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
try
|
||||
using PeriodicTimer timer = new(TimeSpan.FromSeconds(2));
|
||||
while (true)
|
||||
{
|
||||
using PeriodicTimer timer = new(TimeSpan.FromSeconds(2));
|
||||
while (await timer.WaitForNextTickAsync(_cts.Token).ConfigureAwait(false))
|
||||
try
|
||||
{
|
||||
if (!await timer.WaitForNextTickAsync(cancellationToken).ConfigureAwait(false))
|
||||
{
|
||||
return;
|
||||
}
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
if (!Circuit.IsConnected)
|
||||
{
|
||||
continue;
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
await InvokeAsync(RefreshValuesAsync).ConfigureAwait(false);
|
||||
}
|
||||
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
|
||||
{
|
||||
return;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
await ReportAsync(() => _readError = ex.Message).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
}
|
||||
|
||||
// Best-effort: the fault being reported may itself be an InvokeAsync against a
|
||||
// disposed renderer, in which case the dispatch fails the same way and there is no
|
||||
// page left to show it on.
|
||||
private async Task ReportAsync(Action update)
|
||||
{
|
||||
try
|
||||
{
|
||||
await InvokeAsync(() =>
|
||||
{
|
||||
update();
|
||||
StateHasChanged();
|
||||
}).ConfigureAwait(false);
|
||||
}
|
||||
catch
|
||||
{
|
||||
}
|
||||
}
|
||||
@@ -445,7 +617,7 @@
|
||||
}
|
||||
|
||||
string[] tags = [.. _subscribed];
|
||||
DashboardLiveReadResult result = await LiveData.ReadAsync(tags, _cts.Token);
|
||||
DashboardLiveReadResult result = await LiveData.ReadAsync(tags, _viewerId, _cts.Token);
|
||||
_readError = result.Error;
|
||||
_workerPid = result.WorkerProcessId;
|
||||
DateTimeOffset now = DateTimeOffset.UtcNow;
|
||||
@@ -454,7 +626,7 @@
|
||||
// Stamp the observed-change time only when the value/quality
|
||||
// signature actually changes, so the Updated column does not
|
||||
// tick on every poll for a static tag.
|
||||
string signature = $"{value.ValueText}{value.Quality}{value.Ok}";
|
||||
string signature = $"{value.ValueText}{value.Quality}{value.Ok}";
|
||||
if (!_valueSignature.TryGetValue(value.TagAddress, out string? previous)
|
||||
|| previous != signature)
|
||||
{
|
||||
@@ -471,31 +643,49 @@
|
||||
/// <inheritdoc />
|
||||
public async ValueTask DisposeAsync()
|
||||
{
|
||||
await _cts.CancelAsync();
|
||||
await _deployCts.CancelAsync();
|
||||
if (_pollTask is not null)
|
||||
try
|
||||
{
|
||||
try
|
||||
{
|
||||
await _pollTask;
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
}
|
||||
await _cts.CancelAsync().ConfigureAwait(false);
|
||||
await _deployCts.CancelAsync().ConfigureAwait(false);
|
||||
|
||||
// Drained together, not in sequence: a wedged dispatcher blocks every loop at
|
||||
// once, so sequential drains would multiply the bound. DrainAsync tolerates null.
|
||||
await Task.WhenAll(DrainAsync(_pollTask), DrainAsync(_deployTask), DrainAsync(_searchTask))
|
||||
.ConfigureAwait(false);
|
||||
}
|
||||
if (_deployTask is not null)
|
||||
finally
|
||||
{
|
||||
try
|
||||
{
|
||||
await _deployTask;
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
}
|
||||
// Always released, even if cancellation or a drain throws. A search still in
|
||||
// flight disposes its own linked source.
|
||||
_cts.Dispose();
|
||||
_deployCts.Dispose();
|
||||
GC.SuppressFinalize(this);
|
||||
}
|
||||
}
|
||||
|
||||
// The accepted cost of the bound is an abandoned loop that keeps its timer or
|
||||
// notifier subscription until it does unwind; waiting forever on a dispatcher that
|
||||
// is not draining would wedge the circuit teardown itself.
|
||||
private static async Task DrainAsync(Task? loop)
|
||||
{
|
||||
if (loop is null)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
_cts.Dispose();
|
||||
_deployCts.Dispose();
|
||||
GC.SuppressFinalize(this);
|
||||
try
|
||||
{
|
||||
await loop.WaitAsync(LoopDrainTimeout).ConfigureAwait(false);
|
||||
}
|
||||
catch (TimeoutException)
|
||||
{
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
}
|
||||
catch
|
||||
{
|
||||
// Other disposal-time errors are best-effort.
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -10,7 +10,7 @@
|
||||
"Ready" or "Healthy" or "Active" => StatusState.Ok,
|
||||
"Creating" or "StartingWorker" or "WaitingForPipe" or "InitializingWorker" or "Closing"
|
||||
or "Stale" or "Degraded" or "Expiring" => StatusState.Warn,
|
||||
"Faulted" or "Unavailable" or "Expired" => StatusState.Bad,
|
||||
"Faulted" or "Unavailable" or "Expired" or "Unhealthy" => StatusState.Bad,
|
||||
_ => StatusState.Idle,
|
||||
};
|
||||
}
|
||||
|
||||
@@ -0,0 +1,125 @@
|
||||
using ZB.MOM.WW.MxGateway.Contracts.Proto;
|
||||
|
||||
namespace ZB.MOM.WW.MxGateway.Server.Dashboard;
|
||||
|
||||
/// <summary>The Alarms page's filter inputs, compared by value to decide whether a filtered view is stale.</summary>
|
||||
/// <param name="ShowActive">Include unacknowledged <c>Active</c> alarms.</param>
|
||||
/// <param name="ShowAcked">Include <c>ActiveAcked</c> alarms.</param>
|
||||
/// <param name="Area">Area to match (case-insensitive), or empty for all.</param>
|
||||
/// <param name="MinSeverity">Lowest severity shown.</param>
|
||||
/// <param name="MaxSeverity">Highest severity shown.</param>
|
||||
/// <param name="Search">Reference/source/description text search; trimmed before matching.</param>
|
||||
public sealed record DashboardAlarmFilterCriteria(
|
||||
bool ShowActive,
|
||||
bool ShowAcked,
|
||||
string Area,
|
||||
int MinSeverity,
|
||||
int MaxSeverity,
|
||||
string Search);
|
||||
|
||||
/// <summary>
|
||||
/// Memoized projections of the Alarms page's alarm list: the filtered, sorted rows and
|
||||
/// the distinct area list. A render used to run the full filter-and-sort twice (the
|
||||
/// "Showing" card and the table) and the area projection once more, on every render —
|
||||
/// including the renders that typing in the search box triggers per keystroke. Each
|
||||
/// projection is now recomputed only when the alarm list (its generation) or, for the
|
||||
/// rows, the filter criteria change.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// Owned by one page instance and touched only from its render dispatcher, so it is not
|
||||
/// synchronized.
|
||||
/// </remarks>
|
||||
public sealed class DashboardAlarmView
|
||||
{
|
||||
private long _filteredGeneration = -1;
|
||||
private DashboardAlarmFilterCriteria? _filteredCriteria;
|
||||
private IReadOnlyList<DashboardActiveAlarm> _filtered = [];
|
||||
private long _areasGeneration = -1;
|
||||
private IReadOnlyList<string> _areas = [];
|
||||
|
||||
/// <summary>Gets how many times the filtered rows were actually computed (diagnostic, for tests).</summary>
|
||||
internal int FilterComputations { get; private set; }
|
||||
|
||||
/// <summary>Gets how many times the area list was actually computed (diagnostic, for tests).</summary>
|
||||
internal int AreaComputations { get; private set; }
|
||||
|
||||
/// <summary>Returns the alarms matching <paramref name="criteria"/>, highest severity then newest first.</summary>
|
||||
/// <param name="generation">
|
||||
/// Version of <paramref name="alarms"/>; the caller bumps it whenever it replaces the list's contents.
|
||||
/// </param>
|
||||
/// <param name="criteria">The current filter inputs.</param>
|
||||
/// <param name="alarms">The full active-alarm list.</param>
|
||||
/// <returns>The filtered rows; the same instance while generation and criteria are unchanged.</returns>
|
||||
public IReadOnlyList<DashboardActiveAlarm> Filtered(
|
||||
long generation,
|
||||
DashboardAlarmFilterCriteria criteria,
|
||||
IReadOnlyList<DashboardActiveAlarm> alarms)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(criteria);
|
||||
ArgumentNullException.ThrowIfNull(alarms);
|
||||
|
||||
if (generation == _filteredGeneration && criteria == _filteredCriteria)
|
||||
{
|
||||
return _filtered;
|
||||
}
|
||||
|
||||
_filtered = Apply(criteria, alarms);
|
||||
_filteredGeneration = generation;
|
||||
_filteredCriteria = criteria;
|
||||
FilterComputations++;
|
||||
return _filtered;
|
||||
}
|
||||
|
||||
/// <summary>Returns the distinct non-blank areas of <paramref name="alarms"/>, sorted case-insensitively.</summary>
|
||||
/// <param name="generation">Version of <paramref name="alarms"/>.</param>
|
||||
/// <param name="alarms">The full active-alarm list.</param>
|
||||
/// <returns>The area list; the same instance while the generation is unchanged.</returns>
|
||||
public IReadOnlyList<string> Areas(long generation, IReadOnlyList<DashboardActiveAlarm> alarms)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(alarms);
|
||||
|
||||
if (generation == _areasGeneration)
|
||||
{
|
||||
return _areas;
|
||||
}
|
||||
|
||||
_areas = alarms
|
||||
.Select(alarm => alarm.Area)
|
||||
.Where(area => !string.IsNullOrWhiteSpace(area))
|
||||
.Distinct(StringComparer.OrdinalIgnoreCase)
|
||||
.OrderBy(area => area, StringComparer.OrdinalIgnoreCase)
|
||||
.ToArray();
|
||||
_areasGeneration = generation;
|
||||
AreaComputations++;
|
||||
return _areas;
|
||||
}
|
||||
|
||||
private static DashboardActiveAlarm[] Apply(
|
||||
DashboardAlarmFilterCriteria criteria,
|
||||
IReadOnlyList<DashboardActiveAlarm> alarms)
|
||||
{
|
||||
string query = criteria.Search.Trim();
|
||||
return alarms
|
||||
.Where(alarm => MatchesState(criteria, alarm))
|
||||
.Where(alarm => criteria.Area.Length == 0
|
||||
|| string.Equals(alarm.Area, criteria.Area, StringComparison.OrdinalIgnoreCase))
|
||||
.Where(alarm => alarm.Severity >= criteria.MinSeverity && alarm.Severity <= criteria.MaxSeverity)
|
||||
.Where(alarm => query.Length == 0
|
||||
|| alarm.Reference.Contains(query, StringComparison.OrdinalIgnoreCase)
|
||||
|| alarm.Source.Contains(query, StringComparison.OrdinalIgnoreCase)
|
||||
|| alarm.Description.Contains(query, StringComparison.OrdinalIgnoreCase))
|
||||
.OrderByDescending(alarm => alarm.Severity)
|
||||
.ThenByDescending(alarm => alarm.LastTransition ?? DateTimeOffset.MinValue)
|
||||
.ToArray();
|
||||
}
|
||||
|
||||
private static bool MatchesState(DashboardAlarmFilterCriteria criteria, DashboardActiveAlarm alarm)
|
||||
{
|
||||
return alarm.State switch
|
||||
{
|
||||
AlarmConditionState.Active => criteria.ShowActive,
|
||||
AlarmConditionState.ActiveAcked => criteria.ShowAcked,
|
||||
_ => true,
|
||||
};
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,117 @@
|
||||
using ZB.MOM.WW.GalaxyRepository;
|
||||
using ZB.MOM.WW.GalaxyRepository.Grpc;
|
||||
|
||||
namespace ZB.MOM.WW.MxGateway.Server.Dashboard;
|
||||
|
||||
/// <summary>
|
||||
/// Flat, pre-lowercased attribute index behind the Browse page's filter box, shared by
|
||||
/// every circuit and rebuilt only when the Galaxy cache sequence changes.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// The filter used to walk every object and attribute of the cached Galaxy, doing two
|
||||
/// case-insensitive <c>Contains</c> per attribute, synchronously in the property setter
|
||||
/// bound to <c>oninput</c> — so a large Galaxy was rescanned on the render dispatcher
|
||||
/// once per keystroke, per circuit. The page now debounces and runs the search off the
|
||||
/// dispatcher; this index makes each search a single pass of ordinal substring checks
|
||||
/// over strings lowercased once per Galaxy deploy rather than once per comparison.
|
||||
/// Lowercasing is invariant-culture, which matches the previous
|
||||
/// <see cref="StringComparison.OrdinalIgnoreCase"/> behaviour for the ASCII tag names
|
||||
/// Galaxy uses.
|
||||
/// </remarks>
|
||||
public sealed class DashboardBrowseSearchIndex
|
||||
{
|
||||
// How often a scan checks its cancellation token. A superseded keystroke stops
|
||||
// within this many attributes rather than finishing the whole Galaxy.
|
||||
private const int CancellationCheckInterval = 1024;
|
||||
|
||||
private IndexSnapshot? _snapshot;
|
||||
private int _buildCount;
|
||||
|
||||
/// <summary>Gets how many times the index has been (re)built (diagnostic, for tests).</summary>
|
||||
internal int BuildCount => Volatile.Read(ref _buildCount);
|
||||
|
||||
/// <summary>
|
||||
/// Returns up to <paramref name="limit"/> attributes whose full tag reference or
|
||||
/// attribute name contains <paramref name="rawQuery"/>, case-insensitively, in
|
||||
/// Galaxy object order.
|
||||
/// </summary>
|
||||
/// <param name="entry">The current Galaxy cache entry; its sequence keys the index.</param>
|
||||
/// <param name="rawQuery">The filter text as typed; it is trimmed.</param>
|
||||
/// <param name="limit">Maximum number of matches.</param>
|
||||
/// <param name="cancellationToken">Cancelled when a newer query supersedes this one.</param>
|
||||
/// <returns>The matches; empty for a blank query.</returns>
|
||||
/// <exception cref="OperationCanceledException"><paramref name="cancellationToken"/> was cancelled.</exception>
|
||||
public IReadOnlyList<GalaxyAttribute> Search(
|
||||
GalaxyHierarchyCacheEntry entry,
|
||||
string rawQuery,
|
||||
int limit,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(entry);
|
||||
|
||||
string query = (rawQuery ?? string.Empty).Trim().ToLowerInvariant();
|
||||
if (query.Length == 0 || limit <= 0)
|
||||
{
|
||||
return [];
|
||||
}
|
||||
|
||||
IndexedAttribute[] attributes = GetOrBuild(entry, cancellationToken);
|
||||
List<GalaxyAttribute> matches = [];
|
||||
for (int i = 0; i < attributes.Length; i++)
|
||||
{
|
||||
if (i % CancellationCheckInterval == 0)
|
||||
{
|
||||
cancellationToken.ThrowIfCancellationRequested();
|
||||
}
|
||||
|
||||
IndexedAttribute candidate = attributes[i];
|
||||
if (candidate.FullTagReference.Contains(query, StringComparison.Ordinal)
|
||||
|| candidate.AttributeName.Contains(query, StringComparison.Ordinal))
|
||||
{
|
||||
matches.Add(candidate.Attribute);
|
||||
if (matches.Count >= limit)
|
||||
{
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return matches;
|
||||
}
|
||||
|
||||
private IndexedAttribute[] GetOrBuild(GalaxyHierarchyCacheEntry entry, CancellationToken cancellationToken)
|
||||
{
|
||||
IndexSnapshot? current = Volatile.Read(ref _snapshot);
|
||||
if (current is not null && current.Sequence == entry.Sequence)
|
||||
{
|
||||
return current.Attributes;
|
||||
}
|
||||
|
||||
// A racing build for the same sequence is harmless: the projection is pure, so
|
||||
// either winner stores equivalent content.
|
||||
List<IndexedAttribute> attributes = [];
|
||||
foreach (GalaxyObject galaxyObject in entry.Objects)
|
||||
{
|
||||
cancellationToken.ThrowIfCancellationRequested();
|
||||
foreach (GalaxyAttribute attribute in galaxyObject.Attributes)
|
||||
{
|
||||
attributes.Add(new IndexedAttribute(
|
||||
attribute.FullTagReference.ToLowerInvariant(),
|
||||
attribute.AttributeName.ToLowerInvariant(),
|
||||
attribute));
|
||||
}
|
||||
}
|
||||
|
||||
IndexSnapshot built = new(entry.Sequence, [.. attributes]);
|
||||
Volatile.Write(ref _snapshot, built);
|
||||
Interlocked.Increment(ref _buildCount);
|
||||
return built.Attributes;
|
||||
}
|
||||
|
||||
private sealed record IndexSnapshot(long Sequence, IndexedAttribute[] Attributes);
|
||||
|
||||
private readonly record struct IndexedAttribute(
|
||||
string FullTagReference,
|
||||
string AttributeName,
|
||||
GalaxyAttribute Attribute);
|
||||
}
|
||||
@@ -0,0 +1,42 @@
|
||||
using Microsoft.AspNetCore.Components.Server.Circuits;
|
||||
|
||||
namespace ZB.MOM.WW.MxGateway.Server.Dashboard;
|
||||
|
||||
/// <summary>
|
||||
/// Per-circuit view of whether the browser is still attached. Registered scoped, so a
|
||||
/// page injecting it sees its own circuit, and also registered as that circuit's
|
||||
/// <see cref="CircuitHandler"/> so Blazor drives it.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// A closed or backgrounded browser tab leaves its circuit alive, disconnected, for
|
||||
/// <see cref="DashboardServiceCollectionExtensions.DisconnectedCircuitRetention"/>
|
||||
/// before Blazor disposes it, and every timer on the page keeps running in that
|
||||
/// window. The Browse page's two-second <c>ReadBulk</c> poll is the costly one — it
|
||||
/// drives the shared dashboard worker — so it skips its tick while
|
||||
/// <see cref="IsConnected"/> is false and resumes on reconnect.
|
||||
/// </remarks>
|
||||
public sealed class DashboardCircuitConnectionTracker : CircuitHandler
|
||||
{
|
||||
private volatile bool _connected = true;
|
||||
|
||||
/// <summary>
|
||||
/// Gets a value indicating whether the circuit's browser connection is up. True
|
||||
/// until Blazor reports the connection down, and also outside a circuit (prerender),
|
||||
/// where no handler callbacks ever arrive.
|
||||
/// </summary>
|
||||
public bool IsConnected => _connected;
|
||||
|
||||
/// <inheritdoc />
|
||||
public override Task OnConnectionUpAsync(Circuit circuit, CancellationToken cancellationToken)
|
||||
{
|
||||
_connected = true;
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public override Task OnConnectionDownAsync(Circuit circuit, CancellationToken cancellationToken)
|
||||
{
|
||||
_connected = false;
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
}
|
||||
@@ -34,20 +34,38 @@ public sealed class DashboardLiveDataService : IDashboardLiveDataService, IAsync
|
||||
//
|
||||
// The bound is per-read, not absolute: a read may never evict a tag it is itself
|
||||
// about to return, so a single read of more distinct tags than the cap leaves the
|
||||
// set that large. The invariant EvictForAsync actually maintains is
|
||||
// set that large. With a single viewer the invariant EvictForAsync maintains is
|
||||
//
|
||||
// |advise set| after a read <= max(MaxSubscribedTags, distinct tags in that read)
|
||||
//
|
||||
// and any overshoot is squeezed back out by the next read that subscribes a tag
|
||||
// (with several active viewers the per-viewer share below can hold it higher, by
|
||||
// at most what the other viewers hold within their shares) and any overshoot is squeezed back out by the next read that subscribes a tag
|
||||
// (see EvictForAsync). A browse page requests far fewer tags than the cap, so in
|
||||
// practice the set settles at MaxSubscribedTags.
|
||||
private const int MaxSubscribedTags = 256;
|
||||
|
||||
// Per-viewer soft cap: MaxSubscribedTags is shared by every active viewer (Browse
|
||||
// circuit), so each owns at most MaxSubscribedTags / activeViewers tags before its
|
||||
// own least-recently-read tags become the first eviction candidates — never fewer
|
||||
// than this floor. Without the split, two circuits browsing different areas
|
||||
// thrashed the single global LRU: each read evicted the other circuit's whole set,
|
||||
// which then re-advised it two seconds later.
|
||||
private const int MinTagsPerViewer = 32;
|
||||
|
||||
// A viewer that has not read within this window no longer counts as active: its
|
||||
// share is released and its tags become the first eviction candidates. Generous
|
||||
// against the Browse page's 2 s poll, so a briefly paused circuit keeps its share.
|
||||
private static readonly TimeSpan ActiveViewerWindow = TimeSpan.FromSeconds(30);
|
||||
|
||||
// Viewer id for callers that do not identify themselves.
|
||||
private const string SharedViewerId = "";
|
||||
|
||||
private static readonly TimeSpan ReadTimeout = TimeSpan.FromSeconds(5);
|
||||
|
||||
private readonly ISessionManager _sessionManager;
|
||||
private readonly IGatewayAlarmService _alarmService;
|
||||
private readonly ILogger<DashboardLiveDataService> _logger;
|
||||
private readonly TimeProvider _timeProvider;
|
||||
private readonly SemaphoreSlim _gate = new(1, 1);
|
||||
|
||||
/// <summary>
|
||||
@@ -70,6 +88,10 @@ public sealed class DashboardLiveDataService : IDashboardLiveDataService, IAsync
|
||||
|
||||
private readonly LinkedList<SubscribedTag> _recency = new();
|
||||
|
||||
// Per-viewer ownership and activity. A tag is owned by the viewer that read it
|
||||
// last; only touched under _gate.
|
||||
private readonly Dictionary<string, ViewerState> _viewers = new(StringComparer.Ordinal);
|
||||
|
||||
private GatewaySession? _session;
|
||||
private int _serverHandle;
|
||||
private bool _disposed;
|
||||
@@ -79,22 +101,31 @@ public sealed class DashboardLiveDataService : IDashboardLiveDataService, IAsync
|
||||
/// <param name="alarmService">Gateway central alarm service.</param>
|
||||
/// <param name="options">Gateway options supplying <c>Dashboard:ShowTagValues</c>.</param>
|
||||
/// <param name="logger">Diagnostic logger.</param>
|
||||
/// <param name="timeProvider">Clock for viewer activity; defaults to system time.</param>
|
||||
public DashboardLiveDataService(
|
||||
ISessionManager sessionManager,
|
||||
IGatewayAlarmService alarmService,
|
||||
IOptions<GatewayOptions> options,
|
||||
ILogger<DashboardLiveDataService> logger)
|
||||
ILogger<DashboardLiveDataService> logger,
|
||||
TimeProvider? timeProvider = null)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(options);
|
||||
_sessionManager = sessionManager ?? throw new ArgumentNullException(nameof(sessionManager));
|
||||
_alarmService = alarmService ?? throw new ArgumentNullException(nameof(alarmService));
|
||||
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
|
||||
_showTagValues = options.Value.Dashboard.ShowTagValues;
|
||||
_timeProvider = timeProvider ?? TimeProvider.System;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public Task<DashboardLiveReadResult> ReadAsync(
|
||||
IReadOnlyCollection<string> tagAddresses,
|
||||
CancellationToken cancellationToken) => ReadAsync(tagAddresses, viewerId: null, cancellationToken);
|
||||
|
||||
/// <inheritdoc />
|
||||
public async Task<DashboardLiveReadResult> ReadAsync(
|
||||
IReadOnlyCollection<string> tagAddresses,
|
||||
string? viewerId,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(tagAddresses);
|
||||
@@ -109,15 +140,19 @@ public sealed class DashboardLiveDataService : IDashboardLiveDataService, IAsync
|
||||
(GatewaySession session, int serverHandle) = await EnsureReadyAsync(cancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
|
||||
string[] toSubscribe = TouchAndCollectNewTags(tagAddresses, out int justReadCount);
|
||||
string viewer = viewerId ?? SharedViewerId;
|
||||
DateTimeOffset now = _timeProvider.GetUtcNow();
|
||||
TouchViewer(viewer, now);
|
||||
|
||||
string[] toSubscribe = TouchAndCollectNewTags(tagAddresses, viewer, out HashSet<string> justRead);
|
||||
if (toSubscribe.Length > 0)
|
||||
{
|
||||
await EvictForAsync(session, serverHandle, toSubscribe.Length, justReadCount, cancellationToken)
|
||||
await EvictForAsync(session, serverHandle, toSubscribe.Length, justRead, viewer, now, cancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
IReadOnlyList<SubscribeResult> subscribeResults = await session
|
||||
.SubscribeBulkAsync(serverHandle, toSubscribe, cancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
TrackSubscribed(toSubscribe, subscribeResults);
|
||||
TrackSubscribed(toSubscribe, subscribeResults, viewer);
|
||||
}
|
||||
|
||||
IReadOnlyList<BulkReadResult> results = await session
|
||||
@@ -175,18 +210,20 @@ public sealed class DashboardLiveDataService : IDashboardLiveDataService, IAsync
|
||||
}
|
||||
|
||||
// Promotes every already-advised tag in this read to the front of the recency
|
||||
// list and returns the tags that still need subscribing (distinct, in request
|
||||
// order). `justReadCount` is how many distinct tags of this read were already
|
||||
// advised — they now occupy the front of the list and must never be evicted to
|
||||
// make room for the same read's new tags. Callers must hold _gate.
|
||||
// list, hands its ownership to the reading viewer, and returns the tags that
|
||||
// still need subscribing (distinct, in request order). `justRead` is every
|
||||
// distinct tag of this read — none of them may be evicted to make room for the
|
||||
// same read's new tags. Callers must hold _gate.
|
||||
//
|
||||
// Every tag of one read is equally recently read; the recency list needs a total
|
||||
// order anyway, so the whole service uses one tie-break: later in the request wins.
|
||||
// Promoting in request order gives that here, and TrackSubscribed inserts new tags
|
||||
// the same way.
|
||||
private string[] TouchAndCollectNewTags(IReadOnlyCollection<string> tagAddresses, out int justReadCount)
|
||||
private string[] TouchAndCollectNewTags(
|
||||
IReadOnlyCollection<string> tagAddresses,
|
||||
string viewer,
|
||||
out HashSet<string> justRead)
|
||||
{
|
||||
int touched = 0;
|
||||
List<string> toSubscribe = [];
|
||||
HashSet<string> seen = new(StringComparer.OrdinalIgnoreCase);
|
||||
|
||||
@@ -200,10 +237,14 @@ public sealed class DashboardLiveDataService : IDashboardLiveDataService, IAsync
|
||||
_recency.AddFirst(node);
|
||||
}
|
||||
|
||||
if (seen.Add(tag))
|
||||
if (!string.Equals(node.Value.Owner, viewer, StringComparison.Ordinal))
|
||||
{
|
||||
touched++;
|
||||
AdjustOwned(node.Value.Owner, -1);
|
||||
AdjustOwned(viewer, +1);
|
||||
node.Value = node.Value with { Owner = viewer };
|
||||
}
|
||||
|
||||
seen.Add(tag);
|
||||
}
|
||||
else if (seen.Add(tag))
|
||||
{
|
||||
@@ -211,7 +252,7 @@ public sealed class DashboardLiveDataService : IDashboardLiveDataService, IAsync
|
||||
}
|
||||
}
|
||||
|
||||
justReadCount = touched;
|
||||
justRead = seen;
|
||||
return [.. toSubscribe];
|
||||
}
|
||||
|
||||
@@ -221,7 +262,17 @@ public sealed class DashboardLiveDataService : IDashboardLiveDataService, IAsync
|
||||
// from tracking regardless, and the session-invalidation path already handles
|
||||
// gateway/worker drift. Callers must hold _gate.
|
||||
//
|
||||
// Eviction stops at the tags this read just touched (`justReadCount`), so a read
|
||||
// Victims are chosen in two passes, each least-recently-read first:
|
||||
// 1. tags of OTHER viewers that are over their share (PerViewerCap; an
|
||||
// inactive viewer's share is zero);
|
||||
// 2. the reading viewer's own tags that this read did not touch.
|
||||
// A viewer within its share is never evicted by someone else's read, so the
|
||||
// cap is soft: when every other viewer sits within its share and the reader's
|
||||
// own older tags run out, the set ends over MaxSubscribedTags rather than
|
||||
// stripping a neighbour. With one viewer, pass 1 is empty and pass 2 is the
|
||||
// plain global LRU.
|
||||
//
|
||||
// Eviction never touches the tags this read just touched (`justRead`), so a read
|
||||
// whose own distinct tags outnumber the cap ends over it — see MaxSubscribedTags
|
||||
// for the exact invariant. That overshoot is not sticky: the next read that
|
||||
// subscribes anything computes `overflow` against the oversized set and evicts the
|
||||
@@ -236,26 +287,56 @@ public sealed class DashboardLiveDataService : IDashboardLiveDataService, IAsync
|
||||
GatewaySession session,
|
||||
int serverHandle,
|
||||
int incomingCount,
|
||||
int justReadCount,
|
||||
HashSet<string> justRead,
|
||||
string reader,
|
||||
DateTimeOffset now,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
int overflow = _subscribed.Count + incomingCount - MaxSubscribedTags;
|
||||
int evictable = _subscribed.Count - justReadCount;
|
||||
int evictCount = Math.Min(overflow, evictable);
|
||||
if (evictCount <= 0)
|
||||
if (overflow <= 0)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
List<int> evictedHandles = new(evictCount);
|
||||
for (int i = 0; i < evictCount && _recency.Last is { } oldest; i++)
|
||||
int share = PerViewerCap(now);
|
||||
List<int> evictedHandles = [];
|
||||
int evictCount = 0;
|
||||
|
||||
// Pass 1: other viewers over their share.
|
||||
for (LinkedListNode<SubscribedTag>? node = _recency.Last; node is not null && overflow > 0;)
|
||||
{
|
||||
_recency.RemoveLast();
|
||||
_subscribed.Remove(oldest.Value.TagAddress);
|
||||
if (oldest.Value.ItemHandle != 0)
|
||||
LinkedListNode<SubscribedTag>? previous = node.Previous;
|
||||
string owner = node.Value.Owner;
|
||||
if (!string.Equals(owner, reader, StringComparison.Ordinal)
|
||||
&& !justRead.Contains(node.Value.TagAddress)
|
||||
&& OwnedCount(owner) > (IsActive(owner, now) ? share : 0))
|
||||
{
|
||||
evictedHandles.Add(oldest.Value.ItemHandle);
|
||||
Evict(node, evictedHandles);
|
||||
overflow--;
|
||||
evictCount++;
|
||||
}
|
||||
|
||||
node = previous;
|
||||
}
|
||||
|
||||
// Pass 2: the reader's own tags this read did not touch.
|
||||
for (LinkedListNode<SubscribedTag>? node = _recency.Last; node is not null && overflow > 0;)
|
||||
{
|
||||
LinkedListNode<SubscribedTag>? previous = node.Previous;
|
||||
if (string.Equals(node.Value.Owner, reader, StringComparison.Ordinal)
|
||||
&& !justRead.Contains(node.Value.TagAddress))
|
||||
{
|
||||
Evict(node, evictedHandles);
|
||||
overflow--;
|
||||
evictCount++;
|
||||
}
|
||||
|
||||
node = previous;
|
||||
}
|
||||
|
||||
if (evictCount == 0)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
_logger.LogDebug(
|
||||
@@ -282,12 +363,81 @@ public sealed class DashboardLiveDataService : IDashboardLiveDataService, IAsync
|
||||
}
|
||||
}
|
||||
|
||||
// Removes one tag from tracking, releasing its owner's count and collecting its
|
||||
// item handle for the batch unadvise. Callers must hold _gate.
|
||||
private void Evict(LinkedListNode<SubscribedTag> node, List<int> evictedHandles)
|
||||
{
|
||||
_recency.Remove(node);
|
||||
_subscribed.Remove(node.Value.TagAddress);
|
||||
AdjustOwned(node.Value.Owner, -1);
|
||||
if (node.Value.ItemHandle != 0)
|
||||
{
|
||||
evictedHandles.Add(node.Value.ItemHandle);
|
||||
}
|
||||
}
|
||||
|
||||
// Each active viewer's share of MaxSubscribedTags, floored at MinTagsPerViewer.
|
||||
// Callers must hold _gate.
|
||||
private int PerViewerCap(DateTimeOffset now)
|
||||
{
|
||||
int active = _viewers.Values.Count(state => now - state.LastReadAt <= ActiveViewerWindow);
|
||||
return Math.Max(MinTagsPerViewer, MaxSubscribedTags / Math.Max(1, active));
|
||||
}
|
||||
|
||||
private bool IsActive(string viewer, DateTimeOffset now) =>
|
||||
_viewers.TryGetValue(viewer, out ViewerState? state) && now - state.LastReadAt <= ActiveViewerWindow;
|
||||
|
||||
private int OwnedCount(string viewer) =>
|
||||
_viewers.TryGetValue(viewer, out ViewerState? state) ? state.OwnedCount : 0;
|
||||
|
||||
// Records a read by `viewer` and forgets viewers that are both inactive and own
|
||||
// nothing, so the map does not grow one entry per Browse page ever opened.
|
||||
// Callers must hold _gate.
|
||||
private void TouchViewer(string viewer, DateTimeOffset now)
|
||||
{
|
||||
if (!_viewers.TryGetValue(viewer, out ViewerState? state))
|
||||
{
|
||||
state = new ViewerState();
|
||||
_viewers[viewer] = state;
|
||||
}
|
||||
|
||||
state.LastReadAt = now;
|
||||
|
||||
foreach (KeyValuePair<string, ViewerState> entry in _viewers.ToArray())
|
||||
{
|
||||
if (entry.Value.OwnedCount <= 0 && now - entry.Value.LastReadAt > ActiveViewerWindow)
|
||||
{
|
||||
_viewers.Remove(entry.Key);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Callers must hold _gate.
|
||||
private void AdjustOwned(string viewer, int delta)
|
||||
{
|
||||
if (!_viewers.TryGetValue(viewer, out ViewerState? state))
|
||||
{
|
||||
if (delta <= 0)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
state = new ViewerState { LastReadAt = DateTimeOffset.MinValue };
|
||||
_viewers[viewer] = state;
|
||||
}
|
||||
|
||||
state.OwnedCount = Math.Max(0, state.OwnedCount + delta);
|
||||
}
|
||||
|
||||
// Records the freshly advised tags as the most recently read, keeping each
|
||||
// tag's item handle so eviction can unadvise it. Tags the worker failed to
|
||||
// advise are still tracked (matching the pre-cap behaviour of not retrying
|
||||
// them on every read) but carry no handle, so eviction just forgets them.
|
||||
// Callers must hold _gate.
|
||||
private void TrackSubscribed(IReadOnlyList<string> tagAddresses, IReadOnlyList<SubscribeResult> results)
|
||||
private void TrackSubscribed(
|
||||
IReadOnlyList<string> tagAddresses,
|
||||
IReadOnlyList<SubscribeResult> results,
|
||||
string viewer)
|
||||
{
|
||||
Dictionary<string, int> handles = new(results.Count, StringComparer.OrdinalIgnoreCase);
|
||||
foreach (SubscribeResult result in results)
|
||||
@@ -303,7 +453,8 @@ public sealed class DashboardLiveDataService : IDashboardLiveDataService, IAsync
|
||||
foreach (string tag in tagAddresses)
|
||||
{
|
||||
handles.TryGetValue(tag, out int itemHandle);
|
||||
_subscribed[tag] = _recency.AddFirst(new SubscribedTag(tag, itemHandle));
|
||||
_subscribed[tag] = _recency.AddFirst(new SubscribedTag(tag, itemHandle, viewer));
|
||||
AdjustOwned(viewer, +1);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -314,6 +465,10 @@ public sealed class DashboardLiveDataService : IDashboardLiveDataService, IAsync
|
||||
{
|
||||
_subscribed.Clear();
|
||||
_recency.Clear();
|
||||
foreach (ViewerState state in _viewers.Values)
|
||||
{
|
||||
state.OwnedCount = 0;
|
||||
}
|
||||
}
|
||||
|
||||
// Returns a Ready session + its Register server handle, opening a fresh
|
||||
@@ -424,5 +579,14 @@ public sealed class DashboardLiveDataService : IDashboardLiveDataService, IAsync
|
||||
|
||||
// One entry of the advise set. ItemHandle is the handle the worker bound for
|
||||
// the tag, or 0 when the subscribe failed and there is nothing to unadvise.
|
||||
private readonly record struct SubscribedTag(string TagAddress, int ItemHandle);
|
||||
// Owner is the viewer that read the tag most recently.
|
||||
private readonly record struct SubscribedTag(string TagAddress, int ItemHandle, string Owner);
|
||||
|
||||
// One viewer's slice of the advise set.
|
||||
private sealed class ViewerState
|
||||
{
|
||||
public int OwnedCount { get; set; }
|
||||
|
||||
public DateTimeOffset LastReadAt { get; set; }
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
using Microsoft.AspNetCore.Authentication;
|
||||
using Microsoft.AspNetCore.Authentication.Cookies;
|
||||
using Microsoft.AspNetCore.Authorization;
|
||||
using Microsoft.AspNetCore.Components.Server.Circuits;
|
||||
using Microsoft.Extensions.Configuration;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Microsoft.Extensions.Options;
|
||||
@@ -16,6 +17,16 @@ namespace ZB.MOM.WW.MxGateway.Server.Dashboard;
|
||||
/// </summary>
|
||||
public static class DashboardServiceCollectionExtensions
|
||||
{
|
||||
/// <summary>
|
||||
/// How long a disconnected dashboard circuit is kept for reconnect before Blazor
|
||||
/// disposes it. The framework default is three minutes, during which a closed Browse
|
||||
/// tab's page — and its two-second live-read poll against the shared dashboard
|
||||
/// worker — keeps running. Thirty seconds still rides out a network blip or a
|
||||
/// laptop lid; a longer outage reloads the page, which costs nothing here because
|
||||
/// every page re-seeds from in-process state.
|
||||
/// </summary>
|
||||
public static readonly TimeSpan DisconnectedCircuitRetention = TimeSpan.FromSeconds(30);
|
||||
|
||||
/// <summary>
|
||||
/// Registers all dashboard services, authentication, and Razor components.
|
||||
/// </summary>
|
||||
@@ -54,6 +65,14 @@ public static class DashboardServiceCollectionExtensions
|
||||
// nothing in this process builds a hub connection or needs a token for one.
|
||||
services.AddSingleton<HubTokenService>();
|
||||
services.AddScoped<IDashboardBrowseService, DashboardBrowseService>();
|
||||
// Singleton: the lowercased attribute index is keyed on the Galaxy cache sequence
|
||||
// and shared by every circuit's Browse filter, so it is built once per deploy.
|
||||
services.AddSingleton<DashboardBrowseSearchIndex>();
|
||||
// Scoped (one per circuit) and registered as that circuit's CircuitHandler, so the
|
||||
// Browse page can pause its live-read poll while its browser is disconnected.
|
||||
services.AddScoped<DashboardCircuitConnectionTracker>();
|
||||
services.AddScoped<CircuitHandler>(
|
||||
static provider => provider.GetRequiredService<DashboardCircuitConnectionTracker>());
|
||||
// Singleton: EventsHub instances are transient (one per hub invocation), so the
|
||||
// subscriber bookkeeping they share with the broadcaster must outlive them.
|
||||
services.AddSingleton<Hubs.EventsHubViewerRegistry>();
|
||||
@@ -76,7 +95,8 @@ public static class DashboardServiceCollectionExtensions
|
||||
services.AddAntiforgery();
|
||||
services.AddCascadingAuthenticationState();
|
||||
services.AddRazorComponents()
|
||||
.AddInteractiveServerComponents();
|
||||
.AddInteractiveServerComponents(circuit =>
|
||||
circuit.DisconnectedCircuitRetentionPeriod = DisconnectedCircuitRetention);
|
||||
services.AddSignalR();
|
||||
|
||||
// DEV/TEST ONLY. Read directly from configuration here because authentication scheme
|
||||
|
||||
@@ -1,7 +1,33 @@
|
||||
using System.Text.Json.Serialization;
|
||||
using ZB.MOM.WW.MxGateway.Server.Configuration;
|
||||
|
||||
namespace ZB.MOM.WW.MxGateway.Server.Dashboard;
|
||||
|
||||
/// <summary>
|
||||
/// One dashboard refresh: gateway status, sessions, workers, metrics, faults, API
|
||||
/// keys, the effective configuration, and the Galaxy summary.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <see cref="Configuration"/> is excluded from JSON serialization. It is
|
||||
/// startup-static, yet it is the largest part of the record, and the only wire this
|
||||
/// record crosses is the <c>/hubs/snapshot</c> tick sent every snapshot interval.
|
||||
/// Remote clients receive it once per connection instead, as
|
||||
/// <see cref="Hubs.DashboardSnapshotHub.ConfigurationMessage"/> (or on demand via
|
||||
/// <see cref="Hubs.DashboardSnapshotHub.GetConfiguration"/>). In-process pages read the
|
||||
/// property directly and are unaffected.
|
||||
/// </remarks>
|
||||
/// <param name="GeneratedAt">When the snapshot was built.</param>
|
||||
/// <param name="GatewayStartedAt">When the gateway process started.</param>
|
||||
/// <param name="GatewayUptime">Elapsed time since start.</param>
|
||||
/// <param name="GatewayStatus">Aggregate health-check status, or <c>Unknown</c>.</param>
|
||||
/// <param name="GatewayVersion">Server assembly version.</param>
|
||||
/// <param name="Sessions">Most recent sessions, newest first.</param>
|
||||
/// <param name="Workers">Workers that are not closed.</param>
|
||||
/// <param name="Metrics">Gateway counters and gauges.</param>
|
||||
/// <param name="Faults">Recent session and worker faults.</param>
|
||||
/// <param name="ApiKeys">API key summaries (no secrets).</param>
|
||||
/// <param name="Configuration">Effective configuration; not serialized on the hub tick.</param>
|
||||
/// <param name="Galaxy">Galaxy hierarchy cache summary.</param>
|
||||
public sealed record DashboardSnapshot(
|
||||
DateTimeOffset GeneratedAt,
|
||||
DateTimeOffset GatewayStartedAt,
|
||||
@@ -13,5 +39,5 @@ public sealed record DashboardSnapshot(
|
||||
IReadOnlyList<DashboardMetricSummary> Metrics,
|
||||
IReadOnlyList<DashboardFaultSummary> Faults,
|
||||
IReadOnlyList<DashboardApiKeySummary> ApiKeys,
|
||||
EffectiveGatewayConfiguration Configuration,
|
||||
[property: JsonIgnore] EffectiveGatewayConfiguration Configuration,
|
||||
DashboardGalaxySummary Galaxy);
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
using System.Runtime.CompilerServices;
|
||||
using Microsoft.Extensions.Diagnostics.HealthChecks;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using Microsoft.Extensions.Options;
|
||||
@@ -14,7 +15,16 @@ namespace ZB.MOM.WW.MxGateway.Server.Dashboard;
|
||||
|
||||
public sealed class DashboardSnapshotService : IDashboardSnapshotService
|
||||
{
|
||||
private const string HealthyStatus = "Healthy";
|
||||
/// <summary>
|
||||
/// Status shown before the first health evaluation completes, when the health
|
||||
/// evaluation fails, and when no <see cref="HealthCheckService"/> is registered.
|
||||
/// Deliberately not "Healthy": claiming health that was never checked is how the
|
||||
/// dashboard used to show a green gateway status throughout an outage.
|
||||
/// </summary>
|
||||
internal const string UnknownStatus = "Unknown";
|
||||
|
||||
/// <summary>Upper bound on one health evaluation, so a hung check cannot stall the snapshot tick.</summary>
|
||||
private static readonly TimeSpan GatewayStatusRefreshTimeout = TimeSpan.FromSeconds(2);
|
||||
|
||||
/// <summary>
|
||||
/// Minimum spacing between API key list reads. The list is a SQLite query whose
|
||||
@@ -39,7 +49,10 @@ public sealed class DashboardSnapshotService : IDashboardSnapshotService
|
||||
private readonly int _recentFaultLimit;
|
||||
private readonly int _recentSessionLimit;
|
||||
private readonly ILogger<DashboardSnapshotService> _logger;
|
||||
private readonly HealthCheckService? _healthCheckService;
|
||||
private readonly SemaphoreSlim _apiKeySummaryRefreshGate = new(1, 1);
|
||||
private readonly SemaphoreSlim _gatewayStatusRefreshGate = new(1, 1);
|
||||
private string _gatewayStatus = UnknownStatus;
|
||||
private IReadOnlyList<DashboardApiKeySummary> _apiKeySummaries = Array.Empty<DashboardApiKeySummary>();
|
||||
private long _apiKeySummariesRefreshedAtTicks = NeverRefreshedTicks;
|
||||
// The effective configuration is built from IOptions<GatewayOptions> and is startup-static:
|
||||
@@ -65,6 +78,11 @@ public sealed class DashboardSnapshotService : IDashboardSnapshotService
|
||||
/// <param name="options">Gateway configuration options.</param>
|
||||
/// <param name="timeProvider">Provider for current time; defaults to system time.</param>
|
||||
/// <param name="logger">Optional logger for the dashboard snapshot service.</param>
|
||||
/// <param name="healthCheckService">
|
||||
/// The gateway's registered health checks. <see cref="DashboardSnapshot.GatewayStatus"/>
|
||||
/// is the aggregate status of one evaluation per snapshot tick; without it the
|
||||
/// status stays <see cref="UnknownStatus"/>.
|
||||
/// </param>
|
||||
public DashboardSnapshotService(
|
||||
ISessionRegistry sessionRegistry,
|
||||
GatewayMetrics metrics,
|
||||
@@ -73,7 +91,8 @@ public sealed class DashboardSnapshotService : IDashboardSnapshotService
|
||||
IApiKeyAdminStore apiKeyAdminStore,
|
||||
IOptions<GatewayOptions> options,
|
||||
TimeProvider? timeProvider = null,
|
||||
ILogger<DashboardSnapshotService>? logger = null)
|
||||
ILogger<DashboardSnapshotService>? logger = null,
|
||||
HealthCheckService? healthCheckService = null)
|
||||
{
|
||||
_sessionRegistry = sessionRegistry ?? throw new ArgumentNullException(nameof(sessionRegistry));
|
||||
_metrics = metrics ?? throw new ArgumentNullException(nameof(metrics));
|
||||
@@ -88,6 +107,7 @@ public sealed class DashboardSnapshotService : IDashboardSnapshotService
|
||||
_recentFaultLimit = options.Value.Dashboard.RecentFaultLimit;
|
||||
_recentSessionLimit = options.Value.Dashboard.RecentSessionLimit;
|
||||
_logger = logger ?? NullLogger<DashboardSnapshotService>.Instance;
|
||||
_healthCheckService = healthCheckService;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
@@ -111,7 +131,7 @@ public sealed class DashboardSnapshotService : IDashboardSnapshotService
|
||||
GeneratedAt: generatedAt,
|
||||
GatewayStartedAt: _gatewayStartedAt,
|
||||
GatewayUptime: generatedAt - _gatewayStartedAt,
|
||||
GatewayStatus: HealthyStatus,
|
||||
GatewayStatus: Volatile.Read(ref _gatewayStatus),
|
||||
GatewayVersion: typeof(DashboardSnapshotService).Assembly.GetName().Version?.ToString() ?? "unknown",
|
||||
Sessions: sessionSummaries,
|
||||
Workers: workerSummaries,
|
||||
@@ -171,6 +191,7 @@ public sealed class DashboardSnapshotService : IDashboardSnapshotService
|
||||
}
|
||||
|
||||
await RefreshApiKeySummariesAsync(cancellationToken).ConfigureAwait(false);
|
||||
await RefreshGatewayStatusAsync(cancellationToken).ConfigureAwait(false);
|
||||
yield return GetSnapshot();
|
||||
|
||||
using PeriodicTimer timer = new(_snapshotInterval, _timeProvider);
|
||||
@@ -192,10 +213,57 @@ public sealed class DashboardSnapshotService : IDashboardSnapshotService
|
||||
}
|
||||
|
||||
await RefreshApiKeySummariesAsync(cancellationToken).ConfigureAwait(false);
|
||||
await RefreshGatewayStatusAsync(cancellationToken).ConfigureAwait(false);
|
||||
yield return GetSnapshot();
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Evaluates the registered health checks once and caches the aggregate status
|
||||
/// for <see cref="GetSnapshot"/>, which is synchronous and also serves page seeds
|
||||
/// and hub connects between ticks. One evaluation per tick, not per
|
||||
/// <see cref="GetSnapshot"/> call: the checks include an auth-store query, and the
|
||||
/// tick is already the cadence at which the dashboard refreshes.
|
||||
/// </summary>
|
||||
/// <param name="cancellationToken">Token to cancel the evaluation.</param>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
private async Task RefreshGatewayStatusAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
if (_healthCheckService is null)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
// Zero-timeout gate: a concurrent enumeration (the hub publisher and the page
|
||||
// feed each run one) skips rather than queueing a second evaluation.
|
||||
if (!await _gatewayStatusRefreshGate.WaitAsync(0, cancellationToken).ConfigureAwait(false))
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
using CancellationTokenSource timeout = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
|
||||
timeout.CancelAfter(GatewayStatusRefreshTimeout);
|
||||
HealthReport report = await _healthCheckService.CheckHealthAsync(timeout.Token).ConfigureAwait(false);
|
||||
Volatile.Write(ref _gatewayStatus, report.Status.ToString());
|
||||
}
|
||||
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
|
||||
{
|
||||
throw;
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
// A health evaluation that cannot complete is not evidence of health.
|
||||
Volatile.Write(ref _gatewayStatus, UnknownStatus);
|
||||
_logger.LogDebug(exception, "Dashboard gateway health evaluation failed.");
|
||||
}
|
||||
finally
|
||||
{
|
||||
_gatewayStatusRefreshGate.Release();
|
||||
}
|
||||
}
|
||||
|
||||
private static DashboardSessionSummary CreateSessionSummary(
|
||||
GatewaySession session,
|
||||
GatewayMetricsSnapshot metricsSnapshot)
|
||||
|
||||
@@ -0,0 +1,14 @@
|
||||
namespace ZB.MOM.WW.MxGateway.Server.Dashboard.Hubs;
|
||||
|
||||
/// <summary>
|
||||
/// Payload of the <see cref="AlarmsHub.ResetMessage"/> frame: the publisher has
|
||||
/// (re)subscribed to the central alarm monitor, which primes the subscription with
|
||||
/// a full active-alarm snapshot. A client discards its accumulated alarm state on
|
||||
/// receipt and rebuilds it from the <c>active_alarm</c> rows up to the next
|
||||
/// <c>snapshot_complete</c>.
|
||||
/// </summary>
|
||||
/// <param name="Generation">
|
||||
/// Monotonic subscription counter for this gateway process, starting at 1. Lets a
|
||||
/// client tell a resubscribe from a duplicate delivery.
|
||||
/// </param>
|
||||
public sealed record AlarmFeedReset(long Generation);
|
||||
@@ -23,11 +23,19 @@ namespace ZB.MOM.WW.MxGateway.Server.Dashboard.Hubs;
|
||||
[Authorize(Policy = DashboardAuthenticationDefaults.AlarmsPolicy)]
|
||||
public sealed class AlarmsHub : Hub
|
||||
{
|
||||
/// <summary>Group every admitted connection joins on connect.</summary>
|
||||
public const string AllAlarmsGroup = "__alarms__";
|
||||
|
||||
/// <summary>Method name used to push <c>AlarmFeedMessage</c> values to clients.</summary>
|
||||
public const string AlarmMessage = "AlarmFeed";
|
||||
|
||||
/// <summary>
|
||||
/// Method name of the <see cref="AlarmFeedReset"/> marker sent ahead of every
|
||||
/// (re)subscription to the central monitor. A client discards its accumulated
|
||||
/// alarm state on receipt and rebuilds it from the snapshot that follows.
|
||||
/// </summary>
|
||||
public const string ResetMessage = "AlarmFeedReset";
|
||||
|
||||
/// <inheritdoc />
|
||||
public override async Task OnConnectedAsync()
|
||||
{
|
||||
|
||||
@@ -31,18 +31,69 @@ namespace ZB.MOM.WW.MxGateway.Server.Dashboard.Hubs;
|
||||
/// narrowing this loop could apply even in principle. Adding a filter here would imply a
|
||||
/// per-caller scope the feed does not carry.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// Each broadcast is time-bounded (<see cref="DefaultSendTimeout"/>). A group send
|
||||
/// completes only once every member accepted the frame, so awaiting it unbounded
|
||||
/// let one slow browser stall this loop; the stalled loop stopped draining its
|
||||
/// monitor subscription, the monitor dropped it as a slow subscriber, and the
|
||||
/// resubscribe replayed the whole active-alarm snapshot to every client. A send that
|
||||
/// misses the bound is abandoned and the loop moves on.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// Every subscription — the first and each resubscribe — is preceded by an
|
||||
/// <see cref="AlarmsHub.ResetMessage"/> frame. The monitor primes every new
|
||||
/// subscription with a full snapshot, so a client that keeps accumulating
|
||||
/// <c>active_alarm</c> rows across a resubscribe would double-count or keep alarms
|
||||
/// that cleared while the feed was down; the marker tells it to discard its state
|
||||
/// and rebuild from the snapshot that follows.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
/// <param name="alarmService">The gateway's central alarm feed.</param>
|
||||
/// <param name="hubContext">Hub context used to broadcast to the alarms group.</param>
|
||||
/// <param name="options">Gateway options supplying <c>Dashboard:ShowTagValues</c>.</param>
|
||||
/// <param name="logger">Logger for best-effort broadcast failures.</param>
|
||||
public sealed class AlarmsHubPublisher(
|
||||
IGatewayAlarmService alarmService,
|
||||
IHubContext<AlarmsHub> hubContext,
|
||||
IOptions<GatewayOptions> options,
|
||||
ILogger<AlarmsHubPublisher> logger) : BackgroundService
|
||||
public sealed class AlarmsHubPublisher : BackgroundService
|
||||
{
|
||||
private readonly bool _showTagValues = options.Value.Dashboard.ShowTagValues;
|
||||
/// <summary>Upper bound on one alarm broadcast before it is abandoned.</summary>
|
||||
internal static readonly TimeSpan DefaultSendTimeout = TimeSpan.FromSeconds(2);
|
||||
|
||||
private readonly IGatewayAlarmService _alarmService;
|
||||
private readonly IHubContext<AlarmsHub> _hubContext;
|
||||
private readonly ILogger<AlarmsHubPublisher> _logger;
|
||||
private readonly TimeSpan _sendTimeout;
|
||||
private readonly bool _showTagValues;
|
||||
private long _subscriptionGeneration;
|
||||
|
||||
/// <summary>Initializes a new instance of the <see cref="AlarmsHubPublisher"/> class.</summary>
|
||||
/// <param name="alarmService">The gateway's central alarm feed.</param>
|
||||
/// <param name="hubContext">Hub context used to broadcast to the alarms group.</param>
|
||||
/// <param name="options">Gateway options supplying <c>Dashboard:ShowTagValues</c>.</param>
|
||||
/// <param name="logger">Logger for best-effort broadcast failures.</param>
|
||||
public AlarmsHubPublisher(
|
||||
IGatewayAlarmService alarmService,
|
||||
IHubContext<AlarmsHub> hubContext,
|
||||
IOptions<GatewayOptions> options,
|
||||
ILogger<AlarmsHubPublisher> logger)
|
||||
: this(alarmService, hubContext, options, logger, DefaultSendTimeout)
|
||||
{
|
||||
}
|
||||
|
||||
/// <summary>Initializes a new instance with a custom send timeout (test hook).</summary>
|
||||
/// <param name="alarmService">The gateway's central alarm feed.</param>
|
||||
/// <param name="hubContext">Hub context used to broadcast to the alarms group.</param>
|
||||
/// <param name="options">Gateway options supplying <c>Dashboard:ShowTagValues</c>.</param>
|
||||
/// <param name="logger">Logger for best-effort broadcast failures.</param>
|
||||
/// <param name="sendTimeout">Upper bound on one broadcast.</param>
|
||||
internal AlarmsHubPublisher(
|
||||
IGatewayAlarmService alarmService,
|
||||
IHubContext<AlarmsHub> hubContext,
|
||||
IOptions<GatewayOptions> options,
|
||||
ILogger<AlarmsHubPublisher> logger,
|
||||
TimeSpan sendTimeout)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(options);
|
||||
_alarmService = alarmService ?? throw new ArgumentNullException(nameof(alarmService));
|
||||
_hubContext = hubContext ?? throw new ArgumentNullException(nameof(hubContext));
|
||||
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
|
||||
_sendTimeout = sendTimeout;
|
||||
_showTagValues = options.Value.Dashboard.ShowTagValues;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
|
||||
@@ -54,7 +105,16 @@ public sealed class AlarmsHubPublisher(
|
||||
{
|
||||
try
|
||||
{
|
||||
await foreach (AlarmFeedMessage message in alarmService
|
||||
// Sent before the subscription opens, so it reaches clients ahead of
|
||||
// the snapshot the monitor primes the new subscription with.
|
||||
long generation = Interlocked.Increment(ref _subscriptionGeneration);
|
||||
await BroadcastAsync(
|
||||
AlarmsHub.ResetMessage,
|
||||
new AlarmFeedReset(generation),
|
||||
stoppingToken)
|
||||
.ConfigureAwait(false);
|
||||
|
||||
await foreach (AlarmFeedMessage message in _alarmService
|
||||
.StreamAsync(alarmFilterPrefix: null, stoppingToken)
|
||||
.ConfigureAwait(false))
|
||||
{
|
||||
@@ -63,18 +123,8 @@ public sealed class AlarmsHubPublisher(
|
||||
break;
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
AlarmFeedMessage outbound = _showTagValues ? message : RedactValues(message);
|
||||
await hubContext.Clients
|
||||
.Group(AlarmsHub.AllAlarmsGroup)
|
||||
.SendAsync(AlarmsHub.AlarmMessage, outbound, stoppingToken)
|
||||
.ConfigureAwait(false);
|
||||
}
|
||||
catch (Exception ex) when (ex is not OperationCanceledException)
|
||||
{
|
||||
logger.LogWarning(ex, "Alarm broadcast failed; continuing.");
|
||||
}
|
||||
AlarmFeedMessage outbound = _showTagValues ? message : RedactValues(message);
|
||||
await BroadcastAsync(AlarmsHub.AlarmMessage, outbound, stoppingToken).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
@@ -83,7 +133,7 @@ public sealed class AlarmsHubPublisher(
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
logger.LogWarning(ex, "Alarm subscription faulted; reconnecting in 5s.");
|
||||
_logger.LogWarning(ex, "Alarm subscription faulted; reconnecting in 5s.");
|
||||
try
|
||||
{
|
||||
await Task.Delay(TimeSpan.FromSeconds(5), stoppingToken).ConfigureAwait(false);
|
||||
@@ -96,6 +146,40 @@ public sealed class AlarmsHubPublisher(
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Sends one frame to the alarms group within the send timeout. A slow or failed
|
||||
/// send is logged and dropped so the feed keeps draining; only host shutdown
|
||||
/// propagates.
|
||||
/// </summary>
|
||||
/// <param name="method">Client method name.</param>
|
||||
/// <param name="argument">The frame payload.</param>
|
||||
/// <param name="stoppingToken">Host shutdown token.</param>
|
||||
/// <returns>A task that completes when the send finished or was abandoned.</returns>
|
||||
private async Task BroadcastAsync(string method, object argument, CancellationToken stoppingToken)
|
||||
{
|
||||
try
|
||||
{
|
||||
bool sent = await BoundedHubSend.SendAsync(
|
||||
_hubContext.Clients.Group(AlarmsHub.AllAlarmsGroup),
|
||||
method,
|
||||
argument,
|
||||
_sendTimeout,
|
||||
stoppingToken)
|
||||
.ConfigureAwait(false);
|
||||
if (!sent)
|
||||
{
|
||||
_logger.LogDebug(
|
||||
"Alarm broadcast {Method} exceeded {Timeout} and was abandoned; continuing.",
|
||||
method,
|
||||
_sendTimeout);
|
||||
}
|
||||
}
|
||||
catch (Exception ex) when (ex is not OperationCanceledException)
|
||||
{
|
||||
_logger.LogWarning(ex, "Alarm broadcast failed; continuing.");
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Produces a copy of <paramref name="source"/> with the alarm value fields
|
||||
/// cleared, leaving every other field — reference, severity, state, category,
|
||||
|
||||
@@ -0,0 +1,52 @@
|
||||
using Microsoft.AspNetCore.SignalR;
|
||||
|
||||
namespace ZB.MOM.WW.MxGateway.Server.Dashboard.Hubs;
|
||||
|
||||
/// <summary>
|
||||
/// Time-bounded SignalR send shared by the dashboard hub publishers. A group or
|
||||
/// broadcast send completes only once every target connection has accepted the
|
||||
/// frame, so one remote viewer with a full transport buffer would otherwise hold
|
||||
/// the publisher's single drain loop for as long as that viewer stays slow.
|
||||
/// </summary>
|
||||
internal static class BoundedHubSend
|
||||
{
|
||||
/// <summary>
|
||||
/// Sends <paramref name="argument"/> as <paramref name="method"/> and waits at most
|
||||
/// <paramref name="timeout"/> for it. The timeout cancels the send's own token, so
|
||||
/// the lifetime manager abandons the pending write to the slow connection, and the
|
||||
/// wait itself is bounded too, so a proxy that ignores the token cannot stall the
|
||||
/// caller either.
|
||||
/// </summary>
|
||||
/// <param name="proxy">Target client proxy (a group or all connections).</param>
|
||||
/// <param name="method">Client method name.</param>
|
||||
/// <param name="argument">The single message argument.</param>
|
||||
/// <param name="timeout">Upper bound on the send.</param>
|
||||
/// <param name="stoppingToken">Host shutdown token; its cancellation propagates.</param>
|
||||
/// <returns>
|
||||
/// <see langword="true"/> when the send completed in time; <see langword="false"/>
|
||||
/// when it was abandoned at the timeout.
|
||||
/// </returns>
|
||||
/// <exception cref="OperationCanceledException"><paramref name="stoppingToken"/> was cancelled.</exception>
|
||||
public static async Task<bool> SendAsync(
|
||||
IClientProxy proxy,
|
||||
string method,
|
||||
object argument,
|
||||
TimeSpan timeout,
|
||||
CancellationToken stoppingToken)
|
||||
{
|
||||
using CancellationTokenSource bounded = CancellationTokenSource.CreateLinkedTokenSource(stoppingToken);
|
||||
bounded.CancelAfter(timeout);
|
||||
|
||||
try
|
||||
{
|
||||
await proxy.SendAsync(method, argument, bounded.Token)
|
||||
.WaitAsync(bounded.Token)
|
||||
.ConfigureAwait(false);
|
||||
return true;
|
||||
}
|
||||
catch (OperationCanceledException) when (!stoppingToken.IsCancellationRequested)
|
||||
{
|
||||
return false;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -10,10 +10,11 @@ namespace ZB.MOM.WW.MxGateway.Server.Dashboard.Hubs;
|
||||
/// Broadcasts MxEvents to the two dashboard audiences for a session: remote
|
||||
/// <see cref="EventsHub"/> clients subscribed to the session's group, and
|
||||
/// in-process subscribers opened through
|
||||
/// <see cref="IDashboardSessionEventSubscriber.Subscribe"/>. Fire-and-forget: we
|
||||
/// hand the send to the hub context and return immediately so the source gRPC
|
||||
/// stream is never blocked. Errors are logged once and dropped — keeping the
|
||||
/// SignalR mirror best-effort preserves the gRPC contract that exists today.
|
||||
/// <see cref="IDashboardSessionEventSubscriber.Subscribe"/>. <see cref="Publish"/>
|
||||
/// never blocks: both audiences are fed through bounded drop-oldest queues, so the
|
||||
/// source gRPC stream is never held up by a slow dashboard viewer. Errors are
|
||||
/// logged and dropped — keeping the SignalR mirror best-effort preserves the gRPC
|
||||
/// contract.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// When <c>MxGateway:Dashboard:ShowTagValues</c> is false (the default), tag
|
||||
@@ -26,31 +27,45 @@ namespace ZB.MOM.WW.MxGateway.Server.Dashboard.Hubs;
|
||||
/// <see cref="EventsHub"/>), while the redaction decides what a permitted
|
||||
/// subscriber sees — so the value-leak seam stays closed whatever the ACL
|
||||
/// admits.
|
||||
/// <para>
|
||||
/// Hub delivery goes through one <see cref="HubPump"/> per watched session: a
|
||||
/// <see cref="HubQueueCapacity"/>-deep drop-oldest channel drained by a single task
|
||||
/// that sends one event at a time, each send time-bounded. An earlier version fired
|
||||
/// a group send per event and returned, so a slow remote viewer accumulated an
|
||||
/// unbounded number of pending sends — each holding a cloned event — inside the
|
||||
/// SignalR lifetime manager. The pump is started by the first event published while
|
||||
/// a hub connection watches the session and torn down when
|
||||
/// <see cref="EventsHubViewerRegistry.HubViewersReleased"/> reports the last one gone.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
/// <param name="hubContext">Hub context used to send to the session's group.</param>
|
||||
/// <param name="viewerRegistry">
|
||||
/// Live-subscriber registry consulted before any per-event work is done. Both
|
||||
/// audiences register here — hub connections by their SignalR connection id,
|
||||
/// in-process subscriptions by a synthetic one — so the gate stays a single
|
||||
/// source of truth.
|
||||
/// </param>
|
||||
/// <param name="options">Gateway options supplying <c>Dashboard:ShowTagValues</c>.</param>
|
||||
/// <param name="logger">Logger for best-effort mirror failures.</param>
|
||||
public sealed class DashboardEventBroadcaster(
|
||||
IHubContext<EventsHub> hubContext,
|
||||
EventsHubViewerRegistry viewerRegistry,
|
||||
IOptions<GatewayOptions> options,
|
||||
ILogger<DashboardEventBroadcaster> logger) : IDashboardEventBroadcaster, IDashboardSessionEventSubscriber
|
||||
public sealed class DashboardEventBroadcaster : IDashboardEventBroadcaster, IDashboardSessionEventSubscriber, IDisposable
|
||||
{
|
||||
/// <summary>
|
||||
/// Queue depth per in-process subscriber. The consumer is a Blazor page
|
||||
/// rendering the newest handful of events, so a burst it cannot keep up with
|
||||
/// is dropped oldest-first rather than allowed to grow — same best-effort
|
||||
/// contract the SignalR mirror already has.
|
||||
/// contract the SignalR mirror has.
|
||||
/// </summary>
|
||||
private const int InProcessQueueCapacity = 256;
|
||||
|
||||
private readonly bool _showTagValues = options.Value.Dashboard.ShowTagValues;
|
||||
/// <summary>
|
||||
/// Queue depth of a session's hub-delivery pump. Same depth and drop policy as
|
||||
/// the in-process feed: the remote viewer is a live view, so a burst it cannot
|
||||
/// keep up with loses its oldest events rather than buffering without bound.
|
||||
/// </summary>
|
||||
internal const int HubQueueCapacity = 256;
|
||||
|
||||
private static readonly TimeSpan DefaultHubSendTimeout = TimeSpan.FromSeconds(5);
|
||||
|
||||
private readonly IHubContext<EventsHub> _hubContext;
|
||||
|
||||
private readonly EventsHubViewerRegistry _viewerRegistry;
|
||||
|
||||
private readonly ILogger<DashboardEventBroadcaster> _logger;
|
||||
|
||||
private readonly TimeSpan _hubSendTimeout;
|
||||
|
||||
private readonly bool _showTagValues;
|
||||
|
||||
private readonly object _syncRoot = new();
|
||||
|
||||
@@ -63,6 +78,9 @@ public sealed class DashboardEventBroadcaster(
|
||||
private readonly Dictionary<string, InProcessSubscription[]> _inProcessSubscribers =
|
||||
new(StringComparer.Ordinal);
|
||||
|
||||
/// <summary>Live hub-delivery pumps per session. Guarded by <see cref="_syncRoot"/>.</summary>
|
||||
private readonly Dictionary<string, HubPump> _hubPumps = new(StringComparer.Ordinal);
|
||||
|
||||
/// <summary>
|
||||
/// Total live in-process subscribers, read without the lock so the common
|
||||
/// case — nobody has a session-details page open — never contends on it.
|
||||
@@ -70,6 +88,61 @@ public sealed class DashboardEventBroadcaster(
|
||||
/// </summary>
|
||||
private int _inProcessSubscriberCount;
|
||||
|
||||
private bool _disposed;
|
||||
|
||||
/// <summary>Initializes a new instance of the <see cref="DashboardEventBroadcaster"/> class.</summary>
|
||||
/// <param name="hubContext">Hub context used to send to the session's group.</param>
|
||||
/// <param name="viewerRegistry">
|
||||
/// Live-subscriber registry consulted before any per-event work is done. Both
|
||||
/// audiences register here — hub connections by their SignalR connection id,
|
||||
/// in-process subscriptions by a synthetic one — so the gate stays a single
|
||||
/// source of truth.
|
||||
/// </param>
|
||||
/// <param name="options">Gateway options supplying <c>Dashboard:ShowTagValues</c>.</param>
|
||||
/// <param name="logger">Logger for best-effort mirror failures.</param>
|
||||
public DashboardEventBroadcaster(
|
||||
IHubContext<EventsHub> hubContext,
|
||||
EventsHubViewerRegistry viewerRegistry,
|
||||
IOptions<GatewayOptions> options,
|
||||
ILogger<DashboardEventBroadcaster> logger)
|
||||
: this(hubContext, viewerRegistry, options, logger, DefaultHubSendTimeout)
|
||||
{
|
||||
}
|
||||
|
||||
/// <summary>Initializes a new instance with a custom hub send timeout (test hook).</summary>
|
||||
/// <param name="hubContext">Hub context used to send to the session's group.</param>
|
||||
/// <param name="viewerRegistry">Live-subscriber registry.</param>
|
||||
/// <param name="options">Gateway options supplying <c>Dashboard:ShowTagValues</c>.</param>
|
||||
/// <param name="logger">Logger for best-effort mirror failures.</param>
|
||||
/// <param name="hubSendTimeout">Upper bound on one group send before it is abandoned.</param>
|
||||
internal DashboardEventBroadcaster(
|
||||
IHubContext<EventsHub> hubContext,
|
||||
EventsHubViewerRegistry viewerRegistry,
|
||||
IOptions<GatewayOptions> options,
|
||||
ILogger<DashboardEventBroadcaster> logger,
|
||||
TimeSpan hubSendTimeout)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(options);
|
||||
_hubContext = hubContext ?? throw new ArgumentNullException(nameof(hubContext));
|
||||
_viewerRegistry = viewerRegistry ?? throw new ArgumentNullException(nameof(viewerRegistry));
|
||||
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
|
||||
_hubSendTimeout = hubSendTimeout;
|
||||
_showTagValues = options.Value.Dashboard.ShowTagValues;
|
||||
_viewerRegistry.HubViewersReleased += OnHubViewersReleased;
|
||||
}
|
||||
|
||||
/// <summary>Gets the number of live hub-delivery pumps (one per hub-watched session).</summary>
|
||||
internal int ActiveHubPumpCount
|
||||
{
|
||||
get
|
||||
{
|
||||
lock (_syncRoot)
|
||||
{
|
||||
return _hubPumps.Count;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public void Publish(string sessionId, MxEvent mxEvent)
|
||||
{
|
||||
@@ -81,46 +154,129 @@ public sealed class DashboardEventBroadcaster(
|
||||
// Every session's dashboard-mirror subscriber calls Publish for every event,
|
||||
// whether or not a browser is on that session's page. Without this gate the
|
||||
// steady state — no dashboard viewer at all — still paid a deep protobuf
|
||||
// clone (redaction is on by default) plus a send to an empty SignalR group
|
||||
// per event. Bail before both.
|
||||
if (!viewerRegistry.HasViewers(sessionId))
|
||||
// clone (redaction is on by default) per event. Bail before it.
|
||||
if (!_viewerRegistry.HasViewers(sessionId))
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
MxEvent outbound = _showTagValues ? mxEvent : RedactValues(mxEvent);
|
||||
|
||||
// In-process delivery first: it is synchronous, cannot throw, and must not be
|
||||
// skipped by the early return the hub send's guard clause takes.
|
||||
DeliverInProcess(sessionId, outbound);
|
||||
|
||||
// Wrap the Task acquisition in a try/catch so a hypothetical synchronous throw
|
||||
// from SendAsync (e.g. an implementation that throws before returning the Task)
|
||||
// cannot escape Publish. The interface contract is never-throw; fire-and-forget.
|
||||
Task send;
|
||||
// In-process-only viewers need no hub send at all; the pump exists only
|
||||
// while a remote connection is in the session's group.
|
||||
if (_viewerRegistry.HasHubViewers(sessionId))
|
||||
{
|
||||
EnqueueForHub(sessionId, outbound);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Stops every hub pump and detaches from the viewer registry. The singleton is
|
||||
/// disposed with the host; a pump's in-flight send is bounded by its timeout.
|
||||
/// </summary>
|
||||
public void Dispose()
|
||||
{
|
||||
HubPump[] pumps;
|
||||
lock (_syncRoot)
|
||||
{
|
||||
if (_disposed)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
_disposed = true;
|
||||
pumps = [.. _hubPumps.Values];
|
||||
_hubPumps.Clear();
|
||||
}
|
||||
|
||||
_viewerRegistry.HubViewersReleased -= OnHubViewersReleased;
|
||||
foreach (HubPump pump in pumps)
|
||||
{
|
||||
pump.Complete();
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Queues an event on the session's hub pump, starting the pump on first use.
|
||||
/// The write happens under the lock so it cannot land on a pump that a
|
||||
/// concurrent teardown has already completed.
|
||||
/// </summary>
|
||||
/// <param name="sessionId">Session the event belongs to.</param>
|
||||
/// <param name="outbound">The event as the dashboard should see it.</param>
|
||||
private void EnqueueForHub(string sessionId, MxEvent outbound)
|
||||
{
|
||||
lock (_syncRoot)
|
||||
{
|
||||
// Re-check under the lock: the last viewer may have left between the
|
||||
// caller's unlocked check and here, and its release handler (which takes
|
||||
// this lock) would then never see a pump created after it ran.
|
||||
if (_disposed || !_viewerRegistry.HasHubViewers(sessionId))
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
if (!_hubPumps.TryGetValue(sessionId, out HubPump? pump))
|
||||
{
|
||||
pump = new HubPump(this, sessionId, HubQueueCapacity);
|
||||
_hubPumps[sessionId] = pump;
|
||||
}
|
||||
|
||||
pump.TryWrite(outbound);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Tears down a session's hub pump once its last hub viewer has gone. Re-checks
|
||||
/// the registry under the lock: a viewer that re-subscribed in between keeps the
|
||||
/// pump, and one that arrives after the teardown gets a fresh pump on the next
|
||||
/// event.
|
||||
/// </summary>
|
||||
/// <param name="sessionId">Session whose last hub viewer left.</param>
|
||||
private void OnHubViewersReleased(string sessionId)
|
||||
{
|
||||
HubPump? pump;
|
||||
lock (_syncRoot)
|
||||
{
|
||||
if (_viewerRegistry.HasHubViewers(sessionId) || !_hubPumps.Remove(sessionId, out pump))
|
||||
{
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
pump.Complete();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Sends one event to the session's group, bounded by the hub send timeout.
|
||||
/// Never throws: a failed or abandoned send is logged and the pump moves on.
|
||||
/// </summary>
|
||||
/// <param name="sessionId">Session whose group receives the event.</param>
|
||||
/// <param name="outbound">The event to send.</param>
|
||||
/// <returns>A task that completes when the send finished or was abandoned.</returns>
|
||||
private async Task SendToHubAsync(string sessionId, MxEvent outbound)
|
||||
{
|
||||
try
|
||||
{
|
||||
send = hubContext.Clients
|
||||
.Group(EventsHub.GroupName(sessionId))
|
||||
.SendAsync(EventsHub.EventMessage, outbound);
|
||||
bool sent = await BoundedHubSend.SendAsync(
|
||||
_hubContext.Clients.Group(EventsHub.GroupName(sessionId)),
|
||||
EventsHub.EventMessage,
|
||||
outbound,
|
||||
_hubSendTimeout,
|
||||
CancellationToken.None)
|
||||
.ConfigureAwait(false);
|
||||
if (!sent)
|
||||
{
|
||||
_logger.LogDebug(
|
||||
"Dashboard event mirror to session {SessionId} timed out after {Timeout}; continuing.",
|
||||
sessionId,
|
||||
_hubSendTimeout);
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
logger.LogDebug(ex, "Dashboard event mirror to session {SessionId} threw synchronously.", sessionId);
|
||||
return;
|
||||
}
|
||||
|
||||
if (!send.IsCompletedSuccessfully)
|
||||
{
|
||||
_ = send.ContinueWith(
|
||||
t =>
|
||||
{
|
||||
if (t.Exception is { } ex)
|
||||
{
|
||||
logger.LogDebug(ex, "Dashboard event mirror to session {SessionId} failed.", sessionId);
|
||||
}
|
||||
},
|
||||
TaskScheduler.Default);
|
||||
_logger.LogDebug(ex, "Dashboard event mirror to session {SessionId} failed.", sessionId);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -133,7 +289,7 @@ public sealed class DashboardEventBroadcaster(
|
||||
// usable for a subscriber that has no SignalR connection behind it. The
|
||||
// "inproc-" prefix cannot collide with a SignalR connection id and makes the
|
||||
// origin obvious in a debugger.
|
||||
string connectionId = "inproc-" + Guid.NewGuid().ToString("N");
|
||||
string connectionId = EventsHubViewerRegistry.InProcessConnectionIdPrefix + Guid.NewGuid().ToString("N");
|
||||
InProcessSubscription subscription = new(this, sessionId, connectionId, InProcessQueueCapacity);
|
||||
|
||||
// Register before the subscriber becomes a delivery target, exactly as
|
||||
@@ -142,7 +298,7 @@ public sealed class DashboardEventBroadcaster(
|
||||
// Publish's gate still reports the session unwatched, silently dropping events
|
||||
// it should receive. The cost of this order is at worst a redaction clone that
|
||||
// reaches nobody for the width of the window.
|
||||
viewerRegistry.AddViewer(connectionId, sessionId);
|
||||
_viewerRegistry.AddViewer(connectionId, sessionId);
|
||||
|
||||
lock (_syncRoot)
|
||||
{
|
||||
@@ -227,12 +383,12 @@ public sealed class DashboardEventBroadcaster(
|
||||
}
|
||||
}
|
||||
|
||||
viewerRegistry.RemoveViewer(subscription.ConnectionId, subscription.SessionId);
|
||||
_viewerRegistry.RemoveViewer(subscription.ConnectionId, subscription.SessionId);
|
||||
|
||||
// The synthetic connection id is used once and never reconnects, so nothing
|
||||
// else will ever call ReleaseConnection for it; without this the registry
|
||||
// would retain an empty per-connection entry per subscription ever opened.
|
||||
viewerRegistry.ReleaseConnection(subscription.ConnectionId);
|
||||
_viewerRegistry.ReleaseConnection(subscription.ConnectionId);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
@@ -328,4 +484,56 @@ public sealed class DashboardEventBroadcaster(
|
||||
_channel.Writer.TryComplete();
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// One session's hub-delivery pump: a bounded drop-oldest channel and the single
|
||||
/// task that drains it into the session's SignalR group, one bounded send at a
|
||||
/// time. The drain is the only reader, so at most one send per session is ever
|
||||
/// pending in the lifetime manager.
|
||||
/// </summary>
|
||||
private sealed class HubPump
|
||||
{
|
||||
private readonly Channel<MxEvent> _channel;
|
||||
|
||||
/// <summary>Initializes a new instance of the <see cref="HubPump"/> class and starts its drain.</summary>
|
||||
/// <param name="owner">Broadcaster whose send path the drain uses.</param>
|
||||
/// <param name="sessionId">Session whose group the pump feeds.</param>
|
||||
/// <param name="capacity">Queue depth before the oldest queued event is dropped.</param>
|
||||
internal HubPump(DashboardEventBroadcaster owner, string sessionId, int capacity)
|
||||
{
|
||||
_channel = Channel.CreateBounded<MxEvent>(new BoundedChannelOptions(capacity)
|
||||
{
|
||||
// DropOldest, not Wait: Publish runs on the gRPC event pipeline and
|
||||
// must never block, and the newest events are the ones a live view wants.
|
||||
FullMode = BoundedChannelFullMode.DropOldest,
|
||||
SingleReader = true,
|
||||
SingleWriter = false,
|
||||
});
|
||||
|
||||
// Task.Run so the drain never starts inline on the publishing thread.
|
||||
Completion = Task.Run(() => DrainAsync(owner, sessionId));
|
||||
}
|
||||
|
||||
/// <summary>Gets the drain task; it completes once the channel is completed and emptied.</summary>
|
||||
internal Task Completion { get; }
|
||||
|
||||
/// <summary>Queues an event, dropping the oldest queued one when full. Never blocks.</summary>
|
||||
/// <param name="mxEvent">The event to queue.</param>
|
||||
internal void TryWrite(MxEvent mxEvent) => _channel.Writer.TryWrite(mxEvent);
|
||||
|
||||
/// <summary>Completes the channel so the drain exits once it is empty.</summary>
|
||||
internal void Complete() => _channel.Writer.TryComplete();
|
||||
|
||||
private async Task DrainAsync(DashboardEventBroadcaster owner, string sessionId)
|
||||
{
|
||||
ChannelReader<MxEvent> reader = _channel.Reader;
|
||||
while (await reader.WaitToReadAsync().ConfigureAwait(false))
|
||||
{
|
||||
while (reader.TryRead(out MxEvent? next))
|
||||
{
|
||||
await owner.SendToHubAsync(sessionId, next).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,13 +1,15 @@
|
||||
using Microsoft.AspNetCore.Authorization;
|
||||
using Microsoft.AspNetCore.SignalR;
|
||||
using ZB.MOM.WW.MxGateway.Server.Configuration;
|
||||
|
||||
namespace ZB.MOM.WW.MxGateway.Server.Dashboard.Hubs;
|
||||
|
||||
/// <summary>
|
||||
/// SignalR hub that pushes a fresh <see cref="DashboardSnapshot"/> on every
|
||||
/// snapshot refresh. New connections receive the current snapshot
|
||||
/// immediately via <see cref="OnConnectedAsync"/>; subsequent refreshes are
|
||||
/// broadcast by <see cref="DashboardSnapshotPublisher"/>.
|
||||
/// snapshot refresh. New connections receive the effective configuration and the
|
||||
/// current snapshot immediately via <see cref="OnConnectedAsync"/>; subsequent
|
||||
/// refreshes are broadcast by <see cref="DashboardSnapshotPublisher"/> without the
|
||||
/// configuration, which is startup-static and serialized only once per connection.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// Connections are counted into <see cref="DashboardSnapshotHubConnectionCounter"/>
|
||||
@@ -22,13 +24,28 @@ public sealed class DashboardSnapshotHub(
|
||||
/// <summary>Method name used to push snapshot updates to clients.</summary>
|
||||
public const string SnapshotMessage = "SnapshotUpdated";
|
||||
|
||||
/// <summary>
|
||||
/// Method name of the one-per-connection effective-configuration push. Snapshot
|
||||
/// ticks omit the configuration, so a client keeps the copy it got here.
|
||||
/// </summary>
|
||||
public const string ConfigurationMessage = "ConfigurationUpdated";
|
||||
|
||||
/// <summary>
|
||||
/// Returns the effective gateway configuration — the same object
|
||||
/// <see cref="ConfigurationMessage"/> carries — for a client that needs it again.
|
||||
/// </summary>
|
||||
/// <returns>The effective configuration.</returns>
|
||||
public EffectiveGatewayConfiguration GetConfiguration() => snapshotService.GetSnapshot().Configuration;
|
||||
|
||||
/// <inheritdoc />
|
||||
public override async Task OnConnectedAsync()
|
||||
{
|
||||
// Count the viewer before seeding it, so the publisher resumes its tick
|
||||
// no later than the first snapshot this connection renders.
|
||||
connectionCounter.Increment();
|
||||
await Clients.Caller.SendAsync(SnapshotMessage, snapshotService.GetSnapshot()).ConfigureAwait(false);
|
||||
DashboardSnapshot snapshot = snapshotService.GetSnapshot();
|
||||
await Clients.Caller.SendAsync(ConfigurationMessage, snapshot.Configuration).ConfigureAwait(false);
|
||||
await Clients.Caller.SendAsync(SnapshotMessage, snapshot).ConfigureAwait(false);
|
||||
await base.OnConnectedAsync().ConfigureAwait(false);
|
||||
}
|
||||
|
||||
|
||||
@@ -26,11 +26,20 @@ namespace ZB.MOM.WW.MxGateway.Server.Dashboard.Hubs;
|
||||
/// connected. While the count is zero the publisher does not advance the snapshot
|
||||
/// enumerator at all, so the producing iterator stays suspended and builds nothing.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// Each broadcast is bounded by a send timeout (one second by default, the default
|
||||
/// snapshot interval). A broadcast completes only once every connection accepted the
|
||||
/// frame, so awaiting it unbounded let one slow remote client delay every tick for
|
||||
/// everyone; a send that misses the bound is abandoned and the next tick goes out on
|
||||
/// time. The tick omits the effective configuration (see
|
||||
/// <see cref="DashboardSnapshot"/>), so each broadcast is the volatile state only.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
public sealed class DashboardSnapshotPublisher : BackgroundService
|
||||
{
|
||||
private static readonly TimeSpan DefaultReconnectDelay = TimeSpan.FromSeconds(5);
|
||||
private static readonly TimeSpan DefaultIdlePollInterval = TimeSpan.FromSeconds(1);
|
||||
private static readonly TimeSpan DefaultSendTimeout = TimeSpan.FromSeconds(1);
|
||||
|
||||
private readonly IDashboardSnapshotService _snapshotService;
|
||||
private readonly IHubContext<DashboardSnapshotHub> _hubContext;
|
||||
@@ -38,6 +47,7 @@ public sealed class DashboardSnapshotPublisher : BackgroundService
|
||||
private readonly ILogger<DashboardSnapshotPublisher> _logger;
|
||||
private readonly TimeSpan _reconnectDelay;
|
||||
private readonly TimeSpan _idlePollInterval;
|
||||
private readonly TimeSpan _sendTimeout;
|
||||
|
||||
/// <summary>Initializes a new instance of the DashboardSnapshotPublisher class.</summary>
|
||||
/// <param name="snapshotService">The snapshot service to subscribe to.</param>
|
||||
@@ -64,13 +74,15 @@ public sealed class DashboardSnapshotPublisher : BackgroundService
|
||||
/// <param name="logger">The logger instance.</param>
|
||||
/// <param name="reconnectDelay">The delay before reconnecting after a subscription failure.</param>
|
||||
/// <param name="idlePollInterval">How often the idle publisher re-checks for a connected viewer.</param>
|
||||
/// <param name="sendTimeout">Upper bound on one broadcast; defaults to one second.</param>
|
||||
internal DashboardSnapshotPublisher(
|
||||
IDashboardSnapshotService snapshotService,
|
||||
IHubContext<DashboardSnapshotHub> hubContext,
|
||||
DashboardSnapshotHubConnectionCounter connectionCounter,
|
||||
ILogger<DashboardSnapshotPublisher> logger,
|
||||
TimeSpan reconnectDelay,
|
||||
TimeSpan idlePollInterval)
|
||||
TimeSpan idlePollInterval,
|
||||
TimeSpan? sendTimeout = null)
|
||||
{
|
||||
_snapshotService = snapshotService;
|
||||
_hubContext = hubContext;
|
||||
@@ -78,6 +90,7 @@ public sealed class DashboardSnapshotPublisher : BackgroundService
|
||||
_logger = logger;
|
||||
_reconnectDelay = reconnectDelay;
|
||||
_idlePollInterval = idlePollInterval;
|
||||
_sendTimeout = sendTimeout ?? DefaultSendTimeout;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
@@ -117,10 +130,19 @@ public sealed class DashboardSnapshotPublisher : BackgroundService
|
||||
|
||||
try
|
||||
{
|
||||
await _hubContext.Clients
|
||||
.All
|
||||
.SendAsync(DashboardSnapshotHub.SnapshotMessage, snapshot, stoppingToken)
|
||||
bool sent = await BoundedHubSend.SendAsync(
|
||||
_hubContext.Clients.All,
|
||||
DashboardSnapshotHub.SnapshotMessage,
|
||||
snapshot,
|
||||
_sendTimeout,
|
||||
stoppingToken)
|
||||
.ConfigureAwait(false);
|
||||
if (!sent)
|
||||
{
|
||||
_logger.LogDebug(
|
||||
"Snapshot broadcast exceeded {Timeout} and was abandoned; the next tick proceeds.",
|
||||
_sendTimeout);
|
||||
}
|
||||
}
|
||||
catch (Exception ex) when (ex is not OperationCanceledException)
|
||||
{
|
||||
|
||||
@@ -19,11 +19,27 @@ namespace ZB.MOM.WW.MxGateway.Server.Dashboard.Hubs;
|
||||
/// simply goes away never calls <c>UnsubscribeSession</c>; the hub's
|
||||
/// <c>OnDisconnectedAsync</c> releases everything the connection held.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// Hub viewers are also counted on their own, separately from the in-process
|
||||
/// subscribers that register under an <see cref="InProcessConnectionIdPrefix"/>
|
||||
/// id: <see cref="DashboardEventBroadcaster"/> runs one bounded hub-delivery pump
|
||||
/// per session only while a remote connection is watching it, and tears that pump
|
||||
/// down on <see cref="HubViewersReleased"/>.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
public sealed class EventsHubViewerRegistry
|
||||
{
|
||||
/// <summary>
|
||||
/// Prefix of the synthetic connection ids in-process subscribers register under.
|
||||
/// A SignalR connection id never carries it, so it is what tells the two
|
||||
/// audiences apart for the hub-only count.
|
||||
/// </summary>
|
||||
public const string InProcessConnectionIdPrefix = "inproc-";
|
||||
|
||||
private readonly ConcurrentDictionary<string, int> _viewersBySession = new(StringComparer.Ordinal);
|
||||
|
||||
private readonly ConcurrentDictionary<string, int> _hubViewersBySession = new(StringComparer.Ordinal);
|
||||
|
||||
private readonly ConcurrentDictionary<string, ConcurrentDictionary<string, byte>> _sessionsByConnection =
|
||||
new(StringComparer.Ordinal);
|
||||
|
||||
@@ -53,8 +69,19 @@ public sealed class EventsHubViewerRegistry
|
||||
}
|
||||
|
||||
_viewersBySession.AddOrUpdate(sessionId, 1, static (_, count) => count + 1);
|
||||
if (IsHubConnection(connectionId))
|
||||
{
|
||||
_hubViewersBySession.AddOrUpdate(sessionId, 1, static (_, count) => count + 1);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Raised with a session id when its last remote hub viewer leaves (unsubscribe
|
||||
/// or disconnect). In-process subscribers do not raise it. Handlers run on the
|
||||
/// releasing thread and must not throw or block.
|
||||
/// </summary>
|
||||
public event Action<string>? HubViewersReleased;
|
||||
|
||||
/// <summary>
|
||||
/// Records that <paramref name="connectionId"/> stopped watching
|
||||
/// <paramref name="sessionId"/>. A removal with no matching
|
||||
@@ -75,7 +102,7 @@ public sealed class EventsHubViewerRegistry
|
||||
return;
|
||||
}
|
||||
|
||||
ReleaseSession(sessionId);
|
||||
Release(connectionId, sessionId);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
@@ -107,7 +134,7 @@ public sealed class EventsHubViewerRegistry
|
||||
// same detached set must not let the session be decremented twice.
|
||||
if (sessions.TryRemove(sessionId, out _))
|
||||
{
|
||||
ReleaseSession(sessionId);
|
||||
Release(connectionId, sessionId);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -120,32 +147,63 @@ public sealed class EventsHubViewerRegistry
|
||||
&& _viewersBySession.TryGetValue(sessionId, out int count)
|
||||
&& count > 0;
|
||||
|
||||
/// <summary>
|
||||
/// Gets a value indicating whether any remote <see cref="EventsHub"/> connection —
|
||||
/// as opposed to an in-process subscriber — is watching the session.
|
||||
/// </summary>
|
||||
/// <param name="sessionId">Session id to test.</param>
|
||||
/// <returns><see langword="true"/> when at least one hub connection is subscribed.</returns>
|
||||
public bool HasHubViewers(string sessionId) =>
|
||||
!string.IsNullOrEmpty(sessionId)
|
||||
&& _hubViewersBySession.TryGetValue(sessionId, out int count)
|
||||
&& count > 0;
|
||||
|
||||
private static bool IsHubConnection(string connectionId) =>
|
||||
!connectionId.StartsWith(InProcessConnectionIdPrefix, StringComparison.Ordinal);
|
||||
|
||||
/// <summary>
|
||||
/// Releases one subscription from both counts it was recorded in, and raises
|
||||
/// <see cref="HubViewersReleased"/> when it was the session's last hub viewer.
|
||||
/// </summary>
|
||||
/// <param name="connectionId">Connection that held the subscription.</param>
|
||||
/// <param name="sessionId">Session id whose counts are released.</param>
|
||||
private void Release(string connectionId, string sessionId)
|
||||
{
|
||||
ReleaseSession(_viewersBySession, sessionId);
|
||||
if (IsHubConnection(connectionId) && ReleaseSession(_hubViewersBySession, sessionId))
|
||||
{
|
||||
HubViewersReleased?.Invoke(sessionId);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Decrements the session's viewer count, dropping the entry entirely at
|
||||
/// zero so the dictionary does not grow one key per session ever viewed.
|
||||
/// The compare-and-swap loop keeps the decrement correct against a
|
||||
/// concurrent <see cref="AddViewer"/> on the same session.
|
||||
/// </summary>
|
||||
/// <param name="counts">The count map to decrement.</param>
|
||||
/// <param name="sessionId">Session id whose count is released.</param>
|
||||
private void ReleaseSession(string sessionId)
|
||||
/// <returns><see langword="true"/> when this release dropped the count to zero.</returns>
|
||||
private static bool ReleaseSession(ConcurrentDictionary<string, int> counts, string sessionId)
|
||||
{
|
||||
while (true)
|
||||
{
|
||||
if (!_viewersBySession.TryGetValue(sessionId, out int count))
|
||||
if (!counts.TryGetValue(sessionId, out int count))
|
||||
{
|
||||
return;
|
||||
return false;
|
||||
}
|
||||
|
||||
if (count <= 1)
|
||||
{
|
||||
if (_viewersBySession.TryRemove(new KeyValuePair<string, int>(sessionId, count)))
|
||||
if (counts.TryRemove(new KeyValuePair<string, int>(sessionId, count)))
|
||||
{
|
||||
return;
|
||||
return true;
|
||||
}
|
||||
}
|
||||
else if (_viewersBySession.TryUpdate(sessionId, count - 1, count))
|
||||
else if (counts.TryUpdate(sessionId, count - 1, count))
|
||||
{
|
||||
return;
|
||||
return false;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -20,6 +20,24 @@ public interface IDashboardLiveDataService
|
||||
IReadOnlyCollection<string> tagAddresses,
|
||||
CancellationToken cancellationToken);
|
||||
|
||||
/// <summary>
|
||||
/// Reads the supplied tags on behalf of one viewer (a Browse page circuit).
|
||||
/// The shared advise set is budgeted per active viewer, so the viewer id is
|
||||
/// what stops one circuit's reads from evicting another circuit's whole set.
|
||||
/// Never throws, as <see cref="ReadAsync(IReadOnlyCollection{string}, CancellationToken)"/>.
|
||||
/// </summary>
|
||||
/// <param name="tagAddresses">Fully-qualified tag references to read.</param>
|
||||
/// <param name="viewerId">
|
||||
/// Stable id of the reading viewer for the life of its page; <see langword="null"/>
|
||||
/// reads under one shared anonymous viewer.
|
||||
/// </param>
|
||||
/// <param name="cancellationToken">Token to cancel the read.</param>
|
||||
/// <returns>The read result, or an error-bearing result on failure.</returns>
|
||||
Task<DashboardLiveReadResult> ReadAsync(
|
||||
IReadOnlyCollection<string> tagAddresses,
|
||||
string? viewerId,
|
||||
CancellationToken cancellationToken) => ReadAsync(tagAddresses, cancellationToken);
|
||||
|
||||
/// <summary>
|
||||
/// Queries the currently-active alarm set for the dashboard session.
|
||||
/// Never throws — failures are surfaced in
|
||||
|
||||
Reference in New Issue
Block a user