Potential deadlock with NetworkStreamSession with Websockets

[mbrandenburger]

The current implementation of the NetworkStreamSession acquires the sessionsMutex and calls close on each stream when closing a session.

With websockets, it may happen that the close call hangs when trying to write c.reads <- streamEOF within the multiplexedBaseConn implementation, see here. This may happen when the connection will closed before c.reads is consumed.

This causes the caller of close to block forever, thus, the sessionMutex in [NetworkStreamSession] is not released anymore.

One approach to overcome this issue is to perform a non-blocking write to c.reads and then complete the close() call.

select {
case c.reads <- streamEOF:
default:
	logger.Warnf("cannot send EOF to close connection")
}

[alexandrosfilios]

@mbrandenburger Currently we are writing tests for the comm layer on the branch test-comm. Let's try to reproduce and capture it in a unit test.

[mbrandenburger]

#887 adds a test that triggers the experiences issue. The test fails.

The following patch helped to make the test passing

diff --git a/platform/view/services/comm/host/rest/websocket/multiplexed_provider.go b/platform/view/services/comm/host/rest/websocket/multiplexed_provider.go
index 01189428..fa54d270 100644
--- a/platform/view/services/comm/host/rest/websocket/multiplexed_provider.go
+++ b/platform/view/services/comm/host/rest/websocket/multiplexed_provider.go
@@ -424,7 +424,11 @@ func (c *subConn) close(client bool) {
 			c.writes <- MultiplexedMessage{ID: c.id, Err: "EOF"}
 			<-c.writeErrs
 		}
-		c.reads <- streamEOF
+		select {
+		case c.reads <- streamEOF:
+		default:
+			logger.Warnf("cannot send EOF to close connection")
+		}
 		c.closes <- c.id
 	})
 }
diff --git a/platform/view/services/comm/host/rest/websocket/stream.go b/platform/view/services/comm/host/rest/websocket/stream.go
index c4635544..1900e155 100644
--- a/platform/view/services/comm/host/rest/websocket/stream.go
+++ b/platform/view/services/comm/host/rest/websocket/stream.go
@@ -58,18 +58,19 @@ func (s *stream) readMessages(ctx context.Context) {
 	for {
 		select {
 		case <-ctx.Done():
-			logger.Warnf("Context for stream [%s] closed by us. Error: %w", s.Hash(), ctx.Err())
+			logger.Debugf("Context for stream [%s] closed by us. Error: %v", s.Hash(), ctx.Err())
 			s.reads <- streamEOF
+			s.Close()
 			return
 		default:
 			_, msg, err := s.conn.ReadMessage()
 			if err != nil && (websocket.IsCloseError(err, websocket.CloseAbnormalClosure)) {
-				logger.Debugf("Websocket connection closed unexpectedly")
+				logger.Debugf("Websocket connection closed unexpectedly: %v", err)
 				s.reads <- streamEOF
 				s.Close()
 				return
 			}
-			logger.Debugf("Read message of length [%d] on [%s]. Error encountered: %w", len(msg), s.Hash(), err)
+			logger.Debugf("Read message of length [%d] on [%s]", len(msg), s.Hash())
 			s.reads <- result{value: msg, err: err}
 		}
 	}

However, note that the patch is more a quick fix than a proper solution as I haven't checked the consequences of NOT sending the EOF. :/ Moreover, the patch contains some log taming.