feat(cluster): 增加跨节点任务执行与请求素材场景

新增 AI_GATEWAY_ASYNC_QUEUE_WORKER_ENABLED,默认保持启用;关闭时仅停止 River 异步任务执行器,HTTP、健康检查、结算与其他后台任务继续运行。

新增 0089 迁移并同步管理端默认场景,将 request_asset 加入现有上传渠道,确保 multipart 请求素材先转换为跨节点共享 URL。

验证:gofmt、go test ./... -count=1、go vet ./...、pnpm lint、pnpm test、pnpm build、迁移测试。
This commit is contained in:
2026-07-28 04:36:58 +08:00
parent 2c7f8c14eb
commit edf7e66941
9 changed files with 60 additions and 3 deletions
+2
View File
@@ -57,6 +57,7 @@ type Config struct {
GlobalHTTPProxySource string
LogLevel slog.Level
BillingEngineMode string
AsyncQueueWorkerEnabled bool
AsyncWorkerHardLimit int
AsyncWorkerRefreshIntervalSeconds int
}
@@ -112,6 +113,7 @@ func Load() Config {
GlobalHTTPProxySource: globalProxy.Source,
LogLevel: logLevel(env("LOG_LEVEL", "info")),
BillingEngineMode: strings.ToLower(env("BILLING_ENGINE_MODE", "observe")),
AsyncQueueWorkerEnabled: env("AI_GATEWAY_ASYNC_QUEUE_WORKER_ENABLED", "true") == "true",
AsyncWorkerHardLimit: envIntValidated("AI_GATEWAY_ASYNC_WORKER_HARD_LIMIT", 2048),
AsyncWorkerRefreshIntervalSeconds: envIntValidated("AI_GATEWAY_ASYNC_WORKER_REFRESH_INTERVAL_SECONDS", 5),
}
+12
View File
@@ -82,6 +82,18 @@ func TestValidateAsyncWorkerSettings(t *testing.T) {
}
}
func TestLoadAsyncQueueWorkerEnabled(t *testing.T) {
cfg := Load()
if !cfg.AsyncQueueWorkerEnabled {
t.Fatal("async queue worker must be enabled by default")
}
t.Setenv("AI_GATEWAY_ASYNC_QUEUE_WORKER_ENABLED", "false")
cfg = Load()
if cfg.AsyncQueueWorkerEnabled {
t.Fatal("async queue worker remained enabled after explicit disable")
}
}
func TestValidateTaskHistorySettings(t *testing.T) {
cfg := Load()
cfg.TaskRetentionDays = 30
+5 -1
View File
@@ -120,7 +120,11 @@ func NewServerWithContext(ctx context.Context, cfg config.Config, db *store.Stor
return server.identityRuntime.LegacyJWTEnabled()
}
server.auth.LocalAPIKeyVerifier = db.VerifyLocalAPIKey
server.runner.StartAsyncQueueWorker(ctx)
if cfg.AsyncQueueWorkerEnabled {
server.runner.StartAsyncQueueWorker(ctx)
} else if logger != nil {
logger.Info("asynchronous queue worker disabled for this process")
}
server.runner.StartBillingSettlementWorker(ctx)
server.runner.StartTaskHistoryWorkers(ctx)
server.startLocalTempAssetCleanup(ctx)
@@ -578,7 +578,7 @@ func normalizeFileStorageScene(scene string) string {
}
func defaultFileStorageScenes() []string {
return []string{FileStorageSceneUpload, FileStorageSceneImageResult}
return []string{FileStorageSceneUpload, FileStorageSceneImageResult, FileStorageSceneRequestAsset}
}
func defaultFileStorageRetryPolicyIfEmpty(policy map[string]any) map[string]any {
@@ -0,0 +1,17 @@
package store
import (
"reflect"
"testing"
)
func TestDefaultFileStorageScenesIncludeCrossNodeRequestAssets(t *testing.T) {
want := []string{
FileStorageSceneUpload,
FileStorageSceneImageResult,
FileStorageSceneRequestAsset,
}
if got := defaultFileStorageScenes(); !reflect.DeepEqual(got, want) {
t.Fatalf("default file storage scenes = %#v, want %#v", got, want)
}
}