mxaccessgw/src/ZB.MOM.WW.MxGateway.Server/Grpc/EventStreamService.cs

using System.Runtime.CompilerServices;
using System.Threading.Channels;
using Microsoft.Extensions.Options;
using ZB.MOM.WW.MxGateway.Contracts.Proto;
using ZB.MOM.WW.MxGateway.Server.Configuration;
using ZB.MOM.WW.MxGateway.Server.Dashboard.Hubs;
using ZB.MOM.WW.MxGateway.Server.Metrics;
using ZB.MOM.WW.MxGateway.Server.Sessions;
using ZB.MOM.WW.MxGateway.Server.Workers;

namespace ZB.MOM.WW.MxGateway.Server.Grpc;

public sealed class EventStreamService(
    ISessionManager sessionManager,
    IOptions<GatewayOptions> options,
    MxAccessGrpcMapper mapper,
    GatewayMetrics metrics,
    IDashboardEventBroadcaster dashboardEventBroadcaster,
    ILogger<EventStreamService> logger) : IEventStreamService
{
    /// <summary>
    /// Streams events from a session to the client asynchronously.
    /// </summary>
    /// <param name="request">Stream events request.</param>
    /// <param name="cancellationToken">Cancellation token.</param>
    /// <returns>Async enumerable of MX events.</returns>
    public async IAsyncEnumerable<MxEvent> StreamEventsAsync(
        StreamEventsRequest request,
        [EnumeratorCancellation] CancellationToken cancellationToken)
    {
        if (!sessionManager.TryGetSession(request.SessionId, out GatewaySession? session) || session is null)
        {
            throw new SessionManagerException(
                SessionManagerErrorCode.SessionNotFound,
                $"Session {request.SessionId} was not found.");
        }

        using IDisposable subscriber = session.AttachEventSubscriber(
            options.Value.Sessions.AllowMultipleEventSubscribers);
        using CancellationTokenSource streamCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);

        int streamQueueDepth = 0;
        Channel<MxEvent> eventQueue = Channel.CreateBounded<MxEvent>(
            new BoundedChannelOptions(options.Value.Events.QueueCapacity)
            {
                SingleReader = true,
                SingleWriter = true,
                FullMode = BoundedChannelFullMode.Wait,
                AllowSynchronousContinuations = false,
            });
        Task producerTask = ProduceEventsAsync(
            session,
            request.AfterWorkerSequence,
            eventQueue.Writer,
            () =>
            {
                Interlocked.Increment(ref streamQueueDepth);
                metrics.AdjustGrpcEventStreamQueueDepth(1);
            },
            streamCts.Token);

        try
        {
            await foreach (MxEvent mxEvent in eventQueue.Reader.ReadAllAsync(cancellationToken).ConfigureAwait(false))
            {
                Interlocked.Decrement(ref streamQueueDepth);
                metrics.AdjustGrpcEventStreamQueueDepth(-1);
                yield return mxEvent;
            }

            await producerTask.ConfigureAwait(false);
        }
        finally
        {
            await streamCts.CancelAsync().ConfigureAwait(false);
            subscriber.Dispose();

            try
            {
                await producerTask.ConfigureAwait(false);
            }
            catch (OperationCanceledException) when (streamCts.IsCancellationRequested)
            {
            }
            catch (Exception exception)
            {
                logger.LogDebug(
                    exception,
                    "Event stream producer stopped for session {SessionId}.",
                    request.SessionId);
            }

            int remainingDepth = Interlocked.Exchange(ref streamQueueDepth, 0);
            if (remainingDepth > 0)
            {
                metrics.AdjustGrpcEventStreamQueueDepth(-remainingDepth);
            }

            metrics.StreamDisconnected("Detached");
        }
    }

    private async Task ProduceEventsAsync(
        GatewaySession session,
        ulong afterWorkerSequence,
        ChannelWriter<MxEvent> writer,
        Action eventQueued,
        CancellationToken cancellationToken)
    {
        try
        {
            await foreach (WorkerEvent workerEvent in session
                .ReadEventsAsync(cancellationToken)
                .WithCancellation(cancellationToken)
                .ConfigureAwait(false))
            {
                MxEvent publicEvent = mapper.MapEvent(workerEvent);
                if (publicEvent.WorkerSequence <= afterWorkerSequence)
                {
                    continue;
                }

                // Mirror the event to the dashboard EventsHub group for this
                // session. Fire-and-forget — broadcast errors must not affect
                // the source gRPC stream. Server-041: the
                // IDashboardEventBroadcaster contract documents Publish as
                // never-throw, but we enforce that at the seam too, so a
                // future implementation that adds synchronous validation or
                // a serializer hop cannot fault the producer loop and end
                // this client's gRPC stream.
                try
                {
                    dashboardEventBroadcaster.Publish(session.SessionId, publicEvent);
                }
                catch (Exception ex)
                {
                    logger.LogDebug(
                        ex,
                        "Dashboard event mirror threw for session {SessionId}; continuing.",
                        session.SessionId);
                }

                if (!writer.TryWrite(publicEvent))
                {
                    string message = $"Session {session.SessionId} event stream queue overflowed.";
                    metrics.QueueOverflow("grpc-event-stream");
                    if (options.Value.Events.BackpressurePolicy == EventBackpressurePolicy.FailFast)
                    {
                        session.MarkFaulted(message);
                        metrics.Fault(SessionManagerErrorCode.EventQueueOverflow.ToString());
                    }
                    else
                    {
                        logger.LogDebug(
                            "Disconnecting event stream for session {SessionId} after queue overflow.",
                            session.SessionId);
                    }

                    writer.TryComplete(new SessionManagerException(
                        SessionManagerErrorCode.EventQueueOverflow,
                        message));
                    return;
                }

                eventQueued();
            }

            writer.TryComplete();
        }
        catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
        {
            writer.TryComplete();
        }
        catch (Exception exception)
        {
            if (exception is WorkerClientException)
            {
                session.MarkFaulted(exception.Message);
                metrics.Fault(WorkerClientErrorCode.WorkerFaulted.ToString());
            }

            writer.TryComplete(exception);
        }
    }
}