TEL-392: cache the drop and respond to retried failed calls (#694)

* TEL-392: cache the drop and respond to retried failed calls

* add config knob

Author: hechen-eng

pkg/config/config.go

@@ -116,6 +116,10 @@ type Config struct {
 	// Setting it to true makes SIP server silently drop INVITE requests if it gets a negative Auth or Dispatch response.
 	// Doing so hides our SIP endpoint from (a low effort) port scanners.
 	HideInboundPort bool `yaml:"hide_inbound_port"`
+	// DisableRejectedInviteCache turns off the per-server cache that replays
+	// a final INVITE rejection (keyed by Call-ID + From-tag) for retries
+	// reusing the same identifiers.
+	DisableRejectedInviteCache bool `yaml:"disable_rejected_invite_cache"`
 	// AddRecordRoute forces SIP to add Record-Route headers to the responses.
 	AddRecordRoute bool `yaml:"add_record_route"`
 

pkg/sip/inbound.go

@@ -362,6 +362,17 @@ func (s *Server) processInvite(req *sip.Request, tx sip.ServerTransaction) (retE
 	log := cc.log.WithValues("transport", tr, "tid", tid.String())
 	cc.log = log
 
+	// Replay cached final rejection for retries reusing the same Call-ID +
+	// From-tag (e.g. provider-level failover after a 4xx). Skips creating
+	// a duplicate call object and the OnSessionEnd side-effects that follow.
+	if s.rejectedInvites != nil {
+		if prev, ok := s.rejectedInvites.Get([2]string{cc.SIPCallID(), string(cc.Tag())}); ok {
+			log.Debugw("replaying cached INVITE rejection", "status", prev.status, "reason", prev.reason)
+			cc.RespondAndDrop(prev.status, prev.reason)
+			return nil
+		}
+	}
+
 	log.Infow("processing invite")
 
 	s.cmu.RLock()
@@ -1720,6 +1731,15 @@ func (c *sipInbound) RespondAndDrop(status sip.StatusCode, reason string) {
 	c.stopRinging()
 	c.respond(status, reason)
 	c.drop()
+	// Cache the response so a retry reusing the same Call-ID + From-tag
+	// (e.g. provider failover after a 4xx) gets the cached reply replayed
+	// instead of running through the handler again.
+	if c.s != nil && c.s.rejectedInvites != nil && status >= 300 && c.sipCallID != "" {
+		c.s.rejectedInvites.Add(
+			[2]string{c.sipCallID, string(c.tag)},
+			rejectedInviteResponse{status: status, reason: reason},
+		)
+	}
 }
 
 func (c *sipInbound) Address() sip.Uri {

pkg/sip/server.go

@@ -157,6 +157,7 @@ type Server struct {
 	cmu                sync.RWMutex
 	byLocalTag         map[LocalTag]*inboundCall
 	provisionalInvites *expirable.LRU[[2]string, LocalTag]
+	rejectedInvites    *expirable.LRU[[2]string, rejectedInviteResponse]
 
 	infos struct {
 		sync.Mutex
@@ -178,6 +179,11 @@ type inProgressInvite struct {
 	authResolved atomic.Bool
 }
 
+type rejectedInviteResponse struct {
+	status sip.StatusCode
+	reason string
+}
+
 type ServerOption func(s *Server)
 
 func WithGetRoomServer(fn GetRoomFunc) ServerOption {
@@ -208,6 +214,10 @@ func NewServer(region string, conf *config.Config, log logger.Logger, mon *stats
 		byLocalTag:         make(map[LocalTag]*inboundCall),
 		provisionalInvites: expirable.NewLRU[[2]string, LocalTag](maxCallCache, nil, callCacheTTL),
 	}
+	// Initialize the rejected-invite replay cache unless explicitly disabled.
+	if !conf.DisableRejectedInviteCache {
+		s.rejectedInvites = expirable.NewLRU[[2]string, rejectedInviteResponse](maxCallCache, nil, callCacheTTL)
+	}
 	for _, option := range options {
 		option(s)
 	}

pkg/sip/service_test.go

@@ -7,6 +7,7 @@ import (
 	"math/rand"
 	"net/netip"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -242,6 +243,99 @@ func TestService_AuthDrop(t *testing.T) {
 	})
 }
 
+// TestService_RejectedInviteCacheReplay verifies that a second INVITE
+// reusing the same Call-ID and From-tag after a final 4xx response gets
+// the cached response replayed without invoking the auth/dispatch
+// handlers a second time. This guards the dedup that absorbs
+// provider-level retries (same Call-ID + From-tag, new SIP transaction)
+// after we've already sent a terminal rejection.
+func TestService_RejectedInviteCacheReplay(t *testing.T) {
+	const (
+		fromUser = "caller@example.com"
+		toUser   = "callee@example.com"
+		callID   = "rejected-invite-replay-test@example.com"
+		fromTag  = "fixed-from-tag-replay"
+	)
+
+	var authCalls, dispatchCalls atomic.Int32
+
+	h := &TestHandler{
+		GetAuthCredentialsFunc: func(ctx context.Context, call *rpc.SIPCall) (AuthInfo, error) {
+			authCalls.Add(1)
+			return AuthInfo{Result: AuthAccept}, nil
+		},
+		DispatchCallFunc: func(ctx context.Context, info *CallInfo) CallDispatch {
+			dispatchCalls.Add(1)
+			return CallDispatch{Result: DispatchNoRuleReject}
+		},
+		OnSessionEndFunc: func(ctx context.Context, callIdentifier *CallIdentifier, callInfo *livekit.SIPCallInfo, reason string) {
+			// no-op
+		},
+	}
+
+	sipPort := rand.Intn(testPortSIPMax-testPortSIPMin) + testPortSIPMin
+	localIP, err := config.GetLocalIP()
+	require.NoError(t, err)
+	sipServerAddress := fmt.Sprintf("%s:%d", localIP, sipPort)
+
+	mon, err := stats.NewMonitor(&config.Config{MaxCpuUtilization: 0.9})
+	require.NoError(t, err)
+
+	log := logger.LogRLogger(logr.Discard())
+	s, err := NewService("", &config.Config{
+		SIPPort:       sipPort,
+		SIPPortListen: sipPort,
+		RTPPort:       rtcconfig.PortRange{Start: testPortRTPMin, End: testPortRTPMax},
+	}, mon, log, func(projectID string) rpc.IOInfoClient { return nil })
+	require.NoError(t, err)
+	require.NotNil(t, s)
+	s.SetHandler(h)
+	require.NoError(t, s.Start())
+	t.Cleanup(s.Stop)
+
+	ua, err := sipgo.NewUA(sipgo.WithUserAgent(fromUser),
+		sipgo.WithUserAgentLogger(slog.New(logger.ToSlogHandler(s.log))))
+	require.NoError(t, err)
+	client, err := sipgo.NewClient(ua)
+	require.NoError(t, err)
+
+	offer, err := sdp.NewOfferWith(defaultCodecs, localIP, 0xB0B, sdp.EncryptionNone)
+	require.NoError(t, err)
+	offerData, err := offer.SDP.Marshal()
+	require.NoError(t, err)
+
+	sendInvite := func() *sip.Response {
+		recipient := sip.Uri{User: toUser, Host: sipServerAddress}
+		req := sip.NewRequest(sip.INVITE, recipient)
+		req.SetDestination(sipServerAddress)
+		req.SetBody(offerData)
+		req.AppendHeader(sip.NewHeader("Content-Type", "application/sdp"))
+		req.AppendHeader(sip.NewHeader("Call-ID", callID))
+		req.AppendHeader(&sip.FromHeader{
+			DisplayName: fromUser,
+			Address:     sip.Uri{User: fromUser, Host: sipServerAddress},
+			Params:      sip.HeaderParams{{K: "tag", V: fromTag}},
+		})
+		tx, err := client.TransactionRequest(req)
+		require.NoError(t, err)
+		t.Cleanup(tx.Terminate)
+		return getFinalResponseOrFail(t, tx, req)
+	}
+
+	// First INVITE: full handler invocation, 404 from DispatchNoRuleReject.
+	res1 := sendInvite()
+	require.Equal(t, sip.StatusCode(404), res1.StatusCode)
+	require.Equal(t, int32(1), authCalls.Load())
+	require.Equal(t, int32(1), dispatchCalls.Load())
+
+	// Second INVITE with the same Call-ID + From-tag should be served from
+	// the cache: same 404, but handlers must NOT be invoked again.
+	res2 := sendInvite()
+	require.Equal(t, sip.StatusCode(404), res2.StatusCode)
+	require.Equal(t, int32(1), authCalls.Load(), "auth handler must not be re-invoked on replay")
+	require.Equal(t, int32(1), dispatchCalls.Load(), "dispatch handler must not be re-invoked on replay")
+}
+
 func TestService_OnSessionEnd(t *testing.T) {
 	const (
 		expectedCallID    = "test-call-id"