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. } } }