Return 503 for OPTIONS when SIP node is not healthy (#764)

pkg/sip/inbound.go

@@ -420,6 +420,20 @@ func (s *Server) processInvite(req *sip.Request, tx sip.ServerTransaction) (retE
 		}
 	}
 
+	// New dialogs only: reject while draining / not started so graceful
+	// shutdown and OPTIONS-based load balancers stop sending fresh calls (#764).
+	// Do not reject on HealthUnderLoad — transient CPU spikes must not drop
+	// inbound INVITEs (outbound CreateSIPParticipant already gates on HealthOK).
+	// In-dialog re-INVITEs for active calls were handled above.
+	if h := s.mon.Health(); h == stats.HealthStopped || h == stats.HealthNotStarted {
+		log.Infow("rejecting new invite, node not ready", "health", h.String())
+		cmon := s.mon.NewCall(stats.Inbound, cc.From().Host, cc.To().Host)
+		cmon.InviteReq()
+		cmon.InviteErrorShort(stats.ServerError("not-ready"))
+		cc.RespondAndDrop(sip.StatusServiceUnavailable, "Service Unavailable")
+		return psrpc.NewErrorf(psrpc.Unavailable, "sip node health: %s", h.String())
+	}
+
 	from, to := cc.From(), cc.To()
 
 	cmon := s.mon.NewCall(stats.Inbound, from.Host, to.Host)
@@ -560,6 +574,31 @@ func (s *Server) processInvite(req *sip.Request, tx sip.ServerTransaction) (retE
 }
 
 func (s *Server) onOptions(log *slog.Logger, req *sip.Request, tx sip.ServerTransaction) {
+	// In-dialog OPTIONS are session keepalives from SBCs for active calls.
+	// Answer 200 even while draining so existing calls are not torn down (#764).
+	if tag, err := GetLocalTagUAS(req); err == nil {
+		s.cmu.RLock()
+		c := s.byLocalTag[tag]
+		s.cmu.RUnlock()
+		if c != nil {
+			_ = tx.Respond(sip.NewResponseFromRequest(req, sip.StatusOK, "OK", nil))
+			return
+		}
+		if s.cli != nil {
+			if oc := s.cli.getActiveCall(tag); oc != nil {
+				_ = tx.Respond(sip.NewResponseFromRequest(req, sip.StatusOK, "OK", nil))
+				return
+			}
+		}
+	}
+
+	// Out-of-dialog OPTIONS are used by SIP proxies as health probes.
+	// Mirror CreateSIPParticipant / HTTP health: only answer 200 when ready.
+	if h := s.mon.Health(); h != stats.HealthOK {
+		log.Debug("OPTIONS rejected", "health", h.String())
+		_ = tx.Respond(sip.NewResponseFromRequest(req, sip.StatusServiceUnavailable, "Service Unavailable", nil))
+		return
+	}
 	_ = tx.Respond(sip.NewResponseFromRequest(req, sip.StatusOK, "OK", nil))
 }
 

pkg/sip/options_test.go

@@ -0,0 +1,96 @@
+package sip
+
+import (
+	"log/slog"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/livekit/sipgo/sip"
+
+	"github.com/livekit/sip/pkg/config"
+	"github.com/livekit/sip/pkg/stats"
+)
+
+type captureServerTx struct {
+	resp *sip.Response
+}
+
+func (t *captureServerTx) Respond(res *sip.Response) error {
+	t.resp = res
+	return nil
+}
+func (t *captureServerTx) Terminate()            {}
+func (t *captureServerTx) Done() <-chan struct{} { return nil }
+func (t *captureServerTx) Err() error            { return nil }
+func (t *captureServerTx) Acks() <-chan *sip.Request {
+	return nil
+}
+func (t *captureServerTx) Cancels() <-chan *sip.Request {
+	return nil
+}
+
+func newOPTIONSRequest() *sip.Request {
+	return sip.NewRequest(sip.OPTIONS, sip.Uri{Host: "sip.test", Port: 5060})
+}
+
+func TestOnOptionsHealth(t *testing.T) {
+	// MaxCpuUtilization=1.0 disables the under-load path so the test is stable on busy hosts.
+	cfg := &config.Config{MaxCpuUtilization: 1.0, NodeID: "test-options"}
+	mon, err := stats.NewMonitor(cfg)
+	require.NoError(t, err)
+	require.NoError(t, mon.Start(cfg))
+	t.Cleanup(mon.Stop)
+
+	s := &Server{mon: mon, byLocalTag: make(map[LocalTag]*inboundCall)}
+	log := slog.Default()
+
+	t.Run("healthy returns 200", func(t *testing.T) {
+		require.Equal(t, stats.HealthOK, mon.Health())
+		tx := &captureServerTx{}
+		s.onOptions(log, newOPTIONSRequest(), tx)
+		require.NotNil(t, tx.resp)
+		require.Equal(t, sip.StatusCode(200), tx.resp.StatusCode)
+	})
+
+	t.Run("shutdown returns 503 for out-of-dialog", func(t *testing.T) {
+		mon.Shutdown()
+		require.Equal(t, stats.HealthStopped, mon.Health())
+		tx := &captureServerTx{}
+		s.onOptions(log, newOPTIONSRequest(), tx)
+		require.NotNil(t, tx.resp)
+		require.Equal(t, sip.StatusCode(503), tx.resp.StatusCode)
+	})
+
+	t.Run("shutdown returns 200 for in-dialog keepalive", func(t *testing.T) {
+		require.Equal(t, stats.HealthStopped, mon.Health())
+		const tag = "active-call-tag"
+		s.byLocalTag[LocalTag(tag)] = &inboundCall{}
+
+		req := newOPTIONSRequest()
+		to := &sip.ToHeader{
+			Address: sip.Uri{User: "agent", Host: "sip.test", Port: 5060},
+			Params:  sip.NewParams(),
+		}
+		to.Params.Add("tag", tag)
+		req.AppendHeader(to)
+
+		tx := &captureServerTx{}
+		s.onOptions(log, req, tx)
+		require.NotNil(t, tx.resp)
+		require.Equal(t, sip.StatusCode(200), tx.resp.StatusCode)
+	})
+}
+
+func TestOnOptionsNotStarted(t *testing.T) {
+	mon, err := stats.NewMonitor(&config.Config{MaxCpuUtilization: 0.9, NodeID: "test-ns"})
+	require.NoError(t, err)
+	// Intentionally do not Start — HealthNotStarted.
+	require.Equal(t, stats.HealthNotStarted, mon.Health())
+
+	s := &Server{mon: mon, byLocalTag: make(map[LocalTag]*inboundCall)}
+	tx := &captureServerTx{}
+	s.onOptions(slog.Default(), newOPTIONSRequest(), tx)
+	require.NotNil(t, tx.resp)
+	require.Equal(t, sip.StatusCode(503), tx.resp.StatusCode)
+}