From 372ea4df1b0af2acfaccfdf7f4d8c61c0590e918 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Sat, 15 Aug 2026 11:42:31 -0400 Subject: [PATCH] feat(transport): observability for the import-session cap (register row 53) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Nothing counted open import sessions and nothing logged a rejection at MaxConcurrentImportSessions (default 8), so the register-row-43 session leak could only be inferred from rotating test failures. - Gauge scadabridge.central.transport.import_sessions.open on the existing ZB.MOM.WW.ScadaBridge meter — already in SiteServiceRegistration.ObservedMeters (the ZbTelemetryOptions.Meters allowlist), so no telemetry config change is needed and the series is not silently dropped. Provider-backed, mirroring the StoreAndForward queue-depth gauge: registered by BundleSessionEvictionService .StartAsync (Transport's only lifecycle seam) and identity-checked-cleared on StopAsync so a late stop cannot clobber a newer registrant. - IBundleSessionStore.OpenSessionCount counts only UNEXPIRED sessions. The raw dictionary count would report the node as pinned at the cap for up to a one-minute sweep interval after a burst had drained — the exact false positive the instrument exists to rule out. - BundleSessionStore.Open logs Warning with open count, cap and TTL when it rejects. Fires only on a real cap hit: a store full of expired sessions is reclaimed first and logs nothing. - Docs: docs/components/Transport.md + docs/requirements/Component-Transport.md. Tests: Transport 173 passed (+6 new — Warning content, two negative controls, unexpired-only counting, gauge register/clear, no-clobber-on-late-stop). --- docs/components/Transport.md | 5 + docs/requirements/Component-Transport.md | 2 +- .../Transport/IBundleSessionStore.cs | 9 + .../Observability/ScadaBridgeTelemetry.cs | 60 +++++ .../Import/BundleSessionEvictionService.cs | 39 +++ .../Import/BundleSessionStore.cs | 46 +++- .../Import/BundleSessionObservabilityTests.cs | 248 ++++++++++++++++++ 7 files changed, 407 insertions(+), 2 deletions(-) create mode 100644 tests/ZB.MOM.WW.ScadaBridge.Transport.Tests/Import/BundleSessionObservabilityTests.cs diff --git a/docs/components/Transport.md b/docs/components/Transport.md index 65dc5d24..91cb2d19 100644 --- a/docs/components/Transport.md +++ b/docs/components/Transport.md @@ -49,6 +49,11 @@ After `LoadAsync` validates and decrypts a bundle, the plaintext content is stor Without this, an abandoned wizard or a plain CLI preview held one of the eight `MaxConcurrentImportSessions` slots — and the decrypted bundle content — for the full TTL; eight of them locked out every import on the active node (deferred register row 43, observed live). +**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. +- **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 `PreviewAsync` compares each bundle artifact against the target environment using `ArtifactDiff`, classifying items as `Identical`, `Modified`, `New`, or `Blocker`. The operator assigns a `ResolutionAction` (`Add`, `Overwrite`, `Skip`, `Rename`) per item. `ApplyAsync` honours those resolutions in a single EF transaction and threads a new `BundleImportId` GUID through every per-entity audit row via the scoped `IAuditCorrelationContext`. This makes every configuration row written by a bundle import queryable as a group from the audit log. diff --git a/docs/requirements/Component-Transport.md b/docs/requirements/Component-Transport.md index 8dba53b0..8da888d1 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 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. **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/Interfaces/Transport/IBundleSessionStore.cs b/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Transport/IBundleSessionStore.cs index cdc493ec..efcb5c46 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Transport/IBundleSessionStore.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Transport/IBundleSessionStore.cs @@ -18,6 +18,15 @@ public interface IBundleSessionStore /// Removes all sessions whose expiry has passed. void EvictExpired(); + /// + /// Register row 53: the number of sessions currently held that have not yet expired. + /// Backs the scadabridge.central.transport.import_sessions.open gauge, so it + /// must be cheap and side-effect-free (it is read from the metrics collection thread) + /// and must NOT count expired-but-unswept entries — otherwise the gauge would read + /// as "at the cap" for up to a sweep interval after a burst drained. + /// + int OpenSessionCount { get; } + /// /// T-003: returns the current unlock-failure count for a bundle keyed by its /// content hash. The counter is server-owned so a second tab / CLI caller diff --git a/src/ZB.MOM.WW.ScadaBridge.Commons/Observability/ScadaBridgeTelemetry.cs b/src/ZB.MOM.WW.ScadaBridge.Commons/Observability/ScadaBridgeTelemetry.cs index 0ffbf176..b2431fe4 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Commons/Observability/ScadaBridgeTelemetry.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Commons/Observability/ScadaBridgeTelemetry.cs @@ -118,6 +118,14 @@ public static class ScadaBridgeTelemetry /// Provider that yields the live StoreAndForward queue depth; set by a later task. private static Func? _queueDepthProvider; + /// + /// Register row 53: provider that yields the number of currently open (unexpired) + /// Transport bundle-import sessions, registered by the Transport component's + /// session-eviction hosted service. Null on a site node (Transport is central-only), + /// where the gauge simply reports 0. + /// + private static Func? _importSessionCountProvider; + #pragma warning disable IDE0052 // Held to keep the observable gauges alive for the meter's lifetime. /// Gauge reporting the number of currently open site connections. private static readonly ObservableGauge _siteConnectionUp = @@ -138,6 +146,20 @@ public static class ScadaBridgeTelemetry () => Interlocked.Read(ref _liveAlarmAggregatorsActive), unit: "1", description: "Number of per-site live-alarm aggregators currently running on the active central node."); + + /// + /// Register row 53: gauge reporting the number of open Transport bundle-import + /// sessions on this central node. Each open session pins a fully decrypted bundle + /// in memory and consumes one of TransportOptions.MaxConcurrentImportSessions + /// (default 8), so a series that plateaus at the cap is the signature of leaked + /// sessions — the failure mode register row 43 could only be inferred from rotating + /// test failures because nothing counted them. + /// + private static readonly ObservableGauge _transportImportSessionsOpen = + Meter.CreateObservableGauge("scadabridge.central.transport.import_sessions.open", + () => Volatile.Read(ref _importSessionCountProvider) is { } p ? p() : 0L, + unit: "1", + description: "Number of currently open bundle-import sessions on this central node."); #pragma warning restore IDE0052 // ---------------- Emit helpers ---------------- @@ -242,4 +264,42 @@ public static class ScadaBridgeTelemetry // delegate, so a stale stop cannot clobber a successor's provider. Interlocked.CompareExchange(ref _queueDepthProvider, null, provider); } + + /// + /// Register row 53: registers the provider the open-import-sessions gauge reads on + /// each observation. Called by the Transport component's session-eviction hosted + /// service on start. A null provider is ignored so the gauge falls back to 0. + /// The callback is invoked synchronously on the metrics collection thread, so it + /// must stay cheap — the Transport implementation reads a + /// it + /// already owns. + /// + /// A callback returning the current open-session count. + public static void SetImportSessionCountProvider(Func provider) + { + if (provider is null) + { + return; + } + + Volatile.Write(ref _importSessionCountProvider, provider); + } + + /// + /// Clears the open-import-sessions provider, but only if the currently registered + /// provider is the exact delegate passed in + /// (reference-equal compare-and-clear), so a late stop of a superseded instance + /// cannot stomp a newer registrant. Mirrors . + /// + /// The provider delegate to remove; ignored unless it is the + /// one currently registered. + public static void ClearImportSessionCountProvider(Func provider) + { + if (provider is null) + { + return; + } + + Interlocked.CompareExchange(ref _importSessionCountProvider, null, provider); + } } diff --git a/src/ZB.MOM.WW.ScadaBridge.Transport/Import/BundleSessionEvictionService.cs b/src/ZB.MOM.WW.ScadaBridge.Transport/Import/BundleSessionEvictionService.cs index a0c36115..cae75a32 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Transport/Import/BundleSessionEvictionService.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Transport/Import/BundleSessionEvictionService.cs @@ -1,6 +1,7 @@ using Microsoft.Extensions.Hosting; using Microsoft.Extensions.Logging; using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Transport; +using ZB.MOM.WW.ScadaBridge.Commons.Observability; namespace ZB.MOM.WW.ScadaBridge.Transport.Import; @@ -21,6 +22,15 @@ internal sealed class BundleSessionEvictionService : BackgroundService private readonly IBundleSessionStore _sessionStore; private readonly ILogger _logger; + /// + /// Register row 53: the open-session gauge callback, held in a field so + /// can pass the SAME delegate reference to + /// — the clear is + /// an identity-checked compare-and-clear, so a stale stop cannot unregister a newer + /// instance's provider. Mirrors StoreAndForwardService's queue-depth registration. + /// + private readonly Func _openSessionCountProvider; + /// /// Initializes a new instance of . /// @@ -32,6 +42,35 @@ internal sealed class BundleSessionEvictionService : BackgroundService { _sessionStore = sessionStore ?? throw new ArgumentNullException(nameof(sessionStore)); _logger = logger ?? throw new ArgumentNullException(nameof(logger)); + _openSessionCountProvider = () => _sessionStore.OpenSessionCount; + } + + /// + /// Register row 53: registers the open-import-sessions gauge provider before the + /// sweep loop starts. This service is the Transport component's only lifecycle seam + /// (the session store itself is a plain singleton), and it is registered by + /// AddTransport, which only the Central branch of the host calls — so the + /// gauge reports 0 on a site node, where there are no import sessions to count. + /// + /// Host startup cancellation token. + /// A task that completes when the base has started. + public override Task StartAsync(CancellationToken cancellationToken) + { + ScadaBridgeTelemetry.SetImportSessionCountProvider(_openSessionCountProvider); + return base.StartAsync(cancellationToken); + } + + /// + /// Deregisters this instance's gauge provider on graceful stop so the gauge falls + /// back to 0 rather than reading through a dead store. The clear is identity-checked, + /// so a late stop cannot clobber a successor that already re-registered. + /// + /// Host shutdown cancellation token. + /// A task that completes when the base has stopped. + public override async Task StopAsync(CancellationToken cancellationToken) + { + ScadaBridgeTelemetry.ClearImportSessionCountProvider(_openSessionCountProvider); + await base.StopAsync(cancellationToken).ConfigureAwait(false); } /// diff --git a/src/ZB.MOM.WW.ScadaBridge.Transport/Import/BundleSessionStore.cs b/src/ZB.MOM.WW.ScadaBridge.Transport/Import/BundleSessionStore.cs index 1d52735b..ccac6231 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Transport/Import/BundleSessionStore.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Transport/Import/BundleSessionStore.cs @@ -1,4 +1,6 @@ using System.Collections.Concurrent; +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.Types.Transport; @@ -48,6 +50,7 @@ public sealed class BundleSessionStore : IBundleSessionStore private readonly TimeProvider _timeProvider; private readonly IOptions _options; + private readonly ILogger _logger; /// T-003: per-bundle unlock-failure entry with expiry. private sealed class UnlockFailureRecord @@ -61,12 +64,42 @@ public sealed class BundleSessionStore : IBundleSessionStore /// /// Transport options. is also used as the TTL for the T-003 per-bundle unlock-failure tracker. /// Time provider used to evaluate session expiry. - public BundleSessionStore(IOptions options, TimeProvider timeProvider) + /// Optional logger. Register row 53: a rejection at is logged at Warning, so a session leak is diagnosable from the node log instead of being inferred from rotating test failures. + public BundleSessionStore( + IOptions options, + TimeProvider timeProvider, + ILogger? logger = null) { ArgumentNullException.ThrowIfNull(options); _options = options; _ = options.Value; _timeProvider = timeProvider ?? throw new ArgumentNullException(nameof(timeProvider)); + _logger = logger ?? NullLogger.Instance; + } + + /// + public int OpenSessionCount + { + get + { + // Register row 53: count only unexpired entries. _sessions.Count would + // include expired-but-unswept sessions (the eviction sweep runs once a + // minute), so the gauge would read "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. Read-only: the gauge callback runs + // on the metrics collection thread and must not mutate the store. + var now = _timeProvider.GetUtcNow(); + var count = 0; + foreach (var kv in _sessions) + { + if (kv.Value.ExpiresAt > now) + { + count++; + } + } + + return count; + } } /// @@ -82,6 +115,17 @@ public sealed class BundleSessionStore : IBundleSessionStore EvictExpired(); if (_sessions.Count >= cap) { + // 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 + // row 43's — the count-vs-cap pair plus the recurrence rate in the + // log is what distinguishes them. + _logger.LogWarning( + "Bundle import session rejected: {OpenSessions} of {Cap} concurrent import sessions are open " + + "(ScadaBridge:Transport:MaxConcurrentImportSessions). Sessions release on apply, cancel or " + + "the {TtlMinutes}-minute TTL; a count that stays pinned at the cap indicates leaked sessions.", + _sessions.Count, cap, _options.Value.BundleSessionTtlMinutes); + throw new InvalidOperationException( $"Too many concurrent import sessions ({cap}). Complete or cancel an in-progress import before starting another."); } diff --git a/tests/ZB.MOM.WW.ScadaBridge.Transport.Tests/Import/BundleSessionObservabilityTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Transport.Tests/Import/BundleSessionObservabilityTests.cs new file mode 100644 index 00000000..ca9f0939 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.Transport.Tests/Import/BundleSessionObservabilityTests.cs @@ -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; + +/// +/// Register row 53: MaxConcurrentImportSessions 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 +/// scadabridge.central.transport.import_sessions.open gauge and the Warning-level +/// log emitted when rejects at the cap. +/// +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(() => 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.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.Instance); + var service = new BundleSessionEvictionService(store, NullLogger.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.Instance); + var newStore = new BundleSessionStore(Options.Create(new TransportOptions()), clock, NullLogger.Instance); + var oldService = new BundleSessionEvictionService(oldStore, NullLogger.Instance); + var newService = new BundleSessionEvictionService(newStore, NullLogger.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 + // ------------------------------------------------------------------ + + /// + /// 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. + /// + /// The gauge's current value, or 0 if it never reported. + 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((_, measurement, _, _) => observed = measurement); + listener.Start(); + listener.RecordObservableInstruments(); + return observed; + } + + private static BundleSession SessionExpiringAt(DateTimeOffset expiresAt) => new() + { + SessionId = Guid.NewGuid(), + Manifest = StubManifest(), + DecryptedContent = Array.Empty(), + 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()); + + /// Minimal mutable-clock for deterministic expiry. + 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); + } + + /// Captures formatted log entries so the cap-rejection Warning can be asserted. + private sealed class CapturingLogger : ILogger + { + public List<(LogLevel Level, string Message)> Entries { get; } = []; + + public IDisposable? BeginScope(TState state) where TState : notnull => null; + + public bool IsEnabled(LogLevel logLevel) => true; + + public void Log( + LogLevel logLevel, + EventId eventId, + TState state, + Exception? exception, + Func formatter) + { + Entries.Add((logLevel, formatter(state, exception))); + } + } +}