From d91dd71870d17f2c4ff8f05534be32c6e6bbf226 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Mon, 3 Aug 2026 11:04:16 +0000 Subject: [PATCH 1/2] Return 503 for OPTIONS when SIP node is not healthy (#764) SIP proxies often probe backends with OPTIONS. Always answering 200 kept draining nodes in the pool. Mirror CreateSIPParticipant/HTTP health and answer 503 unless HealthOK. Also reject new inbound INVITEs while unhealthy; in-dialog re-INVITEs for active calls still succeed. Co-authored-by: li xuanqun <793005378@qq.com> --- pkg/sip/inbound.go | 20 +++++++++++ pkg/sip/options_test.go | 75 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 95 insertions(+) create mode 100644 pkg/sip/options_test.go diff --git a/pkg/sip/inbound.go b/pkg/sip/inbound.go index 7da5fe54..614d4165 100644 --- a/pkg/sip/inbound.go +++ b/pkg/sip/inbound.go @@ -420,6 +420,18 @@ func (s *Server) processInvite(req *sip.Request, tx sip.ServerTransaction) (retE } } + // New dialogs only: reject while draining / overloaded so graceful shutdown + // and OPTIONS-based load balancers stop sending fresh calls here (#764). + // In-dialog re-INVITEs for active calls were handled above. + if h := s.mon.Health(); h != stats.HealthOK { + 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 +572,14 @@ func (s *Server) processInvite(req *sip.Request, tx sip.ServerTransaction) (retE } func (s *Server) onOptions(log *slog.Logger, req *sip.Request, tx sip.ServerTransaction) { + // SIP proxies often probe backends with OPTIONS and remove unhealthy + // destinations from the pool. Mirror CreateSIPParticipant / HTTP health: + // only answer 200 when the node is ready to take new work (#764). + 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..3f9072b8 --- /dev/null +++ b/pkg/sip/options_test.go @@ -0,0 +1,75 @@ +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 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} + log := slog.Default() + req := sip.NewRequest(sip.OPTIONS, sip.Uri{Host: "sip.test", Port: 5060}) + + t.Run("healthy returns 200", func(t *testing.T) { + require.Equal(t, stats.HealthOK, mon.Health()) + tx := &captureServerTx{} + s.onOptions(log, req, tx) + require.NotNil(t, tx.resp) + require.Equal(t, sip.StatusCode(200), tx.resp.StatusCode) + }) + + t.Run("shutdown returns 503", func(t *testing.T) { + mon.Shutdown() + require.Equal(t, stats.HealthStopped, mon.Health()) + tx := &captureServerTx{} + s.onOptions(log, req, tx) + require.NotNil(t, tx.resp) + require.Equal(t, sip.StatusCode(503), 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} + tx := &captureServerTx{} + req := sip.NewRequest(sip.OPTIONS, sip.Uri{Host: "sip.test", Port: 5060}) + s.onOptions(slog.Default(), req, tx) + require.NotNil(t, tx.resp) + require.Equal(t, sip.StatusCode(503), tx.resp.StatusCode) +} From 3fc69eaaad8cc5f8b910677439b30d4b2853312b Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Mon, 3 Aug 2026 11:12:46 +0000 Subject: [PATCH 2/2] Fix OPTIONS drain: keep in-dialog keepalives, don't reject INVITE on UnderLoad Out-of-dialog OPTIONS still return 503 when unhealthy. In-dialog OPTIONS for active calls answer 200 during drain so SBCs don't tear down live sessions. New INVITEs are rejected only when Stopped/NotStarted, not on transient HealthUnderLoad (which broke CI on busy runners). Co-authored-by: li xuanqun <793005378@qq.com> --- pkg/sip/inbound.go | 31 +++++++++++++++++++++++------ pkg/sip/options_test.go | 43 ++++++++++++++++++++++++++++++----------- 2 files changed, 57 insertions(+), 17 deletions(-) diff --git a/pkg/sip/inbound.go b/pkg/sip/inbound.go index 614d4165..59637971 100644 --- a/pkg/sip/inbound.go +++ b/pkg/sip/inbound.go @@ -420,10 +420,12 @@ func (s *Server) processInvite(req *sip.Request, tx sip.ServerTransaction) (retE } } - // New dialogs only: reject while draining / overloaded so graceful shutdown - // and OPTIONS-based load balancers stop sending fresh calls here (#764). + // 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.HealthOK { + 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() @@ -572,9 +574,26 @@ func (s *Server) processInvite(req *sip.Request, tx sip.ServerTransaction) (retE } func (s *Server) onOptions(log *slog.Logger, req *sip.Request, tx sip.ServerTransaction) { - // SIP proxies often probe backends with OPTIONS and remove unhealthy - // destinations from the pool. Mirror CreateSIPParticipant / HTTP health: - // only answer 200 when the node is ready to take new work (#764). + // 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)) diff --git a/pkg/sip/options_test.go b/pkg/sip/options_test.go index 3f9072b8..0f496134 100644 --- a/pkg/sip/options_test.go +++ b/pkg/sip/options_test.go @@ -20,9 +20,9 @@ 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) 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 } @@ -30,6 +30,10 @@ 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"} @@ -38,26 +42,44 @@ func TestOnOptionsHealth(t *testing.T) { require.NoError(t, mon.Start(cfg)) t.Cleanup(mon.Stop) - s := &Server{mon: mon} + s := &Server{mon: mon, byLocalTag: make(map[LocalTag]*inboundCall)} log := slog.Default() - req := sip.NewRequest(sip.OPTIONS, sip.Uri{Host: "sip.test", Port: 5060}) t.Run("healthy returns 200", func(t *testing.T) { require.Equal(t, stats.HealthOK, mon.Health()) tx := &captureServerTx{} - s.onOptions(log, req, tx) + s.onOptions(log, newOPTIONSRequest(), tx) require.NotNil(t, tx.resp) require.Equal(t, sip.StatusCode(200), tx.resp.StatusCode) }) - t.Run("shutdown returns 503", func(t *testing.T) { + 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, req, tx) + 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) { @@ -66,10 +88,9 @@ func TestOnOptionsNotStarted(t *testing.T) { // Intentionally do not Start — HealthNotStarted. require.Equal(t, stats.HealthNotStarted, mon.Health()) - s := &Server{mon: mon} + s := &Server{mon: mon, byLocalTag: make(map[LocalTag]*inboundCall)} tx := &captureServerTx{} - req := sip.NewRequest(sip.OPTIONS, sip.Uri{Host: "sip.test", Port: 5060}) - s.onOptions(slog.Default(), req, tx) + s.onOptions(slog.Default(), newOPTIONSRequest(), tx) require.NotNil(t, tx.resp) require.Equal(t, sip.StatusCode(503), tx.resp.StatusCode) }