ScadaBridge/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/Scenarios/SlowSubscriberScenario.cs

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;

/// <summary>Per-subscriber outcome from the slow-subscriber isolation probe.</summary>
/// <param name="Name">Probe name.</param>
/// <param name="IsSlow">Whether this probe's reader was deliberately stalled.</param>
/// <param name="Received">Events the reader drained.</param>
/// <param name="Dropped">Events evicted by this probe's bounded DropOldest channel.</param>
/// <param name="DeliveryRatio">Received / published, before accounting for the site-stream buffer.</param>
public sealed record SubscriberOutcome(
    string Name,
    bool IsSlow,
    long Received,
    long Dropped,
    double DeliveryRatio);

/// <summary>Result of the slow-subscriber isolation measurement.</summary>
/// <param name="PublishedEvents">Events published to the site stream during the probe.</param>
/// <param name="PublishSeconds">Wall time the publisher took.</param>
/// <param name="PublishPerSecond">Publish throughput observed by the producer.</param>
/// <param name="Outcomes">Per-subscriber outcomes.</param>
/// <param name="HealthyMinDeliveryRatio">Worst delivery ratio among the healthy subscribers.</param>
/// <param name="SlowDeliveryRatio">Delivery ratio of the stalled subscriber.</param>
public sealed record SlowSubscriberResult(
    int PublishedEvents,
    double PublishSeconds,
    double PublishPerSecond,
    IReadOnlyList<SubscriberOutcome> Outcomes,
    double HealthyMinDeliveryRatio,
    double SlowDeliveryRatio);

/// <summary>
/// Register row 50, second half: what does a slow or stalled gRPC subscriber do to
/// per-subscriber buffering when several subscribers are attached?
///
/// <para>
/// 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 <c>SiteStreamManager</c>.
/// </para>
/// <para>
/// 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 <c>Buffer(DropHead)</c> stage and its own bounded
/// <c>DropOldest</c> channel? The design intends the latter; this measures it.
/// </para>
/// </summary>
public static class SlowSubscriberScenario
{
    /// <summary>Per-event reader delay applied to the stalled subscriber.</summary>
    public const int SlowReaderDelayMicroseconds = 50_000;

    /// <summary>
    /// Publish rate for the probe. Deliberately paced rather than a tight burst: the
    /// publish source is a single <c>Source.ActorRef(StreamBufferSize, DropHead)</c>
    /// 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.
    /// </summary>
    public const int PublishRatePerSecond = 2_000;

    /// <summary>Runs the isolation probe on a dedicated site.</summary>
    /// <param name="site">Site whose stream manager is used.</param>
    /// <param name="subscriberCount">Total subscribers to attach (one of them is stalled).</param>
    /// <param name="eventCount">Events to publish.</param>
    /// <param name="cancellationToken">Cancels the measurement.</param>
    /// <returns>The measured result.</returns>
    public static async Task<SlowSubscriberResult> RunAsync(
        SiteRuntimeFixture site,
        int subscriberCount,
        int eventCount,
        CancellationToken cancellationToken)
    {
        var instanceName = site.InstanceName(0);
        var probes = new List<StreamSubscriberProbe>(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();
        }
    }
}