fix(gateway): bound Apollo video ingress

gateway/apollo_native.go

@@ -33,6 +33,9 @@ const (
 	nativeApolloVideoQueueLatency = 250 * time.Millisecond
 	nativeApolloAudioQueuePackets = 16
 	nativeApolloEventQueuePackets = 16
+	nativeApolloVideoIngressSlots = 2048
+	nativeApolloVideoPacketBytes  = apolloVideoHeaderSize + apolloVideoRawPacketSize
+	nativeApolloVideoReadBuffer   = nativeApolloVideoIngressSlots * nativeApolloVideoPacketBytes
 )
 
 // NativeApolloBackend keeps provider sockets inside the gateway process. The
@@ -41,8 +44,9 @@ const (
 type NativeApolloBackend struct {
 	Dialer *net.Dialer
 
-	mu      sync.Mutex
-	pending map[string]*apolloRTSPSetup
+	mu             sync.Mutex
+	pending        map[string]*apolloRTSPSetup
+	configureMedia func(*apolloMediaCodec)
 }
 
 func NewNativeApolloBackend() *NativeApolloBackend {
@@ -268,7 +272,7 @@ func (b *NativeApolloBackend) Open(ctx context.Context, request LaunchRequest, _
 	if !ok || setup == nil {
 		return nil, ErrProviderDisconnected
 	}
-	return newNativeApolloProviderSession(ctx, setup)
+	return newNativeApolloProviderSession(ctx, setup, b.configureMedia)
 }
 
 func readBounded(reader io.Reader, max int) ([]byte, error) {
@@ -336,7 +340,7 @@ func newNativeApolloSession(sessionID string) *nativeApolloSession {
 	}
 }
 
-func newNativeApolloProviderSession(ctx context.Context, setup *apolloRTSPSetup) (*nativeApolloSession, error) {
+func newNativeApolloProviderSession(ctx context.Context, setup *apolloRTSPSetup, configureMedia func(*apolloMediaCodec)) (*nativeApolloSession, error) {
 	if setup == nil || len(setup.streamKey) != 16 || setup.controlPort == 0 || setup.audioPort == 0 || setup.videoPort == 0 {
 		return nil, ErrProviderMalformed
 	}
@@ -360,6 +364,9 @@ func newNativeApolloProviderSession(ctx context.Context, setup *apolloRTSPSetup)
 		peer.close(err)
 		return nil, err
 	}
+	if configureMedia != nil {
+		configureMedia(media)
+	}
 	session := newNativeApolloSession(setup.sessionID)
 	managementClient, err := newPinnedApolloHTTPClient(setup.providerWork)
 	if err != nil {
@@ -391,6 +398,12 @@ func newNativeApolloProviderSession(ctx context.Context, setup *apolloRTSPSetup)
 		peer.close(err)
 		return nil, err
 	}
+	if err := videoConn.SetReadBuffer(nativeApolloVideoReadBuffer); err != nil {
+		_ = audioConn.Close()
+		_ = videoConn.Close()
+		peer.close(err)
+		return nil, fmt.Errorf("set Apollo video receive buffer: %w", err)
+	}
 	if _, err := audioConn.Write(apolloMediaPing(setup.audioPing, 1)); err != nil {
 		_ = audioConn.Close()
 		_ = videoConn.Close()
@@ -896,19 +909,20 @@ func (s *nativeApolloSession) readUDPMedia() {
 		s.closeMediaChannels()
 		return
 	}
-	var readers sync.WaitGroup
-	readers.Add(2)
-	read := func(conn *net.UDPConn, output chan ProviderMedia, video bool) {
-		defer readers.Done()
+	videoIngress := newNativeApolloVideoIngress()
+	var workers sync.WaitGroup
+	workers.Add(3)
+	go func() {
+		defer workers.Done()
 		buffer := make([]byte, apolloMediaMaximumPacket+1)
 		for {
 			if s.mediaQuiesced.Load() {
 				return
 			}
-			if err := conn.SetReadDeadline(time.Now().Add(250 * time.Millisecond)); err != nil {
+			if err := s.audioConn.SetReadDeadline(time.Now().Add(250 * time.Millisecond)); err != nil {
 				return
 			}
-			count, err := conn.Read(buffer)
+			count, err := s.audioConn.Read(buffer)
 			receivedAt := time.Now()
 			if err != nil {
 				if networkErr, ok := err.(net.Error); ok && networkErr.Timeout() {
@@ -928,45 +942,121 @@ func (s *nativeApolloSession) readUDPMedia() {
 				return
 			}
 			s.mediaIngress.Add(1)
-			var payloads [][]byte
-			if video {
-				shard, openErr := s.media.OpenVideo(buffer[:count])
-				if openErr != nil {
-					continue
-				}
-				payload, err := s.videoFEC.Add(shard)
-				if err != nil || len(payload) == 0 {
-					continue
-				}
-				payloads = [][]byte{payload}
-			} else {
-				shard, openErr := s.media.OpenAudio(buffer[:count])
-				if openErr != nil {
-					continue
-				}
-				var evicted bool
-				payloads, evicted, err = s.audioFEC.Add(s.media, shard)
-				if evicted {
-					s.mediaDrops.Add(1)
-				}
+			shard, openErr := s.media.OpenAudio(buffer[:count])
+			if openErr != nil {
+				continue
+			}
+			payloads, evicted, err := s.audioFEC.Add(s.media, shard)
+			if evicted {
+				s.mediaDrops.Add(1)
 			}
 			if err != nil {
 				continue
 			}
 			for _, payload := range payloads {
-				s.enqueueMedia(output, payload, receivedAt)
+				s.enqueueMedia(s.audio, payload, receivedAt)
 			}
 		}
-	}
-	go read(s.audioConn, s.audio, false)
-	go read(s.videoConn, s.video, true)
+	}()
 	go func() {
-		readers.Wait()
+		defer workers.Done()
+		s.drainApolloVideo(videoIngress)
+	}()
+	go func() {
+		defer workers.Done()
+		s.processApolloVideo(videoIngress)
+	}()
+	go func() {
+		workers.Wait()
 		close(s.readDone)
 		s.closeMediaChannels()
 	}()
 }
 
+type nativeApolloVideoIngressSlot struct {
+	packet     [apolloMediaMaximumPacket + 1]byte
+	count      int
+	receivedAt time.Time
+}
+
+type nativeApolloVideoIngress struct {
+	slots   [nativeApolloVideoIngressSlots]nativeApolloVideoIngressSlot
+	free    chan uint16
+	ready   chan uint16
+	scratch [apolloMediaMaximumPacket + 1]byte
+}
+
+func newNativeApolloVideoIngress() *nativeApolloVideoIngress {
+	ingress := &nativeApolloVideoIngress{
+		free:  make(chan uint16, nativeApolloVideoIngressSlots),
+		ready: make(chan uint16, nativeApolloVideoIngressSlots),
+	}
+	for index := range nativeApolloVideoIngressSlots {
+		ingress.free <- uint16(index)
+	}
+	return ingress
+}
+
+func (s *nativeApolloSession) drainApolloVideo(ingress *nativeApolloVideoIngress) {
+	defer close(ingress.ready)
+	for {
+		if s.mediaQuiesced.Load() {
+			return
+		}
+		select {
+		case index := <-ingress.free:
+			slot := &ingress.slots[index]
+			count, err := s.videoConn.Read(slot.packet[:])
+			receivedAt := time.Now()
+			if err != nil {
+				ingress.free <- index
+				return
+			}
+			if count > apolloMediaMaximumPacket {
+				ingress.free <- index
+				continue
+			}
+			if s.mediaQuiesced.Load() {
+				ingress.free <- index
+				return
+			}
+			s.mediaIngress.Add(1)
+			slot.count, slot.receivedAt = count, receivedAt
+			ingress.ready <- index
+		default:
+			count, err := s.videoConn.Read(ingress.scratch[:])
+			if err != nil {
+				return
+			}
+			if count > apolloMediaMaximumPacket {
+				continue
+			}
+			if s.mediaQuiesced.Load() {
+				return
+			}
+			s.mediaIngress.Add(1)
+			s.mediaDrops.Add(1)
+		}
+	}
+}
+
+func (s *nativeApolloSession) processApolloVideo(ingress *nativeApolloVideoIngress) {
+	for index := range ingress.ready {
+		slot := &ingress.slots[index]
+		if !s.mediaQuiesced.Load() {
+			shard, err := s.media.OpenVideo(slot.packet[:slot.count])
+			if err == nil {
+				payload, fecErr := s.videoFEC.Add(shard)
+				if fecErr == nil && len(payload) > 0 {
+					s.enqueueMedia(s.video, payload, slot.receivedAt)
+				}
+			}
+		}
+		slot.count, slot.receivedAt = 0, time.Time{}
+		ingress.free <- index
+	}
+}
+
 func pushLatest[T any](channel chan T, payload T) bool {
 	select {
 	case channel <- payload:

gateway/apollo_native_test.go

@@ -17,8 +17,10 @@ import (
 	"net"
 	"net/http"
 	"net/http/httptest"
+	"runtime"
 	"strconv"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -760,6 +762,34 @@ func TestNativeApolloSessionRelaysOnlyAuthenticatedEncodedUDPMedia(t *testing.T)
 	if drops := session.Telemetry().MediaDrops; drops < 2 {
 		t.Fatalf("stale FEC eviction drops = %d, want at least 2", drops)
 	}
+	beforeIngress := session.mediaIngress.Load()
+	beforeDrops := session.mediaDrops.Load()
+	if _, err := videoServer.WriteToUDP(make([]byte, apolloMediaMaximumPacket+2), videoClient.LocalAddr().(*net.UDPAddr)); err != nil {
+		t.Fatal(err)
+	}
+	marker := []byte{0xde, 0xad, 0xbe, 0xef}
+	markerPayload := make([]byte, apolloVideoShardPayloadSize)
+	markerPayload[0], markerPayload[3] = 0x01, 0x01
+	binary.LittleEndian.PutUint16(markerPayload[4:6], uint16(8+len(marker)))
+	copy(markerPayload[8:], marker)
+	markerPacket := sourceEncryptVideoRaw(t, key, sourceShapedVideoRaw(44, 103, 103, 0x07, 1, 0, 0, markerPayload), "0123456789dV")
+	if _, err := videoServer.WriteToUDP(markerPacket, videoClient.LocalAddr().(*net.UDPAddr)); err != nil {
+		t.Fatal(err)
+	}
+	select {
+	case media := <-session.Video():
+		if !bytes.Equal(media.Payload, marker) {
+			t.Fatalf("post-oversize marker relay = %x, want %x", media.Payload, marker)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("post-oversize marker was not relayed")
+	}
+	if got := session.mediaIngress.Load() - beforeIngress; got != 1 {
+		t.Fatalf("post-oversize video ingress = %d, want 1 valid marker", got)
+	}
+	if got := session.mediaDrops.Load(); got != beforeDrops {
+		t.Fatalf("post-oversize video drops = %d, want unchanged %d", got, beforeDrops)
+	}
 	terminateCtx, cancel := context.WithTimeout(context.Background(), time.Second)
 	defer cancel()
 	if err := session.Terminate(terminateCtx); err != nil {
@@ -767,6 +797,166 @@ func TestNativeApolloSessionRelaysOnlyAuthenticatedEncodedUDPMedia(t *testing.T)
 	}
 }
 
+func TestNativeApolloVideoIngressSaturationIsBoundedAndAccounted(t *testing.T) {
+	if nativeApolloVideoIngressSlots != 2048 || nativeApolloVideoReadBuffer != 2_195_456 {
+		t.Fatalf("video ingress bounds = %d slots/%d bytes, want 2048/2195456", nativeApolloVideoIngressSlots, nativeApolloVideoReadBuffer)
+	}
+	ingress := newNativeApolloVideoIngress()
+	if len(ingress.slots) != 2048 || len(ingress.free) != 2048 || cap(ingress.ready) != 2048 || len(ingress.slots[0].packet) != apolloMediaMaximumPacket+1 {
+		t.Fatalf("video ingress pool = slots:%d free:%d ready-cap:%d packet:%d",
+			len(ingress.slots), len(ingress.free), cap(ingress.ready), len(ingress.slots[0].packet))
+	}
+
+	key := []byte("0123456789abcdef")
+	session, audioServer, videoServer := newNativeApolloMediaTestSession(t, key)
+	defer audioServer.Close()
+	defer videoServer.Close()
+	blockCtx, cancelBlock := context.WithCancel(context.Background())
+	defer cancelBlock()
+	blocked := make(chan struct{})
+	release := make(chan struct{})
+	var releaseOnce sync.Once
+	releaseAEAD := func() { releaseOnce.Do(func() { close(release) }) }
+	defer releaseAEAD()
+	session.media.aead = &qualificationBlockingAEAD{
+		AEAD: session.media.aead, ctx: blockCtx, blocked: blocked, release: release,
+	}
+	go session.readUDPMedia()
+
+	packet := sourceShapedEncryptedVideoPacket(t, key, []byte{1})
+	if _, err := videoServer.WriteToUDP(packet, session.videoConn.LocalAddr().(*net.UDPAddr)); err != nil {
+		t.Fatal(err)
+	}
+	select {
+	case <-blocked:
+	case <-time.After(time.Second):
+		t.Fatal("video processor did not block")
+	}
+	waitIngress := func(want uint64) {
+		t.Helper()
+		deadline := time.NewTimer(time.Second)
+		defer deadline.Stop()
+		for session.mediaIngress.Load() < want {
+			select {
+			case <-deadline.C:
+				t.Fatalf("video ingress = %d, want at least %d", session.mediaIngress.Load(), want)
+			default:
+				runtime.Gosched()
+			}
+		}
+	}
+	waitIngress(1)
+	const overflow = 33
+	sent := uint64(1)
+	for remaining := nativeApolloVideoIngressSlots - 1 + overflow; remaining > 0; {
+		batch := min(32, remaining)
+		for range batch {
+			if _, err := videoServer.WriteToUDP([]byte{1}, session.videoConn.LocalAddr().(*net.UDPAddr)); err != nil {
+				t.Fatal(err)
+			}
+		}
+		sent += uint64(batch)
+		waitIngress(sent)
+		remaining -= batch
+	}
+	if got := session.mediaDrops.Load(); got != overflow {
+		t.Fatalf("saturated video ingress drops = %d, want %d", got, overflow)
+	}
+
+	session.quiesceMedia()
+	cancelBlock()
+	select {
+	case <-session.readDone:
+	case <-time.After(time.Second):
+		t.Fatal("video ingress workers did not stop")
+	}
+}
+
+func TestNativeApolloTerminateStopsActiveVideoDrainAndProcessor(t *testing.T) {
+	key := []byte("0123456789abcdef")
+	session, audioServer, videoServer := newNativeApolloMediaTestSession(t, key)
+	defer audioServer.Close()
+	defer videoServer.Close()
+	blockCtx, cancelBlock := context.WithCancel(context.Background())
+	t.Cleanup(cancelBlock)
+	blocked := make(chan struct{})
+	session.media.aead = &qualificationBlockingAEAD{
+		AEAD: session.media.aead, ctx: blockCtx, blocked: blocked, release: make(chan struct{}),
+	}
+	go session.readUDPMedia()
+	if _, err := videoServer.WriteToUDP(sourceShapedEncryptedVideoPacket(t, key, []byte{1}), session.videoConn.LocalAddr().(*net.UDPAddr)); err != nil {
+		t.Fatal(err)
+	}
+	select {
+	case <-blocked:
+	case <-time.After(time.Second):
+		t.Fatal("video processor did not block")
+	}
+
+	terminateCtx, cancelTerminate := context.WithTimeout(context.Background(), time.Second)
+	defer cancelTerminate()
+	terminated := make(chan error, 1)
+	go func() { terminated <- session.Terminate(terminateCtx) }()
+	cancelBlock()
+	if err := <-terminated; err != nil {
+		t.Fatalf("Terminate() error = %v", err)
+	}
+	select {
+	case <-session.readDone:
+	default:
+		t.Fatal("Terminate returned before video drain and processor stopped")
+	}
+	select {
+	case _, ok := <-session.Video():
+		if ok {
+			t.Fatal("video channel remained open after worker shutdown")
+		}
+	default:
+		t.Fatal("video channel was not closed after worker shutdown")
+	}
+}
+
+func newNativeApolloMediaTestSession(t *testing.T, key []byte) (*nativeApolloSession, *net.UDPConn, *net.UDPConn) {
+	t.Helper()
+	audioServer, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.ParseIP("127.0.0.1")})
+	if err != nil {
+		t.Fatal(err)
+	}
+	videoServer, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.ParseIP("127.0.0.1")})
+	if err != nil {
+		_ = audioServer.Close()
+		t.Fatal(err)
+	}
+	audioClient, err := net.DialUDP("udp", nil, audioServer.LocalAddr().(*net.UDPAddr))
+	if err != nil {
+		_ = audioServer.Close()
+		_ = videoServer.Close()
+		t.Fatal(err)
+	}
+	videoClient, err := net.DialUDP("udp", nil, videoServer.LocalAddr().(*net.UDPAddr))
+	if err != nil {
+		_ = audioClient.Close()
+		_ = audioServer.Close()
+		_ = videoServer.Close()
+		t.Fatal(err)
+	}
+	session := newNativeApolloSession("session-video-ingress")
+	session.media, err = newApolloMediaCodec(key, 1)
+	if err != nil {
+		_ = audioClient.Close()
+		_ = videoClient.Close()
+		_ = audioServer.Close()
+		_ = videoServer.Close()
+		t.Fatal(err)
+	}
+	session.audioConn, session.videoConn = audioClient, videoClient
+	t.Cleanup(func() {
+		_ = audioClient.Close()
+		_ = videoClient.Close()
+	})
+	return session, audioServer, videoServer
+}
+
 func TestNativeApolloTerminateReleasesPressedProviderInput(t *testing.T) {
 	server, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.ParseIP("127.0.0.1")})
 	if err != nil {

gateway/qualification_contract_test.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"compress/gzip"
 	"context"
+	"crypto/cipher"
 	"encoding/binary"
 	"errors"
 	"fmt"
@@ -15,10 +16,35 @@ import (
 	"runtime"
 	"strconv"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 )
 
+type qualificationBlockingAEAD struct {
+	cipher.AEAD
+	ctx     context.Context
+	blocked chan struct{}
+	release chan struct{}
+	once    sync.Once
+	waitErr error
+}
+
+func (a *qualificationBlockingAEAD) Open(dst, nonce, ciphertext, additionalData []byte) ([]byte, error) {
+	a.once.Do(func() {
+		close(a.blocked)
+		select {
+		case <-a.release:
+		case <-a.ctx.Done():
+			a.waitErr = a.ctx.Err()
+		}
+	})
+	if a.waitErr != nil {
+		return nil, a.waitErr
+	}
+	return a.AEAD.Open(dst, nonce, ciphertext, additionalData)
+}
+
 func TestQualificationCatalogMatchesSection7(t *testing.T) {
 	media := qualificationMediaProfiles()
 	if len(media) != 3 {
@@ -757,6 +783,94 @@ func TestQualificationLateSourceBatchDoesNotCollapseThroughPublicPath(t *testing
 		path.session.mediaRecovered.Load(), path.session.mediaEnqueued.Load())
 }
 
+func TestQualificationBlockedVideoAEADDoesNotBlockProviderIngress(t *testing.T) {
+	profile := qualificationMediaProfiles()[2]
+	blockCtx, cancelBlock := context.WithCancel(context.Background())
+	defer cancelBlock()
+	blocked := make(chan struct{})
+	release := make(chan struct{})
+	var releaseOnce sync.Once
+	releaseAEAD := func() { releaseOnce.Do(func() { close(release) }) }
+	t.Cleanup(releaseAEAD)
+	native := NewNativeApolloBackend()
+	native.configureMedia = func(media *apolloMediaCodec) {
+		media.aead = &qualificationBlockingAEAD{
+			AEAD: media.aead, ctx: blockCtx, blocked: blocked, release: release,
+		}
+	}
+	path := newQualificationPathWithNativeBackend(t, profile, profile.BitrateKbps, nil, native)
+	t.Cleanup(func() {
+		path.Close()
+		select {
+		case <-path.session.readDone:
+		case <-time.After(time.Second):
+			t.Error("native media workers did not stop during qualification cleanup")
+		}
+	})
+
+	payload := qualificationFramePayload(profile, 0)
+	packets := qualificationSourceVideoPackets(t, path.key, 1, payload)
+	if len(payload) != 666_664 || len(packets) != 662 {
+		t.Fatalf("4K60 keyframe = %d bytes/%d shards, want 666664/662", len(payload), len(packets))
+	}
+
+	beforeMetrics := path.server.Metrics()
+	beforeIngress := path.session.mediaIngress.Load()
+	beforeRecovered := path.session.mediaRecovered.Load()
+	beforeEnqueued := path.session.mediaEnqueued.Load()
+	beforePacer := path.server.pacer.reservations.Load()
+	if _, err := path.emit(t, payload); err != nil {
+		t.Fatal(err)
+	}
+	select {
+	case <-blocked:
+	case <-time.After(time.Second):
+		t.Fatal("video AEAD did not block")
+	}
+	if sent := path.fixture.sentPackets.Load(); sent != 662 {
+		t.Fatalf("fixture sent packets = %d, want 662", sent)
+	}
+	deadline := time.NewTimer(250 * time.Millisecond)
+	defer deadline.Stop()
+	for path.session.mediaIngress.Load()-beforeIngress != 662 {
+		select {
+		case <-deadline.C:
+			t.Fatalf("blocked-AEAD ingress = %d, want 662 after 662 successful fixture writes",
+				path.session.mediaIngress.Load()-beforeIngress)
+		default:
+			runtime.Gosched()
+		}
+	}
+
+	releaseAEAD()
+	recovered, err := path.receivePayload(context.Background())
+	if err != nil {
+		t.Fatal(err)
+	}
+	afterMetrics := path.server.Metrics()
+	stageDeadline := time.Now().Add(2 * time.Second)
+	for (path.session.mediaRecovered.Load() <= beforeRecovered ||
+		path.session.mediaEnqueued.Load() <= beforeEnqueued ||
+		afterMetrics.ProcessingSamples <= beforeMetrics.ProcessingSamples ||
+		afterMetrics.MediaPackets <= beforeMetrics.MediaPackets) && time.Now().Before(stageDeadline) {
+		runtime.Gosched()
+		afterMetrics = path.server.Metrics()
+	}
+	if path.backend.setups.Load() != 1 || path.backend.opens.Load() != 1 ||
+		path.session.mediaRecovered.Load()-beforeRecovered != 1 ||
+		path.session.mediaEnqueued.Load()-beforeEnqueued != 1 ||
+		afterMetrics.ProcessingSamples <= beforeMetrics.ProcessingSamples ||
+		path.server.pacer.reservations.Load() <= beforePacer ||
+		afterMetrics.MediaPackets <= beforeMetrics.MediaPackets || !bytes.Equal(recovered, payload) {
+		t.Fatalf("blocked-AEAD public path: setup=%d open=%d ingress=%d recovered=%d enqueued=%d processing=%d pacer=%d media=%d payload=%t",
+			path.backend.setups.Load(), path.backend.opens.Load(), path.session.mediaIngress.Load()-beforeIngress,
+			path.session.mediaRecovered.Load()-beforeRecovered, path.session.mediaEnqueued.Load()-beforeEnqueued,
+			afterMetrics.ProcessingSamples-beforeMetrics.ProcessingSamples,
+			path.server.pacer.reservations.Load()-beforePacer, afterMetrics.MediaPackets-beforeMetrics.MediaPackets,
+			bytes.Equal(recovered, payload))
+	}
+}
+
 func TestQualificationProcessingRetainsProviderIngressDiagnostics(t *testing.T) {
 	profile := qualificationMediaProfiles()[2]
 	path := newQualificationProcessingPath(t, profile, profile.BitrateKbps)

gateway/qualification_harness_test.go

@@ -1055,13 +1055,17 @@ func newQualificationImpairedPath(t *testing.T, profile qualificationMediaProfil
 }
 
 func newQualificationPathWithImpairment(t *testing.T, profile qualificationMediaProfile, pacerKbps int64, impairment *qualificationImpairmentProfile) *qualificationPath {
+	return newQualificationPathWithNativeBackend(t, profile, pacerKbps, impairment, NewNativeApolloBackend())
+}
+
+func newQualificationPathWithNativeBackend(t *testing.T, profile qualificationMediaProfile, pacerKbps int64, impairment *qualificationImpairmentProfile, native *NativeApolloBackend) *qualificationPath {
 	t.Helper()
 	serverTLS, clientTLS := testTLS(t)
 	fixture := newQualificationApolloFixture(t, serverTLS, clientTLS, "qualification-session", profile)
 	if impairment != nil {
 		fixture.setControlImpairment(*impairment)
 	}
-	backend := &qualificationTracingBackend{native: NewNativeApolloBackend()}
+	backend := &qualificationTracingBackend{native: native}
 	provider := NewApolloAdapter(backend, ProviderIdentity{})
 	authority := protocol.SessionAuthority{
 		Version: "1", SessionID: "qualification-session", GatewayID: "gateway-1",

openspec/changes/truthful-fixed-frame-qualification/design.md

@@ -10,6 +10,15 @@ The retained v6 qualification run passed its then-current checks but is supersed
 
 The later Darwin non-sustained pre-CI invocation was not green and was not retried. Its 1440p120 profile delivered the exact 6,250,000 bytes in 120 frames plus all 6,483 source and warm-up shards with zero drops, but measured 46,973.13 kbps over an implied approximately 1.0644383 seconds and failed the 5% throughput gate. Private Linux full verification/artifact retention and the replacement v8 normative run remain open.
 
+Private Linux run 125 at the frozen v8 harness head is retained as failed evidence. Its exact 33-datagram gap between successful fixture writes and production `MediaIngress` equaled the Linux socket's 33 measured kernel UDP drops. The complete-frame queue, fair pacer, QUIC fragmentation, and public decoder were downstream and did not account for the loss.
+
+The ingress repair follows reviewed behavior rather than copying implementation source:
+
+- Apollo `adc5c5a0bd80831ce495434bb16aee2cd4175fb8`, GPL-3.0, `src/stream.cpp:1463-1474,1573-1627`, supplies the 80%-of-1-Gbps raw-block pacing, 64-KiB/64-packet batch cap, and cross-frame send schedule used by the fixture.
+- Moonlight common-C pin `2ea47752c3051d72a64bcca190024e8b354fa1ef`, GPL-3.0, `src/VideoStream.c:28-35,331-333` and `src/PlatformSockets.c:364-405`, supplies the reviewed 2,048-video-packet receive-buffer request and dedicated receive-thread behavior. The cited `VideoStream.c` blob is byte-identical at the local standalone `703a06946861ff82cd33e5e13c59c1b017f7ded9` checkout.
+
+The native provider therefore requests `2,048 * 1,072 = 2,195,456` bytes with `SetReadBuffer()` on the connected video socket immediately after dialing it. A setter error aborts setup; an OS-imposed cap is accepted without privilege or getter dependence. A dedicated drain owns a fixed 2,048-slot FIFO pool. Every slot is 1,433 bytes (`apolloMediaMaximumPacket + 1`), so oversized datagrams remain observably invalid rather than being truncated into the accepted range; packet storage is 2,934,784 bytes (about 2.80 MiB) plus fixed index and timestamp metadata. The existing single decrypt/FEC processor consumes those slots. When every slot is occupied, the drain keeps reading into one fixed 1,433-byte scratch buffer and counts each accepted-size discard in both ingress and drop telemetry; oversized datagrams retain the existing rejection semantics. Socket close cancels the blocking read, and media channels close only after the unchanged audio reader, video drain, and video processor exit. Audio and control behavior are unchanged.
+
 ## Goals / Non-Goals
 
 **Goals:**
@@ -27,8 +36,10 @@ The later Darwin non-sustained pre-CI invocation was not green and was not retri
 - Derive bytes per fixed interval from bitrate and FPS, distribute integer remainder deterministically, and shift bounded bytes into periodic keyframes while keeping the interval total exact.
 - Carry a deterministic frame index/pattern only in the generated payload bytes; no codec semantics are claimed.
 - Keep the existing path/impairment/resource driver and change its unit from datagram payload to complete frame.
+- Keep video decrypt/FEC single-threaded; only the bounded connected-socket drain is separated so crypto stalls cannot become unexplained kernel loss.
 
 ## Risks / Trade-offs
 
 - [Keyframes can exceed queue budget] → use the reviewed 1 MiB frame ceiling and production byte-bound queue.
 - [Short smoke windows have rounding effects] → assert exact generated totals and report measured duration separately from normative ten-minute gates.
+- [A stalled video processor exhausts the user-space pool] → keep draining into one fixed scratch buffer and attribute accepted-size overflow to existing ingress/drop counters rather than kernel loss or unbounded allocation.

openspec/changes/truthful-fixed-frame-qualification/proposal.md

@@ -4,11 +4,14 @@ The existing fixed-profile harness treats each 1,179-byte datagram as an encoded
 
 The v6 complete-frame fixture subsequently exposed a source-fidelity defect on ordinary Linux runners: it emitted every UDP shard in one tight loop, unlike pinned Apollo's bounded intra-frame rate and batch schedule. The affected v6 qualification evidence remains retained but is superseded for candidate-readiness purposes.
 
+Private Linux run 125 then demonstrated a separate production-ingress defect after source pacing was corrected: 33 successful fixture writes missing from `MediaIngress` matched 33 measured kernel UDP drops while decrypt/FEC, queue, pacer, QUIC, and client counters remained downstream of the shortfall.
+
 ## What Changes
 
 - Generate deterministic variable-size encoded frame units at the named frame rates and target bitrates, including bounded keyframes.
 - Traverse native Apollo recovery, production queues, the production pacer, QUIC framing, and independent reassembly.
 - Assert frame count/rate, bitrate, exact bytes and boundaries, clean loss attribution, latency, and resource bounds.
+- Decouple native video socket draining from the single decrypt/FEC processor with a fixed provider-scoped receive pool and request the source-backed video receive-buffer size.
 - Keep short smoke tests separate and leave all prior normative artifacts unchanged.
 
 ## Capabilities
@@ -23,4 +26,4 @@ None.
 
 ## Impact
 
-The existing qualification harness and canonical qualification specification only. No codec operation, production dependency, or normative run before immutable consumer publication. Requirements: P3C-002, P3C-008, P3C-029, P3C-030, P3C-033, VER-009, VER-010, OPS-015.
+The qualification harness, its canonical specification, and native Apollo video ingress in `gateway/apollo_native.go`. Downstream complete-frame queues, audio/control ingress, codec/FEC formats, pacing, QUIC, dependencies, and public interfaces remain unchanged. No codec operation or normative rerun is included. Requirements: P3C-002, P3C-008, P3C-029, P3C-030, P3C-033, VER-009, VER-010, OPS-015.

openspec/changes/truthful-fixed-frame-qualification/specs/gateway-qualification/spec.md

@@ -5,6 +5,8 @@ The qualification harness SHALL drive pinned-mTLS Apollo management, encrypted R
 
 Within each complete frame the source fixture SHALL reproduce pinned Apollo's source schedule by deriving packets per millisecond from the raw UDP block size at 80% of 1 Gbps, bounding each source batch to the smaller of 64 KiB or 64 packets, capturing the monotonic batch start immediately after the first successful shard write, scheduling the next batch no earlier than that start plus the current batch's raw-block serialization interval, carrying that schedule across frames, and making pacing waits context-cancellable. A delayed batch SHALL remain late rather than trigger an overdue catch-up burst.
 
+Native Apollo video ingress SHALL request a 2,195,456-byte socket receive buffer before media ping or worker startup and SHALL drain the connected video socket into a fixed FIFO pool of exactly 2,048 slots before the existing single decrypt/FEC processor. Each slot and the saturation scratch buffer SHALL be `apolloMediaMaximumPacket + 1` bytes so oversized datagrams remain rejected. A full pool SHALL NOT stop socket draining: each successfully read accepted-size discard SHALL increment both media-ingress and media-drop telemetry without allocation, while oversized reads SHALL retain the existing rejection accounting. Socket closure SHALL cancel the video read, and video/audio channels SHALL close only after the audio reader, video drain, and video processor exit. Audio and control ingress SHALL remain unchanged.
+
 #### Scenario: Healthy fixed profile
 - **WHEN** a frozen candidate runs one fixed profile for the normative duration in the isolated qualification command
 - **THEN** the harness emits compressed raw frame/path and gateway-process resource samples plus a summary tied to the exact command, CPU scope, timing-overhead method, topology, source commit, immutable Protocol version, environment, and payload hash
@@ -16,3 +18,11 @@ Within each complete frame the source fixture SHALL reproduce pinned Apollo's so
 #### Scenario: Source-shaped Apollo pacing is preserved
 - **WHEN** the fixture emits 1,072-byte encrypted video shards with 1,040-byte raw blocks for consecutive complete frames
 - **THEN** it uses 96 packets per millisecond, batches at most 63 shards, records each batch after its first successful shard write, starts each later batch no earlier than the prior batch's raw serialization interval, carries the schedule into the following frame, and emits no shard after a cancelled pacing wait
+
+#### Scenario: Video crypto processing stalls
+- **WHEN** the first video AEAD operation is blocked while a 662-shard keyframe arrives
+- **THEN** all 662 successful connected-socket reads reach media-ingress accounting before processing resumes, and after release the exact complete frame traverses recovery, the bounded production queue, pacer, QUIC, and independent reassembly
+
+#### Scenario: Video ingress pool saturates
+- **WHEN** all 2,048 fixed video slots are occupied
+- **THEN** accepted-size datagrams are deliberately discarded through the fixed scratch buffer and counted as ingress plus drops, oversized datagrams remain rejected, and cancellation closes every media worker without a race or leak

openspec/changes/truthful-fixed-frame-qualification/tasks.md

@@ -23,5 +23,11 @@
 - [x] 5.1 Retain the v6 qualification attempt and mark its passing result superseded by the tight-loop source defect
 - [x] 5.2 Implement v8 post-first-write, non-collapsing complete-frame UDP pacing with persistent cross-frame carry and verify the focused, race, short-resource, and cross-platform compile checks that passed
 - [x] 5.3 Preserve private Linux runs 123 and 124 as failed evidence, the passing local v8 Darwin sustained run as non-Linux and non-normative, and the un-retried failed Darwin non-sustained pre-CI invocation with its exact throughput evidence
-- [ ] 5.4 Run private Linux full verification and retain deterministic Linux artifacts for the frozen v8 harness descendant
+- [ ] 5.4 Run private Linux full verification and retain deterministic Linux artifacts for the frozen v8 harness and video-ingress descendant
 - [ ] 5.5 Run one separately approved replacement v8 normative Section 7 qualification
+
+## 6. Native video ingress remediation
+
+- [x] 6.1 Preserve run 125 and reproduce its pre-decrypt shortfall with a public 662-shard blocked-AEAD regression
+- [x] 6.2 Add the video-only 2,195,456-byte socket-buffer request, fixed 2,048-slot drain, single processor, overflow accounting, and bounded cancellation tests
+- [ ] 6.3 Freeze the reviewed production repair through the still-open private Linux full-verification and artifact gate before any replacement normative run