diff --git a/apps/api/cmd/acceptance-load/main.go b/apps/api/cmd/acceptance-load/main.go index b4b3335..388ef68 100644 --- a/apps/api/cmd/acceptance-load/main.go +++ b/apps/api/cmd/acceptance-load/main.go @@ -279,7 +279,7 @@ func parseOptions() (options, error) { return options{}, errors.New("requests must be between 0 and 10000") } switch opts.profile { - case "simulated-smoke", "simulated-all", "gemini-baseline", "gemini-multi-image", "gemini-large", "gemini-peak", "video-throughput", "video-capacity", "video-recovery", + case "simulated-smoke", "simulated-all", "gemini-baseline", "gemini-multi-image", "gemini-large", "gemini-peak", "video-throughput", "video-capacity", "video-provider-quota", "video-recovery", "mixed-soak", "mixed-overload": if opts.emulatorURL == "" { return options{}, errors.New("AI_GATEWAY_ACCEPTANCE_EMULATOR_URL is required for simulated profiles") @@ -371,6 +371,12 @@ func run(ctx context.Context, opts options) ([]phaseResult, error) { requests = opts.requestCount } result = runVideo(ctx, client, opts, name, opts.shardRequestCount(requests), false, opts.emulatorFixtureURLs(), false, opts.shardIndex, opts.shardCount) + case "video-provider-quota": + requests := 24 + if opts.requestCount > 0 { + requests = opts.requestCount + } + result = runVideo(ctx, client, opts, name, opts.shardRequestCount(requests), false, opts.emulatorFixtureURLs(), false, opts.shardIndex, opts.shardCount) case "video-recovery": result = runVideo(ctx, client, opts, name, opts.shardRequestCount(96), true, opts.emulatorFixtureURLs(), false, opts.shardIndex, opts.shardCount) case "mixed-soak", "mixed-overload": @@ -657,18 +663,21 @@ func runVideo( if len(combo) > imageCount { combo = combo[:imageCount] } - if !realUpstream && logicalIndex%4 == 0 { + if !realUpstream && name != "video-provider-quota" && logicalIndex%4 == 0 { combo[0] = imageURLs[12+(logicalIndex/4)%4] } content := make([]any, 0, len(combo)+1) prompt := "多参考图生成连续运镜视频,保持人物、服装和场景一致" + if name == "video-provider-quota" { + prompt = "acceptance-provider-quota" + } if longRun { prompt += " acceptance-long-recovery" } if name == "video-capacity" { prompt += " acceptance-capacity-ladder" } - if !realUpstream && logicalIndex%4 == 0 { + if !realUpstream && name != "video-provider-quota" && logicalIndex%4 == 0 { prompt += " acceptance-force-conversion" } content = append(content, map[string]any{"type": "text", "text": prompt}) @@ -716,7 +725,7 @@ func runVideo( err = withOperation("video_submit", err) mu.Lock() latencies[index] = time.Since(requestStarted) - if !realUpstream && logicalIndex%4 == 0 { + if !realUpstream && name != "video-provider-quota" && logicalIndex%4 == 0 { forcedConversions++ } if err != nil { diff --git a/apps/api/internal/acceptanceemulator/server.go b/apps/api/internal/acceptanceemulator/server.go index d7aa913..609095e 100644 --- a/apps/api/internal/acceptanceemulator/server.go +++ b/apps/api/internal/acceptanceemulator/server.go @@ -78,11 +78,12 @@ type Report struct { } type videoTask struct { - ID string - Model string - CreatedAt time.Time - ReadyAt time.Time - ImageRefs []imageReference + ID string + Model string + CreatedAt time.Time + ReadyAt time.Time + UsageTokens int + ImageRefs []imageReference } type imageReference struct { @@ -243,7 +244,7 @@ func (s *Server) submitVideo(w http.ResponseWriter, r *http.Request) { id := "acceptance-video-" + strconv.FormatUint(s.nextID.Add(1), 10) task := videoTask{ ID: id, Model: strings.TrimSpace(stringValue(body["model"])), - CreatedAt: s.now(), ReadyAt: s.now().Add(delay), ImageRefs: refs, + CreatedAt: s.now(), ReadyAt: s.now().Add(delay), UsageTokens: videoUsageTokens(body), ImageRefs: refs, } s.mu.Lock() if existingID := s.videoByIdempotency[idempotencyKey]; existingID != "" { @@ -298,7 +299,7 @@ func (s *Server) getVideo(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusOK, map[string]any{ "id": id, "model": task.Model, "status": status, "content": content, "created_at": task.CreatedAt.Unix(), - "usage": map[string]any{"completion_tokens": 1, "total_tokens": 1}, + "usage": map[string]any{"completion_tokens": task.UsageTokens, "total_tokens": task.UsageTokens}, }) } @@ -674,12 +675,22 @@ func videoDelay(body map[string]any, longRun bool) time.Duration { if longRun { return 2*time.Minute + time.Duration(seed%61)*time.Second } + if videoPromptContains(body, "acceptance-provider-quota") { + return 5*time.Second + time.Duration(seed%6)*time.Second + } if videoPromptContains(body, "acceptance-capacity-ladder") { return 30*time.Second + time.Duration(seed%16)*time.Second } return 5*time.Second + time.Duration(seed%11)*time.Second } +func videoUsageTokens(body map[string]any) int { + if videoPromptContains(body, "acceptance-provider-quota") { + return 7 + } + return 1 +} + func videoPromptContains(body map[string]any, marker string) bool { content, _ := body["content"].([]any) for _, raw := range content { diff --git a/apps/api/internal/acceptanceemulator/server_test.go b/apps/api/internal/acceptanceemulator/server_test.go index 48a3d54..0e21f7b 100644 --- a/apps/api/internal/acceptanceemulator/server_test.go +++ b/apps/api/internal/acceptanceemulator/server_test.go @@ -242,6 +242,23 @@ func TestVideoCapacityDelay(t *testing.T) { } } +func TestVideoProviderQuotaDelayAndUsage(t *testing.T) { + body := map[string]any{ + "seed": json.Number("5"), + "content": []any{map[string]any{ + "type": "text", + "text": "acceptance-provider-quota", + }}, + } + delay := videoDelay(body, false) + if delay < 5*time.Second || delay > 10*time.Second { + t.Fatalf("video provider quota delay=%s, want 5s..10s", delay) + } + if tokens := videoUsageTokens(body); tokens != 7 { + t.Fatalf("video provider quota usage tokens=%d, want 7", tokens) + } +} + func TestVolcesProtocolAcceptsThreeSixAndNineReferenceImages(t *testing.T) { server := New(Config{Wait: func(context.Context, time.Duration) error { return nil }}) httpServer := httptest.NewServer(server.Handler()) diff --git a/apps/api/internal/runner/limits_test.go b/apps/api/internal/runner/limits_test.go index 0169ad7..a67eb15 100644 --- a/apps/api/internal/runner/limits_test.go +++ b/apps/api/internal/runner/limits_test.go @@ -134,3 +134,15 @@ func TestReservationsFromPolicySkipsNonPositiveLimits(t *testing.T) { t.Fatalf("expected concurrent reservation with limit 2, got %+v", reservations[0]) } } + +func TestProviderQuotaPromptReservesSevenTokens(t *testing.T) { + body := map[string]any{ + "content": []any{map[string]any{ + "type": "text", + "text": "acceptance-provider-quota", + }}, + } + if got := estimateRequestTokens(body); got != 7 { + t.Fatalf("provider quota prompt tokens=%d, want 7", got) + } +} diff --git a/scripts/acceptance/provider-burst.sh b/scripts/acceptance/provider-burst.sh index 777b670..37ecfb2 100644 --- a/scripts/acceptance/provider-burst.sh +++ b/scripts/acceptance/provider-burst.sh @@ -44,11 +44,11 @@ WITH source AS ( AND COALESCE((platform.config->>'acceptanceSnapshot')::boolean, false) ORDER BY platform.priority, platform.created_at LIMIT 1 -), variants(platform_id, platform_key, platform_name, concurrency_limit) AS ( +), variants(platform_id, platform_key, platform_name, concurrency_limit, rpm_limit, tpm_limit, priority) AS ( VALUES - ('${provider_burst_platform_ids[0]}'::uuid, 'acceptance-provider-burst-2', 'Acceptance Provider Burst C2', 2), - ('${provider_burst_platform_ids[1]}'::uuid, 'acceptance-provider-burst-4', 'Acceptance Provider Burst C4', 4), - ('${provider_burst_platform_ids[2]}'::uuid, 'acceptance-provider-burst-6', 'Acceptance Provider Burst C6', 6) + ('${provider_burst_platform_ids[0]}'::uuid, 'acceptance-provider-burst-2', 'Acceptance Provider Burst C2', 2, 4, 28, 100), + ('${provider_burst_platform_ids[1]}'::uuid, 'acceptance-provider-burst-4', 'Acceptance Provider Burst C4', 4, 8, 56, 100), + ('${provider_burst_platform_ids[2]}'::uuid, 'acceptance-provider-burst-6', 'Acceptance Provider Burst C6', 6, 12, 84, 200) ) INSERT INTO integration_platforms ( id, provider, platform_key, name, base_url, auth_type, credentials, config, @@ -61,12 +61,14 @@ SELECT variants.platform_id, source.provider, variants.platform_key, variants.pl source.config || jsonb_build_object( 'acceptanceProviderBurst', true, 'acceptanceBurstRunId', '$run_id', - 'acceptanceConcurrencyLimit', variants.concurrency_limit + 'acceptanceConcurrencyLimit', variants.concurrency_limit, + 'acceptanceRPMLimit', variants.rpm_limit, + 'acceptanceTPMLimit', variants.tpm_limit ), source.visibility_scope, source.tenant_id, source.tenant_key, source.default_pricing_mode, source.default_discount_factor, source.pricing_rule_set_id, source.retry_policy, '{\"rules\":[]}'::jsonb, - 100, NULL, 'enabled', NULL, NULL, variants.platform_name + variants.priority, NULL, 'enabled', NULL, NULL, variants.platform_name FROM source CROSS JOIN variants ON CONFLICT (id) DO UPDATE SET provider = EXCLUDED.provider, @@ -102,11 +104,11 @@ WITH source AS ( AND COALESCE((platform.config->>'acceptanceSnapshot')::boolean, false) ORDER BY platform.priority, platform.created_at LIMIT 1 -), variants(model_id, platform_id, concurrency_limit) AS ( +), variants(model_id, platform_id, concurrency_limit, rpm_limit, tpm_limit) AS ( VALUES - ('${provider_burst_model_ids[0]}'::uuid, '${provider_burst_platform_ids[0]}'::uuid, 2), - ('${provider_burst_model_ids[1]}'::uuid, '${provider_burst_platform_ids[1]}'::uuid, 4), - ('${provider_burst_model_ids[2]}'::uuid, '${provider_burst_platform_ids[2]}'::uuid, 6) + ('${provider_burst_model_ids[0]}'::uuid, '${provider_burst_platform_ids[0]}'::uuid, 2, 4, 28), + ('${provider_burst_model_ids[1]}'::uuid, '${provider_burst_platform_ids[1]}'::uuid, 4, 8, 56), + ('${provider_burst_model_ids[2]}'::uuid, '${provider_burst_platform_ids[2]}'::uuid, 6, 12, 84) ) INSERT INTO platform_models ( id, platform_id, base_model_id, model_name, model_alias, model_type, @@ -127,7 +129,8 @@ SELECT variants.model_id, variants.platform_id, source.base_model_id, 'metric', 'concurrent', 'limit', variants.concurrency_limit, 'leaseTtlSeconds', 120 ), - jsonb_build_object('metric', 'rpm', 'limit', 600, 'windowSeconds', 60) + jsonb_build_object('metric', 'rpm', 'limit', variants.rpm_limit, 'windowSeconds', 60), + jsonb_build_object('metric', 'tpm_total', 'limit', variants.tpm_limit, 'windowSeconds', 60) )), 'override', NULL, '{}'::jsonb, source.provider_model_name, NULL, true FROM source CROSS JOIN variants @@ -219,13 +222,18 @@ DO UPDATE SET status = 'active', metadata = EXCLUDED.metadata, SELECT jsonb_build_object( 'platforms', count(*), 'limits', jsonb_agg((platform.config->>'acceptanceConcurrencyLimit')::int ORDER BY (platform.config->>'acceptanceConcurrencyLimit')::int), + 'rpmLimits', jsonb_agg((platform.config->>'acceptanceRPMLimit')::int ORDER BY (platform.config->>'acceptanceConcurrencyLimit')::int), + 'tpmLimits', jsonb_agg((platform.config->>'acceptanceTPMLimit')::int ORDER BY (platform.config->>'acceptanceConcurrencyLimit')::int), + 'priorities', jsonb_agg(platform.priority ORDER BY (platform.config->>'acceptanceConcurrencyLimit')::int), 'enabled', count(*) FILTER (WHERE platform.status='enabled' AND model.enabled) ) FROM integration_platforms platform JOIN platform_models model ON model.platform_id = platform.id WHERE COALESCE((platform.config->>'acceptanceProviderBurst')::boolean, false) AND platform.config->>'acceptanceBurstRunId' = '$run_id';") - jq -e '.platforms == 3 and .enabled == 3 and .limits == [2,4,6]' <<<"$configured" >/dev/null + jq -e '.platforms == 3 and .enabled == 3 and .limits == [2,4,6] and + .rpmLimits == [4,8,12] and .tpmLimits == [28,56,84] and .priorities == [100,100,200]' \ + <<<"$configured" >/dev/null } sample_provider_burst() { @@ -235,7 +243,10 @@ sample_provider_burst() { database_query " WITH configured AS ( SELECT platform.id platform_id, platform.platform_key, model.id platform_model_id, - (platform.config->>'acceptanceConcurrencyLimit')::int concurrency_limit + platform.priority, + (platform.config->>'acceptanceConcurrencyLimit')::int concurrency_limit, + (platform.config->>'acceptanceRPMLimit')::int rpm_limit, + (platform.config->>'acceptanceTPMLimit')::int tpm_limit FROM integration_platforms platform JOIN platform_models model ON model.platform_id = platform.id WHERE COALESCE((platform.config->>'acceptanceProviderBurst')::boolean, false) @@ -302,7 +313,10 @@ build_provider_burst_report() { platforms=$(database_query " WITH configured AS ( SELECT platform.id platform_id, platform.platform_key, model.id platform_model_id, - (platform.config->>'acceptanceConcurrencyLimit')::int concurrency_limit + platform.priority, + (platform.config->>'acceptanceConcurrencyLimit')::int concurrency_limit, + (platform.config->>'acceptanceRPMLimit')::int rpm_limit, + (platform.config->>'acceptanceTPMLimit')::int tpm_limit FROM integration_platforms platform JOIN platform_models model ON model.platform_id=platform.id WHERE COALESCE((platform.config->>'acceptanceProviderBurst')::boolean, false) @@ -332,18 +346,34 @@ WITH configured AS ( FROM gateway_task_attempts attempt WHERE attempt.task_id IN (SELECT id FROM gateway_tasks WHERE acceptance_run_id='$run_id'::uuid) GROUP BY attempt.platform_model_id +), rate_peaks AS ( + SELECT split_part(counter.scope_key, ':', 3)::uuid platform_model_id, + max(counter.limit_value) FILTER (WHERE counter.metric='rpm') rpm_limit, + max(counter.used_value + counter.reserved_value) FILTER (WHERE counter.metric='rpm') rpm_peak, + max(counter.limit_value) FILTER (WHERE counter.metric='tpm_total') tpm_limit, + max(counter.used_value + counter.reserved_value) FILTER (WHERE counter.metric='tpm_total') tpm_peak + FROM gateway_rate_limit_counters counter + WHERE counter.scope_type='platform_model' + AND counter.scope_key LIKE 'acceptance:$run_id:%' + GROUP BY split_part(counter.scope_key, ':', 3)::uuid ) SELECT COALESCE(jsonb_agg(jsonb_build_object( 'platformId', configured.platform_id, 'platformModelId', configured.platform_model_id, 'platformKey', configured.platform_key, + 'priority', configured.priority, 'limit', configured.concurrency_limit, 'peak', COALESCE(peaks.peak,0), + 'rpmLimit', configured.rpm_limit, + 'rpmPeak', COALESCE(rate_peaks.rpm_peak,0), + 'tpmLimit', configured.tpm_limit, + 'tpmPeak', COALESCE(rate_peaks.tpm_peak,0), 'attempts', COALESCE(attempts.attempts,0) ) ORDER BY configured.concurrency_limit), '[]'::jsonb) FROM configured LEFT JOIN peaks ON peaks.platform_model_id=configured.platform_model_id -LEFT JOIN attempts ON attempts.platform_model_id=configured.platform_model_id;") +LEFT JOIN attempts ON attempts.platform_model_id=configured.platform_model_id +LEFT JOIN rate_peaks ON rate_peaks.platform_model_id=configured.platform_model_id;") queue=$(jq -s '{ samples:length, maxQueuedTasks:(map(.queuedTasks)|max), @@ -437,7 +467,16 @@ SELECT jsonb_build_object( FROM gateway_task_callback_outbox callback WHERE callback.task_id IN (SELECT id FROM gateway_tasks WHERE acceptance_run_id='$run_id'::uuid);") resource_summary=$(provider_burst_resource_summary "$resources") - if ! jq -e 'length==3 and all(.[]; .peak == .limit and .attempts > 0)' <<<"$platforms" >/dev/null; then passed=false; fi + if ! jq -e 'length==3 and + [.[].priority] == [100,100,200] and + [.[].attempts] == [4,8,12] and + all(.[]; .peak == .limit and .rpmPeak == .rpmLimit and .tpmPeak == .tpmLimit)' \ + <<<"$platforms" >/dev/null; then passed=false; fi + if ! jq -e '.[0].priority==.[1].priority and + .[0].attempts*2==.[1].attempts and + .[2].priority>.[1].priority and + .[0].peak==.[0].limit and .[1].peak==.[1].limit and .[2].attempts>0' \ + <<<"$platforms" >/dev/null; then passed=false; fi if ! jq -e '.queueObserved and .admissionWaitObserved and .drainedAtEnd and .maxQueuedTasks > 0 and .maxWaitingAdmissions > 0' <<<"$queue" >/dev/null; then passed=false; fi if ! jq -e --argjson requests "$requests" 'length==3 and all(.[]; .tasks>0 and .workerInstanceId!="unclaimed") and ([.[].tasks]|add)==$requests' <<<"$distribution" >/dev/null; then passed=false; fi if ! jq -e 'length==3 and all(.[]; .peakActiveTasks>0 and (.pressureStates|index("critical")|not))' <<<"$worker_peaks" >/dev/null; then passed=false; fi @@ -458,11 +497,20 @@ WHERE callback.task_id IN (SELECT id FROM gateway_tasks WHERE acceptance_run_id= --argjson resources "$resource_summary" --argjson leaks "$leaks" \ --argjson duplicates "$duplicates" --argjson callbacks "$callbacks" \ '{ - schemaVersion:"acceptance-provider-burst-report/v1", + schemaVersion:"acceptance-provider-burst-report/v2", runId:$runId,profile:"provider-burst",model:$model, startedAt:(now|todateiso8601),finishedAt:(now|todateiso8601), passed:$passed,secretSafe:true,requests:$requests,phases:$phases, - providerRouting:{platforms:$platforms,totalConfiguredConcurrency:([$platforms[].limit]|add)}, + providerRouting:{ + platforms:$platforms, + totalConfiguredConcurrency:([$platforms[].limit]|add), + samePriorityLoadBalanced:($platforms[0].priority==$platforms[1].priority and + $platforms[0].attempts*2==$platforms[1].attempts), + lowerPrioritySpilloverObserved:($platforms[2].priority>$platforms[1].priority and + $platforms[0].peak==$platforms[0].limit and + $platforms[1].peak==$platforms[1].limit and + $platforms[2].attempts>0) + }, queue:$queue, cluster:{workers:$workers,workerPeaks:$workerPeaks,taskDistribution:$distribution, rateLimits:$rateLimits,concurrencyPeaks:$concurrencyPeaks, @@ -473,14 +521,14 @@ WHERE callback.task_id IN (SELECT id FROM gateway_tasks WHERE acceptance_run_id= } provider_burst() { - local requests=${AI_GATEWAY_LOCAL_PROVIDER_BURST_REQUESTS:-48} + local requests=24 local load_report="$report_root/provider-burst-load.json" local samples="$report_root/provider-burst-samples.ndjson" local worker_samples="$report_root/provider-burst-worker-samples.ndjson" local resources="$report_root/provider-burst-resources.csv" local report="$report_root/provider-burst.json" worker_sample_stop="$report_root/provider-burst-samples.stop" - [[ $requests =~ ^[0-9]+$ ]] && ((requests >= 24 && requests <= 256)) + [[ $requests == 24 ]] [[ ! -e $load_report && ! -e $samples && ! -e $worker_samples && ! -e $resources && ! -e $report && ! -e $worker_sample_stop ]] || fail_gate acceptance_report_exists "provider burst evidence already exists" current_phase=provider_burst_platforms @@ -495,7 +543,7 @@ provider_burst() { sample_resources "$resources" "$worker_sample_stop" & active_resource_sampler_pid=$! current_phase=provider_burst_load - run_load video-throughput "$load_report" -requests "$requests" + run_load video-provider-quota "$load_report" -requests "$requests" sleep 2 touch "$worker_sample_stop" wait "$active_provider_sampler_pid"