WebSocketException (0x80004005): The WebSocket is in an invalid state ('Closed') for this operation. Valid states are: 'Open, CloseSent'

[sepehr1014]

Using Microsoft.AspNetCore 2.1.0-preview1-final we get this exception on ReceiveAsync. The usage is quite ordinary and we didn't get any errors on the earlier builds. The first thing we do after we accept the socket is:

result = await currentSocket.ReceiveAsync(new ArraySegment<byte>(buffer), CancellationToken.None)

And here's the exception:

System.Net.WebSockets.WebSocketException (0x80004005): The WebSocket is in an invalid state ('Closed') for this operation. Valid states are: 'Open, CloseSent'
   at System.Net.WebSockets.WebSocketValidate.ThrowIfInvalidState(WebSocketState currentState, Boolean isDisposed, WebSocketState[] validStates)
   at System.Net.WebSockets.ManagedWebSocket.ReceiveAsync(ArraySegment`1 buffer, CancellationToken cancellationToken)
--- End of stack trace from previous location where exception was thrown ---
   at System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw()
   at System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)

[davidfowl]

What does your websocket loop look like?

[sepehr1014]

Well, here's the loop. But this line is NOT in the loop. It's the first read ...

while (currentSocket.State == WebSocketState.Open)
            {
                var temp = fullResult;
                fullResult = new byte[fullResult.Length + result.Count];
                Buffer.BlockCopy(temp, 0, fullResult, 0, temp.Length);
                Buffer.BlockCopy(buffer, 0, fullResult, temp.Length, result.Count);

                if (result.EndOfMessage)
                {
                    long requestId = 0;
                    try
                    {
                        var json = Encoding.UTF8.GetString(fullResult, 0, fullResult.Length);
                        // Process ...
                    }
                    finally
                    {
                        fullResult = new byte[0];
                    }
                }

                try
                {
                    result = await currentSocket.ReceiveAsync(new ArraySegment<byte>(buffer), CancellationToken.None);
                }
                catch
                {
                    break;
                }
            }

[davidfowl]

I'd like to see the whole loop

[sepehr1014]

@davidfowl Actually that was our web socket loop. I could send you our messaging controller if I had your email address.

[davidfowl]

Where do you handle close messages?

[sepehr1014]

@davidfowl Here's the controller code:

    public class MessagingController : Controller
    {
        WebSocket currentSocket;
        string currentSession;

        public async Task Connect([FromQuery]string clientId)
        {
            if (Request.HttpContext.WebSockets.IsWebSocketRequest)
            {
                currentSocket = await Request.HttpContext.WebSockets.AcceptWebSocketAsync();

                var currentId = IdentityHelper.GetUserId(User);
                currentSession = $"{currentId}:{clientId}";

                List<string> sessionList;
                if (cacheHandler.MessagingSessions.TryGetValue(currentId, out sessionList))
                {
                    if (!sessionList.Contains(currentSession))
                        sessionList.Add(currentSession);
                }
                else
                    cacheHandler.MessagingSessions.TryAdd(currentId, new List<string> { currentSession });

                cacheHandler.MessagingRequests.TryAdd(currentSession, new ConcurrentQueue<string>());

                WebSocket lastSocket;
                cacheHandler.MessagingSockets.TryRemove(currentSession, out lastSocket);
                if (lastSocket != currentSocket)
                    lastSocket?.Dispose();
                cacheHandler.MessagingSockets.TryAdd(currentSession, currentSocket);

                var taskRequests = Task.Factory.StartNew(HandleRequests, TaskCreationOptions.LongRunning);

                try
                {
                    await HandleSocket();
                }
                catch (WebSocketException ex) when (ex.Message.Contains("The remote party closed the WebSocket connection without completing the close handshake")
                    || ex.Message.Contains("The WebSocket is in an invalid state ('Closed') for this operation")
                    //|| (ex.InnerException is IOException && ex.InnerException?.Message?.Contains("connection reset by peer") == true)
                    //|| (ex is WebSocketException && ex.Message.Contains("net_WebSockets_InvalidState; (Closed,Open, CloseSent)"))
                    )
                {
                }
                finally
                {
                    try
                    {
                        if (currentSocket.State != WebSocketState.Closed)
                            await currentSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, null, CancellationToken.None);
                    }
                    catch (Exception ex) when (ex is ObjectDisposedException || ex.Message.Contains("The remote party closed the WebSocket connection without completing the close handshake"))
                    { }
                    catch (Exception ex)
                    {
                        _logger.LogError(ex, null);
                    }

                    if (cacheHandler.MessagingSockets.ContainsKey(currentSession) && cacheHandler.MessagingSockets[currentSession] == currentSocket)
                    {
                        var tryResult = cacheHandler.MessagingSockets.TryRemove(currentSession, out currentSocket);
                        if (!tryResult)
                            _logger.LogError(new Exception("Could not remove current socket from dictionary!"), null);
                    }

                    currentSocket?.Dispose();
                }
            }
        }
    
        private async Task HandleRequests()
        {
            string payload;
            var retries = 10;
            while (cacheHandler.MessagingSockets.ContainsKey(currentSession) && cacheHandler.MessagingSockets[currentSession] == currentSocket)
            {
                if (currentSocket.State == WebSocketState.Open && cacheHandler.MessagingRequests[currentSession].TryPeek(out payload))
                {
                    try
                    {
                        await SendAsync(currentSocket, payload);
                    }
                    catch (Exception ex)
                    {
                        if (retries-- > 0)
                        {
                            await Task.Delay(100);
                            continue;
                        }
                        else
                            _logger.LogError(LoggingEvents.MessagingError, ex, null);
                    }

                    var tryResult = cacheHandler.MessagingRequests[currentSession].TryDequeue(out payload);
                    if (!tryResult)
                        _logger.LogError(new Exception("Could not dequeue payload from dictionary!"), null);

                    retries = 10;
                }
                else
                    await Task.Delay(25);
            }
        }

        private async Task HandleSocket()
        {
            var buffer = new byte[1024 * 16];
            var fullResult = new byte[0];
            WebSocketReceiveResult result = null;

            try
            {
                result = await currentSocket.ReceiveAsync(new ArraySegment<byte>(buffer), CancellationToken.None);
            }
            catch (ObjectDisposedException)
            {
                return;
            }

            while (currentSocket.State == WebSocketState.Open)
            {
                var temp = fullResult;
                fullResult = new byte[fullResult.Length + result.Count];
                Buffer.BlockCopy(temp, 0, fullResult, 0, temp.Length);
                Buffer.BlockCopy(buffer, 0, fullResult, temp.Length, result.Count);

                if (result.EndOfMessage)
                {
                    long requestId = 0;
                    try
                    {
                        var json = Encoding.UTF8.GetString(fullResult, 0, fullResult.Length);
                        _logger.LogWarning("Received from " + currentSocket.GetHashCode() + ": " + json);

                        dynamic req = JsonConvert.DeserializeObject(json);
                        string action = req.a.ToString();
                        requestId = Int64.Parse(req.i?.ToString() ?? 0);
                        Task<object> task;

                        var actionClassType = this.GetType().Assembly.GetType($"Pasaj.Hubs.{action.Split('.')[0]}Hub");
                        using (var hubInstance = Activator.CreateInstance(actionClassType, args: new dynamic[] { User, requestId, this, cacheHandler, _logger }) as IDisposable)
                        {
                            var actionMethodName = action.Split('.')[1][0].ToString().ToUpper() + action.Split('.')[1].Substring(1);
                            var actionMethod = actionClassType.GetMethod(actionMethodName);
                            var actionParameters = actionMethod.GetParameters();

                            if (actionParameters.Count() > 0)
                            {
                                var model = req.p != null ? JsonConvert.DeserializeObject(req.p.ToString(), actionParameters[0].ParameterType) : null;
                                task = (Task<object>)actionMethod.Invoke(hubInstance, new[] { model });
                            }
                            else
                                task = (Task<object>)actionMethod.Invoke(hubInstance, null);
                            var res = await task.ConfigureAwait(false);

                            if (res != null)
                            {
                                ConcurrentQueue<string> reqQueue;
                                if (cacheHandler.MessagingRequests.TryGetValue(currentSession, out reqQueue))
                                    reqQueue.Enqueue(res.ToJSON());
                                else
                                    throw new Exception("The given key was not present in the dictionary: MessagingRequests: HandleSocket");
                            }
                        }
                    }
                    catch (Exception ex)
                    {
                        _logger.LogError(LoggingEvents.MessagingError, ex, "Socket " + currentSocket.GetHashCode());
                        var err = new MessagingErrorObject
                        {
                            Id = requestId,
                            Parameters = new ErrorObject { ErrorCode = HttpStatusCode.InternalServerError }
                        };
                        cacheHandler.MessagingRequests[currentSession].Enqueue(err.ToJSON());
                    }
                    finally
                    {
                        fullResult = new byte[0];
                    }
                }

                try
                {
                    result = await currentSocket.ReceiveAsync(new ArraySegment<byte>(buffer), CancellationToken.None);
                }
                catch
                {
                    break;
                }
            }
        }

        private async Task SendAsync(WebSocket socket, string payload)
        {
            byte[] bytes = Encoding.UTF8.GetBytes(payload);
            var buffer = new byte[1024 * 16];
            int readCount;
            using (var ms = new MemoryStream(bytes))
                while ((readCount = ms.Read(buffer, 0, buffer.Length)) > 0)
                {
                    await socket.SendAsync(new ArraySegment<byte>(buffer, 0, readCount), WebSocketMessageType.Text, ms.Position == ms.Length, CancellationToken.None); //new CancellationTokenSource(TimeSpan.FromHours(1)).Token);//.ConfigureAwait(false);
                    _logger.LogWarning("Sent to " + socket.GetHashCode() + ": " + Encoding.UTF8.GetString(buffer));
                    buffer = new byte[1024 * 16];
                }
        }
    }

[davidfowl]

@sepehr1014 it doesn't look like you handle receiving the close frame?

[sepehr1014]

@davidfowl We're reading from the socket while (currentSocket.State == WebSocketState.Open) then try to send the close frame from our side and dispose of it explicitly.
What other measures should we take?

[davidfowl]

• You need to handle the incoming close frame and send a close back (to end gracefully).
• You need to check the state of the websocket before you send.
• You need to be a wary of how much you buffer and what a bad client could do (if this app has untrusted clients) by just sending you messages without and fin (end of message true) frame.
• You need to make sure you coordinate sending with shutting down the websocket.
• You need to make sure you dispose the websocket when you're done with it
• You can use the WebSocketException.WebSocketErrorCode enum instead of looking at exception text to detect certain kinds of error.

That's just off the top of my head.

Thank you for contacting us. Due to a lack of activity on this discussion issue we're closing it in an effort to keep our backlog clean. If you believe there is a concern related to the ASP.NET Core framework, which hasn't been addressed yet, please file a new issue.

This issue will be locked after 30 more days of inactivity. If you still wish to discuss this subject after then, please create a new issue!