test(gateway): freeze v8 qualification harness

gateway/qualification_contract_test.go

@@ -6,11 +6,13 @@ import (
 	"context"
 	"encoding/binary"
 	"errors"
+	"fmt"
 	"io"
 	"net"
 	"os"
 	"path/filepath"
 	"reflect"
+	"runtime"
 	"strconv"
 	"strings"
 	"testing"
@@ -121,10 +123,10 @@ func TestQualificationApolloFixturePacesSourceShapedVideo(t *testing.T) {
 	}()
 
 	started = time.Now()
-	if err := fixture.sendVideo(context.Background(), packets); err != nil {
+	if err := fixture.sendVideo(context.Background(), 1, packets); err != nil {
 		t.Fatal(err)
 	}
-	if err := fixture.sendVideo(context.Background(), packets[:1]); err != nil {
+	if err := fixture.sendVideo(context.Background(), 2, packets[:1]); err != nil {
 		t.Fatal(err)
 	}
 	elapsed := time.Since(started)
@@ -154,7 +156,7 @@ func TestQualificationApolloFixturePacesSourceShapedVideo(t *testing.T) {
 	beforeCancel := fixture.sentPackets.Load()
 	cancelled, cancel := context.WithCancel(context.Background())
 	cancel()
-	if err := fixture.sendVideo(cancelled, packets[:1]); !errors.Is(err, context.Canceled) {
+	if err := fixture.sendVideo(cancelled, 3, packets[:1]); !errors.Is(err, context.Canceled) {
 		t.Fatalf("cancelled pacing wait returned %v, want context.Canceled", err)
 	}
 	if fixture.sentPackets.Load() != beforeCancel {
@@ -690,3 +692,200 @@ func TestQualificationCarriesCompleteLargeFramesThroughPublicPath(t *testing.T)
 		}
 	}
 }
+
+func TestQualificationLateSourceBatchDoesNotCollapseThroughPublicPath(t *testing.T) {
+	profile := qualificationMediaProfiles()[2]
+	path := newQualificationPath(t, profile, profile.BitrateKbps)
+	defer path.Close()
+
+	const shardCount = 662
+	payload := qualificationFramePayload(profile, 0)
+	if len(payload) != 666_664 {
+		t.Fatalf("4K60 keyframe bytes = %d, want 666664", len(payload))
+	}
+	packets := qualificationSourceVideoPackets(t, path.key, 1, payload)
+	if len(packets) != shardCount {
+		t.Fatalf("source packet count = %d, want %d", len(packets), shardCount)
+	}
+
+	var batchIndices []int
+	var batchStarts []time.Time
+	var firstBatch time.Time
+	path.fixture.observeVideoBatch = func(index int, at time.Time) {
+		if firstBatch.IsZero() {
+			firstBatch = at
+		}
+		batchIndices = append(batchIndices, index)
+		batchStarts = append(batchStarts, at)
+	}
+	stalled := false
+	path.fixture.beforeVideoBatch = func(ctx context.Context, index int) error {
+		if index != 63 || stalled {
+			return nil
+		}
+		stalled = true
+		return qualificationWaitContext(ctx, firstBatch.Add(5*time.Millisecond))
+	}
+
+	trace, _, err := path.traverse(t, payload)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !trace.NativeSetup || !trace.NativeOpen || !trace.NativeUDPIngress || !trace.ApolloRecovered ||
+		!trace.ProductionQueue || !trace.ProductionMediaLoop || !trace.ProductionPacer ||
+		!trace.VerseQUIC || !trace.PublicClientDecode || !trace.PayloadPreserved {
+		t.Fatalf("late-batch path skipped production stages: %#v", trace)
+	}
+	if got := path.sourceUDP.Load(); got != shardCount {
+		t.Fatalf("source UDP count = %d, want %d", got, shardCount)
+	}
+	if len(batchStarts) != (shardCount+62)/63 {
+		t.Fatalf("batch count = %d, want %d", len(batchStarts), (shardCount+62)/63)
+	}
+	packetsPerMillisecond, _ := qualificationApolloVideoPacing(apolloVideoRawPacketSize)
+	for index := 1; index < len(batchStarts); index++ {
+		previousPackets := min(63, shardCount-batchIndices[index-1])
+		minimum := qualificationApolloVideoOffset(previousPackets, packetsPerMillisecond)
+		actual := batchStarts[index].Sub(batchStarts[index-1])
+		if actual < minimum {
+			t.Fatalf("batch %d at packet %d started %s after packet %d, before raw serialization interval %s; starts=%v",
+				index, batchIndices[index], actual, batchIndices[index-1], minimum, batchStarts)
+		}
+	}
+	t.Logf("late-batch public path: packets=%d starts=%v ingress=%d recovered=%d enqueued=%d",
+		path.sourceUDP.Load(), batchStarts, path.session.mediaIngress.Load(),
+		path.session.mediaRecovered.Load(), path.session.mediaEnqueued.Load())
+}
+
+func TestQualificationProcessingRetainsProviderIngressDiagnostics(t *testing.T) {
+	profile := qualificationMediaProfiles()[2]
+	path := newQualificationProcessingPath(t, profile, profile.BitrateKbps)
+	defer path.Close()
+
+	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))
+	}
+	before, err := path.process.snapshot()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := path.emit(t, payload); err != nil {
+		t.Fatal(err)
+	}
+	recovered, err := path.receivePayload(context.Background())
+	if err != nil {
+		t.Fatal(err)
+	}
+	diagnostics := path.processingDiagnostics(1)
+	if diagnostics.SnapshotError != "" {
+		t.Fatal(diagnostics.SnapshotError)
+	}
+	after := diagnostics.Gateway
+	batches := diagnostics.BatchHistory
+	if diagnostics.ProcessedFrames != 1 || diagnostics.SourceFrames != 1 ||
+		diagnostics.ExpectedWritesThroughLastFrame != 662 ||
+		diagnostics.FixtureSentPackets != 662 || diagnostics.SourceUDP != 662 ||
+		after.MediaIngress-before.MediaIngress != 662 ||
+		after.MediaRecovered-before.MediaRecovered != 1 ||
+		after.MediaEnqueued-before.MediaEnqueued != 1 || !bytes.Equal(recovered, payload) {
+		t.Fatalf("provider ingress diagnostics: sent=%d source=%d ingress=%d recovered=%d enqueued=%d payload=%t",
+			diagnostics.FixtureSentPackets, diagnostics.SourceUDP, after.MediaIngress-before.MediaIngress,
+			after.MediaRecovered-before.MediaRecovered, after.MediaEnqueued-before.MediaEnqueued,
+			bytes.Equal(recovered, payload))
+	}
+	if len(batches) != 11 {
+		t.Fatalf("provider batch history = %d entries, want 11", len(batches))
+	}
+	for index := range batches {
+		wantPacket := index * 63
+		if batches[index].ProviderFrame != 1 || batches[index].PacketWithinFrame != wantPacket ||
+			batches[index].SourcePacket != uint64(wantPacket) {
+			t.Fatalf("provider batch %d = %#v, want frame 1 source/within %d", index, batches[index], wantPacket)
+		}
+		if index > 0 {
+			previousPackets := min(63, 662-batches[index-1].PacketWithinFrame)
+			minimum := qualificationApolloVideoOffset(previousPackets, 96)
+			if batches[index].StartedAfter-batches[index-1].StartedAfter < minimum {
+				t.Fatalf("provider batch %d collapsed: %#v", index, batches)
+			}
+		}
+	}
+	if (runtime.GOOS == "darwin" || runtime.GOOS == "linux") && !after.VideoReceiveBufferAvailable {
+		t.Fatal("provider video SO_RCVBUF unavailable on a supported diagnostic platform")
+	}
+	if after.VideoReceiveBufferAvailable && after.VideoReceiveBuffer <= 0 {
+		t.Fatalf("provider video SO_RCVBUF = %d available=%t, want measured >0",
+			after.VideoReceiveBuffer, after.VideoReceiveBufferAvailable)
+	}
+	if after.KernelDropsAvailable && after.KernelDrops != 0 {
+		t.Fatalf("provider UDP kernel drops = %d, want 0", after.KernelDrops)
+	}
+	t.Logf("provider ingress diagnostics: %#v", diagnostics)
+}
+
+func TestQualificationActualEmissionDoesNotCatchUpAfterPreWritePause(t *testing.T) {
+	key := bytes.Repeat([]byte{0x4d}, 16)
+	packets := qualificationSourceVideoPackets(t, key, 1, make([]byte, 189*apolloVideoShardPayloadSize-8))
+	receiver, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.ParseIP("127.0.0.1")})
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer receiver.Close()
+	sender, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.ParseIP("127.0.0.1")})
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer sender.Close()
+	fixture := &qualificationApolloFixture{video: sender, failures: make(chan error, 1)}
+	remote := *receiver.LocalAddr().(*net.UDPAddr)
+	fixture.videoRemote.Store(&remote)
+	stalled := false
+	fixture.beforeVideoFirstWrite = func(ctx context.Context, packetWithinFrame int) error {
+		if packetWithinFrame != 63 || stalled {
+			return nil
+		}
+		stalled = true
+		return qualificationWaitContext(ctx, time.Now().Add(5*time.Millisecond))
+	}
+	var emitted []time.Time
+	fixture.observeVideoBatch = func(_ int, at time.Time) { emitted = append(emitted, at) }
+	drained := make(chan struct{})
+	go func() {
+		defer close(drained)
+		buffer := make([]byte, 2048)
+		for range len(packets) {
+			if _, _, readErr := receiver.ReadFromUDP(buffer); readErr != nil {
+				return
+			}
+		}
+	}()
+	if err := fixture.sendVideo(context.Background(), 1, packets); err != nil {
+		t.Fatal(err)
+	}
+	select {
+	case <-drained:
+	case <-time.After(time.Second):
+		t.Fatal("actual-emission receiver did not drain")
+	}
+	if len(emitted) != 3 {
+		t.Fatalf("actual emitted batch starts = %d, want 3", len(emitted))
+	}
+	minimum := qualificationApolloVideoOffset(63, 96)
+	if actual := emitted[2].Sub(emitted[1]); actual < minimum {
+		t.Fatalf("post-write batch 126 started %s after batch 63, before %s; emitted=%v", actual, minimum, emitted)
+	}
+	t.Logf("post-write emitted batch starts: %v", emitted)
+}
+
+func TestQualificationProcessingDiagnosticsFormatUsesMonotonicOffsets(t *testing.T) {
+	diagnostics := qualificationProcessingDiagnostics{
+		BatchHistory: []qualificationVideoBatchObservation{{StartedAfter: 1234567 * time.Nanosecond}},
+	}
+	failureText := fmt.Errorf("qualification failed; diagnostics=%#v", diagnostics).Error()
+	if !strings.Contains(failureText, "StartedAfter") || !strings.Contains(failureText, "1234567") ||
+		strings.Contains(failureText, "time.Date(") {
+		t.Fatalf("processing failure text does not retain explicit monotonic offsets: %s", failureText)
+	}
+}