feat(sessions): route event streaming through SessionEventDistributor

src/ZB.MOM.WW.MxGateway.Server/Sessions/GatewaySession.cs

@@ -1,4 +1,7 @@
+using System.Runtime.CompilerServices;
 using ZB.MOM.WW.MxGateway.Contracts.Proto;
+using ZB.MOM.WW.MxGateway.Server.Configuration;
+using ZB.MOM.WW.MxGateway.Server.Grpc;
 using ZB.MOM.WW.MxGateway.Server.Workers;
 
 namespace ZB.MOM.WW.MxGateway.Server.Sessions;
@@ -7,6 +10,7 @@ public sealed class GatewaySession
 {
     private readonly object _syncRoot = new();
     private readonly SemaphoreSlim _closeLock = new(1, 1);
+    private readonly SessionEventStreaming _eventStreaming;
     private IWorkerClient? _workerClient;
     private SessionState _state = SessionState.Creating;
     private string? _finalFault;
@@ -14,6 +18,8 @@ public sealed class GatewaySession
     private DateTimeOffset? _leaseExpiresAt;
     private bool _closeStarted;
     private int _activeEventSubscriberCount;
+    private SessionEventDistributor? _eventDistributor;
+    private bool _eventDistributorStarted;
     private readonly Dictionary<(int ServerHandle, int ItemHandle), SessionItemRegistration> _items = [];
 
     /// <summary>
@@ -80,6 +86,15 @@ public sealed class GatewaySession
     /// <param name="shutdownTimeout">Timeout for worker process shutdown.</param>
     /// <param name="leaseDuration">Duration of the session lease.</param>
     /// <param name="openedAt">Timestamp when the session opened.</param>
+    /// <param name="eventStreaming">
+    ///     Dependencies the session uses to construct and own its
+    ///     <see cref="SessionEventDistributor"/> (the single per-session worker-event pump
+    ///     that fans raw mapped <see cref="MxEvent"/>s to every subscriber lease). When
+    ///     <see langword="null"/>, defaults are used (no replay logger, system clock, a
+    ///     fresh mapper, and default <see cref="EventOptions"/>) so unit tests that build a
+    ///     session directly still get a working distributor. Production passes the
+    ///     DI-resolved dependencies.
+    /// </param>
     public GatewaySession(
         string sessionId,
         string backendName,
@@ -93,7 +108,8 @@ public sealed class GatewaySession
         TimeSpan startupTimeout,
         TimeSpan shutdownTimeout,
         TimeSpan leaseDuration,
-        DateTimeOffset openedAt)
+        DateTimeOffset openedAt,
+        SessionEventStreaming? eventStreaming = null)
     {
         if (string.IsNullOrWhiteSpace(sessionId))
         {
@@ -130,6 +146,7 @@ public sealed class GatewaySession
         OpenedAt = openedAt;
         _lastClientActivityAt = openedAt;
         _leaseExpiresAt = openedAt + leaseDuration;
+        _eventStreaming = eventStreaming ?? SessionEventStreaming.Default;
     }
 
     /// <summary>
@@ -337,6 +354,72 @@ public sealed class GatewaySession
         TransitionTo(SessionState.Ready);
     }
 
+    // Constructs and starts the distributor exactly once, registering the subscriber under
+    // the same start so no event the pump fans can be missed between start and register.
+    // Started lazily on the FIRST AttachEventSubscriber rather than at MarkReady: today the
+    // worker event stream is only drained when a client begins streaming, so deferring the
+    // single drain to first-attach preserves that "events start flowing on subscribe"
+    // behavior and avoids draining a fast-completing source into the void before any
+    // subscriber exists. The source factory mirrors the mapping/ordering/start that
+    // EventStreamService.ProduceEventsAsync used before Task 4: it drains the worker event
+    // stream in source order and maps each WorkerEvent to the public MxEvent with the same
+    // mapper, with no skip/filter — per-RPC filtering (e.g. AfterWorkerSequence) stays at the
+    // subscriber boundary in EventStreamService. Returns a registered lease atomically with
+    // the start so the very first subscriber sees the stream from its beginning.
+    private IEventSubscriberLease StartDistributorAndRegister()
+    {
+        SessionEventDistributor distributor;
+        bool startNow = false;
+        lock (_syncRoot)
+        {
+            if (_eventDistributor is null)
+            {
+                EventOptions eventOptions = _eventStreaming.EventOptions;
+                _eventDistributor = new SessionEventDistributor(
+                    SessionId,
+                    MapWorkerEventsAsync,
+                    eventOptions.QueueCapacity,
+                    eventOptions.ReplayBufferCapacity,
+                    eventOptions.ReplayRetentionSeconds,
+                    _eventStreaming.DistributorLogger,
+                    _eventStreaming.TimeProvider);
+            }
+
+            distributor = _eventDistributor;
+            if (!_eventDistributorStarted)
+            {
+                _eventDistributorStarted = true;
+                startNow = true;
+            }
+        }
+
+        // Register BEFORE starting the pump so a subscriber is present when the pump begins
+        // draining — no event is fanned to an empty subscriber set and then missed by this
+        // first subscriber. StartAsync only schedules the pump task; it never blocks.
+        IEventSubscriberLease lease = distributor.Register();
+        if (startNow)
+        {
+            distributor.StartAsync(CancellationToken.None).GetAwaiter().GetResult();
+        }
+
+        return lease;
+    }
+
+    // The distributor's single event source. Drains the worker event stream once (the
+    // distributor guarantees a single consumer) and maps each frame to the public MxEvent,
+    // preserving worker order. Mirrors the former ProduceEventsAsync mapping exactly.
+    private async IAsyncEnumerable<MxEvent> MapWorkerEventsAsync(
+        [EnumeratorCancellation] CancellationToken cancellationToken)
+    {
+        MxAccessGrpcMapper mapper = _eventStreaming.Mapper;
+        await foreach (WorkerEvent workerEvent in ReadEventsAsync(cancellationToken)
+            .WithCancellation(cancellationToken)
+            .ConfigureAwait(false))
+        {
+            yield return mapper.MapEvent(workerEvent);
+        }
+    }
+
     /// <summary>
     /// Transitions the session to the Faulted state with a fault description.
     /// </summary>
@@ -395,10 +478,15 @@ public sealed class GatewaySession
     }
 
     /// <summary>
-    /// Attaches an event subscriber and returns a disposable lease.
+    ///     Attaches an event subscriber and returns a lease whose
+    ///     <see cref="IEventSubscriberLease.Reader"/> reads the fanned public
+    ///     <see cref="MxEvent"/>s for this subscriber. The single-subscriber guard
+    ///     (Tasks 7/8 relax it) is unchanged: with multi-subscriber disabled a second
+    ///     attach is rejected. The returned lease, when disposed, unregisters the
+    ///     distributor subscriber AND decrements the active-subscriber count.
     /// </summary>
     /// <param name="allowMultipleSubscribers">If true, allows multiple concurrent event subscribers.</param>
-    public IDisposable AttachEventSubscriber(bool allowMultipleSubscribers)
+    public IEventSubscriberLease AttachEventSubscriber(bool allowMultipleSubscribers)
     {
         lock (_syncRoot)
         {
@@ -417,7 +505,20 @@ public sealed class GatewaySession
             }
 
             _activeEventSubscriberCount++;
-            return new EventSubscriberLease(this);
+        }
+
+        // Construct/start the distributor and register this subscriber. Done outside the
+        // guard lock (StartDistributorAndRegister takes _syncRoot itself for construction).
+        // On any failure roll back the count we just took so the guard stays consistent.
+        try
+        {
+            IEventSubscriberLease distributorLease = StartDistributorAndRegister();
+            return new EventSubscriberLease(this, distributorLease);
+        }
+        catch
+        {
+            DetachEventSubscriber();
+            throw;
         }
     }
 
@@ -974,6 +1075,23 @@ public sealed class GatewaySession
         {
         }
 
+        // Stop the event pump and complete every subscriber channel before tearing down the
+        // worker client (the pump's source). DisposeAsync is the single session teardown
+        // point (SessionManager.RemoveSessionAsync awaits it after close), so awaiting it
+        // here guarantees the distributor's pump task is observed and subscribers are
+        // completed rather than left dangling.
+        SessionEventDistributor? distributor;
+        lock (_syncRoot)
+        {
+            distributor = _eventDistributor;
+            _eventDistributor = null;
+        }
+
+        if (distributor is not null)
+        {
+            await distributor.DisposeAsync().ConfigureAwait(false);
+        }
+
         if (_workerClient is not null)
         {
             await _workerClient.DisposeAsync().ConfigureAwait(false);
@@ -1115,12 +1233,19 @@ public sealed class GatewaySession
         }
     }
 
-    private sealed class EventSubscriberLease(GatewaySession session) : IDisposable
+    private sealed class EventSubscriberLease(GatewaySession session, IEventSubscriberLease distributorLease)
+        : IEventSubscriberLease
     {
         private bool _disposed;
 
+        /// <inheritdoc />
+        public System.Threading.Channels.ChannelReader<MxEvent> Reader => distributorLease.Reader;
+
         /// <summary>
-        /// Disposes the lease and detaches the event subscriber.
+        ///     Disposes the lease: unregisters this subscriber from the distributor (completing
+        ///     its channel) and decrements the session's active-subscriber count. Ordering is
+        ///     not significant — the count guard and the distributor registration are
+        ///     independent — but both must run exactly once.
         /// </summary>
         public void Dispose()
         {
@@ -1129,8 +1254,9 @@ public sealed class GatewaySession
                 return;
             }
 
-            session.DetachEventSubscriber();
             _disposed = true;
+            distributorLease.Dispose();
+            session.DetachEventSubscriber();
         }
     }
 }

src/ZB.MOM.WW.MxGateway.Server/Sessions/SessionEventStreaming.cs

@@ -0,0 +1,42 @@
+using Microsoft.Extensions.Logging.Abstractions;
+using ZB.MOM.WW.MxGateway.Server.Configuration;
+using ZB.MOM.WW.MxGateway.Server.Grpc;
+
+namespace ZB.MOM.WW.MxGateway.Server.Sessions;
+
+/// <summary>
+///     Dependencies a <see cref="GatewaySession"/> needs to construct and own its
+///     <see cref="SessionEventDistributor"/>. Bundled so the session constructor stays a
+///     single optional parameter rather than four, and so unit tests that build a session
+///     directly get a working distributor from <see cref="Default"/> without wiring DI.
+/// </summary>
+/// <param name="Mapper">
+///     Maps worker IPC <c>WorkerEvent</c> frames to public <c>MxEvent</c>s. The distributor
+///     pump applies this once per event in worker order, mirroring the mapping
+///     <c>EventStreamService.ProduceEventsAsync</c> used before Task 4.
+/// </param>
+/// <param name="EventOptions">
+///     Supplies the distributor's per-subscriber queue capacity and replay ring-buffer
+///     bounds (<see cref="EventOptions.QueueCapacity"/>,
+///     <see cref="EventOptions.ReplayBufferCapacity"/>,
+///     <see cref="EventOptions.ReplayRetentionSeconds"/>).
+/// </param>
+/// <param name="DistributorLogger">Logger for the distributor pump lifecycle.</param>
+/// <param name="TimeProvider">Clock used to timestamp and age-evict replay entries.</param>
+public sealed record SessionEventStreaming(
+    MxAccessGrpcMapper Mapper,
+    EventOptions EventOptions,
+    ILogger<SessionEventDistributor> DistributorLogger,
+    TimeProvider TimeProvider)
+{
+    /// <summary>
+    ///     Defaults used when a session is constructed without explicit streaming
+    ///     dependencies (unit tests). Uses a fresh mapper, default event options, a no-op
+    ///     logger, and the system clock.
+    /// </summary>
+    public static SessionEventStreaming Default { get; } = new(
+        new MxAccessGrpcMapper(),
+        new EventOptions(),
+        NullLogger<SessionEventDistributor>.Instance,
+        TimeProvider.System);
+}

src/ZB.MOM.WW.MxGateway.Server/Sessions/SessionManager.cs

@@ -25,6 +25,8 @@ public sealed class SessionManager : ISessionManager
     private readonly ILogger<SessionManager> _logger;
     private readonly GatewayOptions _options;
     private readonly SemaphoreSlim _sessionSlots;
+    private readonly Grpc.MxAccessGrpcMapper _eventMapper;
+    private readonly ILogger<SessionEventDistributor> _distributorLogger;
 
     /// <summary>
     /// Initializes a new instance of <see cref="SessionManager"/>.
@@ -35,13 +37,17 @@ public sealed class SessionManager : ISessionManager
     /// <param name="metrics">Gateway metrics.</param>
     /// <param name="timeProvider">Time provider for timestamps.</param>
     /// <param name="logger">Logger.</param>
+    /// <param name="eventMapper">Mapper used by each session's event distributor to map worker events to public events.</param>
+    /// <param name="distributorLogger">Logger passed to each session's event distributor pump.</param>
     public SessionManager(
         ISessionRegistry registry,
         ISessionWorkerClientFactory workerClientFactory,
         IOptions<GatewayOptions> options,
         GatewayMetrics metrics,
         TimeProvider? timeProvider = null,
-        ILogger<SessionManager>? logger = null)
+        ILogger<SessionManager>? logger = null,
+        Grpc.MxAccessGrpcMapper? eventMapper = null,
+        ILogger<SessionEventDistributor>? distributorLogger = null)
     {
         _registry = registry ?? throw new ArgumentNullException(nameof(registry));
         _workerClientFactory = workerClientFactory ?? throw new ArgumentNullException(nameof(workerClientFactory));
@@ -49,6 +55,8 @@ public sealed class SessionManager : ISessionManager
         _metrics = metrics ?? throw new ArgumentNullException(nameof(metrics));
         _timeProvider = timeProvider ?? TimeProvider.System;
         _logger = logger ?? NullLogger<SessionManager>.Instance;
+        _eventMapper = eventMapper ?? new Grpc.MxAccessGrpcMapper();
+        _distributorLogger = distributorLogger ?? NullLogger<SessionEventDistributor>.Instance;
         _options = options.Value;
         _sessionSlots = new SemaphoreSlim(_options.Sessions.MaxSessions, _options.Sessions.MaxSessions);
     }
@@ -438,6 +446,12 @@ public sealed class SessionManager : ISessionManager
         DateTimeOffset openedAt = _timeProvider.GetUtcNow();
         string clientCorrelationId = CreateClientCorrelationId(request.ClientSessionName, sessionId);
 
+        SessionEventStreaming eventStreaming = new(
+            _eventMapper,
+            _options.Events,
+            _distributorLogger,
+            _timeProvider);
+
         return new GatewaySession(
             sessionId,
             backendName,
@@ -451,7 +465,8 @@ public sealed class SessionManager : ISessionManager
             startupTimeout,
             shutdownTimeout,
             leaseDuration,
-            openedAt);
+            openedAt,
+            eventStreaming);
     }
 
     private static string CreateClientCorrelationId(