TEL-886: Fix tests and other misc

pkg/sip/inbound.go

@@ -697,36 +697,35 @@ func (s *Server) onNotify(log *slog.Logger, req *sip.Request, tx sip.ServerTrans
 }
 
 type inboundCall struct {
-	s                *Server
-	tid              traceid.ID
-	logPtr           atomic.Pointer[logger.Logger]
-	cc               *sipInbound
-	mon              *stats.CallMonitor
-	state            *CallState
-	callStart        time.Time
-	extraAttrs       map[string]string
-	attrsToHdr       map[string]string
-	ctx              context.Context
-	cancel           func()
-	closeReason      atomic.Pointer[ReasonHeader]
-	call             *rpc.SIPCall
-	mmu              sync.Mutex
-	media            MediaPort
-	mediaCodecs      *msdk.CodecSet
-	dtmf             chan dtmf.Event // buffered
-	endCall          chan EndCall    // buffered
-	lkRoom           RoomInterface   // LiveKit room; only active after correct pin is entered
-	callDur          func() time.Duration
-	joinDur          func() time.Duration
-	forwardDTMF      atomic.Bool
-	done             atomic.Bool
-	started          core.Fuse
-	stats            Stats
-	sigTs            SignalingTimestamps
-	jitterBuf        bool
-	projectID        string
-	audioOut         *msdk.WriteCloserSwitch[msdk.PCM16Sample] // inner writer owned by MediaPort
-	audioInProcessor msdk.PCM16Processor
+	s           *Server
+	tid         traceid.ID
+	logPtr      atomic.Pointer[logger.Logger]
+	cc          *sipInbound
+	mon         *stats.CallMonitor
+	state       *CallState
+	callStart   time.Time
+	extraAttrs  map[string]string
+	attrsToHdr  map[string]string
+	ctx         context.Context
+	cancel      func()
+	closeReason atomic.Pointer[ReasonHeader]
+	call        *rpc.SIPCall
+	mmu         sync.Mutex
+	media       MediaPort
+	mediaCodecs *msdk.CodecSet
+	dtmf        chan dtmf.Event // buffered
+	endCall     chan EndCall    // buffered
+	lkRoom      RoomInterface   // LiveKit room; only active after correct pin is entered
+	callDur     func() time.Duration
+	joinDur     func() time.Duration
+	forwardDTMF atomic.Bool
+	done        atomic.Bool
+	started     core.Fuse
+	stats       Stats
+	sigTs       SignalingTimestamps
+	jitterBuf   bool
+	projectID   string
+	audioOut    *msdk.WriteCloserSwitch[msdk.PCM16Sample] // inner writer owned by MediaPort
 }
 
 func (s *Server) newInboundCall(
@@ -911,7 +910,7 @@ func (c *inboundCall) handleInvite(ctx context.Context, tid traceid.ID, req *sip
 		}
 		rawSDP := req.Body()
 		tmedia := c.mon.StageDurTimer("start-media")
-		answerData, err := c.runMediaConn(tid, rawSDP, m, conf, disp.EnabledFeatures, disp.FeatureFlags)
+		answerData, err := c.runMediaConn(tid, rawSDP, m, conf, disp.FeatureFlags)
 		tmedia()
 		if err != nil {
 			sipReason := sip.StatusInternalServerError
@@ -1036,7 +1035,7 @@ func (c *inboundCall) handleInvite(ctx context.Context, tid traceid.ID, req *sip
 		return fmt.Errorf("failed joining room: %w", err)
 	}
 	// Publish our own track.
-	if err := c.publishTrack(); err != nil {
+	if err := c.publishTrack(disp.EnabledFeatures, disp.FeatureFlags); err != nil {
 		c.log().Errorw("Cannot publish track", err)
 		c.closeWithTerm(ctx, stats.ServerError("publish-failed"))
 		return fmt.Errorf("publishing track to room failed: %w", err)
@@ -1136,7 +1135,7 @@ func (w *dtmfEventWriter) WriteSample(sample *livekit.SipDTMF) error {
 	return nil
 }
 
-func (c *inboundCall) runMediaConn(tid traceid.ID, offerData []byte, mconf *sipMediaConfig, conf *config.Config, features []livekit.SIPFeature, featureFlags map[string]string) ([]byte, error) {
+func (c *inboundCall) runMediaConn(tid traceid.ID, offerData []byte, mconf *sipMediaConfig, conf *config.Config, featureFlags map[string]string) ([]byte, error) {
 	c.mmu.Lock()
 	defer c.mmu.Unlock()
 	c.mon.SDPSize(len(offerData), true)
@@ -1174,7 +1173,6 @@ func (c *inboundCall) runMediaConn(tid traceid.ID, offerData []byte, mconf *sipM
 	c.mon.SDPSize(len(answerData), false)
 	c.log().Debugw("SDP answer", "sdp", string(answerData))
 
-	c.audioInProcessor = c.s.handler.GetMediaProcessor(features, featureFlags, string(c.cc.ID()), MediaProcessorOpts{InputSampleRate: RoomSampleRate})
 	mp.WriteDTMFTo(&dtmfEventWriter{handler: c.handleDTMF})
 
 	// Must be set earlier to send the pin prompts.
@@ -1192,6 +1190,7 @@ func (c *inboundCall) runMediaConn(tid traceid.ID, offerData []byte, mconf *sipM
 	})
 	return answerData, nil
 }
+
 func (c *inboundCall) waitMedia(ctx context.Context) (bool, error) {
 	defer c.mon.StageDurTimer("wait-media")()
 	ctx, span := Tracer.Start(ctx, "sip.inbound.waitMedia")
@@ -1606,16 +1605,16 @@ func (c *inboundCall) createLiveKitParticipant(ctx context.Context, rconf RoomCo
 	return nil
 }
 
-func (c *inboundCall) publishTrack() error {
+func (c *inboundCall) publishTrack(features []livekit.SIPFeature, featureFlags map[string]string) error {
 	defer c.mon.StageDurTimer("track-publish")()
 	local, err := c.lkRoom.NewParticipantTrack(RoomSampleRate)
 	if err != nil {
 		_ = c.lkRoom.Close()
 		return err
 	}
 
-	if c.audioInProcessor != nil {
-		local = c.audioInProcessor(local)
+	if audioInProcessor := c.s.handler.GetMediaProcessor(features, featureFlags, string(c.cc.ID()), MediaProcessorOpts{InputSampleRate: RoomSampleRate}); audioInProcessor != nil {
+		local = audioInProcessor(local)
 	}
 	c.media.WriteAudioTo(local)
 	return nil

pkg/sip/media_pipeline.go

@@ -129,9 +129,6 @@ func (p *mediaPortPipeline) init(
 	if p.conf.opts.IgnoreLocalAddrInSDP && mc.Remote.Addr().IsPrivate() {
 		port.SetSymmetric(true) // Already initialized with opts, turn on for edge case
 	}
-
-	// stopDiscarding() must be done ahead of SRTP session creation, or risk dead read without read deadlines
-	port.stopDiscarding()
 	p.lastDTMFTimestamp.Store(math.MaxUint32)
 
 	var err error