feat(transport): counter for import-session cap rejections (register row 66)

Row 53 shipped the scadabridge.central.transport.import_sessions.open gauge
and a Warning log on a cap hit, but a log line is not something an operator
can alert on. Adds the companion counter

    scadabridge.central.transport.import_sessions.rejected

on the existing ZB.MOM.WW.ScadaBridge meter, so no ObservedMeters allowlist
change is needed (an unlisted meter exports nothing, silently).

The emit point is deliberately narrow. BundleSessionStore.Open tests the cap
TWICE — once on entry and again after EvictExpired() — and only the second
test is a rejection: a store full of merely expired sessions reclaims a slot
and admits the caller. Counting the first would make the series fire
routinely on any node whose sweep had not yet run, turning a page-worthy
signal into background noise. Re-Opening an already-open session id skips
the cap block entirely and is likewise not counted.

The two instruments are diagnostic as a pair: rejections rising while the
gauge oscillates below the cap is genuine concurrency (raise the cap);
rejections rising while the gauge sits pinned at the cap is a session leak
(register row 43's signature), where raising the cap only defers the problem.

Tests assert a delta over a scoped action rather than an absolute value, so
they do not inherit the register row 32 static-state flake that makes
QueueDepthGaugeTests fail only under the full concurrent suite. Negative
controls cover under-cap, expired-reclaim, and duplicate-session-id.

Tests: 11 passed in BundleSessionObservabilityTests (6 existing + 5 new).
This commit is contained in:
Joseph Doherty
2026-08-15 12:42:56 -04:00
parent ef876b38c5
commit 90ce099928
5 changed files with 175 additions and 1 deletions
+3
View File
@@ -52,6 +52,9 @@ Without this, an abandoned wizard or a plain CLI preview held one of the eight `
**Session-cap observability (register row 53).** That leak was only ever inferable from rotating test failures, because nothing counted open sessions and nothing logged a rejection. Two signals now close the gap: **Session-cap observability (register row 53).** That leak was only ever inferable from rotating test failures, because nothing counted open sessions and nothing logged a rejection. Two signals now close the gap:
- **Gauge `scadabridge.central.transport.import_sessions.open`** — open (unexpired) sessions on this node, on the `ZB.MOM.WW.ScadaBridge` meter, which is already in `SiteServiceRegistration.ObservedMeters` (the `ZbTelemetryOptions.Meters` allowlist; an unlisted meter exports nothing, silently). The provider is registered by `BundleSessionEvictionService.StartAsync` and identity-checked-cleared on `StopAsync`, mirroring the StoreAndForward queue-depth gauge. `IBundleSessionStore.OpenSessionCount` deliberately excludes expired-but-unswept entries — counting the raw dictionary would report the node as pinned at the cap for up to a sweep interval after a burst had actually drained, which is the exact false positive this instrument exists to rule out. Transport is registered only in the host's Central branch, so the gauge reports 0 on a site node. - **Gauge `scadabridge.central.transport.import_sessions.open`** — open (unexpired) sessions on this node, on the `ZB.MOM.WW.ScadaBridge` meter, which is already in `SiteServiceRegistration.ObservedMeters` (the `ZbTelemetryOptions.Meters` allowlist; an unlisted meter exports nothing, silently). The provider is registered by `BundleSessionEvictionService.StartAsync` and identity-checked-cleared on `StopAsync`, mirroring the StoreAndForward queue-depth gauge. `IBundleSessionStore.OpenSessionCount` deliberately excludes expired-but-unswept entries — counting the raw dictionary would report the node as pinned at the cap for up to a sweep interval after a burst had actually drained, which is the exact false positive this instrument exists to rule out. Transport is registered only in the host's Central branch, so the gauge reports 0 on a site node.
- **Counter `scadabridge.central.transport.import_sessions.rejected`** (register row 66) — bundle-import sessions turned away because the cap was exhausted, on the same meter (so likewise no allowlist change). Row 53 shipped the gauge and a Warning log, but a log line cannot be alerted on; this is the series to alert on. It is incremented **only at the post-eviction recheck** in `BundleSessionStore.Open`: the cap is tested twice, and a store full of merely *expired* sessions reclaims and admits the caller, so counting the first test would fire the signal routinely on any node whose sweep has not yet run. Re-`Open`ing an already-open session id skips the cap block entirely and is likewise never counted.
Read the two instruments **together** — the pair is what discriminates the failure modes. Rejections climbing while the gauge oscillates *below* the cap is genuine concurrency: raise `MaxConcurrentImportSessions`. Rejections climbing while the gauge sits *pinned at* the cap is a session leak (the register row 43 signature); raising the cap only defers it, so find the leak instead.
- **Warning log on rejection** — `BundleSessionStore.Open` logs the open count, the cap, and the TTL when it rejects. It fires only on a real cap hit: a store full of *expired* sessions is reclaimed first and logs nothing. A count that stays pinned at the cap across repeated warnings is a leak; an occasional warning is genuine concurrency, and the fix there is to raise `ScadaBridge:Transport:MaxConcurrentImportSessions` (each slot pins up to `MaxBundleSizeMb` of decrypted plaintext, so the cap is a memory budget). - **Warning log on rejection** — `BundleSessionStore.Open` logs the open count, the cap, and the TTL when it rejects. It fires only on a real cap hit: a store full of *expired* sessions is reclaimed first and logs nothing. A count that stays pinned at the cap across repeated warnings is a leak; an occasional warning is genuine concurrency, and the fix there is to raise `ScadaBridge:Transport:MaxConcurrentImportSessions` (each slot pins up to `MaxBundleSizeMb` of decrypted plaintext, so the cap is a memory budget).
### Conflict resolution and `BundleImportId` correlation ### Conflict resolution and `BundleImportId` correlation
+1 -1
View File
@@ -134,7 +134,7 @@ flowchart TD
class T muted class T muted
``` ```
The component is central-hosted. It is registered in `ZB.MOM.WW.ScadaBridge.Host` for central roles only, never for site roles — even when a bundle carries site-scoped config (T18), Transport only writes the central configuration database, never the site clusters. All persistence flows through existing audited repository interfaces in `ZB.MOM.WW.ScadaBridge.ConfigurationDatabase` — the component does not call `DbContext.SaveChangesAsync` directly. `BundleSessionStore` is in-process on the active central node (matching Blazor Server circuit affinity): 30-minute TTL, eviction on expiry, 3-strike passphrase lockout per session, and a `MaxConcurrentImportSessions` cap (default 8; arch-review 05 Task 24 — startup-validated as `> 0` by `TransportOptionsValidator`, since a zero/negative cap would reject every import session forever) bounding the decrypted-content footprint since each open session pins a fully-decrypted bundle in memory. **The cap is observable (register row 53):** open (unexpired) sessions are published as the `scadabridge.central.transport.import_sessions.open` gauge on the `ZB.MOM.WW.ScadaBridge` meter — already in the `ObservedMeters` allowlist, so no telemetry config change is needed — and a rejection at the cap logs at Warning with the current count, the cap and the TTL. Reclaiming a store full of *expired* sessions is not a cap hit and logs nothing. The component is central-hosted. It is registered in `ZB.MOM.WW.ScadaBridge.Host` for central roles only, never for site roles — even when a bundle carries site-scoped config (T18), Transport only writes the central configuration database, never the site clusters. All persistence flows through existing audited repository interfaces in `ZB.MOM.WW.ScadaBridge.ConfigurationDatabase` — the component does not call `DbContext.SaveChangesAsync` directly. `BundleSessionStore` is in-process on the active central node (matching Blazor Server circuit affinity): 30-minute TTL, eviction on expiry, 3-strike passphrase lockout per session, and a `MaxConcurrentImportSessions` cap (default 8; arch-review 05 Task 24 — startup-validated as `> 0` by `TransportOptionsValidator`, since a zero/negative cap would reject every import session forever) bounding the decrypted-content footprint since each open session pins a fully-decrypted bundle in memory. **The cap is observable (register row 53):** open (unexpired) sessions are published as the `scadabridge.central.transport.import_sessions.open` gauge on the `ZB.MOM.WW.ScadaBridge` meter — already in the `ObservedMeters` allowlist, so no telemetry config change is needed — and a rejection at the cap logs at Warning with the current count, the cap and the TTL. Reclaiming a store full of *expired* sessions is not a cap hit and logs nothing. **A log line is not alertable, so register row 66 adds the counter `scadabridge.central.transport.import_sessions.rejected`** on the same meter, incremented at the same single genuine-rejection point (the post-eviction recheck — never on the reclaim-then-succeed path, and never when re-opening an already-open session id). Alert on the counter; interpret it against the gauge — rejections rising while the gauge stays below the cap means genuine concurrency (raise the cap), whereas rejections rising while the gauge is pinned at the cap means a leaked session (raising the cap only defers it).
**Session release is explicit, not TTL-only (register row 43).** `ApplyAsync` releases the session it applied (zero-then-remove, T-007), and `IBundleImporter.Abandon(sessionId)` does the same for a session that will never be applied. Every abandonment path calls it: the Central UI wizard on new upload / Back-to-Upload / `ResetSessionState` / lockout reset / `Dispose` (navigation away or circuit teardown), the CLI `transport preview` handler (a preview never applies), and the CLI `transport import` handler for the aborts that precede apply (blocker rows, unresolved mappings). `Abandon` is idempotent — double release, an unknown id, and an already-expired id are all silent no-ops — and since session ids are random GUIDs handed only to their opener, abandoning the id you hold can never evict another operator's in-progress import. Relying on the TTL alone was an availability *and* secret-lifetime defect: eight abandoned wizards exhausted `MaxConcurrentImportSessions` and locked out every import on the active node for 30 minutes, each still holding its decrypted bundle content. **Session release is explicit, not TTL-only (register row 43).** `ApplyAsync` releases the session it applied (zero-then-remove, T-007), and `IBundleImporter.Abandon(sessionId)` does the same for a session that will never be applied. Every abandonment path calls it: the Central UI wizard on new upload / Back-to-Upload / `ResetSessionState` / lockout reset / `Dispose` (navigation away or circuit teardown), the CLI `transport preview` handler (a preview never applies), and the CLI `transport import` handler for the aborts that precede apply (blocker rows, unresolved mappings). `Abandon` is idempotent — double release, an unknown id, and an already-expired id are all silent no-ops — and since session ids are random GUIDs handed only to their opener, abandoning the id you hold can never evict another operator's in-progress import. Relying on the TTL alone was an availability *and* secret-lifetime defect: eight abandoned wizards exhausted `MaxConcurrentImportSessions` and locked out every import on the active node for 30 minutes, each still holding its decrypted bundle content.
@@ -93,6 +93,23 @@ public static class ScadaBridgeTelemetry
Meter.CreateCounter<long>("scadabridge.site.stream.events_dropped", unit: "1", Meter.CreateCounter<long>("scadabridge.site.stream.events_dropped", unit: "1",
description: "Events evicted from a site gRPC stream's bounded send channel, tagged by stream kind."); description: "Events evicted from a site gRPC stream's bounded send channel, tagged by stream kind.");
/// <summary>
/// Register row 66: incremented each time <c>BundleSessionStore.Open</c> refuses a
/// bundle-import session because <c>TransportOptions.MaxConcurrentImportSessions</c>
/// (default 8) is genuinely exhausted — i.e. the store was still at the cap AFTER
/// expired sessions were reclaimed. Reclaim-then-succeed and re-opening an already
/// open session id are deliberately NOT counted: neither turns an operator away.
///
/// Row 53 shipped the companion open-sessions gauge and a Warning log, but a log line
/// is not alertable. The pair is what distinguishes the two failure modes: rejections
/// climbing while the gauge oscillates below the cap is genuine concurrency (raise the
/// cap), whereas rejections climbing while the gauge sits pinned at the cap is a
/// session leak like register row 43's (find the leak — raising the cap only defers it).
/// </summary>
private static readonly Counter<long> _transportImportSessionsRejected =
Meter.CreateCounter<long>("scadabridge.central.transport.import_sessions.rejected", unit: "1",
description: "Bundle-import sessions rejected because the concurrent-import-session cap was exhausted.");
// ---------------- Histograms ---------------- // ---------------- Histograms ----------------
/// <summary> /// <summary>
@@ -186,6 +203,14 @@ public static class ScadaBridgeTelemetry
/// <summary>Records one resync whose ack window expired.</summary> /// <summary>Records one resync whose ack window expired.</summary>
public static void RecordSfResyncAckMissing() => _sfResyncAckMissing.Add(1); public static void RecordSfResyncAckMissing() => _sfResyncAckMissing.Add(1);
/// <summary>
/// Records one bundle-import session rejected on a genuine concurrent-session cap hit
/// (register row 66). Call ONLY from the post-eviction recheck in
/// <c>BundleSessionStore.Open</c> — counting the pre-eviction check would inflate the
/// series with reclaim-then-succeed cases that never turned an operator away.
/// </summary>
public static void RecordImportSessionRejected() => _transportImportSessionsRejected.Add(1);
/// <summary>Records that a site connection opened (increments the up-count gauge).</summary> /// <summary>Records that a site connection opened (increments the up-count gauge).</summary>
public static void SiteConnectionOpened() => Interlocked.Increment(ref _siteConnectionsUp); public static void SiteConnectionOpened() => Interlocked.Increment(ref _siteConnectionsUp);
@@ -3,6 +3,7 @@ using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions; using Microsoft.Extensions.Logging.Abstractions;
using Microsoft.Extensions.Options; using Microsoft.Extensions.Options;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Transport; using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Transport;
using ZB.MOM.WW.ScadaBridge.Commons.Observability;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Transport; using ZB.MOM.WW.ScadaBridge.Commons.Types.Transport;
namespace ZB.MOM.WW.ScadaBridge.Transport.Import; namespace ZB.MOM.WW.ScadaBridge.Transport.Import;
@@ -115,6 +116,15 @@ public sealed class BundleSessionStore : IBundleSessionStore
EvictExpired(); EvictExpired();
if (_sessions.Count >= cap) if (_sessions.Count >= cap)
{ {
// Register row 66: the alertable half of row 53's observability. The
// Warning below carries the diagnosis but a log line cannot be alerted
// on; this counter can. Deliberately incremented HERE, after
// EvictExpired() has already failed to free a slot — the outer check
// at the top of this block is not a rejection (a store full of stale
// entries reclaims and succeeds), so counting there would inflate the
// series with imports that were never actually turned away.
ScadaBridgeTelemetry.RecordImportSessionRejected();
// Register row 53: this rejection used to be silent apart from the // Register row 53: this rejection used to be silent apart from the
// exception the caller saw. A cap hit is either genuine concurrency // exception the caller saw. A cap hit is either genuine concurrency
// (many operators importing at once) or a session leak like register // (many operators importing at once) or a session leak like register
@@ -19,6 +19,7 @@ namespace ZB.MOM.WW.ScadaBridge.Transport.Tests.Import;
public sealed class BundleSessionObservabilityTests public sealed class BundleSessionObservabilityTests
{ {
private const string OpenSessionsGauge = "scadabridge.central.transport.import_sessions.open"; private const string OpenSessionsGauge = "scadabridge.central.transport.import_sessions.open";
private const string RejectedSessionsCounter = "scadabridge.central.transport.import_sessions.rejected";
// ------------------------------------------------------------------ // ------------------------------------------------------------------
// Warning log on cap rejection // Warning log on cap rejection
@@ -83,6 +84,108 @@ public sealed class BundleSessionObservabilityTests
Assert.Empty(logger.Entries); Assert.Empty(logger.Entries);
} }
// ------------------------------------------------------------------
// Rejection counter (register row 66)
// ------------------------------------------------------------------
[Fact]
public void Open_AtCap_IncrementsTheRejectionCounter()
{
// Row 53 shipped the gauge + Warning; a log line is not alertable. This counter is
// the series an operator can page on, so it must move on a genuine cap hit.
var clock = new TestClock(DateTimeOffset.UtcNow);
var opts = Options.Create(new TransportOptions { MaxConcurrentImportSessions = 3 });
var sut = new BundleSessionStore(opts, clock, NullLogger<BundleSessionStore>.Instance);
var future = clock.GetUtcNow().AddMinutes(30);
for (var i = 0; i < 3; i++)
{
sut.Open(SessionExpiringAt(future));
}
var delta = ReadRejectionCounter(() =>
Assert.Throws<InvalidOperationException>(() => sut.Open(SessionExpiringAt(future))));
Assert.Equal(1, delta);
}
[Fact]
public void Open_RepeatedCapHits_AccumulateOnTheRejectionCounter()
{
// A counter that saturated at 1 would hide exactly the case it exists for: a
// sustained rejection rate is what distinguishes real contention from a one-off.
var clock = new TestClock(DateTimeOffset.UtcNow);
var opts = Options.Create(new TransportOptions { MaxConcurrentImportSessions = 1 });
var sut = new BundleSessionStore(opts, clock, NullLogger<BundleSessionStore>.Instance);
var future = clock.GetUtcNow().AddMinutes(30);
sut.Open(SessionExpiringAt(future));
var delta = ReadRejectionCounter(() =>
{
for (var i = 0; i < 3; i++)
{
Assert.Throws<InvalidOperationException>(() => sut.Open(SessionExpiringAt(future)));
}
});
Assert.Equal(3, delta);
}
[Fact]
public void Open_UnderCap_DoesNotIncrementTheRejectionCounter()
{
// Negative control: the counter is a rejection signal, not an open-session signal.
var clock = new TestClock(DateTimeOffset.UtcNow);
var opts = Options.Create(new TransportOptions { MaxConcurrentImportSessions = 3 });
var sut = new BundleSessionStore(opts, clock, NullLogger<BundleSessionStore>.Instance);
var delta = ReadRejectionCounter(() =>
sut.Open(SessionExpiringAt(clock.GetUtcNow().AddMinutes(30))));
Assert.Equal(0, delta);
}
[Fact]
public void Open_AtCapOfExpiredSessions_ReclaimsWithoutIncrementingTheRejectionCounter()
{
// The load-bearing negative control. The cap check runs TWICE — once before
// EvictExpired and once after — and only the post-eviction check is a rejection.
// Counting the first would make the series fire routinely on any node whose sweep
// has not yet run, turning a page-worthy signal into background noise.
var clock = new TestClock(DateTimeOffset.UtcNow);
var opts = Options.Create(new TransportOptions { MaxConcurrentImportSessions = 2 });
var sut = new BundleSessionStore(opts, clock, NullLogger<BundleSessionStore>.Instance);
sut.Open(SessionExpiringAt(clock.GetUtcNow().AddMinutes(5)));
sut.Open(SessionExpiringAt(clock.GetUtcNow().AddMinutes(5)));
clock.Advance(TimeSpan.FromMinutes(10));
var delta = ReadRejectionCounter(() =>
sut.Open(SessionExpiringAt(clock.GetUtcNow().AddMinutes(30))));
Assert.Equal(0, delta);
}
[Fact]
public void Open_ReopeningAnExistingSessionIdAtCap_DoesNotIncrementTheRejectionCounter()
{
// Re-Opening an id already in the store skips the cap block entirely and overwrites
// in place. It consumes no new slot and turns nobody away, so it must not count.
var clock = new TestClock(DateTimeOffset.UtcNow);
var opts = Options.Create(new TransportOptions { MaxConcurrentImportSessions = 2 });
var sut = new BundleSessionStore(opts, clock, NullLogger<BundleSessionStore>.Instance);
var future = clock.GetUtcNow().AddMinutes(30);
var first = SessionExpiringAt(future);
sut.Open(first);
sut.Open(SessionExpiringAt(future));
// Store is now exactly at the cap of 2.
var delta = ReadRejectionCounter(() => sut.Open(first));
Assert.Equal(0, delta);
}
// ------------------------------------------------------------------ // ------------------------------------------------------------------
// OpenSessionCount / gauge // OpenSessionCount / gauge
// ------------------------------------------------------------------ // ------------------------------------------------------------------
@@ -197,6 +300,39 @@ public sealed class BundleSessionObservabilityTests
return observed; return observed;
} }
/// <summary>
/// Runs <paramref name="act"/> with a listener attached to the import-session rejection
/// counter and returns the number of rejections recorded DURING that action.
/// </summary>
/// <remarks>
/// Deliberately a delta over a scoped action rather than an absolute read. The counter
/// is a process-global static, so an absolute assertion would inherit the register
/// row 32 flake that makes <c>QueueDepthGaugeTests</c> fail only under the full
/// concurrent suite. A counter measured this way is immune: the listener exists only
/// for the duration of <paramref name="act"/>, and every increment is driven
/// synchronously by this test's own <c>Open</c> call on its own store instance, so a
/// concurrently-running test class cannot contribute a measurement to this window.
/// </remarks>
/// <param name="act">The action to measure.</param>
/// <returns>Rejections counted while <paramref name="act"/> ran.</returns>
private static long ReadRejectionCounter(Action act)
{
long delta = 0;
using var listener = new MeterListener();
listener.InstrumentPublished = (instrument, l) =>
{
if (instrument.Meter.Name == ScadaBridgeTelemetry.MeterName
&& instrument.Name == RejectedSessionsCounter)
{
l.EnableMeasurementEvents(instrument);
}
};
listener.SetMeasurementEventCallback<long>((_, measurement, _, _) => delta += measurement);
listener.Start();
act();
return delta;
}
private static BundleSession SessionExpiringAt(DateTimeOffset expiresAt) => new() private static BundleSession SessionExpiringAt(DateTimeOffset expiresAt) => new()
{ {
SessionId = Guid.NewGuid(), SessionId = Guid.NewGuid(),