feat: migrate push to SSE with WebSocket rollout compatibility

k8s/exceptionless/templates/api.yaml

@@ -28,6 +28,11 @@ spec:
       annotations:
         checksum/config: {{ include (print $.Template.BasePath "/config.yaml") . | sha256sum }}
     spec:
+      # SSE connections are long-lived; give the pod enough time to drain before SIGTERM.
+      # The preStop sleep lets the ALB/ingress controller deregister the pod before traffic stops,
+      # then the remaining window allows ASP.NET Core to cancel RequestAborted tokens and clean up.
+      # When push is eventually enabled behind a Gateway API RoutePolicy, revisit this value.
+      terminationGracePeriodSeconds: 60
       topologySpreadConstraints:
         - maxSkew: 1
           topologyKey: kubernetes.io/hostname
@@ -39,6 +44,13 @@ spec:
         - name: {{ template "exceptionless.name" . }}-api
           image: "{{ .Values.api.image.repository }}:{{ .Values.version }}"
           imagePullPolicy: {{ .Values.api.image.pullPolicy }}
+          lifecycle:
+            preStop:
+              # Give the ALB ~15s to deregister this pod before SIGTERM fires.
+              # The total graceful window is terminationGracePeriodSeconds (60s) minus
+              # this sleep, leaving 45s. ASP.NET Core uses 40s and keeps 5s of safety margin.
+              exec:
+                command: ["sleep", "15"]
           livenessProbe:
             httpGet:
               path: /health
@@ -82,7 +94,10 @@ spec:
 {{- include "exceptionless.otel-env" . | indent 12 }}
             - name: RunJobsInProcess
               value: 'false'
-            - name: EnableWebSockets
+            # SSE rollout prerequisite: Azure Application Gateway for Containers Ingress API
+            # does not support the routeTimeout=0s override required for long-lived SSE streams.
+            # Keep push disabled here until this route moves to Gateway API + RoutePolicy.
+            - name: EnablePush
               value: 'false'
 {{- if (empty .Values.storage.connectionString) }}
           volumeMounts:
@@ -163,6 +178,8 @@ metadata:
     alb.networking.azure.io/alb-namespace: {{ .Values.ingress.albNamespace }}
     alb.networking.azure.io/alb-frontend: {{ template "exceptionless.fullname" . }}-fe
     cert-manager.io/cluster-issuer: {{ .Values.ingress.clusterIssuer }}
+    # SSE is not safe to enable behind the current AGC Ingress API path.
+    # Migrate to Gateway API and attach a RoutePolicy with routeTimeout: 0s before enabling push.
 spec:
   ingressClassName: azure-alb-external
   tls:

src/Exceptionless.Core/Bootstrapper.cs

@@ -112,6 +112,7 @@ public static void RegisterServices(IServiceCollection services, AppOptions appO
         services.TryAddEnumerable(ServiceDescriptor.Singleton<IQueueBehavior<WorkItemData>, WorkItemDuplicateDetectionQueueBehavior>());
 
         services.AddSingleton<IConnectionMapping, ConnectionMapping>();
+        services.AddSingleton<IConnectionLeaseStore, ConnectionLeaseStore>();
         services.AddSingleton<MessageService>();
         services.AddStartupAction<MessageService>();
         services.AddSingleton<IMessageBus>(s => new InMemoryMessageBus(new InMemoryMessageBusOptions
@@ -242,8 +243,8 @@ public static void LogConfiguration(IServiceProvider serviceProvider, AppOptions
         if (String.IsNullOrEmpty(appOptions.StorageOptions.Provider))
             logger.LogWarning("Distributed storage is NOT enabled on {MachineName}", Environment.MachineName);
 
-        if (!appOptions.EnableWebSockets)
-            logger.LogWarning("Web Sockets is NOT enabled on {MachineName}", Environment.MachineName);
+        if (!appOptions.EnablePush)
+            logger.LogWarning("Real-time push (SSE) is NOT enabled on {MachineName}", Environment.MachineName);
 
         if (String.IsNullOrEmpty(appOptions.EmailOptions.SmtpHost))
             logger.LogWarning("Emails will NOT be sent until the SmtpHost is configured on {MachineName}", Environment.MachineName);

src/Exceptionless.Core/Configuration/AppOptions.cs

@@ -55,7 +55,18 @@ public class AppOptions
 
     public bool EnableRepositoryNotifications { get; internal set; }
 
-    public bool EnableWebSockets { get; internal set; }
+    /// <summary>
+    /// Controls whether real-time push (SSE) is enabled. Reads from either 'EnablePush'
+    /// or legacy 'EnableWebSockets' config key for backward compatibility.
+    /// </summary>
+    public bool EnablePush { get; internal set; }
+
+    [Obsolete("Use EnablePush instead. This alias remains for source compatibility during the push transport rollout.")]
+    public bool EnableWebSockets
+    {
+        get => EnablePush;
+        internal set => EnablePush = value;
+    }
 
     public string? Version { get; internal set; }
 
@@ -112,7 +123,8 @@ public static AppOptions ReadFromConfiguration(IConfiguration config)
         options.BulkBatchSize = config.GetValue(nameof(options.BulkBatchSize), 1000);
 
         options.EnableRepositoryNotifications = config.GetValue(nameof(options.EnableRepositoryNotifications), true);
-        options.EnableWebSockets = config.GetValue(nameof(options.EnableWebSockets), true);
+        // Support both new 'EnablePush' and legacy 'EnableWebSockets' config keys
+        options.EnablePush = config.GetValue(nameof(options.EnablePush), config.GetValue("EnableWebSockets", true));
 
         try
         {

src/Exceptionless.Core/Services/MessageService.cs

@@ -6,7 +6,6 @@
 using Foundatio.Repositories.Elasticsearch;
 using Foundatio.Repositories.Models;
 using Microsoft.Extensions.Logging;
-using Microsoft.Extensions.Logging.Abstractions;
 
 namespace Exceptionless.Core.Services;
 
@@ -19,9 +18,7 @@ public class MessageService : IDisposable, IStartupAction
     private readonly IEventRepository _eventRepository;
     private readonly ITokenRepository _tokenRepository;
     private readonly IWebHookRepository _webHookRepository;
-    private readonly IConnectionMapping _connectionMapping;
     private readonly AppOptions _options;
-    private readonly ILogger _logger;
     private readonly List<Action> _disposeActions = [];
 
     public MessageService(
@@ -43,9 +40,12 @@ public MessageService(
         _eventRepository = eventRepository;
         _tokenRepository = tokenRepository;
         _webHookRepository = webHookRepository;
-        _connectionMapping = connectionMapping;
         _options = options;
-        _logger = loggerFactory.CreateLogger<MessageService>() ?? NullLogger<MessageService>.Instance;
+
+        // Preserve the public constructor shape for existing composition roots while push
+        // publication no longer depends on distributed connection state or trace logging.
+        _ = connectionMapping;
+        _ = loggerFactory;
     }
 
     public Task RunAsync(CancellationToken shutdownToken = default)
@@ -74,22 +74,13 @@ public Task RunAsync(CancellationToken shutdownToken = default)
         _disposeActions.Add(() => repo.BeforePublishEntityChanged.RemoveHandler(handler));
     }
 
-    private async Task OnBeforePublishEntityChangedAsync<T>(object sender, BeforePublishEntityChangedEventArgs<T> args)
+    private Task OnBeforePublishEntityChangedAsync<T>(object sender, BeforePublishEntityChangedEventArgs<T> args)
         where T : class, IIdentity, new()
     {
-        int listenerCount = await GetNumberOfListeners(args.Message);
-        args.Cancel = listenerCount == 0;
-        if (args.Cancel)
-            _logger.LogTrace("Cancelled {EntityType} Entity Changed Message: {@Message}", typeof(T).Name, args.Message);
-    }
-
-    private Task<int> GetNumberOfListeners(EntityChanged message)
-    {
-        var entityChanged = ExtendedEntityChanged.Create(message, false);
-        if (String.IsNullOrEmpty(entityChanged.OrganizationId))
-            return Task.FromResult(1); // Return 1 as we have no idea if people are listening.
-
-        return _connectionMapping.GetGroupConnectionCountAsync(entityChanged.OrganizationId);
+        // Push routing is replica-local. Publishing must not depend on immortal distributed
+        // connection indexes because another replica may own the interested client.
+        args.Cancel = false;
+        return Task.CompletedTask;
     }
 
     public void Dispose()

src/Exceptionless.Core/Utility/AppDiagnostics.cs

@@ -125,6 +125,11 @@ public GaugeInfo(Meter meter, string name)
 
     internal static readonly Counter<int> SavedViewsSize = Meter.CreateCounter<int>("ex.savedviews.size", description: "Size of user saved views");
     internal static readonly Counter<int> SavedViewsViewTypeSize = Meter.CreateCounter<int>("ex.savedviews.viewtype.size", description: "Size of user saved views by view type");
+
+    internal static readonly Counter<int> PushSseConnectionsOpened = Meter.CreateCounter<int>("ex.push.connections.sse.opened", description: "SSE push connections opened");
+    internal static readonly Counter<int> PushSseConnectionsClosed = Meter.CreateCounter<int>("ex.push.connections.sse.closed", description: "SSE push connections closed");
+    internal static readonly Counter<int> PushWebSocketConnectionsOpened = Meter.CreateCounter<int>("ex.push.connections.websocket.opened", description: "WebSocket push connections opened");
+    internal static readonly Counter<int> PushWebSocketConnectionsClosed = Meter.CreateCounter<int>("ex.push.connections.websocket.closed", description: "WebSocket push connections closed");
 }
 
 public static class MetricsClientExtensions

src/Exceptionless.Core/Utility/IConnectionLeaseStore.cs

@@ -0,0 +1,77 @@
+namespace Exceptionless.Core.Utility;
+
+public interface IConnectionLeaseStore
+{
+    Task<bool> TryAcquireAsync(string userId, string connectionId, int maxConnections, TimeSpan leaseDuration);
+    Task<bool> RenewAsync(string userId, string connectionId, TimeSpan leaseDuration);
+    Task ReleaseAsync(string userId, string connectionId);
+}
+
+public sealed class ConnectionLeaseStore(TimeProvider timeProvider) : IConnectionLeaseStore
+{
+    private readonly Dictionary<string, Dictionary<string, DateTimeOffset>> _leases = [];
+    private readonly object _lock = new();
+
+    public Task<bool> TryAcquireAsync(string userId, string connectionId, int maxConnections, TimeSpan leaseDuration)
+    {
+        if (maxConnections <= 0)
+            return Task.FromResult(false);
+
+        lock (_lock)
+        {
+            var now = timeProvider.GetUtcNow();
+            if (!_leases.TryGetValue(userId, out var userLeases))
+            {
+                userLeases = [];
+                _leases[userId] = userLeases;
+            }
+
+            RemoveExpired(userLeases, now);
+            if (!userLeases.ContainsKey(connectionId) && userLeases.Count >= maxConnections)
+                return Task.FromResult(false);
+
+            userLeases[connectionId] = now + leaseDuration;
+            return Task.FromResult(true);
+        }
+    }
+
+    public Task<bool> RenewAsync(string userId, string connectionId, TimeSpan leaseDuration)
+    {
+        lock (_lock)
+        {
+            var now = timeProvider.GetUtcNow();
+            if (!_leases.TryGetValue(userId, out var userLeases))
+                return Task.FromResult(false);
+
+            RemoveExpired(userLeases, now);
+            if (!userLeases.ContainsKey(connectionId))
+                return Task.FromResult(false);
+
+            userLeases[connectionId] = now + leaseDuration;
+            return Task.FromResult(true);
+        }
+    }
+
+    public Task ReleaseAsync(string userId, string connectionId)
+    {
+        lock (_lock)
+        {
+            if (_leases.TryGetValue(userId, out var userLeases))
+            {
+                userLeases.Remove(connectionId);
+                if (userLeases.Count is 0)
+                    _leases.Remove(userId);
+            }
+        }
+
+        return Task.CompletedTask;
+    }
+
+    private static void RemoveExpired(Dictionary<string, DateTimeOffset> leases, DateTimeOffset now)
+    {
+        foreach (string connectionId in leases.Where(pair => pair.Value <= now).Select(pair => pair.Key).ToArray())
+            leases.Remove(connectionId);
+    }
+}
+
+public sealed class ConnectionLeaseStoreException(string message, Exception innerException) : Exception(message, innerException);

src/Exceptionless.Insulation/Bootstrapper.cs

@@ -112,6 +112,7 @@ private static void RegisterCache(IServiceCollection container, CacheOptions opt
                 container.ReplaceSingleton<ICacheClient>(CreateRedisCacheClient);
 
             container.ReplaceSingleton<IConnectionMapping, RedisConnectionMapping>();
+            container.ReplaceSingleton<IConnectionLeaseStore, RedisConnectionLeaseStore>();
         }
     }
 

src/Exceptionless.Insulation/Redis/RedisConnectionLeaseStore.cs

@@ -0,0 +1,84 @@
+using Exceptionless.Core;
+using Exceptionless.Core.Utility;
+using StackExchange.Redis;
+
+namespace Exceptionless.Insulation.Redis;
+
+public sealed class RedisConnectionLeaseStore : IConnectionLeaseStore
+{
+    private const string AcquireScript = """
+        local now = redis.call('TIME')
+        local now_ms = (now[1] * 1000) + math.floor(now[2] / 1000)
+        redis.call('ZREMRANGEBYSCORE', KEYS[1], '-inf', now_ms)
+        if redis.call('ZSCORE', KEYS[1], ARGV[1]) then
+          redis.call('ZADD', KEYS[1], now_ms + ARGV[3], ARGV[1])
+          redis.call('PEXPIRE', KEYS[1], ARGV[3] * 2)
+          return 1
+        end
+        if redis.call('ZCARD', KEYS[1]) >= tonumber(ARGV[2]) then return 0 end
+        redis.call('ZADD', KEYS[1], now_ms + ARGV[3], ARGV[1])
+        redis.call('PEXPIRE', KEYS[1], ARGV[3] * 2)
+        return 1
+        """;
+    private const string RenewScript = """
+        local now = redis.call('TIME')
+        local now_ms = (now[1] * 1000) + math.floor(now[2] / 1000)
+        local expires = redis.call('ZSCORE', KEYS[1], ARGV[1])
+        if not expires or tonumber(expires) <= now_ms then
+          redis.call('ZREM', KEYS[1], ARGV[1])
+          return 0
+        end
+        redis.call('ZADD', KEYS[1], now_ms + ARGV[2], ARGV[1])
+        redis.call('PEXPIRE', KEYS[1], ARGV[2] * 2)
+        return 1
+        """;
+    private readonly IConnectionMultiplexer _multiplexer;
+    private readonly string _keyPrefix;
+
+    public RedisConnectionLeaseStore(IConnectionMultiplexer multiplexer, AppOptions options)
+    {
+        _multiplexer = multiplexer;
+        _keyPrefix = $"PushLease:{options.AppScope}:";
+    }
+
+    public async Task<bool> TryAcquireAsync(string userId, string connectionId, int maxConnections, TimeSpan leaseDuration)
+    {
+        try
+        {
+            var result = await Database.ScriptEvaluateAsync(AcquireScript, [GetKey(userId)], [connectionId, maxConnections, (long)leaseDuration.TotalMilliseconds]);
+            return (long)result == 1;
+        }
+        catch (RedisException ex)
+        {
+            throw new ConnectionLeaseStoreException("Unable to acquire a push connection lease.", ex);
+        }
+    }
+
+    public async Task<bool> RenewAsync(string userId, string connectionId, TimeSpan leaseDuration)
+    {
+        try
+        {
+            var result = await Database.ScriptEvaluateAsync(RenewScript, [GetKey(userId)], [connectionId, (long)leaseDuration.TotalMilliseconds]);
+            return (long)result == 1;
+        }
+        catch (RedisException ex)
+        {
+            throw new ConnectionLeaseStoreException("Unable to renew a push connection lease.", ex);
+        }
+    }
+
+    public async Task ReleaseAsync(string userId, string connectionId)
+    {
+        try
+        {
+            await Database.SortedSetRemoveAsync(GetKey(userId), connectionId);
+        }
+        catch (RedisException ex)
+        {
+            throw new ConnectionLeaseStoreException("Unable to release a push connection lease.", ex);
+        }
+    }
+
+    private IDatabase Database => _multiplexer.GetDatabase();
+    private RedisKey GetKey(string userId) => _keyPrefix + userId;
+}

src/Exceptionless.Web/Bootstrapper.cs

@@ -13,7 +13,9 @@ public class Bootstrapper
 {
     public static void RegisterServices(IServiceCollection services, AppOptions appOptions, ILoggerFactory loggerFactory)
     {
+        services.AddSingleton<SseConnectionManager>();
         services.AddSingleton<WebSocketConnectionManager>();
+        services.AddSingleton<PushConnectionRegistry>();
         services.AddSingleton<MessageBusBroker>();
 
         services.AddSingleton<ApiMapper>();

src/Exceptionless.Web/ClientApp.angular/components/websocket/websocket-service.js

@@ -1,6 +1,7 @@
 (function () {
     "use strict";
 
+    // Deprecated: keep the legacy Angular client on WebSocket during the SSE rollout.
     angular
         .module("exceptionless.websocket", ["app.config", "exceptionless", "exceptionless.auth"])
         .factory("websocketService", function ($ExceptionlessClient, $rootScope, $timeout, authService, BASE_URL) {