fix: SDP correctness, RTP goroutine race condition, and non-RFC SRC compatibility

pkg/media/rtp.go

@@ -196,11 +196,19 @@ func StartRTPForwarding(ctx context.Context, forwarder *RTPForwarder, callUUID s
 		defer rtpSpan.End()
 		// Use original ctx for cancellation - don't overwrite with tracing context!
 
-		// Log when goroutine exits (before cleanup)
+		// Defer execution order (LIFO – last declared runs first):
+		//   4th (runs last):  log "exited"
+		//   3rd:              recover + Cleanup  (no-op if stopAndCleanup already ran it)
+		//   2nd:              close(doneChan)    ← unblocks WaitDone in BYE handler
+		//   1st (runs first): WAV Finalize       ← goroutine stopped writing, header safe to write
+
+		// Log when goroutine exits (declared 1st → runs last)
 		defer func() {
 			forwarder.Logger.WithField("call_uuid", callUUID).Info("Main RTP goroutine exited (defer)")
 		}()
 
+		// Goroutine-owned cleanup; will be a no-op if BYE handler already called Cleanup().
+		// recover() here catches any panic from the WAV Finalize or doneChan defers above.
 		defer func() {
 			if r := recover(); r != nil {
 				forwarder.Logger.WithFields(logrus.Fields{
@@ -213,8 +221,13 @@ func StartRTPForwarding(ctx context.Context, forwarder *RTPForwarder, callUUID s
 			forwarder.Cleanup()
 		}()
 
-		// Finalize WAV before Cleanup so the header is updated.
-		// Ensures recordings are playable if the goroutine exits for any reason.
+		// Signal WaitDone() after WAV Finalize but before Cleanup/Azure-upload.
+		// The BYE handler waits here before calling its own Cleanup(). By the time
+		// this fires the goroutine has stopped writing, so no file-closed race.
+		defer forwarder.doneOnce.Do(func() { close(forwarder.doneChan) })
+
+		// Finalize WAV before everything else so the header is always written.
+		// Declared last → runs first in LIFO order.
 		defer func() {
 			if forwarder.WAVWriter != nil {
 				if err := forwarder.WAVWriter.Finalize(); err != nil && forwarder.Logger != nil {
@@ -263,6 +276,16 @@ func StartRTPForwarding(ctx context.Context, forwarder *RTPForwarder, callUUID s
 
 		SetUDPSocketBuffers(udpConn, forwarder.Logger)
 
+		// Send an initial probe to trigger symmetric-RTP / RTP-latching on the SRC.
+		// The SRC (e.g. Cognigy/jambonz) may not start sending until it sees the
+		// first UDP datagram arriving from our advertised RTP endpoint.
+		forwarder.remoteMutex.Lock()
+		probeRTPAddr := forwarder.RemoteRTPAddr
+		forwarder.remoteMutex.Unlock()
+		if probeRTPAddr != nil {
+			sendRTPProbe(udpConn, probeRTPAddr, forwarder.LocalSSRC, forwarder.Logger)
+		}
+
 		var rtcpConn *net.UDPConn
 		if !forwarder.UseRTCPMux && forwarder.RTCPPort > 0 {
 			rtcpAddr := &net.UDPAddr{Port: forwarder.RTCPPort}
@@ -1539,6 +1562,41 @@ func sendRTCPPackets(forwarder *RTPForwarder, packets ...rtcp.Packet) error {
 	return err
 }
 
+// sendRTPProbe sends a single minimal RTP packet from the local socket to the
+// remote RTP address. This is used to "unlatch" symmetric-RTP implementations
+// that wait for the first packet before starting to send.
+func sendRTPProbe(conn *net.UDPConn, remoteAddr *net.UDPAddr, localSSRC uint32, logger *logrus.Logger) {
+	if conn == nil || remoteAddr == nil {
+		return
+	}
+	pkt := &rtp.Packet{
+		Header: rtp.Header{
+			Version:        2,
+			PayloadType:    13, // CN (Comfort Noise) – safe probe payload type
+			SequenceNumber: 0,
+			Timestamp:      0,
+			SSRC:           localSSRC,
+		},
+		Payload: []byte{0}, // 1-byte minimal payload
+	}
+	raw, err := pkt.Marshal()
+	if err != nil {
+		return
+	}
+	if _, err = conn.WriteTo(raw, remoteAddr); err != nil {
+		if logger != nil {
+			logger.WithError(err).Debug("RTP probe send failed")
+		}
+		return
+	}
+	if logger != nil {
+		logger.WithFields(logrus.Fields{
+			"remote": remoteAddr.String(),
+			"ssrc":   localSSRC,
+		}).Info("Sent initial RTP probe (symmetric-RTP trigger)")
+	}
+}
+
 func sendRTCPBye(forwarder *RTPForwarder) {
 	if forwarder == nil {
 		return

pkg/media/sdp_utils.go

@@ -2,6 +2,7 @@ package media
 
 import (
 	"encoding/base64"
+	"net"
 	"strconv"
 	"strings"
 
@@ -17,6 +18,51 @@ func ConfigureForwarderForMediaDescription(forwarder *RTPForwarder, sdpDesc *sdp
 
 	applySessionAttributes(forwarder, sdpDesc, logger)
 	applyMediaAttributes(forwarder, md, logger)
+	setRemoteAddressHints(forwarder, sdpDesc, md, logger)
+}
+
+// setRemoteAddressHints pre-populates the forwarder's remote addresses from the SDP offer.
+// This enables the initial RTP probe for symmetric-RTP / RTP-latching implementations.
+func setRemoteAddressHints(forwarder *RTPForwarder, sdpDesc *sdp.SessionDescription, md *sdp.MediaDescription, logger *logrus.Logger) {
+	connAddr := ""
+	if md.ConnectionInformation != nil && md.ConnectionInformation.Address != nil {
+		connAddr = md.ConnectionInformation.Address.Address
+	} else if sdpDesc.ConnectionInformation != nil && sdpDesc.ConnectionInformation.Address != nil {
+		connAddr = sdpDesc.ConnectionInformation.Address.Address
+	}
+	if connAddr == "" {
+		return
+	}
+	ip := net.ParseIP(connAddr)
+	if ip == nil {
+		return
+	}
+
+	rtpPort := md.MediaName.Port.Value
+	rtcpPort := rtpPort + 1
+	for _, attr := range md.Attributes {
+		if attr.Key == "rtcp" {
+			fields := strings.Fields(attr.Value)
+			if len(fields) > 0 {
+				if p, err := strconv.Atoi(fields[0]); err == nil {
+					rtcpPort = p
+				}
+			}
+			break
+		}
+	}
+
+	forwarder.remoteMutex.Lock()
+	forwarder.RemoteRTPAddr = &net.UDPAddr{IP: ip, Port: rtpPort}
+	forwarder.RemoteRTCPAddr = &net.UDPAddr{IP: ip, Port: rtcpPort}
+	forwarder.remoteMutex.Unlock()
+
+	if logger != nil {
+		logger.WithFields(logrus.Fields{
+			"remote_rtp":  net.JoinHostPort(connAddr, strconv.Itoa(rtpPort)),
+			"remote_rtcp": net.JoinHostPort(connAddr, strconv.Itoa(rtcpPort)),
+		}).Info("Pre-set remote RTP/RTCP address from SDP offer (symmetric-RTP probe target)")
+	}
 }
 
 func applySessionAttributes(forwarder *RTPForwarder, sdpDesc *sdp.SessionDescription, logger *logrus.Logger) {

pkg/media/types.go

@@ -104,6 +104,12 @@ type RTPForwarder struct {
 	rtcpStopChan chan struct{}
 	remoteMutex  sync.Mutex
 
+	// doneChan is closed when the StartRTPForwarding goroutine has fully exited.
+	// Callers that need to ensure the goroutine is done before calling Cleanup()
+	// can select/wait on this channel (see WaitDone).
+	doneChan chan struct{}
+	doneOnce sync.Once
+
 	// Start-time alignment for WAV combining (Fix G)
 	FirstRTPTimestamp uint32    // RTP timestamp of the first packet
 	FirstRTPWallClock time.Time // Wall-clock time when first RTP packet arrived
@@ -230,9 +236,21 @@ func NewRTPForwarder(timeout time.Duration, recordingSession *siprec.RecordingSe
 		LocalSSRC:         generateRandomSSRC(),
 		RTPStats:          newRTPStreamStats(),
 		rtcpStopChan:      make(chan struct{}, 1),
+		doneChan:          make(chan struct{}),
 	}, nil
 }
 
+// WaitDone blocks until the StartRTPForwarding goroutine has fully exited or
+// the timeout elapses. Returns true if the goroutine exited, false on timeout.
+func (f *RTPForwarder) WaitDone(timeout time.Duration) bool {
+	select {
+	case <-f.doneChan:
+		return true
+	case <-time.After(timeout):
+		return false
+	}
+}
+
 // SetCodecInfo configures payload format information used for recording.
 // This method is thread-safe and can be called while RTP processing is active.
 func (f *RTPForwarder) SetCodecInfo(payloadType byte, codecName string, sampleRate, channels int) {

pkg/sip/custom_server.go

@@ -11,6 +11,7 @@ import (
 	"mime"
 	"mime/multipart"
 	"net"
+	"os"
 	"path/filepath"
 	"strconv"
 	"strings"
@@ -1659,6 +1660,7 @@ func (s *CustomSIPServer) handleSiprecInvite(message *SIPMessage) {
 		s.sendResponse(message, 488, "Not Acceptable Here", nil, nil)
 		return
 	}
+	logger.WithField("sdp_offer", string(sdpData)).Info("Received SDP offer from SRC")
 
 	// Create RTP forwarders for this SIPREC call
 	var responseSDP []byte
@@ -1841,6 +1843,17 @@ func (s *CustomSIPServer) handleSiprecInvite(message *SIPMessage) {
 			}
 		}
 
+		// Store the actual SDP stream labels so the RS-metadata response uses matching labels
+		if recordingSession != nil {
+			for idx := range forwarders {
+				label := audioStreams[idx].label
+				if label == "" {
+					label = fmt.Sprintf("%d", idx+1)
+				}
+				recordingSession.SDPStreamLabels = append(recordingSession.SDPStreamLabels, label)
+			}
+		}
+
 		for idx, forwarder := range forwarders {
 			streamID := audioStreams[idx].label
 			if streamID == "" {
@@ -1876,6 +1889,7 @@ func (s *CustomSIPServer) handleSiprecInvite(message *SIPMessage) {
 			}
 			responseSDP = responseSDPBytes
 			logger.WithField("response_sdp_size", len(responseSDP)).Debug("Generated multi-stream SDP response")
+			logger.WithField("sdp_answer", string(responseSDP)).Info("SDP answer sent to SRC")
 		} else {
 			logger.Warn("Using single-stream SDP response (receivedSDP is nil)")
 			sdpResponse := s.handler.generateSDPResponseWithPort(nil, mediaIP, forwarders[0].LocalPort, forwarders[0])
@@ -1933,8 +1947,10 @@ func (s *CustomSIPServer) handleSiprecInvite(message *SIPMessage) {
 		"Accept":    "application/sdp, application/rs-metadata+xml, multipart/mixed",
 	}
 
-	// If we have a recording session, generate proper SIPREC response with metadata
-	if recordingSession != nil {
+	// If we have a recording session, generate proper SIPREC response with metadata.
+	// PLAIN_SDP_RESPONSE=true sends only application/sdp (debug/compat mode for jambonz).
+	plainSDPMode := os.Getenv("PLAIN_SDP_RESPONSE") == "true"
+	if recordingSession != nil && !plainSDPMode {
 		contentType, multipartBody, err := s.generateSiprecResponse(responseSDP, recordingSession, logger)
 		if err != nil {
 			inviteErr = err
@@ -1973,7 +1989,11 @@ func (s *CustomSIPServer) handleSiprecInvite(message *SIPMessage) {
 			"transport": transport,
 		})
 	} else {
-		// Regular response without SIPREC metadata
+		// Plain SDP response – either non-SIPREC call or PLAIN_SDP_RESPONSE=true debug mode.
+		if plainSDPMode && recordingSession != nil {
+			logger.Warn("PLAIN_SDP_RESPONSE mode: sending application/sdp only (no rs-metadata) for SIPREC session")
+		}
+		responseHeaders["Content-Type"] = "application/sdp"
 		callState.State = "awaiting_ack"
 		callState.PendingAckCSeq = callState.RemoteCSeq
 		callState.LastActivity = time.Now()
@@ -2319,6 +2339,7 @@ func (s *CustomSIPServer) handleSiprecReInvite(message *SIPMessage, callState *C
 					continue
 				}
 				extra.Stop()
+				extra.WaitDone(100 * time.Millisecond)
 				extra.Cleanup()
 			}
 			forwarders = forwarders[:len(audioStreams)]
@@ -3639,6 +3660,18 @@ func (s *CustomSIPServer) doFinalizeCall(callID string, callState *CallState, re
 		"single_forwarder":  callState.RTPForwarder != nil,
 	}).Debug("Starting forwarder cleanup in finalizeCall")
 
+	// stopAndCleanup signals a forwarder to stop, waits up to 100 ms for its
+	// goroutine to exit (so the WAV file is no longer being written), then
+	// runs Cleanup. The wait prevents the "file already closed" race where
+	// Cleanup closes the WAV file while the goroutine is still in a Write call.
+	stopAndCleanup := func(forwarder *media.RTPForwarder) {
+		forwarder.Stop()
+		if !forwarder.WaitDone(100 * time.Millisecond) {
+			s.logger.WithField("call_id", callID).Warn("RTP goroutine did not exit within timeout; proceeding with cleanup")
+		}
+		forwarder.Cleanup()
+	}
+
 	// Collect recording legs with timing info for proper alignment (Fix G)
 	recordingLegs := make([]media.LegTiming, 0, len(callState.RTPForwarders))
 	if len(callState.RTPForwarders) > 0 {
@@ -3661,8 +3694,7 @@ func (s *CustomSIPServer) doFinalizeCall(callID string, callState *CallState, re
 				}
 				recordingLegs = append(recordingLegs, leg)
 			}
-			forwarder.Stop()
-			forwarder.Cleanup()
+			stopAndCleanup(forwarder)
 		}
 		callState.RTPForwarders = nil
 	} else if callState.RTPForwarder != nil {
@@ -3681,8 +3713,7 @@ func (s *CustomSIPServer) doFinalizeCall(callID string, callState *CallState, re
 			}
 			recordingLegs = append(recordingLegs, leg)
 		}
-		callState.RTPForwarder.Stop()
-		callState.RTPForwarder.Cleanup()
+		stopAndCleanup(callState.RTPForwarder)
 	}
 
 	callState.RTPForwarder = nil
@@ -3724,8 +3755,7 @@ func (s *CustomSIPServer) doFinalizeCall(callID string, callState *CallState, re
 					recordingLegs = append(recordingLegs, leg)
 				}
 			}
-			forwarder.Stop()
-			forwarder.Cleanup()
+			stopAndCleanup(forwarder)
 			s.logger.WithFields(logrus.Fields{
 				"call_id":   callID,
 				"stream_id": streamID,
@@ -4163,6 +4193,7 @@ func (s *CustomSIPServer) parseSiprecMetadata(rsMetadata []byte, contentType str
 
 	validation := siprec.ValidateSiprecMessage(&metadata)
 	if len(validation.Errors) > 0 {
+		s.logger.WithField("raw_metadata", string(rsMetadata)).Error("SIPREC metadata validation failed – raw XML body")
 		return nil, fmt.Errorf("critical SIPREC metadata validation failure: %v", validation.Errors)
 	}
 
@@ -4785,13 +4816,25 @@ func (s *CustomSIPServer) generateSiprecResponse(sdp []byte, session *siprec.Rec
 		CallID:    session.SIPID,
 	}
 
-	// Add stream information if available
-	for i, streamType := range session.MediaStreamTypes {
+	// Add stream elements using the actual SDP a=label values so jambonz/SRC can correlate.
+	// SDPStreamLabels is populated from the SDP offer's a=label attributes.
+	// Fall back to MediaStreamTypes for backward compat with non-Cognigy SRCs.
+	sdpLabels := session.SDPStreamLabels
+	if len(sdpLabels) == 0 {
+		for i := range session.MediaStreamTypes {
+			sdpLabels = append(sdpLabels, fmt.Sprintf("%d", i+1))
+		}
+	}
+	for i, label := range sdpLabels {
+		streamType := ""
+		if i < len(session.MediaStreamTypes) {
+			streamType = session.MediaStreamTypes[i]
+		}
 		stream := siprec.Stream{
-			Label:    fmt.Sprintf("stream_%d", i),
-			StreamID: fmt.Sprintf("stream_%d_%s", i, streamType),
+			Label:    label,
+			StreamID: fmt.Sprintf("stream-%d", i+1),
 			Type:     streamType,
-			Mode:     "separate", // Default to separate streams
+			Mode:     "separate",
 		}
 		responseMetadata.Streams = append(responseMetadata.Streams, stream)
 	}
@@ -4821,6 +4864,7 @@ func (s *CustomSIPServer) generateSiprecResponse(sdp []byte, session *siprec.Rec
 		"participant_count": len(responseMetadata.Participants),
 		"stream_count":      len(responseMetadata.Streams),
 	}).Info("Generated SIPREC multipart response")
+	logger.WithField("multipart_200ok_body", multipartBody).Info("Full 200 OK multipart body sent to SRC")
 
 	return contentType, multipartBody, nil
 }
@@ -4904,6 +4948,7 @@ func (s *CustomSIPServer) Shutdown(ctx context.Context) error {
 			}).Debug("Cleaning up RTP forwarder during shutdown")
 
 			forwarder.Stop()
+			forwarder.WaitDone(100 * time.Millisecond)
 			forwarder.Cleanup()
 		}