admin: fix 'send on closed channel' panic in worker gRPC server (#10175)

* admin: never close the worker outgoing channel while senders are live conn.outgoing has multiple concurrent senders (heartbeat, task assignment, log request, registration handlers). Closing it on connection teardown raced a sender and paniced with "send on closed channel" — reliably reproduced when a laptop goes idle: heartbeats stall past the 2-minute stale cutoff, the cleanup routine closes the channel, and the resumed worker's heartbeat is received and handled at the same moment. The connection context is already the sole teardown signal, so stop closing the channel entirely. handleOutgoingMessages exits on conn.ctx.Done(), and the buffered channel is GC'd once the connection drops. Route sends through a sendToWorker helper that also selects on conn.ctx.Done() so they bail on teardown instead of blocking for the full timeout. * admin: bail on ctx.Done() while waiting for a worker log response
2026-08-19 21:56:54 +00:00 · 2026-06-30 21:22:28 -07:00
parent bdcc3154ed
commit 36e51e5542
2 changed files with 122 additions and 35 deletions
@@ -154,11 +154,11 @@ func (s *WorkerGrpcServer) Stop() error {
 	s.running = false
 	close(s.stopChan)

-	// Close all worker connections
+	// Close all worker connections. Cancelling the context stops each
+	// handleOutgoingMessages goroutine; the outgoing channel is never closed.
 	s.connMutex.Lock()
 	for _, conn := range s.connections {
 		conn.cancel()
-		s.safeCloseOutgoingChannel(conn, "Stop")
 	}
 	s.connections = make(map[string]*WorkerConnection)
 	s.connMutex.Unlock()
@@ -227,10 +227,9 @@ func (s *WorkerGrpcServer) WorkerStream(stream worker_pb.WorkerService_WorkerStr
 	s.connMutex.Lock()
 	if oldConn, exists := s.connections[workerID]; exists {
 		glog.Infof("Worker %s reconnected, cleaning up old connection", workerID)
-		// Cancel old connection to stop its goroutines
+		// Cancel old connection to stop its goroutines. Its handleOutgoingMessages
+		// exits on the cancelled context; the outgoing channel is never closed.
 		oldConn.cancel()
-		// Don't close oldConn.outgoing here as it may cause panic in handleOutgoingMessages
-		// Let the goroutine exit naturally when it detects context cancellation
 	}
 	s.connections[workerID] = conn
 	s.connMutex.Unlock()
@@ -255,11 +254,8 @@ func (s *WorkerGrpcServer) WorkerStream(stream worker_pb.WorkerService_WorkerStr
 		},
 	}

-	select {
-	case conn.outgoing <- regResponse:
+	if s.sendToWorker(conn, regResponse, 5*time.Second, "registration response") {
 		glog.V(1).Infof("Registration response sent to worker %s", workerID)
-	case <-time.After(5 * time.Second):
-		glog.Errorf("Failed to send registration response to worker %s", workerID)
 	}

 	// Handle incoming messages
@@ -384,11 +380,7 @@ func (s *WorkerGrpcServer) handleHeartbeat(conn *WorkerConnection, heartbeat *wo
 		},
 	}

-	select {
-	case conn.outgoing <- response:
-	case <-time.After(time.Second):
-		glog.Warningf("Failed to send heartbeat response to worker %s", conn.workerID)
-	}
+	s.sendToWorker(conn, response, time.Second, "heartbeat response")
 }

 // handleTaskRequest processes task requests from workers
@@ -435,11 +427,7 @@ func (s *WorkerGrpcServer) handleTaskRequest(conn *WorkerConnection, request *wo
 			},
 		}

-		select {
-		case conn.outgoing <- assignment:
-		case <-time.After(time.Second):
-			glog.Warningf("Failed to send task assignment to worker %s", conn.workerID)
-		}
+		s.sendToWorker(conn, assignment, time.Second, "task assignment")
 	} else {
 		// Send explicit "No Task" response to prevent worker timeout
 		// Workers expect a TaskAssignment message but will sleep if TaskId is empty
@@ -452,11 +440,8 @@ func (s *WorkerGrpcServer) handleTaskRequest(conn *WorkerConnection, request *wo
 			},
 		}

-		select {
-		case conn.outgoing <- noTaskAssignment:
+		if s.sendToWorker(conn, noTaskAssignment, time.Second, "no-task response") {
 			glog.V(4).Infof("Sent 'No Task' response to worker %s", conn.workerID)
-		case <-time.After(time.Second):
-			// If we can't send, the worker will eventually time out and reconnect, which is fine
 		}
 	}
 }
@@ -596,14 +581,22 @@ func (s *WorkerGrpcServer) handleTaskLogResponse(conn *WorkerConnection, respons
 	s.logRequestsMutex.Unlock()
 }

-// safeCloseOutgoingChannel safely closes the outgoing channel for a worker connection.
-func (s *WorkerGrpcServer) safeCloseOutgoingChannel(conn *WorkerConnection, source string) {
-	defer func() {
-		if r := recover(); r != nil {
-			glog.V(1).Infof("%s: recovered from panic closing outgoing channel for worker %s: %v", source, conn.workerID, r)
-		}
-	}()
-	close(conn.outgoing)
+// sendToWorker queues a message on conn.outgoing, which is deliberately never
+// closed: it has multiple concurrent senders, so closing it could panic one of
+// them with "send on closed channel". Teardown is signaled via conn.ctx instead;
+// handleOutgoingMessages drains the channel until that context is cancelled.
+// Returns false if the connection closed or the send timed out.
+func (s *WorkerGrpcServer) sendToWorker(conn *WorkerConnection, msg *worker_pb.AdminMessage, timeout time.Duration, description string) bool {
+	select {
+	case conn.outgoing <- msg:
+		return true
+	case <-conn.ctx.Done():
+		glog.V(2).Infof("Dropped %s for worker %s: connection closed", description, conn.workerID)
+		return false
+	case <-time.After(timeout):
+		glog.Warningf("Failed to send %s to worker %s: timeout", description, conn.workerID)
+		return false
+	}
 }

 // unregisterWorker removes a worker connection
@@ -628,12 +621,11 @@ func (s *WorkerGrpcServer) unregisterWorker(conn *WorkerConnection, event string
 	s.connMutex.Unlock()
 	stats_collect.AdminWorkerEventsTotal.WithLabelValues(event).Inc()

-	// Cancel context to signal goroutines to stop
+	// Cancel context to signal goroutines to stop. The outgoing channel is
+	// never closed (it has multiple senders); handleOutgoingMessages exits on
+	// the cancelled context.
 	conn.cancel()

-	// Safely close the outgoing channel with recover to handle potential double-close
-	s.safeCloseOutgoingChannel(conn, "unregisterWorker")
-
 	glog.V(1).Infof("Unregistered worker %s", conn.workerID)
 }

@@ -732,6 +724,13 @@ func (s *WorkerGrpcServer) RequestTaskLogs(workerID, taskID string, maxEntries i
 	select {
 	case conn.outgoing <- logRequest:
 		glog.V(1).Infof("Log request sent to worker %s for task %s", workerID, taskID)
+	case <-conn.ctx.Done():
+		s.logRequestsMutex.Lock()
+		if s.pendingLogRequests[requestKey] == requestContext {
+			delete(s.pendingLogRequests, requestKey)
+		}
+		s.logRequestsMutex.Unlock()
+		return nil, fmt.Errorf("worker %s connection closed", workerID)
 	case <-time.After(logSendTimeout):
 		// Clean up pending request on timeout
 		s.logRequestsMutex.Lock()
@@ -750,6 +749,13 @@ func (s *WorkerGrpcServer) RequestTaskLogs(workerID, taskID string, maxEntries i
 		}
 		glog.V(1).Infof("Received %d log entries for task %s from worker %s", len(response.LogEntries), taskID, workerID)
 		return response.LogEntries, nil
+	case <-conn.ctx.Done():
+		s.logRequestsMutex.Lock()
+		if s.pendingLogRequests[requestKey] == requestContext {
+			delete(s.pendingLogRequests, requestKey)
+		}
+		s.logRequestsMutex.Unlock()
+		return nil, fmt.Errorf("worker %s connection closed", workerID)
 	case <-time.After(logResponseTimeout):
 		// Clean up pending request on timeout
 		s.logRequestsMutex.Lock()
@@ -0,0 +1,81 @@
+package dash
+
+import (
+	"context"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/worker_pb"
+)
+
+func newTestWorkerConn(bufferSize int) *WorkerConnection {
+	ctx, cancel := context.WithCancel(context.Background())
+	return &WorkerConnection{
+		workerID: "w-test",
+		outgoing: make(chan *worker_pb.AdminMessage, bufferSize),
+		ctx:      ctx,
+		cancel:   cancel,
+	}
+}
+
+// TestSendToWorkerDropsOnCancelledContext verifies a send bails out via the
+// connection context instead of blocking for the full timeout once the
+// connection is torn down and its buffer is full.
+func TestSendToWorkerDropsOnCancelledContext(t *testing.T) {
+	s := &WorkerGrpcServer{adminServer: &AdminServer{}}
+	conn := newTestWorkerConn(1)
+
+	// Fill the buffer so the next send cannot enqueue.
+	conn.outgoing <- &worker_pb.AdminMessage{}
+	conn.cancel()
+
+	start := time.Now()
+	if s.sendToWorker(conn, &worker_pb.AdminMessage{}, 5*time.Second, "heartbeat response") {
+		t.Fatal("expected send to be dropped when context cancelled and buffer full")
+	}
+	if elapsed := time.Since(start); elapsed > time.Second {
+		t.Fatalf("send blocked for %v; expected an immediate context-cancelled bail", elapsed)
+	}
+}
+
+// TestHeartbeatDuringUnregisterDoesNotPanic is a regression test for the
+// "panic: send on closed channel" crash: unregisterWorker (called from the
+// stale-connection cleanup routine) used to close conn.outgoing while the
+// WorkerStream goroutine was still sending heartbeat responses to it.
+func TestHeartbeatDuringUnregisterDoesNotPanic(t *testing.T) {
+	s := &WorkerGrpcServer{
+		adminServer: &AdminServer{},
+		connections: make(map[string]*WorkerConnection),
+	}
+	conn := newTestWorkerConn(100)
+	s.connections[conn.workerID] = conn
+
+	// Drain outgoing like handleOutgoingMessages would, until teardown.
+	go func() {
+		for {
+			select {
+			case <-conn.ctx.Done():
+				return
+			case <-conn.outgoing:
+			}
+		}
+	}()
+
+	var wg sync.WaitGroup
+	wg.Add(2)
+	go func() {
+		defer wg.Done()
+		for i := 0; i < 2000; i++ {
+			s.handleHeartbeat(conn, &worker_pb.WorkerHeartbeat{})
+		}
+	}()
+	go func() {
+		defer wg.Done()
+		s.unregisterWorker(conn, "test")
+	}()
+
+	// Under the old close()-based teardown this would panic the sender goroutine
+	// and crash the test binary.
+	wg.Wait()
+}