start of fix

Overlord-Client/cmd/agent/webrtcpub/p2p_pion.go

@@ -147,7 +147,7 @@ func StartP2POffer(ctx context.Context, kind Kind, sessionID string, offerSDP st
 			return "", fmt.Errorf("add video transceiver: %w", err)
 		}
 		if sender := tx.Sender(); sender != nil {
-			go drainRTCP(sender)
+			go drainRTCP(sender, kind)
 		}
 	}
 	if hasAudio {
@@ -172,7 +172,7 @@ func StartP2POffer(ctx context.Context, kind Kind, sessionID string, offerSDP st
 			return "", fmt.Errorf("add audio transceiver: %w", err)
 		}
 		if sender := tx.Sender(); sender != nil {
-			go drainRTCP(sender)
+			go drainRTCP(sender, "")
 		}
 	}
 

Overlord-Client/cmd/agent/webrtcpub/state.go

@@ -110,7 +110,7 @@ func registerVideoWriter(kind Kind, id string, w VideoWriter) {
 	entry.video = newLatestVideoWriter(w)
 	bucket[id] = entry
 	writersMu.Unlock()
-	RequestKeyframe()
+	RequestKeyframe(kind)
 }
 
 func registerAudioWriter(kind Kind, id string, w AudioWriter) {
@@ -155,14 +155,29 @@ func IsActive(kind Kind) bool {
 	return len(writers[string(kind)]) > 0
 }
 
-var keyframeWanted atomic.Bool
+var (
+	desktopKeyframeWanted   atomic.Bool
+	backstageKeyframeWanted atomic.Bool
+	webcamKeyframeWanted    atomic.Bool
+)
 
-func RequestKeyframe() {
-	keyframeWanted.Store(true)
+func keyframeRequestFlag(kind Kind) *atomic.Bool {
+	switch kind {
+	case Kindbackstage:
+		return &backstageKeyframeWanted
+	case KindWebcam:
+		return &webcamKeyframeWanted
+	default:
+		return &desktopKeyframeWanted
+	}
 }
 
-func ConsumeKeyframeRequest() bool {
-	return keyframeWanted.Swap(false)
+func RequestKeyframe(kind Kind) {
+	keyframeRequestFlag(kind).Store(true)
+}
+
+func ConsumeKeyframeRequest(kind Kind) bool {
+	return keyframeRequestFlag(kind).Swap(false)
 }
 
 func WriteH264(kind Kind, nalu []byte, dur time.Duration) error {

Overlord-Client/cmd/agent/webrtcpub/state_test.go

@@ -67,3 +67,21 @@ func TestWriteH264DoesNotBlockAndKeepsLatestPendingFrame(t *testing.T) {
 	}
 	t.Fatal("latest pending frame was not delivered")
 }
+
+func TestKeyframeRequestsAreIsolatedByStreamKind(t *testing.T) {
+	for _, kind := range []Kind{KindDesktop, Kindbackstage, KindWebcam} {
+		_ = ConsumeKeyframeRequest(kind)
+	}
+
+	RequestKeyframe(Kindbackstage)
+
+	if ConsumeKeyframeRequest(KindDesktop) {
+		t.Fatal("backstage keyframe request leaked into desktop stream")
+	}
+	if ConsumeKeyframeRequest(KindWebcam) {
+		t.Fatal("backstage keyframe request leaked into webcam stream")
+	}
+	if !ConsumeKeyframeRequest(Kindbackstage) {
+		t.Fatal("backstage keyframe request was not retained for backstage stream")
+	}
+}

Overlord-Client/cmd/agent/webrtcpub/whip_pion.go

@@ -25,7 +25,7 @@ import (
 	"github.com/pion/webrtc/v4/pkg/media"
 )
 
-func drainRTCP(sender *webrtc.RTPSender) {
+func drainRTCP(sender *webrtc.RTPSender, kind Kind) {
 	buf := make([]byte, 1500)
 	for {
 		n, _, err := sender.Read(buf)
@@ -39,7 +39,9 @@ func drainRTCP(sender *webrtc.RTPSender) {
 		for _, p := range packets {
 			switch p.(type) {
 			case *rtcp.PictureLossIndication, *rtcp.FullIntraRequest:
-				handleRTCPKeyframeFeedback()
+				if kind != "" {
+					handleRTCPKeyframeFeedback(kind)
+				}
 			}
 		}
 	}
@@ -48,11 +50,15 @@ func drainRTCP(sender *webrtc.RTPSender) {
 const h264SDPFmtpLine = "level-asymmetry-allowed=1;packetization-mode=1;profile-level-id=640034"
 
 var rtcpKeyframeLogOnce sync.Once
-var lastRTCPKeyframeRequest atomic.Int64
+var (
+	lastDesktopRTCPKeyframeRequest   atomic.Int64
+	lastBackstageRTCPKeyframeRequest atomic.Int64
+	lastWebcamRTCPKeyframeRequest    atomic.Int64
+)
 
 const rtcpKeyframeMinInterval = 750 * time.Millisecond
 
-func handleRTCPKeyframeFeedback() {
+func handleRTCPKeyframeFeedback(kind Kind) {
 	if !honorRTCPKeyframes() {
 		rtcpKeyframeLogOnce.Do(func() {
 			log.Printf("webrtcpub: RTCP video keyframe recovery disabled by OVERLORD_WEBRTC_RTCP_KEYFRAMES")
@@ -61,15 +67,26 @@ func handleRTCPKeyframeFeedback() {
 	}
 	now := time.Now().UnixNano()
 	for {
-		last := lastRTCPKeyframeRequest.Load()
+		last := rtcpKeyframeTimestamp(kind).Load()
 		if last > 0 && time.Duration(now-last) < rtcpKeyframeMinInterval {
 			return
 		}
-		if lastRTCPKeyframeRequest.CompareAndSwap(last, now) {
+		if rtcpKeyframeTimestamp(kind).CompareAndSwap(last, now) {
 			break
 		}
 	}
-	RequestKeyframe()
+	RequestKeyframe(kind)
+}
+
+func rtcpKeyframeTimestamp(kind Kind) *atomic.Int64 {
+	switch kind {
+	case Kindbackstage:
+		return &lastBackstageRTCPKeyframeRequest
+	case KindWebcam:
+		return &lastWebcamRTCPKeyframeRequest
+	default:
+		return &lastDesktopRTCPKeyframeRequest
+	}
 }
 
 func honorRTCPKeyframes() bool {
@@ -188,7 +205,7 @@ func Start(ctx context.Context, kind Kind, opts Options) (*Publisher, error) {
 			return nil, fmt.Errorf("add video transceiver: %w", err)
 		}
 		if sender := tx.Sender(); sender != nil {
-			go drainRTCP(sender)
+			go drainRTCP(sender, kind)
 		}
 	}
 	if opts.HasAudio {
@@ -213,7 +230,7 @@ func Start(ctx context.Context, kind Kind, opts Options) (*Publisher, error) {
 			return nil, fmt.Errorf("add audio transceiver: %w", err)
 		}
 		if sender := tx.Sender(); sender != nil {
-			go drainRTCP(sender)
+			go drainRTCP(sender, "")
 		}
 	}