diff --git a/src/ZB.MOM.WW.ScadaBridge.CLI/ManagementHttpClient.cs b/src/ZB.MOM.WW.ScadaBridge.CLI/ManagementHttpClient.cs index e1b62e62..18304a41 100644 --- a/src/ZB.MOM.WW.ScadaBridge.CLI/ManagementHttpClient.cs +++ b/src/ZB.MOM.WW.ScadaBridge.CLI/ManagementHttpClient.cs @@ -9,13 +9,53 @@ public class ManagementHttpClient : IDisposable private readonly HttpClient _httpClient; /// - /// Initializes a new instance of the class. + /// WP2.6e (arch-review misc — CLI HttpClient timeout): default overall + /// for the shared client construction (30 s). This + /// bounds a hung/black-holed connection — before this, the public constructor left + /// at its framework default (100 s), silently longer + /// than most CLI callers' own per-request TimeSpan timeout argument + /// (// + /// already bound each call via their own , but a + /// connection attempt that never completes at all — no response headers, ever — is + /// bounded by instead, since that governs the whole + /// request/response including connect). Config-overridable via the + /// SCADABRIDGE_HTTP_TIMEOUT_SECONDS environment variable, consistent with how + /// every other CLI setting is overridden (see ) — kept + /// self-contained here (no /command-file plumbing) since CLI + /// commands are owned by a separate work package this phase. + /// + public static readonly TimeSpan DefaultTimeout = TimeSpan.FromSeconds(30); + + /// Test seam (WP2.6e) — the effective this instance was constructed with. + internal TimeSpan EffectiveTimeout { get; } + + /// + /// Resolves the effective default timeout: the SCADABRIDGE_HTTP_TIMEOUT_SECONDS + /// environment variable when set to a positive integer, otherwise . + /// + private static TimeSpan ResolveDefaultTimeout() + { + var env = Environment.GetEnvironmentVariable("SCADABRIDGE_HTTP_TIMEOUT_SECONDS"); + if (!string.IsNullOrWhiteSpace(env) + && int.TryParse(env, out var seconds) + && seconds > 0) + { + return TimeSpan.FromSeconds(seconds); + } + + return DefaultTimeout; + } + + /// + /// Initializes a new instance of the class, with + /// set to + /// (30 s, or the SCADABRIDGE_HTTP_TIMEOUT_SECONDS override). /// /// The base URL for the management API. /// The username for HTTP Basic authentication. /// The password for HTTP Basic authentication. public ManagementHttpClient(string baseUrl, string username, string password) - : this(new HttpClient(), baseUrl, username, password) + : this(new HttpClient { Timeout = ResolveDefaultTimeout() }, baseUrl, username, password) { } @@ -31,6 +71,9 @@ public class ManagementHttpClient : IDisposable internal ManagementHttpClient(HttpClient httpClient, string baseUrl, string username, string password) { _httpClient = httpClient; + // Test seam (WP2.6e): exposes the constructed HttpClient's effective Timeout + // without requiring reflection. + EffectiveTimeout = httpClient.Timeout; _httpClient.BaseAddress = new Uri(baseUrl.TrimEnd('/') + "/"); var credentials = Convert.ToBase64String(Encoding.UTF8.GetBytes($"{username}:{password}")); _httpClient.DefaultRequestHeaders.Authorization = diff --git a/src/ZB.MOM.WW.ScadaBridge.Commons/Messages/Health/SiteHealthReport.cs b/src/ZB.MOM.WW.ScadaBridge.Commons/Messages/Health/SiteHealthReport.cs index 8306ffdc..78d77ea0 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Commons/Messages/Health/SiteHealthReport.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Commons/Messages/Health/SiteHealthReport.cs @@ -104,6 +104,17 @@ public record SiteHealthReport( /// pair as perfectly healthy. Collapsing null to 0 anywhere on this path is a bug. /// public long? LocalDbOplogBacklog { get; init; } + + /// + /// WP2.6d (arch-review misc — site stream alarm-vs-attribute shared buffer): + /// cumulative count of alarm state changes dropped from + /// SiteStreamManager's dedicated alarm publish queue because it was at + /// capacity — i.e. the site-wide alarm hand-off itself fell behind, not a + /// per-subscriber buffer. Point-in-time (not reset on collect), refreshed by + /// the site-side SiteStreamAlarmDropReporter hosted service. Defaults to + /// 0 so existing producers/tests that don't populate it stay valid. + /// + public long SiteStreamAlarmDropCount { get; init; } } /// diff --git a/src/ZB.MOM.WW.ScadaBridge.HealthMonitoring/ISiteHealthCollector.cs b/src/ZB.MOM.WW.ScadaBridge.HealthMonitoring/ISiteHealthCollector.cs index 8781553a..a1a19a4d 100644 --- a/src/ZB.MOM.WW.ScadaBridge.HealthMonitoring/ISiteHealthCollector.cs +++ b/src/ZB.MOM.WW.ScadaBridge.HealthMonitoring/ISiteHealthCollector.cs @@ -178,6 +178,22 @@ public interface ISiteHealthCollector // Default no-op so test fakes do not need to be updated. } + /// + /// WP2.6d (arch-review misc — site stream alarm-vs-attribute shared buffer): + /// replace the latest cumulative count of alarm state changes dropped from + /// SiteStreamManager's dedicated alarm publish queue, used by the next + /// call. Refreshed periodically by the site-side + /// SiteStreamAlarmDropReporter hosted service. Point-in-time: NOT reset on + /// . Default interface implementation is a no-op so + /// existing test fakes continue to compile without per-fake updates. + /// + /// The cumulative dropped-alarm count from SiteStreamManager.AlarmPublishDroppedCount. + void SetSiteStreamAlarmDropCount(long count) + { + // Default no-op so test fakes do not need to be updated. The real + // SiteHealthCollector overrides this with the Interlocked.Exchange store. + } + /// /// Sets the hostname of this node. /// diff --git a/src/ZB.MOM.WW.ScadaBridge.HealthMonitoring/SiteHealthCollector.cs b/src/ZB.MOM.WW.ScadaBridge.HealthMonitoring/SiteHealthCollector.cs index be008182..505a4bd1 100644 --- a/src/ZB.MOM.WW.ScadaBridge.HealthMonitoring/SiteHealthCollector.cs +++ b/src/ZB.MOM.WW.ScadaBridge.HealthMonitoring/SiteHealthCollector.cs @@ -39,6 +39,10 @@ public class SiteHealthCollector : ISiteHealthCollector private int _scriptQueueDepth; private int _scriptBusyThreads; private long _scriptOldestBusyAgeBits = BitConverter.DoubleToInt64Bits(double.NaN); + // WP2.6d: cumulative alarm-publish-queue drop count, refreshed by + // SiteStreamAlarmDropReporter. Point-in-time (not reset on collect), like the + // scheduler gauges above. + private long _siteStreamAlarmDropCount; private volatile string _nodeHostname = ""; private volatile IReadOnlyList? _clusterNodes; private volatile bool _isActiveNode; @@ -173,6 +177,12 @@ public class SiteHealthCollector : ISiteHealthCollector return double.IsNaN(value) ? null : value; } + /// + public void SetSiteStreamAlarmDropCount(long count) + { + Interlocked.Exchange(ref _siteStreamAlarmDropCount, count); + } + /// public void SetNodeHostname(string hostname) => _nodeHostname = hostname; @@ -277,7 +287,8 @@ public class SiteHealthCollector : ISiteHealthCollector // not run) leaves both report fields null — "no data", not "disconnected // with an empty backlog". LocalDbReplicationConnected = localDbReplication?.Item1, - LocalDbOplogBacklog = localDbReplication?.Item2 + LocalDbOplogBacklog = localDbReplication?.Item2, + SiteStreamAlarmDropCount = Interlocked.Read(ref _siteStreamAlarmDropCount) }; } } diff --git a/src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs b/src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs index bc9fdc2e..66d6996d 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs @@ -836,12 +836,18 @@ akka {{ // writes (static overrides, native_alarm_state) could be cut off on // graceful failover. Names unchanged: deployment-manager-singleton / // deployment-manager-proxy. + // WP2.6a: the shared cache backing SiteExternalSystemRepository reads + // (registered singleton by AddSiteRuntime); DeploymentManagerActor invalidates + // it after an artifact deploy applies external-system changes. + var externalSystemCache = _serviceProvider + .GetService(); var dm = SingletonRegistrar.Start( _actorSystem!, "deployment-manager", Props.Create(() => new DeploymentManagerActor( storage, compilationService, sharedScriptLibrary, streamManager, siteRuntimeOptionsValue, dmLogger, dclManager, - siteHealthCollector, _serviceProvider, null, deploymentConfigFetcher)), + siteHealthCollector, _serviceProvider, null, deploymentConfigFetcher, + null, null, externalSystemCache)), _logger, role: siteRole); var dmProxy = dm.Proxy; diff --git a/src/ZB.MOM.WW.ScadaBridge.InboundAPI/ApiMethodCache.cs b/src/ZB.MOM.WW.ScadaBridge.InboundAPI/ApiMethodCache.cs new file mode 100644 index 00000000..6322b3d0 --- /dev/null +++ b/src/ZB.MOM.WW.ScadaBridge.InboundAPI/ApiMethodCache.cs @@ -0,0 +1,85 @@ +using System.Collections.Concurrent; +using ZB.MOM.WW.ScadaBridge.Commons.Entities.InboundApi; + +namespace ZB.MOM.WW.ScadaBridge.InboundAPI; + +/// +/// WP2.6b (arch-review misc — Inbound API per-request SQL): a short-TTL cache of +/// resolved ApiMethod rows, sitting in front of IInboundApiRepository.GetMethodByNameAsync +/// so a repeated POST /api/{methodName} burst for the same method does not each pay a +/// database round-trip. +/// +/// Two invalidation paths, by design: +/// +/// +/// Fast path. calls +/// for every name in an ApiMethod +/// ScriptArtifactsChanged notification (bundle-import changes today). +/// Fallback. Every entry — including a negative (not-found) result — +/// expires after regardless of whether an +/// invalidation notification ever arrives. This is what keeps an operator edit via the +/// Management API/UI (which does not publish to the change bus) correct within a bounded +/// window, and is also the correctness backstop for the bus's at-least-once/in-process-only +/// contract — a missed or duplicate notification is harmless. +/// +/// +/// A negative (method not found) result is cached too — bounded by the same short TTL — so a +/// caller with a provisioned-but-nonexistent method scope (a real misconfiguration case, not an +/// enumeration probe: EndpointExtensions only reaches this cache after the in-memory +/// scope check already passed) does not repeatedly pay a miss query. +/// +/// +public sealed class ApiMethodCache +{ + private sealed record Entry(ApiMethod? Method, DateTimeOffset ExpiresAtUtc); + + private readonly ConcurrentDictionary _entries = new(StringComparer.Ordinal); + private readonly TimeSpan _ttl; + private readonly TimeProvider _timeProvider; + + /// Initializes the cache. + /// How long a resolved (or not-found) entry stays valid. + /// Optional time source; defaults to . + public ApiMethodCache(TimeSpan ttl, TimeProvider? timeProvider = null) + { + if (ttl <= TimeSpan.Zero) + throw new ArgumentOutOfRangeException(nameof(ttl), ttl, "ApiMethodCache TTL must be positive."); + + _ttl = ttl; + _timeProvider = timeProvider ?? TimeProvider.System; + } + + /// + /// Returns the cached entry for if it has not expired; + /// otherwise invokes , caches the result (found or not-found), and + /// returns it. Concurrent callers on a cache miss for the SAME name may each invoke + /// once (last write wins) — an acceptable, rare race in exchange + /// for not serializing every miss behind a lock; the point of the cache is to absorb + /// repeated hits, not to fully dedupe a first-request thundering herd. + /// + /// The method name to resolve. + /// Loads the current row from the repository on a cache miss. + /// Cancellation token for the fetch. + public async Task GetOrFetchAsync( + string methodName, + Func> fetch, + CancellationToken cancellationToken) + { + var now = _timeProvider.GetUtcNow(); + if (_entries.TryGetValue(methodName, out var cached) && cached.ExpiresAtUtc > now) + { + return cached.Method; + } + + var method = await fetch(cancellationToken).ConfigureAwait(false); + _entries[methodName] = new Entry(method, now + _ttl); + return method; + } + + /// Drops the cached entry for one method name. Safe to call for an unknown name. + /// The method name to invalidate. + public void Invalidate(string methodName) => _entries.TryRemove(methodName, out _); + + /// Diagnostic helper — the number of currently-cached entries (including expired-but-not-evicted ones). + internal int Count => _entries.Count; +} diff --git a/src/ZB.MOM.WW.ScadaBridge.InboundAPI/EndpointExtensions.cs b/src/ZB.MOM.WW.ScadaBridge.InboundAPI/EndpointExtensions.cs index 0f753ef5..abb6e0cd 100644 --- a/src/ZB.MOM.WW.ScadaBridge.InboundAPI/EndpointExtensions.cs +++ b/src/ZB.MOM.WW.ScadaBridge.InboundAPI/EndpointExtensions.cs @@ -91,6 +91,7 @@ public static class EndpointExtensions var logger = httpContext.RequestServices.GetRequiredService>(); var verifier = httpContext.RequestServices.GetRequiredService(); var repository = httpContext.RequestServices.GetRequiredService(); + var methodCache = httpContext.RequestServices.GetRequiredService(); var executor = httpContext.RequestServices.GetRequiredService(); var routeHelper = httpContext.RequestServices.GetRequiredService(); var options = httpContext.RequestServices.GetRequiredService>().Value; @@ -150,9 +151,17 @@ public static class EndpointExtensions // "Echo" does not grant "echo". This is the intended invariant: method // names are case-sensitive identifiers, and a key's granted scopes must be // provisioned with the exact casing of the methods they authorize. + // WP2.6b: the per-request repository round-trip is fronted by a short-TTL + // ApiMethodCache (invalidated by name on an ApiMethod ScriptArtifactsChanged + // notification, and self-healing via TTL expiry otherwise — see ApiMethodCache + // and ScriptArtifactChangeSubscriber). Only reached after the in-scope check + // passes, preserving the existing enumeration-safety ordering below. var inScope = identity.Scopes.Contains(methodName); var method = inScope - ? await repository.GetMethodByNameAsync(methodName, httpContext.RequestAborted) + ? await methodCache.GetOrFetchAsync( + methodName, + ct => repository.GetMethodByNameAsync(methodName, ct), + httpContext.RequestAborted) : null; if (method == null || !inScope) diff --git a/src/ZB.MOM.WW.ScadaBridge.InboundAPI/InboundApiOptions.cs b/src/ZB.MOM.WW.ScadaBridge.InboundAPI/InboundApiOptions.cs index d9ce3ddf..cd172a61 100644 --- a/src/ZB.MOM.WW.ScadaBridge.InboundAPI/InboundApiOptions.cs +++ b/src/ZB.MOM.WW.ScadaBridge.InboundAPI/InboundApiOptions.cs @@ -34,4 +34,18 @@ public class InboundApiOptions /// /// public string ApiKeyPepper { get; set; } = string.Empty; + + /// + /// WP2.6b (arch-review misc — Inbound API per-request SQL): how long a resolved + /// ApiMethod row (found OR not-found) is served from + /// before the next request re-fetches it from the repository. The + /// ScriptArtifactChangeSubscriber/IScriptArtifactChangeBus pipeline + /// invalidates a specific method's entry immediately on a bundle-import change (the fast + /// path); this TTL is the self-healing fallback for changes the bus does not cover (e.g. + /// an operator edit via the Management API/UI, which does not publish a bus notification), + /// bounding staleness without requiring every mutation path to remember to invalidate. + /// Short by design — the whole point is to absorb repeated-request-storm SQL, not to + /// serve minutes-old method definitions. + /// + public TimeSpan ApiMethodCacheTtl { get; set; } = TimeSpan.FromSeconds(5); } diff --git a/src/ZB.MOM.WW.ScadaBridge.InboundAPI/InboundApiOptionsValidator.cs b/src/ZB.MOM.WW.ScadaBridge.InboundAPI/InboundApiOptionsValidator.cs index 23bc7b0b..574dcd37 100644 --- a/src/ZB.MOM.WW.ScadaBridge.InboundAPI/InboundApiOptionsValidator.cs +++ b/src/ZB.MOM.WW.ScadaBridge.InboundAPI/InboundApiOptionsValidator.cs @@ -27,5 +27,9 @@ public sealed class InboundApiOptionsValidator : OptionsValidatorBase 0, $"ScadaBridge:InboundApi:MaxRequestBodyBytes must be greater than zero " + $"(was {options.MaxRequestBodyBytes}); it caps the accepted request body before buffering."); + + builder.RequireThat(options.ApiMethodCacheTtl > TimeSpan.Zero, + $"ScadaBridge:InboundApi:ApiMethodCacheTtl must be a positive duration " + + $"(was {options.ApiMethodCacheTtl}); it bounds how long a resolved ApiMethod row is cached."); } } diff --git a/src/ZB.MOM.WW.ScadaBridge.InboundAPI/ScriptArtifactChangeSubscriber.cs b/src/ZB.MOM.WW.ScadaBridge.InboundAPI/ScriptArtifactChangeSubscriber.cs index dff2f3cf..8d8bd5c1 100644 --- a/src/ZB.MOM.WW.ScadaBridge.InboundAPI/ScriptArtifactChangeSubscriber.cs +++ b/src/ZB.MOM.WW.ScadaBridge.InboundAPI/ScriptArtifactChangeSubscriber.cs @@ -27,20 +27,27 @@ namespace ZB.MOM.WW.ScadaBridge.InboundAPI; public sealed class ScriptArtifactChangeSubscriber : IHostedService { private readonly InboundScriptExecutor _executor; + private readonly ApiMethodCache _methodCache; private readonly ILogger _logger; private readonly IScriptArtifactChangeBus? _bus; private IDisposable? _subscription; /// Initializes the subscriber. /// The compiled-handler cache to invalidate. + /// + /// WP2.6b: the short-TTL resolved-ApiMethod-row cache to invalidate alongside + /// the compiled handler. + /// /// Logger instance. /// The change bus, or null when the host registers none (site roles, tests). public ScriptArtifactChangeSubscriber( InboundScriptExecutor executor, + ApiMethodCache methodCache, ILogger logger, IScriptArtifactChangeBus? bus = null) { _executor = executor; + _methodCache = methodCache; _logger = logger; _bus = bus; } @@ -78,6 +85,7 @@ public sealed class ScriptArtifactChangeSubscriber : IHostedService foreach (var name in notification.Names) { _executor.InvalidateMethod(name); + _methodCache.Invalidate(name); } _logger.LogInformation( diff --git a/src/ZB.MOM.WW.ScadaBridge.InboundAPI/ServiceCollectionExtensions.cs b/src/ZB.MOM.WW.ScadaBridge.InboundAPI/ServiceCollectionExtensions.cs index bf694c50..57669485 100644 --- a/src/ZB.MOM.WW.ScadaBridge.InboundAPI/ServiceCollectionExtensions.cs +++ b/src/ZB.MOM.WW.ScadaBridge.InboundAPI/ServiceCollectionExtensions.cs @@ -1,4 +1,5 @@ using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Options; namespace ZB.MOM.WW.ScadaBridge.InboundAPI; @@ -19,6 +20,13 @@ public static class ServiceCollectionExtensions services.AddSingleton(); services.AddScoped(); + // WP2.6b: short-TTL cache of resolved ApiMethod rows, sitting in front of the + // per-request repository fetch in EndpointExtensions. Invalidated by name via + // ScriptArtifactChangeSubscriber (below) and self-heals via TTL expiry for + // changes the bus does not cover. + services.AddSingleton(sp => new ApiMethodCache( + sp.GetRequiredService>().Value.ApiMethodCacheTtl)); + // Routed calls go through the IInstanceRouter seam; the // production implementation delegates to CommunicationService. services.AddScoped(); diff --git a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/DeploymentManagerActor.cs b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/DeploymentManagerActor.cs index 338818c6..938de1fa 100644 --- a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/DeploymentManagerActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/DeploymentManagerActor.cs @@ -16,6 +16,7 @@ using ZB.MOM.WW.ScadaBridge.SiteEventLogging; using ZB.MOM.WW.ScadaBridge.SiteRuntime.Deployment; using ZB.MOM.WW.ScadaBridge.SiteRuntime.Messages; using ZB.MOM.WW.ScadaBridge.SiteRuntime.Persistence; +using ZB.MOM.WW.ScadaBridge.SiteRuntime.Repositories; using ZB.MOM.WW.ScadaBridge.SiteRuntime.Scripts; using ZB.MOM.WW.ScadaBridge.SiteRuntime.Streaming; @@ -40,6 +41,14 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers // DeployInstanceCommand is validated before any actor/persistence work. private readonly DeployCompileValidator _deployCompileValidator; private readonly SharedScriptLibrary _sharedScriptLibrary; + /// + /// WP2.6a: the shared cache backing SiteExternalSystemRepository reads. + /// Invalidated wholesale after applies + /// external-system changes so a script's next call re-reads fresh definitions/method + /// lists instead of serving a stale cached snapshot. Null in tests that do not wire + /// external-system caching — the invalidation call is then simply skipped. + /// + private readonly ExternalSystemDefinitionCache? _externalSystemCache; private readonly SiteStreamManager? _streamManager; private readonly SiteRuntimeOptions _options; private readonly ILogger _logger; @@ -157,6 +166,12 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers /// deployed configuration at startup; defaults to 5 seconds when null. /// Optional override for loading all deployed configurations /// at startup; defaults to reading from . Primarily for tests. + /// + /// WP2.6a: the shared backing + /// SiteExternalSystemRepository reads, invalidated after this actor applies + /// external-system changes. Optional/null in tests that do not exercise external-system + /// caching. + /// public DeploymentManagerActor( SiteStorageService storage, ScriptCompilationService compilationService, @@ -170,12 +185,14 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers ILoggerFactory? loggerFactory = null, IDeploymentConfigFetcher? configFetcher = null, TimeSpan? startupLoadRetryInterval = null, - Func>>? configLoader = null) + Func>>? configLoader = null, + ExternalSystemDefinitionCache? externalSystemCache = null) { _storage = storage; _compilationService = compilationService; _deployCompileValidator = new DeployCompileValidator(compilationService); _sharedScriptLibrary = sharedScriptLibrary; + _externalSystemCache = externalSystemCache; _streamManager = streamManager; _options = options; _dclManager = dclManager; @@ -1914,6 +1931,11 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers await _storage.DeleteExternalSystemsExceptAsync( command.ExternalSystems.Select(es => es.Name).ToList()); + + // WP2.6a: drop the shared read cache so the next script call sees + // the freshly-applied definitions/method lists instead of a stale + // snapshot loaded before this deploy. + _externalSystemCache?.InvalidateAll(); } // Store database connection definitions diff --git a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Repositories/ExternalSystemDefinitionCache.cs b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Repositories/ExternalSystemDefinitionCache.cs new file mode 100644 index 00000000..b4f6b512 --- /dev/null +++ b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Repositories/ExternalSystemDefinitionCache.cs @@ -0,0 +1,89 @@ +using ZB.MOM.WW.ScadaBridge.Commons.Entities.ExternalSystems; + +namespace ZB.MOM.WW.ScadaBridge.SiteRuntime.Repositories; + +/// +/// WP2.6a (arch-review misc — site external-system resolution): the shared, +/// redeploy-invalidated snapshot backing 's +/// reads. +/// +/// Registered as a DI singleton in production (AddSiteRuntime) so every +/// scoped instance — one per script +/// execution's DI scope — shares ONE snapshot instead of re-querying and +/// re-parsing the external_systems table (including its JSON +/// method_definitions column) on every call. The repository's default +/// constructor also accepts a private, unshared instance of this class so tests that +/// construct the repository directly keep seeing freshly-written rows without any +/// caching surprise. +/// +/// +/// Loaded lazily and wholesale on first read (one query covering every system's +/// definition AND its parsed method list, indexed by name and by +/// for O(1) by-ID lookups) and dropped wholesale by +/// — called by DeploymentManagerActor after an +/// artifact deploy applies new/changed/removed external systems, so the cache never +/// serves a stale method list or a deleted system past the next read. +/// +/// +public sealed class ExternalSystemDefinitionCache +{ + /// + /// One fully-loaded, immutable view of the site's external-system catalogue. + /// + /// System definitions keyed by name (ordinal). + /// Synthetic system ID → name, for O(1) by-ID resolution. + /// Parsed method list per system name. + /// Synthetic method ID → (system name, method name), for O(1) by-ID resolution. + internal sealed record Snapshot( + IReadOnlyDictionary ByName, + IReadOnlyDictionary IdToName, + IReadOnlyDictionary> MethodsByName, + IReadOnlyDictionary MethodIdIndex); + + // Guards concurrent cache-miss loads so a burst of first-callers after a redeploy + // collapses onto a single SQLite read rather than each racing its own full scan. + private readonly SemaphoreSlim _loadGate = new(1, 1); + + // Reference reads/writes of a class type are already atomic in .NET; volatile only + // adds the memory-visibility guarantee across threads, which is all the + // check-then-load double-check pattern below needs. + private volatile Snapshot? _snapshot; + + /// + /// Drops the cached snapshot. The next read reloads a fresh one from storage. + /// Safe to call at any time, including with no snapshot yet loaded. + /// + public void InvalidateAll() => _snapshot = null; + + /// + /// Returns the cached snapshot, loading it via on a + /// cache miss (first read, or the first read after ). + /// Concurrent misses collapse onto a single load. + /// + /// Loads a fresh snapshot from storage; invoked at most once per miss. + /// Cancellation token for the load. + internal async Task GetOrLoadAsync( + Func> loader, CancellationToken cancellationToken) + { + var snapshot = _snapshot; + if (snapshot != null) + return snapshot; + + await _loadGate.WaitAsync(cancellationToken).ConfigureAwait(false); + try + { + // Re-check: another caller may have already loaded while we waited. + snapshot = _snapshot; + if (snapshot != null) + return snapshot; + + snapshot = await loader(cancellationToken).ConfigureAwait(false); + _snapshot = snapshot; + return snapshot; + } + finally + { + _loadGate.Release(); + } + } +} diff --git a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Repositories/SiteExternalSystemRepository.cs b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Repositories/SiteExternalSystemRepository.cs index 7daba84e..e03801fb 100644 --- a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Repositories/SiteExternalSystemRepository.cs +++ b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Repositories/SiteExternalSystemRepository.cs @@ -11,27 +11,45 @@ namespace ZB.MOM.WW.ScadaBridge.SiteRuntime.Repositories; /// backed by the local SQLite database via . /// Write operations throw because site-local /// artifacts are managed exclusively through deployment from Central. +/// +/// WP2.6a (arch-review misc): every ExternalSystemDefinition/ExternalSystemMethod +/// read is served from an snapshot instead of +/// a fresh SQLite scan per call — the by-ID lookups in particular used to re-fetch every +/// row (and, for methods, re-parse every system's JSON method list) just to find one +/// match. See for the sharing/invalidation +/// contract; DeploymentManagerActor invalidates it after an artifact deploy +/// applies external-system changes. +/// /// public class SiteExternalSystemRepository : IExternalSystemRepository { private readonly SiteStorageService _storage; + private readonly ExternalSystemDefinitionCache _cache; /// - /// Shared options for the per-row MethodDefinitionDto JSON parse in - /// (perf remediation, arch-review WP1.5) — - /// avoids allocating a new per call. Settings - /// preserved exactly; the query shape itself is untouched (WP2.6). - /// - private static readonly JsonSerializerOptions MethodDefinitionJsonOptions = - new() { PropertyNameCaseInsensitive = true }; - - /// - /// Initializes a new site-side external system repository. + /// Initializes a new site-side external system repository with a private, + /// unshared cache — used by tests and any caller that wants each instance to see + /// its own fresh view with no cross-instance sharing. /// /// Storage service providing database access. public SiteExternalSystemRepository(SiteStorageService storage) + : this(storage, new ExternalSystemDefinitionCache()) + { + } + + /// + /// Initializes a new site-side external system repository backed by a shared + /// — the production DI registration + /// (AddSiteRuntime) injects one singleton cache shared by every scoped + /// repository instance, so repeated resolutions (e.g. one per script execution's + /// DI scope) do not each pay a fresh SQLite scan. + /// + /// Storage service providing database access. + /// The shared (or private) definition cache backing reads. + public SiteExternalSystemRepository(SiteStorageService storage, ExternalSystemDefinitionCache cache) { _storage = storage ?? throw new ArgumentNullException(nameof(storage)); + _cache = cache ?? throw new ArgumentNullException(nameof(cache)); } // ── ExternalSystemDefinition (read) ── @@ -40,55 +58,27 @@ public class SiteExternalSystemRepository : IExternalSystemRepository public async Task> GetAllExternalSystemsAsync( CancellationToken cancellationToken = default) { - // Already open — SiteStorageService.CreateConnection now hands out a - // LocalDb-managed connection. Opening it again would throw. - await using var connection = CreateConnection(); - - await using var command = connection.CreateCommand(); - command.CommandText = @" - SELECT name, endpoint_url, auth_type, auth_configuration, timeout_seconds - FROM external_systems"; - - var results = new List(); - await using var reader = await command.ExecuteReaderAsync(cancellationToken); - while (await reader.ReadAsync(cancellationToken)) - { - results.Add(MapExternalSystem(reader)); - } - - return results; + var snapshot = await _cache.GetOrLoadAsync(LoadSnapshotAsync, cancellationToken); + return snapshot.ByName.Values.ToList(); } /// public async Task GetExternalSystemByIdAsync( int id, CancellationToken cancellationToken = default) { - // The SQLite table is keyed by name, not by integer ID. - // Scan all rows and match on the synthetic ID derived from the name. - var all = await GetAllExternalSystemsAsync(cancellationToken); - return all.FirstOrDefault(e => e.Id == id); + var snapshot = await _cache.GetOrLoadAsync(LoadSnapshotAsync, cancellationToken); + return snapshot.IdToName.TryGetValue(id, out var name) + && snapshot.ByName.TryGetValue(name, out var definition) + ? definition + : null; } /// public async Task GetExternalSystemByNameAsync( string name, CancellationToken cancellationToken = default) { - // Already open — SiteStorageService.CreateConnection now hands out a - // LocalDb-managed connection. Opening it again would throw. - await using var connection = CreateConnection(); - - await using var command = connection.CreateCommand(); - command.CommandText = @" - SELECT name, endpoint_url, auth_type, auth_configuration, timeout_seconds - FROM external_systems - WHERE name = @name"; - command.Parameters.AddWithValue("@name", name); - - await using var reader = await command.ExecuteReaderAsync(cancellationToken); - if (!await reader.ReadAsync(cancellationToken)) - return null; - - return MapExternalSystem(reader); + var snapshot = await _cache.GetOrLoadAsync(LoadSnapshotAsync, cancellationToken); + return snapshot.ByName.TryGetValue(name, out var definition) ? definition : null; } // ── ExternalSystemMethod (read) ── @@ -97,27 +87,13 @@ public class SiteExternalSystemRepository : IExternalSystemRepository public async Task> GetMethodsByExternalSystemIdAsync( int externalSystemId, CancellationToken cancellationToken = default) { - // Find the parent system to get its name, then parse its method_definitions JSON. - var system = await GetExternalSystemByIdAsync(externalSystemId, cancellationToken); - if (system is null) + var snapshot = await _cache.GetOrLoadAsync(LoadSnapshotAsync, cancellationToken); + if (!snapshot.IdToName.TryGetValue(externalSystemId, out var name)) return Array.Empty(); - // Already open — SiteStorageService.CreateConnection now hands out a - // LocalDb-managed connection. Opening it again would throw. - await using var connection = CreateConnection(); - - await using var command = connection.CreateCommand(); - command.CommandText = @" - SELECT method_definitions - FROM external_systems - WHERE name = @name"; - command.Parameters.AddWithValue("@name", system.Name); - - var json = (string?)await command.ExecuteScalarAsync(cancellationToken); - if (string.IsNullOrWhiteSpace(json)) - return Array.Empty(); - - return ParseMethodDefinitions(json, externalSystemId); + return snapshot.MethodsByName.TryGetValue(name, out var methods) + ? methods + : Array.Empty(); } /// @@ -133,17 +109,13 @@ public class SiteExternalSystemRepository : IExternalSystemRepository public async Task GetExternalSystemMethodByIdAsync( int id, CancellationToken cancellationToken = default) { - // Scan all systems and their methods to find the matching synthetic ID. - var systems = await GetAllExternalSystemsAsync(cancellationToken); - foreach (var system in systems) - { - var methods = await GetMethodsByExternalSystemIdAsync(system.Id, cancellationToken); - var match = methods.FirstOrDefault(m => m.Id == id); - if (match is not null) - return match; - } + var snapshot = await _cache.GetOrLoadAsync(LoadSnapshotAsync, cancellationToken); + if (!snapshot.MethodIdIndex.TryGetValue(id, out var key)) + return null; + if (!snapshot.MethodsByName.TryGetValue(key.SystemName, out var methods)) + return null; - return null; + return methods.FirstOrDefault(m => m.Id == id); } // ── DatabaseConnectionDefinition (read) ── @@ -282,13 +254,63 @@ public class SiteExternalSystemRepository : IExternalSystemRepository }; } + /// + /// Loads a full, fresh snapshot of every external system and its parsed method + /// list in ONE query (name, endpoint config, and the method_definitions JSON + /// column together) — the cache-miss path behind every + /// read. Building both the by-name and + /// by-ID indexes here, in a single pass, is what turns the by-ID lookups into O(1) + /// dictionary gets instead of the prior full-table-then-LINQ-scan. + /// + private async Task LoadSnapshotAsync( + CancellationToken cancellationToken) + { + // Already open — SiteStorageService.CreateConnection now hands out a + // LocalDb-managed connection. Opening it again would throw. + await using var connection = CreateConnection(); + + await using var command = connection.CreateCommand(); + command.CommandText = @" + SELECT name, endpoint_url, auth_type, auth_configuration, timeout_seconds, method_definitions + FROM external_systems"; + + var byName = new Dictionary(StringComparer.Ordinal); + var idToName = new Dictionary(); + var methodsByName = new Dictionary>(StringComparer.Ordinal); + var methodIdIndex = new Dictionary(); + + await using var reader = await command.ExecuteReaderAsync(cancellationToken); + while (await reader.ReadAsync(cancellationToken)) + { + var definition = MapExternalSystem(reader); + byName[definition.Name] = definition; + idToName[definition.Id] = definition.Name; + + var methodsJson = reader.IsDBNull(5) ? null : reader.GetString(5); + var methods = string.IsNullOrWhiteSpace(methodsJson) + ? Array.Empty() + : ParseMethodDefinitions(methodsJson, definition.Id); + methodsByName[definition.Name] = methods; + foreach (var method in methods) + methodIdIndex[method.Id] = (definition.Name, method.Name); + } + + return new ExternalSystemDefinitionCache.Snapshot(byName, idToName, methodsByName, methodIdIndex); + } + + // Built ONCE per assembly (Phase 1, arch-review misc): JsonSerializerOptions + // construction is not free and every method-list parse used to mint a new + // instance — now shared across every LoadSnapshotAsync call. + private static readonly JsonSerializerOptions MethodDefinitionsJsonOptions = + new() { PropertyNameCaseInsensitive = true }; + private static IReadOnlyList ParseMethodDefinitions( string json, int externalSystemId) { try { var methods = JsonSerializer.Deserialize>(json, - MethodDefinitionJsonOptions); + MethodDefinitionsJsonOptions); if (methods is null) return Array.Empty(); diff --git a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/ServiceCollectionExtensions.cs b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/ServiceCollectionExtensions.cs index 6b018167..796a5e41 100644 --- a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/ServiceCollectionExtensions.cs +++ b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/ServiceCollectionExtensions.cs @@ -65,8 +65,16 @@ public static class ServiceCollectionExtensions // construction, which is what runs SiteLocalDbSetup.OnReady. services.AddSingleton(); - // Site-local repository implementations backed by SQLite - services.AddScoped(); + // Site-local repository implementations backed by SQLite. + // WP2.6a: ExternalSystemDefinitionCache is a singleton shared by every scoped + // SiteExternalSystemRepository instance (one per script-execution DI scope), so + // resolving the repository repeatedly does not each pay a fresh SQLite + // fetch-all — DeploymentManagerActor invalidates the shared cache after an + // artifact deploy applies external-system changes. + services.AddSingleton(); + services.AddScoped(sp => new SiteExternalSystemRepository( + sp.GetRequiredService(), + sp.GetRequiredService())); // Notify-and-fetch: typed HttpClient for fetching deployment configs from central. services.AddHttpClient() @@ -83,6 +91,13 @@ public static class ServiceCollectionExtensions sp.GetRequiredService>().Value, sp.GetRequiredService>())); + // WP2.6d: periodically lift SiteStreamManager's alarm-publish-queue drop count + // onto the site health report. + services.AddHostedService(sp => new Streaming.SiteStreamAlarmDropReporter( + sp.GetRequiredService(), + sp.GetRequiredService(), + sp.GetRequiredService>())); + return services; } diff --git a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/SiteRuntimeOptions.cs b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/SiteRuntimeOptions.cs index 42503ca5..ba2506f3 100644 --- a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/SiteRuntimeOptions.cs +++ b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/SiteRuntimeOptions.cs @@ -77,4 +77,17 @@ public class SiteRuntimeOptions /// bounded script-execution threads is gone. Default: 30000ms. /// public int StuckScriptGraceMs { get; set; } = 30000; + + /// + /// WP2.6d (arch-review misc — site stream alarm-vs-attribute shared buffer): + /// capacity of the bounded hand-off queue feeding 's + /// dedicated alarm publish path. Alarm state changes are staged through this + /// queue into their OWN broadcast source — separate from the (far higher-volume) + /// attribute publish source — so an attribute storm cannot evict a pending alarm + /// transition. DropOldest overflow; drops are counted (surfaced on the site health + /// report) and logged. Default: 2000 (alarm volume is inherently far lower than + /// attribute volume, so this rarely needs to be anywhere near + /// ). + /// + public int AlarmPublishQueueCapacity { get; set; } = 2000; } diff --git a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/SiteRuntimeOptionsValidator.cs b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/SiteRuntimeOptionsValidator.cs index f3ed4789..6515c648 100644 --- a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/SiteRuntimeOptionsValidator.cs +++ b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/SiteRuntimeOptionsValidator.cs @@ -61,5 +61,9 @@ public sealed class SiteRuntimeOptionsValidator : OptionsValidatorBase= 0, $"ScadaBridge:SiteRuntime:StuckScriptGraceMs must be >= 0 " + $"(was {options.StuckScriptGraceMs}); a negative grace throws inside the stuck-script watchdog's delay."); + + builder.RequireThat(options.AlarmPublishQueueCapacity > 0, + $"ScadaBridge:SiteRuntime:AlarmPublishQueueCapacity must be greater than 0 " + + $"(was {options.AlarmPublishQueueCapacity}); it bounds SiteStreamManager's dedicated alarm publish queue."); } } diff --git a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Streaming/SiteStreamAlarmDropReporter.cs b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Streaming/SiteStreamAlarmDropReporter.cs new file mode 100644 index 00000000..f3784c50 --- /dev/null +++ b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Streaming/SiteStreamAlarmDropReporter.cs @@ -0,0 +1,78 @@ +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using ZB.MOM.WW.ScadaBridge.HealthMonitoring; + +namespace ZB.MOM.WW.ScadaBridge.SiteRuntime.Streaming; + +/// +/// WP2.6d (arch-review misc — site stream alarm-vs-attribute shared buffer): site-side +/// hosted service that periodically reads +/// and pushes it into so +/// the next carries a fresh snapshot on +/// the site health report. Mirrors ScriptSchedulerStatsReporter: immediate first +/// probe, fixed cadence, exceptions logged and swallowed so the loop survives every probe +/// failure. +/// +public sealed class SiteStreamAlarmDropReporter : BackgroundService +{ + /// Default poll cadence (10 s) — coarse enough to amortise across health reports. + internal static readonly TimeSpan DefaultPollInterval = TimeSpan.FromSeconds(10); + + private readonly ISiteHealthCollector _collector; + private readonly SiteStreamManager _streamManager; + private readonly ILogger _logger; + private readonly TimeSpan _pollInterval; + + /// Initializes a new instance of . + /// The site health collector that receives the drop count. + /// The site stream manager whose alarm-queue drop count is sampled. + /// Logger instance. + /// Poll interval override; defaults to (10 s). + public SiteStreamAlarmDropReporter( + ISiteHealthCollector collector, + SiteStreamManager streamManager, + ILogger logger, + TimeSpan? pollInterval = null) + { + _collector = collector ?? throw new ArgumentNullException(nameof(collector)); + _streamManager = streamManager ?? throw new ArgumentNullException(nameof(streamManager)); + _logger = logger ?? throw new ArgumentNullException(nameof(logger)); + _pollInterval = pollInterval ?? DefaultPollInterval; + } + + /// + protected override async Task ExecuteAsync(CancellationToken stoppingToken) + { + // Immediate first probe so the first health report after start carries a + // real snapshot instead of a zero. + Probe(); + + while (!stoppingToken.IsCancellationRequested) + { + try + { + await Task.Delay(_pollInterval, stoppingToken).ConfigureAwait(false); + } + catch (OperationCanceledException) + { + break; + } + + Probe(); + } + } + + private void Probe() + { + try + { + _collector.SetSiteStreamAlarmDropCount(_streamManager.AlarmPublishDroppedCount); + } + catch (Exception ex) + { + // Catch-all is deliberate: the hosted service must survive every class + // of probe failure so the next tick gets a chance. + _logger.LogWarning(ex, "SiteStreamAlarmDropReporter probe failed; next tick will retry."); + } + } +} diff --git a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Streaming/SiteStreamManager.cs b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Streaming/SiteStreamManager.cs index a49fedb4..5a91247b 100644 --- a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Streaming/SiteStreamManager.cs +++ b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Streaming/SiteStreamManager.cs @@ -1,3 +1,4 @@ +using System.Threading.Channels; using Akka; using Akka.Actor; using Akka.Streams; @@ -18,6 +19,21 @@ namespace ZB.MOM.WW.ScadaBridge.SiteRuntime.Streaming; /// Implements ISiteStreamSubscriber so the gRPC server can subscribe actors /// to instance events without referencing SiteRuntime directly. /// +/// +/// WP2.6d (arch-review misc — site stream alarm-vs-attribute shared buffer): attribute +/// and alarm events used to share ONE upstream Source.ActorRef + BroadcastHub +/// pair, so a burst of attribute changes could fill that shared buffer and evict a +/// pending alarm transition before any subscriber ever saw it. Attribute and alarm +/// events now travel through entirely SEPARATE publish sources/hubs +/// (/ vs. +/// /), so an attribute storm +/// can only ever evict other attribute events. The alarm path additionally stages +/// through a bounded () so drops +/// are counted () — Source.ActorRef's +/// built-in DropHead overflow has no drop callback to instrument. Publishing is also +/// skipped entirely when nobody is subscribed to that source (see +/// /). +/// public class SiteStreamManager : ISiteStreamSubscriber { /// Sentinel instance name recorded for site-wide (non-instance-scoped) subscriptions. @@ -26,11 +42,44 @@ public class SiteStreamManager : ISiteStreamSubscriber private ActorSystem? _system; private IMaterializer? _materializer; private readonly int _bufferSize; + private readonly int _alarmPublishQueueCapacity; private readonly ILogger _logger; private readonly object _lock = new(); - private IActorRef? _sourceActor; - private Source? _hubSource; + // Attribute-only publish source/hub. By far the higher-volume of the two — + // deliberately kept on Source.ActorRef's built-in bounded-mailbox + DropHead + // behavior (unchanged from before WP2.6d); losing an occasional attribute update + // under a storm is the accepted, pre-existing trade-off. + private IActorRef? _attributeSourceActor; + private Source? _attributeHubSource; + + // Alarm-only publish source/hub — isolated from the attribute path above so an + // attribute storm can never evict a pending alarm transition. + private IActorRef? _alarmSourceActor; + private Source? _alarmHubSource; + + // Bounded hand-off queue feeding the alarm source actor. Staging alarm publishes + // through this (rather than Tell-ing _alarmSourceActor directly) is what makes + // AlarmPublishDroppedCount possible — Source.ActorRef's own DropHead overflow has + // no drop callback to observe. + private Channel? _alarmPublishQueue; + private Task? _alarmPump; + private long _alarmPublishDroppedCount; + + /// + /// Cumulative count of alarm state changes dropped because + /// was at capacity (WP2.6d) — i.e. the site-wide + /// alarm hand-off queue itself fell behind, not the downstream per-subscriber + /// buffers. Surfaced on the site health report. + /// + public long AlarmPublishDroppedCount => Interlocked.Read(ref _alarmPublishDroppedCount); + + // Per-source subscriber counts (WP2.6d "skip publish at zero subscribers"). + // A per-instance Subscribe touches BOTH sources (it needs both attribute and alarm + // events for that instance); SubscribeSiteAlarms touches only the alarm source. + private int _attributeSubscriberCount; + private int _alarmSubscriberCount; + private readonly Dictionary _subscriptions = new(); /// Initializes the stream manager with configuration and logger; the Akka stream is not started until is called. @@ -41,66 +90,138 @@ public class SiteStreamManager : ISiteStreamSubscriber ILogger logger) { _bufferSize = options.StreamBufferSize; + _alarmPublishQueueCapacity = options.AlarmPublishQueueCapacity; _logger = logger; } /// - /// Initializes the broadcast stream. Must be called after ActorSystem is ready. - /// The ActorSystem is passed here rather than via the constructor so that - /// SiteStreamManager can be created by DI before the actor system exists. + /// Initializes the broadcast streams (one for attribute events, one for alarm + /// events — see the type-level remarks) and starts the alarm hand-off pump. Must be + /// called after ActorSystem is ready. The ActorSystem is passed here rather than via + /// the constructor so that SiteStreamManager can be created by DI before the actor + /// system exists. /// - /// The running Akka used to materialize the broadcast stream. + /// The running Akka used to materialize the broadcast streams. public void Initialize(ActorSystem system) { _system = system; _materializer = _system.Materializer(); - var (sourceActor, hubSource) = Source.ActorRef( + var (attributeSourceActor, attributeHubSource) = Source.ActorRef( _bufferSize, OverflowStrategy.DropHead) .ToMaterialized( BroadcastHub.Sink(bufferSize: 256), Keep.Both) .Run(_materializer); + _attributeSourceActor = attributeSourceActor; + _attributeHubSource = attributeHubSource; - _sourceActor = sourceActor; - _hubSource = hubSource; + var (alarmSourceActor, alarmHubSource) = Source.ActorRef( + _bufferSize, + OverflowStrategy.DropHead) + .ToMaterialized( + BroadcastHub.Sink(bufferSize: 256), + Keep.Both) + .Run(_materializer); + _alarmSourceActor = alarmSourceActor; + _alarmHubSource = alarmHubSource; + + _alarmPublishQueue = CreateAlarmPublishQueue(_alarmPublishQueueCapacity, () => + { + var total = Interlocked.Increment(ref _alarmPublishDroppedCount); + _logger.LogWarning( + "Alarm publish queue exceeded its bounded capacity ({Capacity}); " + + "oldest pending alarm transition dropped (total dropped: {Dropped})", + _alarmPublishQueueCapacity, total); + }); + + var alarmQueue = _alarmPublishQueue; + var alarmActor = _alarmSourceActor; + _alarmPump = Task.Run(async () => + { + await foreach (var changed in alarmQueue.Reader.ReadAllAsync()) + { + alarmActor.Tell(changed); + } + }); _logger.LogInformation( - "SiteStreamManager initialized with publish buffer size {BufferSize}", _bufferSize); + "SiteStreamManager initialized with publish buffer size {BufferSize} " + + "and alarm publish queue capacity {AlarmQueueCapacity}", + _bufferSize, _alarmPublishQueueCapacity); } /// - /// Publishes an attribute value change to the broadcast hub. - /// Fire-and-forget — never blocks the calling actor. + /// Builds the bounded, single-reader alarm hand-off queue with DropOldest overflow, + /// invoking on every eviction. Extracted as a pure, + /// testable helper (internal — see InternalsVisibleTo) so the drop-counting + /// wiring can be verified deterministically, without racing the async pump/actor + /// system against a burst of publishes. + /// + internal static Channel CreateAlarmPublishQueue(int capacity, Action onDropped) => + Channel.CreateBounded( + new BoundedChannelOptions(Math.Max(1, capacity)) + { + SingleReader = true, + SingleWriter = false, + FullMode = BoundedChannelFullMode.DropOldest, + }, + itemDropped: _ => onDropped()); + + /// + /// Publishes an attribute value change to the attribute broadcast hub. + /// Fire-and-forget — never blocks the calling actor. Skipped entirely when no + /// subscriber is currently interested in attribute events (WP2.6d). /// /// The attribute value change event to publish. public void PublishAttributeValueChanged(AttributeValueChanged changed) { - _sourceActor?.Tell(changed); + if (Volatile.Read(ref _attributeSubscriberCount) == 0) + return; + + _attributeSourceActor?.Tell(changed); } /// - /// Publishes an alarm state change to the broadcast hub. - /// Fire-and-forget — never blocks the calling actor. + /// Publishes an alarm state change to the DEDICATED alarm broadcast hub — isolated + /// from the (far higher-volume) attribute path, so an attribute storm can never + /// evict a pending alarm transition. Fire-and-forget — never blocks the calling + /// actor. Skipped entirely when no subscriber is currently interested in alarm + /// events (WP2.6d). /// /// The alarm state change event to publish. public void PublishAlarmStateChanged(AlarmStateChanged changed) { - _sourceActor?.Tell(changed); + if (Volatile.Read(ref _alarmSubscriberCount) == 0) + return; + + _alarmPublishQueue?.Writer.TryWrite(changed); } /// public string Subscribe(string instanceName, IActorRef subscriber) { - if (_hubSource is null || _materializer is null) + if (_attributeHubSource is null || _alarmHubSource is null || _materializer is null) throw new InvalidOperationException("SiteStreamManager.Initialize must be called before Subscribe"); var subscriptionId = Guid.NewGuid().ToString(); var capturedInstance = instanceName; var capturedSubscriber = subscriber; - var killSwitch = _hubSource + // Two independent graphs — one per source — both forwarding to the same + // subscriber actor. The actor's own mailbox serializes delivery; this instance + // subscription (used by Debug View) does not need cross-source ordering + // guarantees, only that alarm events for it are never lost to an attribute + // storm on a DIFFERENT instance sharing the same (now attribute-only) hub. + var attributeKillSwitch = _attributeHubSource + .Where(ev => ev.InstanceUniqueName == capturedInstance) + .Buffer(_bufferSize, OverflowStrategy.DropHead) + .ViaMaterialized(KillSwitches.Single(), Keep.Right) + .To(Sink.ForEach(ev => capturedSubscriber.Tell(ev))) + .Run(_materializer); + + var alarmKillSwitch = _alarmHubSource .Where(ev => ev.InstanceUniqueName == capturedInstance) .Buffer(_bufferSize, OverflowStrategy.DropHead) .ViaMaterialized(KillSwitches.Single(), Keep.Right) @@ -110,9 +231,15 @@ public class SiteStreamManager : ISiteStreamSubscriber lock (_lock) { _subscriptions[subscriptionId] = new SubscriptionInfo( - instanceName, subscriber, killSwitch, DateTimeOffset.UtcNow); + instanceName, subscriber, + new[] { attributeKillSwitch, alarmKillSwitch }, + TouchesAttributes: true, TouchesAlarms: true, + DateTimeOffset.UtcNow); } + Interlocked.Increment(ref _attributeSubscriberCount); + Interlocked.Increment(ref _alarmSubscriberCount); + _logger.LogDebug( "Subscriber {SubscriptionId} registered for instance {Instance}", subscriptionId, instanceName); @@ -122,10 +249,10 @@ public class SiteStreamManager : ISiteStreamSubscriber /// /// Subscribe to ALARM events for ALL instances on the site (no per-instance - /// filter). Only events are forwarded; - /// events are dropped (attributes are far - /// higher-volume and the aggregated Alarm Summary never shows them). Same - /// broadcast-hub wiring as , and the returned + /// filter). The dedicated alarm hub carries only + /// events (the Where below is a defensive no-op, not a load-bearing filter — + /// see the type-level remarks); the aggregated Alarm Summary never sees attribute + /// events. Same broadcast-hub wiring as , and the returned /// subscription id is torn down via exactly like the /// per-instance variant. /// @@ -133,13 +260,13 @@ public class SiteStreamManager : ISiteStreamSubscriber /// A subscription id to pass to . public string SubscribeSiteAlarms(IActorRef subscriber) { - if (_hubSource is null || _materializer is null) + if (_alarmHubSource is null || _materializer is null) throw new InvalidOperationException("SiteStreamManager.Initialize must be called before SubscribeSiteAlarms"); var subscriptionId = Guid.NewGuid().ToString(); var capturedSubscriber = subscriber; - var killSwitch = _hubSource + var killSwitch = _alarmHubSource .Where(ev => ev is AlarmStateChanged) .Buffer(_bufferSize, OverflowStrategy.DropHead) .ViaMaterialized(KillSwitches.Single(), Keep.Right) @@ -149,9 +276,14 @@ public class SiteStreamManager : ISiteStreamSubscriber lock (_lock) { _subscriptions[subscriptionId] = new SubscriptionInfo( - SiteWideInstanceName, subscriber, killSwitch, DateTimeOffset.UtcNow); + SiteWideInstanceName, subscriber, + new[] { killSwitch }, + TouchesAttributes: false, TouchesAlarms: true, + DateTimeOffset.UtcNow); } + Interlocked.Increment(ref _alarmSubscriberCount); + _logger.LogDebug( "Subscriber {SubscriptionId} registered for site-wide alarm events", subscriptionId); @@ -160,7 +292,7 @@ public class SiteStreamManager : ISiteStreamSubscriber /// /// Unsubscribe from instance events. Shuts down the per-subscriber - /// stream graph via its KillSwitch. + /// stream graph(s) via their KillSwitch(es). /// /// The subscription ID returned by . /// true if the subscription was found and removed; false if it was already gone. @@ -173,7 +305,7 @@ public class SiteStreamManager : ISiteStreamSubscriber return false; } - info.KillSwitch.Shutdown(); + TearDown(info); _logger.LogDebug("Subscriber {SubscriptionId} removed", subscriptionId); return true; } @@ -193,7 +325,7 @@ public class SiteStreamManager : ISiteStreamSubscriber } foreach (var info in toShutdown) - info.KillSwitch.Shutdown(); + TearDown(info); if (toShutdown.Count > 0) { @@ -202,6 +334,18 @@ public class SiteStreamManager : ISiteStreamSubscriber } } + /// Shuts down every kill switch for a subscription and decrements the matching per-source counters. + private void TearDown(SubscriptionInfo info) + { + foreach (var killSwitch in info.KillSwitches) + killSwitch.Shutdown(); + + if (info.TouchesAttributes) + Interlocked.Decrement(ref _attributeSubscriberCount); + if (info.TouchesAlarms) + Interlocked.Decrement(ref _alarmSubscriberCount); + } + /// /// Returns the count of active subscriptions (for diagnostics/testing). /// @@ -213,6 +357,8 @@ public class SiteStreamManager : ISiteStreamSubscriber private record SubscriptionInfo( string InstanceName, IActorRef Subscriber, - IKillSwitch KillSwitch, + IReadOnlyList KillSwitches, + bool TouchesAttributes, + bool TouchesAlarms, DateTimeOffset SubscribedAt); } diff --git a/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardOptions.cs b/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardOptions.cs index 5425c10b..f91c8335 100644 --- a/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardOptions.cs +++ b/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardOptions.cs @@ -38,4 +38,18 @@ public class StoreAndForwardOptions /// 1 = legacy serial. Within a lane delivery stays sequential (per-target FIFO). /// public int SweepTargetParallelism { get; set; } = 4; + + /// + /// WP2.6c (arch-review misc — unbounded S&F observer queue): capacity of the + /// cached-call audit-observer pump's queue. It was the one unbounded + /// Channel<T> left in the system — a slow/stuck + /// ICachedCallLifecycleObserver (a SQLite audit write) could not stretch + /// the retry sweep, but nothing stopped it growing without bound while the sweep kept + /// posting. Bounded with + /// — the same overflow policy as its sibling bounded channels + /// (SiteEventLogger's write queue, the debug/alarm stream hubs) — so a stuck + /// observer sheds the oldest unprocessed notifications instead of leaking memory; drops + /// are counted (see StoreAndForwardService.ObserverQueueDroppedCount). + /// + public int ObserverQueueCapacity { get; set; } = 10_000; } diff --git a/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardOptionsValidator.cs b/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardOptionsValidator.cs index 41bef58b..a2da666f 100644 --- a/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardOptionsValidator.cs +++ b/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardOptionsValidator.cs @@ -41,5 +41,9 @@ public sealed class StoreAndForwardOptionsValidator : OptionsValidatorBase= 1, $"ScadaBridge:StoreAndForward:SweepTargetParallelism must be >= 1 " + $"(was {options.SweepTargetParallelism}); it caps concurrent (category,target) sweep lanes — 1 means serial."); + + builder.RequireThat(options.ObserverQueueCapacity >= 1, + $"ScadaBridge:StoreAndForward:ObserverQueueCapacity must be >= 1 " + + $"(was {options.ObserverQueueCapacity}); it bounds the cached-call audit-observer pump's queue."); } } diff --git a/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardService.cs b/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardService.cs index 53a6ad2f..e0143d4c 100644 --- a/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardService.cs +++ b/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardService.cs @@ -122,9 +122,52 @@ public class StoreAndForwardService /// completes it (a restarted instance needs a fresh /// channel). Before starts the pump, posts fall back /// to inline processing (see ). + /// + /// WP2.6c: bounded (, + /// default 10,000) with — this was + /// the one unbounded channel left in the system; a pump that falls behind (a stuck + /// observer) now sheds the oldest unprocessed notification instead of growing + /// without bound. Drops are counted via + /// and exposed by . + /// /// - private Channel> _observerQueue = - Channel.CreateUnbounded>(new UnboundedChannelOptions { SingleReader = true }); + private Channel> _observerQueue = CreateObserverQueue( + FieldInitializerObserverQueueCapacity, onDropped: null); + + /// + /// Capacity used only for the field initializer, before + /// re-creates the channel sized from + /// (unavailable at field-init + /// time — is assigned in the constructor body). Irrelevant in + /// practice: a post before falls back to inline processing + /// (see ), so nothing is ever queued at this size. + /// + private const int FieldInitializerObserverQueueCapacity = 1; + + /// + /// Cumulative count of cached-call audit-observer notifications dropped because + /// was at capacity (WP2.6c). Not reset across + /// / cycles — a diagnostic total for + /// the lifetime of this service instance. + /// + private long _observerQueueDroppedCount; + + /// Diagnostic counter — see . + public long ObserverQueueDroppedCount => Interlocked.Read(ref _observerQueueDroppedCount); + + /// + /// Builds a bounded, single-reader observer queue with DropOldest overflow, invoking + /// (if supplied) on every eviction. + /// + private static Channel> CreateObserverQueue(int capacity, Action? onDropped) => + Channel.CreateBounded>( + new BoundedChannelOptions(Math.Max(1, capacity)) + { + SingleReader = true, + SingleWriter = false, + FullMode = BoundedChannelFullMode.DropOldest, + }, + itemDropped: _ => onDropped?.Invoke()); /// /// The single-reader pump draining , or @@ -371,8 +414,17 @@ public class StoreAndForwardService // StopAsync completes the channel, so a restarted instance needs a fresh // one. The pump is best-effort: an observer that throws is logged and // swallowed so a failing audit pipeline never corrupts retry bookkeeping. - _observerQueue = Channel.CreateUnbounded>( - new UnboundedChannelOptions { SingleReader = true }); + // WP2.6c: bounded + DropOldest, sized from options; a drop increments + // _observerQueueDroppedCount (surfaced via ObserverQueueDroppedCount) and is + // logged at Warning so a stuck observer is visible, not just silently lossy. + _observerQueue = CreateObserverQueue(_options.ObserverQueueCapacity, onDropped: () => + { + Interlocked.Increment(ref _observerQueueDroppedCount); + _logger.LogWarning( + "Cached-call audit-observer queue exceeded its bounded capacity ({Capacity}); " + + "oldest pending notification dropped (total dropped: {Dropped})", + _options.ObserverQueueCapacity, Interlocked.Read(ref _observerQueueDroppedCount)); + }); _observerPump = Task.Run(async () => { await foreach (var work in _observerQueue.Reader.ReadAllAsync()) diff --git a/tests/ZB.MOM.WW.ScadaBridge.CLI.Tests/ManagementHttpClientTests.cs b/tests/ZB.MOM.WW.ScadaBridge.CLI.Tests/ManagementHttpClientTests.cs index 60013abb..e35da643 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.CLI.Tests/ManagementHttpClientTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.CLI.Tests/ManagementHttpClientTests.cs @@ -104,3 +104,78 @@ public class ManagementHttpClientTests Assert.Equal("TIMEOUT", response.ErrorCode); } } + +/// +/// WP2.6e (arch-review misc — CLI HttpClient timeout): the public +/// constructor must bound its underlying +/// explicitly (30 s default) rather than leaving the +/// framework's 100 s default in place, and must honor the +/// SCADABRIDGE_HTTP_TIMEOUT_SECONDS override — consistent with how every other +/// CLI setting is environment-overridable (). Runs in the shared +/// "Environment" collection (see ) so it never races another +/// test mutating process-wide environment variables. +/// +[Collection("Environment")] +public class ManagementHttpClientTimeoutTests +{ + private const string EnvVar = "SCADABRIDGE_HTTP_TIMEOUT_SECONDS"; + + [Fact] + public void DefaultConstructor_SetsThirtySecondTimeout_WhenEnvVarUnset() + { + var original = Environment.GetEnvironmentVariable(EnvVar); + try + { + Environment.SetEnvironmentVariable(EnvVar, null); + + using var client = new ManagementHttpClient("http://localhost:9001", "user", "pass"); + + Assert.Equal(TimeSpan.FromSeconds(30), ManagementHttpClient.DefaultTimeout); + Assert.Equal(TimeSpan.FromSeconds(30), client.EffectiveTimeout); + } + finally + { + Environment.SetEnvironmentVariable(EnvVar, original); + } + } + + [Theory] + [InlineData("0")] + [InlineData("-5")] + [InlineData("not-a-number")] + [InlineData("")] + public void InvalidOrNonPositiveEnvValue_FallsBackToDefault(string value) + { + var original = Environment.GetEnvironmentVariable(EnvVar); + try + { + Environment.SetEnvironmentVariable(EnvVar, value); + + using var client = new ManagementHttpClient("http://localhost:9001", "user", "pass"); + + Assert.Equal(TimeSpan.FromSeconds(30), client.EffectiveTimeout); + } + finally + { + Environment.SetEnvironmentVariable(EnvVar, original); + } + } + + [Fact] + public void PositiveEnvValue_OverridesDefaultTimeout() + { + var original = Environment.GetEnvironmentVariable(EnvVar); + try + { + Environment.SetEnvironmentVariable(EnvVar, "5"); + + using var client = new ManagementHttpClient("http://localhost:9001", "user", "pass"); + + Assert.Equal(TimeSpan.FromSeconds(5), client.EffectiveTimeout); + } + finally + { + Environment.SetEnvironmentVariable(EnvVar, original); + } + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.InboundAPI.Tests/ApiMethodCacheTests.cs b/tests/ZB.MOM.WW.ScadaBridge.InboundAPI.Tests/ApiMethodCacheTests.cs new file mode 100644 index 00000000..99a4e262 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.InboundAPI.Tests/ApiMethodCacheTests.cs @@ -0,0 +1,110 @@ +using ZB.MOM.WW.ScadaBridge.Commons.Entities.InboundApi; + +namespace ZB.MOM.WW.ScadaBridge.InboundAPI.Tests; + +/// +/// WP2.6b (arch-review misc — Inbound API per-request SQL): +/// hit/miss/expiry/invalidation behavior, independent of the endpoint and subscriber wiring +/// (covered separately by ). +/// +public class ApiMethodCacheTests +{ + private static ApiMethod Method(string name) => new(name, "return 1;") { Id = 1 }; + + [Fact] + public async Task GetOrFetchAsync_SecondCallWithinTtl_ServesFromCache_NoRefetch() + { + var cache = new ApiMethodCache(TimeSpan.FromMinutes(5)); + var fetchCount = 0; + Task Fetch(CancellationToken _) + { + fetchCount++; + return Task.FromResult(Method("m")); + } + + var first = await cache.GetOrFetchAsync("m", Fetch, CancellationToken.None); + var second = await cache.GetOrFetchAsync("m", Fetch, CancellationToken.None); + + Assert.Equal(1, fetchCount); + Assert.Same(first, second); + } + + [Fact] + public async Task GetOrFetchAsync_AfterTtlExpiry_Refetches() + { + var time = new FakeTimeProvider(DateTimeOffset.UtcNow); + var cache = new ApiMethodCache(TimeSpan.FromSeconds(1), time); + var fetchCount = 0; + Task Fetch(CancellationToken _) + { + fetchCount++; + return Task.FromResult(Method("m")); + } + + await cache.GetOrFetchAsync("m", Fetch, CancellationToken.None); + time.Advance(TimeSpan.FromSeconds(2)); + await cache.GetOrFetchAsync("m", Fetch, CancellationToken.None); + + Assert.Equal(2, fetchCount); + } + + [Fact] + public async Task GetOrFetchAsync_NegativeResult_IsCachedUntilTtlExpires() + { + var cache = new ApiMethodCache(TimeSpan.FromMinutes(5)); + var fetchCount = 0; + Task Fetch(CancellationToken _) + { + fetchCount++; + return Task.FromResult(null); + } + + var first = await cache.GetOrFetchAsync("missing", Fetch, CancellationToken.None); + var second = await cache.GetOrFetchAsync("missing", Fetch, CancellationToken.None); + + Assert.Null(first); + Assert.Null(second); + Assert.Equal(1, fetchCount); + } + + [Fact] + public async Task Invalidate_DropsEntry_NextCallRefetches() + { + var cache = new ApiMethodCache(TimeSpan.FromMinutes(5)); + var fetchCount = 0; + Task Fetch(CancellationToken _) + { + fetchCount++; + return Task.FromResult(Method("m")); + } + + await cache.GetOrFetchAsync("m", Fetch, CancellationToken.None); + cache.Invalidate("m"); + await cache.GetOrFetchAsync("m", Fetch, CancellationToken.None); + + Assert.Equal(2, fetchCount); + } + + [Fact] + public void Invalidate_UnknownName_DoesNotThrow() + { + var cache = new ApiMethodCache(TimeSpan.FromMinutes(5)); + cache.Invalidate("never-cached"); + } + + [Fact] + public void Constructor_NonPositiveTtl_Throws() + { + Assert.Throws(() => new ApiMethodCache(TimeSpan.Zero)); + Assert.Throws(() => new ApiMethodCache(TimeSpan.FromSeconds(-1))); + } + + /// Minimal controllable for TTL-expiry tests. + private sealed class FakeTimeProvider : TimeProvider + { + private DateTimeOffset _now; + public FakeTimeProvider(DateTimeOffset start) => _now = start; + public void Advance(TimeSpan by) => _now += by; + public override DateTimeOffset GetUtcNow() => _now; + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.InboundAPI.Tests/InboundApiOptionsValidatorTests.cs b/tests/ZB.MOM.WW.ScadaBridge.InboundAPI.Tests/InboundApiOptionsValidatorTests.cs index 65af18bc..a5c9f381 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.InboundAPI.Tests/InboundApiOptionsValidatorTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.InboundAPI.Tests/InboundApiOptionsValidatorTests.cs @@ -65,4 +65,22 @@ public class InboundApiOptionsValidatorTests Assert.True(result.Failed); Assert.Contains("MaxRequestBodyBytes", result.FailureMessage); } + + [Fact] + public void ZeroApiMethodCacheTtl_IsRejected() + { + var result = Validate(new InboundApiOptions { ApiMethodCacheTtl = TimeSpan.Zero }); + + Assert.True(result.Failed); + Assert.Contains("ApiMethodCacheTtl", result.FailureMessage); + } + + [Fact] + public void NegativeApiMethodCacheTtl_IsRejected() + { + var result = Validate(new InboundApiOptions { ApiMethodCacheTtl = TimeSpan.FromSeconds(-1) }); + + Assert.True(result.Failed); + Assert.Contains("ApiMethodCacheTtl", result.FailureMessage); + } } diff --git a/tests/ZB.MOM.WW.ScadaBridge.InboundAPI.Tests/ScriptArtifactChangeSubscriberTests.cs b/tests/ZB.MOM.WW.ScadaBridge.InboundAPI.Tests/ScriptArtifactChangeSubscriberTests.cs index 37b3076f..d03d7f52 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.InboundAPI.Tests/ScriptArtifactChangeSubscriberTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.InboundAPI.Tests/ScriptArtifactChangeSubscriberTests.cs @@ -41,12 +41,13 @@ public class ScriptArtifactChangeSubscriberTests private readonly InboundScriptExecutor _executor = new( NullLogger.Instance, Substitute.For()); + private readonly ApiMethodCache _methodCache = new(TimeSpan.FromMinutes(5)); private readonly RecordingBus _bus = new(); private readonly RouteHelper _route = new( Substitute.For(), Substitute.For()); private ScriptArtifactChangeSubscriber CreateSubscriber(IScriptArtifactChangeBus? bus) => - new(_executor, NullLogger.Instance, bus); + new(_executor, _methodCache, NullLogger.Instance, bus); private Task Run(ApiMethod m) => _executor.ExecuteAsync( m, new Dictionary(), _route, TimeSpan.FromSeconds(10)); @@ -121,6 +122,34 @@ public class ScriptArtifactChangeSubscriberTests Assert.Equal(0, _bus.SubscriberCount); } + /// + /// WP2.6b: an ApiMethod change notification must also drop the resolved-row cache + /// entry (not just the compiled handler) — proving EndpointExtensions' next lookup + /// re-fetches instead of serving a stale cached row for the rest of the TTL window. + /// + [Fact] + public async Task ApiMethodPublish_InvalidatesResolvedMethodCache() + { + var subscriber = CreateSubscriber(_bus); + await subscriber.StartAsync(CancellationToken.None); + + var fetchCount = 0; + Task Fetch(CancellationToken _) + { + fetchCount++; + return Task.FromResult(new ApiMethod("m", "return 1;") { Id = 1 }); + } + + await _methodCache.GetOrFetchAsync("m", Fetch, CancellationToken.None); + await _methodCache.GetOrFetchAsync("m", Fetch, CancellationToken.None); + Assert.Equal(1, fetchCount); // second call served from cache + + _bus.Publish(ApiMethodChanged("m")); + + await _methodCache.GetOrFetchAsync("m", Fetch, CancellationToken.None); + Assert.Equal(2, fetchCount); // cache entry was dropped — re-fetched + } + [Fact] public async Task NullBus_NoOps() { diff --git a/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Repositories/SiteRepositoryTests.cs b/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Repositories/SiteRepositoryTests.cs index 366d5eb8..201bf176 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Repositories/SiteRepositoryTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Repositories/SiteRepositoryTests.cs @@ -215,4 +215,99 @@ public class SiteRepositoryTests : IDisposable // (arch-review 08 §1.3/#23) along with the vestigial SiteNotificationRepository — // notification config is central-only and never lives on a site. The synthetic-ID // stability guarantee is still exercised by ExternalSystemRepository_SyntheticId_IsStableAcrossRestart. + + // ── WP2.6a: ExternalSystemDefinitionCache (arch-review misc — site external-system resolution) ── + + /// + /// WP2.6a: two repository instances sharing one + /// (the production DI shape — one singleton cache, many scoped repository instances) must + /// see the SAME cached snapshot: a row written after the cache was already populated by + /// the first repository is invisible to the second until the cache is invalidated. + /// + [Fact] + public async Task ExternalSystemRepository_SharedCache_HitServesStaleSnapshotUntilInvalidated() + { + var storage = NewStorage(); + await storage.InitializeAsync(); + await storage.StoreExternalSystemAsync("Alpha", "https://alpha.test", "None", null, null); + + var cache = new ExternalSystemDefinitionCache(); + var repo1 = new SiteExternalSystemRepository(storage, cache); + var repo2 = new SiteExternalSystemRepository(storage, cache); + + // repo1 populates the shared cache. + var initial = await repo1.GetAllExternalSystemsAsync(); + Assert.Single(initial); + + // A new system is written directly to storage — repo2, sharing the same + // cache, must NOT see it yet (cache hit serves the stale snapshot). + await storage.StoreExternalSystemAsync("Beta", "https://beta.test", "None", null, null); + var stillStale = await repo2.GetAllExternalSystemsAsync(); + Assert.Single(stillStale); + + // After invalidation, the next read reloads from storage and sees both rows. + cache.InvalidateAll(); + var fresh = await repo2.GetAllExternalSystemsAsync(); + Assert.Equal(2, fresh.Count); + } + + /// + /// WP2.6a: the by-ID lookups (system and method) must resolve correctly out of the + /// cached snapshot — proving the O(1) id-index build in LoadSnapshotAsync is + /// wired correctly for both entity kinds, not just the by-name path already covered + /// by ExternalSystemGateway-011. + /// + [Fact] + public async Task ExternalSystemRepository_SharedCache_ByIdLookups_ResolveSystemAndMethod() + { + var storage = NewStorage(); + await storage.InitializeAsync(); + var methodDefs = "[{\"Name\":\"getData\",\"HttpMethod\":\"GET\",\"Path\":\"/data\"}]"; + await storage.StoreExternalSystemAsync( + "WeatherApi", "https://api.example.com", "ApiKey", null, methodDefs); + + var cache = new ExternalSystemDefinitionCache(); + var repo = new SiteExternalSystemRepository(storage, cache); + + var system = await repo.GetExternalSystemByNameAsync("WeatherApi"); + Assert.NotNull(system); + + var byId = await repo.GetExternalSystemByIdAsync(system!.Id); + Assert.NotNull(byId); + Assert.Equal("WeatherApi", byId!.Name); + + var methods = await repo.GetMethodsByExternalSystemIdAsync(system.Id); + Assert.Single(methods); + + var methodById = await repo.GetExternalSystemMethodByIdAsync(methods[0].Id); + Assert.NotNull(methodById); + Assert.Equal("getData", methodById!.Name); + + Assert.Null(await repo.GetExternalSystemByIdAsync(-1)); + Assert.Null(await repo.GetExternalSystemMethodByIdAsync(-1)); + } + + /// + /// WP2.6a: a repository constructed via the single-arg (no shared cache) constructor + /// gets its own private cache — a second such repository over the same storage must + /// see rows written after the first repository's cache was already populated, + /// preserving the pre-cache "always fresh" behavior for callers that opt out of + /// sharing. + /// + [Fact] + public async Task ExternalSystemRepository_PrivateCache_DoesNotShareAcrossInstances() + { + var storage = NewStorage(); + await storage.InitializeAsync(); + await storage.StoreExternalSystemAsync("Alpha", "https://alpha.test", "None", null, null); + + var repo1 = new SiteExternalSystemRepository(storage); + Assert.Single(await repo1.GetAllExternalSystemsAsync()); + + await storage.StoreExternalSystemAsync("Beta", "https://beta.test", "None", null, null); + + // A brand-new repository instance (its own private cache, unpopulated) sees both rows. + var repo2 = new SiteExternalSystemRepository(storage); + Assert.Equal(2, (await repo2.GetAllExternalSystemsAsync()).Count); + } } diff --git a/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/SiteRuntimeOptionsValidatorTests.cs b/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/SiteRuntimeOptionsValidatorTests.cs index 1d627316..13a6b419 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/SiteRuntimeOptionsValidatorTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/SiteRuntimeOptionsValidatorTests.cs @@ -75,4 +75,13 @@ public class SiteRuntimeOptionsValidatorTests Assert.True(result.Failed); Assert.Contains("StartupBatchDelayMs", result.FailureMessage); } + + [Fact] + public void ZeroAlarmPublishQueueCapacity_IsRejected() + { + var result = Validate(new SiteRuntimeOptions { AlarmPublishQueueCapacity = 0 }); + + Assert.True(result.Failed); + Assert.Contains("AlarmPublishQueueCapacity", result.FailureMessage); + } } diff --git a/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Streaming/SiteStreamAlarmDropReporterTests.cs b/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Streaming/SiteStreamAlarmDropReporterTests.cs new file mode 100644 index 00000000..acdad930 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Streaming/SiteStreamAlarmDropReporterTests.cs @@ -0,0 +1,57 @@ +using Akka.TestKit.Xunit2; +using Microsoft.Extensions.Logging.Abstractions; +using ZB.MOM.WW.ScadaBridge.HealthMonitoring; +using ZB.MOM.WW.ScadaBridge.SiteRuntime.Streaming; + +namespace ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests.Streaming; + +/// +/// WP2.6d: the hosted reporter must lift +/// onto the site health report via . Uses the real +/// collector (mirrors ScriptSchedulerStatsReporterTests — NSubstitute is not +/// referenced by this test project). The queue's own drop-counting mechanism is verified +/// deterministically and separately by +/// ; +/// this test proves the reporter's poll-and-push wiring runs correctly. +/// +public class SiteStreamAlarmDropReporterTests : TestKit, IDisposable +{ + void IDisposable.Dispose() => Shutdown(); + + [Fact] + public async Task Reporter_PushesAlarmDropCountToCollector() + { + var options = new SiteRuntimeOptions { StreamBufferSize = 100 }; + var collector = new SiteHealthCollector(); + var streamManager = new SiteStreamManager(options, NullLogger.Instance); + streamManager.Initialize(Sys); + + using var reporter = new SiteStreamAlarmDropReporter( + collector, streamManager, NullLogger.Instance, + pollInterval: TimeSpan.FromMilliseconds(50)); + + await reporter.StartAsync(CancellationToken.None); + try + { + // The immediate first probe (no drops yet) must reach the report as 0 — + // proving the reporter actually ran and pushed a value, not that the field + // simply defaulted. + await WaitUntilAsync(() => + collector.CollectReport("site-1").SiteStreamAlarmDropCount == streamManager.AlarmPublishDroppedCount); + + var report = collector.CollectReport("site-1"); + Assert.Equal(streamManager.AlarmPublishDroppedCount, report.SiteStreamAlarmDropCount); + } + finally + { + await reporter.StopAsync(CancellationToken.None); + } + } + + private static async Task WaitUntilAsync(Func condition) + { + for (var i = 0; i < 100 && !condition(); i++) + await Task.Delay(50); + Assert.True(condition(), "condition not met within timeout"); + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Streaming/SiteStreamManagerTests.cs b/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Streaming/SiteStreamManagerTests.cs index e4dbfbe7..f187e6a0 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Streaming/SiteStreamManagerTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Streaming/SiteStreamManagerTests.cs @@ -159,4 +159,86 @@ public class SiteStreamManagerTests : TestKit, IDisposable _streamManager.RemoveSubscriber(probe.Ref); Assert.Equal(0, _streamManager.SubscriptionCount); } + + // ── WP2.6d: separate alarm publish path, drop counter, skip-at-zero-subscribers ── + + /// + /// WP2.6d: alarm state changes now travel a dedicated publish source, isolated from + /// the (far higher-volume) attribute path — a burst of attribute events for OTHER + /// instances, interleaved with an alarm-only subscriber's events, must not cause any + /// alarm to be lost. + /// + [Fact] + public void PublishAlarmStateChanged_SurvivesConcurrentAttributeStorm_ForUnrelatedInstances() + { + var alarmProbe = CreateTestProbe(); + _streamManager.SubscribeSiteAlarms(alarmProbe.Ref); + + // A storm of attribute events for a DIFFERENT instance — none of which the + // alarm subscriber is even listening to — interleaved with alarm events. Before + // WP2.6d these shared one upstream buffer; now they are fully separate sources. + for (var i = 0; i < 500; i++) + { + _streamManager.PublishAttributeValueChanged(new AttributeValueChanged( + "NoisyPump", "Temperature", "Temperature", i.ToString(), "Good", DateTimeOffset.UtcNow)); + } + + _streamManager.PublishAlarmStateChanged(new AlarmStateChanged( + "Pump1", "HighTemp", AlarmState.Active, 1, DateTimeOffset.UtcNow)); + + var received = alarmProbe.ExpectMsg(TimeSpan.FromSeconds(3)); + Assert.Equal("Pump1", received.InstanceUniqueName); + } + + /// + /// WP2.6d: with zero subscribers of any kind, PublishAlarmStateChanged/ + /// PublishAttributeValueChanged must be no-ops — no exception, and (for alarms) the + /// event never reaches the bounded hand-off queue, so + /// stays at zero rather than + /// counting events nobody could ever have received anyway. + /// + [Fact] + public void Publish_WithNoSubscribers_IsNoOp_AndDoesNotCountAsDropped() + { + _streamManager.PublishAlarmStateChanged(new AlarmStateChanged( + "Pump1", "HighTemp", AlarmState.Active, 1, DateTimeOffset.UtcNow)); + _streamManager.PublishAttributeValueChanged(new AttributeValueChanged( + "Pump1", "Temperature", "Temperature", "1", "Good", DateTimeOffset.UtcNow)); + + Assert.Equal(0, _streamManager.AlarmPublishDroppedCount); + + // Publishing resumes working normally once a subscriber exists. + var probe = CreateTestProbe(); + _streamManager.SubscribeSiteAlarms(probe.Ref); + _streamManager.PublishAlarmStateChanged(new AlarmStateChanged( + "Pump1", "HighTemp", AlarmState.Active, 2, DateTimeOffset.UtcNow)); + + probe.ExpectMsg(TimeSpan.FromSeconds(3)); + } + + /// + /// WP2.6d: once the bounded alarm hand-off queue (the exact factory + /// wires into the drop-count callback) is + /// full, a further write evicts the oldest queued item and fires the drop callback. + /// Exercised directly against the queue factory — deterministic, no actor + /// system/async pump race involved (see the factory's own doc comment). + /// + [Fact] + public void CreateAlarmPublishQueue_OverCapacity_DropsOldestAndInvokesCallback() + { + var dropped = 0; + var queue = SiteStreamManager.CreateAlarmPublishQueue(capacity: 2, () => Interlocked.Increment(ref dropped)); + + // Nothing reads from this queue, so all five writes stay purely upstream — + // capacity 2 means the 3rd/4th/5th writes must each evict the oldest. + for (var i = 0; i < 5; i++) + { + var wrote = queue.Writer.TryWrite(new AlarmStateChanged( + "Pump1", "HighTemp", AlarmState.Active, i, DateTimeOffset.UtcNow)); + Assert.True(wrote); // DropOldest TryWrite always succeeds once capacity > 0 + } + + Assert.Equal(3, dropped); + Assert.Equal(2, queue.Reader.Count); + } } diff --git a/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardOptionsValidatorTests.cs b/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardOptionsValidatorTests.cs index 2a7da0e5..bec17fcb 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardOptionsValidatorTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardOptionsValidatorTests.cs @@ -83,4 +83,16 @@ public class StoreAndForwardOptionsValidatorTests Assert.True(result.Failed); Assert.Contains("SweepTargetParallelism", result.FailureMessage); } + + // ── WP2.6c: bounded observer queue capacity ── + + [Theory] + [InlineData(0)] + [InlineData(-1)] + public void Validate_NonPositiveObserverQueueCapacity_Fails(int capacity) + { + var result = Validate(new StoreAndForwardOptions { ObserverQueueCapacity = capacity }); + Assert.True(result.Failed); + Assert.Contains("ObserverQueueCapacity", result.FailureMessage); + } } diff --git a/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardServiceTests.cs b/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardServiceTests.cs index 434b3d60..82f6c724 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardServiceTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardServiceTests.cs @@ -1,4 +1,6 @@ using Microsoft.Extensions.Logging.Abstractions; +using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services; +using ZB.MOM.WW.ScadaBridge.Commons.Types; using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; using ZB.MOM.WW.ScadaBridge.TestSupport; @@ -885,4 +887,82 @@ public class StoreAndForwardServiceTests : IAsyncLifetime, IDisposable release.TrySetResult(); await stop.WaitAsync(TimeSpan.FromSeconds(5)); // drains the real sweep promptly once released } + + // ── WP2.6c: bounded, DropOldest observer queue + drop counter ── + + private sealed class BlockingObserver : ICachedCallLifecycleObserver + { + private readonly TaskCompletionSource _gate; + public BlockingObserver(TaskCompletionSource gate) => _gate = gate; + public async Task OnAttemptCompletedAsync(CachedCallAttemptContext context, CancellationToken ct = default) + => await _gate.Task; + } + + /// + /// WP2.6c: the cached-call audit-observer queue is bounded — once the single-reader + /// pump is stuck awaiting a slow/stuck observer, further posted notifications must + /// evict the oldest queued one (DropOldest) instead of growing without bound, and + /// every eviction must increment . + /// + [Fact] + public async Task ObserverQueue_BoundedCapacity_DropsOldestAndCountsDrops() + { + var gate = new TaskCompletionSource(); + var observer = new BlockingObserver(gate); + + var localDb = TestLocalDb.CreateTemp("ObsQueueBound"); + var storage = new StoreAndForwardStorage(localDb.Db, NullLogger.Instance); + await storage.InitializeAsync(); + var service = new StoreAndForwardService( + storage, + new StoreAndForwardOptions + { + DefaultRetryInterval = TimeSpan.Zero, + DefaultMaxRetries = 5, + RetryTimerInterval = TimeSpan.FromHours(1), // timer never fires in-test + ObserverQueueCapacity = 2, + }, + NullLogger.Instance, + cachedCallObserver: observer, + siteId: "site-77"); + + await service.StartAsync(); + try + { + service.RegisterDeliveryHandler(StoreAndForwardCategory.ExternalSystem, + _ => throw new HttpRequestException("transient")); + + // Enqueue more messages than the bounded capacity (2) — the pump dequeues + // the first notification and blocks on the observer gate, so every + // subsequent notification posted during this sweep queues (and, past + // capacity, evicts the oldest still-queued one) rather than being + // processed. + for (var i = 0; i < 6; i++) + { + await service.EnqueueAsync( + StoreAndForwardCategory.ExternalSystem, $"t{i}", "{}", + attemptImmediateDelivery: false, retryInterval: TimeSpan.Zero, + messageId: TrackedOperationId.New().ToString()); + } + + await service.RetryPendingMessagesAsync(); + + // Give the bounded channel a moment to have absorbed/evicted every post + // (the pump itself stays blocked on the gate throughout). + var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(5); + while (DateTime.UtcNow < deadline && service.ObserverQueueDroppedCount == 0) + await Task.Delay(10); + + Assert.True(service.ObserverQueueDroppedCount > 0, + "expected at least one notification to be dropped once the bounded queue filled"); + } + finally + { + gate.TrySetResult(); + await service.StopAsync(); + var path = localDb.Path; + localDb.Dispose(); + TestLocalDb.DeleteFiles(path); + } + } }