dohertj2/ScadaBridge
1
0
Fork 0
Code Issues 17 Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(deployment): move PendingUpdatesCache to DeploymentManager so the CLI shares it (ScadaBridge#91)

Browse Source 
`health pending-updates` (#63) called IPendingUpdatesEvaluator.EvaluateAsync directly
from ManagementActor because the cache lived in CentralUI, so every CLI call was a full
uncached fleet flatten. IPendingUpdatesCache/PendingUpdatesCache now live in
DeploymentManager.PendingUpdates and are registered by AddDeploymentManager; the
SingleFlightMemo<T> primitive moves to Commons.Caching (public) so both CentralUI's KPI /
alarm-summary caches and DeploymentManager can use it. HandleGetPendingUpdates reads the
shared cache; the UI circuits and the CLI now share one memo per node.

Tests: ManagementActor repeated-calls test (fails against the direct evaluator call:
2 evaluations instead of 1), AddDeploymentManager singleton-registration test; the cache
and memo tests move with their types.

(ScadaBridge#91)

Claude-Session: https://claude.ai/code/session_01GT67AEHd2Acv1YPwGfXEoV
This commit is contained in:
Joseph Doherty
2026-11-09 13:30:38 -05:00
parent c7d868c52e
commit 1ee2e171cf
16 changed files with 102 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ZB.MOM.WW.ScadaBridge.CentralUI/ServiceCollectionExtensions.cs
+2 -3
View File
@@ -120,9 +120,8 @@ public static class ServiceCollectionExtensions
// circuit for N dashboards to share a single SQL round per refresh window.
services.AddSingleton<IKpiSnapshotCache, KpiSnapshotCache>();
// Same single-flight TTL shape as IKpiSnapshotCache, over the fleet-wide
// pending-updates flatten (IPendingUpdatesEvaluator) instead of the KPI Asks.
services.AddSingleton<IPendingUpdatesCache, PendingUpdatesCache>();
// IPendingUpdatesCache (the fleet-wide pending-updates memo) is registered by
// AddDeploymentManager, not here, so the management API shares it (Gitea #91).
// Secured Writes: dispatches the two-person secured-write commands
// (submit / approve / reject / list) to the central ManagementActor through the
src/ZB.MOM.WW.ScadaBridge.CentralUI/Services/IPendingUpdatesCache.cs
-45
View File
@@ -1,45 +0,0 @@
using ZB.MOM.WW.ScadaBridge.DeploymentManager.PendingUpdates;
namespace ZB.MOM.WW.ScadaBridge.CentralUI.Services;
/// <summary>
/// Process-level memoized fleet-wide pending-updates snapshot shared by every Blazor circuit
/// on this node — the same single-flight TTL shape as <see cref="IKpiSnapshotCache"/>, applied
/// to <see cref="IPendingUpdatesEvaluator"/>.
/// <para>
/// <see cref="IPendingUpdatesEvaluator.EvaluateAsync"/> flattens every site and every deployed
/// instance, so it is a whole-fleet read, not a per-circuit one. Without memoization, N
/// Deployer/Administrator circuits watching the rail indicator or the Sites page would each
/// re-flatten the fleet on their own poll. This cache collapses them: the freshness window is
/// <c>DeploymentManagerOptions.PendingUpdatesCacheTtl</c> (default 60s), and however many
/// circuits ask inside one window, the fleet is evaluated at most once.
/// </para>
/// <para>
/// A failed evaluation is never memoized: the fault is surfaced to every caller of that flight,
/// and the next caller starts a fresh one rather than re-serving the fault for the rest of the
/// TTL window.
/// </para>
/// </summary>
public interface IPendingUpdatesCache
{
/// <summary>
/// Returns the memoized fleet-wide pending-updates snapshot, evaluating a fresh one if none
/// is still within the TTL window.
/// </summary>
/// <param name="forceRefresh">
/// When <see langword="true"/>, discards a still-fresh value and starts a new evaluation
/// (the operator's "Re-check" action). An evaluation already in flight is still joined
/// rather than duplicated, so a burst of Re-check clicks remains one fleet flatten.
/// </param>
/// <param name="cancellationToken">Cancels this caller's wait only, never the shared flight.</param>
/// <returns>The memoized (or freshly evaluated) fleet-wide snapshot.</returns>
Task<PendingUpdatesSnapshot> GetAsync(bool forceRefresh = false, CancellationToken cancellationToken = default);
/// <summary>
/// Raised (on a thread-pool thread) once deployments have finished on this node and the
/// memoized snapshot was dropped — after a short quiet period, so a bulk deploy's burst of
/// completions raises it once. Subscribers re-read via <see cref="GetAsync"/> instead of
/// waiting out their poll.
/// </summary>
event Action? Invalidated;
}
src/ZB.MOM.WW.ScadaBridge.CentralUI/Services/KpiSnapshotCache.cs
+1
View File
@@ -1,4 +1,5 @@
using Microsoft.Extensions.DependencyInjection;
using ZB.MOM.WW.ScadaBridge.Commons.Caching;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Audit;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Notification;
using ZB.MOM.WW.ScadaBridge.Commons.Types;
src/ZB.MOM.WW.ScadaBridge.CentralUI/Services/PendingUpdatesCache.cs
-127
View File
@@ -1,127 +0,0 @@
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Options;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
using ZB.MOM.WW.ScadaBridge.DeploymentManager;
using ZB.MOM.WW.ScadaBridge.DeploymentManager.PendingUpdates;
namespace ZB.MOM.WW.ScadaBridge.CentralUI.Services;
/// <summary>
/// Default <see cref="IPendingUpdatesCache"/> — a DI <b>singleton</b> holding one
/// <see cref="SingleFlightMemo{T}"/> over <see cref="IPendingUpdatesEvaluator"/>.
/// <para>
/// Registered as a singleton on purpose: the memo has to outlive any one circuit for N
/// dashboards to share a single fleet flatten per refresh window. It depends only on
/// <see cref="IServiceScopeFactory"/> — the evaluator is scoped (it opens repositories), so
/// every flight resolves it from a fresh scope, off any circuit-scoped context.
/// </para>
/// <para>
/// A finished deployment (<see cref="DeploymentStatus.Success"/> or
/// <see cref="DeploymentStatus.Failed"/>, from <see cref="IDeploymentStatusNotifier"/>) drops the
/// memoized snapshot at once and, after <see cref="DefaultInvalidationQuietPeriod"/> without
/// another completion, raises <see cref="Invalidated"/>. Before this, a Deploy all left the rail
/// showing the pre-deploy count for up to the TTL plus one poll. Only deployments run on THIS
/// node are seen; one run through the peer central still clears within the TTL.
/// </para>
/// </summary>
public sealed class PendingUpdatesCache : IPendingUpdatesCache, IDisposable
{
private readonly IServiceScopeFactory _scopeFactory;
private readonly SingleFlightMemo<PendingUpdatesSnapshot> _memo;
/// <summary>Negative TTL after a faulted fleet evaluation, so polling does not hammer a sick DB.</summary>
internal static readonly TimeSpan FaultBackoff = TimeSpan.FromSeconds(10);
/// <summary>How long completions must stop arriving before <see cref="Invalidated"/> is raised.</summary>
internal static readonly TimeSpan DefaultInvalidationQuietPeriod = TimeSpan.FromSeconds(1);
private readonly IDeploymentStatusNotifier _statusNotifier;
private readonly TimeSpan _invalidationQuietPeriod;
private readonly Timer _invalidatedTimer;
private volatile bool _disposed;
/// <inheritdoc/>
public event Action? Invalidated;
/// <summary>
/// Initializes the shared pending-updates cache.
/// </summary>
/// <param name="scopeFactory">Opens a fresh DI scope per fleet evaluation.</param>
/// <param name="options">Supplies <see cref="DeploymentManagerOptions.PendingUpdatesCacheTtl"/>.</param>
/// <param name="statusNotifier">Reports finished deployments, which invalidate the snapshot.</param>
public PendingUpdatesCache(
IServiceScopeFactory scopeFactory,
IOptions<DeploymentManagerOptions> options,
IDeploymentStatusNotifier statusNotifier)
: this(scopeFactory, options, statusNotifier, DefaultInvalidationQuietPeriod)
{
}
/// <summary>Initializes the cache with an explicit quiet period (tests).</summary>
/// <param name="scopeFactory">Opens a fresh DI scope per fleet evaluation.</param>
/// <param name="options">Supplies <see cref="DeploymentManagerOptions.PendingUpdatesCacheTtl"/>.</param>
/// <param name="statusNotifier">Reports finished deployments, which invalidate the snapshot.</param>
/// <param name="invalidationQuietPeriod">How long completions must stop before <see cref="Invalidated"/> is raised.</param>
internal PendingUpdatesCache(
IServiceScopeFactory scopeFactory,
IOptions<DeploymentManagerOptions> options,
IDeploymentStatusNotifier statusNotifier,
TimeSpan invalidationQuietPeriod)
{
_scopeFactory = scopeFactory ?? throw new ArgumentNullException(nameof(scopeFactory));
ArgumentNullException.ThrowIfNull(options);
_statusNotifier = statusNotifier ?? throw new ArgumentNullException(nameof(statusNotifier));
_invalidationQuietPeriod = invalidationQuietPeriod;
_memo = new SingleFlightMemo<PendingUpdatesSnapshot>(
options.Value.PendingUpdatesCacheTtl, faultBackoff: FaultBackoff);
_invalidatedTimer = new Timer(_ => RaiseInvalidated(), null, Timeout.Infinite, Timeout.Infinite);
_statusNotifier.StatusChanged += OnDeploymentStatusChanged;
}
/// <inheritdoc/>
public Task<PendingUpdatesSnapshot> GetAsync(
bool forceRefresh = false, CancellationToken cancellationToken = default) =>
_memo.GetAsync(EvaluateAsync, forceRefresh, cancellationToken);
/// <summary>Stops listening for deployment status changes.</summary>
public void Dispose()
{
_disposed = true;
_statusNotifier.StatusChanged -= OnDeploymentStatusChanged;
_invalidatedTimer.Dispose();
}
private void OnDeploymentStatusChanged(DeploymentStatusChange change)
{
if (_disposed || change.Status is not (DeploymentStatus.Success or DeploymentStatus.Failed))
{
return;
}
// Drop the snapshot at once so no caller is served the pre-deploy count, and (re)arm the
// quiet-period timer so a bulk deploy's burst of completions notifies subscribers once.
_memo.Invalidate();
try
{
_invalidatedTimer.Change(_invalidationQuietPeriod, Timeout.InfiniteTimeSpan);
}
catch (ObjectDisposedException)
{
// Disposed between the check and the re-arm.
}
}
private void RaiseInvalidated()
{
if (!_disposed)
{
Invalidated?.Invoke();
}
}
private async Task<PendingUpdatesSnapshot> EvaluateAsync()
{
await using var scope = _scopeFactory.CreateAsyncScope();
return await scope.ServiceProvider.GetRequiredService<IPendingUpdatesEvaluator>().EvaluateAsync();
}
}
src/ZB.MOM.WW.ScadaBridge.CentralUI/Services/SharedAlarmSummaryService.cs
+1
View File
@@ -1,5 +1,6 @@
using System.Collections.Concurrent;
using Microsoft.Extensions.DependencyInjection;
using ZB.MOM.WW.ScadaBridge.Commons.Caching;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Streaming;
using ZB.MOM.WW.ScadaBridge.Communication;
src/ZB.MOM.WW.ScadaBridge.CentralUI/Services/SingleFlightMemo.cs
-180
View File
@@ -1,180 +0,0 @@
namespace ZB.MOM.WW.ScadaBridge.CentralUI.Services;
/// <summary>
/// A process-level memo slot: one value, a time-to-live, and single-flight production.
/// <para>
/// Blazor Server runs one circuit per connected browser, so a page that polls on a timer
/// runs its query <em>per circuit</em>. Ten operators on the Health dashboard meant ten
/// identical KPI round trips every ten seconds. This slot collapses them: the first caller
/// in a TTL window starts the flight, every concurrent caller awaits the <em>same</em>
/// <see cref="Task{TResult}"/>, and callers arriving after it completes are served the
/// memoized value until the TTL expires.
/// </para>
/// <para>
/// By default a faulted flight is discarded so the next caller retries. A slot built with a
/// <c>faultBackoff</c> instead remembers the fault for that short negative TTL so a sick
/// dependency is not hammered by every poll; a forced refresh bypasses the back-off.
/// The caller's own exception handling is unchanged: it still sees the fault.
/// </para>
/// </summary>
/// <typeparam name="T">The memoized value type.</typeparam>
internal sealed class SingleFlightMemo<T>
{
private readonly object _lock = new();
private readonly TimeSpan _ttl;
private readonly Func<DateTimeOffset> _clock;
/// <summary>The current flight, completed or not. Null until the first call.</summary>
private Task<T>? _current;
/// <summary>Wall-clock instant after which <see cref="_current"/> is stale. Only meaningful when it completed successfully.</summary>
private DateTimeOffset _freshUntil;
/// <summary>
/// How long a faulted flight is remembered (zero = never). While inside the window, callers
/// are served the same fault instead of starting another flight against a sick dependency.
/// </summary>
private readonly TimeSpan _faultBackoff;
/// <summary>Wall-clock instant until which a faulted <see cref="_current"/> is re-served.</summary>
private DateTimeOffset _faultedUntil;
/// <summary>
/// Bumped by <see cref="Invalidate"/>. A flight records the value it started under and is
/// memoized only if no invalidation happened while it ran.
/// </summary>
private long _generation;
/// <summary>
/// Initializes a memo slot.
/// </summary>
/// <param name="ttl">How long a successfully produced value stays fresh.</param>
/// <param name="clock">Clock used for freshness, injectable so tests need no real delay.</param>
/// <param name="faultBackoff">Negative TTL after a faulted flight; null/zero disables it.</param>
public SingleFlightMemo(TimeSpan ttl, Func<DateTimeOffset>? clock = null, TimeSpan? faultBackoff = null)
{
_ttl = ttl;
_faultBackoff = faultBackoff ?? TimeSpan.Zero;
_clock = clock ?? (() => DateTimeOffset.UtcNow);
}
/// <summary>
/// Number of times the factory has actually been invoked. Diagnostics and tests only.
/// </summary>
public int FlightCount { get; private set; }
/// <summary>
/// Marks the memoized value stale, so the next <see cref="GetAsync"/> starts a new flight.
/// A flight already in progress still answers the callers waiting on it, but its result is
/// not memoized: it may have read the state this invalidation reports as changed.
/// </summary>
public void Invalidate()
{
lock (_lock)
{
_generation++;
_freshUntil = default;
_faultedUntil = default;
}
}
/// <summary>
/// Returns the memoized value, starting a flight if none is fresh.
/// </summary>
/// <param name="factory">Produces a fresh value. Invoked outside the slot's lock.</param>
/// <param name="forceRefresh">
/// When <see langword="true"/>, a completed-but-still-fresh value is discarded and a new
/// flight starts. An <em>in-flight</em> round is still joined rather than duplicated, so a
/// burst of operator "Refresh" clicks remains one round trip.
/// </param>
/// <param name="cancellationToken">
/// Cancels this caller's wait only — never the shared flight, which other callers may still
/// be awaiting.
/// </param>
/// <param name="ttlOverride">
/// Freshness window for <em>this</em> flight, overriding the slot's default. Used where the
/// acceptable staleness depends on state the caller knows (the Alarm Summary poll needs a
/// far shorter window while the live alarm cache is cold than while it is serving deltas).
/// </param>
/// <returns>The memoized (or freshly produced) value.</returns>
public Task<T> GetAsync(
Func<Task<T>> factory,
bool forceRefresh = false,
CancellationToken cancellationToken = default,
TimeSpan? ttlOverride = null)
{
ArgumentNullException.ThrowIfNull(factory);
TaskCompletionSource<T>? started = null;
Task<T> result;
long generation = 0;
lock (_lock)
{
var current = _current;
if (current is { IsCompleted: false })
{
// A round is already in progress — join it, force or not.
result = current;
}
else if (!forceRefresh
&& current is { IsFaulted: true }
&& _clock() < _faultedUntil)
{
result = current;
}
else if (!forceRefresh
&& current is { IsCompletedSuccessfully: true }
&& _clock() < _freshUntil)
{
result = current;
}
else
{
// RunContinuationsAsynchronously: a caller's continuation must never run
// inline on the thread that completes the flight (that thread is another
// circuit's dispatcher).
started = new TaskCompletionSource<T>(TaskCreationOptions.RunContinuationsAsynchronously);
_current = started.Task;
FlightCount++;
generation = _generation;
result = started.Task;
}
}
// The factory runs OUTSIDE the lock: it does I/O (an Ask, a SQL round trip) and must
// never be able to re-enter this slot on the calling thread while the lock is held.
if (started is not null)
{
_ = RunAsync(factory, started, ttlOverride ?? _ttl, generation);
}
return cancellationToken.CanBeCanceled ? result.WaitAsync(cancellationToken) : result;
}
private async Task RunAsync(Func<Task<T>> factory, TaskCompletionSource<T> completion, TimeSpan ttl, long generation)
{
try
{
var value = await factory().ConfigureAwait(false);
lock (_lock)
{
// Invalidated while running: answer this flight's callers, but do not serve it
// to anyone else.
_freshUntil = generation == _generation ? _clock() + ttl : default;
}
completion.SetResult(value);
}
catch (Exception ex)
{
// A failure is memoized only for the optional short fault back-off so a sick dependency
// is not retried by every poll; _freshUntil stays in the past.
lock (_lock)
{
_freshUntil = default;
_faultedUntil = _faultBackoff > TimeSpan.Zero ? _clock() + _faultBackoff : default;
}
completion.SetException(ex);
}
}
}