test(gateway): pace qualification video source

gateway/qualification_contract_test.go

@@ -7,6 +7,7 @@ import (
 	"encoding/binary"
 	"errors"
 	"io"
+	"net"
 	"os"
 	"path/filepath"
 	"reflect"
@@ -67,6 +68,72 @@ func TestQualificationRecordsLinkedToolVersions(t *testing.T) {
 	}
 }
 
+func TestQualificationApolloFixturePacesSourceShapedVideo(t *testing.T) {
+	key := bytes.Repeat([]byte{0x3c}, 16)
+	encoded := make([]byte, 1000*apolloVideoShardPayloadSize-8)
+	packets := qualificationSourceVideoPackets(t, key, 1, encoded)
+	if len(packets) != 1000 || len(packets[0]) != 1072 {
+		t.Fatalf("source vector = %d packets of %d bytes, want 1000 packets of 1072 bytes", len(packets), len(packets[0]))
+	}
+	packetsPerMillisecond, batchSize := qualificationApolloVideoPacing(len(packets[0]))
+	if packetsPerMillisecond != 93 || batchSize != 61 {
+		t.Fatalf("Apollo pacing vector = %d packets/ms, batch %d; want 93 and 61", packetsPerMillisecond, batchSize)
+	}
+
+	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)
+	drained := make(chan struct{})
+	go func() {
+		defer close(drained)
+		buffer := make([]byte, 2048)
+		for range len(packets) + 1 {
+			if _, _, readErr := receiver.ReadFromUDP(buffer); readErr != nil {
+				return
+			}
+		}
+	}()
+
+	started := time.Now()
+	if err := fixture.sendVideo(context.Background(), packets); err != nil {
+		t.Fatal(err)
+	}
+	if err := fixture.sendVideo(context.Background(), packets[:1]); err != nil {
+		t.Fatal(err)
+	}
+	elapsed := time.Since(started)
+	wantCarry := 10 * time.Millisecond // floor(1000 / 93) ms at Apollo's pinned 80%-of-1-Gbps rate.
+	if elapsed < wantCarry {
+		t.Fatalf("source fixture sent the next frame after %s, before Apollo pacing carry %s", elapsed, wantCarry)
+	}
+	select {
+	case <-drained:
+	case <-time.After(time.Second):
+		t.Fatal("source-shaped UDP receiver did not drain the fixed vector")
+	}
+	fixture.videoNext = time.Now().Add(time.Second)
+	beforeCancel := fixture.sentPackets.Load()
+	cancelled, cancel := context.WithCancel(context.Background())
+	cancel()
+	if err := fixture.sendVideo(cancelled, packets[:1]); !errors.Is(err, context.Canceled) {
+		t.Fatalf("cancelled pacing wait returned %v, want context.Canceled", err)
+	}
+	if fixture.sentPackets.Load() != beforeCancel {
+		t.Fatal("cancelled pacing wait emitted a UDP shard")
+	}
+}
+
 func TestQualificationOutputAndStatisticsFailClosed(t *testing.T) {
 	if err := validateQualificationOutputDir("relative/evidence"); err == nil {
 		t.Fatal("relative evidence directory was accepted")