fix(multiplayer): bound websocket writes

server/api/events.go

@@ -23,6 +23,7 @@ const (
 	maxWebSocketFrame            = 64 << 10
 	eventQueueCapacity           = 32
 	webSocketIdleLimit           = 2 * time.Minute
+	webSocketWriteLimit          = 10 * time.Second
 	webSocketMessageLimit        = 120
 	webSocketMessageWindow       = time.Minute
 	maxEventConnectionsPerPlayer = 2
@@ -198,7 +199,7 @@ func (s *Service) controlPlaneEvent(w http.ResponseWriter, r *http.Request) {
 				return
 			}
 			writeMu.Lock()
-			err := writeWebSocketFrame(connection, 0x1, payload)
+			err := writeWebSocketFrameWithDeadline(connection, 0x1, payload, webSocketWriteLimit)
 			writeMu.Unlock()
 			if err != nil {
 				return
@@ -284,7 +285,7 @@ func readWebSocketFrames(connection net.Conn, writeMu *sync.Mutex) {
 		}
 		if opcode == 0x9 {
 			writeMu.Lock()
-			_ = writeWebSocketFrame(connection, 0xA, nil)
+			_ = writeWebSocketFrameWithDeadline(connection, 0xA, nil, webSocketWriteLimit)
 			writeMu.Unlock()
 		}
 	}
@@ -381,3 +382,13 @@ func writeWebSocketFrame(connection net.Conn, opcode byte, payload []byte) error
 	_, err := connection.Write(payload)
 	return err
 }
+
+func writeWebSocketFrameWithDeadline(connection net.Conn, opcode byte, payload []byte, timeout time.Duration) error {
+	if timeout <= 0 {
+		return errors.New("invalid websocket write timeout")
+	}
+	if err := connection.SetWriteDeadline(time.Now().Add(timeout)); err != nil {
+		return err
+	}
+	return writeWebSocketFrame(connection, opcode, payload)
+}

server/api/events_test.go

@@ -3,6 +3,7 @@ package api
 import (
 	"bufio"
 	"bytes"
+	"net"
 	"net/http"
 	"net/http/httptest"
 	"testing"
@@ -25,6 +26,24 @@ func TestWebSocketReaderRejectsOversizedControlFrame(t *testing.T) {
 	}
 }
 
+func TestWebSocketWriterDoesNotBlockForeverOnSlowClient(t *testing.T) {
+	sender, receiver := net.Pipe()
+	defer sender.Close()
+	defer receiver.Close()
+	done := make(chan error, 1)
+	go func() {
+		done <- writeWebSocketFrameWithDeadline(sender, 0x1, bytes.Repeat([]byte{'x'}, maxWebSocketFrame), 20*time.Millisecond)
+	}()
+	select {
+	case err := <-done:
+		if err == nil {
+			t.Fatal("write to a non-reading client unexpectedly succeeded")
+		}
+	case <-time.After(time.Second):
+		t.Fatal("write to a non-reading client blocked past its deadline")
+	}
+}
+
 func TestWebSocketHandshakeRequiresRFC6455Version(t *testing.T) {
 	service := &Service{}
 	request := httptest.NewRequest(http.MethodGet, "/v1/events", nil)