chore(followups): reviewer-recommended tests, comments, and hardening from the remediation reviews
The remediation reviews approved every task but left a tail of small notes. This lands the gateway-side half of them. Hardening (behavior changes, all narrow): - BuildFilteredWriteBulkCommand's unreachable default case failed OPEN: a fifth bulk-write kind added upstream without a filter case here would have shipped the DENIED entries to the worker while reporting them denied to the caller. It now throws UnreachableException. - SqliteCanonicalAuditStore.ListRecentAsync no longer throws on a row it cannot date. The retention sweep deliberately preserves such rows (SQLite's datetime() yields NULL, so the DELETE never matches), which guaranteed the dashboard's recent-audit view would meet one eventually and lose the whole page to it. The row is now reported at DateTimeOffset.MinValue with every other column intact, behind an optional logger. - The audit drain loop's finally now completes the channel writer alongside detaching the drain, so a producer that raced past the attached check takes the write-through branch instead of stranding its event in a buffer nobody reads until shutdown. TryComplete is idempotent, so StopAsync is unaffected. Tests: - MapCommandReply ownership (Assert.Same on the inner reply), mirroring the existing MapEvent ownership test. - Redactor key-id length boundary at exactly 64 and 65 characters, pinning which way it fails. Nothing validates key-id length at creation, so docs/Diagnostics.md's "which no issued key id does" is now stated as the heuristic it is. - ApiKeyFailureLimiter.Reset with a PartitionResolution whose partition was evicted between the Check and the Reset: inert, and clears nobody else's block. - Constraint-cache concurrency stress: the cap is enforced by the inserting thread, so overshoot must be transient and proportional to the in-flight inserters, and the cache must settle at or under the cap. - ListRecentAsync against a raw-SQL undateable row. Comment/doc accuracy: - EventsHubViewerRegistry.ReleaseConnection records that it relies on SignalR's default sequential per-connection dispatch (MaximumParallelInvocationsPerClient = 1). - A PERF(followup) note on Invoke's double session resolve and why removing it needs a SessionManager overload. - SessionEventDistributor: the volatile-field comment named the pump as the lock-free reader, but the pump's single capture point is inside _replayLock; the genuinely lock-free reader is SubscriberCount. OnSubscriberOverflow's "cannot be observed here" now excepts the DisposeAsync abandon path. The churn test names its ConcurrentDictionary bucket-order assumption and that a violation surfaces as a read timeout, not a silent pass. - The two "restores the sequential drain's behavior" claims (SessionManager, docs/Sessions.md) were wrong: the sequential drain leaked too, because KillWorkerAsync's entry ThrowIfCancellationRequested aborted the whole loop on the first session for zero kills. Reworded to "fixes a leak the sequential drain also had", with the sweep-bound/shutdown-unbound ParallelOptions asymmetry explained. - ISessionManager.ShutdownAsync's token doc: it degrades the drain to a kill sweep rather than cancelling it, with the bounded overrun stated. SessionShutdownHostedService.StopAsync records that its cancellation-logging branch is now unreachable.
This commit is contained in:
Joseph Doherty
@@ -91,6 +91,11 @@ public sealed class EventsHubViewerRegistry
|
||||
return;
|
||||
}
|
||||
|
||||
// Detaching the set is safe against a SubscribeSession that arrives after the disconnect
|
||||
// only because SignalR dispatches a connection's hub invocations sequentially by default
|
||||
// (MaximumParallelInvocationsPerClient = 1): OnDisconnectedAsync cannot overlap an
|
||||
// AddViewer for the same connection, so no late add can re-create the entry and leak a
|
||||
// count that nothing will ever release. Raising that option would break this.
|
||||
if (!_sessionsByConnection.TryRemove(connectionId, out ConcurrentDictionary<string, byte>? sessions))
|
||||
{
|
||||
return;
|
||||
|
||||
@@ -101,6 +101,12 @@ public sealed class MxAccessGatewayService(
|
||||
try
|
||||
{
|
||||
requestValidator.ValidateInvoke(request);
|
||||
|
||||
// PERF(followup): this resolve and the sessionManager.InvokeAsync below look the same
|
||||
// session up twice (a dictionary hit each, so measured cost is negligible). Collapsing
|
||||
// them needs a SessionManager overload taking an already-resolved GatewaySession, which
|
||||
// would duplicate InvokeAsync's fault mapping (SessionNotFound / state checks / metrics)
|
||||
// at a second entry point — deliberately not worth it until a profile says otherwise.
|
||||
GatewaySession session = ResolveSession(request.SessionId);
|
||||
MxCommand command = request.Command;
|
||||
BulkConstraintPlan? bulkConstraintPlan = await ApplyConstraintsAsync(
|
||||
@@ -697,9 +703,13 @@ public sealed class MxAccessGatewayService(
|
||||
|
||||
default:
|
||||
// Only the four bulk-write kinds above reach FilterWriteBulkAsync, so this is
|
||||
// unreachable; keep the previous behaviour (the unmodified command) rather than
|
||||
// emitting a payload-less one if that ever stops holding.
|
||||
return command.Clone();
|
||||
// unreachable. It throws rather than falling back to the unmodified command,
|
||||
// because that fallback failed OPEN: a fifth bulk-write kind added upstream
|
||||
// without a case here would silently ship the DENIED entries to the worker while
|
||||
// still reporting them denied to the caller. Failing loud on a kind nobody can
|
||||
// reach today is strictly safer than a constraint bypass nobody would notice.
|
||||
throw new UnreachableException(
|
||||
$"Command kind {command.Kind} reached bulk-write constraint filtering without a filter case.");
|
||||
}
|
||||
|
||||
return filtered;
|
||||
|
||||
@@ -246,6 +246,13 @@ public sealed class AuditDrainService(
|
||||
finally
|
||||
{
|
||||
writer.DetachDrain();
|
||||
|
||||
// Detaching alone leaves a racer that already passed the attached check enqueueing into
|
||||
// a channel this loop will never read again — those events would sit in the buffer until
|
||||
// StopAsync's final drain. Completing the writer as well makes that racer's TryWrite
|
||||
// return false, which is the write-through branch, so the event reaches the store now.
|
||||
// TryComplete is idempotent, so StopAsync's own CompleteWriting stays safe either way.
|
||||
writer.CompleteWriting();
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -87,8 +87,11 @@ public sealed class ChannelAuditWriter : IAuditWriter
|
||||
public void DetachDrain() => Volatile.Write(ref _drainAttached, 0);
|
||||
|
||||
/// <summary>
|
||||
/// Enqueues a canonical audit event for the drain to persist. Never blocks, never throws,
|
||||
/// and never touches the store on the caller's thread while a drain is attached.
|
||||
/// Enqueues a canonical audit event for the drain to persist. Never blocks and never throws.
|
||||
/// It also never touches the store on the caller's thread while a drain is attached — with one
|
||||
/// exception: once the channel has been completed (shutdown, or a drain loop that died), the
|
||||
/// enqueue fails and this falls through to the synchronous write, which is what keeps the event
|
||||
/// rather than stranding it in a buffer nobody reads.
|
||||
/// </summary>
|
||||
/// <param name="auditEvent">The canonical audit event to persist.</param>
|
||||
/// <param name="cancellationToken">Token honoured only by the direct write-through path.</param>
|
||||
|
||||
@@ -31,7 +31,14 @@ namespace ZB.MOM.WW.MxGateway.Server.Security.Audit;
|
||||
/// retries.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
public sealed class SqliteCanonicalAuditStore(AuthSqliteConnectionFactory connectionFactory) : IAuditEventSink
|
||||
/// <param name="connectionFactory">Factory for connections to the shared auth database file.</param>
|
||||
/// <param name="logger">
|
||||
/// Optional logger for row-level read diagnostics. Optional because the store is also constructed
|
||||
/// directly by the <c>apikey</c> CLI path and by DI-free tests, which have no logger to hand.
|
||||
/// </param>
|
||||
public sealed class SqliteCanonicalAuditStore(
|
||||
AuthSqliteConnectionFactory connectionFactory,
|
||||
ILogger<SqliteCanonicalAuditStore>? logger = null) : IAuditEventSink
|
||||
{
|
||||
private const string CreateTableSql =
|
||||
"""
|
||||
@@ -208,7 +215,7 @@ public sealed class SqliteCanonicalAuditStore(AuthSqliteConnectionFactory connec
|
||||
events.Add(new AuditEvent
|
||||
{
|
||||
EventId = Guid.Parse(reader.GetString(0)),
|
||||
OccurredAtUtc = ParseUtc(reader.GetString(1)),
|
||||
OccurredAtUtc = ParseUtcOrMinValue(reader.GetString(1), reader.GetString(0)),
|
||||
Actor = reader.GetString(2),
|
||||
Action = reader.GetString(3),
|
||||
Outcome = Enum.Parse<AuditOutcome>(reader.GetString(4)),
|
||||
@@ -239,6 +246,26 @@ public sealed class SqliteCanonicalAuditStore(AuthSqliteConnectionFactory connec
|
||||
Volatile.Write(ref _tableEnsured, 1);
|
||||
}
|
||||
|
||||
private static DateTimeOffset ParseUtc(string value) =>
|
||||
DateTimeOffset.Parse(value, CultureInfo.InvariantCulture, DateTimeStyles.RoundtripKind);
|
||||
// Reading is defensive where writing is not: an insert always round-trips "O", but the table is
|
||||
// append-only shared state that an operator (or a future migration) can put an unparseable
|
||||
// timestamp into, and the retention sweep deliberately keeps such a row — SQLite's datetime()
|
||||
// yields NULL for it, so the DELETE's comparison is never true. A throwing Parse here would let
|
||||
// that single row take out the dashboard's whole recent-audit view. MinValue instead sorts the
|
||||
// row to the far past and keeps every other column readable, which is what an operator looking
|
||||
// at the view actually needs.
|
||||
private DateTimeOffset ParseUtcOrMinValue(string value, string eventId)
|
||||
{
|
||||
if (DateTimeOffset.TryParse(value, CultureInfo.InvariantCulture, DateTimeStyles.RoundtripKind, out DateTimeOffset parsed))
|
||||
{
|
||||
return parsed;
|
||||
}
|
||||
|
||||
// Debug, not warning: the row is still returned and the timestamp text itself is not logged
|
||||
// (audit rows are not secrets, but the value is attacker-influenceable in the worst case).
|
||||
logger?.LogDebug(
|
||||
"Audit event {EventId} has an unparseable occurred_at_utc; reporting it as DateTimeOffset.MinValue.",
|
||||
eventId);
|
||||
|
||||
return DateTimeOffset.MinValue;
|
||||
}
|
||||
}
|
||||
|
||||
+6
-1
@@ -97,8 +97,13 @@ public static class AuthStoreServiceCollectionExtensions
|
||||
sp.GetService<TimeProvider>() ?? TimeProvider.System));
|
||||
DecorateVerifierWithCache(services, security);
|
||||
|
||||
// GetService, not GetRequiredService, for the same reason the writer registration below
|
||||
// gives: the DI-only unit tests build a bare ServiceCollection with no AddLogging(). The
|
||||
// store's logger is optional and only carries row-level read diagnostics.
|
||||
services.AddSingleton(sp =>
|
||||
new SqliteCanonicalAuditStore(sp.GetRequiredService<AuthSqliteConnectionFactory>()));
|
||||
new SqliteCanonicalAuditStore(
|
||||
sp.GetRequiredService<AuthSqliteConnectionFactory>(),
|
||||
sp.GetService<ILogger<SqliteCanonicalAuditStore>>()));
|
||||
services.AddSingleton<IAuditEventSink>(sp => sp.GetRequiredService<SqliteCanonicalAuditStore>());
|
||||
// Resolve the logger defensively: the production host always registers ILogger<T>, but the
|
||||
// DI-only auth/CLI/dashboard unit tests build a bare ServiceCollection without AddLogging().
|
||||
|
||||
@@ -79,7 +79,14 @@ public interface ISessionManager
|
||||
CancellationToken cancellationToken);
|
||||
|
||||
/// <summary>Shuts down all sessions and the session manager.</summary>
|
||||
/// <param name="cancellationToken">Token to cancel the asynchronous operation.</param>
|
||||
/// <param name="cancellationToken">
|
||||
/// Token that <em>degrades</em> the drain rather than cancelling it. It is passed only to each
|
||||
/// session's graceful close; the drain loop and the kill fallback are not bound to it, so
|
||||
/// cancelling turns the drain into a kill sweep instead of abandoning the untried sessions as
|
||||
/// leaked workers. The call therefore overruns a cancelled token by a bounded amount —
|
||||
/// roughly <c>ceil(sessionCount / 4)</c> batches of the worker shutdown timeout in the worst
|
||||
/// case, where 4 is <c>MaxParallelSessionCloses</c>.
|
||||
/// </param>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
Task ShutdownAsync(CancellationToken cancellationToken);
|
||||
}
|
||||
|
||||
@@ -112,8 +112,11 @@ public sealed class SessionEventDistributor : IAsyncDisposable
|
||||
// inside the _lifecycleLock section of every register/unregister; never mutated in
|
||||
// place, so the pump can walk the array it captured with no lock and no allocation.
|
||||
// Volatile.Write / Volatile.Read ORDER the access — they keep the publishing store from
|
||||
// sinking past the lock release and keep the pump's read from being hoisted out of the
|
||||
// fan-out loop. They do NOT promise freshness, and nothing here needs them to: a reader
|
||||
// sinking past the lock release, and they keep a lock-free reader's load from being hoisted
|
||||
// or cached. The pump is NOT that reader: its single capture point sits inside the
|
||||
// _replayLock section of AppendToReplayBufferAndCaptureSubscribers, so the lock edge already
|
||||
// orders it. The genuinely lock-free reader is SubscriberCount, which loads the field with no
|
||||
// lock at all. They do NOT promise freshness, and nothing here needs them to: a reader
|
||||
// may legitimately observe the previous array, which IS the documented "late subscribers
|
||||
// see events after they register" window. Where visibility must be guaranteed — the
|
||||
// RegisterWithReplay handoff — it comes from the _replayLock edge, not from Volatile.
|
||||
@@ -700,8 +703,11 @@ public sealed class SessionEventDistributor : IAsyncDisposable
|
||||
// every path that completes a channel during fan-out (lease disposal via Unregister,
|
||||
// and this method) removes the subscriber from the set BEFORE completing it, so a
|
||||
// completed channel implies the subscriber is already gone and RemoveSubscriber
|
||||
// returns false. (CompleteAllSubscribers completes without removing, but only after
|
||||
// the pump has left its loop, so it cannot be observed here.)
|
||||
// returns false. (CompleteAllSubscribers completes without removing, but only after the
|
||||
// pump has left its loop, so it cannot be observed here — except on the DisposeAsync
|
||||
// abandon path: a source factory that ignores cancellation past the 5 s shutdown timeout
|
||||
// leaves the pump fanning while DisposeAsync completes subscribers, so a spurious overflow
|
||||
// report is possible there. It is harmless, because the session is already being disposed.)
|
||||
//
|
||||
// Bailing out on false is what keeps a normal stream ending mid-traffic from emitting
|
||||
// a bogus EventQueueOverflow metric and — under the default single-subscriber FailFast
|
||||
|
||||
@@ -365,11 +365,19 @@ public sealed class SessionManager : ISessionManager
|
||||
// rather than adopting them).
|
||||
//
|
||||
// For the same reason the loop itself is NOT bound to cancellationToken: a cancelled
|
||||
// ParallelOptions token stops dispatching the remaining sessions entirely, whereas the
|
||||
// sequential drain this replaced let every remaining session fail its graceful close fast
|
||||
// and still kill its worker. The token is passed to the graceful close instead, which
|
||||
// preserves that behavior — a host stop deadline turns the drain into a kill sweep rather
|
||||
// than into a leak.
|
||||
// ParallelOptions token stops dispatching the remaining sessions entirely, so a stop
|
||||
// deadline would leave the untried tail neither closed nor killed. Note this FIXES a leak
|
||||
// the sequential drain also had rather than restoring its behavior: there the kill fallback
|
||||
// ran on the caller's cancelled token, and KillWorkerAsync's entry
|
||||
// ThrowIfCancellationRequested threw out of the loop on the first session — zero kills, not
|
||||
// "fail fast and still kill". The token is passed to the graceful close only, and the kill
|
||||
// runs on CancellationToken.None, so a host stop deadline turns the drain into a kill sweep
|
||||
// rather than into a leak.
|
||||
//
|
||||
// The asymmetry with CloseExpiredLeasesAsync (whose ParallelOptions IS token-bound) is
|
||||
// intentional: that sweep is periodic maintenance whose missed sessions are picked up by
|
||||
// the next pass and, ultimately, by this drain. This drain is terminal — nothing runs after
|
||||
// it — so it must not be abandoned partway.
|
||||
await Parallel.ForEachAsync(
|
||||
_registry.Snapshot(),
|
||||
new ParallelOptions { MaxDegreeOfParallelism = MaxParallelSessionCloses },
|
||||
|
||||
@@ -16,7 +16,14 @@ public sealed class SessionShutdownHostedService(
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
/// <summary>Shuts down all gateway sessions as the host stops, logging (without throwing) if the host's shutdown timeout cancels the operation first.</summary>
|
||||
/// <summary>Shuts down all gateway sessions as the host stops.</summary>
|
||||
/// <remarks>
|
||||
/// The catch below is now effectively unreachable: <see cref="ISessionManager.ShutdownAsync"/>
|
||||
/// no longer aborts on the host's shutdown timeout, it degrades to a kill sweep and logs a
|
||||
/// per-session warning for each session that failed its graceful close. The clause is kept as
|
||||
/// a cheap guard against that contract regressing, not as an expected path — the operator
|
||||
/// signal for a timed-out shutdown is now those per-session warnings.
|
||||
/// </remarks>
|
||||
/// <param name="cancellationToken">Token that signals the host's shutdown timeout has elapsed.</param>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
public async Task StopAsync(CancellationToken cancellationToken)
|
||||
|
||||
@@ -50,6 +50,33 @@ public sealed class GatewayLogRedactorTests
|
||||
Assert.DoesNotContain("super-secret", redacted, StringComparison.Ordinal);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Pins the 64-character key-id boundary in both directions. Nothing validates key-id length at
|
||||
/// creation time (neither <c>ApiKeyAdminCommandLineParser.IsValidKeyId</c> nor
|
||||
/// <c>DashboardApiKeyManagementService.ValidateKeyId</c> caps it), so the cap here is a redaction
|
||||
/// heuristic that a real key id can cross. This test fixes which way it fails when it does: at
|
||||
/// exactly 64 the id is still an identifier and survives; at 65 the whole run is treated as secret
|
||||
/// material and goes. Losing an identifier is the cheap failure; logging a secret is not.
|
||||
/// </summary>
|
||||
/// <param name="keyIdLength">Length of the key id presented before the secret separator.</param>
|
||||
/// <param name="expectsKeyIdPreserved">Whether the key id must survive redaction at that length.</param>
|
||||
[Theory]
|
||||
[InlineData(64, true)]
|
||||
[InlineData(65, false)]
|
||||
public void RedactClientIdentity_KeyIdLengthBoundary_FailsTowardRedaction(
|
||||
int keyIdLength,
|
||||
bool expectsKeyIdPreserved)
|
||||
{
|
||||
string keyId = new('a', keyIdLength);
|
||||
|
||||
string? redacted = GatewayLogRedactor.RedactClientIdentity($"Bearer mxgw_{keyId}_super-secret");
|
||||
|
||||
Assert.Equal(
|
||||
expectsKeyIdPreserved ? $"Bearer mxgw_{keyId}_[redacted]" : "Bearer mxgw_[redacted]",
|
||||
redacted);
|
||||
Assert.DoesNotContain("super-secret", redacted, StringComparison.Ordinal);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Verifies that anything not recognized as a gateway API key fails closed: the scheme word
|
||||
/// survives only when it looks like an auth scheme, and the credential never does.
|
||||
|
||||
@@ -120,6 +120,30 @@ public sealed class MxAccessGrpcMapperTests
|
||||
Assert.Equal(ProtocolStatusCode.ProtocolViolation, publicReply.ProtocolStatus.Code);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Verifies MapCommandReply transfers ownership of the inner MxCommandReply the same way
|
||||
/// MapEvent does: the returned reference is the instance carried by the WorkerCommandReply,
|
||||
/// not a clone. The WorkerCommandReply is discarded after mapping and the awaiting Invoke
|
||||
/// call is its single consumer, so moving the inner reply out is safe and avoids a deep copy
|
||||
/// of a potentially large bulk-read payload.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void MapCommandReply_TransfersOwnershipOfInnerReplyWithoutCloning()
|
||||
{
|
||||
MxCommandReply innerReply = new()
|
||||
{
|
||||
SessionId = "session-1",
|
||||
Kind = MxCommandKind.Register,
|
||||
ProtocolStatus = MxAccessGrpcMapper.Ok(),
|
||||
Register = new RegisterReply { ServerHandle = 50 },
|
||||
};
|
||||
WorkerCommandReply workerReply = new() { Reply = innerReply };
|
||||
|
||||
MxCommandReply mapped = new MxAccessGrpcMapper().MapCommandReply(workerReply);
|
||||
|
||||
Assert.Same(innerReply, mapped);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Verifies MapEvent transfers ownership of the inner MxEvent (GWC-07 / IPC-05): the
|
||||
/// returned reference is the same instance carried by the WorkerEvent, not a clone. The
|
||||
|
||||
@@ -1066,6 +1066,14 @@ public sealed class SessionEventDistributorTests
|
||||
/// racing the fan-out must never drop, duplicate, or reorder an event for a subscriber
|
||||
/// registered throughout — nor leave the array and the dictionary disagreeing on the
|
||||
/// subscriber count once the churn stops.
|
||||
/// <para>
|
||||
/// One assumption worth naming: the snapshot is rebuilt from
|
||||
/// <c>ConcurrentDictionary.Values</c>, whose bucket order happens to keep the long-lived
|
||||
/// stable subscriber ahead of the churned ones here, so it is written to before a churned
|
||||
/// subscriber's disposal can interleave. If that ever stops holding the test does not
|
||||
/// silently pass — it fails as a <see cref="ReadTimeout"/> expiry on the read below, because
|
||||
/// an event dropped for the stable subscriber never arrives.
|
||||
/// </para>
|
||||
/// </summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[Fact]
|
||||
|
||||
@@ -219,6 +219,33 @@ public sealed class ChannelAuditWriterTests : IDisposable
|
||||
(await ListActionsAsync(factory)).OrderBy(action => action, StringComparer.Ordinal));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The sweep above deliberately preserves rows it cannot date, so the dashboard's recent-audit
|
||||
/// view is guaranteed to meet one eventually. Reading must therefore be defensive: the
|
||||
/// undateable row is reported at <see cref="DateTimeOffset.MinValue"/> with every other column
|
||||
/// intact, rather than one bad row throwing the whole page away.
|
||||
/// </summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[Fact]
|
||||
public async Task SqliteStore_ListRecent_WithUnparseableRow_ReportsMinValueInsteadOfThrowing()
|
||||
{
|
||||
(SqliteCanonicalAuditStore store, AuthSqliteConnectionFactory factory) = CreateStore();
|
||||
|
||||
DateTimeOffset occurred = new(2026, 5, 18, 0, 0, 0, TimeSpan.Zero);
|
||||
await store.InsertAsync(MakeEvent("dateable", occurred), CancellationToken.None);
|
||||
await InsertRawRowAsync(factory, "undateable", "0000-not-a-timestamp");
|
||||
|
||||
IReadOnlyList<AuditEvent> recent = await store.ListRecentAsync(10, CancellationToken.None);
|
||||
|
||||
Assert.Equal(2, recent.Count);
|
||||
Assert.Equal(occurred, recent.Single(auditEvent => auditEvent.Action == "dateable").OccurredAtUtc);
|
||||
|
||||
AuditEvent undateable = recent.Single(auditEvent => auditEvent.Action == "undateable");
|
||||
Assert.Equal(DateTimeOffset.MinValue, undateable.OccurredAtUtc);
|
||||
Assert.Equal("operator01", undateable.Actor);
|
||||
Assert.Equal(AuditOutcome.Denied, undateable.Outcome);
|
||||
}
|
||||
|
||||
private static (ChannelAuditWriter Writer, AuditDrainService Drain) CreateWriterAndDrain(
|
||||
IAuditEventSink sink,
|
||||
SecurityOptions? security = null,
|
||||
@@ -265,7 +292,8 @@ public sealed class ChannelAuditWriterTests : IDisposable
|
||||
await command.ExecuteNonQueryAsync(CancellationToken.None);
|
||||
}
|
||||
|
||||
// Reads actions straight from SQL: ListRecentAsync would throw on the unparseable timestamp.
|
||||
// Reads actions straight from SQL so the sweep assertion depends on the DELETE alone, with no
|
||||
// opinion from the store's read path about how an undateable row is surfaced.
|
||||
private static async Task<List<string>> ListActionsAsync(AuthSqliteConnectionFactory factory)
|
||||
{
|
||||
await using SqliteConnection connection = await factory.OpenConnectionAsync(CancellationToken.None);
|
||||
|
||||
@@ -281,6 +281,55 @@ public sealed class CachingApiKeyVerifierTests
|
||||
Assert.Same(reparsed, MapConstraints(firstJson));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Stress: the constraint cache's bound is enforced by the inserting thread itself
|
||||
/// (<c>GetOrAdd</c>, then enqueue, then evict), so concurrent inserters can each land an
|
||||
/// entry before any of them reaches the eviction step. That overshoot is real but must be
|
||||
/// transient and proportional to the in-flight inserters — not unbounded growth — and once
|
||||
/// the churn stops the cache must be back at or under the cap. Hammered on both the
|
||||
/// converging path (every iteration maps the same blob, so all but one <c>GetOrAdd</c>
|
||||
/// loses) and the growth path (a distinct blob per iteration, which is what forces
|
||||
/// eviction).
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void ToGatewayIdentity_ConcurrentBlobs_OvershootIsTransientAndCacheSettlesUnderCap()
|
||||
{
|
||||
const int cap = GatewayApiKeyIdentityMapper.MaxCachedConstraintBlobs;
|
||||
|
||||
// Overshoot is bounded by how many inserters can sit between their GetOrAdd and their own
|
||||
// EvictIfOverCapacity, so scale the allowance with the available parallelism rather than
|
||||
// pinning a magic number. Generous on purpose: the assertion under test is "bounded", not
|
||||
// "bounded by exactly this".
|
||||
int transientAllowance = (Environment.ProcessorCount * 8) + 64;
|
||||
string sharedJson = ConstraintsJson("Area_StressShared");
|
||||
int peak = 0;
|
||||
|
||||
Parallel.For(0, (cap * 2) + 64, index =>
|
||||
{
|
||||
MapConstraints(sharedJson);
|
||||
MapConstraints(ConstraintsJson($"Area_Stress_{index}"));
|
||||
|
||||
int size = GatewayApiKeyIdentityMapper.CurrentCacheSize;
|
||||
int seen = Volatile.Read(ref peak);
|
||||
while (size > seen && Interlocked.CompareExchange(ref peak, size, seen) != seen)
|
||||
{
|
||||
seen = Volatile.Read(ref peak);
|
||||
}
|
||||
});
|
||||
|
||||
Assert.True(
|
||||
peak <= cap + transientAllowance,
|
||||
$"cache peaked at {peak} entries, past the {cap} cap plus the {transientAllowance} transient allowance");
|
||||
|
||||
// The cache is process-wide static and other test classes in this assembly map identities
|
||||
// too, so poll rather than asserting on the instant the loop returns.
|
||||
Assert.True(
|
||||
SpinWait.SpinUntil(
|
||||
() => GatewayApiKeyIdentityMapper.CurrentCacheSize <= cap,
|
||||
TimeSpan.FromSeconds(5)),
|
||||
$"cache settled at {GatewayApiKeyIdentityMapper.CurrentCacheSize} entries, past the {cap} cap");
|
||||
}
|
||||
|
||||
private static ApiKeyConstraints MapConstraints(string constraintsJson) =>
|
||||
GatewayApiKeyIdentityMapper.ToGatewayIdentity(new LibApiKeyIdentity(
|
||||
KeyId: "operator01",
|
||||
|
||||
@@ -414,6 +414,46 @@ public sealed class ApiKeyFailureLimiterTests
|
||||
Assert.True(limiter.IsTracked(arriving));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The <c>PartitionResolution</c> handed back by <c>Check</c> is carried across an await (the
|
||||
/// inner verification) before <c>Reset</c> consumes it, so the partition it names can be evicted
|
||||
/// in between. Applying a stale resolution must be inert — never throw, and never clear a
|
||||
/// partition or aggregate that belongs to somebody else. The failure direction that matters is
|
||||
/// "clears too little", which costs the caller a probe wait; "clears somebody else's block"
|
||||
/// would be a throttle bypass.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void Reset_WithStaleResolutionForEvictedPartition_IsInertAndClearsNoOtherBlock()
|
||||
{
|
||||
ManualTimeProvider clock = new(DateTimeOffset.UnixEpoch);
|
||||
ApiKeyFailureLimiter limiter = CreateLimiter(clock, limit: 3, aggregateLimit: 3, maxPartitions: 2);
|
||||
ApiKeyThrottlePartition victim = new("ipv4:10.0.0.1:1", "victim");
|
||||
ApiKeyThrottlePartition other = new("ipv4:10.0.0.2:1", "other");
|
||||
ApiKeyThrottlePartition arriving = new("ipv4:10.0.0.3:1", "arriving");
|
||||
|
||||
RecordFailures(limiter, victim, 3);
|
||||
Assert.Equal(
|
||||
ApiKeyThrottleDecision.ThrottledByPeer,
|
||||
limiter.Check(victim, out ApiKeyFailureLimiter.PartitionResolution stale));
|
||||
Assert.True(stale.IsResolved);
|
||||
|
||||
// Age the victim's window out, then push the map past its cap so eviction takes it —
|
||||
// expired windows are the first eviction preference, so this is deterministic. The
|
||||
// resolution captured above now names a partition that no longer exists.
|
||||
clock.Advance(Window + TimeSpan.FromSeconds(1));
|
||||
RecordFailures(limiter, other, 3);
|
||||
RecordFailures(limiter, arriving, 3);
|
||||
Assert.False(limiter.IsTracked(victim));
|
||||
|
||||
limiter.Reset(victim, stale);
|
||||
|
||||
Assert.True(limiter.IsTracked(other));
|
||||
Assert.True(limiter.IsTracked(arriving));
|
||||
Assert.Equal(2, limiter.TrackedAggregateCount);
|
||||
Assert.Equal(ApiKeyThrottleDecision.ThrottledByPeer, limiter.Check(other));
|
||||
Assert.Equal(ApiKeyThrottleDecision.ThrottledByPeer, limiter.Check(arriving));
|
||||
}
|
||||
|
||||
private static void RecordFailures(ApiKeyFailureLimiter limiter, ApiKeyThrottlePartition partition, int count)
|
||||
{
|
||||
for (int i = 0; i < count; i++)
|
||||
|
||||
Reference in New Issue
Block a user