feat: implement graceful draining and outbound call control for SIP

.gitignore

@@ -5,3 +5,5 @@ cmd/server/server
 
 test/config.yaml
 test/*/*.mkv
+
+.claude

cmd/livekit-sip/main.go

@@ -103,7 +103,7 @@ func runService(ctx context.Context, c *cli.Command) error {
 	if err != nil {
 		return err
 	}
-	svc := service.NewService(conf, log, sipsrv, sipsrv.Stop, sipsrv.ActiveCalls, psrpcClient, bus, mon)
+	svc := service.NewService(conf, log, sipsrv, sipsrv.Stop, sipsrv.StartDrain, sipsrv.ActiveCalls, psrpcClient, bus, mon)
 	sipsrv.SetHandler(svc)
 
 	if err = sipsrv.Start(); err != nil {

pkg/config/config.go

@@ -108,6 +108,10 @@ type Config struct {
 	HideInboundPort bool `yaml:"hide_inbound_port"`
 	// AddRecordRoute forces SIP to add Record-Route headers to the responses.
 	AddRecordRoute bool `yaml:"add_record_route"`
+	// DisableOutboundCalls prevents creation of new outbound SIP calls.
+	// When enabled, CreateSIPParticipant requests will be rejected.
+	// The client component still runs to handle responses (e.g., BYE) for inbound calls.
+	DisableOutboundCalls bool `yaml:"disable_outbound_calls"`
 
 	// AudioDTMF forces SIP to generate audio DTMF tones in addition to digital.
 	AudioDTMF              bool    `yaml:"audio_dtmf"`

pkg/service/service.go

@@ -42,6 +42,7 @@ import (
 )
 
 type sipServiceStopFunc func()
+type sipServiceStartDrainFunc func()
 type sipServiceActiveCallsFunc func() sip.ActiveCalls
 
 type Service struct {
@@ -58,6 +59,7 @@ type Service struct {
 	rpcSIPServer rpc.SIPInternalServer
 
 	sipServiceStop        sipServiceStopFunc
+	sipServiceStartDrain  sipServiceStartDrainFunc
 	sipServiceActiveCalls sipServiceActiveCallsFunc
 
 	mon      *stats.Monitor
@@ -67,7 +69,7 @@ type Service struct {
 
 func NewService(
 	conf *config.Config, log logger.Logger, srv rpc.SIPInternalServerImpl, sipServiceStop sipServiceStopFunc,
-	sipServiceActiveCalls sipServiceActiveCallsFunc, cli rpc.IOInfoClient, bus psrpc.MessageBus, mon *stats.Monitor,
+	sipServiceStartDrain sipServiceStartDrainFunc, sipServiceActiveCalls sipServiceActiveCallsFunc, cli rpc.IOInfoClient, bus psrpc.MessageBus, mon *stats.Monitor,
 ) *Service {
 	s := &Service{
 		conf: conf,
@@ -78,6 +80,7 @@ func NewService(
 		bus:         bus,
 
 		sipServiceStop:        sipServiceStop,
+		sipServiceStartDrain:  sipServiceStartDrain,
 		sipServiceActiveCalls: sipServiceActiveCalls,
 
 		mon: mon,
@@ -107,7 +110,7 @@ func NewService(
 			Handler: mux,
 		}
 
-		mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
+		healthHandler := func(w http.ResponseWriter, r *http.Request) {
 			st := s.Health()
 			var code int
 			switch st {
@@ -121,7 +124,9 @@ func NewService(
 			w.Header().Set("Content-Type", "text/plain")
 			w.WriteHeader(code)
 			_, _ = w.Write([]byte(st.String()))
-		})
+		}
+		mux.HandleFunc("/", healthHandler)
+		mux.HandleFunc("/healthz", healthHandler)
 	}
 	return s
 }
@@ -188,6 +193,13 @@ func (s *Service) Run() error {
 			s.log.Infow("shutting down")
 			s.DeregisterCreateSIPParticipantTopic()
 
+			// Start draining: stop accepting new SIP calls immediately
+			// This ensures load balancers stop routing new traffic while existing calls finish
+			if s.sipServiceStartDrain != nil {
+				s.log.Infow("starting drain: rejecting new SIP calls")
+				s.sipServiceStartDrain()
+			}
+
 			if !s.killed.Load() {
 				shutdownTicker := time.NewTicker(5 * time.Second)
 				defer shutdownTicker.Stop()
@@ -208,6 +220,14 @@ func (s *Service) Run() error {
 			}
 
 			s.sipServiceStop()
+
+			// Keep health server running for a grace period to allow load balancer
+			// to detect unhealthy status. Health endpoint already returns 503.
+			if s.healthServer != nil {
+				s.log.Infow("waiting for load balancer to detect unhealthy status", "grace_period", "30s")
+				time.Sleep(30 * time.Second)
+			}
+
 			return nil
 		}
 	}

pkg/sip/client.go

@@ -138,6 +138,12 @@ func (c *Client) Start(agent *sipgo.UserAgent, sc *ServiceConfig) error {
 	return nil
 }
 
+// StartDrain stops accepting new outbound call requests but keeps existing calls open.
+// This allows the client to gracefully drain while completing ongoing calls.
+func (c *Client) StartDrain() {
+	c.closing.Break()
+}
+
 func (c *Client) Stop() {
 	ctx := context.Background()
 	ctx, span := Tracer.Start(ctx, "sip.Client.Stop")

pkg/sip/inbound.go

@@ -303,6 +303,13 @@ func (s *Server) processInvite(req *sip.Request, tx sip.ServerTransaction) (retE
 			info.EndedAtNs = time.Now().UnixNano()
 		})
 	}()
+	// Reject new INVITE requests if server is shutting down
+	if s.closing.IsBroken() {
+		s.mon.InviteReqRaw(stats.Inbound)
+		_ = tx.Respond(sip.NewResponseFromRequest(req, sip.StatusServiceUnavailable, "Service Unavailable - Server Shutting Down", nil))
+		s.log.Infow("rejecting INVITE, server is shutting down", "fromIP", req.Source(), "toIP", req.Destination())
+		return psrpc.NewError(psrpc.Unavailable, errors.New("server is shutting down"))
+	}
 	s.mon.InviteReqRaw(stats.Inbound)
 
 	src, err := netip.ParseAddrPort(req.Source())

pkg/sip/server.go

@@ -337,6 +337,12 @@ func (s *Server) Start(agent *sipgo.UserAgent, sc *ServiceConfig, tlsConf *tls.C
 	return nil
 }
 
+// StartDrain stops accepting new INVITE requests but keeps existing calls and listeners open.
+// This allows the server to gracefully drain while completing ongoing calls.
+func (s *Server) StartDrain() {
+	s.closing.Break()
+}
+
 func (s *Server) Stop() {
 	s.closing.Break()
 	s.cmu.Lock()

pkg/sip/service.go

@@ -182,6 +182,13 @@ func (s *Service) ActiveCalls() ActiveCalls {
 	return st
 }
 
+// StartDrain stops accepting new SIP calls (both inbound and outbound) but keeps existing calls
+// and listeners open. This allows the service to gracefully drain while completing ongoing calls.
+func (s *Service) StartDrain() {
+	s.cli.StartDrain()
+	s.srv.StartDrain()
+}
+
 func (s *Service) Stop() {
 	s.cli.Stop()
 	s.srv.Stop()
@@ -307,11 +314,18 @@ func (s *Service) Start() error {
 }
 
 func (s *Service) CreateSIPParticipant(ctx context.Context, req *rpc.InternalCreateSIPParticipantRequest) (*rpc.InternalCreateSIPParticipantResponse, error) {
+	if s.conf.DisableOutboundCalls {
+		return nil, psrpc.NewErrorf(psrpc.Unimplemented, "outbound calls are disabled on this instance")
+	}
 	resp, err := s.cli.CreateSIPParticipant(ctx, req)
 	return resp, siperrors.ApplySIPStatus(err)
 }
 
 func (s *Service) CreateSIPParticipantAffinity(ctx context.Context, req *rpc.InternalCreateSIPParticipantRequest) float32 {
+	if s.conf.DisableOutboundCalls {
+		// Return 0 affinity when outbound calls are disabled to prevent routing to this instance
+		return 0
+	}
 	// TODO: scale affinity based on a number or active calls?
 	return 0.5
 }

test/cloud/service.go

@@ -21,7 +21,7 @@ func NewService(conf *IntegrationConfig, bus psrpc.MessageBus) (*service.Service
 	if err != nil {
 		return nil, err
 	}
-	svc := service.NewService(conf.Config, logger.GetLogger(), sipsrv, sipsrv.Stop, sipsrv.ActiveCalls, psrpcClient, bus, mon)
+	svc := service.NewService(conf.Config, logger.GetLogger(), sipsrv, sipsrv.Stop, sipsrv.StartDrain, sipsrv.ActiveCalls, psrpcClient, bus, mon)
 	sipsrv.SetHandler(svc)
 
 	if err = sipsrv.Start(); err != nil {