Adding single goroutine to batch-delete invites at intervals

Author: alexlivekit

pkg/sip/inbound.go

@@ -137,6 +137,24 @@ func (s *Server) getCallInfo(id string) *inboundCallInfo {
 	return c
 }
 
+func (s *Server) cleanupInvites() {
+	ticker := time.NewTicker(5 * time.Minute) // Periodic cleanup every 5 minutes
+	defer ticker.Stop()
+	for {
+		select {
+		case <-s.closing.Watch():
+			return
+		case <-ticker.C:
+			s.imu.Lock()
+			for it := s.inviteTimeoutQueue.IterateRemoveAfter(inviteCredentialValidity); it.Next(); {
+				key := it.Item().Value
+				delete(s.inProgressInvites, *key)
+			}
+			s.imu.Unlock()
+		}
+	}
+}
+
 func (s *Server) getInvite(sipCallID, toTag, fromTag string) *inProgressInvite {
 	key := dialogKey{
 		sipCallID: sipCallID,
@@ -149,15 +167,13 @@ func (s *Server) getInvite(sipCallID, toTag, fromTag string) *inProgressInvite {
 	if ok {
 		return is
 	}
-	is = &inProgressInvite{sipCallID: sipCallID}
+	is = &inProgressInvite{
+		sipCallID: sipCallID,
+		expireAt:  time.Now().Add(inviteCredentialValidity),
+	}
 	s.inProgressInvites[key] = is
+	s.inviteTimeoutQueue.Reset(&utils.TimeoutQueueItem[*dialogKey]{Value: &key})
 
-	go func() {
-		time.Sleep(inviteCredentialValidity)
-		s.imu.Lock()
-		defer s.imu.Unlock()
-		delete(s.inProgressInvites, key)
-	}()
 	return is
 }
 
@@ -1403,7 +1419,6 @@ func (s *Server) newInbound(log logger.Logger, contact URI, invite *sip.Request,
 	c := &sipInbound{
 		log:      log,
 		s:        s,
-		id:       "unassigned",
 		invite:   invite,
 		inviteTx: inviteTx,
 		legTr:    legTransportFromReq(invite),

pkg/sip/server.go

@@ -24,6 +24,7 @@ import (
 	"net"
 	"net/netip"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/frostbyte73/core"
@@ -35,6 +36,7 @@ import (
 	"github.com/livekit/protocol/livekit"
 	"github.com/livekit/protocol/logger"
 	"github.com/livekit/protocol/rpc"
+	"github.com/livekit/protocol/utils"
 	"github.com/livekit/protocol/utils/traceid"
 	"github.com/livekit/sipgo"
 	"github.com/livekit/sipgo/sip"
@@ -141,8 +143,10 @@ type Server struct {
 	sipListeners []io.Closer
 	sipUnhandled RequestHandler
 
-	imu               sync.Mutex
-	inProgressInvites map[dialogKey]*inProgressInvite
+	imu                  sync.Mutex
+	inProgressInvites    map[dialogKey]*inProgressInvite
+	inviteTimeoutQueue   utils.TimeoutQueue[*dialogKey]
+	isCleanupTaskRunning atomic.Bool
 
 	closing     core.Fuse
 	cmu         sync.RWMutex
@@ -166,6 +170,7 @@ type inProgressInvite struct {
 	sipCallID string
 	challenge digest.Challenge
 	lkCallID  string // SCL_* LiveKit call ID assigned to this dialog
+	expireAt  time.Time
 }
 
 func NewServer(region string, conf *config.Config, log logger.Logger, mon *stats.Monitor, getIOClient GetIOInfoClient) *Server {
@@ -322,6 +327,9 @@ func (s *Server) Start(agent *sipgo.UserAgent, sc *ServiceConfig, tlsConf *tls.C
 		}
 	}
 
+	// Start the cleanup task
+	go s.cleanupInvites()
+
 	return nil
 }