Retain final stats after PeerConnection close

icegatherer.go

@@ -31,6 +31,9 @@ type ICEGatherer struct {
 	agent           *ice.Agent
 	localParameters ICEParameters
 
+	// Stats taken just before the ICETransport closes the agent.
+	finalStats *iceAgentStats
+
 	onLocalCandidateHandler atomic.Value // func(candidate *ICECandidate)
 	onStateChangeHandler    atomic.Value // func(state ICEGathererState)
 
@@ -609,86 +612,119 @@ func (g *ICEGatherer) getAgent() *ice.Agent {
 }
 
 func (g *ICEGatherer) collectStats(collector *statsReportCollector, transportStatsID string) {
-	agent := g.getAgent()
+	g.lock.RLock()
+	agent := g.agent
+	finalStats := g.finalStats
+	g.lock.RUnlock()
+
+	if finalStats != nil {
+		finalStats.collect(collector, transportStatsID)
+
+		return
+	}
+
 	if agent == nil {
 		return
 	}
 
 	collector.Collecting()
 	go func(collector *statsReportCollector, agent *ice.Agent) {
-		for _, candidatePairStats := range agent.GetCandidatePairsStats() {
-			collector.Collecting()
+		g.agentStats(agent).collect(collector, transportStatsID)
+		collector.Done()
+	}(collector, agent)
+}
 
-			stats, err := toICECandidatePairStats(candidatePairStats)
-			if err != nil {
-				g.log.Error(err.Error())
-				collector.Done()
+// retainFinalStats stores the stats of the ICE agent so that they stay available after it is closed.
+// It must be called before the agent is closed.
+func (g *ICEGatherer) retainFinalStats() {
+	g.lock.RLock()
+	agent := g.agent
+	alreadyRetained := g.finalStats != nil
+	g.lock.RUnlock()
 
-				continue
-			}
-			stats.TransportID = transportStatsID
+	if agent == nil || alreadyRetained {
+		return
+	}
 
-			collector.Collect(stats.ID, stats)
-		}
+	finalStats := g.agentStats(agent)
 
-		for _, candidateStats := range agent.GetLocalCandidatesStats() {
-			collector.Collecting()
+	g.lock.Lock()
+	g.finalStats = finalStats
+	g.lock.Unlock()
+}
 
-			networkType, err := getNetworkType(candidateStats.NetworkType)
-			if err != nil {
-				g.log.Error(err.Error())
-			}
+// iceAgentStats holds the stats read from an ICE agent.
+type iceAgentStats struct {
+	candidatePairs []ICECandidatePairStats
+	candidates     []ICECandidateStats
+}
 
-			candidateType, err := getCandidateType(candidateStats.CandidateType)
-			if err != nil {
-				g.log.Error(err.Error())
-			}
+func (s *iceAgentStats) collect(collector *statsReportCollector, transportStatsID string) {
+	for _, stats := range s.candidatePairs {
+		collector.Collecting()
+		stats.TransportID = transportStatsID
+		collector.Collect(stats.ID, stats)
+	}
 
-			stats := ICECandidateStats{
-				Timestamp:     statsTimestampFrom(candidateStats.Timestamp),
-				ID:            candidateStats.ID,
-				TransportID:   transportStatsID,
-				Type:          StatsTypeLocalCandidate,
-				Address:       candidateStats.IP,
-				Port:          int32(candidateStats.Port), //nolint:gosec // G115, no overflow, port
-				Protocol:      networkType.Protocol(),
-				CandidateType: candidateType,
-				Priority:      int32(candidateStats.Priority), //nolint:gosec
-				URL:           candidateStats.URL,
-				RelayProtocol: candidateStats.RelayProtocol,
-			}
-			collector.Collect(stats.ID, stats)
-		}
+	for _, stats := range s.candidates {
+		collector.Collecting()
+		stats.TransportID = transportStatsID
+		collector.Collect(stats.ID, stats)
+	}
+}
 
-		for _, candidateStats := range agent.GetRemoteCandidatesStats() {
-			collector.Collecting()
-			networkType, err := getNetworkType(candidateStats.NetworkType)
-			if err != nil {
-				g.log.Error(err.Error())
-			}
+func (g *ICEGatherer) agentStats(agent *ice.Agent) *iceAgentStats {
+	agentStats := &iceAgentStats{}
 
-			candidateType, err := getCandidateType(candidateStats.CandidateType)
-			if err != nil {
-				g.log.Error(err.Error())
-			}
+	for _, candidatePairStats := range agent.GetCandidatePairsStats() {
+		stats, err := toICECandidatePairStats(candidatePairStats)
+		if err != nil {
+			g.log.Error(err.Error())
 
-			stats := ICECandidateStats{
-				Timestamp:     statsTimestampFrom(candidateStats.Timestamp),
-				ID:            candidateStats.ID,
-				TransportID:   transportStatsID,
-				Type:          StatsTypeRemoteCandidate,
-				Address:       candidateStats.IP,
-				Port:          int32(candidateStats.Port), //nolint:gosec // G115, no overflow, port
-				Protocol:      networkType.Protocol(),
-				CandidateType: candidateType,
-				Priority:      int32(candidateStats.Priority), //nolint:gosec // G115
-				URL:           candidateStats.URL,
-				RelayProtocol: candidateStats.RelayProtocol,
-			}
-			collector.Collect(stats.ID, stats)
+			continue
 		}
-		collector.Done()
-	}(collector, agent)
+
+		agentStats.candidatePairs = append(agentStats.candidatePairs, stats)
+	}
+
+	for _, candidateStats := range agent.GetLocalCandidatesStats() {
+		agentStats.candidates = append(
+			agentStats.candidates, g.toICECandidateStats(candidateStats, StatsTypeLocalCandidate),
+		)
+	}
+
+	for _, candidateStats := range agent.GetRemoteCandidatesStats() {
+		agentStats.candidates = append(
+			agentStats.candidates, g.toICECandidateStats(candidateStats, StatsTypeRemoteCandidate),
+		)
+	}
+
+	return agentStats
+}
+
+func (g *ICEGatherer) toICECandidateStats(candidateStats ice.CandidateStats, statsType StatsType) ICECandidateStats {
+	networkType, err := getNetworkType(candidateStats.NetworkType)
+	if err != nil {
+		g.log.Error(err.Error())
+	}
+
+	candidateType, err := getCandidateType(candidateStats.CandidateType)
+	if err != nil {
+		g.log.Error(err.Error())
+	}
+
+	return ICECandidateStats{
+		Timestamp:     statsTimestampFrom(candidateStats.Timestamp),
+		ID:            candidateStats.ID,
+		Type:          statsType,
+		Address:       candidateStats.IP,
+		Port:          int32(candidateStats.Port), //nolint:gosec // G115, no overflow, port
+		Protocol:      networkType.Protocol(),
+		CandidateType: candidateType,
+		Priority:      int32(candidateStats.Priority), //nolint:gosec // G115
+		URL:           candidateStats.URL,
+		RelayProtocol: candidateStats.RelayProtocol,
+	}
 }
 
 func (g *ICEGatherer) getSelectedCandidatePairStats() (ICECandidatePairStats, bool) {

icetransport.go

@@ -263,6 +263,10 @@ func (t *ICETransport) stop(shouldGracefullyClose bool) error {
 	gatherer := t.gatherer
 	t.lock.Unlock()
 
+	if gatherer != nil {
+		gatherer.retainFinalStats()
+	}
+
 	if conn != nil {
 		var closeErrs []error
 		if shouldGracefullyClose && gatherer != nil {

interceptor.go

@@ -109,6 +109,14 @@ func cleanupStats(id string) {
 	statsGetter.Delete(id)
 }
 
+// finalStatsGetter is a stats.Getter that serves the stats streams had when their PeerConnection was closed.
+type finalStatsGetter map[uint32]*stats.Stats
+
+// Get returns the final stats of the stream with ssrc.
+func (g finalStatsGetter) Get(ssrc uint32) *stats.Stats {
+	return g[ssrc]
+}
+
 // key: string (peerconnection.statsId), value: stats.Getter
 var statsGetter sync.Map // nolint:gochecknoglobals
 

peerconnection.go

@@ -101,6 +101,7 @@ type PeerConnection struct {
 
 	interceptorRTCPWriter interceptor.RTCPWriter
 	statsGetter           stats.Getter
+	finalRTPStats         finalStatsGetter
 }
 
 // NewPeerConnection creates a PeerConnection with the default codecs and interceptors.
@@ -2941,6 +2942,19 @@ func (pc *PeerConnection) close(shouldGracefullyClose bool) error { //nolint:cyc
 
 	// https://www.w3.org/TR/webrtc/#dom-rtcpeerconnection-close (step #4)
 	pc.mu.Lock()
+	// Stopping a sender or a receiver drops its stats, so retain them first.
+	if pc.statsGetter != nil {
+		pc.finalRTPStats = finalStatsGetter{}
+		for _, t := range pc.rtpTransceivers {
+			if sender := t.Sender(); sender != nil {
+				sender.retainFinalStats(pc.statsGetter)
+			}
+			if receiver := t.Receiver(); receiver != nil {
+				receiver.retainFinalStats(pc.statsGetter, pc.finalRTPStats)
+			}
+		}
+		pc.statsGetter = nil
+	}
 	for _, t := range pc.rtpTransceivers {
 		closeErrs = append(closeErrs, t.Stop())
 	}
@@ -2976,7 +2990,6 @@ func (pc *PeerConnection) close(shouldGracefullyClose bool) error { //nolint:cyc
 
 	closeErrs = append(closeErrs, doGracefulCloseOps()...)
 
-	pc.statsGetter = nil
 	cleanupStats(pc.id)
 
 	// Interceptor closes at the end to prevent Bind from being called after interceptor is closed
@@ -3142,22 +3155,33 @@ func (pc *PeerConnection) GetStats() StatsReport {
 			continue
 		}
 	}
+	statsGetter := pc.rtpStatsGetter()
 	pc.mu.Unlock()
 
 	receivers := pc.GetReceivers()
 	for _, receiver := range receivers {
-		receiver.collectStats(statsCollector, pc.statsGetter)
+		receiver.collectStats(statsCollector, statsGetter)
 	}
 
 	for _, sender := range pc.GetSenders() {
-		sender.collectStats(statsCollector, pc.statsGetter)
+		sender.collectStats(statsCollector, statsGetter)
 	}
 
 	pc.api.mediaEngine.collectStats(statsCollector, pc.dtlsTransport.statsID)
 
 	return statsCollector.Ready()
 }
 
+// rtpStatsGetter returns the source of the RTP stats, which are the retained ones after close.
+// The caller must hold pc.mu.
+func (pc *PeerConnection) rtpStatsGetter() stats.Getter {
+	if pc.finalRTPStats != nil {
+		return pc.finalRTPStats
+	}
+
+	return pc.statsGetter
+}
+
 // Start all transports. PeerConnection now has enough state.
 func (pc *PeerConnection) startTransports(
 	iceRole ICERole,

rtpreceiver.go

@@ -469,6 +469,19 @@ func (r *RTPReceiver) collectStats(collector *statsReportCollector, statsGetter
 	}
 }
 
+// retainFinalStats copies the stats of the receiver's tracks from statsGetter to finalStats.
+func (r *RTPReceiver) retainFinalStats(statsGetter stats.Getter, finalStats finalStatsGetter) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+
+	for trackIndex := range r.tracks {
+		if remoteTrack := r.tracks[trackIndex].track; remoteTrack != nil {
+			ssrc := uint32(remoteTrack.SSRC())
+			finalStats[ssrc] = statsGetter.Get(ssrc)
+		}
+	}
+}
+
 func (r *RTPReceiver) populateInboundStats(
 	inboundStats *InboundRTPStreamStats,
 	statsGetter stats.Getter,

rtpsender.go

@@ -54,6 +54,9 @@ type RTPSender struct {
 
 	rtpTransceiver *RTPTransceiver
 
+	// Stats taken just before the PeerConnection stops the sender.
+	finalStats []OutboundRTPStreamStats
+
 	mu                     sync.RWMutex
 	sendCalled, stopCalled chan struct{}
 }
@@ -490,25 +493,46 @@ func (r *RTPSender) SetReadDeadlineSimulcast(deadline time.Time, rid string) err
 }
 
 func (r *RTPSender) collectStats(collector *statsReportCollector, statsGetter stats.Getter) {
-	if statsGetter == nil || !r.hasSent() || r.hasStopped() {
-		return
-	}
-
 	r.mu.RLock()
 	defer r.mu.RUnlock()
 
+	outboundStats := r.finalStats
+	if outboundStats == nil {
+		outboundStats = r.outboundStats(statsGetter)
+	}
+
+	for _, stats := range outboundStats {
+		collector.Collecting()
+		collector.Collect(stats.ID, stats)
+	}
+}
+
+// retainFinalStats stores the stats of the sender so that they stay available after it is stopped.
+// It must be called before the sender is stopped.
+func (r *RTPSender) retainFinalStats(statsGetter stats.Getter) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+
+	r.finalStats = r.outboundStats(statsGetter)
+}
+
+// outboundStats returns the stats of the sender's encodings. The caller must hold r.mu.
+func (r *RTPSender) outboundStats(statsGetter stats.Getter) []OutboundRTPStreamStats {
+	if statsGetter == nil || !r.hasSent() || r.hasStopped() {
+		return nil
+	}
+
 	mid := ""
 	if r.rtpTransceiver != nil {
 		mid = r.rtpTransceiver.Mid()
 	}
 	now := statsTimestampNow()
+	result := make([]OutboundRTPStreamStats, 0, len(r.trackEncodings))
 	for _, trackEncoding := range r.trackEncodings {
 		if trackEncoding.track == nil || trackEncoding.context == nil {
 			continue
 		}
 
-		collector.Collecting()
-
 		outboundStats := OutboundRTPStreamStats{
 			Mid:         mid,
 			Rid:         trackEncoding.track.RID(),
@@ -531,8 +555,10 @@ func (r *RTPSender) collectStats(collector *statsReportCollector, statsGetter st
 			outboundStats.PLICount = stats.OutboundRTPStreamStats.PLICount
 		}
 
-		collector.Collect(outboundStats.ID, outboundStats)
+		result = append(result, outboundStats)
 	}
+
+	return result
 }
 
 // hasSent tells if data has been ever sent for this instance.