ScadaBridge/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/HarnessRun.cs

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;

/// <summary>Everything one harness run measured, ready for serialization.</summary>
/// <param name="Config">The configuration the run executed under.</param>
/// <param name="Environment">Host/runtime description.</param>
/// <param name="StartedUtc">Run start.</param>
/// <param name="TotalSeconds">Total wall time including ramp and teardown.</param>
/// <param name="SiteRampSeconds">Wall time to build all site fixtures (before instances).</param>
/// <param name="InstanceRampSeconds">Wall time to create every Instance Actor across all sites.</param>
/// <param name="SlowestSiteInstanceRampSeconds">Slowest single site's instance ramp — the "deploy 500 instances to a site" figure.</param>
/// <param name="TagUpdateLatency">End-to-end DCL-boundary to stream-subscriber latency.</param>
/// <param name="EmittedTagUpdates">Tag updates offered during the whole run.</param>
/// <param name="SteadyStateEmittedTagUpdates">Tag updates offered during the measurement window only.</param>
/// <param name="AchievedUpdatesPerSecond">Offered load actually achieved in the measurement window.</param>
/// <param name="NominalUpdatesPerSecond">Offered load the configuration called for.</param>
/// <param name="DriverLagSeconds">Cumulative driver slice overrun (harness-bound load shortfall).</param>
/// <param name="DriverSkippedNoCallback">Emissions skipped before subscriptions existed.</param>
/// <param name="SteadyStateResources">Resource behaviour over the measurement window.</param>
/// <param name="WholeRunResources">Resource behaviour over the whole run.</param>
/// <param name="ResourceSamples">
/// 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.
/// </param>
/// <param name="HealthReportLatency">Health report collect+ingest latency.</param>
/// <param name="HealthReportsDelivered">Health reports ingested by the central aggregator.</param>
/// <param name="SitesTrackedByAggregator">Sites the central aggregator ended up tracking.</param>
/// <param name="DebugSnapshotLatency">Debug view snapshot round-trip latency under load.</param>
/// <param name="DebugSnapshotsCompleted">Debug snapshots that completed.</param>
/// <param name="DebugSnapshotTimeouts">Debug snapshots that timed out.</param>
/// <param name="StreamProbeReceived">Events delivered to live stream subscribers.</param>
/// <param name="StreamProbeDropped">Events evicted by live subscribers' bounded channels.</param>
/// <param name="StoreAndForwardDrain">Store-and-forward drain measurement (register row 50).</param>
/// <param name="SlowSubscriber">Slow-subscriber isolation measurement (register row 50).</param>
/// <param name="WireStream">
/// Wire-path stream measurement — real Kestrel + real <c>SiteStreamGrpcClient</c> with
/// R2 batching negotiated — or null when <see cref="HarnessConfig.WireProbesPerSite"/> is 0.
/// </param>
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,
    IReadOnlyList<ResourceSample> ResourceSamples,
    LatencySnapshot HealthReportLatency,
    long HealthReportsDelivered,
    int SitesTrackedByAggregator,
    LatencySnapshot DebugSnapshotLatency,
    long DebugSnapshotsCompleted,
    long DebugSnapshotTimeouts,
    long StreamProbeReceived,
    long StreamProbeDropped,
    StoreAndForwardDrainResult? StoreAndForwardDrain,
    SlowSubscriberResult? SlowSubscriber,
    WireStreamResult? WireStream);

/// <summary>
/// Steady-state-window measurement of the wire-path stream probes: the production
/// gRPC server (Kestrel h2c) and client with the R2 batcher, protobuf serialization
/// and HTTP/2 flow control INSIDE the measured path. All figures are deltas over the
/// steady-state window only.
/// </summary>
/// <param name="ProbeCount">Wire subscribers attached across the fleet.</param>
/// <param name="Latency">End-to-end DCL-boundary → post-unpack client callback latency.</param>
/// <param name="EventsReceived">Domain events delivered by the client in the window.</param>
/// <param name="ServerDroppedEvents">Events evicted by the wire servers' bounded send channels in the window.</param>
/// <param name="StreamFaults">Stream faults reported by the client (expected 0).</param>
/// <param name="GracefulCompletions">Graceful server-side stream completions (expected 0 inside the 4 h lifetime).</param>
/// <param name="Batches">The frame/batch-size distribution the R2 batcher actually produced.</param>
public sealed record WireStreamResult(
    int ProbeCount,
    LatencySnapshot Latency,
    long EventsReceived,
    long ServerDroppedEvents,
    long StreamFaults,
    long GracefulCompletions,
    BatchSizeWindow Batches);

/// <summary>Host and runtime facts recorded alongside the numbers.</summary>
/// <param name="MachineName">Host name.</param>
/// <param name="OsDescription">Operating system description.</param>
/// <param name="ProcessorCount">Logical processors visible to the process.</param>
/// <param name="RuntimeVersion">.NET runtime version.</param>
/// <param name="ServerGc">Whether server GC is active.</param>
public sealed record EnvironmentInfo(
    string MachineName,
    string OsDescription,
    int ProcessorCount,
    string RuntimeVersion,
    bool ServerGc);

/// <summary>
/// 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 deferred-register scenarios on a dedicated site.
/// </summary>
public static class HarnessRun
{
    /// <summary>Data connections each site spreads its tags across.</summary>
    public const int ConnectionsPerSite = 5;

    /// <summary>Subscribers attached in the slow-subscriber isolation probe.</summary>
    public const int SlowSubscriberProbeCount = 5;

    /// <summary>Concurrent enqueue tasks in the store-and-forward drain probe.</summary>
    public const int StoreAndForwardConcurrency = 25;

    /// <summary>Executes a run end to end.</summary>
    /// <param name="config">Scale and duration configuration.</param>
    /// <param name="log">Progress sink (stdout in the console app).</param>
    /// <param name="cancellationToken">Cancels the run.</param>
    /// <returns>The measured result.</returns>
    public static async Task<HarnessRunResult> ExecuteAsync(
        HarnessConfig config, Action<string> 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<SiteRuntimeFixture>(config.Sites);
        var latency = new LatencyHistogram();

        TagUpdateDriver? driver = null;
        ObservabilityProbes? probes = null;
        StoreAndForwardDrainResult? drainResult = null;
        SlowSubscriberResult? slowResult = null;
        BatchSizeAggregator? batchAggregator = null;

        var aggregator = new CentralHealthAggregator(
            Options.Create(new HealthMonitoringOptions
            {
                ReportInterval = config.HealthReportInterval,
                OfflineTimeout = config.HealthReportInterval * 2,
            }),
            NullLogger<CentralHealthAggregator>.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");

            // Wire path (real gRPC): the aggregator must exist BEFORE the first stream
            // opens or the MeterListener misses early frames.
            var wireLatency = new LatencyHistogram();
            if (config.WireProbesPerSite > 0)
            {
                batchAggregator = new BatchSizeAggregator();
                foreach (var site in sites)
                    await site.AttachWireStreamProbesAsync(config.WireProbesPerSite, wireLatency, log);
                log($"  {sites.Sum(s => s.WireSubscribers.Count)} wire stream subscribers attached " +
                    "(real Kestrel + SiteStreamGrpcClient, batching negotiated)");
            }

            // 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);
            }

            // Wire path: same ramp/steady histogram separation, plus window-start
            // baselines for the client/server counters and the batch distribution.
            var steadyWireLatency = new LatencyHistogram();
            var wireReceivedAtWindowStart = sites.Sum(s => s.WireSubscribers.Sum(p => p.ReceivedEvents));
            var wireDroppedAtWindowStart = sites.Sum(s => s.WireHost?.Server.DroppedStreamEventCount ?? 0);
            var batchesAtWindowStart = batchAggregator?.Snapshot();
            foreach (var site in sites)
            {
                foreach (var wire in site.WireSubscribers)
                    wire.RetargetLatency(steadyWireLatency);
            }

            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;

            // Wire-path window deltas, captured BEFORE the deferred-register scenarios below so
            // their publishing cannot contaminate the wire figures.
            WireStreamResult? wireResult = null;
            if (batchAggregator is not null)
            {
                wireResult = new WireStreamResult(
                    ProbeCount: sites.Sum(s => s.WireSubscribers.Count),
                    Latency: steadyWireLatency.Snapshot(),
                    EventsReceived: sites.Sum(s => s.WireSubscribers.Sum(p => p.ReceivedEvents)) - wireReceivedAtWindowStart,
                    ServerDroppedEvents: sites.Sum(s => s.WireHost?.Server.DroppedStreamEventCount ?? 0) - wireDroppedAtWindowStart,
                    StreamFaults: sites.Sum(s => s.WireSubscribers.Sum(p => p.Faults)),
                    GracefulCompletions: sites.Sum(s => s.WireSubscribers.Sum(p => p.GracefulCompletions)),
                    Batches: BatchSizeWindow.From(batchesAtWindowStart!, batchAggregator.Snapshot()));
                log($"  wire path: {wireResult.EventsReceived:N0} events in {wireResult.Batches.Frames:N0} frames " +
                    $"(mean {wireResult.Batches.MeanEventsPerFrame:F1} ev/frame, max {wireResult.Batches.Max}), " +
                    $"P99 {wireResult.Latency.P99Ms:F2}ms");
            }

            // 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),
                ResourceSamples: sampler.Snapshot(),
                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,
                WireStream: wireResult);
        }
        finally
        {
            if (driver != null) await driver.DisposeAsync();
            if (probes != null) await probes.DisposeAsync();
            batchAggregator?.Dispose();
            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);
}