perf(worker): 持久化准入快照提升队列吞吐

原因：线上 P24 验收显示 Worker 在每轮准入前逐条恢复媒体、重算候选和查询 attempts，跨地域查询导致 48 个执行槽长期只能使用约 8 至 16 个。\n\n影响：任务首次排队时保存鉴权后的 admission scope 快照，Worker 单次批量读取并只刷新动态总容量；显式重新选路保留完整慢路径。验收模拟器改到非数据库专用 Worker 节点，运行期 85% 作为立即中止硬门禁，80% 继续作为容量认证目标。\n\n验证：Go 全量测试、go vet、govulncheck、真实 PostgreSQL 迁移与跨 Store 集成测试、ShellCheck、发布及验收脚本、OpenAPI、前端 lint/test/build、pnpm audit high、Compose 与 Kubernetes 渲染均通过。
2026-08-01 21:25:54 +08:00
parent c89c56ca65
commit 56a5176c92
10 changed files with 274 additions and 100 deletions
@@ -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, &current)
-		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 {
@@ -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",