fix(gateway): qualify concurrent fair allocation

gateway/gateway_test.go

@@ -847,7 +847,9 @@ func TestProviderDisconnectEndsPublicGatewaySessionReconnectable(t *testing.T) {
 func TestProviderTerminalCleanupFailureReportsCleanupPending(t *testing.T) {
 	h := newGatewayTransportHarnessWithoutClipboard(t)
 	h.drainInitialMedia(t)
+	h.session.mu.Lock()
 	h.session.failure = FakeFailureTerminationTimeout
+	h.session.mu.Unlock()
 	h.session.EmitEvent(ProviderEvent{Kind: ProviderEventTerminated, Payload: []byte{1, 2, 3, 4}})
 
 	h.waitReleased(t)

gateway/provider.go

@@ -449,11 +449,12 @@ type fakeSession struct {
 }
 
 func (s *fakeSession) Ready(ctx context.Context) error {
+	s.mu.Lock()
 	if s.failure == FakeFailureReadinessTimeout {
+		s.mu.Unlock()
 		<-ctx.Done()
 		return ctx.Err()
 	}
-	s.mu.Lock()
 	defer s.mu.Unlock()
 	if s.state.State == ProviderStateDisconnected {
 		return ErrProviderDisconnected
@@ -585,7 +586,10 @@ func (s *fakeSession) ReleaseAll(_ context.Context) error {
 }
 
 func (s *fakeSession) Terminate(ctx context.Context) error {
-	if s.failure == FakeFailureTerminationTimeout {
+	s.mu.Lock()
+	terminationTimeout := s.failure == FakeFailureTerminationTimeout
+	s.mu.Unlock()
+	if terminationTimeout {
 		<-ctx.Done()
 		s.mu.Lock()
 		s.state.State = ProviderStateCleanup

gateway/qualification_contract_test.go

@@ -224,6 +224,21 @@ func TestQualificationUsesPublicQUICAndProductionPacer(t *testing.T) {
 	}
 }
 
+func TestQualificationCapacityStepsConvergeWithinTenSeconds(t *testing.T) {
+	evidence, err := qualificationPacerEvidence(t, filepath.Join(t.TempDir(), "fairness.csv.gz"), 2*time.Second, 10*time.Second)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(evidence.CapacitySteps) != 2 {
+		t.Fatalf("capacity steps = %#v", evidence.CapacitySteps)
+	}
+	for _, step := range evidence.CapacitySteps {
+		if step.Convergence > 10*time.Second || step.MaximumFiveSecond > step.FiveSecondCap*105/100 {
+			t.Fatalf("capacity step = %#v", step)
+		}
+	}
+}
+
 func TestQualificationSmokeTraversesNativeApolloRecoveryQueuePacerAndQUIC(t *testing.T) {
 	trace := qualificationProductionPathSmoke(t, qualificationMediaProfiles()[0])
 	if !trace.ApolloRecovered || !trace.ProductionQueue || !trace.ProductionPacer ||

gateway/qualification_harness_test.go

@@ -1507,30 +1507,49 @@ func runQualificationFleetStage(t *testing.T, fleet *qualificationFleet, profile
 	}
 	end := time.Now().Add(duration)
 	payload := qualificationPayload(profile)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
 	deliveries := make([]qualificationFlowDelivery, 0, int(duration/time.Millisecond))
-	var sequence uint32
+	var wait sync.WaitGroup
-	for time.Now().Before(end) {
+	var mu sync.Mutex
-		for _, path := range fleet.paths {
+	var firstErr error
-			if !time.Now().Before(end) {
+	for flowIndex, path := range fleet.paths {
-				break
+		wait.Add(1)
+		go func(flowIndex int, path *qualificationPath) {
+			defer wait.Done()
+			var sequence uint32
+			for time.Now().Before(end) && ctx.Err() == nil {
+				current := append([]byte(nil), payload...)
+				binary.BigEndian.PutUint32(current[len(current)-4:], uint32(flowIndex)<<24|sequence)
+				sequence++
+				trace, _, err := path.traverse(t, current)
+				if err == nil && (!trace.NativeUDPIngress || !trace.ApolloRecovered || !trace.ProductionQueue ||
+					!trace.ProductionMediaLoop || !trace.ProductionPacer || !trace.VerseQUIC ||
+					!trace.PublicClientDecode || !trace.PayloadPreserved) {
+					err = errors.New("fairness traffic bypassed production gateway traversal")
+				}
+				if err != nil {
+					mu.Lock()
+					if firstErr == nil {
+						firstErr = err
+						cancel()
+					}
+					mu.Unlock()
+					return
+				}
+				mu.Lock()
+				deliveries = append(deliveries, qualificationFlowDelivery{
+					at: time.Now(), flow: path.flow, bytes: int64(len(current) + frameHeaderSize),
+				})
+				mu.Unlock()
 			}
-			current := append([]byte(nil), payload...)
+		}(flowIndex, path)
-			binary.BigEndian.PutUint32(current[len(current)-4:], sequence)
-			sequence++
-			trace, _, err := path.traverse(t, current)
-			if err != nil {
-				return nil, err
-			}
-			if !trace.NativeUDPIngress || !trace.ApolloRecovered || !trace.ProductionQueue ||
-				!trace.ProductionMediaLoop || !trace.ProductionPacer || !trace.VerseQUIC ||
-				!trace.PublicClientDecode || !trace.PayloadPreserved {
-				return nil, errors.New("fairness traffic bypassed production gateway traversal")
-			}
-			deliveries = append(deliveries, qualificationFlowDelivery{
-				at: time.Now(), flow: path.flow, bytes: int64(len(current) + frameHeaderSize),
-			})
-		}
 	}
+	wait.Wait()
+	if firstErr != nil {
+		return nil, firstErr
+	}
+	sort.Slice(deliveries, func(first, second int) bool { return deliveries[first].at.Before(deliveries[second].at) })
 	return deliveries, nil
 }