Fixing up TransferSipParticipant when receiving BYE after REFER

pkg/sip/inbound.go

@@ -1437,7 +1437,7 @@ func (c *inboundCall) transferCall(ctx context.Context, transferTo string, heade
 		}()
 	}
 
-	err = c.cc.TransferCall(ctx, transferTo, headers)
+	err = c.cc.TransferCall(ctx, transferTo, headers, c.ctx.Done())
 	if err != nil {
 		c.log().Infow("inbound call failed to transfer", "error", err, "transferTo", transferTo)
 		return err
@@ -1974,7 +1974,7 @@ func (c *sipInbound) newReferReq(transferTo string, headers map[string]string) (
 	return req, nil
 }
 
-func (c *sipInbound) TransferCall(ctx context.Context, transferTo string, headers map[string]string) error {
+func (c *sipInbound) TransferCall(ctx context.Context, transferTo string, headers map[string]string, callDone <-chan struct{}) error {
 	req, err := c.newReferReq(transferTo, headers)
 	if err != nil {
 		return err
@@ -1988,6 +1988,10 @@ func (c *sipInbound) TransferCall(ctx context.Context, transferTo string, header
 	select {
 	case <-ctx.Done():
 		return psrpc.NewErrorf(psrpc.Canceled, "refer canceled")
+	case <-callDone:
+		// At this point, REFER was accepted, but we received a BYE, nothing to do, also not an error
+		c.log.Infow("refer canceled by BYE from remote")
+		return nil
 	case err := <-c.referDone:
 		if err != nil {
 			return err

pkg/sip/outbound.go

@@ -722,7 +722,7 @@ func (c *outboundCall) transferCall(ctx context.Context, transferTo string, head
 		}()
 	}
 
-	err = c.cc.transferCall(ctx, transferTo, headers)
+	err = c.cc.transferCall(ctx, transferTo, headers, c.closing.Watch())
 	if err != nil {
 		c.log.Infow("outbound call failed to transfer", "error", err, "transferTo", transferTo)
 		return err
@@ -1140,7 +1140,7 @@ func (c *sipOutbound) Drop() {
 	c.drop()
 }
 
-func (c *sipOutbound) transferCall(ctx context.Context, transferTo string, headers map[string]string) error {
+func (c *sipOutbound) transferCall(ctx context.Context, transferTo string, headers map[string]string, callDone <-chan struct{}) error {
 	c.mu.Lock()
 
 	if c.invite == nil || c.inviteOk == nil {
@@ -1176,6 +1176,10 @@ func (c *sipOutbound) transferCall(ctx context.Context, transferTo string, heade
 	select {
 	case <-ctx.Done():
 		return psrpc.NewErrorf(psrpc.Canceled, "refer canceled")
+	case <-callDone:
+		// At this point, REFER was accepted, but we received a BYE, nothing to do, also not an error
+		c.log.Infow("refer canceled by BYE from remote")
+		return nil
 	case err := <-c.referDone:
 		if err != nil {
 			return err

pkg/sip/service.go

@@ -27,7 +27,6 @@ import (
 	"slices"
 	"strings"
 	"sync"
-	"sync/atomic"
 	"time"
 
 	"github.com/livekit/sipgo/transport"
@@ -47,6 +46,12 @@ import (
 	"github.com/livekit/sip/version"
 )
 
+type PendingTransfer struct {
+	CallID     string
+	TransferTo string
+	Done       chan error
+}
+
 type ServiceConfig struct {
 	SignalingIP      netip.Addr
 	SignalingIPLocal netip.Addr
@@ -63,12 +68,7 @@ type Service struct {
 	closers []io.Closer
 
 	mu               sync.Mutex
-	pendingTransfers map[transferKey]chan struct{}
-}
-
-type transferKey struct {
-	SipCallId  string
-	TransferTo string
+	pendingTransfers map[LocalTag]*PendingTransfer
 }
 
 type GetIOInfoClient func(projectID string) rpc.IOInfoClient
@@ -93,7 +93,7 @@ func NewService(region string, conf *config.Config, mon *stats.Monitor, log logg
 		mon:              mon,
 		cli:              cli,
 		srv:              NewServer(region, conf, log, mon, getIOClient, WithClient(cli)),
-		pendingTransfers: make(map[transferKey]chan struct{}),
+		pendingTransfers: make(map[LocalTag]*PendingTransfer),
 	}
 	var err error
 	s.sconf, err = GetServiceConfig(s.conf)
@@ -340,52 +340,64 @@ func (s *Service) transferSIPParticipant(ctx context.Context, req *rpc.InternalT
 		return &emptypb.Empty{}, err
 	}
 
-	var transferResult atomic.Pointer[error]
-
-	s.mu.Lock()
-	k := transferKey{
-		SipCallId:  req.SipCallId,
-		TransferTo: req.TransferTo,
-	}
-	done, ok := s.pendingTransfers[k]
-	if !ok {
-		done = make(chan struct{})
-		s.pendingTransfers[k] = done
-
+	pending, isNew := s.getOrCreatePendingTransfer(req.SipCallId, req.TransferTo)
+	if !isNew {
+		if pending.TransferTo != req.TransferTo {
+			return &emptypb.Empty{}, psrpc.NewErrorf(psrpc.InvalidArgument, "call already being transferred elsewhere")
+		}
+		// Already transferred, resume wait
+		s.log.Debugw("repeated request for call transfer", "callID", req.SipCallId, "transferTo", req.TransferTo)
+		// TODO: Maybe just bump the psrpc timeout? It gets auto retried anyway internally.
+	} else {
+		// Initial transfer request for this call
 		timeout := req.RingingTimeout.AsDuration()
 		if timeout <= 0 {
-			timeout = 80 * time.Second
+			timeout = 120 * time.Second
 		}
 
 		go func() {
 			ctx, cdone := context.WithTimeout(context.WithoutCancel(ctx), timeout)
 			defer cdone()
 
 			err := s.processParticipantTransfer(ctx, req.SipCallId, req.TransferTo, req.Headers, req.PlayDialtone)
-			transferResult.Store(&err)
-			close(done)
+			select {
+			case pending.Done <- err:
+			default:
+				s.log.Errorw("pending transfer channel is full, dropping error", err, "callID", req.SipCallId, "transferTo", req.TransferTo)
+			}
+			close(pending.Done)
 
 			s.mu.Lock()
-			delete(s.pendingTransfers, k)
+			delete(s.pendingTransfers, LocalTag(req.SipCallId))
 			s.mu.Unlock()
 		}()
-	} else {
-		s.log.Debugw("repeated request for call transfer", "callID", req.SipCallId, "transferTo", req.TransferTo)
 	}
-	s.mu.Unlock()
 
 	select {
-	case <-done:
-		var err error
-		errPtr := transferResult.Load()
-		if errPtr != nil {
-			err = *errPtr
-		}
+	case err := <-pending.Done:
 		return &emptypb.Empty{}, err
 	case <-ctx.Done():
 		return &emptypb.Empty{}, psrpc.NewError(psrpc.Canceled, ctx.Err())
 	}
 }
+func (s *Service) getOrCreatePendingTransfer(callID string, transferTo string) (*PendingTransfer, bool) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	keyCall := LocalTag(callID)
+	pending, ok := s.pendingTransfers[keyCall]
+	if ok {
+		return pending, false
+	}
+
+	pending = &PendingTransfer{
+		CallID:     callID,
+		TransferTo: transferTo,
+		Done:       make(chan error, 1),
+	}
+	s.pendingTransfers[keyCall] = pending
+	return pending, true
+}
 
 func (s *Service) processParticipantTransfer(ctx context.Context, callID string, transferTo string, headers map[string]string, dialtone bool) error {
 	// Look for call both in client (outbound) and server (inbound)