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

using System.Runtime.CompilerServices;
using Microsoft.Extensions.Options;
using ZB.MOM.WW.MxGateway.Contracts.Proto;
using ZB.MOM.WW.MxGateway.Server.Configuration;
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,
    GatewayMetrics metrics) : IEventStreamService
{
    /// <inheritdoc />
    public async IAsyncEnumerable<MxEvent> StreamEventsAsync(
        StreamEventsRequest request,
        string? callerKeyId,
        [EnumeratorCancellation] CancellationToken cancellationToken)
    {
        // This reads the subscriber's lease channel fed by the session's single
        // SessionEventDistributor pump. The pump owns the single drain of the worker event
        // stream and the worker->public mapping (mirroring the former ProduceEventsAsync); this
        // loop is the per-subscriber boundary that applies the per-RPC filter (AfterWorkerSequence),
        // queue-depth metrics, and the backpressure/overflow policy.
        //
        // The dashboard mirror runs OFF this per-RPC loop. The dashboard is a first-class internal
        // subscriber on the session's SessionEventDistributor (see GatewaySession.StartDashboardMirror),
        // so it receives session events even when no gRPC client is streaming. This loop does not
        // mirror to the dashboard. One deliberate consequence: the dashboard sees RAW session events,
        // not the per-gRPC-subscriber AfterWorkerSequence-filtered view this loop applies — the
        // dashboard is a separate LDAP-authenticated monitoring view that should see the session's
        // full event activity.
        //
        // Overflow handling: the distributor's per-subscriber channel is bounded and the pump writes
        // non-blocking. When this subscriber's channel is full the pump applies the per-subscriber
        // backpressure policy and completes this subscriber's channel with a SessionManagerException
        // (SessionManagerErrorCode.EventQueueOverflow). That terminal fault surfaces here when the
        // reader's MoveNextAsync throws, and it propagates to the gRPC client unchanged. The overflow
        // metric, and (in the legacy single-subscriber FailFast case) the session fault + fault metric,
        // are recorded by the distributor's overflow handler so the session, the pump, and other
        // subscribers are isolated from this subscriber's slowness.
        if (!sessionManager.TryGetSession(request.SessionId, out GatewaySession? session) || session is null)
        {
            throw new SessionManagerException(
                SessionManagerErrorCode.SessionNotFound,
                $"Session {request.SessionId} was not found.");
        }

        // Owner-scoped attach (security control): a session's event stream may be
        // attached or reattached ONLY by the API key that opened the session. The detach-grace
        // and fan-out retention windows are on by default, so without this check any event-scoped
        // key that learns a session id could attach to another key's retained session and receive
        // its replayed and live data. Ordinal comparison; null owner (session opened with no auth)
        // matches only a null caller key.
        if (!string.Equals(session.OwnerKeyId, callerKeyId, StringComparison.Ordinal))
        {
            throw new SessionManagerException(
                SessionManagerErrorCode.PermissionDenied,
                $"Session {request.SessionId} is owned by a different API key; event-stream attach is owner-scoped.");
        }

        // No `using` here — subscriber.Dispose() is called exactly once in the finally
        // block below, which also disposes the reader. A `using` declaration would add a
        // second Dispose on the same path and double-decrement the session subscriber count.
        // The subscriber mode (single vs. multi) is derived inside AttachEventSubscriber from
        // the session's own SessionEventStreaming.AllowMultipleEventSubscribers field — the
        // same source the distributor uses — so the two cannot diverge.
        //
        // Reconnect/resume: when AfterWorkerSequence > 0 the client is resuming, so
        // attach via the replay variant that atomically snapshots the replay ring AND registers
        // the live subscriber under one lock. That single critical section is the crux of the
        // no-gap/no-duplicate handoff: every replayed event has sequence <= LiveResumeSequence
        // and every live event delivered below is filtered to sequence > LiveResumeSequence, so
        // an event that was both replayed and (racing the registration) fanned into the live
        // channel is dropped exactly once, while no newer event is skipped. See
        // SessionEventDistributor.RegisterWithReplay for the full argument.
        //
        // AfterWorkerSequence == 0 (fresh stream, not a resume) keeps the original behavior:
        // a plain attach, no replay, no sentinel, and the live filter watermark stays 0.
        ulong afterWorkerSequence = request.AfterWorkerSequence;
        IEventSubscriberLease subscriber;
        IReadOnlyList<MxEvent> replayedEvents = [];
        bool replayGap = false;
        ulong oldestAvailableSequence = 0;

        if (afterWorkerSequence > 0)
        {
            EventSubscriberReplayAttachment attachment = session.AttachEventSubscriberWithReplay(
                options.Value.Sessions.MaxEventSubscribersPerSession,
                afterWorkerSequence);
            subscriber = attachment.Lease;
            replayedEvents = attachment.ReplayedEvents;
            replayGap = attachment.Gap;
            oldestAvailableSequence = attachment.OldestAvailableSequence;

            // The live filter resumes strictly after the last replayed sequence (or, when
            // nothing was replayed, after the requested watermark). This is what makes the
            // handoff free of duplicates: anything <= this watermark was already replayed.
            afterWorkerSequence = attachment.LiveResumeSequence;
        }
        else
        {
            subscriber = session.AttachEventSubscriber(
                options.Value.Sessions.MaxEventSubscribersPerSession);
        }

        IAsyncEnumerator<MxEvent> reader = subscriber.Reader
            .ReadAllAsync(cancellationToken)
            .GetAsyncEnumerator(cancellationToken);

        // GWC-15: register this subscriber's channel as a live backlog source instead of
        // reconciling the queue-depth gauge on every event. The gauge previously read the
        // bounded channel's Count (which takes the channel's internal lock) and adjusted the
        // metric under its own lock on every streamed event. Now the metric reads Count only
        // when it is scraped (ObservableGauge callback) or projected (GetSnapshot), summing the
        // live backlog across every registered subscriber — the same "buffered, not yet
        // delivered" aggregate the per-event push reported, but with no per-event lock traffic.
        // Disposing the registration in the finally removes this subscriber's contribution, so
        // the gauge returns to the other subscribers' backlog (zero when none remain) on
        // disconnect. CanCount guards a channel that ever cannot report Count (contributes 0).
        IDisposable backlogRegistration = metrics.RegisterEventStreamBacklogSource(
            () => subscriber.Reader.CanCount ? subscriber.Reader.Count : 0);

        try
        {
            // Emit order for a resume: the ReplayGap sentinel FIRST (only when events were
            // evicted), then the still-retained replay batch, then live. The sentinel is an
            // explicit documented control signal (not a synthesized MXAccess event) and is
            // delivered ONLY to this resuming subscriber — it is never fanned to other
            // subscribers and never appears in DrainEventsReply (that path is untouched).
            if (replayGap)
            {
                yield return CreateReplayGapSentinel(
                    request.SessionId,
                    request.AfterWorkerSequence,
                    oldestAvailableSequence);
            }

            foreach (MxEvent replayedEvent in replayedEvents)
            {
                // RegisterWithReplay already returns only events strictly newer than
                // AfterWorkerSequence, so no per-item sequence guard is needed here.
                // There is no per-event constraint filter on the event stream: events are
                // fanned as-is by the distributor pump. The only dedup watermark is the
                // LiveResumeSequence applied in the live loop below (to drop any event
                // that was both replayed and raced into the live channel).
                yield return replayedEvent;
            }

            while (true)
            {
                MxEvent mxEvent;
                try
                {
                    if (!await reader.MoveNextAsync().ConfigureAwait(false))
                    {
                        break;
                    }

                    mxEvent = reader.Current;
                }
                catch (WorkerClientException workerException)
                {
                    // The distributor pump completes every subscriber channel with the source
                    // fault when the worker event stream terminates abnormally; that surfaces
                    // here. Mirror the original ProduceEventsAsync behavior: fault the
                    // session and record the metric, then propagate the terminal fault to the
                    // gRPC client.
                    session.MarkFaulted(workerException.Message);
                    metrics.Fault(WorkerClientErrorCode.WorkerFaulted.ToString());
                    throw;
                }

                // Per-RPC filter stays at the subscriber boundary: each request may resume
                // from a different AfterWorkerSequence, so the shared pump fans raw events and
                // this loop drops the ones at or below the caller's watermark.
                if (mxEvent.WorkerSequence <= afterWorkerSequence)
                {
                    continue;
                }

                // The queue-depth gauge is maintained lazily via the backlog registration above
                // (GWC-15): the metric reads this subscriber's channel Count only when scraped,
                // so there is no per-event gauge bookkeeping on this hot path.
                yield return mxEvent;
            }
        }
        finally
        {
            await reader.DisposeAsync().ConfigureAwait(false);

            // Remove this subscriber's live backlog contribution before disposing the lease so
            // the gauge stops counting a channel that is about to be completed; after this the
            // gauge reflects only the remaining subscribers (zero when none remain).
            backlogRegistration.Dispose();
            subscriber.Dispose();

            metrics.StreamDisconnected("Detached");
        }
    }

    // Builds the single ReplayGap control sentinel emitted at the head of a resumed
    // StreamEvents stream when the requested AfterWorkerSequence predates the oldest event
    // still retained (events were evicted). Per the proto contract (MxEvent.replay_gap),
    // the sentinel carries the session id and the populated ReplayGap, with family
    // UNSPECIFIED, no body, and no per-item fields. It is a documented control signal — NOT a
    // synthesized MXAccess event — so emitting it does not violate the no-synthesis rule.
    private static MxEvent CreateReplayGapSentinel(
        string sessionId,
        ulong requestedAfterSequence,
        ulong oldestAvailableSequence)
        => new()
        {
            SessionId = sessionId,
            ReplayGap = new ReplayGap
            {
                RequestedAfterSequence = requestedAfterSequence,
                OldestAvailableSequence = oldestAvailableSequence,
            },
        };
}