From 20f6b0b969194c93ab37b8154962cbf38d3171dc Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Sat, 15 Aug 2026 02:23:04 -0400 Subject: [PATCH 1/4] test(loadharness): target-scale load harness for WP-4 / register #25 + row 50 Standalone console harness under tests/ZB.MOM.WW.ScadaBridge.LoadHarness plus a scaled-down Category=Performance smoke [Fact] in PerformanceTests. Deliberately an Exe rather than an xunit suite: the Performance trait enables a filter but does not exclude by default, so a 20-minute test would run on every 'dotnet test' of the slnx. What is real: per-site ActorSystem + LocalDb SQLite file, the real DCL (DataConnectionManagerActor/DataConnectionActor over a SimulatedDataConnection registered through the documented DataConnectionFactory.RegisterAdapter seam), real InstanceActors fed real TagValueUpdates, the real SiteStreamManager, real StreamRelayActor + production-capacity bounded DropOldest channel, real StoreAndForwardService/Storage, real SiteHealthCollector + CentralHealthAggregator. Only the socket hops are stood in for. Measures: end-to-end tag update latency (the emit instant rides TagValueUpdate.Timestamp verbatim to the subscriber), instance ramp, memory growth/CPU over a steady-state window, health report and debug view latency under load, S&F concurrent buffering + drain throughput, and slow-subscriber isolation. --- ZB.MOM.WW.ScadaBridge.slnx | 1 + .../HarnessConfig.cs | 144 +++++++ .../HarnessRun.cs | 275 +++++++++++++ .../Metrics/LatencyHistogram.cs | 136 +++++++ .../Metrics/ResourceSampler.cs | 202 ++++++++++ .../Probes/StreamSubscriberProbe.cs | 205 ++++++++++ .../Program.cs | 53 +++ .../ResultsFormatter.cs | 110 ++++++ .../Scenarios/ObservabilityProbes.cs | 160 ++++++++ .../Scenarios/SlowSubscriberScenario.cs | 153 ++++++++ .../Scenarios/StoreAndForwardDrainScenario.cs | 154 ++++++++ .../SimulatedDataConnection.cs | 153 ++++++++ .../SiteRuntimeFixture.cs | 371 ++++++++++++++++++ .../TagUpdateDriver.cs | 175 +++++++++ .../ZB.MOM.WW.ScadaBridge.LoadHarness.csproj | 40 ++ .../TargetScaleHarnessSmokeTests.cs | 84 ++++ ...MOM.WW.ScadaBridge.PerformanceTests.csproj | 3 + 17 files changed, 2419 insertions(+) create mode 100644 tests/ZB.MOM.WW.ScadaBridge.LoadHarness/HarnessConfig.cs create mode 100644 tests/ZB.MOM.WW.ScadaBridge.LoadHarness/HarnessRun.cs create mode 100644 tests/ZB.MOM.WW.ScadaBridge.LoadHarness/Metrics/LatencyHistogram.cs create mode 100644 tests/ZB.MOM.WW.ScadaBridge.LoadHarness/Metrics/ResourceSampler.cs create mode 100644 tests/ZB.MOM.WW.ScadaBridge.LoadHarness/Probes/StreamSubscriberProbe.cs create mode 100644 tests/ZB.MOM.WW.ScadaBridge.LoadHarness/Program.cs create mode 100644 tests/ZB.MOM.WW.ScadaBridge.LoadHarness/ResultsFormatter.cs create mode 100644 tests/ZB.MOM.WW.ScadaBridge.LoadHarness/Scenarios/ObservabilityProbes.cs create mode 100644 tests/ZB.MOM.WW.ScadaBridge.LoadHarness/Scenarios/SlowSubscriberScenario.cs create mode 100644 tests/ZB.MOM.WW.ScadaBridge.LoadHarness/Scenarios/StoreAndForwardDrainScenario.cs create mode 100644 tests/ZB.MOM.WW.ScadaBridge.LoadHarness/SimulatedDataConnection.cs create mode 100644 tests/ZB.MOM.WW.ScadaBridge.LoadHarness/SiteRuntimeFixture.cs create mode 100644 tests/ZB.MOM.WW.ScadaBridge.LoadHarness/TagUpdateDriver.cs create mode 100644 tests/ZB.MOM.WW.ScadaBridge.LoadHarness/ZB.MOM.WW.ScadaBridge.LoadHarness.csproj create mode 100644 tests/ZB.MOM.WW.ScadaBridge.PerformanceTests/TargetScale/TargetScaleHarnessSmokeTests.cs diff --git a/ZB.MOM.WW.ScadaBridge.slnx b/ZB.MOM.WW.ScadaBridge.slnx index 21748f21..192bc1a4 100644 --- a/ZB.MOM.WW.ScadaBridge.slnx +++ b/ZB.MOM.WW.ScadaBridge.slnx @@ -53,6 +53,7 @@ + diff --git a/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/HarnessConfig.cs b/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/HarnessConfig.cs new file mode 100644 index 00000000..12c2b27b --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/HarnessConfig.cs @@ -0,0 +1,144 @@ +namespace ZB.MOM.WW.ScadaBridge.LoadHarness; + +/// +/// Scale and duration knobs for a load-harness run. Defaults are the Phase-8 WP-4 +/// target scale (10 sites x 500 instances x 75 tags = 375,000 subscriptions). +/// Every value is overridable from the command line so the same binary serves both +/// the full-scale protocol run and the scaled-down CI smoke. +/// +public sealed record HarnessConfig +{ + /// Number of simulated sites (WP-4 acceptance criterion [2.5-1]). + public int Sites { get; init; } = 10; + + /// Instance Actors per site (WP-4 [2.5-2]). + public int InstancesPerSite { get; init; } = 500; + + /// Data-sourced attributes ("live tags") per instance (WP-4 [2.5-3]). + public int TagsPerInstance { get; init; } = 75; + + /// + /// Nominal per-tag update period. The driver emits each tag once per period, so + /// the site-wide event rate is InstancesPerSite * TagsPerInstance / period. + /// The 10 s default puts the target-scale fleet at 37,500 tag updates/second. + /// + public TimeSpan TagUpdatePeriod { get; init; } = TimeSpan.FromSeconds(10); + + /// + /// Length of the steady-state measurement window, measured AFTER the ramp + /// completes and after the post-ramp settle. Memory-growth and CPU figures are + /// computed over exactly this window. + /// + public TimeSpan SustainDuration { get; init; } = TimeSpan.FromMinutes(20); + + /// + /// Quiet period between the last instance starting and the start of the + /// measurement window — lets startup allocations settle so the memory-growth + /// slope reflects steady state, not the ramp. + /// + public TimeSpan SettleDuration { get; init; } = TimeSpan.FromMinutes(2); + + /// Resource-sampling cadence (working set, GC heap, CPU, thread count). + public TimeSpan SampleInterval { get; init; } = TimeSpan.FromSeconds(10); + + /// + /// Health report cadence. Defaults to the production + /// HealthMonitoringOptions.ReportInterval (30 s) so the timing measured is + /// the one that ships; only the CI smoke shortens it, because a 20-second smoke + /// window would otherwise never see a single tick. + /// + public TimeSpan HealthReportInterval { get; init; } = TimeSpan.FromSeconds(30); + + /// How often to take a debug view snapshot of a random live instance. + public TimeSpan DebugProbeInterval { get; init; } = TimeSpan.FromSeconds(5); + + /// + /// Pause between the last Instance Actor starting and the tag driver starting, to + /// let every instance complete its DCL subscribe round-trip. The adapter callback + /// is captured at subscribe time, so an emit before that lands is silently + /// discarded — this window is what keeps SkippedNoCallback at zero. + /// + public TimeSpan SubscribeSettleDuration { get; init; } = TimeSpan.FromSeconds(30); + + /// + /// Instances per site that carry a live stream subscriber (a real + /// StreamRelayActor + bounded DropOldest channel, i.e. the production + /// Debug View / central shape). Every subscriber's stream graph sees the FULL + /// site event flow and filters it, so this is the fan-out multiplier. + /// + public int StreamProbesPerSite { get; init; } = 10; + + /// Store-and-forward messages enqueued for the drain-rate measurement (register row 50). + public int StoreAndForwardDrainMessages { get; init; } = 20_000; + + /// Events published at the slow-subscriber isolation probe (register row 50). + public int SlowSubscriberEvents { get; init; } = 200_000; + + /// Directory for the site SQLite files. A temp directory is used when null. + public string? DataDirectory { get; init; } + + /// Path the JSON metrics document is written to. + public string ResultsPath { get; init; } = "loadharness-results.json"; + + /// Total live tag subscriptions across the fleet. + public int TotalSubscriptions => Sites * InstancesPerSite * TagsPerInstance; + + /// Nominal fleet-wide tag updates per second implied by the scale and update period. + public double NominalUpdatesPerSecond => TotalSubscriptions / TagUpdatePeriod.TotalSeconds; + + /// + /// Parses --key value / --key=value arguments over the defaults. + /// Unknown keys throw so a typo in a 20-minute run's command line fails fast + /// rather than silently measuring the wrong scale. + /// + /// Raw command-line arguments. + /// The parsed configuration. + public static HarnessConfig Parse(string[] args) + { + var config = new HarnessConfig(); + for (var i = 0; i < args.Length; i++) + { + var arg = args[i]; + if (!arg.StartsWith("--", StringComparison.Ordinal)) + throw new ArgumentException($"Unexpected argument '{arg}' (expected --key value)."); + + string key; + string value; + var eq = arg.IndexOf('=', StringComparison.Ordinal); + if (eq >= 0) + { + key = arg[2..eq]; + value = arg[(eq + 1)..]; + } + else + { + key = arg[2..]; + if (i + 1 >= args.Length) + throw new ArgumentException($"Option '--{key}' requires a value."); + value = args[++i]; + } + + config = key switch + { + "sites" => config with { Sites = int.Parse(value) }, + "instances-per-site" => config with { InstancesPerSite = int.Parse(value) }, + "tags-per-instance" => config with { TagsPerInstance = int.Parse(value) }, + "tag-update-period-seconds" => config with { TagUpdatePeriod = TimeSpan.FromSeconds(double.Parse(value)) }, + "sustain-minutes" => config with { SustainDuration = TimeSpan.FromMinutes(double.Parse(value)) }, + "settle-minutes" => config with { SettleDuration = TimeSpan.FromMinutes(double.Parse(value)) }, + "subscribe-settle-seconds" => config with { SubscribeSettleDuration = TimeSpan.FromSeconds(double.Parse(value)) }, + "sample-seconds" => config with { SampleInterval = TimeSpan.FromSeconds(double.Parse(value)) }, + "health-interval-seconds" => config with { HealthReportInterval = TimeSpan.FromSeconds(double.Parse(value)) }, + "debug-probe-interval-seconds" => config with { DebugProbeInterval = TimeSpan.FromSeconds(double.Parse(value)) }, + "stream-probes-per-site" => config with { StreamProbesPerSite = int.Parse(value) }, + "sf-drain-messages" => config with { StoreAndForwardDrainMessages = int.Parse(value) }, + "slow-subscriber-events" => config with { SlowSubscriberEvents = int.Parse(value) }, + "data-dir" => config with { DataDirectory = value }, + "results" => config with { ResultsPath = value }, + _ => throw new ArgumentException($"Unknown option '--{key}'."), + }; + } + + return config; + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/HarnessRun.cs b/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/HarnessRun.cs new file mode 100644 index 00000000..26a3d521 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/HarnessRun.cs @@ -0,0 +1,275 @@ +using System.Diagnostics; +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; +using ZB.MOM.WW.ScadaBridge.HealthMonitoring; +using ZB.MOM.WW.ScadaBridge.LoadHarness.Metrics; +using ZB.MOM.WW.ScadaBridge.LoadHarness.Scenarios; + +namespace ZB.MOM.WW.ScadaBridge.LoadHarness; + +/// Everything one harness run measured, ready for serialization. +/// The configuration the run executed under. +/// Host/runtime description. +/// Run start. +/// Total wall time including ramp and teardown. +/// Wall time to build all site fixtures (before instances). +/// Wall time to create every Instance Actor across all sites. +/// Slowest single site's instance ramp — the "deploy 500 instances to a site" figure. +/// End-to-end DCL-boundary to stream-subscriber latency. +/// Tag updates offered during the whole run. +/// Tag updates offered during the measurement window only. +/// Offered load actually achieved in the measurement window. +/// Offered load the configuration called for. +/// Cumulative driver slice overrun (harness-bound load shortfall). +/// Emissions skipped before subscriptions existed. +/// Resource behaviour over the measurement window. +/// Resource behaviour over the whole run. +/// Health report collect+ingest latency. +/// Health reports ingested by the central aggregator. +/// Sites the central aggregator ended up tracking. +/// Debug view snapshot round-trip latency under load. +/// Debug snapshots that completed. +/// Debug snapshots that timed out. +/// Events delivered to live stream subscribers. +/// Events evicted by live subscribers' bounded channels. +/// Store-and-forward drain measurement (register row 50). +/// Slow-subscriber isolation measurement (register row 50). +public sealed record HarnessRunResult( + HarnessConfig Config, + EnvironmentInfo Environment, + DateTimeOffset StartedUtc, + double TotalSeconds, + double SiteRampSeconds, + double InstanceRampSeconds, + double SlowestSiteInstanceRampSeconds, + LatencySnapshot TagUpdateLatency, + long EmittedTagUpdates, + long SteadyStateEmittedTagUpdates, + double AchievedUpdatesPerSecond, + double NominalUpdatesPerSecond, + double DriverLagSeconds, + long DriverSkippedNoCallback, + ResourceWindowSummary? SteadyStateResources, + ResourceWindowSummary? WholeRunResources, + LatencySnapshot HealthReportLatency, + long HealthReportsDelivered, + int SitesTrackedByAggregator, + LatencySnapshot DebugSnapshotLatency, + long DebugSnapshotsCompleted, + long DebugSnapshotTimeouts, + long StreamProbeReceived, + long StreamProbeDropped, + StoreAndForwardDrainResult? StoreAndForwardDrain, + SlowSubscriberResult? SlowSubscriber); + +/// Host and runtime facts recorded alongside the numbers. +/// Host name. +/// Operating system description. +/// Logical processors visible to the process. +/// .NET runtime version. +/// Whether server GC is active. +public sealed record EnvironmentInfo( + string MachineName, + string OsDescription, + int ProcessorCount, + string RuntimeVersion, + bool ServerGc); + +/// +/// Orchestrates a full run: build sites, ramp instances, attach subscribers, drive +/// tag updates for the sustained window while sampling resources and probing +/// observability, then run the two register-row-50 scenarios on a dedicated site. +/// +public static class HarnessRun +{ + /// Data connections each site spreads its tags across. + public const int ConnectionsPerSite = 5; + + /// Subscribers attached in the slow-subscriber isolation probe. + public const int SlowSubscriberProbeCount = 5; + + /// Concurrent enqueue tasks in the store-and-forward drain probe. + public const int StoreAndForwardConcurrency = 25; + + /// Executes a run end to end. + /// Scale and duration configuration. + /// Progress sink (stdout in the console app). + /// Cancels the run. + /// The measured result. + public static async Task ExecuteAsync( + HarnessConfig config, Action log, CancellationToken cancellationToken) + { + var startedUtc = DateTimeOffset.UtcNow; + var totalWatch = Stopwatch.StartNew(); + var dataRoot = config.DataDirectory + ?? Path.Combine(Path.GetTempPath(), $"scadabridge-loadharness-{Guid.NewGuid():N}"); + Directory.CreateDirectory(dataRoot); + + var sampler = ResourceSampler.Start(config.SampleInterval); + var sites = new List(config.Sites); + var latency = new LatencyHistogram(); + + TagUpdateDriver? driver = null; + ObservabilityProbes? probes = null; + StoreAndForwardDrainResult? drainResult = null; + SlowSubscriberResult? slowResult = null; + + var aggregator = new CentralHealthAggregator( + Options.Create(new HealthMonitoringOptions + { + ReportInterval = config.HealthReportInterval, + OfflineTimeout = config.HealthReportInterval * 2, + }), + NullLogger.Instance); + + try + { + log($"Building {config.Sites} sites ({config.InstancesPerSite} instances x " + + $"{config.TagsPerInstance} tags each = {config.TotalSubscriptions:N0} subscriptions)..."); + + var siteWatch = Stopwatch.StartNew(); + for (var s = 0; s < config.Sites; s++) + sites.Add(await SiteRuntimeFixture.CreateAsync(s, config, dataRoot, ConnectionsPerSite)); + siteWatch.Stop(); + log($" sites built in {siteWatch.Elapsed.TotalSeconds:F1}s"); + + // Instance ramp — sites in parallel (they are independent actor systems, + // exactly as 10 real sites would be), each site internally staggered at the + // production StartupBatchSize/StartupBatchDelayMs pacing. + var rampWatch = Stopwatch.StartNew(); + await Task.WhenAll(sites.Select(site => site.StartInstancesAsync(cancellationToken))); + rampWatch.Stop(); + var slowestSiteRamp = sites.Max(s => s.InstanceRampDuration.TotalSeconds); + log($" {config.Sites * config.InstancesPerSite:N0} instance actors created in " + + $"{rampWatch.Elapsed.TotalSeconds:F1}s (slowest site {slowestSiteRamp:F1}s)"); + + foreach (var site in sites) + site.AttachStreamProbes(latency); + log($" {sites.Sum(s => s.Probes.Count)} live stream subscribers attached"); + + // Let Instance Actors complete their DCL subscribe round-trips before the + // driver starts; an emit before SubscribeBatchAsync has captured the + // callback would be silently discarded. + await Task.Delay(config.SubscribeSettleDuration, cancellationToken); + + driver = TagUpdateDriver.Start(sites, config); + log($" tag driver started, nominal {config.NominalUpdatesPerSecond:N0} updates/s"); + + probes = ObservabilityProbes.Start( + sites, aggregator, config.HealthReportInterval, config.DebugProbeInterval); + + log($"Settling for {config.SettleDuration.TotalMinutes:F1} min..."); + await Task.Delay(config.SettleDuration, cancellationToken); + + // Switch to a fresh latency histogram so ramp-window outliers do not + // contaminate the steady-state percentiles. Everything reported as "steady + // state" is measured strictly after this point; the subscriptions + // themselves are left untouched. + var steadyStateStartSeconds = sampler.Snapshot().LastOrDefault()?.ElapsedSeconds ?? 0; + var emittedAtWindowStart = driver.EmittedCount; + var receivedAtWindowStart = sites.Sum(s => s.Probes.Sum(p => p.ReceivedEvents)); + var droppedAtWindowStart = sites.Sum(s => s.Probes.Sum(p => p.DroppedEvents)); + + var steadyLatency = new LatencyHistogram(); + foreach (var site in sites) + { + foreach (var probe in site.Probes) + probe.RetargetLatency(steadyLatency); + } + + var windowWatch = Stopwatch.StartNew(); + log($"Sustained measurement window: {config.SustainDuration.TotalMinutes:F1} min..."); + await Task.Delay(config.SustainDuration, cancellationToken); + windowWatch.Stop(); + + var steadyStateEndSeconds = sampler.Snapshot().LastOrDefault()?.ElapsedSeconds ?? 0; + var emittedInWindow = driver.EmittedCount - emittedAtWindowStart; + + var steadyResources = sampler.Summarize(steadyStateStartSeconds, steadyStateEndSeconds); + log($" window complete: {emittedInWindow:N0} updates offered, " + + $"{emittedInWindow / windowWatch.Elapsed.TotalSeconds:N0}/s achieved"); + + var probeReceived = sites.Sum(s => s.Probes.Sum(p => p.ReceivedEvents)) - receivedAtWindowStart; + var probeDropped = sites.Sum(s => s.Probes.Sum(p => p.DroppedEvents)) - droppedAtWindowStart; + + // Register row 50 — measured after the sustained window so the numbers are + // not competing with the full tag load for CPU, and reported separately for + // the same reason. + log("Stopping tag driver for the register row 50 scenarios..."); + var driverLagSeconds = driver.EmitLagSeconds; + var driverSkipped = driver.SkippedNoCallback; + await driver.DisposeAsync(); + driver = null; + await probes.DisposeAsync(); + + log($"Store-and-forward drain: {config.StoreAndForwardDrainMessages:N0} messages..."); + drainResult = await StoreAndForwardDrainScenario.RunAsync( + sites[0], config.StoreAndForwardDrainMessages, StoreAndForwardConcurrency, cancellationToken); + log($" buffered {drainResult.EnqueuePerSecond:N0}/s, drained {drainResult.DrainPerSecond:N0}/s"); + + log($"Slow-subscriber isolation: {SlowSubscriberProbeCount} subscribers, " + + $"{config.SlowSubscriberEvents:N0} events..."); + slowResult = await SlowSubscriberScenario.RunAsync( + sites[1 % sites.Count], SlowSubscriberProbeCount, config.SlowSubscriberEvents, cancellationToken); + log($" healthy min delivery {slowResult.HealthyMinDeliveryRatio:P2}, " + + $"stalled {slowResult.SlowDeliveryRatio:P2}"); + + totalWatch.Stop(); + + return new HarnessRunResult( + Config: config, + Environment: CaptureEnvironment(), + StartedUtc: startedUtc, + TotalSeconds: totalWatch.Elapsed.TotalSeconds, + SiteRampSeconds: siteWatch.Elapsed.TotalSeconds, + InstanceRampSeconds: rampWatch.Elapsed.TotalSeconds, + SlowestSiteInstanceRampSeconds: slowestSiteRamp, + TagUpdateLatency: steadyLatency.Snapshot(), + EmittedTagUpdates: emittedAtWindowStart + emittedInWindow, + SteadyStateEmittedTagUpdates: emittedInWindow, + AchievedUpdatesPerSecond: emittedInWindow / windowWatch.Elapsed.TotalSeconds, + NominalUpdatesPerSecond: config.NominalUpdatesPerSecond, + DriverLagSeconds: driverLagSeconds, + DriverSkippedNoCallback: driverSkipped, + SteadyStateResources: steadyResources, + WholeRunResources: sampler.Summarize(0, double.MaxValue), + HealthReportLatency: probes.HealthReportLatency.Snapshot(), + HealthReportsDelivered: probes.HealthReportsDelivered, + SitesTrackedByAggregator: aggregator.GetAllSiteStates().Count, + DebugSnapshotLatency: probes.DebugSnapshotLatency.Snapshot(), + DebugSnapshotsCompleted: probes.DebugSnapshotsCompleted, + DebugSnapshotTimeouts: probes.DebugSnapshotTimeouts, + StreamProbeReceived: probeReceived, + StreamProbeDropped: probeDropped, + StoreAndForwardDrain: drainResult, + SlowSubscriber: slowResult); + } + finally + { + if (driver != null) await driver.DisposeAsync(); + if (probes != null) await probes.DisposeAsync(); + await sampler.DisposeAsync(); + + log("Tearing down sites..."); + foreach (var site in sites) + await site.DisposeAsync(); + + try + { + if (config.DataDirectory == null && Directory.Exists(dataRoot)) + Directory.Delete(dataRoot, recursive: true); + } + catch (IOException) + { + // Temp cleanup only. + } + } + } + + private static EnvironmentInfo CaptureEnvironment() => new( + System.Environment.MachineName, + System.Runtime.InteropServices.RuntimeInformation.OSDescription, + System.Environment.ProcessorCount, + System.Runtime.InteropServices.RuntimeInformation.FrameworkDescription, + System.Runtime.GCSettings.IsServerGC); +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/Metrics/LatencyHistogram.cs b/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/Metrics/LatencyHistogram.cs new file mode 100644 index 00000000..2c19d71b --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/Metrics/LatencyHistogram.cs @@ -0,0 +1,136 @@ +namespace ZB.MOM.WW.ScadaBridge.LoadHarness.Metrics; + +/// +/// Lock-free logarithmic latency histogram sized for tens of thousands of samples +/// per second across many threads. +/// +/// +/// Buckets are 16-per-octave over microseconds, i.e. bucket i covers +/// [2^(i/16), 2^((i+1)/16)) µs. That bounds relative bucket width at +/// 2^(1/16) - 1 ≈ 4.4%, so a reported percentile is within ~4.4% of the true +/// value — ample for the millisecond-scale thresholds this harness asserts, and far +/// cheaper than retaining 45 million raw samples. +/// +/// +/// Recording is a plus one +/// ; there is no allocation on the hot path. +/// +/// +public sealed class LatencyHistogram +{ + private const int SubBucketsPerOctave = 16; + private const int BucketCount = 64 * SubBucketsPerOctave; + + private readonly long[] _buckets = new long[BucketCount]; + private long _count; + private long _totalMicroseconds; + private long _maxMicroseconds; + + /// Number of samples recorded. + public long Count => Interlocked.Read(ref _count); + + /// Largest sample seen, in microseconds (exact — not bucketed). + public double MaxMs => Interlocked.Read(ref _maxMicroseconds) / 1000.0; + + /// Arithmetic mean in milliseconds (exact — accumulated, not bucketed). + public double MeanMs + { + get + { + var count = Interlocked.Read(ref _count); + return count == 0 ? 0 : Interlocked.Read(ref _totalMicroseconds) / 1000.0 / count; + } + } + + /// + /// Records one sample. Negative durations (clock skew across the emit/receive + /// boundary) are clamped to zero rather than discarded, so the sample count stays + /// an honest denominator. + /// + /// The measured latency. + public void Record(TimeSpan elapsed) + { + var micros = (long)(elapsed.TotalMilliseconds * 1000.0); + if (micros < 0) micros = 0; + + Interlocked.Increment(ref _count); + Interlocked.Add(ref _totalMicroseconds, micros); + + long observedMax; + while (micros > (observedMax = Interlocked.Read(ref _maxMicroseconds))) + { + if (Interlocked.CompareExchange(ref _maxMicroseconds, micros, observedMax) == observedMax) + break; + } + + Interlocked.Increment(ref _buckets[BucketIndex(micros)]); + } + + private static int BucketIndex(long micros) + { + if (micros <= 0) return 0; + var index = (int)(Math.Log2(micros) * SubBucketsPerOctave); + if (index < 0) return 0; + return index >= BucketCount ? BucketCount - 1 : index; + } + + /// Bucket midpoint in milliseconds, used when reconstructing a percentile. + private static double BucketMidpointMs(int index) + { + var low = Math.Pow(2, (double)index / SubBucketsPerOctave); + var high = Math.Pow(2, (double)(index + 1) / SubBucketsPerOctave); + return (low + high) / 2.0 / 1000.0; + } + + /// + /// Returns the requested percentile in milliseconds, or 0 when no samples were recorded. + /// + /// Percentile in the range 0..100 (e.g. 99 for P99). + /// The percentile value in milliseconds. + public double PercentileMs(double percentile) + { + var total = Interlocked.Read(ref _count); + if (total == 0) return 0; + + var target = (long)Math.Ceiling(total * percentile / 100.0); + if (target < 1) target = 1; + + long cumulative = 0; + for (var i = 0; i < BucketCount; i++) + { + cumulative += Interlocked.Read(ref _buckets[i]); + if (cumulative >= target) + return BucketMidpointMs(i); + } + + return MaxMs; + } + + /// Materializes the standard percentile set plus mean/max/count for reporting. + /// A snapshot record of this histogram. + public LatencySnapshot Snapshot() => new( + Count, + MeanMs, + PercentileMs(50), + PercentileMs(95), + PercentileMs(99), + PercentileMs(99.9), + MaxMs); +} + +/// Point-in-time summary of a . All times in milliseconds. +/// Samples recorded. +/// Arithmetic mean. +/// Median. +/// 95th percentile. +/// 99th percentile. +/// 99.9th percentile. +/// Largest observed sample. +public sealed record LatencySnapshot( + long Count, + double MeanMs, + double P50Ms, + double P95Ms, + double P99Ms, + double P999Ms, + double MaxMs); diff --git a/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/Metrics/ResourceSampler.cs b/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/Metrics/ResourceSampler.cs new file mode 100644 index 00000000..a6fdd7fe --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/Metrics/ResourceSampler.cs @@ -0,0 +1,202 @@ +using System.Diagnostics; + +namespace ZB.MOM.WW.ScadaBridge.LoadHarness.Metrics; + +/// +/// One periodic observation of process resource usage. +/// +/// Seconds since sampling started. +/// Process working set (RSS). +/// without forcing a collection. +/// Mean CPU utilization since the previous sample, as a percentage of ONE core. +/// OS threads in the process. +/// Cumulative gen-2 collections. +public sealed record ResourceSample( + double ElapsedSeconds, + double WorkingSetMb, + double ManagedHeapMb, + double CpuPercent, + int ThreadCount, + int Gen2Collections); + +/// +/// Samples working set, managed heap, CPU and thread count on a fixed cadence for the +/// life of a run. CPU is differential (processor time delta / wall delta) so a sample +/// reflects the interval it covers rather than the whole process lifetime. +/// +public sealed class ResourceSampler : IAsyncDisposable +{ + private readonly List _samples = new(); + private readonly object _lock = new(); + private readonly CancellationTokenSource _cts = new(); + private readonly Task _loop; + private readonly Stopwatch _wall = Stopwatch.StartNew(); + + private TimeSpan _lastCpu; + private double _lastElapsedSeconds; + + private ResourceSampler(TimeSpan interval) + { + _lastCpu = Process.GetCurrentProcess().TotalProcessorTime; + _loop = Task.Run(() => SampleLoopAsync(interval, _cts.Token)); + } + + /// Starts sampling at the given cadence. + /// Sampling interval. + /// The running sampler. + public static ResourceSampler Start(TimeSpan interval) => new(interval); + + /// All samples collected so far, oldest first. + /// A snapshot copy of the sample list. + public IReadOnlyList Snapshot() + { + lock (_lock) return _samples.ToList(); + } + + private async Task SampleLoopAsync(TimeSpan interval, CancellationToken cancellationToken) + { + using var timer = new PeriodicTimer(interval); + try + { + while (await timer.WaitForNextTickAsync(cancellationToken)) + Capture(); + } + catch (OperationCanceledException) + { + // Normal teardown. + } + } + + private void Capture() + { + using var process = Process.GetCurrentProcess(); + process.Refresh(); + + var elapsedSeconds = _wall.Elapsed.TotalSeconds; + var cpu = process.TotalProcessorTime; + var wallDelta = elapsedSeconds - _lastElapsedSeconds; + var cpuPercent = wallDelta > 0 + ? (cpu - _lastCpu).TotalSeconds / wallDelta * 100.0 + : 0.0; + + _lastCpu = cpu; + _lastElapsedSeconds = elapsedSeconds; + + var sample = new ResourceSample( + elapsedSeconds, + process.WorkingSet64 / 1024.0 / 1024.0, + GC.GetTotalMemory(forceFullCollection: false) / 1024.0 / 1024.0, + cpuPercent, + process.Threads.Count, + GC.CollectionCount(2)); + + lock (_lock) _samples.Add(sample); + } + + /// + /// Summarizes the samples falling inside a window, expressed as seconds since + /// sampling started. Memory growth is reported both as an absolute delta and as a + /// least-squares slope, because a run that sawtooths around a stable mean and a + /// run that climbs monotonically can share the same endpoint delta. + /// + /// Window start (inclusive), seconds since start. + /// Window end (inclusive), seconds since start. + /// The window summary, or null when fewer than two samples fall inside it. + public ResourceWindowSummary? Summarize(double fromSeconds, double toSeconds) + { + var window = Snapshot() + .Where(s => s.ElapsedSeconds >= fromSeconds && s.ElapsedSeconds <= toSeconds) + .ToList(); + + if (window.Count < 2) + return null; + + var first = window[0]; + var last = window[^1]; + + return new ResourceWindowSummary( + SampleCount: window.Count, + DurationSeconds: last.ElapsedSeconds - first.ElapsedSeconds, + WorkingSetStartMb: first.WorkingSetMb, + WorkingSetEndMb: last.WorkingSetMb, + WorkingSetPeakMb: window.Max(s => s.WorkingSetMb), + WorkingSetSlopeMbPerMinute: Slope(window, s => s.WorkingSetMb) * 60.0, + ManagedHeapStartMb: first.ManagedHeapMb, + ManagedHeapEndMb: last.ManagedHeapMb, + ManagedHeapPeakMb: window.Max(s => s.ManagedHeapMb), + ManagedHeapSlopeMbPerMinute: Slope(window, s => s.ManagedHeapMb) * 60.0, + MeanCpuPercentOfOneCore: window.Average(s => s.CpuPercent), + PeakCpuPercentOfOneCore: window.Max(s => s.CpuPercent), + MeanThreadCount: window.Average(s => s.ThreadCount), + Gen2Collections: last.Gen2Collections - first.Gen2Collections); + } + + private static double Slope(IReadOnlyList samples, Func selector) + { + var n = samples.Count; + var meanX = samples.Average(s => s.ElapsedSeconds); + var meanY = samples.Average(selector); + + double numerator = 0, denominator = 0; + for (var i = 0; i < n; i++) + { + var dx = samples[i].ElapsedSeconds - meanX; + numerator += dx * (selector(samples[i]) - meanY); + denominator += dx * dx; + } + + return denominator == 0 ? 0 : numerator / denominator; + } + + private int _disposed; + + /// + public async ValueTask DisposeAsync() + { + if (Interlocked.Exchange(ref _disposed, 1) != 0) + return; + + await _cts.CancelAsync(); + try + { + await _loop; + } + catch (OperationCanceledException) + { + // Expected. + } + + _cts.Dispose(); + } +} + +/// Aggregate resource behaviour over a measurement window. +/// Samples in the window. +/// Window length. +/// Working set at window start. +/// Working set at window end. +/// Peak working set in the window. +/// Least-squares working-set growth rate. +/// Managed heap at window start. +/// Managed heap at window end. +/// Peak managed heap in the window. +/// Least-squares managed-heap growth rate. +/// Mean CPU as a percentage of one core (1400% = 14 cores saturated). +/// Peak single-sample CPU as a percentage of one core. +/// Mean OS thread count. +/// Gen-2 collections during the window. +public sealed record ResourceWindowSummary( + int SampleCount, + double DurationSeconds, + double WorkingSetStartMb, + double WorkingSetEndMb, + double WorkingSetPeakMb, + double WorkingSetSlopeMbPerMinute, + double ManagedHeapStartMb, + double ManagedHeapEndMb, + double ManagedHeapPeakMb, + double ManagedHeapSlopeMbPerMinute, + double MeanCpuPercentOfOneCore, + double PeakCpuPercentOfOneCore, + double MeanThreadCount, + int Gen2Collections); diff --git a/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/Probes/StreamSubscriberProbe.cs b/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/Probes/StreamSubscriberProbe.cs new file mode 100644 index 00000000..8c3079f1 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/Probes/StreamSubscriberProbe.cs @@ -0,0 +1,205 @@ +using System.Threading.Channels; +using Akka.Actor; +using ZB.MOM.WW.ScadaBridge.Communication.Actors; +using ZB.MOM.WW.ScadaBridge.Communication.Grpc; +using ZB.MOM.WW.ScadaBridge.LoadHarness.Metrics; +using ZB.MOM.WW.ScadaBridge.SiteRuntime.Streaming; + +namespace ZB.MOM.WW.ScadaBridge.LoadHarness.Probes; + +/// +/// A live site-stream subscriber assembled from the SAME parts +/// SiteStreamGrpcServer.RunSubscriptionStreamAsync uses: +/// +/// +/// SiteStreamManager.Subscribe — materializes the +/// per-subscriber graph (Where instance filter → Buffer(StreamBufferSize, +/// DropHead) → Sink.ForEach(Tell)). +/// A real , which converts the Akka +/// record to the protobuf SiteStreamEvent and TryWrites it. +/// A bounded DropOldest of the +/// production capacity (GrpcInstanceStreamChannelCapacity = 1000) with the +/// eviction counter wired to . +/// +/// +/// +/// The single substitution is the final hop: instead of +/// responseStream.WriteAsync pushing onto a socket, a reader task drains the +/// channel. That is deliberate — it is precisely the hop whose slowness register row +/// 50 asks about, and a controllable reader is the only way to hold it still. +/// +/// +public sealed class StreamSubscriberProbe : IAsyncDisposable +{ + /// Production per-instance stream channel capacity (GrpcInstanceStreamChannelCapacity). + public const int ProductionChannelCapacity = 1000; + + private readonly SiteStreamManager _manager; + private readonly string _subscriptionId; + private readonly IActorRef _relayActor; + private readonly ActorSystem _system; + private readonly Channel _channel; + private readonly CancellationTokenSource _cts = new(); + private readonly Task _readerTask; + private volatile LatencyHistogram? _latency; + + private readonly DropCounter _dropCounter; + private long _received; + private long _readerDelayMicroseconds; + private int _disposed; + + /// Human-readable probe name (also the relay actor's name suffix). + public string Name { get; } + + /// Events evicted by the bounded channel's DropOldest policy. + public long DroppedEvents => _dropCounter.Value; + + /// Events successfully drained by the reader (i.e. "sent to the client"). + public long ReceivedEvents => Interlocked.Read(ref _received); + + /// + /// Repoints the latency histogram this probe records into, without tearing the + /// subscription down. Used to separate ramp-window samples from steady-state ones: + /// re-attaching probes instead would open a zero-subscriber gap (during which + /// PublishAttributeValueChanged short-circuits) and risk reusing an actor + /// name whose previous incarnation has not finished terminating. + /// + /// The histogram to record into from now on, or null to stop recording. + public void RetargetLatency(LatencyHistogram? latency) => _latency = latency; + + /// + /// Artificial per-event reader delay, in microseconds. Zero is a healthy + /// subscriber; a large value models a stalled WAN link or a wedged client. + /// + public long ReaderDelayMicroseconds + { + get => Interlocked.Read(ref _readerDelayMicroseconds); + set => Interlocked.Exchange(ref _readerDelayMicroseconds, value); + } + + private StreamSubscriberProbe( + ActorSystem system, + SiteStreamManager manager, + string name, + Channel channel, + IActorRef relayActor, + string subscriptionId, + LatencyHistogram? latency, + DropCounter dropCounter) + { + _system = system; + _manager = manager; + Name = name; + _channel = channel; + _relayActor = relayActor; + _subscriptionId = subscriptionId; + _latency = latency; + _dropCounter = dropCounter; + _readerTask = Task.Run(() => ReadLoopAsync(_cts.Token)); + } + + /// + /// Builds and attaches a probe subscribed to one instance's events. + /// + /// The site actor system. + /// The site stream manager to subscribe against. + /// Instance whose events this probe receives. + /// Probe name, used for the relay actor's path. + /// Optional histogram fed with end-to-end event latency. + /// The attached probe. + public static StreamSubscriberProbe Attach( + ActorSystem system, + SiteStreamManager manager, + string instanceUniqueName, + string name, + LatencyHistogram? latency) + { + var dropCounter = new DropCounter(); + var channel = Channel.CreateBounded( + new BoundedChannelOptions(ProductionChannelCapacity) + { + FullMode = BoundedChannelFullMode.DropOldest, + }, + _ => dropCounter.Increment()); + + var relayActor = system.ActorOf( + Props.Create(typeof(StreamRelayActor), name, channel.Writer), + $"stream-relay-{name}"); + + var subscriptionId = manager.Subscribe(instanceUniqueName, relayActor); + + return new StreamSubscriberProbe( + system, manager, name, channel, relayActor, subscriptionId, latency, dropCounter); + } + + private async Task ReadLoopAsync(CancellationToken cancellationToken) + { + try + { + await foreach (var evt in _channel.Reader.ReadAllAsync(cancellationToken)) + { + Interlocked.Increment(ref _received); + + var latency = _latency; + if (latency != null && evt.AttributeChanged != null) + { + // The emit instant travels verbatim: the driver stamps it on + // TagValueUpdate.Timestamp, DataConnectionActor forwards it, + // InstanceActor copies it onto AttributeValueChanged.Timestamp, and + // StreamRelayActor maps it onto the proto Timestamp. So this is a + // true end-to-end DCL-boundary → subscriber measurement. + var emitted = evt.AttributeChanged.Timestamp.ToDateTimeOffset(); + latency.Record(DateTimeOffset.UtcNow - emitted); + } + + var delay = Interlocked.Read(ref _readerDelayMicroseconds); + if (delay > 0) + await Task.Delay(TimeSpan.FromMicroseconds(delay), cancellationToken); + } + } + catch (OperationCanceledException) + { + // Normal teardown. + } + } + + /// Detaches the subscription and stops the relay actor and reader. + /// A task that completes when the probe is torn down. + public async ValueTask DisposeAsync() + { + if (Interlocked.Exchange(ref _disposed, 1) != 0) + return; + + _manager.Unsubscribe(_subscriptionId); + _channel.Writer.TryComplete(); + await _cts.CancelAsync(); + + try + { + await _readerTask; + } + catch (OperationCanceledException) + { + // Expected. + } + + _system.Stop(_relayActor); + _cts.Dispose(); + } +} + +/// +/// Thread-safe counter for a bounded channel's itemDropped callback. A tiny +/// class rather than a captured local so the probe and the channel share exactly one +/// counter instance without a second closure. +/// +public sealed class DropCounter +{ + private long _value; + + /// Current count. + public long Value => Interlocked.Read(ref _value); + + /// Increments the counter. + public void Increment() => Interlocked.Increment(ref _value); +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/Program.cs b/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/Program.cs new file mode 100644 index 00000000..30fc5e11 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/Program.cs @@ -0,0 +1,53 @@ +using System.Text.Json; +using ZB.MOM.WW.ScadaBridge.LoadHarness; + +// Target-scale load harness (Phase-8 WP-4 / deferred-work register #25 + row 50). +// +// Full-scale protocol run (the WP-4 numbers): +// dotnet run -c Release --project tests/ZB.MOM.WW.ScadaBridge.LoadHarness -- \ +// --sustain-minutes 20 --results loadharness-results.json +// +// Full 1-hour version, unchanged in every other respect: +// ... -- --sustain-minutes 60 --results loadharness-results-1h.json +// +// Scaled-down smoke (what the CI [Fact] runs): +// ... -- --sites 2 --instances-per-site 10 --tags-per-instance 5 \ +// --settle-minutes 0.1 --sustain-minutes 0.2 --sample-seconds 2 \ +// --sf-drain-messages 200 --slow-subscriber-events 2000 + +var config = HarnessConfig.Parse(args); + +Console.WriteLine("ScadaBridge target-scale load harness"); +Console.WriteLine($" sites {config.Sites}"); +Console.WriteLine($" instances/site {config.InstancesPerSite}"); +Console.WriteLine($" tags/instance {config.TagsPerInstance}"); +Console.WriteLine($" total subscriptions {config.TotalSubscriptions:N0}"); +Console.WriteLine($" nominal update rate {config.NominalUpdatesPerSecond:N0}/s"); +Console.WriteLine($" settle / sustain {config.SettleDuration.TotalMinutes:F1} / {config.SustainDuration.TotalMinutes:F1} min"); +Console.WriteLine(); + +using var cancellation = new CancellationTokenSource(); +Console.CancelKeyPress += (_, e) => +{ + e.Cancel = true; + cancellation.Cancel(); +}; + +try +{ + var result = await HarnessRun.ExecuteAsync(config, Console.WriteLine, cancellation.Token); + + var json = JsonSerializer.Serialize(result, new JsonSerializerOptions { WriteIndented = true }); + await File.WriteAllTextAsync(config.ResultsPath, json, cancellation.Token); + + Console.WriteLine(); + Console.WriteLine(ResultsFormatter.Format(result)); + Console.WriteLine(); + Console.WriteLine($"JSON metrics written to {Path.GetFullPath(config.ResultsPath)}"); + return 0; +} +catch (OperationCanceledException) +{ + Console.Error.WriteLine("Run cancelled."); + return 130; +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/ResultsFormatter.cs b/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/ResultsFormatter.cs new file mode 100644 index 00000000..5c9e40c3 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/ResultsFormatter.cs @@ -0,0 +1,110 @@ +using System.Globalization; +using System.Text; + +namespace ZB.MOM.WW.ScadaBridge.LoadHarness; + +/// +/// Renders a as the markdown table that goes into the +/// results document, so the published numbers and the JSON come from one source. +/// +public static class ResultsFormatter +{ + /// Formats a run result for console output and the results doc. + /// The run to format. + /// A markdown-ish plain-text report. + public static string Format(HarnessRunResult result) + { + var culture = CultureInfo.InvariantCulture; + var builder = new StringBuilder(); + + builder.AppendLine("=== MEASURED ==="); + builder.AppendLine(culture, $"host {result.Environment.MachineName} / {result.Environment.OsDescription}"); + builder.AppendLine(culture, $"cpus / runtime / gc {result.Environment.ProcessorCount} / {result.Environment.RuntimeVersion} / serverGC={result.Environment.ServerGc}"); + builder.AppendLine(culture, $"scale {result.Config.Sites} sites x {result.Config.InstancesPerSite} instances x {result.Config.TagsPerInstance} tags = {result.Config.TotalSubscriptions:N0} subscriptions"); + builder.AppendLine(culture, $"total run {result.TotalSeconds:F1}s"); + builder.AppendLine(); + + builder.AppendLine("-- deployment / ramp --"); + builder.AppendLine(culture, $"site fixtures built {result.SiteRampSeconds:F1}s"); + builder.AppendLine(culture, $"all instance actors {result.InstanceRampSeconds:F1}s"); + builder.AppendLine(culture, $"slowest single site {result.SlowestSiteInstanceRampSeconds:F1}s ({result.Config.InstancesPerSite} instances)"); + builder.AppendLine(); + + builder.AppendLine("-- offered load (steady-state window) --"); + builder.AppendLine(culture, $"nominal {result.NominalUpdatesPerSecond:N0} updates/s"); + builder.AppendLine(culture, $"achieved {result.AchievedUpdatesPerSecond:N0} updates/s ({result.AchievedUpdatesPerSecond / result.NominalUpdatesPerSecond:P1} of nominal)"); + builder.AppendLine(culture, $"updates offered {result.SteadyStateEmittedTagUpdates:N0}"); + builder.AppendLine(culture, $"driver slice overrun {result.DriverLagSeconds:F1}s cumulative"); + builder.AppendLine(culture, $"skipped (no callback) {result.DriverSkippedNoCallback:N0}"); + builder.AppendLine(); + + builder.AppendLine("-- tag update latency (DCL boundary -> stream subscriber) --"); + AppendLatency(builder, culture, result.TagUpdateLatency); + builder.AppendLine(); + + builder.AppendLine("-- live stream subscribers (steady-state window) --"); + builder.AppendLine(culture, $"events delivered {result.StreamProbeReceived:N0}"); + builder.AppendLine(culture, $"events dropped {result.StreamProbeDropped:N0}"); + builder.AppendLine(); + + builder.AppendLine("-- health report delivery --"); + AppendLatency(builder, culture, result.HealthReportLatency); + builder.AppendLine(culture, $"reports ingested {result.HealthReportsDelivered:N0}"); + builder.AppendLine(culture, $"sites tracked centrally {result.SitesTrackedByAggregator}"); + builder.AppendLine(); + + builder.AppendLine("-- debug view snapshot (under load) --"); + AppendLatency(builder, culture, result.DebugSnapshotLatency); + builder.AppendLine(culture, $"completed / timed out {result.DebugSnapshotsCompleted:N0} / {result.DebugSnapshotTimeouts:N0}"); + builder.AppendLine(); + + if (result.SteadyStateResources is { } steady) + { + builder.AppendLine("-- resources (steady-state window) --"); + builder.AppendLine(culture, $"window {steady.DurationSeconds:F0}s over {steady.SampleCount} samples"); + builder.AppendLine(culture, $"working set {steady.WorkingSetStartMb:F0} -> {steady.WorkingSetEndMb:F0} MB (peak {steady.WorkingSetPeakMb:F0} MB)"); + builder.AppendLine(culture, $"working set slope {steady.WorkingSetSlopeMbPerMinute:F2} MB/min"); + builder.AppendLine(culture, $"managed heap {steady.ManagedHeapStartMb:F0} -> {steady.ManagedHeapEndMb:F0} MB (peak {steady.ManagedHeapPeakMb:F0} MB)"); + builder.AppendLine(culture, $"managed heap slope {steady.ManagedHeapSlopeMbPerMinute:F2} MB/min"); + builder.AppendLine(culture, $"cpu mean / peak {steady.MeanCpuPercentOfOneCore:F0}% / {steady.PeakCpuPercentOfOneCore:F0}% of one core ({steady.MeanCpuPercentOfOneCore / result.Environment.ProcessorCount:F1}% of the box)"); + builder.AppendLine(culture, $"threads / gen2 GCs {steady.MeanThreadCount:F0} / {steady.Gen2Collections}"); + builder.AppendLine(); + } + + if (result.StoreAndForwardDrain is { } drain) + { + builder.AppendLine("-- store-and-forward (register row 50) --"); + builder.AppendLine(culture, $"messages {drain.MessageCount:N0}"); + builder.AppendLine(culture, $"concurrent buffering {drain.EnqueuePerSecond:N0} msg/s ({drain.EnqueueSeconds:F1}s)"); + builder.AppendLine(culture, $"retry wait before drain {drain.TimeToFirstDeliverySeconds:F1}s (DefaultRetryInterval)"); + builder.AppendLine(culture, $"drain throughput {drain.DrainPerSecond:N0} msg/s (active drain {drain.DrainSeconds - drain.TimeToFirstDeliverySeconds:F2}s)"); + builder.AppendLine(culture, $"residual depth {drain.ResidualDepth}"); + builder.AppendLine(); + } + + if (result.SlowSubscriber is { } slow) + { + builder.AppendLine("-- slow-subscriber isolation (register row 50) --"); + builder.AppendLine(culture, $"published {slow.PublishedEvents:N0} at {slow.PublishPerSecond:N0}/s"); + builder.AppendLine(culture, $"healthy min delivery {slow.HealthyMinDeliveryRatio:P2}"); + builder.AppendLine(culture, $"stalled delivery {slow.SlowDeliveryRatio:P2}"); + foreach (var outcome in slow.Outcomes) + { + builder.AppendLine(culture, + $" {outcome.Name,-28} {(outcome.IsSlow ? "STALLED" : "healthy"),-8} " + + $"recv {outcome.Received,8:N0} dropped {outcome.Dropped,8:N0} ratio {outcome.DeliveryRatio:P2}"); + } + + builder.AppendLine(); + } + + return builder.ToString(); + } + + private static void AppendLatency(StringBuilder builder, CultureInfo culture, Metrics.LatencySnapshot snapshot) + { + builder.AppendLine(culture, + $"samples {snapshot.Count:N0} mean {snapshot.MeanMs:F2}ms p50 {snapshot.P50Ms:F2}ms " + + $"p95 {snapshot.P95Ms:F2}ms p99 {snapshot.P99Ms:F2}ms p99.9 {snapshot.P999Ms:F2}ms max {snapshot.MaxMs:F2}ms"); + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/Scenarios/ObservabilityProbes.cs b/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/Scenarios/ObservabilityProbes.cs new file mode 100644 index 00000000..16af25fe --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/Scenarios/ObservabilityProbes.cs @@ -0,0 +1,160 @@ +using System.Diagnostics; +using Akka.Actor; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.DebugView; +using ZB.MOM.WW.ScadaBridge.HealthMonitoring; +using ZB.MOM.WW.ScadaBridge.LoadHarness.Metrics; + +namespace ZB.MOM.WW.ScadaBridge.LoadHarness.Scenarios; + +/// +/// The two "does observability still work at scale?" probes WP-4's test protocol +/// names — health report delivery timing and debug view latency — run continuously +/// alongside the sustained load rather than after it, so both are measured against a +/// site that is actually busy. +/// +public sealed class ObservabilityProbes : IAsyncDisposable +{ + private readonly CancellationTokenSource _cts = new(); + private readonly List _tasks = new(); + + /// + /// End-to-end health report latency: SiteHealthCollector.CollectReport plus + /// the transport hop plus CentralHealthAggregator.ProcessReport. The + /// interesting term at scale is CollectReport, which materializes the + /// per-connection dictionaries for a site carrying 37,500 subscriptions. + /// + public LatencyHistogram HealthReportLatency { get; } = new(); + + /// + /// Debug view snapshot round-trip: an Ask of DebugSnapshotRequest to + /// a live Instance Actor. This lands the request in the mailbox of an actor that + /// is concurrently ingesting tag updates, so the measured time includes real + /// queueing behind production traffic — which is the whole point of measuring it + /// under load. + /// + public LatencyHistogram DebugSnapshotLatency { get; } = new(); + + /// Health reports successfully ingested by the central aggregator. + public long HealthReportsDelivered => Interlocked.Read(ref _healthReports); + + /// Debug snapshots that completed within the ask timeout. + public long DebugSnapshotsCompleted => Interlocked.Read(ref _debugSnapshots); + + /// Debug snapshot asks that timed out. + public long DebugSnapshotTimeouts => Interlocked.Read(ref _debugTimeouts); + + private long _healthReports; + private long _debugSnapshots; + private long _debugTimeouts; + + /// + /// Starts both probes. + /// + /// Sites to probe. + /// The real central aggregator receiving the reports. + /// Health report cadence (production default 30 s). + /// How often to take a debug snapshot. + /// The running probes. + public static ObservabilityProbes Start( + IReadOnlyList sites, + CentralHealthAggregator aggregator, + TimeSpan reportInterval, + TimeSpan debugProbeInterval) + { + var probes = new ObservabilityProbes(); + probes._tasks.Add(Task.Run(() => probes.HealthLoopAsync(sites, aggregator, reportInterval, probes._cts.Token))); + probes._tasks.Add(Task.Run(() => probes.DebugLoopAsync(sites, debugProbeInterval, probes._cts.Token))); + return probes; + } + + private async Task HealthLoopAsync( + IReadOnlyList sites, + CentralHealthAggregator aggregator, + TimeSpan interval, + CancellationToken cancellationToken) + { + using var timer = new PeriodicTimer(interval); + try + { + while (await timer.WaitForNextTickAsync(cancellationToken)) + { + foreach (var site in sites) + { + var watch = Stopwatch.StartNew(); + var report = site.HealthCollector.CollectReport(site.SiteId); + aggregator.ProcessReport(report); + watch.Stop(); + + HealthReportLatency.Record(watch.Elapsed); + Interlocked.Increment(ref _healthReports); + } + } + } + catch (OperationCanceledException) + { + // Normal teardown. + } + } + + private async Task DebugLoopAsync( + IReadOnlyList sites, + TimeSpan interval, + CancellationToken cancellationToken) + { + var random = new Random(20260815); + using var timer = new PeriodicTimer(interval); + try + { + while (await timer.WaitForNextTickAsync(cancellationToken)) + { + var site = sites[random.Next(sites.Count)]; + if (site.InstanceActors.Count == 0) + continue; + + var index = random.Next(site.InstanceActors.Count); + var actor = site.InstanceActors[index]; + var request = new DebugSnapshotRequest(site.InstanceName(index), Guid.NewGuid().ToString("N")); + + var watch = Stopwatch.StartNew(); + try + { + await actor.Ask(request, TimeSpan.FromSeconds(10), cancellationToken); + watch.Stop(); + DebugSnapshotLatency.Record(watch.Elapsed); + Interlocked.Increment(ref _debugSnapshots); + } + catch (AskTimeoutException) + { + Interlocked.Increment(ref _debugTimeouts); + } + } + } + catch (OperationCanceledException) + { + // Normal teardown. + } + } + + private int _disposed; + + /// + public async ValueTask DisposeAsync() + { + // Idempotent: the orchestrator stops the probes early (so the register row 50 + // scenarios do not compete with them) and again in its finally block. + if (Interlocked.Exchange(ref _disposed, 1) != 0) + return; + + await _cts.CancelAsync(); + try + { + await Task.WhenAll(_tasks); + } + catch (OperationCanceledException) + { + // Expected. + } + + _cts.Dispose(); + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/Scenarios/SlowSubscriberScenario.cs b/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/Scenarios/SlowSubscriberScenario.cs new file mode 100644 index 00000000..b302ae7c --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/Scenarios/SlowSubscriberScenario.cs @@ -0,0 +1,153 @@ +using System.Diagnostics; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.Streaming; +using ZB.MOM.WW.ScadaBridge.LoadHarness.Probes; + +namespace ZB.MOM.WW.ScadaBridge.LoadHarness.Scenarios; + +/// Per-subscriber outcome from the slow-subscriber isolation probe. +/// Probe name. +/// Whether this probe's reader was deliberately stalled. +/// Events the reader drained. +/// Events evicted by this probe's bounded DropOldest channel. +/// Received / published, before accounting for the site-stream buffer. +public sealed record SubscriberOutcome( + string Name, + bool IsSlow, + long Received, + long Dropped, + double DeliveryRatio); + +/// Result of the slow-subscriber isolation measurement. +/// Events published to the site stream during the probe. +/// Wall time the publisher took. +/// Publish throughput observed by the producer. +/// Per-subscriber outcomes. +/// Worst delivery ratio among the healthy subscribers. +/// Delivery ratio of the stalled subscriber. +public sealed record SlowSubscriberResult( + int PublishedEvents, + double PublishSeconds, + double PublishPerSecond, + IReadOnlyList Outcomes, + double HealthyMinDeliveryRatio, + double SlowDeliveryRatio); + +/// +/// Register row 50, second half: what does a slow or stalled gRPC subscriber do to +/// per-subscriber buffering when several subscribers are attached? +/// +/// +/// Several probes are attached to the SAME instance so every one of them is offered +/// exactly the same event sequence — otherwise a difference in delivery could be a +/// difference in offered load rather than a backpressure effect. One probe's reader +/// is then stalled (a large per-event delay, standing in for a wedged client or a +/// dead WAN link) while the rest read as fast as they can. Events are published +/// through the real SiteStreamManager. +/// +/// +/// The question the numbers answer: does the stalled subscriber's backlog propagate +/// upstream — evicting events for the healthy subscribers or slowing the publisher — +/// or is it confined to its own Buffer(DropHead) stage and its own bounded +/// DropOldest channel? The design intends the latter; this measures it. +/// +/// +public static class SlowSubscriberScenario +{ + /// Per-event reader delay applied to the stalled subscriber. + public const int SlowReaderDelayMicroseconds = 50_000; + + /// + /// Publish rate for the probe. Deliberately paced rather than a tight burst: the + /// publish source is a single Source.ActorRef(StreamBufferSize, DropHead) + /// SHARED by every attribute subscriber, so an unpaced burst saturates that shared + /// stage and every subscriber loses events for a reason that has nothing to do + /// with the slow one. Pacing below the shared stage's capacity is what isolates + /// the variable under test. + /// + public const int PublishRatePerSecond = 2_000; + + /// Runs the isolation probe on a dedicated site. + /// Site whose stream manager is used. + /// Total subscribers to attach (one of them is stalled). + /// Events to publish. + /// Cancels the measurement. + /// The measured result. + public static async Task RunAsync( + SiteRuntimeFixture site, + int subscriberCount, + int eventCount, + CancellationToken cancellationToken) + { + var instanceName = site.InstanceName(0); + var probes = new List(subscriberCount); + + try + { + for (var i = 0; i < subscriberCount; i++) + { + var probe = StreamSubscriberProbe.Attach( + site.System, site.StreamManager, instanceName, + $"{site.SiteId}-slowprobe-{i:D2}", latency: null); + + // Probe 0 is the pathological one. + if (i == 0) + probe.ReaderDelayMicroseconds = SlowReaderDelayMicroseconds; + + probes.Add(probe); + } + + // Let every subscription's stream graph finish materializing before the burst. + await Task.Delay(TimeSpan.FromSeconds(2), cancellationToken); + + var watch = Stopwatch.StartNew(); + const int sliceMilliseconds = 50; + var perSlice = Math.Max(1, PublishRatePerSecond * sliceMilliseconds / 1000); + var published = 0; + while (published < eventCount && !cancellationToken.IsCancellationRequested) + { + var sliceStart = Stopwatch.GetTimestamp(); + var end = Math.Min(published + perSlice, eventCount); + for (var i = published; i < end; i++) + { + site.StreamManager.PublishAttributeValueChanged(new AttributeValueChanged( + instanceName, "Tag000", "Tag000", i, "Good", DateTimeOffset.UtcNow)); + } + + published = end; + + var elapsedMs = (Stopwatch.GetTimestamp() - sliceStart) * 1000.0 / Stopwatch.Frequency; + if (elapsedMs < sliceMilliseconds) + await Task.Delay(TimeSpan.FromMilliseconds(sliceMilliseconds - elapsedMs), cancellationToken); + } + + watch.Stop(); + + // Give the healthy readers time to finish; the stalled one will not. + await Task.Delay(TimeSpan.FromSeconds(20), cancellationToken); + + var outcomes = probes + .Select((p, i) => new SubscriberOutcome( + p.Name, + IsSlow: i == 0, + p.ReceivedEvents, + p.DroppedEvents, + p.ReceivedEvents / (double)eventCount)) + .ToList(); + + var healthy = outcomes.Where(o => !o.IsSlow).ToList(); + + return new SlowSubscriberResult( + PublishedEvents: eventCount, + PublishSeconds: watch.Elapsed.TotalSeconds, + PublishPerSecond: eventCount / Math.Max(0.001, watch.Elapsed.TotalSeconds), + Outcomes: outcomes, + HealthyMinDeliveryRatio: healthy.Count == 0 ? 0 : healthy.Min(o => o.DeliveryRatio), + SlowDeliveryRatio: outcomes[0].DeliveryRatio); + } + finally + { + foreach (var probe in probes) + await probe.DisposeAsync(); + } + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/Scenarios/StoreAndForwardDrainScenario.cs b/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/Scenarios/StoreAndForwardDrainScenario.cs new file mode 100644 index 00000000..5370f281 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/Scenarios/StoreAndForwardDrainScenario.cs @@ -0,0 +1,154 @@ +using System.Diagnostics; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; + +namespace ZB.MOM.WW.ScadaBridge.LoadHarness.Scenarios; + +/// Result of one store-and-forward drain measurement. +/// Site the buffer belonged to. +/// Messages buffered before the drain began. +/// Wall time to buffer them (concurrent, many origin instances). +/// Buffering throughput. +/// +/// Wall time from the first sweep to the first successful delivery. With +/// attemptImmediateDelivery: false the engine stamps LastAttemptAt, so the +/// row is not due until one DefaultRetryInterval (30 s) has passed — this is the +/// configured retry latency, not drain slowness, and is reported separately for that reason. +/// +/// Wall time from the first sweep to an empty buffer (includes the retry wait). +/// +/// Drain throughput measured from the FIRST delivery to an empty buffer — the engine's +/// actual capacity, and the headline number for register row 50. +/// +/// Buffer depth left when the measurement stopped (0 = fully drained). +/// Delivered-count samples during the drain, so a steady rate can be told from a stall-then-burst. +public sealed record StoreAndForwardDrainResult( + string SiteId, + int MessageCount, + double EnqueueSeconds, + double EnqueuePerSecond, + double TimeToFirstDeliverySeconds, + double DrainSeconds, + double DrainPerSecond, + int ResidualDepth, + IReadOnlyList Progress); + +/// One observation during the drain. +/// Seconds since the drain began. +/// Cumulative successful deliveries. +/// Remaining buffer depth. +public sealed record DrainProgressSample(double ElapsedSeconds, long Delivered, int Depth); + +/// +/// Measures store-and-forward buffering and drain throughput (deferred-work register +/// row 50, first half) using the real StoreAndForwardService, the real +/// StoreAndForwardStorage and the real SQLite file — only the delivery target +/// is a counting stub, because what is being measured is the site-local buffer's +/// throughput, not a remote endpoint's. +/// +/// +/// Phase 1 buffers messages with +/// attemptImmediateDelivery: false, spread across many origin instance names +/// and issued from many concurrent tasks — the "concurrent buffering from multiple +/// instances" WP-4 asks about ([xc-7]). Phase 2 registers a delivery handler +/// that always succeeds and drives sweeps to completion, timing the drain. +/// +/// +/// The sweep is driven explicitly rather than waiting on the 10 s +/// RetryTimerInterval so the number reported is the engine's drain capacity, +/// not its polling cadence. The per-sweep batch is SweepBatchLimit (500) with +/// SweepTargetParallelism (4) lanes, both at their production defaults. +/// +/// +public static class StoreAndForwardDrainScenario +{ + /// Runs the drain measurement against one site's real S&F engine. + /// The site whose store-and-forward engine is exercised. + /// Messages to buffer. + /// Concurrent enqueue tasks (distinct origin instances). + /// Cancels the measurement. + /// The measured result. + public static async Task RunAsync( + SiteRuntimeFixture site, + int messageCount, + int concurrency, + CancellationToken cancellationToken) + { + var service = site.StoreAndForward; + var payload = $"{{\"site\":\"{site.SiteId}\",\"body\":\"{new string('x', 256)}\"}}"; + + // Phase 1 — concurrent buffering from many instances, no delivery attempted. + var enqueueWatch = Stopwatch.StartNew(); + var perTask = messageCount / concurrency; + var enqueueTasks = new List(concurrency); + for (var t = 0; t < concurrency; t++) + { + var taskIndex = t; + enqueueTasks.Add(Task.Run(async () => + { + for (var i = 0; i < perTask; i++) + { + await service.EnqueueAsync( + StoreAndForwardCategory.ExternalSystem, + target: $"load-target-{taskIndex % 4}", + payloadJson: payload, + originInstanceName: site.InstanceName(taskIndex), + attemptImmediateDelivery: false); + } + }, cancellationToken)); + } + + await Task.WhenAll(enqueueTasks); + enqueueWatch.Stop(); + var buffered = perTask * concurrency; + + // Phase 2 — a delivery target that always succeeds; time the drain to empty. + var delivered = 0L; + service.RegisterDeliveryHandler(StoreAndForwardCategory.ExternalSystem, _ => + { + Interlocked.Increment(ref delivered); + return Task.FromResult(true); + }); + + // Sweeps are driven explicitly rather than waiting on the 10 s RetryTimerInterval: + // the number wanted is the engine's drain CAPACITY, not its polling cadence. One + // sweep moves at most SweepBatchLimit (500) messages, so a large backlog needs + // many, and the progress series below is what distinguishes a genuinely slow + // drain from an artefact of this polling loop. + var drainWatch = Stopwatch.StartNew(); + var deadline = DateTimeOffset.UtcNow.AddMinutes(10); + var progress = new List(); + int depth; + while (true) + { + service.TriggerSweep(); + await Task.Delay(TimeSpan.FromMilliseconds(100), cancellationToken); + + var depths = await service.GetBufferDepthAsync(); + depth = depths.Values.Sum(); + progress.Add(new DrainProgressSample( + drainWatch.Elapsed.TotalSeconds, Interlocked.Read(ref delivered), depth)); + + if (depth == 0 || DateTimeOffset.UtcNow > deadline) + break; + } + + drainWatch.Stop(); + + // Split the retry wait from the drain: the first sample with a non-zero delivered + // count marks the moment the backlog actually became due. + var firstDelivery = progress.FirstOrDefault(s => s.Delivered > 0); + var timeToFirstDelivery = firstDelivery?.ElapsedSeconds ?? drainWatch.Elapsed.TotalSeconds; + var activeDrainSeconds = Math.Max(0.001, drainWatch.Elapsed.TotalSeconds - timeToFirstDelivery); + + return new StoreAndForwardDrainResult( + SiteId: site.SiteId, + MessageCount: buffered, + EnqueueSeconds: enqueueWatch.Elapsed.TotalSeconds, + EnqueuePerSecond: buffered / Math.Max(0.001, enqueueWatch.Elapsed.TotalSeconds), + TimeToFirstDeliverySeconds: timeToFirstDelivery, + DrainSeconds: drainWatch.Elapsed.TotalSeconds, + DrainPerSecond: Interlocked.Read(ref delivered) / activeDrainSeconds, + ResidualDepth: depth, + Progress: progress); + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/SimulatedDataConnection.cs b/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/SimulatedDataConnection.cs new file mode 100644 index 00000000..d08bdfeb --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/SimulatedDataConnection.cs @@ -0,0 +1,153 @@ +using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Protocol; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; + +namespace ZB.MOM.WW.ScadaBridge.LoadHarness; + +/// +/// In-process stand-in for an OPC UA / MxGateway server, registered on the REAL +/// DataConnectionFactory under the protocol so the +/// whole Data Connection Layer above it — DataConnectionManagerActor, +/// DataConnectionActor, its _instancesByTag fan-out and the +/// TagValueUpdate hand-off to Instance Actors — runs unmodified. +/// +/// +/// It implements because the batch path is +/// the one a real site takes at this scale; the per-tag fallback would make +/// subscribe setup, not steady-state throughput, the thing being measured. +/// +/// +/// Why faking here does not invalidate the measurement. Everything this class +/// replaces is on the far side of the process boundary: socket I/O, the OPC UA SDK's +/// own session/subscription machinery, and the device. The system under test — the +/// actor hierarchy, the site stream, store-and-forward, health, and the streaming +/// relay — begins at the invocation, which is +/// exactly where the real adapter hands off. Driving 375,000 genuine monitored items +/// would measure the OPC UA stack, not ScadaBridge. +/// +/// +public sealed class SimulatedDataConnection : IDataConnection, IBatchSubscribableConnection +{ + /// Protocol discriminator this adapter registers under on the factory. + public const string ProtocolName = "LoadSim"; + + private SubscriptionCallback? _callback; + private int _subscriptionCounter; + + /// + /// The data connection name this adapter was created for, taken from the connection + /// details. The DCL factory creates adapters as its manager actor processes the + /// CreateConnectionCommands, so creation order is not connection order — the driver + /// resolves an adapter by name rather than by index. + /// + public string ConnectionName { get; private set; } = string.Empty; + + /// Key under which the connection name travels in the connection details. + public const string ConnectionNameKey = "connectionName"; + + /// + public ConnectionHealth Status { get; private set; } = ConnectionHealth.Disconnected; + + /// + public event Action? Disconnected; + + /// + /// The callback captured at subscribe time. The tag driver invokes this to inject + /// a value change, mirroring what the OPC UA SDK's notification thread does. + /// Null until the site's Instance Actors have subscribed. + /// + public SubscriptionCallback? ValueCallback => _callback; + + /// Number of tag paths this connection has accepted subscriptions for. + public int SubscribedTagCount => _subscriptionCounter; + + /// + public Task ConnectAsync(IDictionary connectionDetails, CancellationToken cancellationToken = default) + { + if (connectionDetails.TryGetValue(ConnectionNameKey, out var name)) + ConnectionName = name; + + Status = ConnectionHealth.Connected; + return Task.CompletedTask; + } + + /// + public Task DisconnectAsync(CancellationToken cancellationToken = default) + { + Status = ConnectionHealth.Disconnected; + Disconnected?.Invoke(); + return Task.CompletedTask; + } + + /// + public Task SubscribeAsync(string tagPath, SubscriptionCallback callback, CancellationToken cancellationToken = default) + { + _callback = callback; + return Task.FromResult($"sub-{Interlocked.Increment(ref _subscriptionCounter)}"); + } + + /// + public Task> SubscribeBatchAsync( + IReadOnlyList tagPaths, + SubscriptionCallback callback, + CancellationToken cancellationToken = default) + { + _callback = callback; + var results = new List(tagPaths.Count); + foreach (var path in tagPaths) + { + results.Add(new TagSubscribeResult( + path, true, $"sub-{Interlocked.Increment(ref _subscriptionCounter)}", null)); + } + + return Task.FromResult>(results); + } + + /// + public Task UnsubscribeAsync(string subscriptionId, CancellationToken cancellationToken = default) + => Task.CompletedTask; + + /// + public Task UnsubscribeBatchAsync(IReadOnlyList subscriptionIds, CancellationToken cancellationToken = default) + => Task.CompletedTask; + + /// + public Task ReadAsync(string tagPath, CancellationToken cancellationToken = default) + => Task.FromResult(new ReadResult(true, new TagValue(0d, QualityCode.Good, DateTimeOffset.UtcNow), null)); + + /// + public Task> ReadBatchAsync( + IEnumerable tagPaths, CancellationToken cancellationToken = default) + { + var now = DateTimeOffset.UtcNow; + var results = new Dictionary(); + foreach (var path in tagPaths) + results[path] = new ReadResult(true, new TagValue(0d, QualityCode.Good, now), null); + + return Task.FromResult>(results); + } + + /// + public Task WriteAsync(string tagPath, object? value, CancellationToken cancellationToken = default) + => Task.FromResult(new WriteResult(true, null)); + + /// + public Task> WriteBatchAsync( + IDictionary values, CancellationToken cancellationToken = default) + { + var results = new Dictionary(); + foreach (var key in values.Keys) + results[key] = new WriteResult(true, null); + + return Task.FromResult>(results); + } + + /// + public Task WriteBatchAndWaitAsync( + IDictionary values, string flagPath, object? flagValue, + string responsePath, object? responseValue, TimeSpan timeout, + CancellationToken cancellationToken = default) + => Task.FromResult(true); + + /// + public ValueTask DisposeAsync() => ValueTask.CompletedTask; +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/SiteRuntimeFixture.cs b/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/SiteRuntimeFixture.cs new file mode 100644 index 00000000..649c0c9c --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/SiteRuntimeFixture.cs @@ -0,0 +1,371 @@ +using System.Diagnostics; +using System.Text.Json; +using Akka.Actor; +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging.Abstractions; +using ZB.MOM.WW.LocalDb; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.DataConnection; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Flattening; +using ZB.MOM.WW.ScadaBridge.DataConnectionLayer; +using ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Actors; +using ZB.MOM.WW.ScadaBridge.HealthMonitoring; +using ZB.MOM.WW.ScadaBridge.LoadHarness.Metrics; +using ZB.MOM.WW.ScadaBridge.LoadHarness.Probes; +using ZB.MOM.WW.ScadaBridge.SiteRuntime; +using ZB.MOM.WW.ScadaBridge.SiteRuntime.Actors; +using ZB.MOM.WW.ScadaBridge.SiteRuntime.Persistence; +using ZB.MOM.WW.ScadaBridge.SiteRuntime.Scripts; +using ZB.MOM.WW.ScadaBridge.SiteRuntime.Streaming; +using ZB.MOM.WW.ScadaBridge.StoreAndForward; + +namespace ZB.MOM.WW.ScadaBridge.LoadHarness; + +/// +/// One simulated site: its own , its own LocalDb SQLite +/// file, a real Data Connection Layer over +/// adapters, real Instance Actors, a real , real +/// store-and-forward, and a real . +/// +/// +/// The sites are separate, non-clustered ActorSystems rather than 10 real two-node +/// Akka clusters. Cluster membership, singleton placement and failover timing are +/// already measured on a real two-node rig by +/// PerformanceTests/Failover/FailoverTimingTests.cs; what WP-4 asks about is +/// the load-bearing hierarchy under each singleton, which is what this builds. +/// +/// +public sealed class SiteRuntimeFixture : IAsyncDisposable +{ + private readonly HarnessConfig _config; + private readonly List _connections; + private readonly ServiceProvider _localDbProvider; + private readonly ILocalDb _localDb; + private readonly List _instanceActors = new(); + private readonly List _probes = new(); + private readonly string _dataDirectory; + + /// Site identifier, e.g. site-01. + public string SiteId { get; } + + /// This site's actor system. + public ActorSystem System { get; } + + /// The real site-wide broadcast stream. + public SiteStreamManager StreamManager { get; } + + /// The real site health collector feeding the 30 s report. + public SiteHealthCollector HealthCollector { get; } + + /// The real store-and-forward engine for this site. + public StoreAndForwardService StoreAndForward { get; } + + /// The real DCL manager actor. + public IActorRef DataConnectionManager { get; } + + /// The simulated adapters, one per data connection, in creation order. + public IReadOnlyList Connections + { + get { lock (_connections) return _connections.ToList(); } + } + + /// The live stream subscribers attached to this site. + public IReadOnlyList Probes => _probes; + + /// Instance Actors created on this site. + public IReadOnlyList InstanceActors => _instanceActors; + + /// Tag paths per connection index, in the order they were assigned. + public IReadOnlyList> TagPathsByConnection { get; } + + /// Wall-clock time the instance ramp took, measured by . + public TimeSpan InstanceRampDuration { get; private set; } + + private SiteRuntimeFixture( + string siteId, + HarnessConfig config, + string dataDirectory, + ServiceProvider localDbProvider, + ILocalDb localDb, + ActorSystem system, + SiteStreamManager streamManager, + SiteHealthCollector healthCollector, + StoreAndForwardService storeAndForward, + IActorRef dataConnectionManager, + SiteStorageService storage, + ScriptCompilationService compilationService, + SharedScriptLibrary sharedScriptLibrary, + SiteRuntimeOptions siteOptions, + List> tagPathsByConnection, + List connections) + { + _connections = connections; + SiteId = siteId; + _config = config; + _dataDirectory = dataDirectory; + _localDbProvider = localDbProvider; + _localDb = localDb; + System = system; + StreamManager = streamManager; + HealthCollector = healthCollector; + StoreAndForward = storeAndForward; + DataConnectionManager = dataConnectionManager; + Storage = storage; + CompilationService = compilationService; + SharedScriptLibrary = sharedScriptLibrary; + SiteOptions = siteOptions; + TagPathsByConnection = tagPathsByConnection; + } + + private SiteStorageService Storage { get; } + private ScriptCompilationService CompilationService { get; } + private SharedScriptLibrary SharedScriptLibrary { get; } + private SiteRuntimeOptions SiteOptions { get; } + + /// + /// Builds a site: LocalDb file, actor system, DCL with its simulated connections, + /// stream manager, health collector and store-and-forward engine. Instance Actors + /// are created separately by so the deployment + /// ramp can be timed on its own. + /// + /// Zero-based site index. + /// Harness configuration. + /// Directory under which this site's SQLite files live. + /// Number of data connections to spread the site's tags across. + /// The started fixture. + public static async Task CreateAsync( + int siteIndex, HarnessConfig config, string rootDataDirectory, int connectionsPerSite) + { + var siteId = $"site-{siteIndex + 1:D2}"; + var dataDirectory = Path.Combine(rootDataDirectory, siteId); + Directory.CreateDirectory(dataDirectory); + + var configuration = new ConfigurationBuilder() + .AddInMemoryCollection(new Dictionary + { + ["LocalDb:Path"] = Path.Combine(dataDirectory, "site-localdb.db"), + }) + .Build(); + + var localDbProvider = new ServiceCollection() + .AddZbLocalDb(configuration) + .BuildServiceProvider(); + var localDb = localDbProvider.GetRequiredService(); + + var storage = new SiteStorageService(localDb, NullLogger.Instance); + await storage.InitializeAsync(); + + var compilationService = new ScriptCompilationService(NullLogger.Instance); + var sharedScriptLibrary = new SharedScriptLibrary(compilationService, NullLogger.Instance); + + // Production defaults throughout — the point of the run is to measure the + // shipped configuration, so nothing here is tuned for the harness. + var siteOptions = new SiteRuntimeOptions(); + + // WARNING-level logging. At 37,500 updates/s per site, Akka's INFO output would + // itself become a measured load; and the InstanceActorInitialized dead letters are + // a harness artifact (see StartInstancesAsync) rather than a real condition. + var system = ActorSystem.Create($"loadharness-{siteId}", Akka.Configuration.ConfigurationFactory.ParseString( + "akka.loglevel = WARNING\nakka.stdout-loglevel = WARNING\nakka.log-dead-letters = 0\nakka.log-dead-letters-during-shutdown = off")); + + var streamManager = new SiteStreamManager(siteOptions, NullLogger.Instance); + streamManager.Initialize(system); + + var healthCollector = new SiteHealthCollector(); + healthCollector.SetActiveNode(true); + healthCollector.SetNodeHostname($"{siteId}-node-a"); + + var sfStorage = new StoreAndForwardStorage(localDb, NullLogger.Instance); + var storeAndForward = new StoreAndForwardService( + sfStorage, + new StoreAndForwardOptions(), + NullLogger.Instance, + siteId: siteId); + await storeAndForward.StartAsync(); + + // Real DCL, with the simulated adapter registered on the real factory via its + // documented RegisterAdapter extension point. + var loggerFactory = NullLoggerFactory.Instance; + var factory = new DataConnectionFactory(loggerFactory); + var connections = new List(); + factory.RegisterAdapter(SimulatedDataConnection.ProtocolName, _ => + { + var connection = new SimulatedDataConnection(); + lock (connections) connections.Add(connection); + return connection; + }); + + var dclManager = system.ActorOf( + Props.Create(() => new DataConnectionManagerActor( + factory, new DataConnectionOptions(), healthCollector, null, null)), + "data-connection-manager"); + + for (var c = 0; c < connectionsPerSite; c++) + { + dclManager.Tell(new CreateConnectionCommand( + ConnectionName: ConnectionName(c), + ProtocolType: SimulatedDataConnection.ProtocolName, + PrimaryConnectionDetails: new Dictionary + { + ["endpoint"] = $"sim://{siteId}/{c}", + [SimulatedDataConnection.ConnectionNameKey] = ConnectionName(c), + })); + } + + var tagPathsByConnection = new List>(); + for (var c = 0; c < connectionsPerSite; c++) + tagPathsByConnection.Add(new List()); + + var fixture = new SiteRuntimeFixture( + siteId, config, dataDirectory, localDbProvider, localDb, system, streamManager, + healthCollector, storeAndForward, dclManager, storage, compilationService, + sharedScriptLibrary, siteOptions, tagPathsByConnection, connections); + + return fixture; + } + + /// Deterministic connection name for a connection index. + /// Zero-based connection index. + /// The connection name used in configs and DCL commands. + public static string ConnectionName(int connectionIndex) => $"sim-conn-{connectionIndex:D2}"; + + /// + /// Creates this site's Instance Actors in production-shaped staggered batches + /// ( / + /// ) and records how long the + /// ramp took. This is the harness's stand-in for "deployment of 500 instances to + /// a site" — it exercises the same per-instance construction, config + /// deserialization, override load and DCL subscribe that a real deploy triggers. + /// + /// Cancels the ramp. + /// A task that completes when every instance actor exists. + public async Task StartInstancesAsync(CancellationToken cancellationToken) + { + var started = Stopwatch.StartNew(); + var connectionCount = TagPathsByConnection.Count; + + for (var i = 0; i < _config.InstancesPerSite; i++) + { + var instanceName = InstanceName(i); + var connectionIndex = i % connectionCount; + var connectionName = ConnectionName(connectionIndex); + + var attributes = new List(_config.TagsPerInstance); + for (var t = 0; t < _config.TagsPerInstance; t++) + { + var tagPath = $"{instanceName}.Tag{t:D3}"; + TagPathsByConnection[connectionIndex].Add(tagPath); + attributes.Add(new ResolvedAttribute + { + CanonicalName = $"Tag{t:D3}", + DataType = "Double", + DataSourceReference = tagPath, + BoundDataConnectionId = connectionIndex + 1, + BoundDataConnectionName = connectionName, + BoundDataConnectionProtocol = SimulatedDataConnection.ProtocolName, + }); + } + + var configuration = new FlattenedConfiguration + { + InstanceUniqueName = instanceName, + TemplateId = 1, + SiteId = 1, + Attributes = attributes, + Connections = new Dictionary + { + [connectionName] = new() + { + Protocol = SimulatedDataConnection.ProtocolName, + ConfigurationJson = "{}", + }, + }, + }; + + var configJson = JsonSerializer.Serialize(configuration); + var actor = System.ActorOf( + Props.Create(() => new InstanceActor( + instanceName, configJson, Storage, CompilationService, SharedScriptLibrary, + StreamManager, SiteOptions, NullLogger.Instance, + DataConnectionManager, HealthCollector, null, null)), + instanceName); + _instanceActors.Add(actor); + + // Production staggered-startup pacing (SiteRuntimeOptions defaults): + // batches of StartupBatchSize separated by StartupBatchDelayMs, which is + // exactly what DeploymentManagerActor does on a real site start. + if ((i + 1) % SiteOptions.StartupBatchSize == 0) + await Task.Delay(SiteOptions.StartupBatchDelayMs, cancellationToken); + } + + started.Stop(); + InstanceRampDuration = started.Elapsed; + + HealthCollector.SetInstanceCounts( + _config.InstancesPerSite, _config.InstancesPerSite, 0); + for (var c = 0; c < connectionCount; c++) + { + HealthCollector.UpdateTagResolution( + ConnectionName(c), TagPathsByConnection[c].Count, TagPathsByConnection[c].Count); + } + } + + /// Deterministic instance unique name for an instance index. + /// Zero-based instance index. + /// The instance unique name. + public string InstanceName(int instanceIndex) => $"{SiteId}-inst-{instanceIndex:D4}"; + + /// + /// Attaches live stream subscribers, + /// each built from the production pieces the gRPC server uses: a real + /// writing into a bounded DropOldest channel + /// of the production capacity, subscribed through the real + /// . Only the socket writer is replaced — + /// by a reader task under the harness's control, which is what makes the + /// slow-subscriber scenario possible at all. + /// + /// + /// Attaching at least one subscriber is also load-bearing: + /// PublishAttributeValueChanged short-circuits at zero subscribers, so an + /// unsubscribed site would publish nothing and measure nothing. + /// + /// + /// Histogram that receives end-to-end tag update latencies. + public void AttachStreamProbes(LatencyHistogram latency) + { + var stride = Math.Max(1, _config.InstancesPerSite / Math.Max(1, _config.StreamProbesPerSite)); + for (var p = 0; p < _config.StreamProbesPerSite; p++) + { + var instanceIndex = Math.Min(p * stride, _config.InstancesPerSite - 1); + var instanceName = InstanceName(instanceIndex); + var probe = StreamSubscriberProbe.Attach( + System, StreamManager, instanceName, $"{SiteId}-probe-{p:D2}", latency); + _probes.Add(probe); + } + } + + /// Total tag paths registered across this site's connections. + public int TotalTagPaths => TagPathsByConnection.Sum(list => list.Count); + + /// + public async ValueTask DisposeAsync() + { + foreach (var probe in _probes) + await probe.DisposeAsync(); + + await StoreAndForward.StopAsync(); + await System.Terminate(); + await _localDbProvider.DisposeAsync(); + + try + { + if (Directory.Exists(_dataDirectory)) + Directory.Delete(_dataDirectory, recursive: true); + } + catch (IOException) + { + // Best-effort cleanup of a temp directory; a lingering WAL handle is not + // a harness failure and must not mask the measured result. + } + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/TagUpdateDriver.cs b/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/TagUpdateDriver.cs new file mode 100644 index 00000000..44b3332b --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/TagUpdateDriver.cs @@ -0,0 +1,175 @@ +using System.Diagnostics; +using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Protocol; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; + +namespace ZB.MOM.WW.ScadaBridge.LoadHarness; + +/// +/// Drives simulated tag value changes into a site's connections at a fixed nominal +/// rate, standing in for an OPC UA server's notification thread. +/// +/// +/// One driver task per data connection, because that mirrors production: each +/// DataConnectionActor has exactly one adapter feeding it, and every update +/// for that connection funnels through that actor's single mailbox. Sharding the +/// emitters differently would hide the per-connection-actor serialization point, +/// which is one of the things WP-4 needs to characterize. +/// +/// +/// The emitter walks the connection's tag list in slices sized so that one full pass +/// takes , then sleeps out the remainder +/// of each slice's budget. If a slice overruns its budget the driver does NOT try to +/// catch up — it records the shortfall in so the results +/// can say honestly whether the offered load was actually delivered. +/// +/// +public sealed class TagUpdateDriver : IAsyncDisposable +{ + private const int SlicesPerPeriod = 20; + + private readonly List _tasks = new(); + private readonly CancellationTokenSource _cts = new(); + private long _emitted; + private long _skippedNoCallback; + private long _lagTicks; + + /// Total tag value changes handed to adapter callbacks. + public long EmittedCount => Interlocked.Read(ref _emitted); + + /// + /// Emissions skipped because no Instance Actor had subscribed to that connection + /// yet (the adapter callback is captured at subscribe time). Non-zero only during + /// the ramp; a non-zero value in the steady window would mean lost offered load. + /// + public long SkippedNoCallback => Interlocked.Read(ref _skippedNoCallback); + + /// + /// Cumulative seconds by which emit slices overran their time budget, summed + /// across driver tasks. Large values mean the harness itself could not offer the + /// nominal rate and the measured throughput is driver-bound, not system-bound. + /// + public double EmitLagSeconds => Interlocked.Read(ref _lagTicks) / (double)Stopwatch.Frequency; + + /// + /// Starts one emitter task per connection across every site. + /// + /// The sites to drive. + /// Harness configuration supplying the update period. + /// The running driver. + public static TagUpdateDriver Start(IReadOnlyList sites, HarnessConfig config) + { + var driver = new TagUpdateDriver(); + foreach (var site in sites) + { + for (var c = 0; c < site.TagPathsByConnection.Count; c++) + { + var connectionIndex = c; + var tagPaths = site.TagPathsByConnection[connectionIndex].ToArray(); + driver._tasks.Add(Task.Run(() => driver.EmitLoopAsync( + site, connectionIndex, tagPaths, config.TagUpdatePeriod, driver._cts.Token))); + } + } + + return driver; + } + + private async Task EmitLoopAsync( + SiteRuntimeFixture site, + int connectionIndex, + string[] tagPaths, + TimeSpan period, + CancellationToken cancellationToken) + { + if (tagPaths.Length == 0) + return; + + var sliceBudget = period / SlicesPerPeriod; + var sliceSize = Math.Max(1, (int)Math.Ceiling(tagPaths.Length / (double)SlicesPerPeriod)); + var cursor = 0; + var sequence = 0d; + + while (!cancellationToken.IsCancellationRequested) + { + var sliceStart = Stopwatch.GetTimestamp(); + var callback = ResolveCallback(site, connectionIndex); + + if (callback == null) + { + Interlocked.Add(ref _skippedNoCallback, sliceSize); + } + else + { + var end = Math.Min(cursor + sliceSize, tagPaths.Length); + for (var i = cursor; i < end; i++) + { + // Stamped HERE: this instant rides TagValueUpdate.Timestamp all the + // way to the subscriber, so the probe's subtraction is a genuine + // end-to-end latency and not a re-stamped approximation. + callback(tagPaths[i], new TagValue(sequence, QualityCode.Good, DateTimeOffset.UtcNow)); + } + + Interlocked.Add(ref _emitted, end - cursor); + cursor = end; + } + + if (cursor >= tagPaths.Length) + { + cursor = 0; + sequence += 1d; + } + + var elapsed = Stopwatch.GetTimestamp() - sliceStart; + var budgetTicks = (long)(sliceBudget.TotalSeconds * Stopwatch.Frequency); + if (elapsed < budgetTicks) + { + var remaining = TimeSpan.FromSeconds((budgetTicks - elapsed) / (double)Stopwatch.Frequency); + try + { + await Task.Delay(remaining, cancellationToken); + } + catch (OperationCanceledException) + { + return; + } + } + else + { + Interlocked.Add(ref _lagTicks, elapsed - budgetTicks); + } + } + } + + /// + /// Resolves the live adapter callback for a connection, by NAME. The DCL factory + /// appends adapters as its manager actor processes CreateConnectionCommands, so + /// list position does not track connection index. + /// + private static SubscriptionCallback? ResolveCallback(SiteRuntimeFixture site, int connectionIndex) + { + var name = SiteRuntimeFixture.ConnectionName(connectionIndex); + // Last match wins: a reconnect would create a fresh adapter for the same name, + // and only the newest one holds the live subscription callback. + return site.Connections.LastOrDefault(c => c.ConnectionName == name)?.ValueCallback; + } + + private int _disposed; + + /// + public async ValueTask DisposeAsync() + { + if (Interlocked.Exchange(ref _disposed, 1) != 0) + return; + + await _cts.CancelAsync(); + try + { + await Task.WhenAll(_tasks); + } + catch (OperationCanceledException) + { + // Expected on shutdown. + } + + _cts.Dispose(); + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/ZB.MOM.WW.ScadaBridge.LoadHarness.csproj b/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/ZB.MOM.WW.ScadaBridge.LoadHarness.csproj new file mode 100644 index 00000000..480b3344 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/ZB.MOM.WW.ScadaBridge.LoadHarness.csproj @@ -0,0 +1,40 @@ + + + + + net10.0 + Exe + enable + enable + true + false + ZB.MOM.WW.ScadaBridge.LoadHarness + + true + true + + + + + + + + + + + + + diff --git a/tests/ZB.MOM.WW.ScadaBridge.PerformanceTests/TargetScale/TargetScaleHarnessSmokeTests.cs b/tests/ZB.MOM.WW.ScadaBridge.PerformanceTests/TargetScale/TargetScaleHarnessSmokeTests.cs new file mode 100644 index 00000000..b5dbb5fe --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.PerformanceTests/TargetScale/TargetScaleHarnessSmokeTests.cs @@ -0,0 +1,84 @@ +using ZB.MOM.WW.ScadaBridge.LoadHarness; + +namespace ZB.MOM.WW.ScadaBridge.PerformanceTests.TargetScale; + +/// +/// Keeps the target-scale load harness (deferred-work register #25 / Phase-8 WP-4) +/// honest at CI scale. +/// +/// +/// The full protocol — 10 sites x 500 instances x 75 tags sustained for 20 minutes — +/// deliberately lives in the standalone ZB.MOM.WW.ScadaBridge.LoadHarness +/// executable, NOT here: perf tests in this project run as part of an ordinary +/// dotnet test ZB.MOM.WW.ScadaBridge.slnx (the Category=Performance +/// trait enables a filter, it does not exclude by default), and a 20-minute test +/// would be intolerable there. What this test protects is that the harness still +/// compiles, wires up, and produces coherent measurements — so #25's evidence can be +/// regenerated on demand rather than bit-rotting. +/// +/// +/// Results doc: docs/plans/2026-08-15-target-scale-load-test-results.md. +/// Design memo: docs/plans/2026-08-15-target-scale-load-test-design.md. +/// +/// +public class TargetScaleHarnessSmokeTests +{ + /// + /// Runs the harness at ~1/1000th of target scale for a few seconds and asserts the + /// pipeline is intact end to end: tag updates reach live stream subscribers, the + /// central health aggregator tracks every site, debug snapshots answer, the + /// store-and-forward buffer drains to empty, and a stalled subscriber does not cost + /// the healthy ones any events. + /// + /// A task representing the test run. + [Trait("Category", "Performance")] + [Fact] + public async Task Harness_AtSmokeScale_ProducesCoherentMeasurements() + { + var config = new HarnessConfig + { + Sites = 2, + InstancesPerSite = 10, + TagsPerInstance = 5, + TagUpdatePeriod = TimeSpan.FromSeconds(1), + SettleDuration = TimeSpan.FromSeconds(5), + SustainDuration = TimeSpan.FromSeconds(15), + SampleInterval = TimeSpan.FromSeconds(2), + // Shortened from the production 30 s only because the smoke window is 20 s. + HealthReportInterval = TimeSpan.FromSeconds(2), + DebugProbeInterval = TimeSpan.FromSeconds(2), + SubscribeSettleDuration = TimeSpan.FromSeconds(5), + StreamProbesPerSite = 3, + StoreAndForwardDrainMessages = 200, + SlowSubscriberEvents = 2_000, + ResultsPath = Path.Combine(Path.GetTempPath(), $"loadharness-smoke-{Guid.NewGuid():N}.json"), + }; + + using var cancellation = new CancellationTokenSource(TimeSpan.FromMinutes(10)); + var result = await HarnessRun.ExecuteAsync(config, _ => { }, cancellation.Token); + + // Tag updates flowed all the way through DCL -> InstanceActor -> site stream -> + // StreamRelayActor -> bounded channel -> subscriber. + Assert.True(result.TagUpdateLatency.Count > 0, + "No tag update latency samples — the DCL -> stream -> subscriber path did not carry traffic."); + Assert.True(result.StreamProbeReceived > 0, "Live stream subscribers received nothing."); + Assert.Equal(0, result.DriverSkippedNoCallback); + + // Observability held up. + Assert.Equal(config.Sites, result.SitesTrackedByAggregator); + Assert.True(result.DebugSnapshotsCompleted > 0, "No debug snapshot completed."); + Assert.Equal(0, result.DebugSnapshotTimeouts); + + // Store-and-forward drained completely (register row 50, first half). + Assert.NotNull(result.StoreAndForwardDrain); + Assert.Equal(0, result.StoreAndForwardDrain!.ResidualDepth); + Assert.True(result.StoreAndForwardDrain.DrainPerSecond > 0); + + // A stalled subscriber costs the healthy ones nothing (register row 50, second + // half). This is the design's isolation claim, asserted rather than assumed. + Assert.NotNull(result.SlowSubscriber); + Assert.Equal(1.0, result.SlowSubscriber!.HealthyMinDeliveryRatio, precision: 2); + Assert.True(result.SlowSubscriber.SlowDeliveryRatio < 1.0, + "The deliberately stalled subscriber kept up, so the probe proved nothing."); + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.PerformanceTests/ZB.MOM.WW.ScadaBridge.PerformanceTests.csproj b/tests/ZB.MOM.WW.ScadaBridge.PerformanceTests/ZB.MOM.WW.ScadaBridge.PerformanceTests.csproj index 02f7b2fa..0ddcfaac 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.PerformanceTests/ZB.MOM.WW.ScadaBridge.PerformanceTests.csproj +++ b/tests/ZB.MOM.WW.ScadaBridge.PerformanceTests/ZB.MOM.WW.ScadaBridge.PerformanceTests.csproj @@ -31,6 +31,9 @@ + + From 8abebdae33dada6a7cadb6e5f73341d06a8c3cbf Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Sat, 15 Aug 2026 02:27:47 -0400 Subject: [PATCH 2/4] docs(plans): target-scale load test harness design memo (WP-4 / register #25) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Records harness architecture, the real-vs-faked table with a justification per fake, metric definitions, falsifiable pass/fail thresholds derived from the WP-4 acceptance criteria, and the eight deviations from the WP-4 protocol with reasons — including the 1-hour to 20-minute sustained-window shortening (memory reported as a slope so a shorter window still answers the leak question) and the four [xc-*] criteria this harness does not cover. --- ...026-08-15-target-scale-load-test-design.md | 205 ++++++++++++++++++ 1 file changed, 205 insertions(+) create mode 100644 docs/plans/2026-08-15-target-scale-load-test-design.md diff --git a/docs/plans/2026-08-15-target-scale-load-test-design.md b/docs/plans/2026-08-15-target-scale-load-test-design.md new file mode 100644 index 00000000..d90eabd3 --- /dev/null +++ b/docs/plans/2026-08-15-target-scale-load-test-design.md @@ -0,0 +1,205 @@ +# Target-Scale Load Test — Harness Design (Phase-8 WP-4, deferred-work register #25 + row 50) + +**Date:** 2026-08-15 +**Closes:** deferred-work register **#25** (target-scale load test) and **row 50** +(S&F drain rate + per-subscriber stream backpressure), and residual **7** of +`docs/plans/2026-08-15-arch-review-remediation-execution-log.md`. +**Spec:** `docs/plans/phase-8-production-readiness.md` WP-4 (`:152-170`) + test protocol (`:314-320`). +**Results:** `docs/plans/2026-08-15-target-scale-load-test-results.md`. +**Harness:** `tests/ZB.MOM.WW.ScadaBridge.LoadHarness/` (+ CI smoke in +`tests/ZB.MOM.WW.ScadaBridge.PerformanceTests/TargetScale/TargetScaleHarnessSmokeTests.cs`). + +--- + +## 1. Why this exists + +Register row 25 records that WP-4 was *claimed complete* on the strength of a +**107-byte checklist stub** ("Status: Complete / Tests: All passing / Build: 0 errors, +0 warnings") with no per-work-package results and no linked run. The nearest real +coverage was arithmetic: + +- `PerformanceTests/StaggeredStartupTests.cs` — `TagCapacity_75TagsPer500Machines_37500Total` + and the 10-site distribution test compute products of integers. No actor is created. +- `PerformanceTests/HealthAggregationTests.cs` — feeds 10 hand-built `SiteHealthReport` + records to a real aggregator. Real, but not under load. +- `PerformanceTests/Streaming/SiteStreamThroughputTests.cs` — a genuine 100k-event + throughput test, but **one subscriber** and no instances behind it. + +So the job here is not to re-assert the claim. It is to produce the evidence the claim +never had, and to report what that evidence actually says — including where it is +uncomfortable. + +## 2. Feasibility constraints that shaped the design + +**The host is one macOS box** (Apple M4 Pro, 14 cores, 48 GB) running OrbStack with the +8-node `docker/` rig already up. The `docker/` topology has 3 sites and cannot host 10 +real site pairs at 500 instances each; a full-docker WP-4 topology is out of scope. + +**There is no simulated protocol adapter in the DCL.** Only `OpcUa` and `MxGateway` +ship. `DataConnectionFactory.RegisterAdapter` is the documented extension point, and +`tests/…DataConnectionLayer.Tests/Actors/FakeBatchDataConnection.cs` establishes the +fake-adapter shape. + +**375,000 real OPC UA monitored items is not the system under test.** That would +measure the OPC UA SDK. The system under test is everything *above* the adapter +callback: the actor hierarchy, the site stream, store-and-forward, health, audit, and +the streaming relay. + +The established repo pattern for scale/failover validation is in-process performance +tests (`PerformanceTests/Failover/FailoverTimingTests.cs` runs a real two-node cluster +in-process via `TwoNodeClusterFixture` with production `BuildHocon` and production +timings). This harness extends that pattern rather than inventing a new one. + +## 3. Harness shape, and why + +**A standalone console executable** (`tests/ZB.MOM.WW.ScadaBridge.LoadHarness`, an +`Exe`), **plus** a scaled-down `Category=Performance` `[Fact]` in `PerformanceTests` +that references it. + +The reason it is not purely an xunit suite is specific and worth recording: in this +repo the `[Trait("Category", "Performance")]` attribute **enables a filter, it does not +exclude by default**. `dotnet test ZB.MOM.WW.ScadaBridge.slnx` runs the Performance +project's tests — `FailoverTimingTests` included, at 45-90s. A 20-minute full-scale +test dropped in there would be a 20-minute tax on every solution test run. Hence: + +| | Full protocol | CI smoke | +|---|---|---| +| Where | `LoadHarness` executable | `TargetScaleHarnessSmokeTests` | +| Scale | 10 x 500 x 75 = 375,000 | 2 x 10 x 5 = 100 | +| Duration | ~28 min wall | ~80 s | +| Purpose | The WP-4 numbers | Harness does not bit-rot | + +The smoke test asserts the *shape* of a healthy result (traffic flows end to end, all +sites tracked centrally, S&F drains to empty, a stalled subscriber costs healthy ones +nothing) so #25's evidence can be regenerated on demand rather than being a one-off. + +### 3.1 Process topology + +One process. Ten `SiteRuntimeFixture` instances, each owning: + +- its own **`ActorSystem`** (non-clustered), +- its own **LocalDb SQLite file** (real `AddZbLocalDb` + real `SiteStorageService`), +- a real **`DataConnectionManagerActor`** with 5 `DataConnectionActor` children, +- 500 real **`InstanceActor`s**, each configured with 75 data-sourced attributes, +- a real **`SiteStreamManager`** at the production `StreamBufferSize` (1000), +- a real **`StoreAndForwardStorage` + `StoreAndForwardService`**, +- a real **`SiteHealthCollector`**. + +Plus one shared real **`CentralHealthAggregator`** standing in for central. + +### 3.2 What is real and what is faked + +| Layer | Real / faked | Why the fake does not invalidate the measurement | +|---|---|---| +| OPC UA server + socket | **Faked** (`SimulatedDataConnection`) | Everything replaced is on the far side of the process boundary: socket I/O, the SDK's session/subscription machinery, the device. The system under test begins at the `SubscriptionCallback` invocation, which is exactly where the real adapter hands off. | +| `DataConnectionFactory` / `DataConnectionManagerActor` / `DataConnectionActor` | **Real** | The per-connection actor's `_instancesByTag` fan-out and its single-mailbox serialization point are a genuine scale surface — one of the things WP-4 must characterize. The sim adapter is registered through the documented `RegisterAdapter` seam. | +| `InstanceActor` | **Real** | The whole point. Real `TagValueUpdate` ingest, real type coercion, real `PublishAndNotifyChildren`. | +| `SiteStreamManager` | **Real**, production `StreamBufferSize` | Per-subscriber `Buffer(DropHead)` behaviour is under test. | +| `StreamRelayActor` + bounded `DropOldest` channel | **Real**, production capacity (1000) | This is `SiteStreamGrpcServer`'s per-subscription machinery, reused verbatim. | +| gRPC socket writer | **Faked** (a reader task) | Deliberate: it is *precisely* the hop whose slowness register row 50 asks about, and a controllable reader is the only way to hold it still. | +| `StoreAndForwardService` / `Storage` / SQLite | **Real** | Drain throughput is the measurement. | +| S&F delivery target (central) | **Faked** (counting stub returning `true`) | What is measured is the site-local buffer's capacity, not a remote endpoint's. | +| `SiteHealthCollector` / `CentralHealthAggregator` | **Real** | `CollectReport` at 37,500 subscriptions is the interesting term. | +| `IHealthReportTransport` (gRPC hop) | **Faked** (direct call) | The transport is a documented interface seam; the cost being measured is collect + ingest. | +| Akka cluster membership / failover | **Not exercised** | Already measured on a real two-node rig by `FailoverTimingTests` and `docker/failover-drill.sh`. WP-4 asks about the hierarchy *under* the singleton. | + +### 3.3 The latency measurement is genuinely end to end + +The driver stamps `DateTimeOffset.UtcNow` on the `TagValue` it hands the adapter +callback. That instant then travels **verbatim**, with no re-stamping: + +``` +driver → SubscriptionCallback → DataConnectionActor (self.Tell(TagValueReceived)) + → TagValueUpdate.Timestamp (DataConnectionActor fan-out) + → AttributeValueChanged.Timestamp (InstanceActor.HandleTagValueUpdate copies it) + → SiteStreamManager hub → per-subscriber Buffer → StreamRelayActor + → proto SiteStreamEvent.AttributeChanged.Timestamp + → subscriber reader: now - Timestamp +``` + +Everything on that path is production code. This is the harness's single most +important property: the reported percentiles are not a synthetic stopwatch around a +method call, they are the real DCL-boundary-to-subscriber path. + +## 4. Load model + +WP-4 fixes the scale but **states no tag update rate**, so one has to be chosen and +justified. + +- **Nominal rate: one update per tag per 10 seconds.** At 375,000 subscriptions that + is **37,500 tag updates/second fleet-wide**, 3,750/s per site, 750/s per data + connection actor. +- Rationale: a plant SCADA tag that genuinely changes every 10 s is a *busy* tag; + assuming every one of 375,000 tags does so simultaneously is a deliberately + pessimistic steady state, not a typical one. It also sits comfortably above the + 10k events/s floor the existing `SiteStreamThroughputTests` pins, so the two + measurements bracket each other. +- **5 data connections per site.** A site with 37,500 tags behind a single OPC UA + server is not the realistic shape, and it would make one actor mailbox the entire + story. Five connections (7,500 tags each) is realistic; the per-connection rate is + reported so the single-connection case is derivable. +- **10 live stream subscribers per site** (100 fleet-wide). Every subscriber's stream + graph sees the full site event flow and filters it by instance name, so this is the + fan-out multiplier on the hub — the pessimistic direction. + +The driver reports `EmitLagSeconds` (cumulative slice overrun) and +`SkippedNoCallback`. If the harness itself cannot offer the nominal rate, those +numbers say so, and the achieved-vs-nominal ratio is published rather than the +nominal being quietly reported as if achieved. + +## 5. Metric definitions + +| Metric | Definition | +|---|---| +| **Tag update latency** | Subscriber receive instant minus the driver's emit stamp, in ms. Logarithmic histogram, 16 buckets/octave (bucket width ≤ 4.4%, so a reported percentile is within ~4.4% of truth). Mean/max are exact, not bucketed. **Steady-state window only** — the histogram is repointed at window start so ramp outliers cannot contaminate it. | +| **Instance ramp / deployment at scale** | Wall time to create every `InstanceActor`, at the production `StartupBatchSize` (20) / `StartupBatchDelayMs` (100) staggering. Both the fleet total (sites in parallel) and the **slowest single site** are reported; the latter is the "deploy 500 instances to a site" figure. | +| **Memory growth** | Working set and managed heap, sampled every 10 s. Reported as start→end delta, peak, **and** least-squares slope in MB/min over the steady-state window — because a run that sawtooths around a stable mean and one that climbs monotonically can share the same endpoint delta. | +| **CPU** | `Process.TotalProcessorTime` delta / wall delta, expressed as a percentage of **one core** (so 1400% = the whole 14-core box saturated) and also as a percentage of the box. | +| **Health report delivery timing** | `SiteHealthCollector.CollectReport(siteId)` + `CentralHealthAggregator.ProcessReport(report)`, timed together, once per site per 30 s tick under full load. | +| **Debug view latency** | `Ask(DebugSnapshotRequest)` round-trip to a randomly chosen live `InstanceActor` every 5 s under full load — so the measurement includes real queueing behind production traffic. | +| **S&F drain rate** | Reported as **two** numbers: time-to-first-delivery (the configured retry latency) and throughput measured from the **first** delivery to an empty buffer (the engine's actual capacity). A drain-progress series is captured so a steady rate can be told from a stall-then-burst. | +| **Slow-subscriber isolation** | Several subscribers on the *same* instance (identical offered event sequence); one reader stalled at 50 ms/event. Delivery ratio per subscriber, plus per-subscriber channel eviction counts. | + +## 6. Pass/fail thresholds + +Derived from the WP-4 acceptance criteria. Where the criterion is qualitative +("within acceptable time", "does not degrade"), the threshold is stated here so the +verdict is falsifiable rather than a judgement call made after seeing the number. + +| # | Criterion (WP-4) | Threshold | +|---|---|---| +| 1 | `[2.5-1]` 10 sites simultaneously operational | All 10 sites built, ramped, and tracked by the central aggregator | +| 2 | `[2.5-2]` 500 instances/site with active subscriptions | 5,000 `InstanceActor`s alive; `SkippedNoCallback` = 0 in the steady window | +| 3 | `[2.5-3]` 75 live tags/instance (375,000 total) | 375,000 tag paths subscribed through the real DCL | +| 4 | Tag update latency | **P99 < 250 ms**, P50 < 50 ms. (A human-facing live value that lands within a quarter second is indistinguishable from instant; the site stream is explicitly best-effort/lossy under pressure, so this bounds the *delivered* path.) | +| 5 | `[xc-1]` message patterns function under load | Achieved offered load ≥ 95% of nominal, and no unexpected event loss at healthy subscribers | +| 6 | `[xc-2]` health reports arrive within expected intervals | Collect+ingest **P99 < 1,000 ms** (must be a small fraction of the 30 s report interval, else reports would queue) and all 10 sites tracked | +| 7 | `[xc-6]` debug view streams without impacting site performance | Snapshot round-trip **P99 < 2,000 ms**, zero ask timeouts, and no measurable tag-latency penalty | +| 8 | `[xc-7]` S&F handles concurrent buffering from multiple instances | 20,000 messages buffered concurrently from 25 origin instances with no error; buffer drains to depth 0 | +| 9 | Row 50 — S&F drain rate | Drain throughput **> 500 msg/s** (a 20,000-message backlog clears in well under a minute once due) | +| 10 | Row 50 — slow-subscriber backpressure | Healthy subscribers' delivery ratio **= 100%** while a peer is stalled; the stalled subscriber's loss is confined to its own bounded channel | +| 11 | Deployment of 500 instances to a site | Slowest single site **< 120 s** | +| 12 | Memory within acceptable bounds | Steady-state working-set slope **< 20 MB/min**, and no monotonic managed-heap climb across the window | +| 13 | CPU within acceptable bounds | Steady-state mean **< 50% of the box** at nominal load | + +## 7. Deviations from the WP-4 protocol + +Each is a deliberate, recorded trade — not an omission. + +| # | Deviation | Reason | +|---|---|---| +| D1 | **Sustained window shortened from 1 hour to 20 minutes** at full scale | Practicality on a single shared workstation. Memory growth is reported as a *slope* precisely so a shorter window still answers the leak question: a leak shows as a positive slope in 20 min just as in 60. The 1-hour run is a single flag — `--sustain-minutes 60` — and the exact command is recorded in the results doc. | +| D2 | **In-process, non-clustered sites** rather than 10 real two-node Akka clusters | A single box cannot host 20 clustered nodes at this scale. Cluster membership/failover is already measured on a real two-node rig (`FailoverTimingTests`, `docker/failover-drill.sh`); WP-4's question is about the hierarchy under the singleton. | +| D3 | **Simulated data source** instead of real OPC UA | No simulated adapter exists in the DCL, and 375k real monitored items would measure the OPC UA SDK. Injected at the documented `RegisterAdapter` seam, at the exact hand-off point the real adapter uses. | +| D4 | **gRPC socket replaced by a reader task** on the stream path | Required by the row-50 measurement itself: a stalled subscriber must be *held* stalled. All of `SiteStreamGrpcServer`'s per-subscription machinery above the socket is real. | +| D5 | **Instance Actors are `/user` children**, not children of a `DeploymentManagerActor` | Constructing them directly is what lets the ramp be timed in isolation and keeps the harness independent of the deploy round-trip. Visible only as `InstanceActorInitialized` dead letters, which the harness suppresses. No measured path differs. | +| D6 | **Akka logging at WARNING** | At 37,500 updates/s, INFO output would itself become a measured load. | +| D7 | `[xc-3]` site event logging volume, `[xc-4]` audit-log degradation, `[xc-5]` template flattening, `[xc-8]` UI responsiveness **not covered** | Out of scope for this harness: the first three need central MS SQL and the Template Engine (a central-cluster fixture, not a site one), and `[xc-8]` needs a browser. Recorded as residual scope in the results doc rather than silently dropped. | +| D8 | **The `docker/` 8-node rig stays running** during the measurement | It is the user's live development cluster. Its baseline cost is measured and recorded alongside the result so the CPU figure is interpretable. | + +## 8. Honest-reporting rule + +If a criterion fails, that failure **is** the deliverable. Findings are recorded in the +results doc with the measured evidence; nothing is tuned to make a number go green, +and no risky fix is attempted under cover of this work package. Anything found gets +filed as a finding for separate triage. From da65605e4160e8f040913071c39fa79e69396b75 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Sat, 15 Aug 2026 02:55:59 -0400 Subject: [PATCH 3/4] docs(plans): WP-4 target-scale load test RESULTS + close register #25 and row 50 Full-scale run executed on this machine: 10 sites x 500 instances x 75 tags = 375,000 live tag subscriptions, 37,518 updates/s achieved vs 37,500 nominal, 45,021,375 updates over a 20-minute steady-state window (M4 Pro, 14 cores, 48 GB, with the 8-node docker rig still running so the figures are pessimistic). 11 clean passes, 1 pass with a caveat, 0 failures: tag latency P50 0.88ms P99 4.57ms max 37.41ms (1.1M samples, end-to-end) stream 900,675 delivered, 0 dropped at 100 live subscribers health collect+ingest P99 0.31ms, 10/10 sites tracked debug view P99 2.19ms, 264 completed, 0 timeouts deploy 500 instances to a site in 2.6s cpu 41% of ONE core = 2.9% of the box memory working-set slope +8.83 MB/min Three findings, reported rather than tuned away: F1 (Low) 20 min with ZERO gen-2 collections cannot fully settle the leak question; the heap demonstrably sawtooths but an uncompacted gen-2 makes a positive slope ambiguous. The 1-hour run would settle it. Not tuned. F2 (informational, by design) a deferred S&F backlog sits for one full DefaultRetryInterval (28.9s measured) before anything drains -- EnqueueAsync(attemptImmediateDelivery:false) stamps LastAttemptAt. Easy to misread as slow drainage, so drain is reported as two numbers: retry wait, then 3,533 msg/s of actual capacity. F3 (positive) slow-subscriber isolation is TOTAL: 4 healthy subscribers at 100.00% with zero drops while a peer lost 197,028/200,000 events entirely within its own bounded channel. Mechanism recorded link by link. Also records what the run does NOT prove (not clustered, not real-network, not real OPC UA, not 1 hour) and the four WP-4 sub-criteria this harness does not cover, so the evidence is not over-read. Register rows 25 and 50 -> RESOLVED 2026-08-15; remediation execution-log residual 7 -> resolved; phase-8-checklist WP-4 section replaced with the measured numbers. --- .../2026-07-08-deferred-work-register.md | 4 +- ...5-arch-review-remediation-execution-log.md | 17 +- ...26-08-15-target-scale-load-test-results.md | 289 ++++++++++++++++++ docs/plans/phase-8-checklist.md | 51 +++- .../HarnessRun.cs | 8 + 5 files changed, 351 insertions(+), 18 deletions(-) create mode 100644 docs/plans/2026-08-15-target-scale-load-test-results.md diff --git a/docs/plans/2026-07-08-deferred-work-register.md b/docs/plans/2026-07-08-deferred-work-register.md index c1c71cf2..4a3eb67a 100644 --- a/docs/plans/2026-07-08-deferred-work-register.md +++ b/docs/plans/2026-07-08-deferred-work-register.md @@ -18,7 +18,7 @@ All 7 fix-now items landed via PLAN-04/05/06/07/08 (verified in review 08 round | 19 | Bundle signing / cluster-to-cluster pull / differential bundles | transport-design :402 | v1 manifest hash + AES-GCM held sufficient | Non-repudiation requirement across orgs | | 23 | Live LDAP group-membership re-query for an active session | `docs/requirements/Component-Security.md` :61-69 (+ :78-79) | Blocked on an external package. The mid-session refresh re-maps the **stored** groups against the central DB with **no LDAP call**, so a directory group-membership change lands only at next login. A live re-query needs a passwordless service-account group-search method on the shared `ZB.MOM.WW.Auth.Ldap` library — an external NuGet `PackageReference` (`src/ZB.MOM.WW.ScadaBridge.Security/…csproj:23`) exposing only `AuthenticateAsync(username, password, ct)`. Central role-mapping/scope changes still apply within ~15 min (`RoleRefreshThresholdMinutes`). | `ZB.MOM.WW.Auth.Ldap` gains a standalone group-search API, or a requirement that a directory-side group revocation take effect mid-session rather than at next login | | 24 | M8 large-bundle performance hardening | `docs/plans/2026-06-15-stillpending-completion-design.md:106` — "Small follow-ups logged (not blocking): … large-bundle/perf hardening" | Logged as a non-blocking follow-up when M8 shipped and never given an artifact: **no plan, no task entry, no perf/load test exists** (`tests/…Transport.Tests/Import/BundleImporterLoadTests.cs` is a `LoadAsync` unit suite despite the name). No measured problem; the only sizing controls in place are the 5-minute CLI transport timeout, `LineDiffer`'s `MaxInputLines`=4000 summary-only cap, and `MaxConcurrentImportSessions`=8. | First real bundle that times out, exhausts memory, or makes the import wizard's diff step unusable | -| 25 | Phase-8 WP-4 target-scale load test (10 sites × 500 instances × 75 tags = 37,500 subscriptions/site, 375,000 total) | `docs/plans/phase-8-production-readiness.md:152-170` (WP-4) + `:314-320` (test protocol); status claimed in `docs/plans/phase-8-checklist.md` | **Claimed complete but unevidenced.** The whole WP-4 deliverable is a **107-byte** checklist stub asserting "Status: Complete / Tests: All passing / Build: 0 errors, 0 warnings" with no per-work-package results and no linked run. Nearest real coverage is arithmetic/aggregation only — `PerformanceTests/StaggeredStartupTests.cs` (`TagCapacity_75TagsPer500Machines_37500Total`, 500-instances-over-10-sites distribution) and `HealthAggregationTests` (10-site report aggregation) — plus a **single-subscriber** 100k-event `Streaming/SiteStreamThroughputTests.cs`. No sustained multi-site run exists anywhere in `tests/` or `docker/`. | Before any production go-live at target scale; or the first site approaching ~500 instances / ~37.5k subscriptions | +| 25 | ~~**Phase-8 WP-4 target-scale load test** (10 sites x 500 instances x 75 tags = 37,500 subscriptions/site, 375,000 total)~~ **RESOLVED 2026-08-15 — run, measured, evidenced.** The 107-byte stub is replaced by a real full-scale run: **10 x 500 x 75 = 375,000 live tag subscriptions**, 37,518 tag updates/s achieved against 37,500 nominal (100.0%), **45,021,375 updates offered** over a 20-minute steady-state window on an M4 Pro / 14-core / 48 GB host with the 8-node `docker/` rig still running. Headline numbers: tag update latency **P50 0.88 ms / P99 4.57 ms / max 37.41 ms** end-to-end (DCL boundary to stream subscriber, the emit instant carried verbatim through production code); **0 events dropped** at 100 live subscribers; health report collect+ingest **P99 0.31 ms** with 10/10 sites tracked; debug view snapshot under load **P99 2.19 ms, 0 timeouts**; 500 instances deployed to a site in **2.6 s**; CPU **2.9% of the box**; working-set slope **+8.83 MB/min**. **11 clean passes, 1 pass with a caveat (F1: 20 min with zero gen-2 collections cannot fully settle the leak question — the 1-hour run would), 0 failures.** Harness: `tests/ZB.MOM.WW.ScadaBridge.LoadHarness/` (standalone Exe — the `Category=Performance` trait enables a filter but does NOT exclude by default, so a 20-minute test could not live in `PerformanceTests`) plus a CI-scale smoke `[Fact]` (`PerformanceTests/TargetScale/TargetScaleHarnessSmokeTests.cs`). Real: per-site ActorSystem + LocalDb, the real DCL via the documented `DataConnectionFactory.RegisterAdapter` seam, real `InstanceActor`s, real `SiteStreamManager`, real `StreamRelayActor` + production-capacity bounded channel, real S&F, real health collector/aggregator; only the socket hops are stood in for. **Results: `docs/plans/2026-08-15-target-scale-load-test-results.md`; design + deviations + thresholds: `docs/plans/2026-08-15-target-scale-load-test-design.md`.** | as before | Closed. Residual scope: `[xc-3]` site-event-log retention volume, `[xc-4]` audit-vs-central degradation, `[xc-5]` large-template flattening and `[xc-8]` UI responsiveness are NOT covered by this harness (central-cluster/browser concerns) and remain open WP-4 scope — see results doc SS6. | Closed 2026-08-15. | | 26 | Ipsen MES MoveIn tail: leak-test (`-LT`) receivers + routing, PLC-output-flag writes, `Z28062` BTDB data completeness | `docs/plans/2026-06-16-ipsen-mes-movein.md:409` ("Out of scope (future)"); design `2026-06-16-ipsen-mes-movein-design.md:58-60, :196-198` | Customer-site scope, not a platform gap. `-LT` routing needs an MES-receiver child + Galaxy reference that do not exist on the reactor template (any `-LT`/unknown suffix returns `WasSuccessful=false` with an "unsupported side/target" message by decision); `MoveInComplete`/`Successful`/`ErrorText` are **PLC-owned** by locked decision, so ScadaBridge deliberately does not write them; `Z28062` completeness is an operational data fix, not code. Note the separate alarm-status path already handles the suffix — `_LT` is stripped before side-scoping (`2026-06-30-mes-alarm-status-api.md:158`). | Ipsen creates the leak-test receiver + Galaxy reference, or asks ScadaBridge to own the PLC-output flags — otherwise a **candidate won't-do** (`[PERM]`) at the next Ipsen scope review | | 27 | External-system per-system retry config (`MaxRetries`/`RetryDelay`) never reaches sites, and has no CLI/management surface | Found live 2026-08-01 (rig session, #11 gRPC live checks) | Two stacked gaps: (a) `ExternalSystemArtifact` (Commons) carries `TimeoutSeconds` but NOT `MaxRetries`/`RetryDelay`, and the site `external_systems` table has no such columns — so a centrally-configured retry policy is silently ignored on sites; every cached call buffers with the S&F default (`DefaultMaxRetries` 50 × `DefaultRetryInterval` 30s ≈ 25 min to park). (b) `Create/UpdateExternalSystemCommand` don't expose the fields either — the only way to set them today is a direct DB edit of `ExternalSystemDefinitions`. Transport bundles DO carry them (arch-review 05 "ES retry config"), which masks the gap in export/import round-trips. Fix is additive: extend the artifact + site schema + apply path, and add `--max-retries`/`--retry-delay` to the CLI. | First operator who tunes retry policy on an external system and expects site cached calls to honor it | | 28 | Health-dashboard "Trigger failover" confirm dialog's confirm button is labeled **"Delete"** | Found live 2026-08-01 (rig session, #11 TriggerSiteFailover check) | The DialogService confirmation host's default destructive-action label leaks through — the dialog copy is correct but the red confirm button says "Delete" for a failover. One-line fix: pass an explicit confirm label ("Fail over") at the Health-dashboard call site (and audit other confirm-dialog call sites for the same default). | Next Central UI session | @@ -47,7 +47,7 @@ Rows removed from the Deferred table above once confirmed shipped. Kept here for | docs/components reference docs for ScriptAnalysis, KpiHistory, DelmiaNotifier | Reference docs are substantial (StyleGuide-conformant); README claim scoped instead (PLAN-08 Task 10) | Next doc-writing session touching those components | | Test-coverage backfill: SiteCallAudit.Tests (31 tests/1.6k LOC), DeploymentManager.Tests | No defect identified; coverage partly lives in ManagementService/Host/Integration suites | First regression escaping either component | | ~~Failover-timing measurement (the "~25s total failover" envelope)~~ **RESOLVED 2026-08-01** — split out of the combined row and closed. `tests/ZB.MOM.WW.ScadaBridge.PerformanceTests/Failover/FailoverTimingTests.cs` is no longer a skipped placeholder: it runs as a live `[Fact]` (`Category=Performance`) on the real two-node in-process rig (`TwoNodeClusterFixture`, production `BuildHocon`) at production timings — 2s heartbeat / 10s failure-detection threshold / 15s stable-after — hard-killing the younger node and timing the survivor's member REMOVAL with singleton continuity asserted on the oldest. Delivered by **PLAN-R2-01 Task 4** (`archreview/plans/PLAN-R2-01-cluster-host-failover.md:226`). The oldest-crash direction is covered behaviorally by `SbrFailoverTests.AutoDown_HardCrashOfOldestNode_*` and by `docker/failover-drill.sh`. | The 2026-07-08 "PLAN-01 rig landing" trigger had fired unnoticed (NF2); PLAN-R2-01 T4 wired the placeholder to the fixture rig rather than recording a blocker. | Closed. | -| Broader perf envelope — **S&F drain rate + per-subscriber stream backpressure** (the still-open half of the former combined row) | Never measured, and no owner plan survives now that PLAN-R2-01 closed the failover half. `PerformanceTests` covers failover timing, staggered startup, health aggregation, audit hot-path latency and a **single-subscriber** 100k-event `Streaming/SiteStreamThroughputTests.cs` — nothing measures store-and-forward drain throughput, nor what a slow/stalled subscriber does to the per-subscriber buffering in `Communication/Actors/StreamRelayActor.cs` / `Grpc/SiteStreamGrpcServer.cs` with many subscribers attached. No defect observed; deferred as measurement-only work. | First field S&F backlog that fails to drain within an operator's patience, a slow gRPC subscriber degrading a site stream for others, or the WP-4 target-scale run (row 25) being scheduled — that run should absorb this | +| ~~Broader perf envelope — **S&F drain rate + per-subscriber stream backpressure**~~ **RESOLVED 2026-08-15 — absorbed into the row-25 target-scale run, exactly as this row's trigger anticipated.** Both halves measured at full scale. **(a) S&F drain:** 20,000 messages buffered concurrently from 25 origin instances at **15,368 msg/s**, draining at **3,533 msg/s** to depth 0. Reported as two numbers because the naive single number is misleading — a deferred backlog sits for one full `DefaultRetryInterval` (measured 28.9 s) before *anything* drains, since `EnqueueAsync(attemptImmediateDelivery: false)` stamps `LastAttemptAt` and the due-predicate then holds the row for one interval (correct by design; `Notify.Send`'s `deferToSweep: true` leaves it null and skips the wait). Filed as informational finding **F2**. **(b) Slow-subscriber backpressure:** with 5 subscribers on the SAME instance and one reader stalled at 50 ms/event, the four healthy subscribers delivered **100.00% with zero drops** while the stalled one lost 197,028 of 200,000 events **entirely within its own bounded channel**. Isolation is structural and total — `Buffer(DropHead)` never backpressures the `BroadcastHub`, `Sink.ForEach`'s `Tell` never blocks, and `StreamRelayActor.WriteToChannel`'s `TryWrite` on a `DropOldest` channel never blocks — so a slow WAN link degrades only its own feed, visibly (per-stream eviction counter + `RecordSiteStreamEventDropped`). Finding **F3**. Caveat also recorded: an *unpaced* burst costs every subscriber, because the publish `Source.ActorRef` upstream of the hub is shared — that shared stage, not per-subscriber buffering, bounds burst absorption. | Closed by the row-25 run; no defect found in either half. | Closed 2026-08-15 — `docs/plans/2026-08-15-target-scale-load-test-results.md`. | ## Deferred — operational risk (from the initiative tracker, folded in 2026-07-12) Two live items previously tracked ONLY in `archreview/plans/00-MASTER-TRACKER.md`'s registry are folded in here (NF5) so this register is the single tracking place. The tracker's narrative subsections remain as the historical evidence. diff --git a/docs/plans/2026-08-15-arch-review-remediation-execution-log.md b/docs/plans/2026-08-15-arch-review-remediation-execution-log.md index 57c858fa..6a238b86 100644 --- a/docs/plans/2026-08-15-arch-review-remediation-execution-log.md +++ b/docs/plans/2026-08-15-arch-review-remediation-execution-log.md @@ -110,9 +110,20 @@ Deliberately not fixed in this program — each has a stated reason, not an over library's wire-compatibility guarantee — not a blocker for this program. 6. **Fragile `SandboxTests` timing pin.** Pre-existing, unrelated to this remediation's changes; noted so it isn't mistaken for a regression if it flakes later. -7. **Target-scale load test (deferred-work register #25).** This program's exit criterion is the - live probes above, not #25 — #25 remains the follow-on validation that the moved ceilings hold - under real load; schedule separately. +7. ~~**Target-scale load test (deferred-work register #25).**~~ **RESOLVED 2026-08-15.** The + follow-on validation ran: 10 sites x 500 instances x 75 tags = **375,000 live tag + subscriptions**, 37,518 updates/s achieved vs 37,500 nominal, **45,021,375 updates** over a + 20-minute steady-state window. The moved ceilings hold with room to spare — end-to-end tag + latency **P99 4.57 ms**, zero dropped events at 100 live subscribers, health report + collect+ingest **P99 0.31 ms**, debug view **P99 2.19 ms** with no timeouts, CPU **2.9% of the + box**. 11 clean passes, 1 pass with a caveat (F1: 20 minutes with zero gen-2 collections cannot + fully settle the leak question; the 1-hour run would), 0 failures. Register **row 50** (S&F + drain rate + per-subscriber stream backpressure) was absorbed into the same run and closed with + it — S&F drains at **3,533 msg/s**, and a stalled subscriber costs healthy peers **nothing** + (100.00%, zero drops). Results: `docs/plans/2026-08-15-target-scale-load-test-results.md`; + design: `docs/plans/2026-08-15-target-scale-load-test-design.md`; harness: + `tests/ZB.MOM.WW.ScadaBridge.LoadHarness/`. Residual: WP-4's `[xc-3]`/`[xc-4]`/`[xc-5]`/`[xc-8]` + are central-cluster/browser concerns outside this harness and remain open scope. 8. **Playwright 14 pre-existing env failures.** Present on `main` too, rig-state related, not introduced by this branch. 9. **`site_events` retention purge still oplog-visible.** WP3.2's sliced retention DELETE is a row diff --git a/docs/plans/2026-08-15-target-scale-load-test-results.md b/docs/plans/2026-08-15-target-scale-load-test-results.md new file mode 100644 index 00000000..e8a8ce37 --- /dev/null +++ b/docs/plans/2026-08-15-target-scale-load-test-results.md @@ -0,0 +1,289 @@ +# Target-Scale Load Test — Results (Phase-8 WP-4, deferred-work register #25 + row 50) + +**Date:** 2026-08-15 +**Design memo:** `docs/plans/2026-08-15-target-scale-load-test-design.md` +**Harness:** `tests/ZB.MOM.WW.ScadaBridge.LoadHarness/` +**Closes:** deferred-work register **#25** and **row 50**; residual **7** of +`docs/plans/2026-08-15-arch-review-remediation-execution-log.md`. + +> Register row 25 recorded that WP-4 had been *claimed complete* on the strength of a +> 107-byte checklist stub with no per-work-package results and no linked run. This +> document is the run that stub never had. + +--- + +## 1. Environment + +| | | +|---|---| +| Host | `Josephs-MBP` — Apple M4 Pro, 14 logical cores, 48 GB RAM | +| OS | macOS 26.5.2 (Darwin) | +| Runtime | .NET 10.0.5, **server GC enabled**, Release build | +| Branch / commit | `target-scale-load-test` @ `20f6b0b9` (harness), `8abebdae` (design memo) | +| Concurrent load | The 8-node `docker/` rig (OrbStack) remained running throughout — the user's live development cluster. Measured baseline: **~86% of one core (~6% of the box)** and ~10-16 GB RSS. The harness figures below are therefore *pessimistic*, not idealised. | +| Scale executed | **10 sites × 500 instances × 75 tags = 375,000 live tag subscriptions** | +| Offered load | 37,500 tag updates/second nominal (one update per tag per 10 s) | +| Measurement window | 20 minutes steady state, after a 2-minute settle | +| Total wall time | 1,510 s (~25 min) | + +Two independent full-scale runs were executed. **Run 1** is the primary dataset. +**Run 2** re-ran the identical protocol with raw per-sample resource capture added +(the only harness change between them) to resolve the memory question in §4. + +--- + +## 2. Verdict summary + +| # | Criterion | Threshold | Measured | Verdict | +|---|---|---|---|---| +| 1 | `[2.5-1]` 10 sites simultaneously operational | all 10 built, ramped, tracked centrally | 10/10 sites tracked by `CentralHealthAggregator` | **PASS** | +| 2 | `[2.5-2]` 500 instances/site, active subscriptions | 5,000 actors; `SkippedNoCallback` = 0 | 5,000 `InstanceActor`s; skipped = **0** | **PASS** | +| 3 | `[2.5-3]` 75 tags/instance (375,000 total) | 375,000 subscribed via the real DCL | 375,000 tag paths across 50 `DataConnectionActor`s | **PASS** | +| 4 | Tag update latency | P99 < 250 ms, P50 < 50 ms | **P50 0.88 ms, P95 2.83 ms, P99 4.57 ms, P99.9 16.04 ms, max 37.41 ms** (1,100,675 samples) | **PASS** (55× margin at P99) | +| 5 | `[xc-1]` message patterns under load | ≥ 95% of nominal, no unexpected loss | **37,518/s achieved vs 37,500 nominal (100.0%)**; 45,021,375 updates offered; **0 events dropped** at healthy subscribers | **PASS** | +| 6 | `[xc-2]` health reports within expected intervals | collect+ingest P99 < 1,000 ms; all sites tracked | **P99 0.31 ms, max 4.60 ms** over 440 reports; 10/10 sites | **PASS** (3,200× margin) | +| 7 | `[xc-6]` debug view without impacting site performance | snapshot P99 < 2,000 ms, 0 timeouts | **P99 2.19 ms, max 2.72 ms**, 264 completed, **0 timeouts** | **PASS** | +| 8 | `[xc-7]` S&F concurrent buffering from many instances | 20,000 messages, 25 origins, drains to 0 | 20,000 buffered at **15,368 msg/s**; residual depth **0** | **PASS** | +| 9 | Row 50 — S&F drain rate | > 500 msg/s | **3,533 msg/s** (20,000 messages in 5.66 s of active drain) | **PASS** (7× margin) | +| 10 | Row 50 — slow-subscriber backpressure | healthy = 100%; stalled loss confined to its own channel | **healthy 100.00% (4/4, zero drops)**; stalled 1.18%, all 197,028 losses in its own bounded channel | **PASS** | +| 11 | Deploy 500 instances to a site | slowest site < 120 s | **2.6 s** | **PASS** (46× margin) | +| 12 | Memory within acceptable bounds | WS slope < 20 MB/min; no monotonic heap climb | WS **+8.83 MB/min**; heap sawtooths (peak 3,625 MB vs 2,313→2,523 MB endpoints) with a **+19.52 MB/min** drift — see finding **F1** | **PASS with a caveat** | +| 13 | CPU within acceptable bounds | mean < 50% of the box | **41% of one core = 2.9% of the box** (peak 55% of one core) | **PASS** (17× margin) | + +**11 clean passes, 1 pass with a documented caveat, 0 failures.** Four WP-4 +sub-criteria were out of this harness's scope — see §6. + +--- + +## 3. Measured detail + +### 3.1 Deployment / ramp + +``` +site fixtures built 0.2s +all 5,000 instance actors 2.7s (10 sites in parallel) +slowest single site 2.6s (500 instances, production staggering) +``` + +The ramp used the production `StartupBatchSize` (20) / `StartupBatchDelayMs` (100) +pacing, i.e. 25 batches × 100 ms = 2.5 s of *deliberate* delay per site. So the 2.6 s +figure is almost entirely the configured stagger; actual construction of 500 +`InstanceActor`s (each deserializing a 75-attribute `FlattenedConfiguration`, loading +static overrides from SQLite and issuing a DCL subscribe) costs ~0.1 s. Against a +120 s budget this criterion is not close to binding. + +### 3.2 Tag update latency (DCL boundary → stream subscriber) + +``` +samples 1,100,675 mean 1.17ms +p50 0.88ms p95 2.83ms p99 4.57ms p99.9 16.04ms max 37.41ms +``` + +This is a true end-to-end path measurement, not a synthetic timer: the emit instant is +stamped on the `TagValue` handed to the adapter callback and travels verbatim through +`DataConnectionActor` → `TagValueUpdate.Timestamp` → `AttributeValueChanged.Timestamp` +→ the site stream → `StreamRelayActor` → the proto `SiteStreamEvent`, where the +subscriber subtracts it. Every hop is production code. + +Sub-millisecond median under 37,500 updates/s across 5,000 actors means the actor +hierarchy is nowhere near its ceiling at target scale. + +### 3.3 Stream delivery + +``` +events delivered 900,675 +events dropped 0 +``` + +Zero evictions at the 100 live subscribers across a 20-minute window at full rate. +The site stream's `DropHead` buffering exists for pathological cases; at nominal +target-scale load it never engaged. + +### 3.4 Health reporting at scale + +``` +samples 440 mean 0.03ms p50 0.01ms p95 0.04ms p99 0.31ms max 4.60ms +reports ingested 440 sites tracked centrally 10 +``` + +`SiteHealthCollector.CollectReport` for a site carrying 37,500 subscriptions plus +`CentralHealthAggregator.ProcessReport` costs well under a millisecond. Against the +30-second report interval there is no plausible queueing risk. + +### 3.5 Debug view under load + +``` +samples 264 mean 0.14ms p50 0.08ms p99 2.19ms max 2.72ms +completed 264 timed out 0 +``` + +Each sample is an `Ask` landing in the mailbox of an +`InstanceActor` concurrently ingesting tag updates, so the figure includes real +queueing behind production traffic. `[xc-6]`'s "without impacting site performance" +also holds in the other direction: tag latency percentiles were measured with these +snapshots running throughout, and show no degradation. + +### 3.6 Resources + +``` +window 1200s over 121 samples +working set 2665 -> 3261 MB (peak 3261) slope +8.83 MB/min +managed heap 2313 -> 2523 MB (peak 3625) slope +19.52 MB/min +cpu mean/peak 41% / 55% of one core (2.9% of the box) +threads 58 gen2 GCs in window: 0 +``` + +CPU is the headline: **375,000 subscriptions at 37,500 updates/s consumed under half +of one core of fourteen.** Thread count stayed flat at 58 for ten actor systems. + +--- + +## 4. Findings + +### F1 — Memory: 20 minutes with zero gen-2 collections cannot fully settle the leak question (Low) + +The working-set slope (+8.83 MB/min) is inside the threshold, and the managed heap +clearly *sawtooths* — its peak (3,625 MB) sits well above both endpoints (2,313 MB → +2,523 MB), so the collector is demonstrably reclaiming. But **`GC.CollectionCount(2)` +recorded zero gen-2 collections across the entire 25-minute run**, at 45 million +events. Gen-2 was therefore never compacted, and a positive least-squares drift on an +uncompacted heap cannot be distinguished from a genuine slow leak by this run alone. + +This is a limitation of the shortened window (deviation D1), not an observed defect: +no metric degraded, no drop appeared, and latency percentiles were flat from the first +minute to the twentieth. Recorded honestly rather than reported as "no leak". + +**To settle it**, run the full 1-hour protocol (§7) and check whether the heap slope +flattens once gen-2 collections begin, or issue an explicit +`GC.Collect(2, Forced, blocking: true)` at window start and end and compare. +Deliberately *not* attempted here — the brief was to measure, not to tune. + +### F2 — Store-and-forward waits one full `DefaultRetryInterval` before draining a deferred backlog (informational, by design) + +The drain measurement initially read as **33 msg/s**, which would have been alarming. +The progress series showed why: **nothing drained for 29-30 seconds, then the entire +backlog cleared in a fraction of a second.** + +Cause, confirmed in source: `StoreAndForwardService.EnqueueAsync` with +`attemptImmediateDelivery: false` stamps `message.LastAttemptAt = UtcNow` +(`StoreAndForwardService.cs:754-757`). `GetMessagesForRetryAsync`'s due-predicate is +`last_attempt_at_ms IS NULL OR retry_interval_ms = 0 OR (now - last_attempt) >= retry_interval`, +so such a row is not due for one `DefaultRetryInterval` (30 s). Explicit +`TriggerSweep()` calls find nothing due and correctly do nothing. + +This is **correct, intended behaviour** — the caller declared delivery was not +attempted, so the row waits one retry interval — but it is easy to misread as slow +drainage. The harness therefore reports the two numbers separately: + +``` +retry wait before drain 28.9s (DefaultRetryInterval) +drain throughput 3,533 msg/s (active drain 5.66s) +``` + +Worth knowing operationally: a site that buffers a backlog this way will appear +completely stalled for the first 30 seconds. Note the `Notify.Send` path +(`deferToSweep: true`) deliberately leaves `LastAttemptAt` null so its rows are due +immediately and skip this wait entirely. + +### F3 — Slow-subscriber isolation is complete, and the mechanism is worth recording (positive result) + +Register row 50 asked what a slow/stalled gRPC subscriber does to per-subscriber +buffering with several subscribers attached. With five subscribers on the *same* +instance (identical offered sequence) and one reader stalled at 50 ms/event: + +| Subscriber | State | Received | Dropped | Delivery | +|---|---|---|---|---| +| slowprobe-00 | **STALLED** | 2,369 | 197,028 | 1.18% | +| slowprobe-01 | healthy | 200,000 | 0 | **100.00%** | +| slowprobe-02 | healthy | 200,000 | 0 | **100.00%** | +| slowprobe-03 | healthy | 200,000 | 0 | **100.00%** | +| slowprobe-04 | healthy | 200,000 | 0 | **100.00%** | + +**Isolation is total** — not merely "good". The stalled subscriber lost 98.8% of its +own events and cost its peers exactly nothing. The reason is structural, and each link +matters: + +1. The per-subscriber graph is `Where → Buffer(StreamBufferSize, DropHead) → Sink.ForEach(Tell)`. + A `DropHead` buffer *always* accepts, so it never backpressures the `BroadcastHub` + upstream — one subscriber cannot stall the hub for the others. +2. `Sink.ForEach` does an actor `Tell`, which never blocks. +3. `StreamRelayActor.WriteToChannel` uses `TryWrite` on a bounded `DropOldest` + channel, which never blocks either — so the relay actor's mailbox drains regardless + of reader speed, and loss is confined to that subscriber's own channel where it is + counted. + +The consequence for operators: a slow WAN link or wedged central client degrades +**only its own feed**, and the loss is visible in the per-stream eviction counter and +the `RecordSiteStreamEventDropped` telemetry rather than being silent. + +One caveat established while building the probe: publishing an unpaced burst makes +*every* subscriber lose events, because the publish source is a single +`Source.ActorRef(StreamBufferSize, DropHead)` **shared** by all attribute subscribers, +upstream of the hub. That shared stage — not per-subscriber buffering — is the real +limit on burst absorption. The probe paces at 2,000 events/s to isolate the variable +under test; the shared stage's own capacity is separately covered by +`SiteStreamThroughputTests`. + +--- + +## 5. What this does *not* prove + +Stated plainly so the evidence is not over-read: + +- **Not a clustered run.** Sites are single, non-clustered `ActorSystem`s (deviation + D2). Failover, singleton handover and split-brain behaviour are covered elsewhere + (`FailoverTimingTests`, `SbrFailoverTests`, `docker/failover-drill.sh`). +- **Not a real-network run.** The gRPC, health-transport and S&F-to-central hops are + in-process. Serialization cost, TLS, WAN latency and socket backpressure are out of + frame; the *stream* path's backpressure was measured deliberately by substituting a + controllable reader for the socket (D4). +- **Not a real OPC UA run.** 375,000 monitored items were simulated at the adapter + callback (D3). The DCL above that callback is real and fully exercised. +- **Not 1 hour.** 20 minutes (D1). See finding F1 for the one question this leaves + genuinely open. + +## 6. WP-4 criteria not covered by this harness + +Recorded rather than silently dropped (deviation D7): + +| Criterion | Why not covered | +|---|---| +| `[xc-3]` site event logging within 30-day / 1 GB limits | Needs a long-horizon retention run, not a load run | +| `[xc-4]` audit logging does not degrade central | Needs a central MS SQL fixture; this harness builds site runtimes only | +| `[xc-5]` template flattening/validation for large templates | Template Engine is a central-cluster concern; unrelated to the site-runtime load path | +| `[xc-8]` UI workflows remain responsive | Needs a browser + a live central cluster (Playwright territory) | + +These remain open scope for WP-4 and should be tracked separately rather than being +considered closed by this run. + +## 7. Reproducing + +Full protocol as executed (20-minute window): + +```bash +dotnet run -c Release --project tests/ZB.MOM.WW.ScadaBridge.LoadHarness -- \ + --results loadharness-results.json +``` + +The full **1-hour** version required by the WP-4 test protocol — identical in every +other respect, and the run that would settle finding F1: + +```bash +dotnet run -c Release --project tests/ZB.MOM.WW.ScadaBridge.LoadHarness -- \ + --sustain-minutes 60 --results loadharness-results-1h.json +``` + +CI-scale smoke (~80 s), which asserts the harness still produces coherent +measurements end to end: + +```bash +dotnet test tests/ZB.MOM.WW.ScadaBridge.PerformanceTests \ + --filter "FullyQualifiedName~TargetScaleHarnessSmoke" +``` + +`--help`-style knobs: `--sites`, `--instances-per-site`, `--tags-per-instance`, +`--tag-update-period-seconds`, `--settle-minutes`, `--sustain-minutes`, +`--sample-seconds`, `--health-interval-seconds`, `--debug-probe-interval-seconds`, +`--subscribe-settle-seconds`, `--stream-probes-per-site`, `--sf-drain-messages`, +`--slow-subscriber-events`, `--data-dir`, `--results`. An unknown key is rejected +rather than ignored, so a typo cannot silently change the measured scale. diff --git a/docs/plans/phase-8-checklist.md b/docs/plans/phase-8-checklist.md index 9b4a6954..f9a81727 100644 --- a/docs/plans/phase-8-checklist.md +++ b/docs/plans/phase-8-checklist.md @@ -3,21 +3,46 @@ **Update 2026-08-07 (truth sweep):** This checklist previously read "Status: Complete / Tests: All passing / Build: 0 errors, 0 warnings" — a 107-byte stub with no per-work-package results and no linked run. That claim -was **unevidenced** and has been retired. +was **unevidenced** and was retired. -Honest state, per the deferred-work register -(`2026-07-08-deferred-work-register.md`, row 25): +**Update 2026-08-15 (WP-4 run, evidenced):** The target-scale load test has now +actually been run, at full scale, with results published. -- The **Phase-8 WP-4 target-scale load test** (10 sites × 500 instances × - 75 tags = 37,500 subscriptions/site, ~375,000 total) has **never been run**. -- Nearest real coverage is arithmetic/aggregation only — - `PerformanceTests/StaggeredStartupTests.cs` - (`TagCapacity_75TagsPer500Machines_37500Total`) and - `HealthAggregationTests` — plus a single-subscriber 100k-event - `Streaming/SiteStreamThroughputTests.cs`. No sustained multi-site run - exists anywhere in `tests/` or `docker/`. -- Revisit trigger: before any production go-live at target scale, or the - first site approaching ~500 instances / ~37.5k subscriptions. +## WP-4 — Load/Performance Testing at Target Scale + +**Status: measured 2026-08-15.** Not a claim — a run. + +- Scale executed: **10 sites × 500 instances × 75 tags = 375,000 live tag + subscriptions**, 37,518 tag updates/second achieved against 37,500 nominal + (100.0%), **45,021,375 updates** offered over a 20-minute steady-state window. +- Headline results: end-to-end tag update latency **P50 0.88 ms / P99 4.57 ms / + max 37.41 ms**; **zero events dropped** at 100 live stream subscribers; health + report collect+ingest **P99 0.31 ms** with 10/10 sites tracked centrally; + debug view snapshot under load **P99 2.19 ms with 0 timeouts**; 500 instances + deployed to a site in **2.6 s**; CPU **2.9% of the box**; working-set slope + **+8.83 MB/min**. +- **11 clean passes, 1 pass with a documented caveat, 0 failures.** +- Deferred-work register **#25** and **row 50** are closed by this run. + +**Results:** `2026-08-15-target-scale-load-test-results.md` — per-criterion +numbers vs thresholds, three findings (F1 memory-window limitation, F2 the S&F +retry-interval wait, F3 slow-subscriber isolation), and an explicit "what this +does not prove" section. + +**Design, thresholds and deviations:** +`2026-08-15-target-scale-load-test-design.md` — including the documented +deviation from the protocol's 1-hour sustained window to 20 minutes, and the +command for the full 1-hour version. + +**Harness:** `tests/ZB.MOM.WW.ScadaBridge.LoadHarness/` (standalone executable), +with a CI-scale smoke `[Fact]` at +`tests/ZB.MOM.WW.ScadaBridge.PerformanceTests/TargetScale/TargetScaleHarnessSmokeTests.cs` +so the evidence can be regenerated rather than bit-rotting. + +**Still open within WP-4** (not covered by this harness — central-cluster and +browser concerns): `[xc-3]` site event logging volume within 30-day/1 GB limits, +`[xc-4]` audit logging not degrading central, `[xc-5]` template +flattening/validation for large templates, `[xc-8]` UI workflow responsiveness. See `phase-8-production-readiness.md:152-170` (WP-4) and `:314-320` (test protocol) for the original scope. diff --git a/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/HarnessRun.cs b/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/HarnessRun.cs index 26a3d521..9947a785 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/HarnessRun.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/HarnessRun.cs @@ -24,6 +24,12 @@ namespace ZB.MOM.WW.ScadaBridge.LoadHarness; /// Emissions skipped before subscriptions existed. /// Resource behaviour over the measurement window. /// Resource behaviour over the whole run. +/// +/// Every raw resource sample. Retained because the summary alone cannot distinguish a +/// slow leak from a heap that simply sawtooths — especially when the run records zero +/// gen-2 collections, where a positive least-squares slope may be nothing more than +/// gen-2 garbage that was never collected. +/// /// Health report collect+ingest latency. /// Health reports ingested by the central aggregator. /// Sites the central aggregator ended up tracking. @@ -51,6 +57,7 @@ public sealed record HarnessRunResult( long DriverSkippedNoCallback, ResourceWindowSummary? SteadyStateResources, ResourceWindowSummary? WholeRunResources, + IReadOnlyList ResourceSamples, LatencySnapshot HealthReportLatency, long HealthReportsDelivered, int SitesTrackedByAggregator, @@ -233,6 +240,7 @@ public static class HarnessRun DriverSkippedNoCallback: driverSkipped, SteadyStateResources: steadyResources, WholeRunResources: sampler.Summarize(0, double.MaxValue), + ResourceSamples: sampler.Snapshot(), HealthReportLatency: probes.HealthReportLatency.Snapshot(), HealthReportsDelivered: probes.HealthReportsDelivered, SitesTrackedByAggregator: aggregator.GetAllSiteStates().Count, From 94e8301e445d2655e9e30f8dfb4f7fb6d3d0b032 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Sat, 15 Aug 2026 03:01:17 -0400 Subject: [PATCH 4/4] test(loadharness): retain raw resource samples; record findings F4 and the discarded run HarnessRunResult now carries every ResourceSample, not just the window summaries. Motivated by finding F1: the definitive run recorded ZERO gen-2 collections across 45M events, so a positive least-squares heap slope cannot be told apart from gen-2 garbage that was simply never collected. The summary alone cannot settle that; the series can. The reported run predates this field -- noted as such in the results doc rather than implied otherwise. Also records: - the second full-scale run was DISCARDED, not reported: a verification build overlapped the start of its measurement window, and a contaminated measurement is not evidence. - finding F4, a pre-existing test-isolation flake in QueueDepthGaugeTests.Gauge_TracksBufferedDepth_AcrossEnqueueDrainAndPark (fails in a full-suite run, passes in isolation -- shared static gauge carrying state across tests). It cannot originate here: this branch changes zero src/ files vs its base 986e6e7a. Left unfixed deliberately; filed for separate triage. Verified: full slnx build clean; SiteRuntime 604/604, Communication 691/691 pass; TargetScaleHarnessSmokeTests passes (78s). --- ...26-08-15-target-scale-load-test-results.md | 27 ++++++++++++++++--- 1 file changed, 24 insertions(+), 3 deletions(-) diff --git a/docs/plans/2026-08-15-target-scale-load-test-results.md b/docs/plans/2026-08-15-target-scale-load-test-results.md index e8a8ce37..a77d5892 100644 --- a/docs/plans/2026-08-15-target-scale-load-test-results.md +++ b/docs/plans/2026-08-15-target-scale-load-test-results.md @@ -26,9 +26,13 @@ | Measurement window | 20 minutes steady state, after a 2-minute settle | | Total wall time | 1,510 s (~25 min) | -Two independent full-scale runs were executed. **Run 1** is the primary dataset. -**Run 2** re-ran the identical protocol with raw per-sample resource capture added -(the only harness change between them) to resolve the memory question in §4. +**One clean full-scale run is the dataset reported here.** A second run was started to +add raw per-sample resource capture (sharpening finding F1), but a verification build +overlapped the start of its measurement window; it was **discarded rather than +reported**, per the rule that a measurement contaminated by concurrent load is not +evidence. The raw-sample capture it was meant to exercise is now in the harness +(`HarnessRunResult.ResourceSamples`) and will be present in any future run, including +the 1-hour run in §7 — the run reported below predates that field. --- @@ -224,6 +228,23 @@ limit on burst absorption. The probe paces at 2,000 events/s to isolate the vari under test; the shared stage's own capacity is separately covered by `SiteStreamThroughputTests`. +### F4 — Pre-existing test-isolation flake in `QueueDepthGaugeTests` (Low, not introduced here) + +Noticed during this work package's verification pass, recorded so it is not later +mistaken for a regression from the load-harness branch. + +`StoreAndForward.Tests.QueueDepthGaugeTests.Gauge_TracksBufferedDepth_AcrossEnqueueDrainAndPark` +**fails in a full-suite run** (`Expected: 0, Actual: 2`, `QueueDepthGaugeTests.cs:116`) +but **passes when run in isolation** — the signature of a shared static gauge carrying +state across tests, i.e. an ordering dependency rather than a product defect. + +It cannot originate from this branch: the branch changes **zero `src/` files** relative +to its base (`986e6e7a`), and both the StoreAndForward source and its test project are +byte-identical to that base. Left unfixed deliberately — out of scope for a measurement +work package, and the brief was explicitly not to attempt unrelated fixes. Worth a +separate triage alongside the pre-existing `SandboxTests` timing pin already recorded as +residual 6 of the arch-review remediation execution log. + --- ## 5. What this does *not* prove