diff --git a/README.md b/README.md index bbccf26..7beb1c1 100644 --- a/README.md +++ b/README.md @@ -168,7 +168,7 @@ AI_GATEWAY_DATABASE_URL=postgresql://easyai:easyai2025@localhost:5432/easyai_ai_ 如果现有 `easyai-pgvector` 没有把 `5432` 映射到宿主机，就需要补端口映射，或者把 AI Gateway 后端容器化后接入同一个 `easyai` Docker network。 -异步队列 worker 不使用固定业务并发。服务分别汇总启用平台模型和活跃用户组的有效 `concurrent` 策略，采用两者中更严格的集群容量，默认每 5 秒在线调整 River 执行容量。策略解析兼容历史 `platformLimits/modelLimits.max_concurrent_requests`，运行时统一转换为 `rules`。`AI_GATEWAY_ASYNC_WORKER_HARD_LIMIT` 默认 `2048`，限制策略推导出的集群目标；`AI_GATEWAY_ASYNC_WORKER_INSTANCE_HARD_LIMIT` 默认 `32`，按单 Worker 的内存安全容量限制实例分配，即使其他实例失活也不会突破。Worker 的 `AI_GATEWAY_DATABASE_MAX_CONNS` 必须高于实例执行容量，为心跳、选主、租约续期和健康检查保留连接；生产环境按每实例执行容量 `24` 配置连接池上限 `32`。`AI_GATEWAY_ASYNC_ADMISSION_MICROBATCH_SIZE` 默认 `8`，把短任务准入、容量租约和唯一 River job 合并为有界原子微批次，摊薄跨地域同步提交，同时禁止重新形成整窗大事务。`AI_GATEWAY_DATABASE_MIN_IDLE_CONNS` 控制启动预热连接数，默认 `0`，生产 K3s 配置为 `4`；`AI_GATEWAY_DATABASE_MAX_CONN_IDLE_SECONDS` 生产配置为 `300` 秒。平台模型和用户组的 PostgreSQL concurrency lease 仍是业务并发真值。可通过 `AI_GATEWAY_ASYNC_WORKER_REFRESH_INTERVAL_SECONDS` 调整刷新周期。 +异步队列 worker 不使用固定业务并发。服务分别汇总启用平台模型和活跃用户组的有效 `concurrent` 策略，采用两者中更严格的集群容量，默认每 5 秒在线调整 River 执行容量。策略解析兼容历史 `platformLimits/modelLimits.max_concurrent_requests`，运行时统一转换为 `rules`。`AI_GATEWAY_ASYNC_WORKER_HARD_LIMIT` 默认 `2048`，限制策略推导出的集群目标；`AI_GATEWAY_ASYNC_WORKER_INSTANCE_HARD_LIMIT` 默认 `32`，按单 Worker 的内存安全容量限制实例分配，即使其他实例失活也不会突破。Worker 的 `AI_GATEWAY_DATABASE_MAX_CONNS` 必须高于实例执行容量，为心跳、选主、租约续期和健康检查保留连接；生产环境按每实例执行容量 `24` 配置连接池上限 `32`。`AI_GATEWAY_ASYNC_ADMISSION_MICROBATCH_SIZE` 默认 `8`，把短任务准入、容量租约和唯一 River job 合并为有界原子微批次，摊薄跨地域同步提交，同时禁止重新形成整窗大事务。异步任务首次排队时会持久化已经鉴权的候选和 admission scope 快照；Worker 批量读取快照并仅刷新动态执行容量，只有显式重新选路才重新水化媒体和计算候选，避免队列越大、重复路由查询越多。`AI_GATEWAY_DATABASE_MIN_IDLE_CONNS` 控制启动预热连接数，默认 `0`，生产 K3s 配置为 `4`；`AI_GATEWAY_DATABASE_MAX_CONN_IDLE_SECONDS` 生产配置为 `300` 秒。平台模型和用户组的 PostgreSQL concurrency lease 仍是业务并发真值。可通过 `AI_GATEWAY_ASYNC_WORKER_REFRESH_INTERVAL_SECONDS` 调整刷新周期。 ## 迁移原则 diff --git a/apps/api/internal/runner/admission.go b/apps/api/internal/runner/admission.go index 51c6dfb..40bdefb 100644 --- a/apps/api/internal/runner/admission.go +++ b/apps/api/internal/runner/admission.go @@ -34,8 +34,8 @@ const ( asyncWorkerQueueLimit = 10000 asyncWorkerMaxWaitSeconds = 24 * 60 * 60 // The dispatcher scans enough FIFO entries to fill the certified 2 x P32 - // capacity. Store admission commits each item independently so this scan - // size is never also a multi-task transaction size. + // capacity. Durable scope snapshots make this one bulk read; atomic commits + // remain bounded by the separately configured microbatch size. asyncAdmissionBatchMax = 64 acceptanceQueueLimit = 10000 acceptanceQueueMaxWait = 15 * 60 @@ -690,48 +690,80 @@ func (s *Service) SubmitAsyncTask(ctx context.Context, task store.GatewayTask) e return nil } -func (s *Service) dispatchWaitingAsyncTasks(ctx context.Context, tasks []store.GatewayTask) (bool, error) { - if len(tasks) == 0 { +func admissionInputFromSnapshot(admission store.TaskAdmission, workerCapacity int) store.TaskAdmissionInput { + scopes := append([]store.AdmissionScope(nil), admission.Scopes...) + for index := range scopes { + if scopes[index].ScopeType == "worker_capacity" && scopes[index].ScopeKey == asyncWorkerCapacityScopeKey { + scopes[index].ConcurrentLimit = float64(workerCapacity) + } + } + return store.TaskAdmissionInput{ + TaskID: admission.TaskID, + PlatformID: admission.PlatformID, + PlatformModelID: admission.PlatformModelID, + UserGroupID: admission.UserGroupID, + QueueKey: admission.QueueKey, + Mode: admission.Mode, + Priority: admission.Priority, + Scopes: scopes, + } +} + +func (s *Service) dispatchWaitingAsyncTasks(ctx context.Context, admissions []store.TaskAdmission) (bool, error) { + if len(admissions) == 0 { return false, nil } - inputs := make([]store.TaskAdmissionInput, 0, len(tasks)) - tasksByID := make(map[string]store.GatewayTask, len(tasks)) - for _, task := range tasks { - user := authUserFromTask(task) - current, err := s.store.GetTaskAdmission(ctx, task.ID) - if err != nil { - if errors.Is(err, pgx.ErrNoRows) { - continue - } - return false, err - } - plan, err := s.buildTaskAdmissionPlanForCurrentBinding(ctx, task, user, ¤t) - if err != nil { - return false, err - } - if !plan.Eligible { - if err := s.EnqueueAsyncTask(ctx, task); err != nil { - return false, err - } - continue - } - input, err := s.prepareTaskAdmissionInput(ctx, task, plan, "") - if err != nil { - return false, err - } - inputs = append(inputs, input) - tasksByID[task.ID] = task + workerCapacity, err := s.coordinationStore.ActiveWorkerCapacity(ctx) + if err != nil { + return false, err + } + if workerCapacity <= 0 { + return true, nil } - saturated := false microbatchSize := s.cfg.AsyncAdmissionMicrobatchSize if microbatchSize <= 0 { microbatchSize = 1 } - for start := 0; start < len(inputs) && !saturated; start += microbatchSize { - end := min(start+microbatchSize, len(inputs)) + for start := 0; start < len(admissions); start += microbatchSize { + end := min(start+microbatchSize, len(admissions)) + inputs := make([]store.TaskAdmissionInput, 0, end-start) + tasksByID := make(map[string]store.GatewayTask, end-start) + for _, admission := range admissions[start:end] { + if admission.ReselectRequestedAt.IsZero() && len(admission.Scopes) > 0 { + inputs = append(inputs, admissionInputFromSnapshot(admission, workerCapacity)) + tasksByID[admission.TaskID] = store.GatewayTask{ID: admission.TaskID} + continue + } + task, loadErr := s.store.GetTask(ctx, admission.TaskID) + if loadErr != nil { + if errors.Is(loadErr, pgx.ErrNoRows) { + continue + } + return false, loadErr + } + plan, planErr := s.buildTaskAdmissionPlanForCurrentBinding(ctx, task, authUserFromTask(task), &admission) + if planErr != nil { + return false, planErr + } + if !plan.Eligible { + if enqueueErr := s.EnqueueAsyncTask(ctx, task); enqueueErr != nil { + return false, enqueueErr + } + continue + } + input := taskAdmissionInput(task, plan, "") + if _, rebindErr := s.store.RebindWaitingTaskAdmission(ctx, input); rebindErr != nil && !errors.Is(rebindErr, pgx.ErrNoRows) { + return false, rebindErr + } + inputs = append(inputs, input) + tasksByID[task.ID] = task + } + if len(inputs) == 0 { + continue + } outcomes, err := s.store.TryTaskAdmissionAtomicBatchWithAdmittedHook( ctx, - inputs[start:end], + inputs, func(tx pgx.Tx, input store.TaskAdmissionInput) error { task, ok := tasksByID[input.TaskID] if !ok { @@ -753,12 +785,11 @@ func (s *Service) dispatchWaitingAsyncTasks(ctx context.Context, tasks []store.G return false, outcome.Err } if !outcome.Result.Admitted { - saturated = true - break + return true, nil } } } - return saturated, nil + return false, nil } func (s *Service) cancelAsyncSubmissionIfDisconnected(ctx context.Context, taskID string) bool { @@ -789,34 +820,17 @@ func (s *Service) dispatchWaitingAsyncAdmissions(ctx context.Context) { } for { batchLimit := asyncAdmissionBatchLimit(s.cfg.AsyncWorkerHardLimit) - taskIDs, err := s.store.ListWaitingAsyncAdmissionTaskIDs(ctx, batchLimit) + admissions, err := s.store.ListWaitingAsyncAdmissions(ctx, batchLimit) if err != nil { s.logger.Warn("list waiting async admissions failed", "error", err) break } - if len(taskIDs) == 0 { + if len(admissions) == 0 { break } - tasks := make([]store.GatewayTask, 0, len(taskIDs)) - for _, taskID := range taskIDs { - task, err := s.store.GetTask(ctx, taskID) - if err != nil { - if !errors.Is(err, pgx.ErrNoRows) { - s.logger.Warn("load waiting async task failed", "taskID", taskID, "error", err) - } - continue - } - if task.Status != "queued" { - continue - } - tasks = append(tasks, task) - } - if len(tasks) == 0 { - break - } - saturated, err := s.dispatchWaitingAsyncTasks(ctx, tasks) + saturated, err := s.dispatchWaitingAsyncTasks(ctx, admissions) if err != nil { - s.logger.Warn("dispatch waiting async admission batch failed", "tasks", len(tasks), "error", err) + s.logger.Warn("dispatch waiting async admission batch failed", "tasks", len(admissions), "error", err) break } if saturated { diff --git a/apps/api/internal/runner/admission_test.go b/apps/api/internal/runner/admission_test.go index 2a7b842..357edd2 100644 --- a/apps/api/internal/runner/admission_test.go +++ b/apps/api/internal/runner/admission_test.go @@ -2,10 +2,43 @@ package runner import ( "testing" + "time" "github.com/easyai/easyai-ai-gateway/apps/api/internal/store" ) +func TestAdmissionInputFromSnapshotRefreshesWorkerCapacity(t *testing.T) { + admission := store.TaskAdmission{ + TaskID: "task-1", + PlatformID: "platform-1", + PlatformModelID: "model-1", + UserGroupID: "group-1", + QueueKey: "queue-1", + Mode: "async", + Priority: 7, + Scopes: []store.AdmissionScope{ + {ScopeType: "platform_model", ScopeKey: "model-1", ConcurrentLimit: 10}, + {ScopeType: "worker_capacity", ScopeKey: "global", ConcurrentLimit: 48}, + }, + ReselectRequestedAt: time.Time{}, + } + + input := admissionInputFromSnapshot(admission, 24) + + if input.TaskID != admission.TaskID || input.PlatformModelID != admission.PlatformModelID || len(input.Scopes) != 2 { + t.Fatalf("unexpected admission input: %+v", input) + } + if input.Scopes[1].ConcurrentLimit != 24 { + t.Fatalf("worker capacity=%v, want 24", input.Scopes[1].ConcurrentLimit) + } + if input.Scopes[0].ConcurrentLimit != 10 { + t.Fatalf("business scope changed: %+v", input.Scopes[0]) + } + if admission.Scopes[1].ConcurrentLimit != 48 { + t.Fatalf("snapshot was mutated: %+v", admission.Scopes[1]) + } +} + func TestDistributedAdmissionModelTypeBoundary(t *testing.T) { for _, modelType := range []string{ "image_generate", "image_edit", "image_analysis", "image_vectorize", diff --git a/apps/api/internal/store/admission_queue.go b/apps/api/internal/store/admission_queue.go index d4b7bcf..96c15de 100644 --- a/apps/api/internal/store/admission_queue.go +++ b/apps/api/internal/store/admission_queue.go @@ -3,6 +3,7 @@ package store import ( "context" "database/sql" + "encoding/json" "errors" "fmt" "math" @@ -64,19 +65,21 @@ type TaskAdmissionInput struct { } type TaskAdmission struct { - TaskID string - PlatformID string - PlatformModelID string - UserGroupID string - QueueKey string - Mode string - Status string - Priority int - EnqueuedAt time.Time - WaitDeadlineAt time.Time - AdmittedAt time.Time - WaiterID string - WaiterLeaseExpires time.Time + TaskID string + PlatformID string + PlatformModelID string + UserGroupID string + QueueKey string + Mode string + Status string + Priority int + EnqueuedAt time.Time + WaitDeadlineAt time.Time + AdmittedAt time.Time + WaiterID string + WaiterLeaseExpires time.Time + Scopes []AdmissionScope + ReselectRequestedAt time.Time } type TaskAdmissionResult struct { @@ -697,6 +700,10 @@ func (s *Store) rebindWaitingTaskAdmissionOnce(ctx context.Context, input TaskAd return TaskAdmission{}, admissionErr } deadline := admissionDeadline(scopes) + scopeSnapshot, err := json.Marshal(scopes) + if err != nil { + return TaskAdmission{}, fmt.Errorf("marshal rebound task admission scopes: %w", err) + } row := tx.QueryRow(ctx, ` UPDATE gateway_task_admissions SET platform_id = $2::uuid, @@ -704,6 +711,7 @@ SET platform_id = $2::uuid, user_group_id = NULLIF($4, '')::uuid, queue_key = $5, wait_deadline_at = LEAST(wait_deadline_at, $6), + scope_snapshot = $7::jsonb, reselect_requested_at = NULL, updated_at = now() WHERE task_id = $1::uuid @@ -711,13 +719,15 @@ WHERE task_id = $1::uuid RETURNING task_id::text, platform_id::text, platform_model_id::text, COALESCE(user_group_id::text, ''), queue_key, mode, status, priority, enqueued_at, wait_deadline_at, admitted_at, - COALESCE(waiter_id, ''), waiter_lease_expires_at`, + COALESCE(waiter_id, ''), waiter_lease_expires_at, + scope_snapshot, reselect_requested_at`, input.TaskID, input.PlatformID, input.PlatformModelID, input.UserGroupID, input.QueueKey, deadline, + scopeSnapshot, ) migrated, err := scanTaskAdmission(row) if err != nil { @@ -753,6 +763,10 @@ WHERE task_id = $1::uuid if input.Mode == "sync" { waiterLease = time.Now().Add(admissionWaiterLeaseTTL) } + scopeSnapshot, err := json.Marshal(normalizedAdmissionScopes(input.Scopes)) + if err != nil { + return TaskAdmission{}, fmt.Errorf("marshal reset task admission scopes: %w", err) + } row := tx.QueryRow(ctx, ` UPDATE gateway_task_admissions SET status = 'waiting', @@ -764,12 +778,14 @@ SET status = 'waiting', wait_deadline_at = LEAST(wait_deadline_at, $6), waiter_id = NULLIF($7, ''), waiter_lease_expires_at = $8, + scope_snapshot = $9::jsonb, updated_at = now() WHERE task_id = $1::uuid RETURNING task_id::text, platform_id::text, platform_model_id::text, COALESCE(user_group_id::text, ''), queue_key, mode, status, priority, enqueued_at, wait_deadline_at, admitted_at, - COALESCE(waiter_id, ''), waiter_lease_expires_at`, + COALESCE(waiter_id, ''), waiter_lease_expires_at, + scope_snapshot, reselect_requested_at`, input.TaskID, input.PlatformID, input.PlatformModelID, @@ -778,6 +794,7 @@ RETURNING task_id::text, platform_id::text, platform_model_id::text, admissionDeadline(input.Scopes), strings.TrimSpace(input.WaiterID), waiterLease, + scopeSnapshot, ) return scanTaskAdmission(row) } @@ -933,6 +950,62 @@ LIMIT $1`, limit) return taskIDs, rows.Err() } +// ListWaitingAsyncAdmissions returns the durable routing and admission-scope +// snapshots needed by the Worker dispatcher in one round trip. Normal queued +// tasks must not be fully rehydrated and routed again merely to reserve an +// execution slot; only rows explicitly marked for reselection take that slower +// path in the runner. +func (s *Store) ListWaitingAsyncAdmissions(ctx context.Context, limit int) ([]TaskAdmission, error) { + if limit <= 0 || limit > 1000 { + limit = 100 + } + rows, err := s.pool.Query(ctx, ` +SELECT admission.task_id::text, admission.platform_id::text, admission.platform_model_id::text, + COALESCE(admission.user_group_id::text, ''), admission.queue_key, admission.mode, + admission.status, admission.priority, admission.enqueued_at, admission.wait_deadline_at, + admission.admitted_at, COALESCE(admission.waiter_id, ''), admission.waiter_lease_expires_at, + admission.scope_snapshot, admission.reselect_requested_at +FROM gateway_task_admissions admission +JOIN gateway_tasks task ON task.id = admission.task_id +LEFT JOIN river_job job ON job.id = task.river_job_id +WHERE admission.mode = 'async' + AND task.status = 'queued' + AND task.next_run_at <= now() + AND ( + admission.status = 'waiting' + OR ( + admission.status = 'admitted' + AND NOT EXISTS ( + SELECT 1 + FROM gateway_concurrency_leases lease + WHERE lease.task_id = admission.task_id + AND lease.released_at IS NULL + AND lease.expires_at > now() + ) + ) + ) + AND ( + task.river_job_id IS NULL + OR job.id IS NULL + OR job.state <> 'running' + ) +ORDER BY admission.priority ASC, admission.enqueued_at ASC, admission.task_id ASC +LIMIT $1`, limit) + if err != nil { + return nil, err + } + defer rows.Close() + admissions := make([]TaskAdmission, 0, limit) + for rows.Next() { + admission, scanErr := scanTaskAdmission(rows) + if scanErr != nil { + return nil, scanErr + } + admissions = append(admissions, admission) + } + return admissions, rows.Err() +} + // ListWaitingTaskAdmissionIDs returns one FIFO leader for every independent // platform-model queue, additionally requiring the task to lead its user-group // queue when one exists. It is used after capacity is released so API @@ -1323,6 +1396,10 @@ func queueRetryAfter(next time.Time) time.Duration { func insertTaskAdmissionTx(ctx context.Context, tx pgx.Tx, input TaskAdmissionInput, deadline time.Time) (TaskAdmission, error) { waiterID := strings.TrimSpace(input.WaiterID) + scopeSnapshot, err := json.Marshal(normalizedAdmissionScopes(input.Scopes)) + if err != nil { + return TaskAdmission{}, fmt.Errorf("marshal task admission scopes: %w", err) + } var waiterLease any if input.Mode == "sync" { waiterLease = time.Now().Add(admissionWaiterLeaseTTL) @@ -1330,22 +1407,27 @@ func insertTaskAdmissionTx(ctx context.Context, tx pgx.Tx, input TaskAdmissionIn row := tx.QueryRow(ctx, ` INSERT INTO gateway_task_admissions ( task_id, platform_id, platform_model_id, user_group_id, queue_key, mode, status, - priority, wait_deadline_at, waiter_id, waiter_lease_expires_at + priority, wait_deadline_at, waiter_id, waiter_lease_expires_at, scope_snapshot ) VALUES ( $1::uuid, $2::uuid, $3::uuid, NULLIF($4, '')::uuid, $5, $6, 'waiting', - $7, $8, NULLIF($9, ''), $10 + $7, $8, NULLIF($9, ''), $10, $11::jsonb ) ON CONFLICT (task_id) DO UPDATE SET waiter_id = CASE WHEN gateway_task_admissions.mode = 'sync' THEN EXCLUDED.waiter_id ELSE gateway_task_admissions.waiter_id END, waiter_lease_expires_at = CASE WHEN gateway_task_admissions.mode = 'sync' THEN EXCLUDED.waiter_lease_expires_at ELSE gateway_task_admissions.waiter_lease_expires_at END, + scope_snapshot = CASE + WHEN gateway_task_admissions.scope_snapshot IS NULL OR gateway_task_admissions.scope_snapshot = '[]'::jsonb THEN EXCLUDED.scope_snapshot + ELSE gateway_task_admissions.scope_snapshot + END, updated_at = now() RETURNING task_id::text, platform_id::text, platform_model_id::text, COALESCE(user_group_id::text, ''), queue_key, mode, status, priority, enqueued_at, wait_deadline_at, admitted_at, - COALESCE(waiter_id, ''), waiter_lease_expires_at`, + COALESCE(waiter_id, ''), waiter_lease_expires_at, + scope_snapshot, reselect_requested_at`, input.TaskID, input.PlatformID, input.PlatformModelID, input.UserGroupID, - input.QueueKey, input.Mode, input.Priority, deadline, waiterID, waiterLease) + input.QueueKey, input.Mode, input.Priority, deadline, waiterID, waiterLease, scopeSnapshot) return scanTaskAdmission(row) } @@ -1361,7 +1443,8 @@ WHERE task_id = $1::uuid RETURNING task_id::text, platform_id::text, platform_model_id::text, COALESCE(user_group_id::text, ''), queue_key, mode, status, priority, enqueued_at, wait_deadline_at, admitted_at, - COALESCE(waiter_id, ''), waiter_lease_expires_at`, + COALESCE(waiter_id, ''), waiter_lease_expires_at, + scope_snapshot, reselect_requested_at`, taskID, waiterID, admissionWaiterLeaseTTL.String()) return scanTaskAdmission(row) } @@ -1377,7 +1460,8 @@ WHERE task_id = $1::uuid AND status = 'waiting' RETURNING task_id::text, platform_id::text, platform_model_id::text, COALESCE(user_group_id::text, ''), queue_key, mode, status, priority, enqueued_at, wait_deadline_at, admitted_at, - COALESCE(waiter_id, ''), waiter_lease_expires_at`, + COALESCE(waiter_id, ''), waiter_lease_expires_at, + scope_snapshot, reselect_requested_at`, taskID) return scanTaskAdmission(row) } @@ -1395,7 +1479,8 @@ func loadTaskAdmission(ctx context.Context, q admissionQuerier, taskID string) ( SELECT task_id::text, platform_id::text, platform_model_id::text, COALESCE(user_group_id::text, ''), queue_key, mode, status, priority, enqueued_at, wait_deadline_at, admitted_at, - COALESCE(waiter_id, ''), waiter_lease_expires_at + COALESCE(waiter_id, ''), waiter_lease_expires_at, + scope_snapshot, reselect_requested_at FROM gateway_task_admissions WHERE task_id = $1::uuid`, taskID) admission, err := scanTaskAdmission(row) @@ -1409,6 +1494,8 @@ func scanTaskAdmission(row pgx.Row) (TaskAdmission, error) { var admission TaskAdmission var admittedAt sql.NullTime var waiterLeaseExpires sql.NullTime + var reselectRequestedAt sql.NullTime + var scopeSnapshot []byte err := row.Scan( &admission.TaskID, &admission.PlatformID, @@ -1423,6 +1510,8 @@ func scanTaskAdmission(row pgx.Row) (TaskAdmission, error) { &admittedAt, &admission.WaiterID, &waiterLeaseExpires, + &scopeSnapshot, + &reselectRequestedAt, ) if admittedAt.Valid { admission.AdmittedAt = admittedAt.Time @@ -1430,6 +1519,14 @@ func scanTaskAdmission(row pgx.Row) (TaskAdmission, error) { if waiterLeaseExpires.Valid { admission.WaiterLeaseExpires = waiterLeaseExpires.Time } + if reselectRequestedAt.Valid { + admission.ReselectRequestedAt = reselectRequestedAt.Time + } + if err == nil && len(scopeSnapshot) > 0 { + if unmarshalErr := json.Unmarshal(scopeSnapshot, &admission.Scopes); unmarshalErr != nil { + return TaskAdmission{}, fmt.Errorf("unmarshal task admission scopes: %w", unmarshalErr) + } + } return admission, err } diff --git a/apps/api/internal/store/admission_queue_integration_test.go b/apps/api/internal/store/admission_queue_integration_test.go index b0cbca1..bc96041 100644 --- a/apps/api/internal/store/admission_queue_integration_test.go +++ b/apps/api/internal/store/admission_queue_integration_test.go @@ -434,6 +434,23 @@ WHERE id = $1::uuid`, queuedAtomicTask.ID, queuedSyntheticRiverJobID) if err != nil || queuedAdmission.Status != "waiting" { t.Fatalf("atomic queued admission=%+v err=%v, want waiting", queuedAdmission, err) } + if len(queuedAdmission.Scopes) == 0 { + t.Fatal("queued admission did not persist its scope snapshot") + } + waitingAdmissionSnapshots, err := first.ListWaitingAsyncAdmissions(ctx, 1000) + if err != nil { + t.Fatalf("list waiting async admission snapshots: %v", err) + } + var listedSnapshot *TaskAdmission + for index := range waitingAdmissionSnapshots { + if waitingAdmissionSnapshots[index].TaskID == queuedAtomicTask.ID { + listedSnapshot = &waitingAdmissionSnapshots[index] + break + } + } + if listedSnapshot == nil || len(listedSnapshot.Scopes) != len(queuedAdmission.Scopes) { + t.Fatalf("listed admission snapshot=%+v, want %d scopes", listedSnapshot, len(queuedAdmission.Scopes)) + } var riverJobID int64 if err := first.pool.QueryRow(ctx, ` SELECT diff --git a/apps/api/migrations/0100_task_admission_scope_snapshot.sql b/apps/api/migrations/0100_task_admission_scope_snapshot.sql new file mode 100644 index 0000000..e6b416d --- /dev/null +++ b/apps/api/migrations/0100_task_admission_scope_snapshot.sql @@ -0,0 +1,9 @@ +ALTER TABLE gateway_task_admissions + ADD COLUMN IF NOT EXISTS scope_snapshot jsonb; + +ALTER TABLE gateway_task_admissions + ADD CONSTRAINT gateway_task_admissions_scope_snapshot_array_check + CHECK (scope_snapshot IS NULL OR jsonb_typeof(scope_snapshot) = 'array') NOT VALID; + +ALTER TABLE gateway_task_admissions + VALIDATE CONSTRAINT gateway_task_admissions_scope_snapshot_array_check; diff --git a/deploy/kubernetes/acceptance/protocol-emulator.yaml b/deploy/kubernetes/acceptance/protocol-emulator.yaml index aa3660c..43469f1 100644 --- a/deploy/kubernetes/acceptance/protocol-emulator.yaml +++ b/deploy/kubernetes/acceptance/protocol-emulator.yaml @@ -18,7 +18,11 @@ spec: app.kubernetes.io/part-of: easyai-ai-gateway spec: nodeSelector: - easyai.io/site: ningbo + # Keep the memory-heavy protocol emulator away from PostgreSQL and + # control-plane nodes. Any future dedicated Worker node can host it + # without changing this manifest. + easyai.io/worker: "true" + easyai.io/database: "false" easyai.io/workload: "true" securityContext: runAsNonRoot: true diff --git a/docs/operations/production-acceptance.md b/docs/operations/production-acceptance.md index 0202169..3c33ccb 100644 --- a/docs/operations/production-acceptance.md +++ b/docs/operations/production-acceptance.md @@ -86,26 +86,24 @@ AI_GATEWAY_WORKER_REPLICAS_NINGBO AI_GATEWAY_WORKER_REPLICAS_HONGKONG ``` -`AI_GATEWAY_WORKER_REPLICAS_*` 是发布工具配置，不会传入容器。容量档位先固定宁波、香港 -各一个 2 GiB Worker，再按实时内存、CPU 和 PostgreSQL 预算验证 `1+2`、条件允许时的 -`2+2`，以及安全 drain 回到 `1+1`： - -生产基线使用两个独立物理 Worker 节点：第二台宁波服务器承载 `easyai-worker-ningbo`， -香港服务器承载 `easyai-worker-hongkong`。两个 Deployment 都要求节点带 -`easyai.io/worker=true`，原宁波混部节点不再承载 Worker。默认验收配置为： +`AI_GATEWAY_WORKER_REPLICAS_*` 是发布工具配置，不会传入容器。当前生产基线在香港站点运行 +两个 2 GiB Worker，其中一个位于香港控制面节点，另一个位于带 +`easyai.io/worker=true,easyai.io/database=false` 标签的专用 Worker 节点；宁波混部节点不承载 +Worker。后续副本上限继续按实时内存、CPU 和 PostgreSQL 预算计算，不按站点或代码写死： ```bash -AI_GATEWAY_ACCEPTANCE_BASE_REPLICAS_NINGBO=1 -AI_GATEWAY_ACCEPTANCE_BASE_REPLICAS_HONGKONG=1 -AI_GATEWAY_ACCEPTANCE_AUTOSCALING_MIN_REPLICAS_NINGBO=1 +AI_GATEWAY_ACCEPTANCE_BASE_REPLICAS_NINGBO=0 +AI_GATEWAY_ACCEPTANCE_BASE_REPLICAS_HONGKONG=2 +AI_GATEWAY_ACCEPTANCE_AUTOSCALING_MIN_REPLICAS_NINGBO=0 AI_GATEWAY_ACCEPTANCE_AUTOSCALING_MIN_REPLICAS_HONGKONG=1 -AI_GATEWAY_ACCEPTANCE_AUTOSCALING_MAX_REPLICAS_NINGBO=2 +AI_GATEWAY_ACCEPTANCE_AUTOSCALING_MAX_REPLICAS_NINGBO=0 AI_GATEWAY_ACCEPTANCE_AUTOSCALING_MAX_REPLICAS_HONGKONG=2 ``` -资源前置门禁和容量控制器只采样 `easyai.io/worker=true` 节点，避免把原宁波混部节点的 -可分配资源错误计入弹性上限。每站点能否升到两个副本仍由实测 RSS、CPU 和 PostgreSQL -连接预算决定，不因新增节点而预先承诺。 +资源前置门禁和容量控制器只把已启用站点的 `easyai.io/worker=true` 节点计入弹性预算， +协议模拟器优先固定到 `easyai.io/worker=true,easyai.io/database=false` 的专用节点，避免它与 +宁波主库和控制面争抢内存。每站点能否增加副本仍由实测 RSS、CPU 和 PostgreSQL 连接预算 +决定，不因新增节点而预先承诺。 | 档位 | 单实例执行槽 | 数据库池 | 媒体并发 | 全局执行槽 | |---|---:|---:|---:|---:| @@ -156,7 +154,7 @@ PNG 并通过验收 Key 上传到 Gateway。生产快照默认在已部署的 AP 编辑模型，以及 `omni_video.max_images >= 9` 的视频模型并优先 Seedance 2.0/fast。 验收组按 release SHA 隔离；历史 Run 的用户和分片不会被清理或复用到新 release，本次需要的 分片及其 API Key 会幂等迁移到当前 release 的专属组。 -宁波 15 分钟预热内存门槛默认 65%，可用 +已启用 Worker 节点的 15 分钟预热内存门槛默认 65%，可用 `AI_GATEWAY_ACCEPTANCE_NODE_MEMORY_PRECONDITION_PERCENT` 在 50–79% 之间显式调整；报告必须记录 实际门槛，运行期 80%目标和 85%硬门禁不随之放宽。 diff --git a/scripts/cluster/run-production-acceptance.sh b/scripts/cluster/run-production-acceptance.sh index 4dcde52..7fec103 100755 --- a/scripts/cluster/run-production-acceptance.sh +++ b/scripts/cluster/run-production-acceptance.sh @@ -2630,7 +2630,7 @@ pressure_row_live_gate_reason() { $4 >= 900 {print "oldest_wait_seconds=" $4 " limit=900"; exit} $6 <= 0 {print "postgres_max_connections_unavailable"; exit} $5 * 4 >= $6 * 3 {print "postgres_connections=" $5 " limit_75_percent_of=" $6; exit} - $7 >= 80 {print "node_memory_percent=" $7 " limit=80"; exit} + $7 >= 85 {print "node_memory_percent=" $7 " hard_limit=85"; exit} $8 >= 1536 {print "api_pod_memory_mib=" $8 " limit=1536"; exit} $14 > 0 {print "lease_renewal_failures=" $14; exit} $15 > 0 {print "lease_lost=" $15; exit} diff --git a/tests/release/production-acceptance-script-test.sh b/tests/release/production-acceptance-script-test.sh index dd02bbb..a6ff006 100755 --- a/tests/release/production-acceptance-script-test.sh +++ b/tests/release/production-acceptance-script-test.sh @@ -108,10 +108,12 @@ grep -Fq '# target_host root@hongkong-worker-2.invalid' <<<"$metrics" [[ $(metric_max test_metric <<<"$metrics") == 1 ]] [[ $(metric_sum test_metric <<<"$metrics") == 2 ]] -passing_pressure_row='2026-08-01T00:00:00Z,0,0,899,149,200,79,1535,0,32,32,0,0,0,0,2,48,0,8,8,0,0,0,16,16,0,0,1535,100' +passing_pressure_row='2026-08-01T00:00:00Z,0,0,899,149,200,84,1535,0,32,32,0,0,0,0,2,48,0,8,8,0,0,0,16,16,0,0,1535,100' pressure_row_passes_live_gates "$passing_pressure_row" [[ $(pressure_row_live_gate_reason '2026-08-01T00:00:00Z,0,0,900,149,200,79,1535,0,32,32,0,0,0,0,2,48,0,8,8,0,0,0,16,16,0,0,1535,100') == \ 'oldest_wait_seconds=900 limit=900' ]] +[[ $(pressure_row_live_gate_reason '2026-08-01T00:00:00Z,0,0,899,149,200,85,1535,0,32,32,0,0,0,0,2,48,0,8,8,0,0,0,16,16,0,0,1535,100') == \ + 'node_memory_percent=85 hard_limit=85' ]] for failed_pressure_row in \ '2026-08-01T00:00:00Z,0,0,901,149,200,79,1535,0,32,32,0,0,0,0,2,48,0,8,8,0,0,0,16,16,0,0,1535,100' \