TEL-886: Fix SSRC race to codec

pkg/sip/media_pipeline.go

@@ -176,6 +176,8 @@ func (p *mediaPortPipeline) setupInput(mc *sdp.MediaConfig, audioToRoom msdk.PCM
 	// And these are only available before decoding, hence it wraps both audioHandler & sink
 	audioHandler = newSilenceFiller(audioHandler, sink, codecInfo.RTPClockRate, codecInfo.SampleRate, p.conf.log)
 
+	audioHandler = NewSerializedRTPHandler(audioHandler) // SilenceFiller/Codecs are not thread-safe
+
 	mux := rtp.NewMux(nil)
 	mux.SetDefault(newRTPStatsHandler(p.conf.mon, "", nil))
 
@@ -405,6 +407,31 @@ func (p *mediaPortPipeline) Close() error {
 	return errors.Join(errs...)
 }
 
+func NewSerializedRTPHandler(w rtp.HandlerCloser) rtp.HandlerCloser {
+	return &serializedRTPHandler{w: w}
+}
+
+type serializedRTPHandler struct {
+	mu sync.Mutex
+	w  rtp.HandlerCloser
+}
+
+func (s *serializedRTPHandler) String() string {
+	return s.w.String()
+}
+
+func (s *serializedRTPHandler) HandleRTP(h *rtp.Header, payload []byte) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.w.HandleRTP(h, payload)
+}
+
+func (s *serializedRTPHandler) Close() {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.w.Close()
+}
+
 // dtmfOutWriter sends SipDTMF as RFC 4733 telephone-events (optional in-band audio).
 type dtmfOutWriter struct {
 	log logger.Logger

pkg/sip/media_pipeline_test.go

@@ -528,9 +528,10 @@ func TestMediaPipelineTeardownMultiSSRC(t *testing.T) {
 	h.injectAudio(0x22222222, 1, 160, sample)
 
 	require.Eventually(t, func() bool {
-		return h.ssrcCount.Load() >= 2
-	}, time.Second, 5*time.Millisecond, "expected AcceptStream for two SSRCs")
-	assert.Equal(t, h.packetCount.Load(), uint64(2))
+		return h.ssrcCount.Load() >= 2 && h.packetCount.Load() >= 2
+	}, time.Second, 5*time.Millisecond, "expected AcceptStream and HandleRTP for two SSRCs")
+	assert.Equal(t, uint64(2), h.ssrcCount.Load())
+	assert.Equal(t, uint64(2), h.packetCount.Load())
 
 	done := make(chan error, 1)
 	go func() {
@@ -544,6 +545,57 @@ func TestMediaPipelineTeardownMultiSSRC(t *testing.T) {
 	}
 }
 
+func TestMediaPipelineConcurrentSSRCPump(t *testing.T) {
+	const (
+		ssrcCount = 3
+		packets   = 30 // Currently the built-in limit of media-sdk's ssrc mux
+	)
+	codec := audioCodecByName(t, g711.ULawSDPNameAndRate)
+	h := newPipelineHarness(t, RoomSampleRate)
+	h.configure(codec, testAudioPT(codec), testDTMFPT, false)
+
+	silence := make(msdk.PCM16Sample, codec.Info().SampleRate/int(time.Second/msrtp.DefFrameDur))
+	var encoded msrtp.Buffer
+	clock := codec.Info().RTPClockRate
+	if clock == 0 {
+		clock = codec.Info().SampleRate
+	}
+	enc := msrtp.EncodePCM(msrtp.NewSeqWriter(&encoded).NewStream(h.audioPT, clock), h.codec)
+	require.NoError(t, enc.WriteSample(silence))
+	require.NoError(t, enc.Close())
+	require.NotEmpty(t, encoded, "codec produced no RTP")
+	payload := slices.Clone(encoded[0].Payload)
+
+	pkt := &rtp.Packet{
+		Header: rtp.Header{
+			Version:     2,
+			PayloadType: h.audioPT,
+		},
+		Payload: payload,
+	}
+	for i := range packets {
+		pkt.SequenceNumber = uint16(i)
+		pkt.SSRC = uint32(i % ssrcCount)
+		h.injectRTP(pkt)
+	}
+
+	require.Eventually(t, func() bool { return h.ssrcCount.Load() == ssrcCount }, time.Second, time.Millisecond, "expected %d SSRCs", ssrcCount)
+	assert.Eventually(t, func() bool { return h.packetCount.Load() == packets }, time.Second, time.Millisecond, "expected %d packets", packets)
+	assert.Equal(t, uint64(ssrcCount), h.ssrcCount.Load())
+	assert.Equal(t, uint64(packets), h.packetCount.Load())
+
+	done := make(chan error, 1)
+	go func() {
+		done <- h.pipeline.Close()
+	}()
+	select {
+	case err := <-done:
+		require.NoError(t, err)
+	case <-time.After(2 * time.Second):
+		t.Fatal("pipeline.Close hung under concurrent SSRC pumps")
+	}
+}
+
 func TestMediaPipelineReuseUDPConn(t *testing.T) {
 	const rate = 48000
 	d := pipelineTestDTMF[1] // event-only