diff --git a/pkg/sip/inbound.go b/pkg/sip/inbound.go index 7da5fe54..59637971 100644 --- a/pkg/sip/inbound.go +++ b/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)) } diff --git a/pkg/sip/options_test.go b/pkg/sip/options_test.go new file mode 100644 index 00000000..0f496134 --- /dev/null +++ b/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) +}