Merge pull request #17958 from Shadowghost/fix-websocket

Don't let a torn-down WebSocket take down the request handler

Emby.Server.Implementations/HttpServer/WebSocketConnection.cs

@@ -127,7 +127,7 @@ namespace Emby.Server.Implementations.HttpServer
                 {
                     receiveResult = await _socket.ReceiveAsync(memory, cancellationToken).ConfigureAwait(false);
                 }
-                catch (Exception ex) when (ex is WebSocketException or ObjectDisposedException or OperationCanceledException)
+                catch (Exception ex) when (IsConnectionGone(ex))
                 {
                     // ObjectDisposedException/OperationCanceledException: the socket was torn
                     // down underneath us (e.g. by the keep-alive watchdog after the connection
@@ -158,7 +158,15 @@ namespace Emby.Server.Implementations.HttpServer
 
                 if (receiveResult.EndOfMessage)
                 {
-                    await ProcessInternal(pipe.Reader).ConfigureAwait(false);
+                    try
+                    {
+                        await ProcessInternal(pipe.Reader).ConfigureAwait(false);
+                    }
+                    catch (Exception ex) when (IsConnectionGone(ex))
+                    {
+                        _logger.LogWarning("WS {IP} error sending data: {Message}", RemoteEndPoint, ex.Message);
+                        break;
+                    }
                 }
             }
             while ((_socket.State == WebSocketState.Open || _socket.State == WebSocketState.Connecting)
@@ -170,13 +178,24 @@ namespace Emby.Server.Implementations.HttpServer
                 || _socket.State == WebSocketState.CloseReceived
                 || _socket.State == WebSocketState.CloseSent)
             {
-                await _socket.CloseAsync(
-                    WebSocketCloseStatus.NormalClosure,
-                    string.Empty,
-                    cancellationToken).ConfigureAwait(false);
+                try
+                {
+                    await _socket.CloseAsync(
+                        WebSocketCloseStatus.NormalClosure,
+                        string.Empty,
+                        cancellationToken).ConfigureAwait(false);
+                }
+                catch (Exception ex) when (IsConnectionGone(ex))
+                {
+                    // The peer is already gone, there is nobody left to send the close frame to.
+                    _logger.LogDebug("WS {IP} error closing connection: {Message}", RemoteEndPoint, ex.Message);
+                }
             }
         }
 
+        private static bool IsConnectionGone(Exception ex)
+            => ex is WebSocketException or ObjectDisposedException or OperationCanceledException;
+
         private async Task ProcessInternal(PipeReader reader)
         {
             ReadResult result = await reader.ReadAsync().ConfigureAwait(false);

Emby.Server.Implementations/Session/SessionWebSocketListener.cs

@@ -171,7 +171,7 @@ namespace Emby.Server.Implementations.Session
             {
                 await SendForceKeepAlive(webSocket).ConfigureAwait(false);
             }
-            catch (WebSocketException exception)
+            catch (Exception exception) when (exception is WebSocketException or ObjectDisposedException or OperationCanceledException)
             {
                 _logger.LogWarning(exception, "Cannot send ForceKeepAlive message to WebSocket {0}.", webSocket);
             }
@@ -232,7 +232,7 @@ namespace Emby.Server.Implementations.Session
                 {
                     await SendForceKeepAlive(webSocket).ConfigureAwait(false);
                 }
-                catch (WebSocketException exception)
+                catch (Exception exception) when (exception is WebSocketException or ObjectDisposedException or OperationCanceledException)
                 {
                     _logger.LogInformation(exception, "Error sending ForceKeepAlive message to WebSocket.");
                     lost.Add(webSocket);

MediaBrowser.Controller/Net/WebSocketMessages/Outbound/ForceKeepAliveMessage.cs

@@ -4,7 +4,9 @@ using MediaBrowser.Model.Session;
 namespace MediaBrowser.Controller.Net.WebSocketMessages.Outbound;
 
 /// <summary>
-/// Force keep alive websocket messages.
+/// Force keep alive websocket messages. The data is the timeout in seconds after which the
+/// server considers the connection lost; clients are expected to answer with a KeepAlive
+/// message and to keep sending one at least every half of that timeout.
 /// </summary>
 public class ForceKeepAliveMessage : OutboundWebSocketMessage<int>
 {

tests/Jellyfin.Server.Implementations.Tests/HttpServer/WebSocketConnectionTests.cs

@@ -1,7 +1,11 @@
 using System;
 using System.Buffers;
 using System.IO;
+using System.Net.WebSockets;
+using System.Text;
 using System.Text.Json;
+using System.Threading;
+using System.Threading.Tasks;
 using Emby.Server.Implementations.HttpServer;
 using Microsoft.Extensions.Logging.Abstractions;
 using Xunit;
@@ -48,6 +52,92 @@ namespace Jellyfin.Server.Implementations.Tests.HttpServer
             Assert.Throws<JsonException>(() => con.DeserializeWebSocketMessage(new ReadOnlySequence<byte>(bytes), out var bytesConsumed));
         }
 
+        [Fact]
+        public async Task ReceiveAsync_SocketTornDownWhileAnswering_RaisesClosedWithoutThrowing()
+        {
+            // The keep-alive watchdog can dispose a connection while the receive loop is
+            // answering a message on it. The failing answer must not escape into the request
+            // handler, as that would skip the Closed event the session needs to release it.
+            var socket = new DisposedOnSendWebSocket(Encoding.UTF8.GetBytes("{\"MessageType\":\"KeepAlive\"}"));
+            var con = new WebSocketConnection(new NullLogger<WebSocketConnection>(), socket, null!, null!)
+            {
+                OnReceive = _ => Task.CompletedTask
+            };
+
+            var closed = false;
+            con.Closed += (_, _) => closed = true;
+
+            await con.ReceiveAsync(TestContext.Current.CancellationToken);
+
+            Assert.True(closed);
+            Assert.Equal(1, socket.SendAttempts);
+        }
+
+        /// <summary>
+        /// A socket that hands out a single message and then behaves like a socket that was
+        /// disposed underneath the receive loop.
+        /// </summary>
+        internal sealed class DisposedOnSendWebSocket : WebSocket
+        {
+            private readonly byte[] _message;
+            private bool _received;
+
+            public DisposedOnSendWebSocket(byte[] message)
+            {
+                _message = message;
+            }
+
+            public int SendAttempts { get; private set; }
+
+            public override WebSocketCloseStatus? CloseStatus => null;
+
+            public override string? CloseStatusDescription => null;
+
+            public override string? SubProtocol => null;
+
+            public override WebSocketState State => SendAttempts == 0 ? WebSocketState.Open : WebSocketState.Closed;
+
+            public override void Abort()
+            {
+            }
+
+            public override Task CloseAsync(WebSocketCloseStatus closeStatus, string? statusDescription, CancellationToken cancellationToken)
+                => Task.CompletedTask;
+
+            public override Task CloseOutputAsync(WebSocketCloseStatus closeStatus, string? statusDescription, CancellationToken cancellationToken)
+                => Task.CompletedTask;
+
+            public override void Dispose()
+            {
+            }
+
+            public override ValueTask<ValueWebSocketReceiveResult> ReceiveAsync(Memory<byte> buffer, CancellationToken cancellationToken)
+            {
+                ObjectDisposedException.ThrowIf(_received, this);
+
+                _received = true;
+                _message.CopyTo(buffer);
+                return ValueTask.FromResult(new ValueWebSocketReceiveResult(_message.Length, WebSocketMessageType.Text, true));
+            }
+
+            public override Task<WebSocketReceiveResult> ReceiveAsync(ArraySegment<byte> buffer, CancellationToken cancellationToken)
+                => throw new NotImplementedException();
+
+            public override ValueTask SendAsync(ReadOnlyMemory<byte> buffer, WebSocketMessageType messageType, bool endOfMessage, CancellationToken cancellationToken)
+                => throw FailSend();
+
+            public override Task SendAsync(ArraySegment<byte> buffer, WebSocketMessageType messageType, bool endOfMessage, CancellationToken cancellationToken)
+                => throw FailSend();
+
+            private WebSocketException FailSend()
+            {
+                SendAttempts++;
+                return new WebSocketException(
+                    WebSocketError.InvalidState,
+                    "The WebSocket is in an invalid state ('Closed') for this operation. Valid states are: 'Open, CloseReceived'");
+            }
+        }
+
         internal sealed class BufferSegment : ReadOnlySequenceSegment<byte>
         {
             public BufferSegment(Memory<byte> memory)