将协议模拟验收的供应商并发限额与 Worker 容量门禁分离,真实金丝雀继续保留生产限流。readyz 改用关键连接池并预热关键及 River 池,延长生产连接空闲周期,降低跨地域连接抖动。失败 Run 现在可以原子替换且失败报告记录任务数量,避免 validation 之间短暂放开正式流量。\n\n验证:Go 全量测试、go vet、PostgreSQL 集成测试、ShellCheck、迁移安全、发布脚本、pnpm lint/test/build、OpenAPI 无漂移均通过。
182 lines
6.0 KiB
Go
182 lines
6.0 KiB
Go
package runner
|
|
|
|
import (
|
|
"testing"
|
|
|
|
"github.com/easyai/easyai-ai-gateway/apps/api/internal/store"
|
|
)
|
|
|
|
func TestDistributedAdmissionModelTypeBoundary(t *testing.T) {
|
|
for _, modelType := range []string{
|
|
"image_generate", "image_edit", "image_analysis", "image_vectorize",
|
|
"video_generate", "video_enhance", "image_to_video", "text_to_video",
|
|
"video_edit", "video_reference", "video_first_last_frame", "video_understanding", "omni_video", "omni",
|
|
"audio_generate", "audio_understanding", "text_to_speech", "voice_clone",
|
|
} {
|
|
if !distributedAdmissionModelType(modelType) {
|
|
t.Errorf("%s should use distributed admission", modelType)
|
|
}
|
|
}
|
|
for _, modelType := range []string{"text_generate", "embedding", "rerank", "", "unknown"} {
|
|
if distributedAdmissionModelType(modelType) {
|
|
t.Errorf("%s should preserve immediate execution/rate-limit behavior", modelType)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestAcceptanceAdmissionScopesUseWorkerCapacityInsteadOfSupplierConcurrency(t *testing.T) {
|
|
input := []store.AdmissionScope{{
|
|
ScopeType: "platform_model",
|
|
ScopeKey: "model-1",
|
|
ConcurrentLimit: 10,
|
|
QueueLimit: 0,
|
|
MaxWaitSeconds: 0,
|
|
}, {
|
|
ScopeType: "worker_capacity",
|
|
ScopeKey: "global",
|
|
ConcurrentLimit: 48,
|
|
}}
|
|
|
|
got := acceptanceAdmissionScopes(store.GatewayTask{RunMode: "acceptance"}, input)
|
|
|
|
if got[0].ConcurrentLimit != 0 {
|
|
t.Fatalf("protocol-emulated acceptance must defer to worker capacity, 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 got[1].ConcurrentLimit != 48 {
|
|
t.Fatalf("acceptance worker capacity changed, got %+v", got[1])
|
|
}
|
|
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 TestAcceptanceCanaryPreservesProductionConcurrency(t *testing.T) {
|
|
input := []store.AdmissionScope{{
|
|
ScopeType: "platform_model",
|
|
ScopeKey: "model-1",
|
|
ConcurrentLimit: 10,
|
|
}}
|
|
|
|
got := acceptanceAdmissionScopes(store.GatewayTask{RunMode: "acceptance_canary"}, input)
|
|
|
|
if got[0].ConcurrentLimit != 10 {
|
|
t.Fatalf("real acceptance canary must preserve production concurrency, got %+v", got[0])
|
|
}
|
|
}
|
|
|
|
func TestAcceptanceInfrastructureReservationsOnlyRemoveSupplierConcurrency(t *testing.T) {
|
|
input := []store.RateLimitReservation{
|
|
{ScopeType: "platform_model", Metric: "concurrent", Limit: 10},
|
|
{ScopeType: "platform_model", Metric: "rpm", Limit: 600},
|
|
{ScopeType: "user_group", Metric: "concurrent", Limit: 20},
|
|
}
|
|
|
|
got := acceptanceInfrastructureReservations(store.GatewayTask{RunMode: "acceptance"}, input)
|
|
if len(got) != 2 || got[0].Metric != "rpm" || got[1].ScopeType != "user_group" {
|
|
t.Fatalf("unexpected acceptance reservations: %+v", got)
|
|
}
|
|
canary := acceptanceInfrastructureReservations(store.GatewayTask{RunMode: "acceptance_canary"}, input)
|
|
if len(canary) != len(input) {
|
|
t.Fatalf("real canary reservations changed: %+v", canary)
|
|
}
|
|
}
|
|
|
|
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])
|
|
}
|
|
}
|
|
|
|
func TestAsyncAdmissionBatchCoversCertifiedGlobalCapacity(t *testing.T) {
|
|
for _, testCase := range []struct {
|
|
globalHardLimit int
|
|
want int
|
|
}{
|
|
{globalHardLimit: 48, want: 48},
|
|
{globalHardLimit: 64, want: 64},
|
|
{globalHardLimit: 128, want: 64},
|
|
{globalHardLimit: 0, want: 64},
|
|
} {
|
|
if got := asyncAdmissionBatchLimit(testCase.globalHardLimit); got != testCase.want {
|
|
t.Fatalf("async admission batch limit for %d = %d, want %d", testCase.globalHardLimit, got, testCase.want)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestPinCandidatesToTaskAdmissionPreservesAdmittedCandidate(t *testing.T) {
|
|
input := []store.RuntimeModelCandidate{
|
|
{PlatformID: "platform-a", PlatformModelID: "model-a"},
|
|
{PlatformID: "platform-b", PlatformModelID: "model-b"},
|
|
{PlatformID: "platform-c", PlatformModelID: "model-c"},
|
|
}
|
|
admission := &store.TaskAdmission{
|
|
Status: "admitted",
|
|
PlatformID: "platform-b",
|
|
PlatformModelID: "model-b",
|
|
}
|
|
|
|
got, pinned := pinCandidatesToTaskAdmission(input, admission)
|
|
|
|
if !pinned {
|
|
t.Fatal("expected admitted candidate to be pinned")
|
|
}
|
|
if got[0].PlatformModelID != "model-b" || got[1].PlatformModelID != "model-a" || got[2].PlatformModelID != "model-c" {
|
|
t.Fatalf("unexpected candidate order: %+v", got)
|
|
}
|
|
if input[0].PlatformModelID != "model-a" {
|
|
t.Fatalf("candidate pinning mutated the caller slice: %+v", input)
|
|
}
|
|
}
|
|
|
|
func TestPinCandidatesToTaskAdmissionPreservesWaitingCandidate(t *testing.T) {
|
|
input := []store.RuntimeModelCandidate{
|
|
{PlatformID: "platform-a", PlatformModelID: "model-a"},
|
|
{PlatformID: "platform-b", PlatformModelID: "model-b"},
|
|
}
|
|
admission := &store.TaskAdmission{
|
|
Status: "waiting",
|
|
PlatformID: "platform-b",
|
|
PlatformModelID: "model-b",
|
|
}
|
|
|
|
got, pinned := pinCandidatesToTaskAdmission(input, admission)
|
|
|
|
if !pinned || got[0].PlatformModelID != "model-b" {
|
|
t.Fatalf("waiting candidate was not pinned: pinned=%v candidates=%+v", pinned, got)
|
|
}
|
|
}
|
|
|
|
func TestPinCandidatesToTaskAdmissionIgnoresMissingCandidate(t *testing.T) {
|
|
input := []store.RuntimeModelCandidate{
|
|
{PlatformID: "platform-a", PlatformModelID: "model-a"},
|
|
{PlatformID: "platform-b", PlatformModelID: "model-b"},
|
|
}
|
|
admission := &store.TaskAdmission{
|
|
Status: "admitted",
|
|
PlatformID: "platform-c",
|
|
PlatformModelID: "model-c",
|
|
}
|
|
|
|
got, pinned := pinCandidatesToTaskAdmission(input, admission)
|
|
|
|
if pinned {
|
|
t.Fatalf("unexpected candidate pin for missing binding: %+v", got)
|
|
}
|
|
if got[0].PlatformModelID != "model-a" {
|
|
t.Fatalf("candidate order changed for missing binding: %+v", got)
|
|
}
|
|
}
|