feat(transport): fleet-wide bundle-unlock throttle window (register row 52)
The per-IP unlock throttle's counters were a process-local ConcurrentDictionary on each central node, so the effective fleet cap was N x MaxUnlockAttemptsPerIpPerHour per client (failing over to the other node handed the client a fresh budget) and a node restart cleared the window outright. The window now lives in the central MS SQL BundleUnlockAttempt table — the HA store both central nodes already share, so no new infrastructure, no new port, no consensus protocol: - Commons: BundleUnlockAttempt entity + IBundleUnlockAttemptRepository. - ConfigurationDatabase: entity config (one index, IX_BundleUnlockAttempt_Client_Attempted, serving both the windowed count and the prune), repository (AsNoTracking count, ExecuteDeleteAsync prune), DbSet, DI registration, migration AddBundleUnlockAttemptTable + the hand-written idempotent production script under docs/plans/sql/. - Transport: IBundleUnlockAttemptWindow seam + SqlBundleUnlockAttemptWindow (singleton; resolves the scoped repository from its own scope per call, like the central purge singletons). BundleUnlockRateLimiter keeps the policy and gains TryRegisterAttemptAsync; BundleImporter.LoadAsync calls it. Behaviour preserved from row 44: locked bundles spend no hourly budget (the lockout check still runs first), throttle denials burn no per-bundle strike and are NOT recorded so they cannot roll the window forward, unattributable clients share the stricter <unattributed> bucket, and the budget counts attempts rather than failures. Config keys and their semantics are unchanged, so the rig's 500 override stays valid. Three deliberate trades, documented in both Transport docs: - FAILS OPEN. A store error logs a Warning naming the client and cap, and permits the attempt. This is a best-effort rate cap over an already authenticated, Administrator-gated surface — the availability of legitimate imports outranks the cap, and a DB blip must not make the import wizard unusable. Falling back to the process-local window was considered and rejected: it would still block a legitimate operator's 11th import. - Approximate. Count-then-insert is not atomic across nodes, so simultaneous attempts can overshoot by the number in flight — bounded, self-correcting within the window, and not worth provider-specific raw SQL. - Pruned opportunistically on the unlock path (CAS-claimed at most every 15 min per process, bounded to 1000 rows, retaining 2x the window, best-effort try/catch) rather than by a new background service — Transport owns no central scheduler. Same shape as DeploymentService's terminal-record sweep. Tests: Transport.IntegrationTests 116 passed (+10 — two limiter instances over one store enforcing a single combined window, restart durability, denials burning no budget, sliding release, bucket isolation, store-failure fail-open, caller-cancellation not swallowed, no-shared-window fallback, prune and its negative control); ConfigurationDatabase 395 passed (+5).
This commit is contained in:
Joseph Doherty
+90
@@ -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));
|
||||
}
|
||||
}
|
||||
+263
@@ -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);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user