feat(central-ui): process-wide single-flight cache for the pending-updates snapshot

Claude-Session: https://claude.ai/code/session_01Y5ef34iUUMjmwnMpL9xgmB
This commit is contained in:
Joseph Doherty
2026-09-14 13:52:54 -04:00
parent 0287f8f64f
commit ada91bb82c
4 changed files with 216 additions and 0 deletions
@@ -120,6 +120,10 @@ public static class ServiceCollectionExtensions
// circuit for N dashboards to share a single SQL round per refresh window. // circuit for N dashboards to share a single SQL round per refresh window.
services.AddSingleton<IKpiSnapshotCache, KpiSnapshotCache>(); 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>();
// Secured Writes: dispatches the two-person secured-write commands // Secured Writes: dispatches the two-person secured-write commands
// (submit / approve / reject / list) to the central ManagementActor through the // (submit / approve / reject / list) to the central ManagementActor through the
// in-process ManagementActorHolder seam — the same Ask path the HTTP /management // in-process ManagementActorHolder seam — the same Ask path the HTTP /management
@@ -0,0 +1,37 @@
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);
}
@@ -0,0 +1,46 @@
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Options;
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>
/// </summary>
public sealed class PendingUpdatesCache : IPendingUpdatesCache
{
private readonly IServiceScopeFactory _scopeFactory;
private readonly SingleFlightMemo<PendingUpdatesSnapshot> _memo;
/// <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>
public PendingUpdatesCache(IServiceScopeFactory scopeFactory, IOptions<DeploymentManagerOptions> options)
{
_scopeFactory = scopeFactory ?? throw new ArgumentNullException(nameof(scopeFactory));
ArgumentNullException.ThrowIfNull(options);
_memo = new SingleFlightMemo<PendingUpdatesSnapshot>(options.Value.PendingUpdatesCacheTtl);
}
/// <inheritdoc/>
public Task<PendingUpdatesSnapshot> GetAsync(
bool forceRefresh = false, CancellationToken cancellationToken = default) =>
_memo.GetAsync(EvaluateAsync, forceRefresh, cancellationToken);
private async Task<PendingUpdatesSnapshot> EvaluateAsync()
{
await using var scope = _scopeFactory.CreateAsyncScope();
return await scope.ServiceProvider.GetRequiredService<IPendingUpdatesEvaluator>().EvaluateAsync();
}
}
@@ -0,0 +1,129 @@
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Options;
using NSubstitute;
using ZB.MOM.WW.ScadaBridge.CentralUI.Services;
using ZB.MOM.WW.ScadaBridge.DeploymentManager;
using ZB.MOM.WW.ScadaBridge.DeploymentManager.PendingUpdates;
namespace ZB.MOM.WW.ScadaBridge.CentralUI.Tests.Services;
/// <summary>
/// Unit tests for <see cref="PendingUpdatesCache"/> — the same single-flight TTL contract as
/// <see cref="SingleFlightMemoTests"/> proves for the primitive, exercised here through the
/// full DI-scope-per-flight shape: N concurrent callers before the evaluator resolves must
/// share ONE fleet flatten, a value produced inside the TTL is served without re-evaluating,
/// <c>forceRefresh</c> bypasses a fresh value but still joins an in-flight round, and a
/// faulted evaluation is never memoized.
/// </summary>
public class PendingUpdatesCacheTests : IDisposable
{
private static readonly DateTimeOffset T0 = new(2026, 9, 14, 9, 0, 0, TimeSpan.Zero);
private readonly IPendingUpdatesEvaluator _evaluator = Substitute.For<IPendingUpdatesEvaluator>();
private readonly ServiceProvider _provider;
public PendingUpdatesCacheTests()
{
var services = new ServiceCollection();
services.AddScoped(_ => _evaluator);
_provider = services.BuildServiceProvider();
}
// A generous TTL (1h) so every test's timing is driven by task completion, not the clock.
private static IOptions<DeploymentManagerOptions> Options(TimeSpan? ttl = null) =>
Microsoft.Extensions.Options.Options.Create(
new DeploymentManagerOptions { PendingUpdatesCacheTtl = ttl ?? TimeSpan.FromHours(1) });
private PendingUpdatesCache CreateSut(TimeSpan? ttl = null) =>
new(_provider.GetRequiredService<IServiceScopeFactory>(), Options(ttl));
private static PendingUpdatesSnapshot Snapshot() =>
new(T0, Array.Empty<SitePendingUpdates>());
[Fact]
public async Task ConcurrentCallers_ShareOneEvaluation()
{
var released = new TaskCompletionSource<PendingUpdatesSnapshot>(
TaskCreationOptions.RunContinuationsAsynchronously);
_evaluator.EvaluateAsync(Arg.Any<CancellationToken>()).Returns(released.Task);
var sut = CreateSut();
var waiters = Enumerable.Range(0, 10).Select(_ => sut.GetAsync()).ToArray();
released.SetResult(Snapshot());
var results = await Task.WhenAll(waiters);
await _evaluator.Received(1).EvaluateAsync(Arg.Any<CancellationToken>());
Assert.All(results, r => Assert.Same(results[0], r));
}
[Fact]
public async Task WithinTtl_ServesTheMemoizedValue_WithoutReEvaluating()
{
_evaluator.EvaluateAsync(Arg.Any<CancellationToken>()).Returns(Snapshot());
var sut = CreateSut();
var first = await sut.GetAsync();
var second = await sut.GetAsync();
await _evaluator.Received(1).EvaluateAsync(Arg.Any<CancellationToken>());
Assert.Same(first, second);
}
[Fact]
public async Task ForceRefresh_BypassesAFreshValue()
{
var snapshotA = Snapshot();
var snapshotB = new PendingUpdatesSnapshot(T0.AddMinutes(1), Array.Empty<SitePendingUpdates>());
_evaluator.EvaluateAsync(Arg.Any<CancellationToken>()).Returns(snapshotA, snapshotB);
var sut = CreateSut();
var first = await sut.GetAsync();
var forced = await sut.GetAsync(forceRefresh: true);
await _evaluator.Received(2).EvaluateAsync(Arg.Any<CancellationToken>());
Assert.Same(snapshotA, first);
Assert.Same(snapshotB, forced);
}
[Fact]
public async Task ForceRefresh_StillJoinsAnInFlightRound()
{
var released = new TaskCompletionSource<PendingUpdatesSnapshot>(
TaskCreationOptions.RunContinuationsAsynchronously);
_evaluator.EvaluateAsync(Arg.Any<CancellationToken>()).Returns(released.Task);
var sut = CreateSut();
var first = sut.GetAsync();
var forcedA = sut.GetAsync(forceRefresh: true);
var forcedB = sut.GetAsync(forceRefresh: true);
var snapshot = Snapshot();
released.SetResult(snapshot);
Assert.Same(snapshot, await first);
Assert.Same(snapshot, await forcedA);
Assert.Same(snapshot, await forcedB);
await _evaluator.Received(1).EvaluateAsync(Arg.Any<CancellationToken>());
}
[Fact]
public async Task FailedEvaluation_SurfacesToTheCaller_AndIsNotMemoized()
{
_evaluator.EvaluateAsync(Arg.Any<CancellationToken>())
.Returns(
Task.FromException<PendingUpdatesSnapshot>(new InvalidOperationException("flatten failed")),
Task.FromResult(Snapshot()));
var sut = CreateSut();
await Assert.ThrowsAsync<InvalidOperationException>(() => sut.GetAsync());
// Same instant, well inside the TTL — a cached fault would starve every dashboard
// for the rest of the window, so the next caller must start a new evaluation.
var recovered = await sut.GetAsync();
Assert.NotNull(recovered);
await _evaluator.Received(2).EvaluateAsync(Arg.Any<CancellationToken>());
}
public void Dispose() => _provider.Dispose();
}