TEL-886: Self review

pkg/sip/inbound.go

@@ -72,7 +72,9 @@ const (
 var allowHeader = sip.NewHeader("Allow", "INVITE, ACK, CANCEL, BYE, NOTIFY, REFER, MESSAGE, OPTIONS, INFO, SUBSCRIBE")
 
 var errNoACK = errors.New("no ACK received for 200 OK")
-var errInternal = errors.New("internal error")
+
+// RFC 3261 §21.4.27 / §14.2 — glare: INVITE received while an INVITE we sent is in progress.
+const statusRequestPending sip.StatusCode = 491
 
 // hashPassword creates a SHA256 hash of the password for logging purposes
 func hashPassword(password string) string {
@@ -408,7 +410,11 @@ func (s *Server) processInvite(req *sip.Request, tx sip.ServerTransaction) (retE
 		existing.log().Infow("reinvite", "content-length", req.ContentLength(), "cseq", cc.InviteCSeq())
 		if err := existing.updateRemoteFromSDP(sdpBodyFromRequest(req)); err != nil {
 			log.Errorw("failed to update inbound call SDP", err)
-			cc.RejectAsKeepAlive(sip.StatusBadRequest, "Bad Request")
+			if ok := errors.As(err, &SDPError{}); ok {
+				cc.RejectAsKeepAlive(sip.StatusBadRequest, "Bad Request")
+			} else {
+				cc.RejectAsKeepAlive(sip.StatusInternalServerError, "Internal Server Error")
+			}
 			return nil
 		}
 		// TODO(alexfish): Reply with the new SDP.
@@ -421,17 +427,23 @@ func (s *Server) processInvite(req *sip.Request, tx sip.ServerTransaction) (retE
 		if oc != nil && oc.cc != nil && oc.cc.InviteCSeq() < newCSeq {
 			if oc.media == nil {
 				oc.log.Errorw("outbound call media has not been negotiated", nil)
-				return errInternal
+				cc.RejectAsKeepAlive(statusRequestPending, "Request Pending")
+				return nil
 			}
 			localSDP, err := oc.media.GetLocalSDP()
 			if err != nil || len(localSDP) == 0 {
 				oc.log.Errorw("outbound call does not have an SDP", nil)
-				return errInternal
+				cc.RejectAsKeepAlive(statusRequestPending, "Request Pending")
+				return nil
 			}
 			oc.log.Infow("accepting reinvite", "content-length", req.ContentLength(), "cseq", cc.InviteCSeq())
 			if err := oc.updateRemoteFromSDP(sdpBodyFromRequest(req)); err != nil {
 				log.Errorw("failed to update outbound call SDP", err)
-				cc.RejectAsKeepAlive(sip.StatusBadRequest, "Bad Request")
+				if ok := errors.As(err, &SDPError{}); ok {
+					cc.RejectAsKeepAlive(sip.StatusBadRequest, "Bad Request")
+				} else {
+					cc.RejectAsKeepAlive(sip.StatusInternalServerError, "Internal Server Error")
+				}
 				return nil
 			}
 			oc.cc.RecordInvite(newCSeq)
@@ -1023,13 +1035,13 @@ func (c *inboundCall) handleInvite(ctx context.Context, tid traceid.ID, req *sip
 	if pinPrompt {
 		status = CallActive
 	}
-	if err := c.joinRoom(ctx, disp.Room, status); err != nil {
-		return fmt.Errorf("failed joining room: %w", err)
-	}
 	answerData, err = c.negotiateMedia(rawSDP)
 	if err != nil {
 		return rejectMedia(err)
 	}
+	if err := c.joinRoom(ctx, disp.Room, status); err != nil {
+		return fmt.Errorf("failed joining room: %w", err)
+	}
 	// Publish our own track.
 	if err := c.publishTrack(disp.EnabledFeatures, disp.FeatureFlags); err != nil {
 		c.log().Errorw("Cannot publish track", err)
@@ -1107,11 +1119,11 @@ func (c *inboundCall) waitForCallEnd(ctx context.Context, ackReceived <-chan str
 			// Today we seek to enforce all calls to be ACKed or dropped.
 			// Sometimes, though, we do not see ACKs for invites (e.g due to possible
 			// issues with load balancing).
-			// To accomodate this issue, instead of ending the call right here, we instead
+			// To accommodate this issue, instead of ending the call right here, we instead
 			// set an aggressive timeout as a softer fallback.
 			// If the issue really is a dropped ACK, media is expected to flow shortly,
-			// allowing us to accomodate this eventuality. If, however, there is no media
-			// obserrved, the call still ends quickly.
+			// allowing us to accommodate this eventuality. If, however, there is no media
+			// observed, the call still ends quickly.
 			// Once ACKs are certain to be reliable, we will end the call here.
 			c.media.SetTimeout(min(inviteOkAckLateTimeout, c.s.conf.MediaTimeoutInitial), mediaTimeout)
 		}
@@ -1210,7 +1222,7 @@ func (c *inboundCall) negotiateMedia(offerData []byte) ([]byte, error) {
 	c.mon.SDPSize(len(offerData), true)
 	c.log().Debugw("SDP offer", "sdp", string(offerData))
 
-	answerData, err := c.media.GenerateAnswer(offerData, false)
+	answerData, err := c.media.GenerateAnswer(offerData)
 	if err != nil {
 		return nil, err
 	}
@@ -1571,7 +1583,7 @@ func (c *inboundCall) updateRemoteFromSDP(body []byte) error {
 	if mp == nil {
 		return nil
 	}
-	_, err := mp.GenerateAnswer(body, false)
+	_, err := mp.GenerateAnswer(body)
 	return err
 }
 
@@ -1657,8 +1669,7 @@ func (c *inboundCall) publishTrack(features []livekit.SIPFeature, featureFlags m
 		old.Close()
 	}
 	if old := c.media.WriteInboundDTMFTo(c.lkRoom.GetInboundDTMFWriter()); old != nil {
-		c.log().Warnw("media port has unexpected inbound dtmf writer", nil)
-		old.Close()
+		old.Close() // Can be pinDTMFWriter
 	}
 	return nil
 }

pkg/sip/media_pipeline.go

@@ -152,19 +152,15 @@ func (p *mediaPortPipeline) init(
 	return nil
 }
 
-// Construct the Audio and optionally DTM pipleine from SIP RTP to LK PCM, in reverse order.
+// Construct the Audio and optionally DTMF pipeline from SIP RTP to LK PCM, in reverse order.
 func (p *mediaPortPipeline) setupInput(mc *sdp.MediaConfig, audioToRoom msdk.PCM16Writer, dtmfToRoom msdk.WriteCloser[*livekit.SipDTMF]) error {
 	var err error
 	var inboundLatencyEntry atomic.Int64
 	sink := msdk.NopCloser(audioToRoom) // Prevent pipeline close from closing room
 	sink = newLatencyPCMExit(sink, &inboundLatencyEntry, &p.conf.stats.LatencyInE2E)
-
 	codecInfo := mc.Audio.Codec.Info()
 	sink = msdk.ResampleWriter(sink, codecInfo.SampleRate)
-
-	if p.conf.stats != nil {
-		sink = newMediaWriterCount(sink, &p.conf.stats.AudioInFrames, &p.conf.stats.AudioInSamples)
-	}
+	sink = newMediaWriterCount(sink, &p.conf.stats.AudioInFrames, &p.conf.stats.AudioInSamples)
 
 	if p.conf.opts.LogSignalChanges {
 		sink, err = NewSignalLogger(p.conf.log, "input", sink)
@@ -231,8 +227,8 @@ func (p *mediaPortPipeline) handleEventRTP(h *rtp.Header, payload []byte) error
 	})
 }
 
-// Construct the Audio and optionally DTM pipleine from LK PCM to SIP RTP
-// Retuirns the insulated (nopCloser) connectors, and an error.
+// Construct the Audio and optionally DTMF pipeline from LK PCM to SIP RTP
+// Returns the insulated (nopCloser) connectors, and an error.
 func (p *mediaPortPipeline) setupOutput(mc *sdp.MediaConfig, incomingSampleRate int) error {
 	p.rtpLoopWG.Go(p.rtpLoop)
 	w, err := p.sess.OpenWriteStream()
@@ -440,12 +436,12 @@ func (w *dtmfOutWriter) WriteSample(sample *livekit.SipDTMF) error {
 	if len(digits) == 0 {
 		digit := dtmf.CodeToChar(byte(sample.Code))
 		if digit == 0 {
-			return fmt.Errorf("code %d not supoported", sample.Code)
+			return fmt.Errorf("code %d not supported", sample.Code)
 		}
 		digits = string([]byte{digit})
 	} else if sample.Code > 0 {
 		// We can't distinguish between a code0 and no code, but better have something here
-		w.log.Warnw("code payload detected, ignored due to explicit digits", nil, "code", sample.Code, "digits", sample.Digit)
+		w.log.Debugw("code payload detected, ignored due to explicit digits", "code", sample.Code, "digits", sample.Digit)
 	}
 
 	w.mu.Lock()

pkg/sip/media_pipeline_test.go

@@ -123,6 +123,44 @@ func (c *dtmfCollector) snapshot() []*livekit.SipDTMF {
 	return out
 }
 
+// pcmCollector accumulates decoded room audio. The pipeline writes from the RTP
+// read goroutine while the test reads, so every access is guarded.
+type pcmCollector struct {
+	sampleRate int
+
+	mu  sync.Mutex
+	buf msdk.PCM16Sample
+}
+
+func (c *pcmCollector) String() string { return fmt.Sprintf("pcmCollector(%d)", c.sampleRate) }
+
+func (c *pcmCollector) SampleRate() int { return c.sampleRate }
+
+func (c *pcmCollector) Close() error { return nil }
+
+func (c *pcmCollector) WriteSample(sample msdk.PCM16Sample) error {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.buf = append(c.buf, sample...)
+	return nil
+}
+
+func (c *pcmCollector) len() int {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return len(c.buf)
+}
+
+// since returns a copy of everything written after the first n samples.
+func (c *pcmCollector) since(n int) msdk.PCM16Sample {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if n >= len(c.buf) {
+		return nil
+	}
+	return slices.Clone(c.buf[n:])
+}
+
 // pipelineHarness is the durable side of a mediaPort: UDP pipe, pipeline config,
 // buffer anchors, and a synthesized MediaConfig. The pipeline itself is swapped
 // on configure / reconfigure.
@@ -136,7 +174,7 @@ type pipelineHarness struct {
 	audioOut    *msdk.WriteCloserSwitch[msdk.PCM16Sample]
 	dtmfIn      *msdk.WriteCloserSwitch[*livekit.SipDTMF]
 	dtmfOut     *msdk.WriteCloserSwitch[*livekit.SipDTMF]
-	roomAudio   *msdk.PCM16Sample
+	roomAudio   *pcmCollector
 	roomDTMF    *dtmfCollector
 	pipeline    *mediaPortPipeline
 	ssrcCount   atomic.Uint64
@@ -159,10 +197,10 @@ func newPipelineHarness(t *testing.T, sampleRate int) *pipelineHarness {
 		audioOut:  msdk.NewWriteCloserSwitch[msdk.PCM16Sample](sampleRate),
 		dtmfIn:    msdk.NewWriteCloserSwitch[*livekit.SipDTMF](dtmf.SampleRate),
 		dtmfOut:   msdk.NewWriteCloserSwitch[*livekit.SipDTMF](dtmf.SampleRate),
-		roomAudio: new(msdk.PCM16Sample),
+		roomAudio: &pcmCollector{sampleRate: sampleRate},
 		roomDTMF:  &dtmfCollector{},
 	}
-	h.audioIn.Swap(msdk.NewPCM16BufferWriter(h.roomAudio, sampleRate))
+	h.audioIn.Swap(h.roomAudio)
 	h.dtmfIn.Swap(h.roomDTMF)
 	h.conf = &MediaPortPipelineConfig{
 		log:   log,
@@ -348,7 +386,7 @@ func (h *pipelineHarness) testAudioFromRoom(t *testing.T) {
 }
 
 func (h *pipelineHarness) testAudioFromPort(t *testing.T) {
-	before := len(*h.roomAudio)
+	before := h.roomAudio.len()
 	packetsBefore := h.packetCount.Load()
 	clock := h.codec.Info().RTPClockRate
 	if clock == 0 {
@@ -363,15 +401,15 @@ func (h *pipelineHarness) testAudioFromPort(t *testing.T) {
 		return h.packetCount.Load() >= packetsBefore+5
 	}, time.Second, 5*time.Millisecond, "RTP should be accepted")
 	require.Eventually(t, func() bool {
-		return len(*h.roomAudio) > before
+		return h.roomAudio.len() > before
 	}, time.Second, 5*time.Millisecond, "decoded PCM should reach room (packets=%d input=%d failed=%d ignored=%d room=%d)",
 		h.packetCount.Load(),
 		h.pipeline.conf.stats.InputPackets.Load(),
 		h.pipeline.conf.stats.FailedPackets.Load(),
 		h.pipeline.conf.stats.IgnoredPackets.Load(),
-		len(*h.roomAudio),
+		h.roomAudio.len(),
 	)
-	require.Greater(t, pcmEnergy((*h.roomAudio)[before:]), int64(0), "decoded room audio should carry energy")
+	require.Greater(t, pcmEnergy(h.roomAudio.since(before)), int64(0), "decoded room audio should carry energy")
 }
 
 func (h *pipelineHarness) testDTMFFromRoom(t *testing.T) {