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

merge: transport hardening (register rows 51-53) — BundleLockedException curated to Commons/MapFault, fleet-wide MS SQL unlock-throttle window, import-session gauge + cap Warning

Browse Source
This commit is contained in:
Joseph Doherty
2026-08-15 12:00:28 -04:00
parent 1be08e96cc 73f518d6a6
commit 4785f3cbef
31 changed files with 3678 additions and 101 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/Repositories/BundleUnlockAttemptRepositoryTests.cs
+90
View File
@@ -0,0 +1,90 @@
using ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Repositories;
namespace ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests.Repositories;
/// <summary>
/// Register row 52 coverage for <see cref="BundleUnlockAttemptRepository"/> over the
/// in-memory SQLite harness — the fleet-wide bundle-unlock throttle window's persistence
/// layer. Exercises the windowed count (including the bucket-isolation and window-boundary
/// cases the throttle depends on) and the bounded prune.
/// </summary>
public class BundleUnlockAttemptRepositoryTests
{
// Fixed UTC base so the window assertions are deterministic.
private static readonly DateTime Base = new(2026, 6, 1, 12, 0, 0, DateTimeKind.Utc);
private static ScadaBridgeDbContext NewContext() => SqliteTestHelper.CreateInMemoryContext();
[Fact]
public async Task CountAttemptsSinceAsync_CountsOnlyTheGivenBucketInsideTheWindow()
{
await using var ctx = NewContext();
var repo = new BundleUnlockAttemptRepository(ctx);
// Two inside the trailing hour, one aged out, one from a different client.
await repo.RecordAttemptAsync("10.0.0.1", Base.AddMinutes(-10));
await repo.RecordAttemptAsync("10.0.0.1", Base.AddMinutes(-50));
await repo.RecordAttemptAsync("10.0.0.1", Base.AddMinutes(-90));
await repo.RecordAttemptAsync("10.0.0.2", Base.AddMinutes(-5));
var count = await repo.CountAttemptsSinceAsync("10.0.0.1", Base.AddHours(-1));
Assert.Equal(2, count);
}
[Fact]
public async Task CountAttemptsSinceAsync_IsInclusiveAtTheWindowFloor()
{
// The floor is inclusive, matching the in-memory limiter's prune predicate
// (entries at exactly the cutoff still count). Pinned because an off-by-one here
// would silently hand every client one extra attempt per window.
await using var ctx = NewContext();
var repo = new BundleUnlockAttemptRepository(ctx);
await repo.RecordAttemptAsync("10.0.0.1", Base.AddHours(-1));
Assert.Equal(1, await repo.CountAttemptsSinceAsync("10.0.0.1", Base.AddHours(-1)));
}
[Fact]
public async Task CountAttemptsSinceAsync_UnknownBucket_ReturnsZero()
{
await using var ctx = NewContext();
var repo = new BundleUnlockAttemptRepository(ctx);
Assert.Equal(0, await repo.CountAttemptsSinceAsync("<unattributed>", Base.AddHours(-1)));
}
[Fact]
public async Task PruneAttemptsOlderThanAsync_DeletesOnlyAgedRows_AndIsBounded()
{
await using var ctx = NewContext();
var repo = new BundleUnlockAttemptRepository(ctx);
for (var i = 0; i < 5; i++)
{
await repo.RecordAttemptAsync("10.0.0.1", Base.AddHours(-5).AddMinutes(i));
}
await repo.RecordAttemptAsync("10.0.0.1", Base.AddMinutes(-1));
// Bounded: only 3 of the 5 eligible rows go in this pass.
var deleted = await repo.PruneAttemptsOlderThanAsync(Base.AddHours(-2), maxRows: 3);
Assert.Equal(3, deleted);
// The recent row is untouched, and a second pass clears the remainder.
Assert.Equal(1, await repo.CountAttemptsSinceAsync("10.0.0.1", Base.AddHours(-1)));
Assert.Equal(2, await repo.PruneAttemptsOlderThanAsync(Base.AddHours(-2), maxRows: 3));
Assert.Equal(0, await repo.PruneAttemptsOlderThanAsync(Base.AddHours(-2), maxRows: 3));
}
[Fact]
public async Task PruneAttemptsOlderThanAsync_RejectsANonPositiveBatchSize()
{
await using var ctx = NewContext();
var repo = new BundleUnlockAttemptRepository(ctx);
await Assert.ThrowsAsync<ArgumentOutOfRangeException>(
() => repo.PruneAttemptsOlderThanAsync(Base, maxRows: 0));
}
}
tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/ManagementActorTests.cs
+45
View File
@@ -3219,6 +3219,51 @@ public class ManagementActorTests : TestKit, IDisposable
Assert.Contains("Too many bundle unlock attempts", response.Error);
}
[Fact]
public void ImportBundleCommand_LockedBundle_SurfacesTheCuratedReason()
{
// Register row 51: the per-bundle 3-strike lockout used to reach the CLI as
// MapFault's generic "An internal error occurred (CorrelationId=...)" because
// BundleLockedException lived in Transport, which ManagementService does not
// reference. It now lives in Commons alongside BundleUnlockThrottledException
// and is curated the same way.
var (_, importer) = AddBundleSubstitutes();
importer.LoadAsync(Arg.Any<Stream>(), Arg.Any<string?>(), Arg.Any<CancellationToken>(), Arg.Any<string?>())
.Returns<Commons.Types.Transport.BundleSession>(_ =>
throw new Commons.Types.Transport.BundleLockedException("abc123", 3));
var payload = Convert.ToBase64String(new byte[] { 0x01, 0x02, 0x03 });
var actor = CreateActor();
actor.Tell(Envelope(new ImportBundleCommand(payload, "wrong", "skip"), "Administrator"));
var response = ExpectMsg<ManagementError>(TimeSpan.FromSeconds(5));
Assert.Equal("COMMAND_FAILED", response.ErrorCode);
Assert.Contains("locked after 3 failed unlock attempts", response.Error);
Assert.DoesNotContain("An internal error occurred", response.Error);
}
[Fact]
public void PreviewBundleCommand_LockedBundle_SurfacesTheCuratedReason()
{
// Register row 51: preview shares the same LoadAsync fault surface as import,
// so it needs the same curated arm.
var (_, importer) = AddBundleSubstitutes();
importer.LoadAsync(Arg.Any<Stream>(), Arg.Any<string?>(), Arg.Any<CancellationToken>(), Arg.Any<string?>())
.Returns<Commons.Types.Transport.BundleSession>(_ =>
throw new Commons.Types.Transport.BundleLockedException("abc123", 3));
var payload = Convert.ToBase64String(new byte[] { 0x01, 0x02, 0x03 });
var actor = CreateActor();
actor.Tell(Envelope(new PreviewBundleCommand(payload, "wrong"), "Administrator"));
var response = ExpectMsg<ManagementError>(TimeSpan.FromSeconds(5));
Assert.Equal("COMMAND_FAILED", response.ErrorCode);
Assert.Contains("locked after 3 failed unlock attempts", response.Error);
Assert.DoesNotContain("An internal error occurred", response.Error);
}
// ========================================================================
// Native alarm source CRUD (Task 21)
// ========================================================================
tests/ZB.MOM.WW.ScadaBridge.Transport.IntegrationTests/Import/FleetWideUnlockThrottleTests.cs
+263
View File
@@ -0,0 +1,263 @@
using Microsoft.Data.Sqlite;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.DependencyInjection;
using NSubstitute;
using NSubstitute.ExceptionExtensions;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
using ZB.MOM.WW.ScadaBridge.ConfigurationDatabase;
using ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Repositories;
using ZB.MOM.WW.ScadaBridge.Transport.Import;
namespace ZB.MOM.WW.ScadaBridge.Transport.IntegrationTests.Import;
/// <summary>
/// Register row 52: the bundle-unlock throttle's counters used to be per-node AND
/// per-process, so the effective fleet cap was N x <c>MaxUnlockAttemptsPerIpPerHour</c>
/// per client and a node restart silently cleared the window. These tests stand up TWO
/// independent limiter instances — the stand-in for central-a and central-b — over ONE
/// shared relational store and prove they enforce a single combined window, plus the
/// fail-open behaviour a store outage must produce.
/// </summary>
public sealed class FleetWideUnlockThrottleTests : IDisposable
{
private readonly SqliteConnection _connection;
private readonly ServiceProvider _provider;
public FleetWideUnlockThrottleTests()
{
// One open in-memory SQLite connection for the fixture's lifetime — the database
// lives exactly as long as the connection, and both "nodes" resolve scoped
// contexts over it, which is what makes the store genuinely shared.
_connection = new SqliteConnection("DataSource=:memory:");
_connection.Open();
var services = new ServiceCollection();
services.AddDbContext<ScadaBridgeDbContext>(opts => opts.UseSqlite(_connection));
services.AddScoped<IBundleUnlockAttemptRepository, BundleUnlockAttemptRepository>();
_provider = services.BuildServiceProvider();
using var scope = _provider.CreateScope();
scope.ServiceProvider.GetRequiredService<ScadaBridgeDbContext>().Database.EnsureCreated();
}
public void Dispose()
{
_provider.Dispose();
_connection.Dispose();
}
/// <summary>
/// Builds a limiter over the shared store, standing in for one central node's
/// singleton. Each call returns a fully independent instance (its own process-local
/// fallback buckets, its own prune claim) — exactly the isolation the two real nodes
/// have.
/// </summary>
/// <param name="clock">Clock the node observes.</param>
/// <returns>A limiter whose window is the shared store.</returns>
private BundleUnlockRateLimiter NewNode(TimeProvider clock) => new(
clock,
BundleUnlockRateLimiter.DefaultWindow,
new SqlBundleUnlockAttemptWindow(
_provider.GetRequiredService<IServiceScopeFactory>(), clock),
logger: null);
[Fact]
public async Task TwoNodesSharingOneStore_EnforceASingleCombinedWindow()
{
// The headline row-52 contract: attempts spread across both central nodes count
// against ONE budget. Before this, five attempts per node were both allowed.
var clock = new TestClock(new DateTimeOffset(2026, 6, 1, 12, 0, 0, TimeSpan.Zero));
var nodeA = NewNode(clock);
var nodeB = NewNode(clock);
for (var i = 0; i < 5; i++)
{
Assert.True(await nodeA.TryRegisterAttemptAsync("10.0.0.1", 10));
Assert.True(await nodeB.TryRegisterAttemptAsync("10.0.0.1", 10));
}
// The combined budget of 10 is now spent — on EITHER node.
Assert.False(await nodeA.TryRegisterAttemptAsync("10.0.0.1", 10));
Assert.False(await nodeB.TryRegisterAttemptAsync("10.0.0.1", 10));
}
[Fact]
public async Task DeniedAttempts_BurnNoBudget()
{
// Register row 44's contract, preserved through the move to the shared store: a
// rejected attempt is not recorded, so hammering cannot roll the window forward
// and extend the client's own lockout.
var clock = new TestClock(new DateTimeOffset(2026, 6, 1, 12, 0, 0, TimeSpan.Zero));
var nodeA = NewNode(clock);
for (var i = 0; i < 3; i++)
{
Assert.True(await nodeA.TryRegisterAttemptAsync("10.0.0.1", 3));
}
for (var i = 0; i < 5; i++)
{
Assert.False(await nodeA.TryRegisterAttemptAsync("10.0.0.1", 3));
}
// Exactly the 3 permitted attempts were persisted — the 5 denials added nothing.
using var scope = _provider.CreateScope();
var repo = scope.ServiceProvider.GetRequiredService<IBundleUnlockAttemptRepository>();
var recorded = await repo.CountAttemptsSinceAsync(
"10.0.0.1", clock.GetUtcNow().UtcDateTime.AddHours(-1));
Assert.Equal(3, recorded);
}
[Fact]
public async Task TheWindowIsRestartDurable()
{
// A node restart used to clear the window outright. A brand-new limiter instance
// (the restarted process) must still see the budget the old one spent.
var clock = new TestClock(new DateTimeOffset(2026, 6, 1, 12, 0, 0, TimeSpan.Zero));
var before = NewNode(clock);
for (var i = 0; i < 3; i++)
{
Assert.True(await before.TryRegisterAttemptAsync("10.0.0.1", 3));
}
var afterRestart = NewNode(clock);
Assert.False(await afterRestart.TryRegisterAttemptAsync("10.0.0.1", 3));
}
[Fact]
public async Task TheWindowSlides_SoAnAgedOutBudgetIsReleased()
{
var clock = new TestClock(new DateTimeOffset(2026, 6, 1, 12, 0, 0, TimeSpan.Zero));
var node = NewNode(clock);
for (var i = 0; i < 3; i++)
{
Assert.True(await node.TryRegisterAttemptAsync("10.0.0.1", 3));
}
Assert.False(await node.TryRegisterAttemptAsync("10.0.0.1", 3));
clock.Advance(TimeSpan.FromMinutes(61));
Assert.True(await node.TryRegisterAttemptAsync("10.0.0.1", 3));
}
[Fact]
public async Task BucketsAreIsolatedPerClient()
{
var clock = new TestClock(new DateTimeOffset(2026, 6, 1, 12, 0, 0, TimeSpan.Zero));
var node = NewNode(clock);
Assert.True(await node.TryRegisterAttemptAsync("10.0.0.1", 1));
Assert.False(await node.TryRegisterAttemptAsync("10.0.0.1", 1));
// A different client — and the shared unattributed bucket — are unaffected.
Assert.True(await node.TryRegisterAttemptAsync("10.0.0.2", 1));
Assert.True(await node.TryRegisterAttemptAsync("<unattributed>", 1));
}
[Fact]
public async Task StoreFailure_FailsOpen_AndDoesNotBlockTheUnlock()
{
// Availability of legitimate imports outranks a best-effort rate cap: a database
// error must permit the attempt, not take the import surface down with the DB.
var window = Substitute.For<IBundleUnlockAttemptWindow>();
window
.TryRegisterAttemptAsync(
Arg.Any<string>(), Arg.Any<int>(), Arg.Any<TimeSpan>(), Arg.Any<CancellationToken>())
.ThrowsAsync(new InvalidOperationException("database unavailable"));
var limiter = new BundleUnlockRateLimiter(
TimeProvider.System, BundleUnlockRateLimiter.DefaultWindow, window, logger: null);
for (var i = 0; i < 20; i++)
{
Assert.True(await limiter.TryRegisterAttemptAsync("10.0.0.1", 1));
}
}
[Fact]
public async Task CallerCancellation_IsNotSwallowedByTheFailOpenPath()
{
// Fail-open must not hide the caller's own cancellation — that would turn an
// aborted request into a silently permitted attempt.
var window = Substitute.For<IBundleUnlockAttemptWindow>();
window
.TryRegisterAttemptAsync(
Arg.Any<string>(), Arg.Any<int>(), Arg.Any<TimeSpan>(), Arg.Any<CancellationToken>())
.ThrowsAsync(new OperationCanceledException());
var limiter = new BundleUnlockRateLimiter(
TimeProvider.System, BundleUnlockRateLimiter.DefaultWindow, window, logger: null);
using var cts = new CancellationTokenSource();
await cts.CancelAsync();
await Assert.ThrowsAnyAsync<OperationCanceledException>(
() => limiter.TryRegisterAttemptAsync("10.0.0.1", 10, cts.Token));
}
[Fact]
public async Task WithoutASharedWindow_TheLimiterStillEnforcesAProcessLocalCap()
{
// Hosts that register Transport without a shared window keep the pre-row-52
// behaviour rather than losing the cap entirely.
var clock = new TestClock(new DateTimeOffset(2026, 6, 1, 12, 0, 0, TimeSpan.Zero));
var limiter = new BundleUnlockRateLimiter(clock, BundleUnlockRateLimiter.DefaultWindow);
Assert.False(limiter.IsFleetWide);
Assert.True(await limiter.TryRegisterAttemptAsync("10.0.0.1", 1));
Assert.False(await limiter.TryRegisterAttemptAsync("10.0.0.1", 1));
}
[Fact]
public async Task AgedRowsArePrunedOpportunistically_OnceThePruneIntervalHasElapsed()
{
// No background service: the prune rides the unlock path, claimed at most once per
// PruneMinInterval. Rows are retained for PruneRetentionWindows x window so a prune
// racing a count can never delete a row the count still needs.
var clock = new TestClock(new DateTimeOffset(2026, 6, 1, 12, 0, 0, TimeSpan.Zero));
var node = NewNode(clock);
Assert.True(await node.TryRegisterAttemptAsync("10.0.0.1", 10));
// Move past both the retention horizon (2 hours) and the prune cadence.
clock.Advance(TimeSpan.FromHours(3));
Assert.True(await node.TryRegisterAttemptAsync("10.0.0.2", 10));
using var scope = _provider.CreateScope();
var ctx = scope.ServiceProvider.GetRequiredService<ScadaBridgeDbContext>();
var remaining = await ctx.BundleUnlockAttempts.AsNoTracking().ToListAsync();
// The aged row is gone; the one just written by the pruning attempt survives.
Assert.Equal("10.0.0.2", Assert.Single(remaining).ClientKey);
}
[Fact]
public async Task RecentRowsSurviveAPrunePass()
{
// Negative control for the prune horizon: an attempt still inside the retention
// window must not be deleted, or the throttle would forget live budget.
var clock = new TestClock(new DateTimeOffset(2026, 6, 1, 12, 0, 0, TimeSpan.Zero));
var node = NewNode(clock);
Assert.True(await node.TryRegisterAttemptAsync("10.0.0.1", 10));
// Past the prune cadence (15 min) but inside the 2-hour retention horizon.
clock.Advance(TimeSpan.FromMinutes(30));
Assert.True(await node.TryRegisterAttemptAsync("10.0.0.2", 10));
using var scope = _provider.CreateScope();
var ctx = scope.ServiceProvider.GetRequiredService<ScadaBridgeDbContext>();
Assert.Equal(2, await ctx.BundleUnlockAttempts.AsNoTracking().CountAsync());
}
/// <summary>Minimal mutable-clock <see cref="TimeProvider"/> for deterministic windows.</summary>
private sealed class TestClock : TimeProvider
{
private DateTimeOffset _now;
public TestClock(DateTimeOffset start) { _now = start; }
public override DateTimeOffset GetUtcNow() => _now;
public void Advance(TimeSpan by) => _now = _now.Add(by);
}
}
tests/ZB.MOM.WW.ScadaBridge.Transport.Tests/Import/BundleSessionObservabilityTests.cs
+248
View File
@@ -0,0 +1,248 @@
using System.Diagnostics.Metrics;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
using Microsoft.Extensions.Options;
using ZB.MOM.WW.ScadaBridge.Commons.Observability;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Transport;
using ZB.MOM.WW.ScadaBridge.Transport.Import;
namespace ZB.MOM.WW.ScadaBridge.Transport.Tests.Import;
/// <summary>
/// Register row 53: <c>MaxConcurrentImportSessions</c> had no observability — nothing
/// counted open sessions and nothing logged a rejection at the cap, so the register-row-43
/// session leak could only be inferred from rotating test failures. These tests pin the
/// two signals that close that gap: the
/// <c>scadabridge.central.transport.import_sessions.open</c> gauge and the Warning-level
/// log emitted when <see cref="BundleSessionStore.Open"/> rejects at the cap.
/// </summary>
public sealed class BundleSessionObservabilityTests
{
private const string OpenSessionsGauge = "scadabridge.central.transport.import_sessions.open";
// ------------------------------------------------------------------
// Warning log on cap rejection
// ------------------------------------------------------------------
[Fact]
public void Open_AtCap_LogsWarningWithCountAndCap()
{
// The rejection used to be visible only as the exception the caller saw. The
// operator-facing signal must name BOTH the current count and the cap — that
// pair is what distinguishes genuine concurrency from a leak.
var clock = new TestClock(DateTimeOffset.UtcNow);
var opts = Options.Create(new TransportOptions { MaxConcurrentImportSessions = 3 });
var logger = new CapturingLogger();
var sut = new BundleSessionStore(opts, clock, logger);
var future = clock.GetUtcNow().AddMinutes(30);
for (var i = 0; i < 3; i++)
{
sut.Open(SessionExpiringAt(future));
}
Assert.Throws<InvalidOperationException>(() => sut.Open(SessionExpiringAt(future)));
var warning = Assert.Single(logger.Entries, e => e.Level == LogLevel.Warning);
Assert.Contains("Bundle import session rejected", warning.Message, StringComparison.Ordinal);
Assert.Contains("3 of 3", warning.Message, StringComparison.Ordinal);
Assert.Contains("MaxConcurrentImportSessions", warning.Message, StringComparison.Ordinal);
}
[Fact]
public void Open_UnderCap_LogsNothing()
{
// Negative control: the Warning must be a cap signal, not per-open noise.
var clock = new TestClock(DateTimeOffset.UtcNow);
var opts = Options.Create(new TransportOptions { MaxConcurrentImportSessions = 3 });
var logger = new CapturingLogger();
var sut = new BundleSessionStore(opts, clock, logger);
sut.Open(SessionExpiringAt(clock.GetUtcNow().AddMinutes(30)));
Assert.Empty(logger.Entries);
}
[Fact]
public void Open_AtCapOfExpiredSessions_ReclaimsAndLogsNothing()
{
// A store that is full only of stale entries reclaims and admits the caller —
// that is not a cap hit and must not produce a Warning, or the signal would fire
// routinely on any node whose sweep has not yet run.
var clock = new TestClock(DateTimeOffset.UtcNow);
var opts = Options.Create(new TransportOptions { MaxConcurrentImportSessions = 2 });
var logger = new CapturingLogger();
var sut = new BundleSessionStore(opts, clock, logger);
sut.Open(SessionExpiringAt(clock.GetUtcNow().AddMinutes(5)));
sut.Open(SessionExpiringAt(clock.GetUtcNow().AddMinutes(5)));
clock.Advance(TimeSpan.FromMinutes(10));
sut.Open(SessionExpiringAt(clock.GetUtcNow().AddMinutes(30)));
Assert.Empty(logger.Entries);
}
// ------------------------------------------------------------------
// OpenSessionCount / gauge
// ------------------------------------------------------------------
[Fact]
public void OpenSessionCount_ExcludesExpiredButUnsweptSessions()
{
// The eviction sweep runs once a minute, so a raw dictionary count would report
// the store as pinned at the cap for up to a sweep interval after a burst had
// actually drained — exactly the false signal this instrument exists to rule out.
var clock = new TestClock(DateTimeOffset.UtcNow);
var sut = new BundleSessionStore(Options.Create(new TransportOptions()), clock, NullLogger<BundleSessionStore>.Instance);
sut.Open(SessionExpiringAt(clock.GetUtcNow().AddMinutes(5)));
var live = SessionExpiringAt(clock.GetUtcNow().AddMinutes(60));
sut.Open(live);
Assert.Equal(2, sut.OpenSessionCount);
clock.Advance(TimeSpan.FromMinutes(10));
// The short session has expired but nothing has swept it yet.
Assert.Equal(1, sut.OpenSessionCount);
}
[Fact]
public async Task EvictionService_RegistersTheGaugeProvider_AndClearsItOnStop()
{
// The eviction service is Transport's only lifecycle seam, so it owns the
// process-global gauge slot. The gauge is read as a delta against a baseline
// because the instrument is a process-wide static.
var clock = new TestClock(DateTimeOffset.UtcNow);
var store = new BundleSessionStore(Options.Create(new TransportOptions()), clock, NullLogger<BundleSessionStore>.Instance);
var service = new BundleSessionEvictionService(store, NullLogger<BundleSessionEvictionService>.Instance);
var baseline = ReadGauge();
await service.StartAsync(CancellationToken.None);
try
{
store.Open(SessionExpiringAt(clock.GetUtcNow().AddMinutes(30)));
store.Open(SessionExpiringAt(clock.GetUtcNow().AddMinutes(30)));
Assert.Equal(2, ReadGauge());
}
finally
{
await service.StopAsync(CancellationToken.None);
}
// After the clear the gauge falls back to reporting 0, not through a dead store.
Assert.Equal(0, ReadGauge());
Assert.Equal(0, baseline);
}
[Fact]
public async Task EvictionService_StopDoesNotClobberANewerInstancesProvider()
{
// Identity-checked compare-and-clear: a late stop of a superseded instance must
// leave a newer registrant's provider in place (mirrors the StoreAndForward
// queue-depth gauge's contract).
var clock = new TestClock(DateTimeOffset.UtcNow);
var oldStore = new BundleSessionStore(Options.Create(new TransportOptions()), clock, NullLogger<BundleSessionStore>.Instance);
var newStore = new BundleSessionStore(Options.Create(new TransportOptions()), clock, NullLogger<BundleSessionStore>.Instance);
var oldService = new BundleSessionEvictionService(oldStore, NullLogger<BundleSessionEvictionService>.Instance);
var newService = new BundleSessionEvictionService(newStore, NullLogger<BundleSessionEvictionService>.Instance);
await oldService.StartAsync(CancellationToken.None);
await newService.StartAsync(CancellationToken.None);
try
{
newStore.Open(SessionExpiringAt(clock.GetUtcNow().AddMinutes(30)));
// Late stop of the superseded instance.
await oldService.StopAsync(CancellationToken.None);
Assert.Equal(1, ReadGauge());
}
finally
{
await newService.StopAsync(CancellationToken.None);
}
Assert.Equal(0, ReadGauge());
}
// ------------------------------------------------------------------
// Helpers
// ------------------------------------------------------------------
/// <summary>
/// Forces one synchronous observation of the open-import-sessions gauge and returns
/// the measured value. Mirrors the MeterListener helper used by the Communication
/// and StoreAndForward telemetry suites.
/// </summary>
/// <returns>The gauge's current value, or 0 if it never reported.</returns>
private static long ReadGauge()
{
long observed = 0;
using var listener = new MeterListener();
listener.InstrumentPublished = (instrument, l) =>
{
if (instrument.Meter.Name == ScadaBridgeTelemetry.MeterName
&& instrument.Name == OpenSessionsGauge)
{
l.EnableMeasurementEvents(instrument);
}
};
listener.SetMeasurementEventCallback<long>((_, measurement, _, _) => observed = measurement);
listener.Start();
listener.RecordObservableInstruments();
return observed;
}
private static BundleSession SessionExpiringAt(DateTimeOffset expiresAt) => new()
{
SessionId = Guid.NewGuid(),
Manifest = StubManifest(),
DecryptedContent = Array.Empty<byte>(),
ExpiresAt = expiresAt,
};
private static BundleManifest StubManifest() => new(
BundleFormatVersion: 1,
SchemaVersion: "1.0",
CreatedAtUtc: DateTimeOffset.UnixEpoch,
SourceEnvironment: "test",
ExportedBy: "t",
ScadaBridgeVersion: "1",
ContentHash: "0",
Encryption: null,
Summary: new BundleSummary(0, 0, 0, 0, 0, 0, 0, 0),
Contents: Array.Empty<ManifestContentEntry>());
/// <summary>Minimal mutable-clock <see cref="TimeProvider"/> for deterministic expiry.</summary>
private sealed class TestClock : TimeProvider
{
private DateTimeOffset _now;
public TestClock(DateTimeOffset start) { _now = start; }
public override DateTimeOffset GetUtcNow() => _now;
public void Advance(TimeSpan by) => _now = _now.Add(by);
}
/// <summary>Captures formatted log entries so the cap-rejection Warning can be asserted.</summary>
private sealed class CapturingLogger : ILogger<BundleSessionStore>
{
public List<(LogLevel Level, string Message)> Entries { get; } = [];
public IDisposable? BeginScope<TState>(TState state) where TState : notnull => null;
public bool IsEnabled(LogLevel logLevel) => true;
public void Log<TState>(
LogLevel logLevel,
EventId eventId,
TState state,
Exception? exception,
Func<TState, Exception?, string> formatter)
{
Entries.Add((logLevel, formatter(state, exception)));
}
}
}