diff --git a/pkg/media/rtp.go b/pkg/media/rtp.go index 236e1f8d..28305649 100644 --- a/pkg/media/rtp.go +++ b/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 diff --git a/pkg/media/sdp_utils.go b/pkg/media/sdp_utils.go index f9cb3b58..05ed9209 100644 --- a/pkg/media/sdp_utils.go +++ b/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) { diff --git a/pkg/media/types.go b/pkg/media/types.go index 95a3811d..ecb63f2b 100644 --- a/pkg/media/types.go +++ b/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) { diff --git a/pkg/sip/custom_server.go b/pkg/sip/custom_server.go index 22bb5b55..6d1e985f 100644 --- a/pkg/sip/custom_server.go +++ b/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() } diff --git a/pkg/sip/sdp.go b/pkg/sip/sdp.go index 5395277b..dcc9259a 100644 --- a/pkg/sip/sdp.go +++ b/pkg/sip/sdp.go @@ -282,7 +282,8 @@ func (h *Handler) generateSDPAdvanced(receivedSDP *sdp.SessionDescription, optio foundDirectionAttr = true continue case "recvonly": - directionAttr = sdp.Attribute{Key: "recvonly"} + // Offer says recvonly → they receive, we must send (reversed SIPREC) + directionAttr = sdp.Attribute{Key: "sendonly"} foundDirectionAttr = true continue case "rtcp": @@ -414,16 +415,18 @@ func (h *Handler) generateSDPAdvanced(receivedSDP *sdp.SessionDescription, optio } // Create the complete session description + // Use our own origin (not mirroring the SRC's origin) to avoid loop detection + // on Cognigy/jambonz media stacks that compare session IDs. sessionDesc := &sdp.SessionDescription{ Origin: sdp.Origin{ - Username: receivedSDP.Origin.Username, - SessionID: receivedSDP.Origin.SessionID, - SessionVersion: receivedSDP.Origin.SessionVersion, + Username: "siprec-srs", + SessionID: uint64(time.Now().UnixMicro()), + SessionVersion: 1, NetworkType: "IN", AddressType: "IP4", - UnicastAddress: connectionAddr, // Use NAT-aware address + UnicastAddress: connectionAddr, }, - SessionName: receivedSDP.SessionName, + SessionName: sdp.SessionName("SRS Recording Session"), ConnectionInformation: &sdp.ConnectionInformation{ NetworkType: "IN", AddressType: "IP4", diff --git a/pkg/siprec/parser.go b/pkg/siprec/parser.go index 0c2a17f6..7eb905f4 100644 --- a/pkg/siprec/parser.go +++ b/pkg/siprec/parser.go @@ -548,7 +548,8 @@ func ValidateSiprecMessage(rsMetadata *RSMetadata) ValidationResult { assoc := rsMetadata.SessionRecordingAssoc if (assoc == RSAssociation{}) { - result.addError("missing session recording association element") + // Cognigy VGW (jambonz) does not send – downgraded to warning + result.addWarning("missing session recording association element (non-RFC-compliant SRC – continuing)") } else { assocSessionID := normalizedAssocSessionID(assoc) if assocSessionID == "" { diff --git a/pkg/siprec/types.go b/pkg/siprec/types.go index 3fa61565..4e24ea39 100644 --- a/pkg/siprec/types.go +++ b/pkg/siprec/types.go @@ -23,6 +23,7 @@ type RecordingSession struct { StateReasonRef string StateExpires time.Time MediaStreamTypes []string + SDPStreamLabels []string // Actual a=label values from the SDP offer, used for RS-metadata response SessionGroups []SessionGroupAssociation PolicyUpdates []PolicyUpdate SessionGroupRoles map[string]string