wangbo
edcbbd6a87
将批量准入改为每任务独立事务，避免 P24 压测时 48 条任务共享长事务造成 transactionid 锁潮与 Worker 槽位空转。失败验收在保持 validation 的同时清理未提交任务，并记录精确压力门禁原因。统一生产 0+2 Worker 拓扑、运行时 ConfigMap 与发布配置，补充镜像预检诊断和回归测试。\n\n验证：Go 全量测试、go vet、真实 PostgreSQL 48 任务集成测试、迁移安全检查、发布脚本测试、bash -n、ShellCheck、Compose 配置均通过。
326 lines
13 KiB
Go
326 lines
13 KiB
Go
package store
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"os"
|
|
"strings"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/easyai/easyai-ai-gateway/apps/api/internal/auth"
|
|
)
|
|
|
|
func TestAcceptanceTrafficGateAndCASPromotion(t *testing.T) {
|
|
databaseURL := strings.TrimSpace(os.Getenv("AI_GATEWAY_TEST_DATABASE_URL"))
|
|
if databaseURL == "" {
|
|
t.Skip("set AI_GATEWAY_TEST_DATABASE_URL to run acceptance traffic integration test")
|
|
}
|
|
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
|
|
defer cancel()
|
|
applyOIDCJITTestMigrations(t, ctx, databaseURL)
|
|
db, err := Connect(ctx, databaseURL)
|
|
if err != nil {
|
|
t.Fatalf("connect store: %v", err)
|
|
}
|
|
defer db.Close()
|
|
resetTrafficMode := func() {
|
|
_, _ = db.pool.Exec(context.Background(), `
|
|
UPDATE system_settings
|
|
SET value = '{"mode":"live","revision":0}'::jsonb, updated_at = now()
|
|
WHERE setting_key = $1`, SystemSettingGatewayTrafficMode)
|
|
}
|
|
resetTrafficMode()
|
|
t.Cleanup(resetTrafficMode)
|
|
|
|
token := "acceptance-token-" + time.Now().Format("150405.000000000")
|
|
run, err := db.CreateAcceptanceRun(ctx, CreateAcceptanceRunInput{
|
|
ReleaseSHA: strings.Repeat("a", 40),
|
|
APIImageDigest: "sha256:" + strings.Repeat("b", 64),
|
|
WorkerImageDigest: "sha256:" + strings.Repeat("c", 64),
|
|
APIKeyID: "acceptance-api-key",
|
|
UserID: "acceptance-user",
|
|
Token: token,
|
|
EmulatorBaseURL: "http://acceptance-emulator:8090",
|
|
CallbackURL: "http://acceptance-emulator:8090/callbacks",
|
|
CapacityProfile: "P24",
|
|
Config: map[string]any{"participants": []any{
|
|
map[string]any{"apiKeyId": "acceptance-shard-key", "userId": "acceptance-shard-user"},
|
|
}},
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("create acceptance run: %v", err)
|
|
}
|
|
encoded, _ := json.Marshal(run)
|
|
if strings.Contains(string(encoded), token) {
|
|
t.Fatal("acceptance run response exposed the raw token")
|
|
}
|
|
if failed, err := db.FinishAcceptanceRun(ctx, FinishAcceptanceRunInput{
|
|
RunID: run.ID, Passed: false, FailureReason: "pre-activation gate failed",
|
|
}); err != nil || failed.Status != "failed" {
|
|
t.Fatalf("finish pending acceptance run after pre-activation failure=%+v err=%v", failed, err)
|
|
}
|
|
mode, err := db.ActivateAcceptanceRun(ctx, run.ID)
|
|
if err != nil {
|
|
t.Fatalf("activate acceptance run: %v", err)
|
|
}
|
|
if mode.Mode != "validation" || mode.RunID != run.ID || mode.Revision != 1 {
|
|
t.Fatalf("unexpected validation mode: %+v", mode)
|
|
}
|
|
user := &auth.User{ID: "acceptance-user", APIKeyID: "acceptance-api-key"}
|
|
if _, err := db.AuthorizeAcceptanceTask(ctx, "", "", user); !errors.Is(err, ErrProductionTrafficPaused) {
|
|
t.Fatalf("production request error=%v, want traffic paused", err)
|
|
}
|
|
if _, err := db.AuthorizeAcceptanceTask(ctx, run.ID, "wrong-token", user); !errors.Is(err, ErrAcceptanceNotAuthorized) {
|
|
t.Fatalf("wrong acceptance token error=%v", err)
|
|
}
|
|
if authorizedRunID, err := db.AuthorizeAcceptanceTask(ctx, run.ID, token, user); err != nil || authorizedRunID != run.ID {
|
|
t.Fatalf("authorize acceptance task run=%q err=%v", authorizedRunID, err)
|
|
}
|
|
shardUser := &auth.User{ID: "acceptance-shard-user", APIKeyID: "acceptance-shard-key"}
|
|
if authorizedRunID, err := db.AuthorizeAcceptanceTask(ctx, run.ID, token, shardUser); err != nil || authorizedRunID != run.ID {
|
|
t.Fatalf("authorize acceptance shard run=%q err=%v", authorizedRunID, err)
|
|
}
|
|
unpairedShard := &auth.User{ID: "acceptance-shard-user", APIKeyID: "acceptance-api-key"}
|
|
if _, err := db.AuthorizeAcceptanceTask(ctx, run.ID, token, unpairedShard); !errors.Is(err, ErrAcceptanceNotAuthorized) {
|
|
t.Fatalf("unpaired acceptance shard error=%v, want unauthorized", err)
|
|
}
|
|
if baseURL, credential, err := db.AcceptanceCandidateOverride(ctx, run.ID); err != nil ||
|
|
baseURL != "http://acceptance-emulator:8090" || credential == "" || credential == token {
|
|
t.Fatalf("candidate override base=%q credential=%q err=%v", baseURL, credential, err)
|
|
}
|
|
capacityProfiles := []CapacityProfileInput{
|
|
{
|
|
Kind: "image", StableThroughputPerSecond: 10, AdmittedThroughputPerSecond: 8,
|
|
MaxRunning: 48, MaxQueued: 480, MaxPredictedWaitSeconds: 60,
|
|
},
|
|
{
|
|
Kind: "video", StableThroughputPerSecond: 1, AdmittedThroughputPerSecond: 0.8,
|
|
MaxRunning: 48, MaxQueued: 96, MaxPredictedWaitSeconds: 120,
|
|
},
|
|
}
|
|
if _, err := db.StageAcceptanceCapacityProfiles(ctx, StageAcceptanceCapacityProfilesInput{
|
|
RunID: run.ID, ConfigHash: strings.Repeat("d", 64), CapacityProfiles: capacityProfiles,
|
|
}); err != nil {
|
|
t.Fatalf("stage acceptance capacity profiles: %v", err)
|
|
}
|
|
if _, err := db.FinishAcceptanceRun(ctx, FinishAcceptanceRunInput{
|
|
RunID: run.ID, Passed: true, Report: map[string]any{"queueFinal": 0},
|
|
}); err != nil {
|
|
t.Fatalf("finish acceptance run: %v", err)
|
|
}
|
|
promotion := PromoteAcceptanceRunInput{
|
|
RunID: run.ID, Revision: mode.Revision,
|
|
ReleaseSHA: mode.ReleaseSHA, APIImageDigest: mode.APIImageDigest, WorkerImageDigest: mode.WorkerImageDigest,
|
|
ConfigHash: strings.Repeat("d", 64),
|
|
CapacityProfiles: capacityProfiles,
|
|
}
|
|
stale := promotion
|
|
stale.Revision++
|
|
if _, err := db.PromoteAcceptanceRun(ctx, stale); !errors.Is(err, ErrAcceptanceStateConflict) {
|
|
t.Fatalf("stale promotion error=%v, want state conflict", err)
|
|
}
|
|
live, err := db.PromoteAcceptanceRun(ctx, promotion)
|
|
if err != nil {
|
|
t.Fatalf("promote acceptance run: %v", err)
|
|
}
|
|
if live.Mode != "live" || live.Revision != 2 {
|
|
t.Fatalf("unexpected live mode: %+v", live)
|
|
}
|
|
stalePause := PauseGatewayTrafficInput{
|
|
Revision: live.Revision + 1, ReleaseSHA: live.ReleaseSHA,
|
|
APIImageDigest: live.APIImageDigest, WorkerImageDigest: live.WorkerImageDigest,
|
|
Reason: "integration hard gate",
|
|
}
|
|
if _, err := db.PauseGatewayTraffic(ctx, stalePause); !errors.Is(err, ErrAcceptanceStateConflict) {
|
|
t.Fatalf("stale post-promotion pause error=%v, want state conflict", err)
|
|
}
|
|
stalePause.Revision = live.Revision
|
|
paused, err := db.PauseGatewayTraffic(ctx, stalePause)
|
|
if err != nil {
|
|
t.Fatalf("pause post-promotion traffic: %v", err)
|
|
}
|
|
if paused.Mode != "validation" || paused.RunID != run.ID || paused.Revision != 3 {
|
|
t.Fatalf("unexpected post-promotion pause mode: %+v", paused)
|
|
}
|
|
if _, err := db.AuthorizeAcceptanceTask(ctx, "", "", user); !errors.Is(err, ErrProductionTrafficPaused) {
|
|
t.Fatalf("production request after monitor pause error=%v, want traffic paused", err)
|
|
}
|
|
profiles, err := db.ListCapacityProfiles(ctx)
|
|
if err != nil || len(profiles) < 2 {
|
|
t.Fatalf("certified capacity profiles=%+v err=%v", profiles, err)
|
|
}
|
|
resetTrafficMode()
|
|
|
|
failedRun, err := db.CreateAcceptanceRun(ctx, CreateAcceptanceRunInput{
|
|
ReleaseSHA: strings.Repeat("d", 40),
|
|
APIImageDigest: "sha256:" + strings.Repeat("e", 64),
|
|
WorkerImageDigest: "sha256:" + strings.Repeat("f", 64),
|
|
APIKeyID: "acceptance-api-key",
|
|
UserID: "acceptance-user",
|
|
Token: token + "-retry",
|
|
EmulatorBaseURL: "http://acceptance-emulator:8090",
|
|
CallbackURL: "http://acceptance-emulator:8090/callbacks",
|
|
CapacityProfile: "P24",
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("create retry acceptance run: %v", err)
|
|
}
|
|
failedMode, err := db.ActivateAcceptanceRun(ctx, failedRun.ID)
|
|
if err != nil {
|
|
t.Fatalf("activate retry acceptance run: %v", err)
|
|
}
|
|
createAcceptanceTask := func(label string) GatewayTask {
|
|
t.Helper()
|
|
task, createErr := db.CreateTask(ctx, CreateTaskInput{
|
|
Kind: "images.edits", Model: "acceptance-cleanup-" + label,
|
|
RunMode: "acceptance", AcceptanceRunID: failedRun.ID,
|
|
Request: map[string]any{"prompt": "acceptance cleanup integration test"},
|
|
}, &auth.User{ID: "acceptance-user", Source: "gateway"})
|
|
if createErr != nil {
|
|
t.Fatalf("create acceptance cleanup task: %v", createErr)
|
|
}
|
|
return task
|
|
}
|
|
queuedTask := createAcceptanceTask("queued")
|
|
runningTask := createAcceptanceTask("running")
|
|
notSubmittedTask := createAcceptanceTask("not-submitted")
|
|
submittingTask := createAcceptanceTask("submitting")
|
|
var billingUserID string
|
|
if err := db.pool.QueryRow(ctx, `
|
|
INSERT INTO gateway_users (user_key, username)
|
|
VALUES ($1, $1)
|
|
RETURNING id::text`, "acceptance-cleanup-"+time.Now().Format("150405.000000000")).Scan(&billingUserID); err != nil {
|
|
t.Fatalf("create acceptance cleanup billing user: %v", err)
|
|
}
|
|
if _, err := db.pool.Exec(ctx, `
|
|
UPDATE gateway_tasks
|
|
SET gateway_user_id = $2::uuid,
|
|
reservation_amount = 1,
|
|
billing_status = 'not_started'
|
|
WHERE id = $1::uuid`, queuedTask.ID, billingUserID); err != nil {
|
|
t.Fatalf("reserve acceptance cleanup task billing: %v", err)
|
|
}
|
|
if _, err := db.ClaimTaskExecution(ctx, runningTask.ID, "10000000-0000-4000-8000-000000000001", time.Minute); err != nil {
|
|
t.Fatalf("claim safe running acceptance task: %v", err)
|
|
}
|
|
notSubmittedClaim, err := db.ClaimTaskExecution(ctx, notSubmittedTask.ID, "10000000-0000-4000-8000-000000000002", time.Minute)
|
|
if err != nil {
|
|
t.Fatalf("claim not-submitted acceptance task: %v", err)
|
|
}
|
|
notSubmittedAttemptID, err := db.CreateTaskAttempt(ctx, CreateTaskAttemptInput{
|
|
TaskID: notSubmittedTask.ID, ExecutionToken: notSubmittedClaim.ExecutionToken,
|
|
AttemptNo: 1, Status: "running", Simulated: true,
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("create not-submitted acceptance task attempt: %v", err)
|
|
}
|
|
submittingClaim, err := db.ClaimTaskExecution(ctx, submittingTask.ID, "10000000-0000-4000-8000-000000000003", time.Minute)
|
|
if err != nil {
|
|
t.Fatalf("claim submitting acceptance task: %v", err)
|
|
}
|
|
submittingAttemptID, err := db.CreateTaskAttempt(ctx, CreateTaskAttemptInput{
|
|
TaskID: submittingTask.ID, ExecutionToken: submittingClaim.ExecutionToken,
|
|
AttemptNo: 1, Status: "running", Simulated: true,
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("create submitting acceptance task attempt: %v", err)
|
|
}
|
|
if _, err := db.pool.Exec(ctx, `
|
|
UPDATE gateway_task_attempts
|
|
SET upstream_submission_status = 'submitting', upstream_submission_updated_at = now()
|
|
WHERE id = $1::uuid`, submittingAttemptID); err != nil {
|
|
t.Fatalf("mark acceptance task attempt submitting: %v", err)
|
|
}
|
|
if passed, err := db.FinishAcceptanceRun(ctx, FinishAcceptanceRunInput{
|
|
RunID: failedRun.ID, Passed: true, Report: map[string]any{"testsPassed": true},
|
|
}); err != nil || passed.Status != "passed" {
|
|
t.Fatalf("mark acceptance run passed before external certification=%+v err=%v", passed, err)
|
|
}
|
|
if _, err := db.PromoteAcceptanceRun(ctx, PromoteAcceptanceRunInput{
|
|
RunID: failedRun.ID, Revision: failedMode.Revision,
|
|
ReleaseSHA: failedMode.ReleaseSHA, APIImageDigest: failedMode.APIImageDigest,
|
|
WorkerImageDigest: failedMode.WorkerImageDigest,
|
|
ConfigHash: strings.Repeat("d", 64),
|
|
CapacityProfiles: capacityProfiles,
|
|
}); !errors.Is(err, ErrAcceptancePromotionGates) {
|
|
t.Fatalf("unstaged capacity promotion error=%v, want promotion gates", err)
|
|
}
|
|
if failed, err := db.FinishAcceptanceRun(ctx, FinishAcceptanceRunInput{
|
|
RunID: failedRun.ID, Passed: false, FailureReason: "load failed",
|
|
}); err != nil || failed.Status != "failed" {
|
|
t.Fatalf("downgrade unpromoted acceptance run after external gate failure=%+v err=%v", failed, err)
|
|
}
|
|
var failedRunCancelled int
|
|
if err := db.pool.QueryRow(ctx, `
|
|
SELECT count(*)
|
|
FROM gateway_tasks
|
|
WHERE id = ANY($1::uuid[])
|
|
AND status = 'cancelled'
|
|
AND error_code = 'acceptance_run_aborted'`,
|
|
[]string{queuedTask.ID, runningTask.ID, notSubmittedTask.ID},
|
|
).Scan(&failedRunCancelled); err != nil {
|
|
t.Fatalf("read failed acceptance cleanup states: %v", err)
|
|
}
|
|
if failedRunCancelled != 3 {
|
|
t.Fatalf("failed acceptance cleanup cancelled=%d, want 3", failedRunCancelled)
|
|
}
|
|
if retried, err := db.RetryAcceptanceRun(ctx, failedRun.ID); err != nil || retried.Status != "running" {
|
|
t.Fatalf("retry acceptance run=%+v err=%v", retried, err)
|
|
}
|
|
aborted, err := db.AbortAcceptanceRun(ctx, PromoteAcceptanceRunInput{
|
|
RunID: failedRun.ID, Revision: failedMode.Revision,
|
|
ReleaseSHA: failedMode.ReleaseSHA, APIImageDigest: failedMode.APIImageDigest,
|
|
WorkerImageDigest: failedMode.WorkerImageDigest,
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("abort acceptance run: %v", err)
|
|
}
|
|
if aborted.Mode != "live" || aborted.Revision != failedMode.Revision+1 {
|
|
t.Fatalf("unexpected aborted mode: %+v", aborted)
|
|
}
|
|
var safeCancelled, submittingRunning, notSubmittedAttempts, releaseEvents int
|
|
if err := db.pool.QueryRow(ctx, `
|
|
SELECT
|
|
count(*) FILTER (
|
|
WHERE id = ANY($1::uuid[])
|
|
AND status = 'cancelled'
|
|
AND error_code = 'acceptance_run_aborted'
|
|
),
|
|
count(*) FILTER (
|
|
WHERE id = $2::uuid
|
|
AND status = 'running'
|
|
),
|
|
(SELECT count(*) FROM gateway_task_attempts WHERE id = $3::uuid),
|
|
(SELECT count(*)
|
|
FROM settlement_outbox
|
|
WHERE task_id = $4::uuid
|
|
AND action = 'release'
|
|
AND status = 'pending')
|
|
FROM gateway_tasks`,
|
|
[]string{queuedTask.ID, runningTask.ID, notSubmittedTask.ID},
|
|
submittingTask.ID,
|
|
notSubmittedAttemptID,
|
|
queuedTask.ID,
|
|
).Scan(&safeCancelled, &submittingRunning, ¬SubmittedAttempts, &releaseEvents); err != nil {
|
|
t.Fatalf("read aborted acceptance task states: %v", err)
|
|
}
|
|
if safeCancelled != 3 || submittingRunning != 1 || notSubmittedAttempts != 0 || releaseEvents != 1 {
|
|
t.Fatalf(
|
|
"abort cleanup safe_cancelled=%d submitting_running=%d not_submitted_attempts=%d release_events=%d",
|
|
safeCancelled,
|
|
submittingRunning,
|
|
notSubmittedAttempts,
|
|
releaseEvents,
|
|
)
|
|
}
|
|
if _, err := db.pool.Exec(ctx, `DELETE FROM gateway_tasks WHERE id = ANY($1::uuid[])`,
|
|
[]string{queuedTask.ID, runningTask.ID, notSubmittedTask.ID, submittingTask.ID}); err != nil {
|
|
t.Fatalf("delete acceptance cleanup tasks: %v", err)
|
|
}
|
|
if _, err := db.pool.Exec(ctx, `DELETE FROM gateway_users WHERE id = $1::uuid`, billingUserID); err != nil {
|
|
t.Fatalf("delete acceptance cleanup billing user: %v", err)
|
|
}
|
|
}
|