capture: match Apple's frame queue admission boundary

The shared encoded queue checked the duration after adding the incoming
frame, rejecting one picture before Apple's producer admission threshold.
Check existing queued duration >= 67 ms instead. At 30 fps, two pending
pictures occupy 66.7 ms, so a third is admitted and the fourth is rejected.

Grounding: artifacts/26A5388g__MacOS/decompiled/AirPlaySender.c,
vdsink_ShouldDropFrame (128485-128526). The queue is created at 188416,
passed to APVirtualDisplaySinkCreate at 188571, and consumed by
screenstream_dequeueAndProcessSampleBuffer at 319674. It holds encoded
CMSampleBuffers. The byte/chunk bounds and shared-sink detachment policy
remain unchanged; this does not port Apple's producer-feedback mechanism
or resolve longer stalls reported in PR #36.

Tests: failing-first 20/30/60 fps admission regressions, exact 67 ms equality,
drain/re-enqueue ordering, byte/chunk bounds, and oversized-first-frame
rejection. Full race suite, vet, and executable builds passed before commit.

Author: omarroth

internal/airplay/capture_broadcast.go

@@ -81,10 +81,10 @@ type BroadcastSink struct {
 
 	maxQueuedBytes  int
 	maxQueuedChunks int
-	// Apple's ordinary virtual-display source bounds its upstream frame queue to
-	// 67 ms and drops an incoming source frame at that limit. Doubletake derives
-	// a downstream encoded-relay ceiling from that value and counts configured
-	// sample durations. The byte and chunk limits remain independent safeguards.
+	// Apple's vdsink_ShouldDropFrame checks whether the existing encoded sample
+	// queue has reached 67 ms before signaling the producer to skip a frame.
+	// Match that admission boundary using configured sample durations. Shared
+	// relay overflow still detaches the sink; byte and chunk limits are separate.
 	maxFrameQueueDuration time.Duration
 	backpressure          bool
 	blockedProducers      int // number waiting for queue handoff; guarded by mu
@@ -379,16 +379,16 @@ func (s *BroadcastSink) frameQueueExceedsLimitsLocked(frame VideoAccessUnit) boo
 		return true
 	}
 	return len(s.frameQueue) > 0 && s.maxFrameQueueDuration > 0 &&
-		s.queuedFrameDuration+s.frameDuration > s.maxFrameQueueDuration
+		s.queuedFrameDuration >= s.maxFrameQueueDuration
 }
 
 // enqueueFrame appends an immutable complete access unit without copying it.
 // The same backing bytes can safely be referenced by every receiver queue.
 // An explicitly single-destination sink waits once one AU is pending, rather
 // than accumulating encoded references which cannot safely be dropped. Shared
 // fan-out is nonblocking and detaches only the sink which exceeds Doubletake's
-// nominal-duration relay budget. This policy is distinct from Apple's upstream
-// source-frame dropping behavior.
+// nominal-duration relay budget. Detachment is distinct from Apple's producer
+// admission signal and does not discard individual encoded references.
 func (s *BroadcastSink) enqueueFrame(frame VideoAccessUnit) error {
 	if len(frame.AnnexB) == 0 {
 		return nil

internal/airplay/capture_broadcast_test.go

@@ -192,15 +192,15 @@ func TestBroadcastSinkNonblockingFrameQueueUsesNominalDuration(t *testing.T) {
 	base := time.Now()
 	// A large or backward PTS gap can be caused by the upstream leaky queue; it
 	// must not turn one queued picture into an artificial duration overflow.
-	for i, offset := range []time.Duration{0, time.Second} {
+	for i, offset := range []time.Duration{0, time.Second, -time.Second} {
 		frame := VideoAccessUnit{AnnexB: []byte{byte(i + 1)}, PTS: base.Add(offset)}
 		if err := sink.enqueueFrame(frame); err != nil {
 			t.Fatalf("enqueue frame %d at %v: %v", i, offset, err)
 		}
 	}
-	third := VideoAccessUnit{AnnexB: []byte{3}, PTS: base.Add(-time.Second)}
-	if err := sink.enqueueFrame(third); !errors.Is(err, errBroadcastSinkBacklog) {
-		t.Fatalf("enqueue third nominal 30fps frame = %v, want backlog error", err)
+	fourth := VideoAccessUnit{AnnexB: []byte{4}, PTS: base.Add(2 * time.Second)}
+	if err := sink.enqueueFrame(fourth); !errors.Is(err, errBroadcastSinkBacklog) {
+		t.Fatalf("enqueue fourth nominal 30fps frame = %v, want backlog error", err)
 	}
 }
 
@@ -210,8 +210,9 @@ func TestBroadcastSinkNominalDurationUsesConfiguredFrameRate(t *testing.T) {
 		acceptedFrames  int
 		rejectedOrdinal int
 	}{
-		{fps: 20, acceptedFrames: 1, rejectedOrdinal: 2},
-		{fps: 60, acceptedFrames: 4, rejectedOrdinal: 5},
+		{fps: 20, acceptedFrames: 2, rejectedOrdinal: 3},
+		{fps: 30, acceptedFrames: 3, rejectedOrdinal: 4},
+		{fps: 60, acceptedFrames: 5, rejectedOrdinal: 6},
 	} {
 		t.Run(fmt.Sprintf("%dfps", test.fps), func(t *testing.T) {
 			broadcast := NewBroadcastCaptureWithFrameRate(nil, test.fps)
@@ -229,6 +230,63 @@ func TestBroadcastSinkNominalDurationUsesConfiguredFrameRate(t *testing.T) {
 	}
 }
 
+func TestBroadcastSinkRejectsFrameAtExactDurationThreshold(t *testing.T) {
+	sink := newBroadcastSink(nil)
+	defer sink.Close()
+	// Apple's admission check compares the existing queue to the threshold.
+	// Two samples land exactly on 67 ms; equality must reject the next sample.
+	sink.frameDuration = sink.maxFrameQueueDuration / 2
+	for _, value := range []byte{1, 2} {
+		if err := sink.enqueueFrame(VideoAccessUnit{AnnexB: []byte{value}}); err != nil {
+			t.Fatalf("enqueue frame %d: %v", value, err)
+		}
+	}
+	third := VideoAccessUnit{AnnexB: []byte{3}}
+	if err := sink.enqueueFrame(third); !errors.Is(err, errBroadcastSinkBacklog) {
+		t.Fatalf("enqueue at exact threshold = %v, want backlog error", err)
+	}
+	first, err := sink.ReadVideoAccessUnit()
+	if err != nil || !bytes.Equal(first.AnnexB, []byte{1}) {
+		t.Fatalf("first queued frame after rejected enqueue = (%x, %v), want 01", first.AnnexB, err)
+	}
+	if err := sink.enqueueFrame(third); err != nil {
+		t.Fatalf("enqueue after draining below threshold: %v", err)
+	}
+	for _, want := range []byte{2, 3} {
+		frame, err := sink.ReadVideoAccessUnit()
+		if err != nil || !bytes.Equal(frame.AnnexB, []byte{want}) {
+			t.Fatalf("queued frame = (%x, %v), want %02x", frame.AnnexB, err, want)
+		}
+	}
+}
+
+func TestBroadcastSinkFrameQueueRetainsByteAndChunkLimits(t *testing.T) {
+	for _, test := range []struct {
+		name                string
+		maxBytes, maxChunks int
+		prefill             bool
+	}{
+		{name: "bytes", maxBytes: 3, maxChunks: 10, prefill: true},
+		{name: "chunks", maxBytes: 100, maxChunks: 1, prefill: true},
+		{name: "oversized first frame", maxBytes: 1, maxChunks: 10},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			sink := newBroadcastSink(nil)
+			defer sink.Close()
+			sink.maxQueuedBytes, sink.maxQueuedChunks = test.maxBytes, test.maxChunks
+			frame := VideoAccessUnit{AnnexB: []byte{1, 2}}
+			if test.prefill {
+				if err := sink.enqueueFrame(frame); err != nil {
+					t.Fatalf("enqueue initial frame: %v", err)
+				}
+			}
+			if err := sink.enqueueFrame(frame); !errors.Is(err, errBroadcastSinkBacklog) {
+				t.Fatalf("enqueue below duration threshold = %v, want capacity error", err)
+			}
+		})
+	}
+}
+
 func TestBackpressuredSinkRegistrationIsExclusive(t *testing.T) {
 	sharedBroadcast := NewBroadcastCapture(nil)
 	shared := sharedBroadcast.AddSink()