ScadaBridge/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/Probes/StreamSubscriberProbe.cs

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;

/// <summary>
/// A live site-stream subscriber assembled from the SAME parts
/// <c>SiteStreamGrpcServer.RunSubscriptionStreamAsync</c> uses:
///
/// <list type="number">
///   <item><description><c>SiteStreamManager.Subscribe</c> — materializes the
///   per-subscriber graph (<c>Where</c> instance filter → <c>Buffer(StreamBufferSize,
///   DropHead)</c> → <c>Sink.ForEach(Tell)</c>).</description></item>
///   <item><description>A real <see cref="StreamRelayActor"/>, which converts the Akka
///   record to the protobuf <c>SiteStreamEvent</c> and <c>TryWrite</c>s it.</description></item>
///   <item><description>A bounded <c>DropOldest</c> <see cref="Channel"/> of the
///   production capacity (<c>GrpcInstanceStreamChannelCapacity</c> = 1000) with the
///   eviction counter wired to <see cref="DroppedEvents"/>.</description></item>
/// </list>
///
/// <para>
/// The single substitution is the final hop: instead of
/// <c>responseStream.WriteAsync</c> 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.
/// </para>
/// </summary>
public sealed class StreamSubscriberProbe : IAsyncDisposable
{
    /// <summary>Production per-instance stream channel capacity (<c>GrpcInstanceStreamChannelCapacity</c>).</summary>
    public const int ProductionChannelCapacity = 1000;

    private readonly SiteStreamManager _manager;
    private readonly string _subscriptionId;
    private readonly IActorRef _relayActor;
    private readonly ActorSystem _system;
    private readonly Channel<SiteStreamEvent> _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;

    /// <summary>Human-readable probe name (also the relay actor's name suffix).</summary>
    public string Name { get; }

    /// <summary>Events evicted by the bounded channel's DropOldest policy.</summary>
    public long DroppedEvents => _dropCounter.Value;

    /// <summary>Events successfully drained by the reader (i.e. "sent to the client").</summary>
    public long ReceivedEvents => Interlocked.Read(ref _received);

    /// <summary>
    /// 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
    /// <c>PublishAttributeValueChanged</c> short-circuits) and risk reusing an actor
    /// name whose previous incarnation has not finished terminating.
    /// </summary>
    /// <param name="latency">The histogram to record into from now on, or null to stop recording.</param>
    public void RetargetLatency(LatencyHistogram? latency) => _latency = latency;

    /// <summary>
    /// Artificial per-event reader delay, in microseconds. Zero is a healthy
    /// subscriber; a large value models a stalled WAN link or a wedged client.
    /// </summary>
    public long ReaderDelayMicroseconds
    {
        get => Interlocked.Read(ref _readerDelayMicroseconds);
        set => Interlocked.Exchange(ref _readerDelayMicroseconds, value);
    }

    private StreamSubscriberProbe(
        ActorSystem system,
        SiteStreamManager manager,
        string name,
        Channel<SiteStreamEvent> 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));
    }

    /// <summary>
    /// Builds and attaches a probe subscribed to one instance's events.
    /// </summary>
    /// <param name="system">The site actor system.</param>
    /// <param name="manager">The site stream manager to subscribe against.</param>
    /// <param name="instanceUniqueName">Instance whose events this probe receives.</param>
    /// <param name="name">Probe name, used for the relay actor's path.</param>
    /// <param name="latency">Optional histogram fed with end-to-end event latency.</param>
    /// <returns>The attached probe.</returns>
    public static StreamSubscriberProbe Attach(
        ActorSystem system,
        SiteStreamManager manager,
        string instanceUniqueName,
        string name,
        LatencyHistogram? latency)
    {
        var dropCounter = new DropCounter();
        var channel = Channel.CreateBounded<SiteStreamEvent>(
            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.
        }
    }

    /// <summary>Detaches the subscription and stops the relay actor and reader.</summary>
    /// <returns>A task that completes when the probe is torn down.</returns>
    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();
    }
}

/// <summary>
/// Thread-safe counter for a bounded channel's <c>itemDropped</c> callback. A tiny
/// class rather than a captured local so the probe and the channel share exactly one
/// counter instance without a second closure.
/// </summary>
public sealed class DropCounter
{
    private long _value;

    /// <summary>Current count.</summary>
    public long Value => Interlocked.Read(ref _value);

    /// <summary>Increments the counter.</summary>
    public void Increment() => Interlocked.Increment(ref _value);
}