Merge remote-tracking branch 'origin/main' into agent-1/issue-17-implement-dashboard-authentication

src/MxGateway.Server/GatewayApplication.cs

@@ -37,6 +37,7 @@ public static class GatewayApplication
         builder.Services.AddSingleton<GatewayMetrics>();
         builder.Services.AddSingleton<MxAccessGrpcMapper>();
         builder.Services.AddSingleton<MxAccessGrpcRequestValidator>();
+        builder.Services.AddSingleton<IEventStreamService, EventStreamService>();
         builder.Services.AddWorkerProcessLauncher();
         builder.Services.AddGatewaySessions();
         builder.Services.AddGatewayDashboard();

src/MxGateway.Server/Grpc/EventStreamService.cs

@@ -0,0 +1,140 @@
+using System.Runtime.CompilerServices;
+using System.Threading.Channels;
+using Microsoft.Extensions.Options;
+using MxGateway.Contracts.Proto;
+using MxGateway.Server.Configuration;
+using MxGateway.Server.Metrics;
+using MxGateway.Server.Sessions;
+using MxGateway.Server.Workers;
+
+namespace MxGateway.Server.Grpc;
+
+public sealed class EventStreamService(
+    ISessionManager sessionManager,
+    IOptions<GatewayOptions> options,
+    MxAccessGrpcMapper mapper,
+    GatewayMetrics metrics,
+    ILogger<EventStreamService> logger) : IEventStreamService
+{
+    public async IAsyncEnumerable<MxEvent> StreamEventsAsync(
+        StreamEventsRequest request,
+        [EnumeratorCancellation] CancellationToken cancellationToken)
+    {
+        if (!sessionManager.TryGetSession(request.SessionId, out GatewaySession session))
+        {
+            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,
+            () =>
+            {
+                int depth = Interlocked.Increment(ref streamQueueDepth);
+                metrics.SetEventQueueDepth(depth);
+            },
+            streamCts.Token);
+
+        try
+        {
+            await foreach (MxEvent mxEvent in eventQueue.Reader.ReadAllAsync(cancellationToken).ConfigureAwait(false))
+            {
+                int depth = Math.Max(0, Interlocked.Decrement(ref streamQueueDepth));
+                metrics.SetEventQueueDepth(depth);
+                yield return mxEvent;
+            }
+
+            await producerTask.ConfigureAwait(false);
+        }
+        finally
+        {
+            await streamCts.CancelAsync().ConfigureAwait(false);
+            subscriber.Dispose();
+            metrics.StreamDisconnected("Detached");
+
+            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);
+            }
+        }
+    }
+
+    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;
+                }
+
+                if (!writer.TryWrite(publicEvent))
+                {
+                    string message = $"Session {session.SessionId} event stream queue overflowed.";
+                    session.MarkFaulted(message);
+                    metrics.QueueOverflow("grpc-event-stream");
+                    metrics.Fault(SessionManagerErrorCode.EventQueueOverflow.ToString());
+                    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);
+        }
+    }
+}

src/MxGateway.Server/Grpc/IEventStreamService.cs

@@ -0,0 +1,10 @@
+using MxGateway.Contracts.Proto;
+
+namespace MxGateway.Server.Grpc;
+
+public interface IEventStreamService
+{
+    IAsyncEnumerable<MxEvent> StreamEventsAsync(
+        StreamEventsRequest request,
+        CancellationToken cancellationToken);
+}

src/MxGateway.Server/Grpc/MxAccessGatewayService.cs

@@ -12,6 +12,7 @@ public sealed class MxAccessGatewayService(
     IGatewayRequestIdentityAccessor identityAccessor,
     MxAccessGrpcRequestValidator requestValidator,
     MxAccessGrpcMapper mapper,
+    IEventStreamService eventStreamService,
     ILogger<MxAccessGatewayService> logger) : MxAccessGateway.MxAccessGatewayBase
 {
     public override async Task<OpenSessionReply> OpenSession(
@@ -102,17 +103,11 @@ public sealed class MxAccessGatewayService(
         try
         {
             requestValidator.ValidateStreamEvents(request);
-            await foreach (WorkerEvent workerEvent in sessionManager
-                .ReadEventsAsync(request.SessionId, context.CancellationToken)
+            await foreach (MxEvent publicEvent in eventStreamService
+                .StreamEventsAsync(request, context.CancellationToken)
                 .WithCancellation(context.CancellationToken)
                 .ConfigureAwait(false))
             {
-                MxEvent publicEvent = mapper.MapEvent(workerEvent);
-                if (publicEvent.WorkerSequence <= request.AfterWorkerSequence)
-                {
-                    continue;
-                }
-
                 await responseStream.WriteAsync(publicEvent).ConfigureAwait(false);
             }
         }
@@ -154,6 +149,8 @@ public sealed class MxAccessGatewayService(
         {
             SessionManagerErrorCode.SessionNotFound => StatusCode.NotFound,
             SessionManagerErrorCode.SessionNotReady => StatusCode.FailedPrecondition,
+            SessionManagerErrorCode.EventSubscriberAlreadyActive => StatusCode.ResourceExhausted,
+            SessionManagerErrorCode.EventQueueOverflow => StatusCode.ResourceExhausted,
             SessionManagerErrorCode.SessionLimitExceeded => StatusCode.ResourceExhausted,
             SessionManagerErrorCode.OpenFailed => StatusCode.Unavailable,
             SessionManagerErrorCode.CloseFailed => StatusCode.Unavailable,

src/MxGateway.Server/Sessions/GatewaySession.cs

@@ -13,6 +13,7 @@ public sealed class GatewaySession
     private DateTimeOffset _lastClientActivityAt;
     private DateTimeOffset? _leaseExpiresAt;
     private bool _closeStarted;
+    private int _activeEventSubscriberCount;
 
     public GatewaySession(
         string sessionId,
@@ -131,6 +132,17 @@ public sealed class GatewaySession
         }
     }
 
+    public int ActiveEventSubscriberCount
+    {
+        get
+        {
+            lock (_syncRoot)
+            {
+                return _activeEventSubscriberCount;
+            }
+        }
+    }
+
     public void AttachWorkerClient(IWorkerClient workerClient)
     {
         ArgumentNullException.ThrowIfNull(workerClient);
@@ -202,6 +214,29 @@ public sealed class GatewaySession
         }
     }
 
+    public IDisposable AttachEventSubscriber(bool allowMultipleSubscribers)
+    {
+        lock (_syncRoot)
+        {
+            if (_state != SessionState.Ready || _workerClient?.State != WorkerClientState.Ready)
+            {
+                throw new SessionManagerException(
+                    SessionManagerErrorCode.SessionNotReady,
+                    $"Session {SessionId} is not ready for event streaming. Current state is {_state}.");
+            }
+
+            if (!allowMultipleSubscribers && _activeEventSubscriberCount > 0)
+            {
+                throw new SessionManagerException(
+                    SessionManagerErrorCode.EventSubscriberAlreadyActive,
+                    $"Session {SessionId} already has an active event stream subscriber.");
+            }
+
+            _activeEventSubscriberCount++;
+            return new EventSubscriberLease(this);
+        }
+    }
+
     public async Task<WorkerCommandReply> InvokeAsync(
         WorkerCommand command,
         CancellationToken cancellationToken)
@@ -287,4 +322,31 @@ public sealed class GatewaySession
             return _workerClient;
         }
     }
+
+    private void DetachEventSubscriber()
+    {
+        lock (_syncRoot)
+        {
+            if (_activeEventSubscriberCount > 0)
+            {
+                _activeEventSubscriberCount--;
+            }
+        }
+    }
+
+    private sealed class EventSubscriberLease(GatewaySession session) : IDisposable
+    {
+        private bool _disposed;
+
+        public void Dispose()
+        {
+            if (_disposed)
+            {
+                return;
+            }
+
+            session.DetachEventSubscriber();
+            _disposed = true;
+        }
+    }
 }

src/MxGateway.Server/Sessions/SessionManagerErrorCode.cs

@@ -4,6 +4,8 @@ public enum SessionManagerErrorCode
 {
     SessionNotFound,
     SessionNotReady,
+    EventSubscriberAlreadyActive,
+    EventQueueOverflow,
     SessionLimitExceeded,
     OpenFailed,
     CloseFailed,

src/MxGateway.Server/Workers/WorkerClient.cs

@@ -29,6 +29,7 @@ public sealed class WorkerClient : IWorkerClient
     private WorkerClientState _state;
     private DateTimeOffset _lastHeartbeatAt;
     private int? _processId;
+    private int _eventQueueDepth;
     private Task? _readLoopTask;
     private Task? _writeLoopTask;
     private Task? _heartbeatLoopTask;
@@ -197,6 +198,8 @@ public sealed class WorkerClient : IWorkerClient
     {
         await foreach (WorkerEvent workerEvent in _events.Reader.ReadAllAsync(cancellationToken).ConfigureAwait(false))
         {
+            int queueDepth = Math.Max(0, Interlocked.Decrement(ref _eventQueueDepth));
+            _metrics?.SetEventQueueDepth(queueDepth);
             yield return workerEvent;
         }
     }
@@ -394,11 +397,6 @@ public sealed class WorkerClient : IWorkerClient
             _metrics?.EventReceived(SessionId, workerEvent.Event.Family.ToString());
         }
 
-        if (!await _events.Writer.WaitToWriteAsync(cancellationToken).ConfigureAwait(false))
-        {
-            return;
-        }
-
         if (!_events.Writer.TryWrite(workerEvent))
         {
             _metrics?.QueueOverflow("worker-events");
@@ -406,7 +404,11 @@ public sealed class WorkerClient : IWorkerClient
                 WorkerClientErrorCode.ProtocolViolation,
                 "Worker event channel rejected an event.",
                 null);
+            return;
         }
+
+        int queueDepth = Interlocked.Increment(ref _eventQueueDepth);
+        _metrics?.SetEventQueueDepth(queueDepth);
     }
 
     private void CompleteCommand(WorkerEnvelope envelope)