fix(api): prevent idle TCP connections from blocking the accept loop

Move per-connection protocol detection (TLS handshake, reader.Peek) out
of the accept loop and into a per-connection goroutine. An idle TCP
connection that never sends bytes would previously block Peek(1)
indefinitely, preventing all subsequent connections from being accepted
and making the management/API server unresponsive.

Closes #3267

internal/api/protocol_multiplexer.go

@@ -7,6 +7,7 @@ import (
 	"net"
 	"net/http"
 	"strings"
+	"time"
 
 	log "github.com/sirupsen/logrus"
 )
@@ -48,68 +49,88 @@ func (s *Server) acceptMuxConnections(listener net.Listener, httpListener *muxLi
 			continue
 		}
 
-		tlsConn, ok := conn.(*tls.Conn)
+		// Dispatch each connection to a goroutine so that slow/idle clients
-		if ok {
+		// cannot block the accept loop. Previously, TLS handshake and
-			if errHandshake := tlsConn.Handshake(); errHandshake != nil {
+		// reader.Peek(1) were performed inline; an idle TCP connection that
-				if errClose := conn.Close(); errClose != nil {
+		// never sent bytes would block Peek indefinitely, preventing all
-					log.Errorf("failed to close connection after TLS handshake error: %v", errClose)
+		// subsequent connections from being accepted (issue #3267).
-				}
+		go s.routeMuxConnection(conn, httpListener)
-				continue
+	}
-			}
+}
-			proto := strings.TrimSpace(tlsConn.ConnectionState().NegotiatedProtocol)
+
-			if proto == "h2" || proto == "http/1.1" {
+// routeMuxConnection performs per-connection protocol detection and routing.
-				if httpListener == nil {
+func (s *Server) routeMuxConnection(conn net.Conn, httpListener *muxListener) {
-					if errClose := conn.Close(); errClose != nil {
+	// Set a read deadline so that idle connections that never send bytes do not
-						log.Errorf("failed to close connection: %v", errClose)
+	// leak goroutines and file descriptors. The deadline is cleared once the
-					}
+	// connection is successfully routed to its handler.
-					continue
+	const muxSniffDeadline = 10 * time.Second
-				}
+	_ = conn.SetReadDeadline(time.Now().Add(muxSniffDeadline))
-				if errPut := httpListener.Put(tlsConn); errPut != nil {
+
-					if errClose := conn.Close(); errClose != nil {
+	tlsConn, ok := conn.(*tls.Conn)
-						log.Errorf("failed to close connection after HTTP routing failure: %v", errClose)
+	if ok {
-					}
+		if errHandshake := tlsConn.Handshake(); errHandshake != nil {
-				}
+			if errClose := conn.Close(); errClose != nil {
-				continue
+				log.Errorf("failed to close connection after TLS handshake error: %v", errClose)
-			}
+			}
-		}
+			return
-
+		}
-		reader := bufio.NewReader(conn)
+		proto := strings.TrimSpace(tlsConn.ConnectionState().NegotiatedProtocol)
-		prefix, errPeek := reader.Peek(1)
+		if proto == "h2" || proto == "http/1.1" {
-		if errPeek != nil {
+			if httpListener == nil {
-			if errClose := conn.Close(); errClose != nil {
+				if errClose := conn.Close(); errClose != nil {
-				log.Errorf("failed to close connection after protocol peek failure: %v", errClose)
+					log.Errorf("failed to close connection: %v", errClose)
-			}
+				}
-			continue
+				return
-		}
+			}
-
+			if errPut := httpListener.Put(tlsConn); errPut != nil {
-		if isRedisRESPPrefix(prefix[0]) {
+				if errClose := conn.Close(); errClose != nil {
-			if s.cfg != nil && s.cfg.Home.Enabled {
+					log.Errorf("failed to close connection after HTTP routing failure: %v", errClose)
-				if errClose := conn.Close(); errClose != nil {
+				}
-					log.Errorf("failed to close redis connection while home mode is enabled: %v", errClose)
+			} else {
-				}
+				_ = conn.SetReadDeadline(time.Time{})
-				continue
+			}
-			}
+			return
-			if !s.managementRoutesEnabled.Load() {
+		}
-				if errClose := conn.Close(); errClose != nil {
+	}
-					log.Errorf("failed to close redis connection while management is disabled: %v", errClose)
+
-				}
+	reader := bufio.NewReader(conn)
-				continue
+	prefix, errPeek := reader.Peek(1)
-			}
+	if errPeek != nil {
-			go s.handleRedisConnection(conn, reader)
+		if errClose := conn.Close(); errClose != nil {
-			continue
+			log.Errorf("failed to close connection after protocol peek failure: %v", errClose)
-		}
+		}
-
+		return
-		if httpListener == nil {
+	}
-			if errClose := conn.Close(); errClose != nil {
+
-				log.Errorf("failed to close connection without HTTP listener: %v", errClose)
+	if isRedisRESPPrefix(prefix[0]) {
-			}
+		if s.cfg != nil && s.cfg.Home.Enabled {
-			continue
+			if errClose := conn.Close(); errClose != nil {
-		}
+				log.Errorf("failed to close redis connection while home mode is enabled: %v", errClose)
-
+			}
-		if errPut := httpListener.Put(&bufferedConn{Conn: conn, reader: reader}); errPut != nil {
+			return
-			if errClose := conn.Close(); errClose != nil {
+		}
-				log.Errorf("failed to close connection after HTTP routing failure: %v", errClose)
+		if !s.managementRoutesEnabled.Load() {
-			}
+			if errClose := conn.Close(); errClose != nil {
-		}
+				log.Errorf("failed to close redis connection while management is disabled: %v", errClose)
+			}
+			return
+		}
+		s.handleRedisConnection(conn, reader)
+		return
+	}
+
+	if httpListener == nil {
+		if errClose := conn.Close(); errClose != nil {
+			log.Errorf("failed to close connection without HTTP listener: %v", errClose)
+		}
+		return
+	}
+
+	if errPut := httpListener.Put(&bufferedConn{Conn: conn, reader: reader}); errPut != nil {
+		if errClose := conn.Close(); errClose != nil {
+			log.Errorf("failed to close connection after HTTP routing failure: %v", errClose)
+		}
+	} else {
+		_ = conn.SetReadDeadline(time.Time{})
 	}
 }

internal/api/protocol_multiplexer_test.go

@@ -0,0 +1,65 @@
+package api
+
+import (
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+func TestAcceptMuxNotBlockedByIdleConnection(t *testing.T) {
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("failed to listen: %v", err)
+	}
+	defer listener.Close()
+
+	var routed atomic.Int32
+	handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		routed.Add(1)
+		w.WriteHeader(http.StatusOK)
+	})
+	srv := httptest.NewUnstartedServer(handler)
+	defer srv.Close()
+
+	muxLn := newMuxListener(listener.Addr(), 1024)
+	server := &Server{managementRoutesEnabled: atomic.Bool{}}
+	server.managementRoutesEnabled.Store(false)
+
+	errCh := make(chan error, 1)
+	go func() {
+		errCh <- server.acceptMuxConnections(listener, muxLn)
+	}()
+
+	srv.Listener = muxLn
+	srv.Start()
+
+	// Open an idle TCP connection that never sends any bytes.
+	idleConn, err := net.DialTimeout("tcp", listener.Addr().String(), 2*time.Second)
+	if err != nil {
+		t.Fatalf("failed to dial idle connection: %v", err)
+	}
+	defer idleConn.Close()
+
+	// Give the accept loop time to pick up the idle connection.
+	time.Sleep(50 * time.Millisecond)
+
+	// Send a real HTTP request. Before the fix, the accept loop would be
+	// blocked on Peek(1) for the idle connection, causing this request to
+	// time out.
+	client := &http.Client{Timeout: 3 * time.Second}
+	resp, err := client.Get("http://" + listener.Addr().String() + "/")
+	if err != nil {
+		listener.Close()
+		t.Fatalf("HTTP request failed (accept loop may be blocked by idle connection): %v", err)
+	}
+	resp.Body.Close()
+
+	listener.Close()
+
+	if routed.Load() == 0 {
+		t.Error("expected at least one request to be routed")
+	}
+}