Remove unsafe late-sink replay

Ultraworked with [omo](https://github.com/code-yeongyu/oh-my-openagent)

Co-authored-by: sisyphus-dev-ai <sisyphus-dev-ai@users.noreply.github.com>

Author: minpeter

internal/airplay/capture_broadcast.go

@@ -48,7 +48,6 @@ type BroadcastCapture struct {
 	// adjacent PTS values. A leaky upstream queue can legitimately create large
 	// PTS gaps while only one encoded picture is pending.
 	frameDuration time.Duration
-	now           func() time.Time
 	mu            sync.Mutex
 	done          chan struct{}
 	err           error // set before done is closed
@@ -58,11 +57,6 @@ type BroadcastCapture struct {
 	// following sequence, which gives attachment an exact cutover even when a
 	// source read has completed but has not yet been fanned out.
 	sequence uint64
-	// Timestamped access-unit fan-out caches the latest complete parameter-set
-	// plus random-access AU. Legacy byte-stream fan-out retains its exact
-	// next-read cutover and therefore waits for the encoder's next keyframe.
-	primer VideoAccessUnit
-
 	drainTimeout time.Duration
 
 	sinks []*BroadcastSink
@@ -79,7 +73,6 @@ type BroadcastSink struct {
 
 	queue       [][]byte
 	frameQueue  []VideoAccessUnit
-	primer      VideoAccessUnit
 	headOffset  int
 	queuedBytes int
 	// queuedFrameDuration is a nominal sample-duration sum. It deliberately does
@@ -96,7 +89,6 @@ type BroadcastSink struct {
 	maxFrameQueueDuration time.Duration
 	backpressure          bool
 	blockedProducers      int // number waiting for queue handoff; guarded by mu
-	awaitingRandomAccess  bool
 
 	inputClosed   bool // the source ended; drain queue, then return EOF
 	closed        bool // explicitly removed; discard queue and return EOF
@@ -145,7 +137,6 @@ func NewBroadcastCaptureWithFrameRate(src *ScreenCapture, fps int) *BroadcastCap
 		src:           src,
 		frames:        src != nil && src.frames != nil,
 		frameDuration: time.Second / time.Duration(fps),
-		now:           time.Now,
 		done:          make(chan struct{}),
 		drainTimeout:  broadcastSinkDrainTimeout,
 	}
@@ -164,11 +155,6 @@ func (bc *BroadcastCapture) AddSink() *BroadcastSink {
 		return s
 	}
 	s.startSequence = bc.sequence + 1
-	s.primer = bc.primer
-	if len(s.primer.AnnexB) > 0 {
-		s.primer.PTS = bc.now()
-		s.awaitingRandomAccess = true
-	}
 	bc.sinks = append(bc.sinks, s)
 	bc.mu.Unlock()
 	return s
@@ -194,11 +180,6 @@ func (bc *BroadcastCapture) AddBackpressuredSink() (*BroadcastSink, error) {
 	}
 	bc.exclusive = true
 	s.startSequence = bc.sequence + 1
-	s.primer = bc.primer
-	if len(s.primer.AnnexB) > 0 {
-		s.primer.PTS = bc.now()
-		s.awaitingRandomAccess = true
-	}
 	bc.sinks = append(bc.sinks, s)
 	bc.mu.Unlock()
 	return s, nil
@@ -286,12 +267,6 @@ func (bc *BroadcastCapture) runFrames() error {
 		frame, readErr := bc.src.ReadVideoAccessUnit()
 		if len(frame.AnnexB) > 0 {
 			bc.mu.Lock()
-			if len(frame.AnnexB) <= broadcastSinkQueueBytes && isDecoderPrimer(frame.AnnexB) {
-				bc.primer = VideoAccessUnit{
-					AnnexB: append([]byte(nil), frame.AnnexB...),
-					PTS:    frame.PTS,
-				}
-			}
 			sinks := make([]*BroadcastSink, 0, len(bc.sinks))
 			for _, sink := range bc.sinks {
 				if sink.startSequence <= sequence {
@@ -313,58 +288,6 @@ func (bc *BroadcastCapture) runFrames() error {
 	}
 }
 
-func isDecoderPrimer(annexB []byte) bool {
-	var h264SPS, h264PPS, h264IDR bool
-	var hevcVPS, hevcSPS, hevcPPS, hevcIRAP bool
-	for _, nal := range splitAnnexBAccessUnit(annexB) {
-		raw := stripStartCode(nal)
-		if len(raw) == 0 {
-			continue
-		}
-		switch raw[0] & 0x1f {
-		case 5:
-			h264IDR = true
-		case 7:
-			h264SPS = true
-		case 8:
-			h264PPS = true
-		}
-		if len(raw) < 2 {
-			continue
-		}
-		switch nalType := hevcNALType(raw); nalType {
-		case 32:
-			hevcVPS = true
-		case 33:
-			hevcSPS = true
-		case 34:
-			hevcPPS = true
-		default:
-			hevcIRAP = hevcIRAP || nalType >= 16 && nalType <= 23
-		}
-	}
-	return h264SPS && h264PPS && h264IDR || hevcVPS && hevcSPS && hevcPPS && hevcIRAP
-}
-
-func isRandomAccessUnit(annexB []byte) bool {
-	for _, nal := range splitAnnexBAccessUnit(annexB) {
-		raw := stripStartCode(nal)
-		if len(raw) == 0 {
-			continue
-		}
-		if raw[0]&0x1f == 5 {
-			return true
-		}
-		if len(raw) >= 2 {
-			nalType := hevcNALType(raw)
-			if nalType >= 16 && nalType <= 23 {
-				return true
-			}
-		}
-	}
-	return false
-}
-
 // finish stops accepting sinks, lets existing sinks drain, and only then
 // publishes BroadcastCapture completion.
 func (bc *BroadcastCapture) finish(err error) {
@@ -478,12 +401,6 @@ func (s *BroadcastSink) enqueueFrame(frame VideoAccessUnit) error {
 		if s.closed || s.inputClosed {
 			return io.ErrClosedPipe
 		}
-		if s.awaitingRandomAccess {
-			if !isRandomAccessUnit(frame.AnnexB) {
-				return nil
-			}
-			s.awaitingRandomAccess = false
-		}
 		if len(s.frameQueue) == 0 && len(frame.AnnexB) > s.maxQueuedBytes {
 			return errBroadcastSinkBacklog
 		}
@@ -506,7 +423,7 @@ func (s *BroadcastSink) enqueueFrame(frame VideoAccessUnit) error {
 }
 
 func (s *BroadcastSink) queueEmptyLocked() bool {
-	return len(s.queue) == 0 && len(s.frameQueue) == 0 && len(s.primer.AnnexB) == 0
+	return len(s.queue) == 0 && len(s.frameQueue) == 0
 }
 
 // finish marks source EOF without discarding data already queued.
@@ -536,7 +453,6 @@ func (s *BroadcastSink) abort() {
 			s.frameQueue[i].AnnexB = nil
 		}
 		s.frameQueue = nil
-		s.primer = VideoAccessUnit{}
 		s.headOffset = 0
 		s.queuedBytes = 0
 		s.queuedFrameDuration = 0
@@ -598,20 +514,12 @@ func (s *BroadcastSink) Read(p []byte) (int, error) {
 func (s *BroadcastSink) ReadVideoAccessUnit() (VideoAccessUnit, error) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
-	for len(s.primer.AnnexB) == 0 && len(s.frameQueue) == 0 && !s.inputClosed && !s.closed {
+	for len(s.frameQueue) == 0 && !s.inputClosed && !s.closed {
 		s.cond.Wait()
 	}
 	if s.closed {
 		return VideoAccessUnit{}, io.EOF
 	}
-	if len(s.primer.AnnexB) > 0 {
-		primer := s.primer
-		s.primer = VideoAccessUnit{}
-		if s.inputClosed && len(s.frameQueue) == 0 && len(s.queue) == 0 {
-			s.closeDoneLocked()
-		}
-		return primer, nil
-	}
 	if len(s.frameQueue) == 0 {
 		s.closeDoneLocked()
 		return VideoAccessUnit{}, io.EOF

internal/airplay/capture_broadcast_test.go

@@ -18,11 +18,6 @@ type channelVideoAccessUnitReader struct {
 	frames <-chan VideoAccessUnit
 }
 
-type signaledVideoAccessUnitReader struct {
-	frames <-chan VideoAccessUnit
-	reads  chan<- struct{}
-}
-
 func waitForBroadcastSinkState(t *testing.T, sink *BroadcastSink, predicate func(*BroadcastSink) bool, description string) {
 	t.Helper()
 	deadline := time.Now().Add(time.Second)
@@ -55,15 +50,6 @@ func (r *channelVideoAccessUnitReader) ReadVideoAccessUnit() (VideoAccessUnit, e
 	return frame, nil
 }
 
-func (r *signaledVideoAccessUnitReader) ReadVideoAccessUnit() (VideoAccessUnit, error) {
-	r.reads <- struct{}{}
-	frame, ok := <-r.frames
-	if !ok {
-		return VideoAccessUnit{}, io.EOF
-	}
-	return frame, nil
-}
-
 func (r *sliceVideoAccessUnitReader) ReadVideoAccessUnit() (VideoAccessUnit, error) {
 	if r.index == len(r.frames) {
 		return VideoAccessUnit{}, io.EOF
@@ -115,135 +101,6 @@ func TestBroadcastCapturePreservesTimestampedAccessUnits(t *testing.T) {
 	}
 }
 
-func TestBroadcastCaptureReplaysDecoderPrimerToLateSink(t *testing.T) {
-	frames := make(chan VideoAccessUnit)
-	reads := make(chan struct{}, 4)
-	capture := &ScreenCapture{
-		frames: &signaledVideoAccessUnitReader{frames: frames, reads: reads},
-		waitCh: make(chan struct{}),
-	}
-	broadcast := NewBroadcastCaptureWithFrameRate(capture, 30)
-	replayPTS := time.Unix(110, 0)
-	broadcast.now = func() time.Time { return replayPTS }
-	runDone := make(chan error, 1)
-	go func() { runDone <- broadcast.Run() }()
-
-	<-reads
-	primer := VideoAccessUnit{
-		AnnexB: []byte{
-			0, 0, 0, 1, 0x67, 0x42, 0x00, 0x1f,
-			0, 0, 0, 1, 0x68, 0xce, 0x06, 0xe2,
-			0, 0, 0, 1, 0x65, 0x80,
-		},
-		PTS: time.Unix(100, 0),
-	}
-	frames <- primer
-	<-reads
-
-	sink := broadcast.AddSink()
-	defer sink.Close()
-	replayDone := make(chan struct {
-		frame VideoAccessUnit
-		err   error
-	}, 1)
-	go func() {
-		frame, err := sink.ReadVideoAccessUnit()
-		replayDone <- struct {
-			frame VideoAccessUnit
-			err   error
-		}{frame: frame, err: err}
-	}()
-	var replayed VideoAccessUnit
-	select {
-	case result := <-replayDone:
-		if result.err != nil {
-			t.Fatalf("read replayed decoder primer: %v", result.err)
-		}
-		replayed = result.frame
-	case <-time.After(time.Second):
-		t.Fatal("cached decoder primer was not available immediately")
-	}
-	if !bytes.Equal(replayed.AnnexB, primer.AnnexB) {
-		t.Fatalf("first late-sink frame = %x, want cached decoder primer %x", replayed.AnnexB, primer.AnnexB)
-	}
-	if !replayed.PTS.Equal(replayPTS) {
-		t.Fatalf("replayed primer PTS = %v, want attachment PTS %v", replayed.PTS, replayPTS)
-	}
-
-	boundary := VideoAccessUnit{
-		AnnexB: []byte{0, 0, 0, 1, 0x61, 0x40},
-		PTS:    primer.PTS.Add(10 * time.Second),
-	}
-	frames <- boundary
-	<-reads
-	live := VideoAccessUnit{
-		AnnexB: []byte{0, 0, 0, 1, 0x61, 0x80},
-		PTS:    boundary.PTS.Add(time.Second / 30),
-	}
-	frames <- live
-	<-reads
-	randomAccess := VideoAccessUnit{
-		AnnexB: []byte{0, 0, 0, 1, 0x65, 0x80},
-		PTS:    live.PTS.Add(time.Second / 30),
-	}
-	frames <- randomAccess
-	close(frames)
-
-	next, err := sink.ReadVideoAccessUnit()
-	if err != nil {
-		t.Fatalf("read live frame after decoder primer: %v", err)
-	}
-	if !bytes.Equal(next.AnnexB, randomAccess.AnnexB) || !next.PTS.Equal(randomAccess.PTS) {
-		t.Fatalf("live frame after primer = {%x %v}, want random access {%x %v}", next.AnnexB, next.PTS, randomAccess.AnnexB, randomAccess.PTS)
-	}
-	if err := <-runDone; !errors.Is(err, io.EOF) {
-		t.Fatalf("broadcast run = %v, want EOF", err)
-	}
-}
-
-func TestBroadcastCapturePrimerSnapshotSurvivesRefresh(t *testing.T) {
-	frames := make(chan VideoAccessUnit)
-	reads := make(chan struct{}, 4)
-	capture := &ScreenCapture{
-		frames: &signaledVideoAccessUnitReader{frames: frames, reads: reads},
-		waitCh: make(chan struct{}),
-	}
-	broadcast := NewBroadcastCapture(capture)
-	broadcast.now = func() time.Time { return time.Unix(120, 0) }
-	runDone := make(chan error, 1)
-	go func() { runDone <- broadcast.Run() }()
-
-	first := VideoAccessUnit{AnnexB: []byte{
-		0, 0, 0, 1, 0x67, 0x42,
-		0, 0, 0, 1, 0x68, 0xce,
-		0, 0, 0, 1, 0x65, 0xaa,
-	}}
-	second := VideoAccessUnit{AnnexB: []byte{
-		0, 0, 0, 1, 0x67, 0x64,
-		0, 0, 0, 1, 0x68, 0xee,
-		0, 0, 0, 1, 0x65, 0xbb,
-	}}
-	<-reads
-	frames <- first
-	<-reads
-	sink := broadcast.AddSink()
-	defer sink.Close()
-	frames <- second
-	<-reads
-
-	got, err := sink.ReadVideoAccessUnit()
-	if err != nil {
-		t.Fatalf("read primer snapshot: %v", err)
-	}
-	if !bytes.Equal(got.AnnexB, first.AnnexB) {
-		t.Fatalf("primer snapshot = %x, want %x", got.AnnexB, first.AnnexB)
-	}
-	close(frames)
-	if err := <-runDone; !errors.Is(err, io.EOF) {
-		t.Fatalf("broadcast run = %v, want EOF", err)
-	}
-}
-
 func TestBroadcastSinkBackpressuresWithOnePendingAccessUnit(t *testing.T) {
 	sink := newBroadcastSinkWithPolicy(nil, true)
 	base := time.Now()