From 17433bf2dafc207a3470f58427fdda98231a192b Mon Sep 17 00:00:00 2001
From: wangbo <wangbolhr@163.com>
Date: Fri, 31 Jul 2026 08:32:39 +0800
Subject: [PATCH] =?UTF-8?q?perf(worker):=20=E6=8C=81=E7=BB=AD=E6=89=B9?=
 =?UTF-8?q?=E9=87=8F=E5=A1=AB=E6=BB=A1=E5=BC=82=E6=AD=A5=E6=89=A7=E8=A1=8C?=
 =?UTF-8?q?=E6=A7=BD?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

异步准入通知在突发提交时会合并，原调度器每次只填充一个 8 任务批次，导致全局容量长期空闲并反向拖慢入口登记。\n\n收到通知或周期唤醒后连续拉取并提交批次，直到队列为空或全局执行容量真正饱和；保留每批 8 个的短事务边界。\n\n验证：env -u AI_GATEWAY_TEST_DATABASE_URL go test ./... -count=1；go vet ./...；gofmt -l 无输出。
---
 apps/api/internal/runner/admission.go | 77 ++++++++++++++++-----------
 1 file changed, 45 insertions(+), 32 deletions(-)

diff --git a/apps/api/internal/runner/admission.go b/apps/api/internal/runner/admission.go
index 2c1660b..d198ee6 100644
--- a/apps/api/internal/runner/admission.go
+++ b/apps/api/internal/runner/admission.go
@@ -704,42 +704,55 @@ func (s *Service) dispatchWaitingAsyncAdmissions(ctx context.Context) {
 		case <-s.asyncAdmissionWake:
 		case <-ticker.C:
 		}
-		batchLimit := s.cfg.AsyncWorkerHardLimit
-		if batchLimit <= 0 {
-			batchLimit = 64
-		}
-		if batchLimit > asyncAdmissionBatchMax {
-			batchLimit = asyncAdmissionBatchMax
-		}
-		taskIDs, err := s.store.ListWaitingAsyncAdmissionTaskIDs(ctx, batchLimit)
-		if err != nil {
-			s.logger.Warn("list waiting async admissions failed", "error", err)
-			continue
-		}
-		tasks := make([]store.GatewayTask, 0, len(taskIDs))
-		for _, taskID := range taskIDs {
-			task, err := s.store.GetTask(ctx, taskID)
+		for {
+			batchLimit := s.cfg.AsyncWorkerHardLimit
+			if batchLimit <= 0 {
+				batchLimit = 64
+			}
+			if batchLimit > asyncAdmissionBatchMax {
+				batchLimit = asyncAdmissionBatchMax
+			}
+			taskIDs, err := s.store.ListWaitingAsyncAdmissionTaskIDs(ctx, batchLimit)
 			if err != nil {
-				if !errors.Is(err, pgx.ErrNoRows) {
-					s.logger.Warn("load waiting async task failed", "taskID", taskID, "error", err)
+				s.logger.Warn("list waiting async admissions failed", "error", err)
+				break
+			}
+			if len(taskIDs) == 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
 				}
-				continue
+				if task.Status != "queued" {
+					continue
+				}
+				tasks = append(tasks, task)
 			}
-			if task.Status != "queued" {
-				continue
+			if len(tasks) == 0 {
+				break
 			}
-			tasks = append(tasks, task)
-		}
-		saturated, err := s.dispatchWaitingAsyncTasks(ctx, tasks)
-		if err != nil {
-			s.logger.Warn("dispatch waiting async admission batch failed", "tasks", len(tasks), "error", err)
-			continue
-		}
-		if saturated {
-			// Every asynchronous task shares the global worker-capacity FIFO
-			// scope. Once its current head cannot be admitted, later tasks
-			// cannot make progress until a lease is released.
-			s.observeTaskAdmission("dispatch_saturated")
+			saturated, err := s.dispatchWaitingAsyncTasks(ctx, tasks)
+			if err != nil {
+				s.logger.Warn("dispatch waiting async admission batch failed", "tasks", len(tasks), "error", err)
+				break
+			}
+			if saturated {
+				// Every asynchronous task shares the global worker-capacity FIFO
+				// scope. Once its current head cannot be admitted, later tasks
+				// cannot make progress until a lease is released.
+				s.observeTaskAdmission("dispatch_saturated")
+				break
+			}
+			// PostgreSQL notifications are wake-up hints and collapse while a
+			// batch is being admitted. Keep filling consecutive batches until
+			// the global execution scope is actually saturated so a large burst
+			// cannot strand most Worker slots behind the one-second fallback.
 		}
 	}
 }