dohertj2/ScadaBridge
1
0
Fork 0
Code Issues 2 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

perf(misc): cached hot-path lookups, bounded observer queue, alarm-priority stream path

Browse Source 
WP2.6 (arch-review remediation, cross-cutting misc):
- SiteExternalSystemRepository: name/ID-indexed ExternalSystemDefinitionCache replaces
  the fetch-all + reverse-map scan on every by-ID/method lookup; loaded once per
  redeploy, invalidated by DeploymentManagerActor after HandleDeployArtifacts applies
  external-system changes. Static JsonSerializerOptions for method-list parsing.
- Inbound API: short-TTL ApiMethodCache fronts the per-request ApiMethod repository
  fetch; invalidated by name via the existing ScriptArtifactChangeSubscriber/
  IScriptArtifactChangeBus pipeline, self-healing via TTL for changes the bus
  doesn't cover (e.g. Management API edits).
- StoreAndForward: the cached-call audit-observer queue — the one unbounded channel
  left in the system — is now bounded (ObserverQueueCapacity, default 10,000) with
  DropOldest overflow and a dropped-notification counter.
- SiteStreamManager: alarm state changes now travel a dedicated publish
  source/broadcast hub, isolated from the (far higher-volume) attribute path, so an
  attribute storm can no longer evict a pending alarm transition; the alarm hand-off
  queue is bounded with a drop counter surfaced on the site health report
  (SiteStreamAlarmDropCount via the new SiteStreamAlarmDropReporter), and publishing
  is skipped entirely at zero subscribers on either path.
- CLI ManagementHttpClient: explicit 30s HttpClient.Timeout on the shared
  construction (was the 100s framework default), overridable via
  SCADABRIDGE_HTTP_TIMEOUT_SECONDS.

Deviation: the failback-probe heartbeat item is NOT included — its only viable
surface (CentralChannelProvider.cs / heartbeat consumers) lives entirely in the
Communication project, explicitly off-limits to this work package this phase.

Tests: SiteRuntime.Tests (550), InboundAPI.Tests (278), StoreAndForward.Tests (133),
CLI.Tests (390), HealthMonitoring.Tests (97) — all green after full solution build.
This commit is contained in:
Joseph Doherty
2026-08-14 20:59:43 -04:00
parent ee193cd2bb
commit a212283104
32 changed files with 1361 additions and 111 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Streaming/SiteStreamAlarmDropReporter.cs
+78
View File
@@ -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;
/// <summary>
/// WP2.6d (arch-review misc — site stream alarm-vs-attribute shared buffer): site-side
/// hosted service that periodically reads <see cref="SiteStreamManager.AlarmPublishDroppedCount"/>
/// and pushes it into <see cref="ISiteHealthCollector.SetSiteStreamAlarmDropCount"/> so
/// the next <see cref="ISiteHealthCollector.CollectReport"/> carries a fresh snapshot on
/// the site health report. Mirrors <c>ScriptSchedulerStatsReporter</c>: immediate first
/// probe, fixed cadence, exceptions logged and swallowed so the loop survives every probe
/// failure.
/// </summary>
public sealed class SiteStreamAlarmDropReporter : BackgroundService
{
/// <summary>Default poll cadence (10 s) — coarse enough to amortise across health reports.</summary>
internal static readonly TimeSpan DefaultPollInterval = TimeSpan.FromSeconds(10);
private readonly ISiteHealthCollector _collector;
private readonly SiteStreamManager _streamManager;
private readonly ILogger<SiteStreamAlarmDropReporter> _logger;
private readonly TimeSpan _pollInterval;
/// <summary>Initializes a new instance of <see cref="SiteStreamAlarmDropReporter"/>.</summary>
/// <param name="collector">The site health collector that receives the drop count.</param>
/// <param name="streamManager">The site stream manager whose alarm-queue drop count is sampled.</param>
/// <param name="logger">Logger instance.</param>
/// <param name="pollInterval">Poll interval override; defaults to <see cref="DefaultPollInterval"/> (10 s).</param>
public SiteStreamAlarmDropReporter(
ISiteHealthCollector collector,
SiteStreamManager streamManager,
ILogger<SiteStreamAlarmDropReporter> 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;
}
/// <inheritdoc />
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.");
}
}
}
src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Streaming/SiteStreamManager.cs
+176 -30
View File
@@ -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.
/// </summary>
/// <remarks>
/// WP2.6d (arch-review misc — site stream alarm-vs-attribute shared buffer): attribute
/// and alarm events used to share ONE upstream <c>Source.ActorRef</c> + 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
/// (<see cref="_attributeSourceActor"/>/<see cref="_attributeHubSource"/> vs.
/// <see cref="_alarmSourceActor"/>/<see cref="_alarmHubSource"/>), so an attribute storm
/// can only ever evict other attribute events. The alarm path additionally stages
/// through a bounded <see cref="Channel{T}"/> (<see cref="_alarmPublishQueue"/>) so drops
/// are counted (<see cref="AlarmPublishDroppedCount"/>) — <c>Source.ActorRef</c>'s
/// built-in DropHead overflow has no drop callback to instrument. Publishing is also
/// skipped entirely when nobody is subscribed to that source (see
/// <see cref="_attributeSubscriberCount"/>/<see cref="_alarmSubscriberCount"/>).
/// </remarks>
public class SiteStreamManager : ISiteStreamSubscriber
{
/// <summary>Sentinel instance name recorded for site-wide (non-instance-scoped) subscriptions.</summary>
@@ -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<SiteStreamManager> _logger;
private readonly object _lock = new();
private IActorRef? _sourceActor;
private Source<ISiteStreamEvent, NotUsed>? _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<ISiteStreamEvent, NotUsed>? _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<ISiteStreamEvent, NotUsed>? _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<AlarmStateChanged>? _alarmPublishQueue;
private Task? _alarmPump;
private long _alarmPublishDroppedCount;
/// <summary>
/// Cumulative count of alarm state changes dropped because
/// <see cref="_alarmPublishQueue"/> 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.
/// </summary>
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<string, SubscriptionInfo> _subscriptions = new();
/// <summary>Initializes the stream manager with configuration and logger; the Akka stream is not started until <see cref="Initialize"/> is called.</summary>
@@ -41,66 +90,138 @@ public class SiteStreamManager : ISiteStreamSubscriber
ILogger<SiteStreamManager> logger)
{
_bufferSize = options.StreamBufferSize;
_alarmPublishQueueCapacity = options.AlarmPublishQueueCapacity;
_logger = logger;
}
/// <summary>
/// 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.
/// </summary>
/// <param name="system">The running Akka <see cref="ActorSystem"/> used to materialize the broadcast stream.</param>
/// <param name="system">The running Akka <see cref="ActorSystem"/> used to materialize the broadcast streams.</param>
public void Initialize(ActorSystem system)
{
_system = system;
_materializer = _system.Materializer();
var (sourceActor, hubSource) = Source.ActorRef<ISiteStreamEvent>(
var (attributeSourceActor, attributeHubSource) = Source.ActorRef<ISiteStreamEvent>(
_bufferSize,
OverflowStrategy.DropHead)
.ToMaterialized(
BroadcastHub.Sink<ISiteStreamEvent>(bufferSize: 256),
Keep.Both)
.Run(_materializer);
_attributeSourceActor = attributeSourceActor;
_attributeHubSource = attributeHubSource;
_sourceActor = sourceActor;
_hubSource = hubSource;
var (alarmSourceActor, alarmHubSource) = Source.ActorRef<ISiteStreamEvent>(
_bufferSize,
OverflowStrategy.DropHead)
.ToMaterialized(
BroadcastHub.Sink<ISiteStreamEvent>(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);
}
/// <summary>
/// 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 <paramref name="onDropped"/> on every eviction. Extracted as a pure,
/// testable helper (internal — see <c>InternalsVisibleTo</c>) so the drop-counting
/// wiring can be verified deterministically, without racing the async pump/actor
/// system against a burst of publishes.
/// </summary>
internal static Channel<AlarmStateChanged> CreateAlarmPublishQueue(int capacity, Action onDropped) =>
Channel.CreateBounded<AlarmStateChanged>(
new BoundedChannelOptions(Math.Max(1, capacity))
{
SingleReader = true,
SingleWriter = false,
FullMode = BoundedChannelFullMode.DropOldest,
},
itemDropped: _ => onDropped());
/// <summary>
/// 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).
/// </summary>
/// <param name="changed">The attribute value change event to publish.</param>
public void PublishAttributeValueChanged(AttributeValueChanged changed)
{
_sourceActor?.Tell(changed);
if (Volatile.Read(ref _attributeSubscriberCount) == 0)
return;
_attributeSourceActor?.Tell(changed);
}
/// <summary>
/// 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).
/// </summary>
/// <param name="changed">The alarm state change event to publish.</param>
public void PublishAlarmStateChanged(AlarmStateChanged changed)
{
_sourceActor?.Tell(changed);
if (Volatile.Read(ref _alarmSubscriberCount) == 0)
return;
_alarmPublishQueue?.Writer.TryWrite(changed);
}
/// <inheritdoc />
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<ISiteStreamEvent>(), Keep.Right)
.To(Sink.ForEach<ISiteStreamEvent>(ev => capturedSubscriber.Tell(ev)))
.Run(_materializer);
var alarmKillSwitch = _alarmHubSource
.Where(ev => ev.InstanceUniqueName == capturedInstance)
.Buffer(_bufferSize, OverflowStrategy.DropHead)
.ViaMaterialized(KillSwitches.Single<ISiteStreamEvent>(), 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
/// <summary>
/// Subscribe to ALARM events for ALL instances on the site (no per-instance
/// filter). Only <see cref="AlarmStateChanged"/> events are forwarded;
/// <see cref="AttributeValueChanged"/> events are dropped (attributes are far
/// higher-volume and the aggregated Alarm Summary never shows them). Same
/// broadcast-hub wiring as <see cref="Subscribe"/>, and the returned
/// filter). The dedicated alarm hub carries only <see cref="AlarmStateChanged"/>
/// events (the <c>Where</c> 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 <see cref="Subscribe"/>, and the returned
/// subscription id is torn down via <see cref="Unsubscribe"/> exactly like the
/// per-instance variant.
/// </summary>
@@ -133,13 +260,13 @@ public class SiteStreamManager : ISiteStreamSubscriber
/// <returns>A subscription id to pass to <see cref="Unsubscribe"/>.</returns>
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<ISiteStreamEvent>(), 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
/// <summary>
/// Unsubscribe from instance events. Shuts down the per-subscriber
/// stream graph via its KillSwitch.
/// stream graph(s) via their KillSwitch(es).
/// </summary>
/// <param name="subscriptionId">The subscription ID returned by <see cref="Subscribe"/>.</param>
/// <returns><c>true</c> if the subscription was found and removed; <c>false</c> if it was already gone.</returns>
@@ -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
}
}
/// <summary>Shuts down every kill switch for a subscription and decrements the matching per-source counters.</summary>
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);
}
/// <summary>
/// Returns the count of active subscriptions (for diagnostics/testing).
/// </summary>
@@ -213,6 +357,8 @@ public class SiteStreamManager : ISiteStreamSubscriber
private record SubscriptionInfo(
string InstanceName,
IActorRef Subscriber,
IKillSwitch KillSwitch,
IReadOnlyList<IKillSwitch> KillSwitches,
bool TouchesAttributes,
bool TouchesAlarms,
DateTimeOffset SubscribedAt);
}