fix(dashboard): debounced off-dispatcher Browse search, bounded disposal, circuit-aware polling
- Search: the oninput setter scanned the whole Galaxy synchronously per keystroke. It now debounces 150ms, runs on the thread pool with cancellation (newer keystroke supersedes), and reads a shared singleton DashboardBrowseSearchIndex of pre-lowercased references keyed on the Galaxy cache Sequence. - Disposal: DisposeAsync awaited both loops unbounded and the loops caught only OCE. Loops now handle faults per iteration (poll faults surface in _readError, deploy-notifier faults in a retrying warning), mutate the tree on the dispatcher, drain concurrently with a 5s bound, and the CTSs are disposed in finally. - Circuits: DisconnectedCircuitRetentionPeriod is 30s (was the 3 min default) and a scoped DashboardCircuitConnectionTracker (CircuitHandler) lets the 2s ReadBulk poll skip ticks while the browser is disconnected. - Reads pass a per-page viewer id to the live-data per-viewer share.
This commit is contained in:
Joseph Doherty
@@ -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.
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
|
||||
@@ -0,0 +1,95 @@
|
||||
using ZB.MOM.WW.GalaxyRepository;
|
||||
using ZB.MOM.WW.GalaxyRepository.Grpc;
|
||||
using ZB.MOM.WW.MxGateway.Server.Dashboard;
|
||||
|
||||
namespace ZB.MOM.WW.MxGateway.Tests.Dashboard;
|
||||
|
||||
/// <summary>
|
||||
/// Verifies the Browse filter's shared index: case-insensitive matching on reference
|
||||
/// and attribute name, the result limit, one build per Galaxy cache sequence, and
|
||||
/// cancellation of a superseded scan.
|
||||
/// </summary>
|
||||
public sealed class DashboardBrowseSearchIndexTests
|
||||
{
|
||||
/// <summary>Matches either the full reference or the attribute name, ignoring case.</summary>
|
||||
[Fact]
|
||||
public void Search_MatchesReferenceOrAttributeNameCaseInsensitively()
|
||||
{
|
||||
DashboardBrowseSearchIndex index = new();
|
||||
GalaxyHierarchyCacheEntry entry = CreateEntry(sequence: 1);
|
||||
|
||||
IReadOnlyList<GalaxyAttribute> byReference = index.Search(entry, " PUMP_001.p ", 10, CancellationToken.None);
|
||||
IReadOnlyList<GalaxyAttribute> byName = index.Search(entry, "levels", 10, CancellationToken.None);
|
||||
|
||||
Assert.Equal(["Pump_001.PV"], byReference.Select(attr => attr.FullTagReference));
|
||||
Assert.Equal(["Tank_001.Levels[]"], byName.Select(attr => attr.FullTagReference));
|
||||
Assert.Empty(index.Search(entry, " ", 10, CancellationToken.None));
|
||||
}
|
||||
|
||||
/// <summary>Stops at the limit, in Galaxy object order.</summary>
|
||||
[Fact]
|
||||
public void Search_StopsAtTheLimit()
|
||||
{
|
||||
DashboardBrowseSearchIndex index = new();
|
||||
|
||||
IReadOnlyList<GalaxyAttribute> matches = index.Search(CreateEntry(sequence: 1), "_001", 2, CancellationToken.None);
|
||||
|
||||
Assert.Equal(["Pump_001.PV", "Pump_001.Alarm"], matches.Select(attr => attr.FullTagReference));
|
||||
}
|
||||
|
||||
/// <summary>The index is built once per cache sequence and rebuilt when the sequence moves.</summary>
|
||||
[Fact]
|
||||
public void Search_RebuildsOnlyWhenTheCacheSequenceChanges()
|
||||
{
|
||||
DashboardBrowseSearchIndex index = new();
|
||||
GalaxyHierarchyCacheEntry first = CreateEntry(sequence: 1);
|
||||
|
||||
index.Search(first, "pump", 10, CancellationToken.None);
|
||||
index.Search(first, "tank", 10, CancellationToken.None);
|
||||
Assert.Equal(1, index.BuildCount);
|
||||
|
||||
GalaxyHierarchyCacheEntry redeployed = CreateEntry(sequence: 2, extraTag: "Valve_001.Open");
|
||||
IReadOnlyList<GalaxyAttribute> matches = index.Search(redeployed, "valve", 10, CancellationToken.None);
|
||||
|
||||
Assert.Equal(2, index.BuildCount);
|
||||
Assert.Equal(["Valve_001.Open"], matches.Select(attr => attr.FullTagReference));
|
||||
}
|
||||
|
||||
/// <summary>A cancelled (superseded) search throws instead of finishing the scan.</summary>
|
||||
[Fact]
|
||||
public void Search_WhenCancelled_Throws()
|
||||
{
|
||||
DashboardBrowseSearchIndex index = new();
|
||||
using CancellationTokenSource cancelled = new();
|
||||
cancelled.Cancel();
|
||||
|
||||
Assert.ThrowsAny<OperationCanceledException>(
|
||||
() => index.Search(CreateEntry(sequence: 1), "pump", 10, cancelled.Token));
|
||||
}
|
||||
|
||||
private static GalaxyHierarchyCacheEntry CreateEntry(long sequence, string? extraTag = null)
|
||||
{
|
||||
GalaxyObject pump = new() { GobjectId = 1, TagName = "Pump_001" };
|
||||
pump.Attributes.Add(new GalaxyAttribute { AttributeName = "PV", FullTagReference = "Pump_001.PV" });
|
||||
pump.Attributes.Add(new GalaxyAttribute { AttributeName = "Alarm", FullTagReference = "Pump_001.Alarm" });
|
||||
|
||||
GalaxyObject tank = new() { GobjectId = 2, TagName = "Tank_001" };
|
||||
tank.Attributes.Add(new GalaxyAttribute { AttributeName = "Levels", FullTagReference = "Tank_001.Levels[]" });
|
||||
|
||||
List<GalaxyObject> objects = [pump, tank];
|
||||
if (extraTag is not null)
|
||||
{
|
||||
GalaxyObject extra = new() { GobjectId = 3, TagName = extraTag.Split('.')[0] };
|
||||
extra.Attributes.Add(new GalaxyAttribute { AttributeName = extraTag.Split('.')[1], FullTagReference = extraTag });
|
||||
objects.Add(extra);
|
||||
}
|
||||
|
||||
return GalaxyHierarchyCacheEntry.Empty with
|
||||
{
|
||||
Status = GalaxyCacheStatus.Healthy,
|
||||
Sequence = sequence,
|
||||
Objects = objects,
|
||||
Index = GalaxyHierarchyIndex.Build(objects),
|
||||
};
|
||||
}
|
||||
}
|
||||
@@ -1,6 +1,9 @@
|
||||
using Microsoft.AspNetCore.Builder;
|
||||
using Microsoft.AspNetCore.Components.Server;
|
||||
using Microsoft.AspNetCore.Components.Server.Circuits;
|
||||
using Microsoft.AspNetCore.Routing;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.Options;
|
||||
using ZB.MOM.WW.MxGateway.Server;
|
||||
using ZB.MOM.WW.MxGateway.Server.Dashboard;
|
||||
using ZB.MOM.WW.MxGateway.Server.Dashboard.Hubs;
|
||||
@@ -61,4 +64,42 @@ public sealed class DashboardHubsRegistrationTests
|
||||
Assert.Same(broadcaster, subscriber);
|
||||
Assert.Same(app.Services.GetRequiredService<DashboardEventBroadcaster>(), broadcaster);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A closed tab's circuit (and its Browse poll) is retained for 30 seconds, not the
|
||||
/// framework's three minutes, and each circuit gets a connection tracker that is also
|
||||
/// its <see cref="CircuitHandler"/>, so the Browse poll can pause while disconnected.
|
||||
/// </summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[Fact]
|
||||
public async Task Build_WhenDashboardEnabled_ShortensCircuitRetentionAndTracksConnection()
|
||||
{
|
||||
await using WebApplication app = GatewayApplication.Build([]);
|
||||
|
||||
CircuitOptions circuitOptions = app.Services.GetRequiredService<IOptions<CircuitOptions>>().Value;
|
||||
Assert.Equal(TimeSpan.FromSeconds(30), circuitOptions.DisconnectedCircuitRetentionPeriod);
|
||||
Assert.Equal(DashboardServiceCollectionExtensions.DisconnectedCircuitRetention, circuitOptions.DisconnectedCircuitRetentionPeriod);
|
||||
|
||||
using IServiceScope scope = app.Services.CreateScope();
|
||||
DashboardCircuitConnectionTracker tracker = scope.ServiceProvider.GetRequiredService<DashboardCircuitConnectionTracker>();
|
||||
Assert.Contains(scope.ServiceProvider.GetServices<CircuitHandler>(), handler => ReferenceEquals(handler, tracker));
|
||||
|
||||
Assert.True(tracker.IsConnected);
|
||||
await tracker.OnConnectionDownAsync(null!, CancellationToken.None);
|
||||
Assert.False(tracker.IsConnected);
|
||||
await tracker.OnConnectionUpAsync(null!, CancellationToken.None);
|
||||
Assert.True(tracker.IsConnected);
|
||||
}
|
||||
|
||||
/// <summary>The Browse filter index is one shared instance, built once per Galaxy deploy.</summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[Fact]
|
||||
public async Task Build_WhenDashboardEnabled_SharesOneBrowseSearchIndex()
|
||||
{
|
||||
await using WebApplication app = GatewayApplication.Build([]);
|
||||
|
||||
Assert.Same(
|
||||
app.Services.GetRequiredService<DashboardBrowseSearchIndex>(),
|
||||
app.Services.GetRequiredService<DashboardBrowseSearchIndex>());
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user