BCAI/easyai-ai-gateway
3
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

perf(worker): 解耦准入调度与远端执行

Browse Source 
原因：线上 P24 同构验收确认两台香港 Worker 的全局容量为 48，但跨地域逐行准入事务每轮只能形成 8 个活跃租约，队列最老等待超过 15 分钟。\n\n影响：新增可配置的异步准入 dispatcher 角色；生产两地 API 负责准入和过期回收，Worker 仅执行 River job。当前主库同站点 API 可低延迟填满容量，主库切换后另一地 API 通过既有数据库锁安全接管；未配置环境变量时保持原 Worker 一体化行为。\n\n验证：Go 全量测试、go vet、gofmt、真实 PostgreSQL 1000 任务双进程回归、前端 lint/test/build、Kubernetes server-side dry-run、Compose、ShellCheck、cluster/manual release tests 全部通过。
This commit is contained in:
wangbo
2026-08-01 23:24:30 +08:00
parent 9c093974a1
commit a4046c2b81
10 changed files with 96 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/api/internal/config/config.go
+18
View File
@@ -80,6 +80,8 @@ type Config struct {
AsyncWorkerInstanceHardLimit int
AsyncWorkerRefreshIntervalSeconds int
AsyncAdmissionMicrobatchSize int
AsyncAdmissionDispatcherEnabled bool
AsyncAdmissionDispatcherConfigured bool
WorkerAutoscalingEnabled bool
WorkerReplicasNingbo int
WorkerReplicasHongkong int
@@ -188,6 +190,10 @@ func Load() Config {
AsyncWorkerInstanceHardLimit: envIntValidated("AI_GATEWAY_ASYNC_WORKER_INSTANCE_HARD_LIMIT", 32),
AsyncWorkerRefreshIntervalSeconds: envIntValidated("AI_GATEWAY_ASYNC_WORKER_REFRESH_INTERVAL_SECONDS", 5),
AsyncAdmissionMicrobatchSize: envIntValidated("AI_GATEWAY_ASYNC_ADMISSION_MICROBATCH_SIZE", 8),
AsyncAdmissionDispatcherEnabled: envValue("AI_GATEWAY_ASYNC_ADMISSION_DISPATCHER_ENABLED") == "true",
AsyncAdmissionDispatcherConfigured: envValue(
"AI_GATEWAY_ASYNC_ADMISSION_DISPATCHER_ENABLED",
) != "",
WorkerAutoscalingEnabled: env("AI_GATEWAY_WORKER_AUTOSCALING_ENABLED", "false") == "true",
WorkerReplicasNingbo: envOptionalIntValidated("AI_GATEWAY_WORKER_REPLICAS_NINGBO", 1),
WorkerReplicasHongkong: envOptionalIntValidated("AI_GATEWAY_WORKER_REPLICAS_HONGKONG", 1),
@@ -463,6 +469,18 @@ func (c Config) RunsAsyncExecutionWorker() bool {
}
}
// RunsAsyncAdmissionDispatcher keeps the legacy all-in-one/Worker behaviour
// unless a deployment explicitly separates admission coordination from task
// execution. Production enables this on both API sites so the instance nearest
// the current PostgreSQL primary wins the existing database scope locks, while
// remote Worker replicas only execute River jobs.
func (c Config) RunsAsyncAdmissionDispatcher() bool {
if c.AsyncAdmissionDispatcherConfigured {
return c.AsyncAdmissionDispatcherEnabled
}
return c.RunsAsyncExecutionWorker()
}
func (c Config) RunsBackgroundWorkers() bool {
switch c.EffectiveProcessRole() {
case "api", "capacity-controller":
apps/api/internal/config/config_test.go
+18
View File
@@ -159,12 +159,30 @@ func TestProcessRolePrecedenceAndCompatibility(t *testing.T) {
if cfg.RunsPublicHTTP() || !cfg.RunsAsyncExecutionWorker() || !cfg.RunsBackgroundWorkers() {
t.Fatalf("worker role capabilities are inconsistent: %+v", cfg)
}
if !cfg.RunsAsyncAdmissionDispatcher() {
t.Fatal("legacy Worker role no longer runs the admission dispatcher")
}
t.Setenv("AI_GATEWAY_PROCESS_ROLE", "api")
cfg = Load()
if !cfg.RunsPublicHTTP() || cfg.RunsAsyncExecutionWorker() || cfg.RunsBackgroundWorkers() {
t.Fatalf("api role capabilities are inconsistent: %+v", cfg)
}
if cfg.RunsAsyncAdmissionDispatcher() {
t.Fatal("API role unexpectedly runs the admission dispatcher without explicit configuration")
}
t.Setenv("AI_GATEWAY_ASYNC_ADMISSION_DISPATCHER_ENABLED", "true")
cfg = Load()
if !cfg.RunsAsyncAdmissionDispatcher() || cfg.RunsAsyncExecutionWorker() {
t.Fatalf("API admission-only role is inconsistent: %+v", cfg)
}
t.Setenv("AI_GATEWAY_ASYNC_ADMISSION_DISPATCHER_ENABLED", "false")
t.Setenv("AI_GATEWAY_PROCESS_ROLE", "worker")
cfg = Load()
if cfg.RunsAsyncAdmissionDispatcher() || !cfg.RunsAsyncExecutionWorker() {
t.Fatalf("Worker execution-only role is inconsistent: %+v", cfg)
}
t.Setenv("AI_GATEWAY_PROCESS_ROLE", "capacity-controller")
cfg = Load()
apps/api/internal/httpapi/async_worker_acceptance_integration_test.go
+28 -24
View File
@@ -136,33 +136,37 @@ func TestAsyncWorkerThousandConcurrentSchedulingAcceptance(t *testing.T) {
serverCtx, cancelServer := context.WithCancel(ctx)
defer cancelServer()
server := httptest.NewServer(NewServerWithStores(serverCtx, config.Config{
AppEnv: "test",
HTTPAddr: ":0",
DatabaseURL: databaseURL,
IdentityMode: "hybrid",
JWTSecret: "test-secret",
BillingEngineMode: "observe",
CORSAllowedOrigin: "*",
ProcessRole: "api",
AsyncQueueWorkerEnabled: true,
AsyncWorkerHardLimit: 1000,
AsyncWorkerInstanceHardLimit: 1000,
AsyncWorkerRefreshIntervalSeconds: 1,
AppEnv: "test",
HTTPAddr: ":0",
DatabaseURL: databaseURL,
IdentityMode: "hybrid",
JWTSecret: "test-secret",
BillingEngineMode: "observe",
CORSAllowedOrigin: "*",
ProcessRole: "api",
AsyncQueueWorkerEnabled: true,
AsyncAdmissionDispatcherEnabled: true,
AsyncAdmissionDispatcherConfigured: true,
AsyncWorkerHardLimit: 1000,
AsyncWorkerInstanceHardLimit: 1000,
AsyncWorkerRefreshIntervalSeconds: 1,
}, apiDB, apiCoordinationDB, apiDB, slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelError}))))
defer server.Close()
workerServer := httptest.NewServer(NewServerWithStores(serverCtx, config.Config{
AppEnv: "test",
HTTPAddr: ":0",
DatabaseURL: databaseURL,
IdentityMode: "hybrid",
JWTSecret: "test-secret",
BillingEngineMode: "observe",
CORSAllowedOrigin: "*",
ProcessRole: "worker",
AsyncQueueWorkerEnabled: true,
AsyncWorkerHardLimit: 1000,
AsyncWorkerInstanceHardLimit: 1000,
AsyncWorkerRefreshIntervalSeconds: 1,
AppEnv: "test",
HTTPAddr: ":0",
DatabaseURL: databaseURL,
IdentityMode: "hybrid",
JWTSecret: "test-secret",
BillingEngineMode: "observe",
CORSAllowedOrigin: "*",
ProcessRole: "worker",
AsyncQueueWorkerEnabled: true,
AsyncAdmissionDispatcherEnabled: false,
AsyncAdmissionDispatcherConfigured: true,
AsyncWorkerHardLimit: 1000,
AsyncWorkerInstanceHardLimit: 1000,
AsyncWorkerRefreshIntervalSeconds: 1,
}, workerDB, workerCoordinationDB, workerDB, slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelError}))))
defer workerServer.Close()
apps/api/internal/httpapi/server.go
+3
View File
@@ -138,6 +138,9 @@ func NewServerWithStores(
logger.Info("asynchronous queue worker disabled for this process")
}
}
if cfg.RunsAsyncAdmissionDispatcher() {
server.runner.StartAsyncAdmissionDispatcher(ctx)
}
if cfg.RunsBackgroundWorkers() {
server.runner.StartBillingSettlementWorker(ctx)
server.runner.StartTaskHistoryWorkers(ctx)
apps/api/internal/runner/queue_worker.go
+13 -2
View File
@@ -274,13 +274,24 @@ func (s *Service) startRiverQueue(ctx context.Context, workerEnabled bool) error
return err
}
go s.refreshAsyncWorkerCapacity(ctx)
go s.dispatchWaitingAsyncAdmissions(ctx)
go s.reapExpiredTaskAdmissions(ctx)
go s.recoverOrphanedAsyncRiverJobs(ctx)
go s.stopAsyncWorkersOnShutdown(ctx)
return nil
}
// StartAsyncAdmissionDispatcher starts the lightweight admission/reaping
// coordinator independently from River execution workers. This allows the
// dispatcher to run beside PostgreSQL while Worker pods scale on other nodes;
// the existing database task/scope locks make multiple API-site dispatchers
// safe during primary failover.
func (s *Service) StartAsyncAdmissionDispatcher(ctx context.Context) {
s.asyncAdmissionRunner.Do(func() {
go s.dispatchWaitingAsyncAdmissions(ctx)
go s.reapExpiredTaskAdmissions(ctx)
s.logger.Info("asynchronous admission dispatcher started")
})
}
func (s *Service) newRiverAsyncExecutionClient(capacity int) (*river.Client[pgx.Tx], error) {
workers := river.NewWorkers()
if err := river.AddWorkerSafely(workers, &asyncTaskWorker{service: s}); err != nil {
apps/api/internal/runner/service.go
+1
View File
@@ -45,6 +45,7 @@ type Service struct {
asyncAdmissionWake chan struct{}
admissionTaskWaiters map[string]*admissionTaskWaiter
admissionListener sync.Once
asyncAdmissionRunner sync.Once
asyncClientFactory func(int) (asyncExecutionClient, error)
mediaResultSlots chan struct{}
imageNormalizeSlots chan struct{}