diff --git a/internal/server/health_monitor_internal_test.go b/internal/server/health_monitor_internal_test.go new file mode 100644 index 0000000..6741a61 --- /dev/null +++ b/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?)") + } +} diff --git a/internal/server/server.go b/internal/server/server.go index 4b1729a..9f876ec 100644 --- a/internal/server/server.go +++ b/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)