fix(server): make MonitorHealthChan resilient to slow and closed notifiers

internal/server/health_monitor_internal_test.go

@@ -0,0 +1,95 @@
+// Copyright © 2026, SAS Institute Inc., Cary, NC, USA.  All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package server
+
+import (
+	"testing"
+	"time"
+
+	pb "github.com/sassoftware/arke/api"
+	"github.com/stretchr/testify/assert"
+)
+
+// resetHealthNotifiers empties the package-level notifier registry so each test
+// starts clean and MonitorHealthChan does not fan out to notifiers left over
+// from other tests.
+func resetHealthNotifiers() {
+	for _, addr := range healthNotifiers.GetList() {
+		healthNotifiers.Delete(addr)
+	}
+}
+
+// TestMonitorHealthChanDoesNotBlockOnBusyReceiver verifies that one client whose
+// reader is not ready cannot stall delivery to the others: the busy notifier's
+// send is dropped and the ready client still receives the broadcast.
+func TestMonitorHealthChanDoesNotBlockOnBusyReceiver(t *testing.T) {
+	resetHealthNotifiers()
+	defer resetHealthNotifiers()
+
+	busy := make(chan pb.HealthStatus_Code)     // unbuffered, no reader -> never ready
+	ready := make(chan pb.HealthStatus_Code, 1) // buffered -> always accepts
+	notifyHealth("busy", busy)
+	notifyHealth("ready", ready)
+
+	receiver := make(chan pb.HealthStatus_Code)
+	done := make(chan struct{})
+	go func() {
+		MonitorHealthChan(receiver)
+		close(done)
+	}()
+
+	receiver <- pb.HealthStatus_GOAWAY
+
+	select {
+	case code := <-ready:
+		assert.Equal(t, pb.HealthStatus_GOAWAY, code)
+	case <-time.After(time.Second):
+		t.Fatal("broadcast blocked on a busy receiver; ready client never got the code")
+	}
+
+	close(receiver)
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatal("MonitorHealthChan did not exit after its receiver was closed")
+	}
+}
+
+// TestMonitorHealthChanSurvivesClosedNotifier verifies that a notifier closed
+// concurrently (client disconnect or re-registration) does not panic the monitor
+// goroutine, and that delivery to other clients continues.
+func TestMonitorHealthChanSurvivesClosedNotifier(t *testing.T) {
+	resetHealthNotifiers()
+	defer resetHealthNotifiers()
+
+	closed := make(chan pb.HealthStatus_Code, 1)
+	notifyHealth("closed", closed)
+	close(closed) // simulate a client whose channel was closed on disconnect/re-register
+
+	live := make(chan pb.HealthStatus_Code, 1)
+	notifyHealth("live", live)
+
+	receiver := make(chan pb.HealthStatus_Code)
+	done := make(chan struct{})
+	go func() {
+		MonitorHealthChan(receiver)
+		close(done)
+	}()
+
+	receiver <- pb.HealthStatus_GOAWAY
+
+	select {
+	case code := <-live:
+		assert.Equal(t, pb.HealthStatus_GOAWAY, code)
+	case <-time.After(time.Second):
+		t.Fatal("monitor did not deliver to the live client (panic on the closed notifier?)")
+	}
+
+	close(receiver)
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatal("MonitorHealthChan did not exit (panic on the closed notifier?)")
+	}
+}

internal/server/server.go

@@ -781,12 +781,40 @@ func MonitorHealthChan(receiver chan pb.HealthStatus_Code) {
 		for _, clientAddr := range healthNotifiers.GetList() {
 			if notifierInt, ok := healthNotifiers.Get(clientAddr); ok {
 				notifier := notifierInt.(chan pb.HealthStatus_Code)
-				notifier <- code
+				sendHealthCode(clientAddr, notifier, code)
 			}
 		}
 	}
 }
 
+// sendHealthCode delivers a single health code to one client's notifier without
+// letting that client stall the broadcast to every other client.
+//
+// The send is non-blocking: a client's reader can be busy (e.g. still in
+// stream.Send to a slow connection) and not yet back at its receive, so a
+// blocking send here would hold up the entire broadcast loop and starve all
+// other clients. If the receiver is not ready we drop the code rather than
+// block.
+//
+// It also recovers from a send on a closed channel: a client's notifier is
+// closed when it disconnects (Check's defer) or re-registers (notifyHealth),
+// and that can race with this send because the registry lock is not held across
+// the lookup and the send. Recovering keeps the monitor goroutine alive instead
+// of panicking the whole broadcast on a routine disconnect.
+func sendHealthCode(clientAddr string, notifier chan pb.HealthStatus_Code, code pb.HealthStatus_Code) {
+	defer func() {
+		if r := recover(); r != nil {
+			util.Logger.Debugf("dropped health notification to %s: receiver closed", clientAddr)
+		}
+	}()
+
+	select {
+	case notifier <- code:
+	default:
+		util.Logger.Debugf("dropped health notification to %s: receiver not ready", clientAddr)
+	}
+}
+
 func (s *HealthzServer) Check(stream pb.Healthz_CheckServer) error {
 	ctx := stream.Context()
 
@@ -795,7 +823,11 @@ func (s *HealthzServer) Check(stream pb.Healthz_CheckServer) error {
 		return err
 	}
 
-	notifyHealthChan := make(chan pb.HealthStatus_Code)
+	// Buffer of 1 so a broadcast is not lost while this reader is momentarily
+	// busy (e.g. in stream.Send below). MonitorHealthChan delivers non-blocking
+	// via sendHealthCode, so without a free slot a transient-busy reader would
+	// drop the code — notably a GOAWAY the client needs to see.
+	notifyHealthChan := make(chan pb.HealthStatus_Code, 1)
 	notifyHealth(clientAddr, notifyHealthChan)
 	defer func() {
 		healthNotifiers.Delete(clientAddr)