diff --git a/docs/components/Transport.md b/docs/components/Transport.md index 9c39eb26..6104ad73 100644 --- a/docs/components/Transport.md +++ b/docs/components/Transport.md @@ -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: - **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). ### Conflict resolution and `BundleImportId` correlation diff --git a/docs/requirements/Component-Transport.md b/docs/requirements/Component-Transport.md index de37bd1b..f8bb5f8b 100644 --- a/docs/requirements/Component-Transport.md +++ b/docs/requirements/Component-Transport.md @@ -134,7 +134,7 @@ flowchart TD 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. diff --git a/src/ZB.MOM.WW.ScadaBridge.Commons/Observability/ScadaBridgeTelemetry.cs b/src/ZB.MOM.WW.ScadaBridge.Commons/Observability/ScadaBridgeTelemetry.cs index b2431fe4..34d3bd1d 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Commons/Observability/ScadaBridgeTelemetry.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Commons/Observability/ScadaBridgeTelemetry.cs @@ -93,6 +93,23 @@ public static class ScadaBridgeTelemetry Meter.CreateCounter("scadabridge.site.stream.events_dropped", unit: "1", description: "Events evicted from a site gRPC stream's bounded send channel, tagged by stream kind."); + /// + /// Register row 66: incremented each time BundleSessionStore.Open refuses a + /// bundle-import session because TransportOptions.MaxConcurrentImportSessions + /// (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). + /// + private static readonly Counter _transportImportSessionsRejected = + Meter.CreateCounter("scadabridge.central.transport.import_sessions.rejected", unit: "1", + description: "Bundle-import sessions rejected because the concurrent-import-session cap was exhausted."); + // ---------------- Histograms ---------------- /// @@ -186,6 +203,14 @@ public static class ScadaBridgeTelemetry /// Records one resync whose ack window expired. public static void RecordSfResyncAckMissing() => _sfResyncAckMissing.Add(1); + /// + /// Records one bundle-import session rejected on a genuine concurrent-session cap hit + /// (register row 66). Call ONLY from the post-eviction recheck in + /// BundleSessionStore.Open — counting the pre-eviction check would inflate the + /// series with reclaim-then-succeed cases that never turned an operator away. + /// + public static void RecordImportSessionRejected() => _transportImportSessionsRejected.Add(1); + /// Records that a site connection opened (increments the up-count gauge). public static void SiteConnectionOpened() => Interlocked.Increment(ref _siteConnectionsUp); diff --git a/src/ZB.MOM.WW.ScadaBridge.Transport/Import/BundleSessionStore.cs b/src/ZB.MOM.WW.ScadaBridge.Transport/Import/BundleSessionStore.cs index ccac6231..fd713a69 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Transport/Import/BundleSessionStore.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Transport/Import/BundleSessionStore.cs @@ -3,6 +3,7 @@ using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging.Abstractions; using Microsoft.Extensions.Options; using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Transport; +using ZB.MOM.WW.ScadaBridge.Commons.Observability; using ZB.MOM.WW.ScadaBridge.Commons.Types.Transport; namespace ZB.MOM.WW.ScadaBridge.Transport.Import; @@ -115,6 +116,15 @@ public sealed class BundleSessionStore : IBundleSessionStore EvictExpired(); 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 // exception the caller saw. A cap hit is either genuine concurrency // (many operators importing at once) or a session leak like register diff --git a/tests/ZB.MOM.WW.ScadaBridge.Transport.Tests/Import/BundleSessionObservabilityTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Transport.Tests/Import/BundleSessionObservabilityTests.cs index ca9f0939..ea6d17e2 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Transport.Tests/Import/BundleSessionObservabilityTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Transport.Tests/Import/BundleSessionObservabilityTests.cs @@ -19,6 +19,7 @@ namespace ZB.MOM.WW.ScadaBridge.Transport.Tests.Import; public sealed class BundleSessionObservabilityTests { 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 @@ -83,6 +84,108 @@ public sealed class BundleSessionObservabilityTests 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.Instance); + var future = clock.GetUtcNow().AddMinutes(30); + + for (var i = 0; i < 3; i++) + { + sut.Open(SessionExpiringAt(future)); + } + + var delta = ReadRejectionCounter(() => + Assert.Throws(() => 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.Instance); + var future = clock.GetUtcNow().AddMinutes(30); + sut.Open(SessionExpiringAt(future)); + + var delta = ReadRejectionCounter(() => + { + for (var i = 0; i < 3; i++) + { + Assert.Throws(() => 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.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.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.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 // ------------------------------------------------------------------ @@ -197,6 +300,39 @@ public sealed class BundleSessionObservabilityTests return observed; } + /// + /// Runs with a listener attached to the import-session rejection + /// counter and returns the number of rejections recorded DURING that action. + /// + /// + /// 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 QueueDepthGaugeTests fail only under the full + /// concurrent suite. A counter measured this way is immune: the listener exists only + /// for the duration of , and every increment is driven + /// synchronously by this test's own Open call on its own store instance, so a + /// concurrently-running test class cannot contribute a measurement to this window. + /// + /// The action to measure. + /// Rejections counted while ran. + 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((_, measurement, _, _) => delta += measurement); + listener.Start(); + act(); + return delta; + } + private static BundleSession SessionExpiringAt(DateTimeOffset expiresAt) => new() { SessionId = Guid.NewGuid(),