diff --git a/gateway/gateway_test.go b/gateway/gateway_test.go index b07e924..f429815 100644 --- a/gateway/gateway_test.go +++ b/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) diff --git a/gateway/provider.go b/gateway/provider.go index 2f7403b..2609e22 100644 --- a/gateway/provider.go +++ b/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 diff --git a/gateway/qualification_contract_test.go b/gateway/qualification_contract_test.go index f22e6be..7084711 100644 --- a/gateway/qualification_contract_test.go +++ b/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 || diff --git a/gateway/qualification_harness_test.go b/gateway/qualification_harness_test.go index 72ec722..34d5f76 100644 --- a/gateway/qualification_harness_test.go +++ b/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 - for time.Now().Before(end) { - for _, path := range fleet.paths { - if !time.Now().Before(end) { - break + var wait sync.WaitGroup + var mu sync.Mutex + var firstErr error + for flowIndex, path := range fleet.paths { + 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...) - 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), - }) - } + }(flowIndex, path) } + 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 }