fix(gateway): synchronize qualification stage evidence

gateway/qualification_contract_test.go

@@ -2,6 +2,7 @@ package gateway
 
 import (
 	"compress/gzip"
+	"encoding/binary"
 	"io"
 	"os"
 	"path/filepath"
@@ -142,6 +143,29 @@ func TestQualificationProcessingPreservesPayload(t *testing.T) {
 	}
 }
 
+func TestQualificationRepeatedTraversalTracksEveryProductionStage(t *testing.T) {
+	profile := qualificationMediaProfiles()[0]
+	pacerKbps := (profile.BitrateKbps*int64(profile.PacketBytes+frameHeaderSize) + int64(profile.PacketBytes) - 1) / int64(profile.PacketBytes)
+	path := newQualificationPath(t, profile, pacerKbps)
+	payload := qualificationPayload(profile)
+	if err := runQualificationWarmup(t, path, profile, payload); err != nil {
+		t.Fatal(err)
+	}
+	for index := 0; index < 10_000; index++ {
+		current := append([]byte(nil), payload...)
+		binary.BigEndian.PutUint32(current[len(current)-4:], uint32(index))
+		trace, _, err := path.traverse(t, current)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if !trace.NativeUDPIngress || !trace.ApolloRecovered || !trace.ProductionQueue ||
+			!trace.ProductionMediaLoop || !trace.ProductionPacer || !trace.VerseQUIC ||
+			!trace.PublicClientDecode || !trace.PayloadPreserved {
+			t.Fatalf("traversal %d missed a production stage: %#v", index, trace)
+		}
+	}
+}
+
 func TestQualificationImpairmentIsDeterministicAndBounded(t *testing.T) {
 	profile := qualificationImpairmentProfiles()[3]
 	first, err := runQualificationImpairment(t, profile, qualificationMediaProfiles()[0], 1000, filepath.Join(t.TempDir(), "first.csv.gz"))

gateway/qualification_harness_test.go

@@ -823,6 +823,7 @@ func (p *qualificationPath) traverse(t *testing.T, payload []byte) (qualificatio
 	beforeIngress := p.session.mediaIngress.Load()
 	beforeRecovered := p.session.mediaRecovered.Load()
 	beforeEnqueued := p.session.mediaEnqueued.Load()
+	beforePacer := p.server.pacer.reservations.Load()
 	trace, err := p.emit(t, payload)
 	if err != nil {
 		return trace, 0, err
@@ -832,11 +833,17 @@ func (p *qualificationPath) traverse(t *testing.T, payload []byte) (qualificatio
 		return trace, 0, err
 	}
 	afterMetrics := p.server.Metrics()
+	deadline := time.Now().Add(2 * time.Second)
+	for (afterMetrics.ProcessingSamples < beforeMetrics.ProcessingSamples+1 ||
+		afterMetrics.MediaPackets <= beforeMetrics.MediaPackets) && time.Now().Before(deadline) {
+		runtime.Gosched()
+		afterMetrics = p.server.Metrics()
+	}
 	trace.NativeUDPIngress = p.session.mediaIngress.Load() > beforeIngress
 	trace.ApolloRecovered = p.session.mediaRecovered.Load() > beforeRecovered
 	trace.ProductionQueue = p.session.mediaEnqueued.Load() > beforeEnqueued
 	trace.ProductionMediaLoop = afterMetrics.ProcessingSamples == beforeMetrics.ProcessingSamples+1
-	trace.ProductionPacer = afterMetrics.PacingDelayNanos > beforeMetrics.PacingDelayNanos
+	trace.ProductionPacer = p.server.pacer.reservations.Load() > beforePacer
 	trace.VerseQUIC = afterMetrics.MediaPackets > beforeMetrics.MediaPackets
 	trace.PublicClientDecode = true
 	trace.PayloadPreserved = bytes.Equal(recovered, payload)

gateway/telemetry.go

@@ -125,6 +125,7 @@ type fairPacer struct {
 	mu             sync.Mutex
 	bytesPerSecond int64
 	flows          map[string]fairPacerFlow
+	reservations   atomic.Uint64
 }
 
 type fairPacerFlow struct {
@@ -194,6 +195,7 @@ func (p *fairPacer) reserveAt(now time.Time, flow string, bytes int) time.Time {
 
 func (p *fairPacer) wait(ctx context.Context, flow string, bytes int) error {
 	target := p.reserveAt(time.Now(), flow, bytes)
+	p.reservations.Add(1)
 	if delay := time.Until(target); delay > 0 {
 		timer := time.NewTimer(delay)
 		defer timer.Stop()