fix(acceptance): 为高并发验收启用有界排队
生产同构 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 通过。
This commit is contained in:
@@ -196,6 +196,7 @@ func run(ctx context.Context, opts options) ([]phaseResult, error) {
|
||||
MaxIdleConns: 2048,
|
||||
MaxIdleConnsPerHost: 1024,
|
||||
MaxConnsPerHost: 1024,
|
||||
ForceAttemptHTTP2: true,
|
||||
TLSClientConfig: tlsConfig,
|
||||
},
|
||||
}
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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])
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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 {
|
||||
|
||||
Reference in New Issue
Block a user