Ignore writeasync failures when sending to multiple connections (#1589)

benchmarks/Microsoft.AspNetCore.SignalR.Microbenchmarks/BroadcastBenchmark.cs

@@ -7,6 +7,7 @@ using BenchmarkDotNet.Attributes;
 using Microsoft.AspNetCore.SignalR.Internal;
 using Microsoft.AspNetCore.SignalR.Internal.Protocol;
 using Microsoft.AspNetCore.Sockets;
+using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Logging.Abstractions;
 
 namespace Microsoft.AspNetCore.SignalR.Microbenchmarks
@@ -25,7 +26,7 @@ namespace Microsoft.AspNetCore.SignalR.Microbenchmarks
         [GlobalSetup]
         public void GlobalSetup()
         {
-            _hubLifetimeManager = new DefaultHubLifetimeManager<Hub>();
+            _hubLifetimeManager = new DefaultHubLifetimeManager<Hub>(NullLogger<DefaultHubLifetimeManager<Hub>>.Instance);
 
 
             IHubProtocol protocol;

benchmarks/Microsoft.AspNetCore.SignalR.Microbenchmarks/DefaultHubDispatcherBenchmark.cs

@@ -36,7 +36,7 @@ namespace Microsoft.AspNetCore.SignalR.Microbenchmarks
 
             _dispatcher = new DefaultHubDispatcher<TestHub>(
                 serviceScopeFactory,
-                new HubContext<TestHub>(new DefaultHubLifetimeManager<TestHub>()),
+                new HubContext<TestHub>(new DefaultHubLifetimeManager<TestHub>(NullLogger<DefaultHubLifetimeManager<TestHub>>.Instance)),
                 new Logger<DefaultHubDispatcher<TestHub>>(NullLoggerFactory.Instance));
 
             var pair = DuplexPipe.CreateConnectionPair(PipeOptions.Default, PipeOptions.Default);

src/Microsoft.AspNetCore.SignalR.Core/DefaultHubLifetimeManager.cs

@@ -6,6 +6,7 @@ using System.Collections.Generic;
 using System.Linq;
 using System.Threading.Tasks;
 using Microsoft.AspNetCore.SignalR.Internal.Protocol;
+using Microsoft.Extensions.Logging;
 
 namespace Microsoft.AspNetCore.SignalR
 {
@@ -13,6 +14,12 @@ namespace Microsoft.AspNetCore.SignalR
     {
         private readonly HubConnectionList _connections = new HubConnectionList();
         private readonly HubGroupList _groups = new HubGroupList();
+        private readonly ILogger _logger;
+
+        public DefaultHubLifetimeManager(ILogger<DefaultHubLifetimeManager<THub>> logger)
+        {
+            _logger = logger;
+        }
 
         public override Task AddGroupAsync(string connectionId, string groupName)
         {
@@ -83,7 +90,7 @@ namespace Microsoft.AspNetCore.SignalR
                     continue;
                 }
 
-                tasks.Add(connection.WriteAsync(message));
+                tasks.Add(SafeWriteAsync(connection, message));
             }
 
             return Task.WhenAll(tasks);
@@ -105,7 +112,7 @@ namespace Microsoft.AspNetCore.SignalR
 
             var message = CreateInvocationMessage(methodName, args);
 
-            return connection.WriteAsync(message);
+            return SafeWriteAsync(connection, message);
         }
 
         public override Task SendGroupAsync(string groupName, string methodName, object[] args)
@@ -119,7 +126,7 @@ namespace Microsoft.AspNetCore.SignalR
             if (group != null)
             {
                 var message = CreateInvocationMessage(methodName, args);
-                var tasks = group.Values.Select(c => c.WriteAsync(message));
+                var tasks = group.Values.Select(c => SafeWriteAsync(c, message));
                 return Task.WhenAll(tasks);
             }
 
@@ -142,7 +149,7 @@ namespace Microsoft.AspNetCore.SignalR
                 var group = _groups[groupName];
                 if (group != null)
                 {
-                    tasks.Add(Task.WhenAll(group.Values.Select(c => c.WriteAsync(message))));
+                    tasks.Add(Task.WhenAll(group.Values.Select(c =>  SafeWriteAsync(c, message))));
                 }
             }
 
@@ -161,7 +168,7 @@ namespace Microsoft.AspNetCore.SignalR
             {
                 var message = CreateInvocationMessage(methodName, args);
                 var tasks = group.Values.Where(connection => !excludedIds.Contains(connection.ConnectionId))
-                    .Select(c => c.WriteAsync(message));
+                    .Select(c => SafeWriteAsync(c, message));
                 return Task.WhenAll(tasks);
             }
 
@@ -215,5 +222,30 @@ namespace Microsoft.AspNetCore.SignalR
                 return userIds.Contains(connection.UserIdentifier);
             });
         }
+
+        // This method is to protect against connections throwing synchronously when writing to them and preventing other connections from being written to
+        private async Task SafeWriteAsync(HubConnectionContext connection, InvocationMessage message)
+        {
+            try
+            {
+                await connection.WriteAsync(message);
+            }
+            // This exception isn't interesting to users
+            catch (Exception ex)
+            {
+                Log.FailedWritingMessage(_logger, ex);
+            }
+        }
+
+        private static class Log
+        {
+            private static readonly Action<ILogger, Exception> _failedWritingMessage =
+                LoggerMessage.Define(LogLevel.Warning, new EventId(1, "FailedWritingMessage"), "Failed writing message.");
+
+            public static void FailedWritingMessage(ILogger logger, Exception exception)
+            {
+                _failedWritingMessage(logger, exception);
+            }
+        }
     }
 }

src/Microsoft.AspNetCore.SignalR.Redis/Internal/RedisLoggerExtensions.cs

@@ -12,37 +12,37 @@ namespace Microsoft.AspNetCore.SignalR.Redis.Internal
     {
         // Category: RedisHubLifetimeManager<THub>
         private static readonly Action<ILogger, string, Exception> _connectingToEndpoints =
-            LoggerMessage.Define<string>(LogLevel.Information, new EventId(1, nameof(ConnectingToEndpoints)), "Connecting to Redis endpoints: {Endpoints}.");
+            LoggerMessage.Define<string>(LogLevel.Information, new EventId(1, "ConnectingToEndpoints"), "Connecting to Redis endpoints: {Endpoints}.");
 
         private static readonly Action<ILogger, Exception> _connected =
-            LoggerMessage.Define(LogLevel.Information, new EventId(2, nameof(Connected)), "Connected to Redis.");
+            LoggerMessage.Define(LogLevel.Information, new EventId(2, "Connected"), "Connected to Redis.");
 
         private static readonly Action<ILogger, string, Exception> _subscribing =
-            LoggerMessage.Define<string>(LogLevel.Trace, new EventId(3, nameof(Subscribing)), "Subscribing to channel: {Channel}.");
+            LoggerMessage.Define<string>(LogLevel.Trace, new EventId(3, "Subscribing"), "Subscribing to channel: {Channel}.");
 
         private static readonly Action<ILogger, string, Exception> _receivedFromChannel =
-            LoggerMessage.Define<string>(LogLevel.Trace, new EventId(4, nameof(ReceivedFromChannel)), "Received message from Redis channel {Channel}.");
+            LoggerMessage.Define<string>(LogLevel.Trace, new EventId(4, "ReceivedFromChannel"), "Received message from Redis channel {Channel}.");
 
         private static readonly Action<ILogger, string, Exception> _publishToChannel =
-            LoggerMessage.Define<string>(LogLevel.Trace, new EventId(5, nameof(PublishToChannel)), "Publishing message to Redis channel {Channel}.");
+            LoggerMessage.Define<string>(LogLevel.Trace, new EventId(5, "PublishToChannel"), "Publishing message to Redis channel {Channel}.");
 
         private static readonly Action<ILogger, string, Exception> _unsubscribe =
-            LoggerMessage.Define<string>(LogLevel.Trace, new EventId(6, nameof(Unsubscribe)), "Unsubscribing from channel: {Channel}.");
+            LoggerMessage.Define<string>(LogLevel.Trace, new EventId(6, "Unsubscribe"), "Unsubscribing from channel: {Channel}.");
 
         private static readonly Action<ILogger, Exception> _notConnected =
-            LoggerMessage.Define(LogLevel.Warning, new EventId(7, nameof(Connected)), "Not connected to Redis.");
+            LoggerMessage.Define(LogLevel.Warning, new EventId(7, "Connected"), "Not connected to Redis.");
 
         private static readonly Action<ILogger, Exception> _connectionRestored =
-            LoggerMessage.Define(LogLevel.Information, new EventId(8, nameof(ConnectionRestored)), "Connection to Redis restored.");
+            LoggerMessage.Define(LogLevel.Information, new EventId(8, "ConnectionRestored"), "Connection to Redis restored.");
 
         private static readonly Action<ILogger, Exception> _connectionFailed =
-            LoggerMessage.Define(LogLevel.Warning, new EventId(9, nameof(ConnectionFailed)), "Connection to Redis failed.");
+            LoggerMessage.Define(LogLevel.Warning, new EventId(9, "ConnectionFailed"), "Connection to Redis failed.");
 
         private static readonly Action<ILogger, Exception> _failedWritingMessage =
-            LoggerMessage.Define(LogLevel.Warning, new EventId(10, nameof(FailedWritingMessage)), "Failed writing message.");
+            LoggerMessage.Define(LogLevel.Warning, new EventId(10, "FailedWritingMessage"), "Failed writing message.");
 
         private static readonly Action<ILogger, Exception> _internalMessageFailed =
-            LoggerMessage.Define(LogLevel.Warning, new EventId(11, nameof(InternalMessageFailed)), "Error processing message for internal server message.");
+            LoggerMessage.Define(LogLevel.Warning, new EventId(11, "InternalMessageFailed"), "Error processing message for internal server message.");
 
         public static void ConnectingToEndpoints(this ILogger logger, EndPointCollection endpoints)
         {

src/Microsoft.AspNetCore.SignalR.Redis/RedisHubLifetimeManager.cs

@@ -173,7 +173,7 @@ namespace Microsoft.AspNetCore.SignalR.Redis
             var connection = _connections[connectionId];
             if (connection != null)
             {
-                return connection.WriteAsync(message.CreateInvocation());
+                return SafeWriteAsync(connection, message.CreateInvocation());
             }
 
             return PublishAsync(_channelNamePrefix + "." + connectionId, message);
@@ -402,14 +402,7 @@ namespace Microsoft.AspNetCore.SignalR.Redis
                     var invocation = message.CreateInvocation();
                     foreach (var connection in _connections)
                     {
-                        try
+                        tasks.Add(SafeWriteAsync(connection, invocation));
-                        {
-                            tasks.Add(connection.WriteAsync(invocation));
-                        }
-                        catch (Exception ex)
-                        {
-                            _logger.FailedWritingMessage(ex);
-                        }
                     }
 
                     await Task.WhenAll(tasks);
@@ -441,14 +434,7 @@ namespace Microsoft.AspNetCore.SignalR.Redis
                     {
                         if (!excludedIds.Contains(connection.ConnectionId))
                         {
-                            try
+                            tasks.Add(SafeWriteAsync(connection, invocation));
-                            {
-                                tasks.Add(connection.WriteAsync(invocation));
-                            }
-                            catch (Exception ex)
-                            {
-                                _logger.FailedWritingMessage(ex);
-                            }
                         }
                     }
 
@@ -524,16 +510,9 @@ namespace Microsoft.AspNetCore.SignalR.Redis
             _logger.Subscribing(connectionChannel);
             return _bus.SubscribeAsync(connectionChannel, async (c, data) =>
             {
-                try
+                var message = DeserializeMessage<RedisInvocationMessage>(data);
-                {
-                    var message = DeserializeMessage<RedisInvocationMessage>(data);
 
-                    await connection.WriteAsync(message.CreateInvocation());
+                await SafeWriteAsync(connection, message.CreateInvocation());
-                }
-                catch (Exception ex)
-                {
-                    _logger.FailedWritingMessage(ex);
-                }
             });
         }
 
@@ -545,16 +524,9 @@ namespace Microsoft.AspNetCore.SignalR.Redis
             // TODO: Look at optimizing (looping over connections checking for Name)
             return _bus.SubscribeAsync(userChannel, async (c, data) =>
             {
-                try
+                var message = DeserializeMessage<RedisInvocationMessage>(data);
-                {
-                    var message = DeserializeMessage<RedisInvocationMessage>(data);
 
-                    await connection.WriteAsync(message.CreateInvocation());
+                await SafeWriteAsync(connection, message.CreateInvocation());
-                }
-                catch (Exception ex)
-                {
-                    _logger.FailedWritingMessage(ex);
-                }
             });
         }
 
@@ -576,14 +548,7 @@ namespace Microsoft.AspNetCore.SignalR.Redis
                             continue;
                         }
 
-                        try
+                        tasks.Add(SafeWriteAsync(groupConnection, invocation));
-                        {
-                            tasks.Add(groupConnection.WriteAsync(invocation));
-                        }
-                        catch (Exception ex)
-                        {
-                            _logger.FailedWritingMessage(ex);
-                        }
                     }
 
                     await Task.WhenAll(tasks);
@@ -611,7 +576,7 @@ namespace Microsoft.AspNetCore.SignalR.Redis
                 // This also saves serializing and deserializing the message!
                 if (connection != null)
                 {
-                    publishTasks.Add(connection.WriteAsync(message.CreateInvocation()));
+                    publishTasks.Add(SafeWriteAsync(connection, message.CreateInvocation()));
                 }
                 else
                 {
@@ -662,6 +627,19 @@ namespace Microsoft.AspNetCore.SignalR.Redis
             return Task.CompletedTask;
         }
 
+        // This method is to protect against connections throwing synchronously when writing to them and preventing other connections from being written to
+        private async Task SafeWriteAsync(HubConnectionContext connection, InvocationMessage message)
+        {
+            try
+            {
+                await connection.WriteAsync(message);
+            }
+            catch (Exception ex)
+            {
+                _logger.FailedWritingMessage(ex);
+            }
+        }
+
         private class LoggerTextWriter : TextWriter
         {
             private readonly ILogger _logger;

test/Microsoft.AspNetCore.SignalR.Redis.Tests/RedisHubLifetimeManagerTests.cs

@@ -503,7 +503,7 @@ namespace Microsoft.AspNetCore.SignalR.Redis.Tests
         }
 
         [Fact]
-        public async Task WritingToLocalConnectionThatFailsThrowsException()
+        public async Task WritingToLocalConnectionThatFailsDoesNotThrowException()
         {
             var manager = new RedisHubLifetimeManager<MyHub>(new LoggerFactory().CreateLogger<RedisHubLifetimeManager<MyHub>>(), Options.Create(new RedisOptions()
             {
@@ -519,8 +519,7 @@ namespace Microsoft.AspNetCore.SignalR.Redis.Tests
 
                 await manager.OnConnectedAsync(connection).OrTimeout();
 
-                var exception = await Assert.ThrowsAsync<Exception>(() => manager.SendConnectionAsync(connection.ConnectionId, "Hello", new object[] { "World" }).OrTimeout());
+                await manager.SendConnectionAsync(connection.ConnectionId, "Hello", new object[] { "World" }).OrTimeout();
-                Assert.Equal("Message", exception.Message);
             }
         }
 

test/Microsoft.AspNetCore.SignalR.Tests/DefaultHubLifetimeManagerTests.cs

@@ -3,6 +3,8 @@ using System.Threading;
 using System.Threading.Channels;
 using System.Threading.Tasks;
 using Microsoft.AspNetCore.SignalR.Internal.Protocol;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Logging.Abstractions;
 using Moq;
 using Xunit;
 
@@ -11,12 +13,12 @@ namespace Microsoft.AspNetCore.SignalR.Tests
     public class DefaultHubLifetimeManagerTests
     {
         [Fact]
-        public async Task InvokeAllAsyncWritesToAllConnectionsOutput()
+        public async Task SendAllAsyncWritesToAllConnectionsOutput()
         {
             using (var client1 = new TestClient())
             using (var client2 = new TestClient())
             {
-                var manager = new DefaultHubLifetimeManager<MyHub>();
+                var manager = new DefaultHubLifetimeManager<MyHub>(new Logger<DefaultHubLifetimeManager<MyHub>>(NullLoggerFactory.Instance));
                 var connection1 = HubConnectionContextUtils.Create(client1.Connection);
                 var connection2 = HubConnectionContextUtils.Create(client2.Connection);
 
@@ -38,12 +40,12 @@ namespace Microsoft.AspNetCore.SignalR.Tests
         }
 
         [Fact]
-        public async Task InvokeAllAsyncDoesNotWriteToDisconnectedConnectionsOutput()
+        public async Task SendAllAsyncDoesNotWriteToDisconnectedConnectionsOutput()
         {
             using (var client1 = new TestClient())
             using (var client2 = new TestClient())
             {
-                var manager = new DefaultHubLifetimeManager<MyHub>();
+                var manager = new DefaultHubLifetimeManager<MyHub>(new Logger<DefaultHubLifetimeManager<MyHub>>(NullLoggerFactory.Instance));
                 var connection1 = HubConnectionContextUtils.Create(client1.Connection);
                 var connection2 = HubConnectionContextUtils.Create(client2.Connection);
 
@@ -64,12 +66,12 @@ namespace Microsoft.AspNetCore.SignalR.Tests
         }
 
         [Fact]
-        public async Task InvokeGroupAsyncWritesToAllConnectionsInGroupOutput()
+        public async Task SendGroupAsyncWritesToAllConnectionsInGroupOutput()
         {
             using (var client1 = new TestClient())
             using (var client2 = new TestClient())
             {
-                var manager = new DefaultHubLifetimeManager<MyHub>();
+                var manager = new DefaultHubLifetimeManager<MyHub>(new Logger<DefaultHubLifetimeManager<MyHub>>(NullLoggerFactory.Instance));
                 var connection1 = HubConnectionContextUtils.Create(client1.Connection);
                 var connection2 = HubConnectionContextUtils.Create(client2.Connection);
 
@@ -90,11 +92,11 @@ namespace Microsoft.AspNetCore.SignalR.Tests
         }
 
         [Fact]
-        public async Task InvokeConnectionAsyncWritesToConnectionOutput()
+        public async Task SendConnectionAsyncWritesToConnectionOutput()
         {
             using (var client = new TestClient())
             {
-                var manager = new DefaultHubLifetimeManager<MyHub>();
+                var manager = new DefaultHubLifetimeManager<MyHub>(new Logger<DefaultHubLifetimeManager<MyHub>>(NullLoggerFactory.Instance));
                 var connection = HubConnectionContextUtils.Create(client.Connection);
 
                 await manager.OnConnectedAsync(connection).OrTimeout();
@@ -109,42 +111,71 @@ namespace Microsoft.AspNetCore.SignalR.Tests
         }
 
         [Fact]
-        public async Task InvokeConnectionAsyncThrowsIfConnectionFailsToWrite()
+        public async Task SendConnectionAsyncDoesNotThrowIfConnectionFailsToWrite()
         {
             using (var client = new TestClient())
             {
-                // Force an exception when writing to connection
+                var manager = new DefaultHubLifetimeManager<MyHub>(new Logger<DefaultHubLifetimeManager<MyHub>>(NullLoggerFactory.Instance));
-                var manager = new DefaultHubLifetimeManager<MyHub>();
 
                 var connectionMock = HubConnectionContextUtils.CreateMock(client.Connection);
+                // Force an exception when writing to connection
                 connectionMock.Setup(m => m.WriteAsync(It.IsAny<HubMessage>())).Throws(new Exception("Message"));
                 var connection = connectionMock.Object;
 
                 await manager.OnConnectedAsync(connection).OrTimeout();
 
-                var exception = await Assert.ThrowsAsync<Exception>(() => manager.SendConnectionAsync(connection.ConnectionId, "Hello", new object[] { "World" }).OrTimeout());
+                await manager.SendConnectionAsync(connection.ConnectionId, "Hello", new object[] { "World" }).OrTimeout();
-                Assert.Equal("Message", exception.Message);
             }
         }
 
         [Fact]
-        public async Task InvokeConnectionAsyncOnNonExistentConnectionNoops()
+        public async Task SendAllAsyncSendsToAllConnectionsEvenWhenSomeFailToSend()
         {
-            var manager = new DefaultHubLifetimeManager<MyHub>();
+            using (var client = new TestClient())
+            using (var client2 = new TestClient())
+            {
+                var manager = new DefaultHubLifetimeManager<MyHub>(new Logger<DefaultHubLifetimeManager<MyHub>>(NullLoggerFactory.Instance));
+
+                var connectionMock = HubConnectionContextUtils.CreateMock(client.Connection);
+                var connectionMock2 = HubConnectionContextUtils.CreateMock(client2.Connection);
+
+                var tcs = new TaskCompletionSource<object>();
+                var tcs2 = new TaskCompletionSource<object>();
+                // Force an exception when writing to connection
+                connectionMock.Setup(m => m.WriteAsync(It.IsAny<HubMessage>())).Callback(() => tcs.TrySetResult(null)).Throws(new Exception("Message"));
+                connectionMock2.Setup(m => m.WriteAsync(It.IsAny<HubMessage>())).Callback(() => tcs2.TrySetResult(null)).Throws(new Exception("Message"));
+                var connection = connectionMock.Object;
+                var connection2 = connectionMock2.Object;
+
+                await manager.OnConnectedAsync(connection).OrTimeout();
+                await manager.OnConnectedAsync(connection2).OrTimeout();
+
+                await manager.SendAllAsync("Hello", new object[] { "World" }).OrTimeout();
+
+                // Check that all connections were "written" to
+                await tcs.Task.OrTimeout();
+                await tcs2.Task.OrTimeout();
+            }
+        }
+
+        [Fact]
+        public async Task SendConnectionAsyncOnNonExistentConnectionNoops()
+        {
+            var manager = new DefaultHubLifetimeManager<MyHub>(new Logger<DefaultHubLifetimeManager<MyHub>>(NullLoggerFactory.Instance));
             await manager.SendConnectionAsync("NotARealConnectionId", "Hello", new object[] { "World" }).OrTimeout();
         }
 
         [Fact]
         public async Task AddGroupOnNonExistentConnectionNoops()
         {
-            var manager = new DefaultHubLifetimeManager<MyHub>();
+            var manager = new DefaultHubLifetimeManager<MyHub>(new Logger<DefaultHubLifetimeManager<MyHub>>(NullLoggerFactory.Instance));
             await manager.AddGroupAsync("NotARealConnectionId", "MyGroup").OrTimeout();
         }
 
         [Fact]
         public async Task RemoveGroupOnNonExistentConnectionNoops()
         {
-            var manager = new DefaultHubLifetimeManager<MyHub>();
+            var manager = new DefaultHubLifetimeManager<MyHub>(new Logger<DefaultHubLifetimeManager<MyHub>>(NullLoggerFactory.Instance));
             await manager.RemoveGroupAsync("NotARealConnectionId", "MyGroup").OrTimeout();
         }