Wait for an ACK for inbound calls. (#445)

Author: dennwc

pkg/sip/client.go

@@ -293,7 +293,7 @@ func (c *Client) onBye(req *sip.Request, tx sip.ServerTransaction) bool {
 		_ = tx.Respond(sip.NewResponseFromRequest(req, sip.StatusCallTransactionDoesNotExists, "Call does not exist", nil))
 		return false
 	}
-	call.log.Infow("BYE")
+	call.log.Infow("BYE from remote")
 	go func(call *outboundCall) {
 		call.cc.AcceptBye(req, tx)
 		call.CloseWithReason(CallHangup, "bye", livekit.DisconnectReason_CLIENT_INITIATED)

pkg/sip/inbound.go

@@ -55,6 +55,8 @@ const (
 	// audioBridgeMaxDelay delays sending audio for certain time, unless RTP packet is received.
 	// This is done because of audio cutoff at the beginning of calls observed in the wild.
 	audioBridgeMaxDelay = 1 * time.Second
+
+	inviteOkAckTimeout = 5 * time.Second
 )
 
 // hashPassword creates a SHA256 hash of the password for logging purposes
@@ -364,21 +366,36 @@ func (s *Server) processInvite(req *sip.Request, tx sip.ServerTransaction) (retE
 }
 
 func (s *Server) onOptions(log *slog.Logger, req *sip.Request, tx sip.ServerTransaction) {
-	_ = tx.Respond(sip.NewResponseFromRequest(req, 200, "OK", nil))
+	_ = tx.Respond(sip.NewResponseFromRequest(req, sip.StatusOK, "OK", nil))
+}
+
+func (s *Server) onAck(log *slog.Logger, req *sip.Request, tx sip.ServerTransaction) {
+	tag, err := getFromTag(req)
+	if err != nil {
+		return
+	}
+	s.cmu.RLock()
+	c := s.activeCalls[tag]
+	s.cmu.RUnlock()
+	if c == nil {
+		return
+	}
+	c.log.Infow("ACK from remote")
+	c.cc.AcceptAck(req, tx)
 }
 
 func (s *Server) onBye(log *slog.Logger, req *sip.Request, tx sip.ServerTransaction) {
 	tag, err := getFromTag(req)
 	if err != nil {
-		_ = tx.Respond(sip.NewResponseFromRequest(req, 400, "", nil))
+		_ = tx.Respond(sip.NewResponseFromRequest(req, sip.StatusBadRequest, "", nil))
 		return
 	}
 
 	s.cmu.RLock()
 	c := s.activeCalls[tag]
 	s.cmu.RUnlock()
 	if c != nil {
-		c.log.Infow("BYE")
+		c.log.Infow("BYE from remote")
 		c.cc.AcceptBye(req, tx)
 		_ = c.Close()
 		return
@@ -417,7 +434,7 @@ func (s *Server) OnNoRoute(log *slog.Logger, req *sip.Request, tx sip.ServerTran
 func (s *Server) onNotify(log *slog.Logger, req *sip.Request, tx sip.ServerTransaction) {
 	tag, err := getFromTag(req)
 	if err != nil {
-		_ = tx.Respond(sip.NewResponseFromRequest(req, 400, "", nil))
+		_ = tx.Respond(sip.NewResponseFromRequest(req, sip.StatusBadRequest, "", nil))
 		return
 	}
 
@@ -597,7 +614,7 @@ func (c *inboundCall) handleInvite(ctx context.Context, req *sip.Request, trunkI
 		}
 		c.log.Infow("Accepting the call", "headers", headers)
 		if err := c.cc.Accept(ctx, answerData, headers); err != nil {
-			c.log.Errorw("Cannot respond to INVITE", err)
+			c.log.Errorw("Cannot accept the call", err)
 			return false, err
 		}
 		c.media.EnableTimeout(true)
@@ -1168,6 +1185,7 @@ type sipInbound struct {
 	nextRequestCSeq uint32
 	referCseq       uint32
 	ringing         chan struct{}
+	acked           core.Fuse
 	setHeaders      setHeadersFunc
 }
 
@@ -1343,7 +1361,7 @@ func (c *sipInbound) Accept(ctx context.Context, sdpData []byte, headers map[str
 	if c.inviteTx == nil {
 		return errors.New("call already rejected")
 	}
-	r := sip.NewResponseFromRequest(c.invite, 200, "OK", sdpData)
+	r := sip.NewResponseFromRequest(c.invite, sip.StatusOK, "OK", sdpData)
 
 	// This will effectively redirect future SIP requests to this server instance (if host address is not LB).
 	r.AppendHeader(c.contact)
@@ -1358,11 +1376,23 @@ func (c *sipInbound) Accept(ctx context.Context, sdpData []byte, headers map[str
 	if err := c.inviteTx.Respond(r); err != nil {
 		return err
 	}
+	ackCtx, ackCancel := context.WithTimeout(ctx, inviteOkAckTimeout)
+	defer ackCancel()
+	select {
+	case <-ackCtx.Done():
+		return errors.New("no ACK received for 200 OK")
+	case <-c.inviteTx.Acks():
+	case <-c.acked.Watch():
+	}
 	c.inviteOk = r
 	c.inviteTx = nil // accepted
 	return nil
 }
 
+func (c *sipInbound) AcceptAck(req *sip.Request, tx sip.ServerTransaction) {
+	c.acked.Break()
+}
+
 func (c *sipInbound) AcceptBye(req *sip.Request, tx sip.ServerTransaction) {
 	_ = tx.Respond(sip.NewResponseFromRequest(req, 200, "OK", nil))
 	c.mu.Lock()

pkg/sip/server.go

@@ -267,13 +267,12 @@ func (s *Server) Start(agent *sipgo.UserAgent, sc *ServiceConfig, tlsConf *tls.C
 
 	s.sipSrv.OnOptions(s.onOptions)
 	s.sipSrv.OnInvite(s.onInvite)
+	s.sipSrv.OnAck(s.onAck)
 	s.sipSrv.OnBye(s.onBye)
 	s.sipSrv.OnNotify(s.onNotify)
 	s.sipSrv.OnNoRoute(s.OnNoRoute)
 	s.sipUnhandled = unhandled
 
-	// Ignore ACKs
-	s.sipSrv.OnAck(func(log *slog.Logger, req *sip.Request, tx sip.ServerTransaction) {})
 	listenIP := s.conf.ListenIP
 	if listenIP == "" {
 		listenIP = "0.0.0.0"

pkg/siptest/client.go

@@ -87,7 +87,7 @@ func NewClient(id string, conf ClientConfig) (*Client, error) {
 		conf.Log.Debug("setting local address", "ip", localIP)
 	}
 	if conf.Port == 0 {
-		conf.Port = 5060 + uint16(rand.Intn(100))
+		conf.Port = 5060 + uint16(rand.Intn(1000))
 	}
 	if conf.Number == "" {
 		conf.Number = "1000"

test/integration/sip_test.go

@@ -57,7 +57,7 @@ func runSIPServer(t testing.TB, lk *LiveKit) *SIPServer {
 	if err != nil {
 		t.Fatal(err)
 	}
-	sipPort := 5060 + rand.Intn(100)
+	sipPort := 5060 + rand.Intn(1000)
 	local, err := config.GetLocalIP()
 	require.NoError(t, err)
 	conf := &config.Config{