From 2bfeb3e17953a6515e9ccdfc099f8f883ffd49d5 Mon Sep 17 00:00:00 2001 From: wangbo Date: Fri, 31 Jul 2026 01:34:16 +0800 Subject: [PATCH] =?UTF-8?q?fix(acceptance):=20=E4=B8=BA=E9=AB=98=E5=B9=B6?= =?UTF-8?q?=E5=8F=91=E9=AA=8C=E6=94=B6=E5=90=AF=E7=94=A8=E6=9C=89=E7=95=8C?= =?UTF-8?q?=E6=8E=92=E9=98=9F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 生产同构 Gemini 千请求在真实候选并发饱和后因 queueing disabled 返回 429，同时单台压测机建立千条 WAN TCP 产生连接层 reset。仅对 acceptance 与 acceptance_canary 覆盖为 10000 条、最长 15 分钟的有界等待队列，保留候选原始并发和 RPM/TPM 上限；压测传输强制 HTTP/2 复用连接，避免单源连接风暴。正式 production 策略不变。\n\n验证：acceptance-load 与 runner 定向测试通过，新增生产策略不变和验收队列边界单测；gofmt、bash -n、ShellCheck、git diff --check 通过。 --- apps/api/cmd/acceptance-load/main.go | 1 + apps/api/internal/runner/admission.go | 19 ++++++++++ apps/api/internal/runner/admission_test.go | 44 +++++++++++++++++++++- apps/api/internal/runner/service.go | 1 + 4 files changed, 64 insertions(+), 1 deletion(-) diff --git a/apps/api/cmd/acceptance-load/main.go b/apps/api/cmd/acceptance-load/main.go index b61753f..04dfd87 100644 --- a/apps/api/cmd/acceptance-load/main.go +++ b/apps/api/cmd/acceptance-load/main.go @@ -196,6 +196,7 @@ func run(ctx context.Context, opts options) ([]phaseResult, error) { MaxIdleConns: 2048, MaxIdleConnsPerHost: 1024, MaxConnsPerHost: 1024, + ForceAttemptHTTP2: true, TLSClientConfig: tlsConfig, }, } diff --git a/apps/api/internal/runner/admission.go b/apps/api/internal/runner/admission.go index c1cec15..9ea362d 100644 --- a/apps/api/internal/runner/admission.go +++ b/apps/api/internal/runner/admission.go @@ -32,6 +32,8 @@ const ( asyncWorkerCapacityScopeKey = "global" asyncWorkerQueueLimit = 10000 asyncWorkerMaxWaitSeconds = 24 * 60 * 60 + acceptanceQueueLimit = 10000 + acceptanceQueueMaxWait = 15 * 60 ) func distributedAdmissionModelType(modelType string) bool { @@ -117,6 +119,7 @@ func (s *Service) buildTaskAdmissionPlan(ctx context.Context, task store.Gateway return taskAdmissionPlan{}, err } } + scopes = acceptanceAdmissionScopes(task, scopes) hasConcurrentLimit := false for _, scope := range scopes { if scope.ConcurrentLimit > 0 { @@ -168,6 +171,21 @@ func (s *Service) withAsyncWorkerCapacityScope(ctx context.Context, scopes []sto return out, nil } +func acceptanceAdmissionScopes(task store.GatewayTask, scopes []store.AdmissionScope) []store.AdmissionScope { + if task.RunMode != "acceptance" && task.RunMode != "acceptance_canary" { + return scopes + } + out := append([]store.AdmissionScope(nil), scopes...) + for index := range out { + if out[index].ConcurrentLimit <= 0 { + continue + } + out[index].QueueLimit = acceptanceQueueLimit + out[index].MaxWaitSeconds = acceptanceQueueMaxWait + } + return out +} + func (s *Service) tryTaskAdmission(ctx context.Context, task store.GatewayTask, plan taskAdmissionPlan, waiterID string) (store.TaskAdmissionResult, error) { return s.tryTaskAdmissionWithAdmittedHook(ctx, task, plan, waiterID, nil) } @@ -313,6 +331,7 @@ func (s *Service) ensureCandidateAdmission( return store.TaskAdmissionResult{}, true, err } } + scopes = acceptanceAdmissionScopes(task, scopes) hasConcurrentLimit := false for _, scope := range scopes { if scope.ConcurrentLimit > 0 { diff --git a/apps/api/internal/runner/admission_test.go b/apps/api/internal/runner/admission_test.go index 87d8555..e8ee007 100644 --- a/apps/api/internal/runner/admission_test.go +++ b/apps/api/internal/runner/admission_test.go @@ -1,6 +1,10 @@ package runner -import "testing" +import ( + "testing" + + "github.com/easyai/easyai-ai-gateway/apps/api/internal/store" +) func TestDistributedAdmissionModelTypeBoundary(t *testing.T) { for _, modelType := range []string{ @@ -19,3 +23,41 @@ func TestDistributedAdmissionModelTypeBoundary(t *testing.T) { } } } + +func TestAcceptanceAdmissionScopesEnableBoundedQueueWithoutChangingConcurrency(t *testing.T) { + input := []store.AdmissionScope{{ + ScopeType: "platform_model", + ScopeKey: "model-1", + ConcurrentLimit: 10, + QueueLimit: 0, + MaxWaitSeconds: 0, + }} + + got := acceptanceAdmissionScopes(store.GatewayTask{RunMode: "acceptance"}, input) + + if got[0].ConcurrentLimit != 10 { + t.Fatalf("acceptance must preserve the production concurrency limit, got %+v", got[0]) + } + if got[0].QueueLimit != acceptanceQueueLimit || got[0].MaxWaitSeconds != acceptanceQueueMaxWait { + t.Fatalf("acceptance must enable a bounded queue, got %+v", got[0]) + } + if input[0].QueueLimit != 0 || input[0].MaxWaitSeconds != 0 { + t.Fatalf("acceptance queue overlay must not mutate the production scopes, got %+v", input[0]) + } +} + +func TestAcceptanceAdmissionScopesLeaveProductionPolicyUnchanged(t *testing.T) { + input := []store.AdmissionScope{{ + ScopeType: "platform_model", + ScopeKey: "model-1", + ConcurrentLimit: 10, + QueueLimit: 3, + MaxWaitSeconds: 7, + }} + + got := acceptanceAdmissionScopes(store.GatewayTask{RunMode: "production"}, input) + + if got[0].QueueLimit != 3 || got[0].MaxWaitSeconds != 7 || got[0].ConcurrentLimit != 10 { + t.Fatalf("production admission policy changed: %+v", got[0]) + } +} diff --git a/apps/api/internal/runner/service.go b/apps/api/internal/runner/service.go index c49f639..112c93d 100644 --- a/apps/api/internal/runner/service.go +++ b/apps/api/internal/runner/service.go @@ -525,6 +525,7 @@ func (s *Service) executeWithToken(ctx context.Context, task store.GatewayTask, admittedPlatformModelID := "" if distributedAdmission && len(candidates) > 0 { admissionScopes, groupID := s.admissionScopes(ctx, user, candidates[0]) + admissionScopes = acceptanceAdmissionScopes(task, admissionScopes) hasConcurrentLimit := false for _, scope := range admissionScopes { if scope.ConcurrentLimit > 0 {