From 012f61238bac1e61f3dfc791f0ee799af2304960 Mon Sep 17 00:00:00 2001 From: ddcat666 <2695092045@qq.com> Date: Thu, 23 Jul 2026 18:35:51 +0800 Subject: [PATCH] =?UTF-8?q?fix(platform):=20=E6=B8=A0=E9=81=93=E5=B9=B6?= =?UTF-8?q?=E5=8F=91=20-=20=E6=94=AF=E6=8C=81=E5=90=8E=E5=8F=B0=E7=83=AD?= =?UTF-8?q?=E9=85=8D=E7=BD=AE=E5=B9=B6=E7=AD=89=E5=BE=85=E7=A9=BA=E9=97=B2?= =?UTF-8?q?=E6=A7=BD=E4=BD=8D=20(#34)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: ddCat --- .env.example | 2 + CHANGELOG.md | 2 + README.md | 2 + backend/internal/handler/auth.go | 112 ++++++++++----- backend/internal/model/models.go | 26 ++-- backend/internal/service/admin.go | 82 ++++++----- .../internal/service/admin_channel_test.go | 40 +++++- backend/internal/service/analytics.go | 4 +- .../internal/service/concurrency_settings.go | 122 ++++++++++++++++ .../service/concurrency_settings_test.go | 63 ++++++++ backend/internal/service/coordination.go | 134 +++++++++++++++++- backend/internal/service/coordination_test.go | 54 +++++++ backend/internal/service/provider.go | 25 ++-- backend/internal/service/service.go | 22 ++- docker-compose.deploy.yml | 2 + docker-compose.local.yml | 2 + docker-compose.server.yml | 2 + docker-compose.yml | 2 + .../pages/admin/channels/channels-page.tsx | 14 +- .../pages/admin/components/admin-shell.tsx | 3 +- .../components/api-log-detail-drawer.tsx | 1 + .../settings/concurrency-settings-page.tsx | 73 ++++++++++ web/src/router.tsx | 4 +- web/src/services/api/auth.ts | 21 ++- web/src/stores/use-config-store.ts | 1 + 25 files changed, 703 insertions(+), 112 deletions(-) create mode 100644 backend/internal/service/concurrency_settings.go create mode 100644 backend/internal/service/concurrency_settings_test.go create mode 100644 backend/internal/service/coordination_test.go create mode 100644 web/src/pages/admin/settings/concurrency-settings-page.tsx diff --git a/.env.example b/.env.example index 2b7587387..138e7391c 100644 --- a/.env.example +++ b/.env.example @@ -4,6 +4,8 @@ CANVAS_ALLOW_PRIVATE_UPSTREAMS=false CANVAS_ALLOWED_PRIVATE_UPSTREAM_HOSTS= CANVAS_CORS_ORIGINS=http://localhost:3000 CANVAS_DATA_PATH=./data +CANVAS_WORKER_CONCURRENCY=3 +CANVAS_CHANNEL_CONCURRENCY=3 # 新服务器请直接运行 README 的一键部署命令；脚本会在 .env 中生成以下 PostgreSQL 配置。 # POSTGRES_DB=open_ai_canvas diff --git a/CHANGELOG.md b/CHANGELOG.md index f865ccdd7..a736d595b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,8 @@ ## Unreleased +- 系统后台支持热更新 Worker 与全局渠道并发，各系统渠道可跟随全局值或单独覆盖；渠道满载时请求等待空闲槽位，不再直接失败。 + ## v1.0.3 - 系统公告正文支持受控 HTML，用户可以看到蓝色链接并在新窗口打开，同时支持粗体、斜体、下划线、列表和换行等常见格式。 diff --git a/README.md b/README.md index fed4a1574..cda9fc1ef 100644 --- a/README.md +++ b/README.md @@ -120,6 +120,8 @@ bun run dev 打开 `http://localhost:3000`，注册首个管理员账号，再在系统设置中配置渠道和模型。 +任务 Worker 和全局渠道并发可在“系统配置 → 任务并发”中热更新；系统渠道可选择跟随全局值，或单独设置 `1-100` 的最大并发数。未保存后台配置时分别回退到 `CANVAS_WORKER_CONCURRENCY` 和 `CANVAS_CHANNEL_CONCURRENCY`，两者默认均为 `3`。渠道槽位暂满时任务会等待，不会直接标记失败。 + Docker 一体化运行： ```bash diff --git a/backend/internal/handler/auth.go b/backend/internal/handler/auth.go index cd63ca622..06df6e3af 100644 --- a/backend/internal/handler/auth.go +++ b/backend/internal/handler/auth.go @@ -439,6 +439,37 @@ func RegisterAdminRoutes(r *gin.RouterGroup, svc *service.Service) { } ok(c, gin.H{"setting": setting}) }) + r.GET("/admin/settings/concurrency", func(c *gin.Context) { + user, err := currentUser(c, svc) + if err != nil { + failService(c, err) + return + } + setting, err := svc.AdminRuntimeConcurrencySetting(user) + if err != nil { + failService(c, err) + return + } + ok(c, gin.H{"setting": setting}) + }) + r.PATCH("/admin/settings/concurrency", func(c *gin.Context) { + user, err := currentUser(c, svc) + if err != nil { + failService(c, err) + return + } + var req service.RuntimeConcurrencySettingRequest + if err := c.ShouldBindJSON(&req); err != nil { + fail(c, http.StatusBadRequest, err) + return + } + setting, err := svc.UpdateRuntimeConcurrencySetting(user, req) + if err != nil { + failService(c, err) + return + } + ok(c, gin.H{"setting": setting}) + }) r.GET("/admin/api-logs", func(c *gin.Context) { user, err := currentUser(c, svc) if err != nil { @@ -524,12 +555,35 @@ func proxySystemRequest(c *gin.Context, svc *service.Service, user *model.User, } modelName := proxyRequestModel(c.GetHeader("Content-Type"), body) billingOrderID := "" + query := c.Request.URL.Query() + for _, key := range []string{"key", "api_key", "access_token", "token"} { + query.Del(key) + } + target := strings.TrimRight(channel.BaseURL, "/") + path + if encodedQuery := query.Encode(); encodedQuery != "" { + target += "?" + encodedQuery + } + if _, err := service.ValidateOutboundURL(target); err != nil { + _ = svc.RefundBilling(billingOrderID, "系统渠道地址校验失败") + failService(c, err) + return + } + // 同步代理与后台任务必须共享渠道槽位，否则两条入口会共同超过供应商并发上限。 + releaseChannel, concurrencyLimit, err := svc.AcquireChannelSlot(c.Request.Context(), channel.ID, "", 36*time.Minute) + if err != nil { + log := apiCallLog(user, channel, billingOrderID, c.Request.Method, path, target, body, model.ApiCallStatusFailed, 0, time.Since(startedAt), err.Error(), concurrencyLimit) + log.ErrorCode, log.Error = service.ChannelSlotFailureDetails(err) + logSystemProxyCall(svc, log, nil) + fail(c, http.StatusServiceUnavailable, err) + return + } + defer releaseChannel() if c.Request.Method == http.MethodPost { capability := "text" switch channel.InterfaceType { case model.ChannelInterfaceOpenAIImage: capability = "image" - case model.ChannelInterfaceNewAPIVideo, model.ChannelInterfaceNewAPIChannel1, model.ChannelInterfaceNewAPIChannel2: + case model.ChannelInterfaceNewAPIVideo, model.ChannelInterfaceNewAPIChannel1, model.ChannelInterfaceNewAPIChannel2, model.ChannelInterfaceXAIVideo: capability = "video" } order, err := svc.ReserveProxyBilling(user.ID, channel.ID, strings.TrimPrefix(modelName, "models/"), capability, c.GetHeader("X-Canvas-Scene"), c.GetHeader("X-Idempotency-Key")) @@ -544,19 +598,6 @@ func proxySystemRequest(c *gin.Context, svc *service.Service, user *model.User, return } } - query := c.Request.URL.Query() - for _, key := range []string{"key", "api_key", "access_token", "token"} { - query.Del(key) - } - target := strings.TrimRight(channel.BaseURL, "/") + path - if encodedQuery := query.Encode(); encodedQuery != "" { - target += "?" + encodedQuery - } - if _, err := service.ValidateOutboundURL(target); err != nil { - _ = svc.RefundBilling(billingOrderID, "系统渠道地址校验失败") - failService(c, err) - return - } upstreamReq, err := http.NewRequestWithContext(c.Request.Context(), c.Request.Method, target, bytes.NewReader(body)) if err != nil { _ = svc.RefundBilling(billingOrderID, "系统渠道请求构造失败") @@ -583,7 +624,7 @@ func proxySystemRequest(c *gin.Context, svc *service.Service, user *model.User, status = model.ApiCallStatusFailed errorText = err.Error() _ = svc.MarkBillingUncertain(billingOrderID, "系统渠道连接中断，费用状态待核对") - logSystemProxyCall(svc, apiCallLog(user, channel, billingOrderID, c.Request.Method, path, target, body, status, statusCode, time.Since(startedAt), errorText), nil) + logSystemProxyCall(svc, apiCallLog(user, channel, billingOrderID, c.Request.Method, path, target, body, status, statusCode, time.Since(startedAt), errorText, concurrencyLimit), nil) fail(c, http.StatusBadGateway, errors.New("系统渠道连接失败")) return } @@ -597,7 +638,7 @@ func proxySystemRequest(c *gin.Context, svc *service.Service, user *model.User, status = model.ApiCallStatusFailed errorText = readErr.Error() _ = svc.MarkBillingUncertain(billingOrderID, "系统渠道响应读取失败，费用状态待核对") - logSystemProxyCall(svc, apiCallLog(user, channel, billingOrderID, c.Request.Method, path, target, body, status, statusCode, time.Since(startedAt), errorText), nil) + logSystemProxyCall(svc, apiCallLog(user, channel, billingOrderID, c.Request.Method, path, target, body, status, statusCode, time.Since(startedAt), errorText, concurrencyLimit), nil) fail(c, http.StatusBadGateway, errors.New("系统渠道响应读取失败")) return } @@ -615,7 +656,7 @@ func proxySystemRequest(c *gin.Context, svc *service.Service, user *model.User, } else { _ = svc.RefundBilling(billingOrderID, "上游明确返回失败") } - logSystemProxyCall(svc, apiCallLog(user, channel, billingOrderID, c.Request.Method, path, target, body, status, statusCode, time.Since(startedAt), errorText), responseBody) + logSystemProxyCall(svc, apiCallLog(user, channel, billingOrderID, c.Request.Method, path, target, body, status, statusCode, time.Since(startedAt), errorText, concurrencyLimit), responseBody) for _, key := range []string{"Content-Type", "Cache-Control", "Content-Disposition"} { if value := resp.Header.Get(key); value != "" { c.Header(key, value) @@ -625,12 +666,12 @@ func proxySystemRequest(c *gin.Context, svc *service.Service, user *model.User, c.Data(resp.StatusCode, resp.Header.Get("Content-Type"), responseBody) } -func apiCallLog(user *model.User, channel *model.ModelChannel, billingOrderID string, method string, path string, target string, body []byte, status model.ApiCallStatus, statusCode int, duration time.Duration, errorText string) model.ApiCallLog { +func apiCallLog(user *model.User, channel *model.ModelChannel, billingOrderID string, method string, path string, target string, body []byte, status model.ApiCallStatus, statusCode int, duration time.Duration, errorText string, concurrencyLimit int) model.ApiCallLog { capability := "text" switch channel.InterfaceType { case model.ChannelInterfaceOpenAIImage: capability = "image" - case model.ChannelInterfaceNewAPIVideo, model.ChannelInterfaceNewAPIChannel1, model.ChannelInterfaceNewAPIChannel2: + case model.ChannelInterfaceNewAPIVideo, model.ChannelInterfaceNewAPIChannel1, model.ChannelInterfaceNewAPIChannel2, model.ChannelInterfaceXAIVideo: capability = "video" } requestKind := "create" @@ -641,22 +682,23 @@ func apiCallLog(user *model.User, channel *model.ModelChannel, billingOrderID st } } return model.ApiCallLog{ - UserID: user.ID, - ChannelID: channel.ID, - BillingOrderID: billingOrderID, - Source: "system-channel", - Capability: capability, - RequestKind: requestKind, - Billable: method == http.MethodPost, - APIFormat: channel.APIFormat, - Method: method, - Path: path, - Model: readPayloadModel(body), - Status: status, - StatusCode: statusCode, - DurationMs: duration.Milliseconds(), - Error: errorText, - UpstreamURL: target, + UserID: user.ID, + ChannelID: channel.ID, + BillingOrderID: billingOrderID, + Source: "system-channel", + Capability: capability, + RequestKind: requestKind, + Billable: method == http.MethodPost, + APIFormat: channel.APIFormat, + Method: method, + Path: path, + Model: readPayloadModel(body), + Status: status, + StatusCode: statusCode, + DurationMs: duration.Milliseconds(), + Error: errorText, + ConcurrencyLimit: concurrencyLimit, + UpstreamURL: target, } } diff --git a/backend/internal/model/models.go b/backend/internal/model/models.go index 0a666792c..1936b7cde 100644 --- a/backend/internal/model/models.go +++ b/backend/internal/model/models.go @@ -147,18 +147,19 @@ type EmailVerificationCode struct { } type ModelChannel struct { - ID string `json:"id" gorm:"primaryKey;size:36"` - UserID string `json:"userId" gorm:"index;size:36"` - Scope ChannelScope `json:"scope" gorm:"index;size:24"` - Enabled bool `json:"enabled" gorm:"index"` - Name string `json:"name" gorm:"size:80"` - BaseURL string `json:"baseUrl"` - APIKey string `json:"-"` - APIFormat string `json:"apiFormat" gorm:"size:24"` - InterfaceType ChannelInterfaceType `json:"interfaceType" gorm:"size:32"` - ModelsJSON string `json:"modelsJson" gorm:"type:text"` - CreatedAt time.Time `json:"createdAt"` - UpdatedAt time.Time `json:"updatedAt"` + ID string `json:"id" gorm:"primaryKey;size:36"` + UserID string `json:"userId" gorm:"index;size:36"` + Scope ChannelScope `json:"scope" gorm:"index;size:24"` + Enabled bool `json:"enabled" gorm:"index"` + Name string `json:"name" gorm:"size:80"` + BaseURL string `json:"baseUrl"` + APIKey string `json:"-"` + APIFormat string `json:"apiFormat" gorm:"size:24"` + InterfaceType ChannelInterfaceType `json:"interfaceType" gorm:"size:32"` + ConcurrencyLimit int `json:"concurrencyLimit"` + ModelsJSON string `json:"modelsJson" gorm:"type:text"` + CreatedAt time.Time `json:"createdAt"` + UpdatedAt time.Time `json:"updatedAt"` } type ChannelModel struct { @@ -207,6 +208,7 @@ type ApiCallLog struct { Currency string `json:"currency" gorm:"size:12"` ErrorCode string `json:"errorCode,omitempty" gorm:"index;size:80"` Error string `json:"error"` + ConcurrencyLimit int `json:"concurrencyLimit"` UpstreamURL string `json:"upstreamUrl"` CreatedAt time.Time `json:"createdAt" gorm:"index;index:idx_api_logs_user_created,priority:2;index:idx_api_logs_channel_created,priority:2;index:idx_api_logs_model_created,priority:2;index:idx_api_logs_status_created,priority:2"` } diff --git a/backend/internal/service/admin.go b/backend/internal/service/admin.go index 9be7b0f23..3e444529b 100644 --- a/backend/internal/service/admin.go +++ b/backend/internal/service/admin.go @@ -75,29 +75,32 @@ type AdminReferenceData struct { } type ChannelRequest struct { - Name string `json:"name"` - BaseURL string `json:"baseUrl"` - APIKey string `json:"apiKey"` - InterfaceType string `json:"interfaceType"` - Models []string `json:"models"` - Enabled *bool `json:"enabled"` + Name string `json:"name"` + BaseURL string `json:"baseUrl"` + APIKey string `json:"apiKey"` + InterfaceType string `json:"interfaceType"` + ConcurrencyLimit *int `json:"concurrencyLimit"` + UseGlobalConcurrency *bool `json:"useGlobalConcurrency"` + Models []string `json:"models"` + Enabled *bool `json:"enabled"` } type PublicModelChannel struct { - ID string `json:"id"` - UserID string `json:"userId"` - Scope model.ChannelScope `json:"scope"` - Enabled bool `json:"enabled"` - Name string `json:"name"` - BaseURL string `json:"baseUrl"` - APIKey string `json:"apiKey"` - APIFormat string `json:"apiFormat"` - InterfaceType model.ChannelInterfaceType `json:"interfaceType"` - Models []string `json:"models"` - ModelCosts []PublicChannelModelPrice `json:"modelCosts"` - HasAPIKey bool `json:"hasApiKey"` - CreatedAt time.Time `json:"createdAt"` - UpdatedAt time.Time `json:"updatedAt"` + ID string `json:"id"` + UserID string `json:"userId"` + Scope model.ChannelScope `json:"scope"` + Enabled bool `json:"enabled"` + Name string `json:"name"` + BaseURL string `json:"baseUrl"` + APIKey string `json:"apiKey"` + APIFormat string `json:"apiFormat"` + InterfaceType model.ChannelInterfaceType `json:"interfaceType"` + ConcurrencyLimit int `json:"concurrencyLimit"` + Models []string `json:"models"` + ModelCosts []PublicChannelModelPrice `json:"modelCosts"` + HasAPIKey bool `json:"hasApiKey"` + CreatedAt time.Time `json:"createdAt"` + UpdatedAt time.Time `json:"updatedAt"` } type PublicChannelModelPrice struct { @@ -530,6 +533,16 @@ func channelFromRequest(req ChannelRequest, channel model.ModelChannel) (model.M // 系统渠道均由后端按已声明的接口类型分发，调用格式固定为 Bearer/OpenAI 兼容鉴权。 channel.APIFormat = "openai" channel.InterfaceType = interfaceType + if req.UseGlobalConcurrency != nil && *req.UseGlobalConcurrency { + channel.ConcurrencyLimit = 0 + } else if req.ConcurrencyLimit != nil { + if *req.ConcurrencyLimit < minChannelConcurrencyLimit || *req.ConcurrencyLimit > maxChannelConcurrencyLimit { + return channel, BadAuthRequest("最大并发数必须是 1-100 的整数") + } + channel.ConcurrencyLimit = *req.ConcurrencyLimit + } else if req.UseGlobalConcurrency != nil { + return channel, BadAuthRequest("请填写渠道最大并发数") + } channel.ModelsJSON = string(modelsJSON) if req.Enabled != nil { channel.Enabled = *req.Enabled @@ -611,20 +624,21 @@ func publicChannel(channel model.ModelChannel, admin bool, channelModels []model interfaceType = inferChannelInterfaceType(models) } return PublicModelChannel{ - ID: channel.ID, - UserID: channel.UserID, - Scope: channel.Scope, - Enabled: channel.Enabled, - Name: channel.Name, - BaseURL: baseURL, - APIKey: apiKey, - APIFormat: channel.APIFormat, - InterfaceType: interfaceType, - Models: models, - ModelCosts: modelCosts, - HasAPIKey: strings.TrimSpace(channel.APIKey) != "", - CreatedAt: channel.CreatedAt, - UpdatedAt: channel.UpdatedAt, + ID: channel.ID, + UserID: channel.UserID, + Scope: channel.Scope, + Enabled: channel.Enabled, + Name: channel.Name, + BaseURL: baseURL, + APIKey: apiKey, + APIFormat: channel.APIFormat, + InterfaceType: interfaceType, + ConcurrencyLimit: channel.ConcurrencyLimit, + Models: models, + ModelCosts: modelCosts, + HasAPIKey: strings.TrimSpace(channel.APIKey) != "", + CreatedAt: channel.CreatedAt, + UpdatedAt: channel.UpdatedAt, } } diff --git a/backend/internal/service/admin_channel_test.go b/backend/internal/service/admin_channel_test.go index 4fd6623ad..1d1bd5a03 100644 --- a/backend/internal/service/admin_channel_test.go +++ b/backend/internal/service/admin_channel_test.go @@ -14,11 +14,12 @@ func TestChannelFromRequestStoresInterfaceType(t *testing.T) { defer server.Close() channel, err := channelFromRequest(ChannelRequest{ - Name: "NewAPI 渠道 1", - BaseURL: server.URL + "/v1", - APIKey: "secret", - InterfaceType: "newapi-channel-1", - Models: []string{"seedance-2.0"}, + Name: "NewAPI 渠道 1", + BaseURL: server.URL + "/v1", + APIKey: "secret", + InterfaceType: "newapi-channel-1", + ConcurrencyLimit: intPtr(6), + Models: []string{"seedance-2.0"}, }, model.ModelChannel{}) if err != nil { t.Fatalf("channelFromRequest() error = %v", err) @@ -29,6 +30,9 @@ func TestChannelFromRequestStoresInterfaceType(t *testing.T) { if channel.APIFormat != "openai" { t.Fatalf("APIFormat = %q, want openai", channel.APIFormat) } + if channel.ConcurrencyLimit != 6 { + t.Fatalf("ConcurrencyLimit = %d, want 6", channel.ConcurrencyLimit) + } } func TestChannelFromRequestStoresXAIVideoInterfaceType(t *testing.T) { @@ -70,3 +74,29 @@ func TestChannelFromRequestRejectsUnknownInterfaceType(t *testing.T) { t.Fatal("channelFromRequest() error = nil") } } + +func TestChannelFromRequestRejectsInvalidConcurrencyLimit(t *testing.T) { + for _, limit := range []int{0, 101} { + _, err := channelFromRequest(ChannelRequest{Name: "Bad", BaseURL: "https://example.com/v1", InterfaceType: "newapi", ConcurrencyLimit: &limit}, model.ModelChannel{}) + if err == nil { + t.Fatalf("channelFromRequest() concurrencyLimit = %d, error = nil", limit) + } + } +} + +func TestRuntimeConcurrencyUsesEnvironmentFallback(t *testing.T) { + t.Setenv("CANVAS_CHANNEL_CONCURRENCY", "7") + t.Setenv("CANVAS_WORKER_CONCURRENCY", "9") + setting := runtimeConcurrencyFromEnvironment() + if setting.ChannelConcurrency != 7 || setting.WorkerConcurrency != 9 { + t.Fatalf("runtimeConcurrencyFromEnvironment() = %#v", setting) + } + + useGlobal := true + channel, err := channelFromRequest(ChannelRequest{Name: "Global", BaseURL: "https://example.com/v1", InterfaceType: "newapi", UseGlobalConcurrency: &useGlobal}, model.ModelChannel{ConcurrencyLimit: 4}) + if err != nil || channel.ConcurrencyLimit != 0 { + t.Fatalf("global concurrency channel = %#v, error = %v", channel, err) + } +} + +func intPtr(value int) *int { return &value } diff --git a/backend/internal/service/analytics.go b/backend/internal/service/analytics.go index ff0217aaa..d3b51884d 100644 --- a/backend/internal/service/analytics.go +++ b/backend/internal/service/analytics.go @@ -245,9 +245,9 @@ func (s *Service) AdminAPICallLogsCSV(actor *model.User, query APICallLogQuery) var buffer bytes.Buffer buffer.WriteString("\xEF\xBB\xBF") writer := csv.NewWriter(&buffer) - _ = writer.Write([]string{"时间", "用户ID", "渠道ID", "任务ID", "计费单ID", "能力", "请求阶段", "模型", "状态", "HTTP状态", "耗时毫秒", "供应商任务ID", "错误码", "错误"}) + _ = writer.Write([]string{"时间", "用户ID", "渠道ID", "任务ID", "计费单ID", "能力", "请求阶段", "模型", "状态", "HTTP状态", "耗时毫秒", "渠道并发上限", "供应商任务ID", "错误码", "错误"}) for _, log := range logs { - _ = writer.Write([]string{log.CreatedAt.UTC().Format(time.RFC3339), log.UserID, log.ChannelID, log.TaskID, log.BillingOrderID, log.Capability, log.RequestKind, log.Model, string(log.Status), strconv.Itoa(log.StatusCode), strconv.FormatInt(log.DurationMs, 10), log.ProviderRequestID, log.ErrorCode, log.Error}) + _ = writer.Write([]string{log.CreatedAt.UTC().Format(time.RFC3339), log.UserID, log.ChannelID, log.TaskID, log.BillingOrderID, log.Capability, log.RequestKind, log.Model, string(log.Status), strconv.Itoa(log.StatusCode), strconv.FormatInt(log.DurationMs, 10), strconv.Itoa(log.ConcurrencyLimit), log.ProviderRequestID, log.ErrorCode, log.Error}) } writer.Flush() if err := writer.Error(); err != nil { diff --git a/backend/internal/service/concurrency_settings.go b/backend/internal/service/concurrency_settings.go new file mode 100644 index 000000000..f805a7e57 --- /dev/null +++ b/backend/internal/service/concurrency_settings.go @@ -0,0 +1,122 @@ +package service + +import ( + "encoding/json" + "errors" + "strings" + "time" + + "infinite-canvas/backend/internal/model" + + "gorm.io/gorm" +) + +const runtimeConcurrencySettingKey = "runtime_concurrency" + +type RuntimeConcurrencySettingRequest struct { + WorkerConcurrency int `json:"workerConcurrency"` + ChannelConcurrency int `json:"channelConcurrency"` +} + +type PublicRuntimeConcurrencySetting struct { + WorkerConcurrency int `json:"workerConcurrency"` + ChannelConcurrency int `json:"channelConcurrency"` + UpdatedBy string `json:"updatedBy"` + CreatedAt time.Time `json:"createdAt"` + UpdatedAt time.Time `json:"updatedAt"` +} + +type runtimeConcurrencySettingValue struct { + WorkerConcurrency int `json:"workerConcurrency"` + ChannelConcurrency int `json:"channelConcurrency"` +} + +func (s *Service) AdminRuntimeConcurrencySetting(actor *model.User) (*PublicRuntimeConcurrencySetting, error) { + if err := s.RequireAdmin(actor); err != nil { + return nil, err + } + setting, value, err := s.readRuntimeConcurrencySetting() + if err != nil { + return nil, err + } + return publicRuntimeConcurrencySetting(setting, value), nil +} + +func (s *Service) UpdateRuntimeConcurrencySetting(actor *model.User, req RuntimeConcurrencySettingRequest) (*PublicRuntimeConcurrencySetting, error) { + if err := s.RequireAdmin(actor); err != nil { + return nil, err + } + value := runtimeConcurrencySettingValue{WorkerConcurrency: req.WorkerConcurrency, ChannelConcurrency: req.ChannelConcurrency} + if err := validateRuntimeConcurrencySetting(value); err != nil { + return nil, err + } + encoded, err := json.Marshal(value) + if err != nil { + return nil, err + } + current, _, err := s.readRuntimeConcurrencySetting() + if err != nil { + return nil, err + } + setting := model.SystemSetting{Key: runtimeConcurrencySettingKey, ValueJSON: string(encoded), UpdatedBy: actor.ID} + if current != nil { + setting.CreatedAt = current.CreatedAt + } + if err := s.repo.SaveSystemSetting(&setting); err != nil { + return nil, err + } + if err := s.appendAdminAudit(actor, "runtime_concurrency.update", "system_setting", runtimeConcurrencySettingKey, "更新任务并发配置", value); err != nil { + return nil, err + } + return publicRuntimeConcurrencySetting(&setting, value), nil +} + +func (s *Service) runtimeConcurrencySetting() (runtimeConcurrencySettingValue, error) { + _, value, err := s.readRuntimeConcurrencySetting() + return value, err +} + +func (s *Service) readRuntimeConcurrencySetting() (*model.SystemSetting, runtimeConcurrencySettingValue, error) { + setting, err := s.repo.SystemSetting(runtimeConcurrencySettingKey) + if errors.Is(err, gorm.ErrRecordNotFound) { + return nil, runtimeConcurrencyFromEnvironment(), nil + } + if err != nil { + return nil, runtimeConcurrencySettingValue{}, err + } + value := runtimeConcurrencySettingValue{} + if strings.TrimSpace(setting.ValueJSON) == "" || json.Unmarshal([]byte(setting.ValueJSON), &value) != nil { + return nil, runtimeConcurrencySettingValue{}, errors.New("任务并发配置格式无效") + } + if err := validateRuntimeConcurrencySetting(value); err != nil { + return nil, runtimeConcurrencySettingValue{}, err + } + return setting, value, nil +} + +func runtimeConcurrencyFromEnvironment() runtimeConcurrencySettingValue { + return runtimeConcurrencySettingValue{ + WorkerConcurrency: effectiveChannelConcurrencyLimit(envInt("CANVAS_WORKER_CONCURRENCY", taskWorkerConcurrency)), + ChannelConcurrency: defaultChannelConcurrencyLimit(), + } +} + +func validateRuntimeConcurrencySetting(value runtimeConcurrencySettingValue) error { + if value.WorkerConcurrency < minChannelConcurrencyLimit || value.WorkerConcurrency > maxChannelConcurrencyLimit { + return BadAuthRequest("Worker 并发数必须是 1-100 的整数") + } + if value.ChannelConcurrency < minChannelConcurrencyLimit || value.ChannelConcurrency > maxChannelConcurrencyLimit { + return BadAuthRequest("全局渠道并发数必须是 1-100 的整数") + } + return nil +} + +func publicRuntimeConcurrencySetting(setting *model.SystemSetting, value runtimeConcurrencySettingValue) *PublicRuntimeConcurrencySetting { + result := &PublicRuntimeConcurrencySetting{WorkerConcurrency: value.WorkerConcurrency, ChannelConcurrency: value.ChannelConcurrency} + if setting != nil { + result.UpdatedBy = setting.UpdatedBy + result.CreatedAt = setting.CreatedAt + result.UpdatedAt = setting.UpdatedAt + } + return result +} diff --git a/backend/internal/service/concurrency_settings_test.go b/backend/internal/service/concurrency_settings_test.go new file mode 100644 index 000000000..0f70b61eb --- /dev/null +++ b/backend/internal/service/concurrency_settings_test.go @@ -0,0 +1,63 @@ +package service + +import ( + "context" + "testing" + "time" + + "infinite-canvas/backend/internal/model" + "infinite-canvas/backend/internal/repository" + + "gorm.io/driver/sqlite" + "gorm.io/gorm" +) + +func TestRuntimeConcurrencySettingPersistsAndChannelOverrideWins(t *testing.T) { + t.Setenv("REDIS_URL", "") + t.Setenv("CANVAS_WORKER_CONCURRENCY", "3") + t.Setenv("CANVAS_CHANNEL_CONCURRENCY", "3") + db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{}) + if err != nil { + t.Fatal(err) + } + if sqlDB, err := db.DB(); err == nil { + sqlDB.SetMaxOpenConns(1) + } + if err := db.AutoMigrate(&model.SystemSetting{}, &model.AdminAuditEvent{}, &model.ModelChannel{}); err != nil { + t.Fatal(err) + } + svc := New(repository.New(db), t.TempDir()) + actor := &model.User{ID: "admin-1", Role: model.UserRoleAdmin} + setting, err := svc.UpdateRuntimeConcurrencySetting(actor, RuntimeConcurrencySettingRequest{WorkerConcurrency: 8, ChannelConcurrency: 6}) + if err != nil { + t.Fatal(err) + } + if setting.WorkerConcurrency != 8 || setting.ChannelConcurrency != 6 { + t.Fatalf("setting = %#v", setting) + } + + channel := model.ModelChannel{ID: "channel-1", Scope: model.ChannelScopeSystem, Enabled: true, ConcurrencyLimit: 0} + if err := db.Create(&channel).Error; err != nil { + t.Fatal(err) + } + release, limit, err := svc.AcquireChannelSlot(context.Background(), channel.ID, "", time.Minute) + if err != nil { + t.Fatal(err) + } + release() + if limit != 6 { + t.Fatalf("global channel limit = %d, want 6", limit) + } + + if err := db.Model(&channel).Update("concurrency_limit", 9).Error; err != nil { + t.Fatal(err) + } + release, limit, err = svc.AcquireChannelSlot(context.Background(), channel.ID, "", time.Minute) + if err != nil { + t.Fatal(err) + } + release() + if limit != 9 { + t.Fatalf("overridden channel limit = %d, want 9", limit) + } +} diff --git a/backend/internal/service/coordination.go b/backend/internal/service/coordination.go index 910943f9a..9e625275b 100644 --- a/backend/internal/service/coordination.go +++ b/backend/internal/service/coordination.go @@ -18,11 +18,50 @@ type localRateEntry struct { count int } +const ( + minChannelConcurrencyLimit = 1 + maxChannelConcurrencyLimit = 100 + defaultChannelConcurrencyValue = 3 +) + +type channelSlotError struct { + scope string + limit int + err error +} + +func (e channelSlotError) Error() string { + if errors.Is(e.err, context.DeadlineExceeded) { + return fmt.Sprintf("等待渠道并发槽位超时（渠道 %s，并发上限 %d）", e.scope, e.limit) + } + if errors.Is(e.err, context.Canceled) { + return fmt.Sprintf("等待渠道并发槽位已取消（渠道 %s，并发上限 %d）", e.scope, e.limit) + } + return fmt.Sprintf("获取渠道并发配额失败（渠道 %s，并发上限 %d）：%v", e.scope, e.limit, e.err) +} + +func (e channelSlotError) Unwrap() error { return e.err } + +func ChannelSlotFailureDetails(err error) (string, string) { + var slotErr channelSlotError + if !errors.As(err, &slotErr) { + return "", "" + } + if errors.Is(slotErr, context.DeadlineExceeded) { + return "channel_concurrency_wait_timeout", slotErr.Error() + } + if errors.Is(slotErr, context.Canceled) { + return "channel_concurrency_wait_cancelled", slotErr.Error() + } + return "channel_concurrency_unavailable", slotErr.Error() +} + type runtimeCoordinator struct { redis *redis.Client instanceID string localMu sync.Mutex localRate map[string]localRateEntry + localSlots map[string]map[string]time.Time } var fixedWindowScript = redis.NewScript(` @@ -40,7 +79,7 @@ return 1 `) func newRuntimeCoordinator(dialect string) (*runtimeCoordinator, error) { - coordinator := &runtimeCoordinator{instanceID: newID(), localRate: map[string]localRateEntry{}} + coordinator := &runtimeCoordinator{instanceID: newID(), localRate: map[string]localRateEntry{}, localSlots: map[string]map[string]time.Time{}} redisURL := strings.TrimSpace(os.Getenv("REDIS_URL")) if redisURL == "" { if dialect == "postgres" { @@ -84,7 +123,30 @@ func (c *runtimeCoordinator) allow(ctx context.Context, key string, limit int, w func (c *runtimeCoordinator) acquire(ctx context.Context, scope string, limit int, ttl time.Duration) (func(), bool, error) { if c.redis == nil { - return func() {}, true, nil + c.localMu.Lock() + now := time.Now() + slots := c.localSlots[scope] + if slots == nil { + slots = map[string]time.Time{} + c.localSlots[scope] = slots + } + for token, expiresAt := range slots { + if !expiresAt.After(now) { + delete(slots, token) + } + } + if len(slots) >= limit { + c.localMu.Unlock() + return nil, false, nil + } + token := c.instanceID + ":" + newID() + slots[token] = now.Add(ttl) + c.localMu.Unlock() + return func() { + c.localMu.Lock() + delete(c.localSlots[scope], token) + c.localMu.Unlock() + }, true, nil } // 有过期分数的有序集合避免实例崩溃后永久占槽，业务数据库仍保存任务与账本真相。 key := "canvas:slots:" + scope @@ -97,6 +159,28 @@ func (c *runtimeCoordinator) acquire(ctx context.Context, scope string, limit in return func() { _ = c.redis.ZRem(context.Background(), key, token).Err() }, true, nil } +func (c *runtimeCoordinator) acquireWithWait(ctx context.Context, scope string, limit int, ttl time.Duration) (func(), error) { + ticker := time.NewTicker(200 * time.Millisecond) + defer ticker.Stop() + for { + if err := ctx.Err(); err != nil { + return nil, err + } + release, acquired, err := c.acquire(ctx, scope, limit, ttl) + if err != nil { + return nil, err + } + if acquired { + return release, nil + } + select { + case <-ctx.Done(): + return nil, ctx.Err() + case <-ticker.C: + } + } +} + func (c *runtimeCoordinator) circuitOpen(ctx context.Context, channelID string) (bool, error) { if c.redis == nil || strings.TrimSpace(channelID) == "" { return false, nil @@ -133,6 +217,52 @@ func envInt(key string, fallback int) int { return value } +func defaultChannelConcurrencyLimit() int { + return effectiveChannelConcurrencyLimit(envInt("CANVAS_CHANNEL_CONCURRENCY", defaultChannelConcurrencyValue)) +} + +func effectiveChannelConcurrencyLimit(configured int) int { + if configured < minChannelConcurrencyLimit || configured > maxChannelConcurrencyLimit { + return defaultChannelConcurrencyValue + } + return configured +} + +func (s *Service) AcquireChannelSlot(ctx context.Context, channelID string, fallbackScope string, ttl time.Duration) (func(), int, error) { + setting, err := s.runtimeConcurrencySetting() + limit := defaultChannelConcurrencyLimit() + if err != nil { + return nil, limit, channelSlotError{scope: firstNonEmpty(strings.TrimSpace(channelID), strings.TrimSpace(fallbackScope), "unknown"), limit: limit, err: fmt.Errorf("读取全局并发配置失败：%w", err)} + } + limit = setting.ChannelConcurrency + scope := strings.TrimSpace(channelID) + if scope != "" { + channel, err := s.repo.SystemChannel(scope) + if err != nil { + return nil, limit, channelSlotError{scope: scope, limit: limit, err: fmt.Errorf("读取渠道并发配置失败：%w", err)} + } + if channel.ConcurrencyLimit > 0 { + if channel.ConcurrencyLimit < minChannelConcurrencyLimit || channel.ConcurrencyLimit > maxChannelConcurrencyLimit { + return nil, limit, channelSlotError{scope: scope, limit: limit, err: errors.New("渠道并发配置超出 1-100 范围")} + } + limit = channel.ConcurrencyLimit + } + } else { + scope = strings.TrimSpace(fallbackScope) + } + if scope == "" { + return nil, limit, channelSlotError{scope: "unknown", limit: limit, err: errors.New("渠道并发范围为空")} + } + if s.coordinator == nil { + return nil, limit, channelSlotError{scope: scope, limit: limit, err: errors.New("运行时协调器未初始化")} + } + release, err := s.coordinator.acquireWithWait(ctx, "channel:"+scope, limit, ttl) + if err != nil { + return nil, limit, channelSlotError{scope: scope, limit: limit, err: err} + } + return release, limit, nil +} + func (s *Service) ValidateRuntime() error { return s.runtimeErr } diff --git a/backend/internal/service/coordination_test.go b/backend/internal/service/coordination_test.go new file mode 100644 index 000000000..81e0d7f3d --- /dev/null +++ b/backend/internal/service/coordination_test.go @@ -0,0 +1,54 @@ +package service + +import ( + "context" + "testing" + "time" +) + +func TestRuntimeCoordinatorWaitsUntilChannelSlotIsReleased(t *testing.T) { + coordinator := &runtimeCoordinator{instanceID: "test", localRate: map[string]localRateEntry{}, localSlots: map[string]map[string]time.Time{}} + releaseFirst, acquired, err := coordinator.acquire(context.Background(), "channel:one", 1, time.Minute) + if err != nil || !acquired { + t.Fatalf("first acquire = (%v, %v), want acquired", acquired, err) + } + + result := make(chan error, 1) + go func() { + releaseSecond, waitErr := coordinator.acquireWithWait(context.Background(), "channel:one", 1, time.Minute) + if waitErr == nil { + releaseSecond() + } + result <- waitErr + }() + + select { + case err := <-result: + t.Fatalf("second acquire returned before release: %v", err) + case <-time.After(50 * time.Millisecond): + } + releaseFirst() + select { + case err := <-result: + if err != nil { + t.Fatalf("second acquire after release: %v", err) + } + case <-time.After(time.Second): + t.Fatal("second acquire did not resume after release") + } +} + +func TestRuntimeCoordinatorStopsWaitingWhenContextIsCancelled(t *testing.T) { + coordinator := &runtimeCoordinator{instanceID: "test", localRate: map[string]localRateEntry{}, localSlots: map[string]map[string]time.Time{}} + release, acquired, err := coordinator.acquire(context.Background(), "channel:one", 1, time.Minute) + if err != nil || !acquired { + t.Fatalf("first acquire = (%v, %v), want acquired", acquired, err) + } + defer release() + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + if _, err := coordinator.acquireWithWait(ctx, "channel:one", 1, time.Minute); err == nil { + t.Fatal("acquireWithWait() error = nil after cancellation") + } +} diff --git a/backend/internal/service/provider.go b/backend/internal/service/provider.go index 596af7742..39de9f75e 100644 --- a/backend/internal/service/provider.go +++ b/backend/internal/service/provider.go @@ -97,6 +97,7 @@ type providerAnalyticsContext struct { VideoSeconds int RequestKind string ProviderRequestID string + ConcurrencyLimit int } func withProviderAnalytics(ctx context.Context, service *Service, task model.Task) context.Context { @@ -1274,17 +1275,17 @@ func doBinary(req *http.Request) ([]byte, string, error) { if open { return nil, "", errors.New("当前渠道连续失败，已暂时熔断，请稍后重试") } - var acquired bool slotID := channelID if slotID == "" { slotID = "custom:" + strings.ToLower(req.URL.Host) } - release, acquired, err = coordinator.acquire(req.Context(), "channel:"+slotID, envInt("CANVAS_CHANNEL_CONCURRENCY", 2), providerHTTPTimeout+time.Minute) + var concurrencyLimit int + release, concurrencyLimit, err = metadata.Service.AcquireChannelSlot(req.Context(), channelID, slotID, providerHTTPTimeout+time.Minute) + metadata.ConcurrencyLimit = concurrencyLimit + req = req.WithContext(context.WithValue(req.Context(), providerAnalyticsKey{}, metadata)) if err != nil { - return nil, "", fmt.Errorf("获取渠道并发配额失败：%w", err) - } - if !acquired { - return nil, "", errors.New("当前渠道并发已满，请稍后重试") + recordProviderRequest(req, startedAt, 0, nil, err) + return nil, "", err } defer release() } @@ -1356,7 +1357,13 @@ func recordProviderRequest(req *http.Request, startedAt time.Time, statusCode in RequestKind: requestKind, Billable: req.Method == http.MethodPost, APIFormat: "openai", Method: req.Method, Path: req.URL.Path, Model: metadata.Model, Status: status, StatusCode: statusCode, DurationMs: time.Since(startedAt).Milliseconds(), - Error: errorText, UpstreamURL: req.URL.Scheme + "://" + req.URL.Host + req.URL.Path, + Error: errorText, ConcurrencyLimit: metadata.ConcurrencyLimit, UpstreamURL: req.URL.Scheme + "://" + req.URL.Host + req.URL.Path, + } + channelSlotFailure := false + if code, message := ChannelSlotFailureDetails(requestErr); code != "" { + channelSlotFailure = true + log.ErrorCode = code + log.Error = message } if requestKind == "create" && metadata.Capability == "video" { log.VideoSeconds = metadata.VideoSeconds @@ -1370,7 +1377,9 @@ func recordProviderRequest(req *http.Request, startedAt time.Time, statusCode in } metadata.Service.EnrichAPICallLog(&log, responseBody) if err := metadata.Service.LogAPICall(log); err != nil { - _ = metadata.Service.MarkBillingUncertain(metadata.BillingOrderID, "上游调用日志写入失败，费用状态待核对") + if !channelSlotFailure { + _ = metadata.Service.MarkBillingUncertain(metadata.BillingOrderID, "上游调用日志写入失败，费用状态待核对") + } } } diff --git a/backend/internal/service/service.go b/backend/internal/service/service.go index 5be725e0e..f38d5f577 100644 --- a/backend/internal/service/service.go +++ b/backend/internal/service/service.go @@ -160,10 +160,14 @@ func New(repo *repository.Repository, dataDir string) *Service { func (s *Service) StartWorker() { go func() { - workerConcurrency := envInt("CANVAS_WORKER_CONCURRENCY", taskWorkerConcurrency) - slots := make(chan struct{}, workerConcurrency) + slots := make(chan struct{}, maxChannelConcurrencyLimit) dispatch := func() { - for len(slots) < cap(slots) { + setting, err := s.runtimeConcurrencySetting() + if err != nil { + return + } + workerConcurrency := setting.WorkerConcurrency + for len(slots) < workerConcurrency { releaseGlobal, acquired, err := s.coordinator.acquire(context.Background(), "workers", workerConcurrency, 45*time.Minute) if err != nil || !acquired { return @@ -697,6 +701,10 @@ func (s *Service) processClaimedTask(task *model.Task) error { result, err = s.persistGeneratedMediaResult(task.UserID, result) } if err != nil { + channelSlotFailedBeforeRequest := false + if code, _ := ChannelSlotFailureDetails(err); code != "" { + channelSlotFailedBeforeRequest = true + } select { case leaseErr := <-leaseLost: _ = s.log(task.UserID, task.ID, "warn", "任务租约失效，等待其他 worker 恢复", leaseErr.Error()) @@ -709,7 +717,11 @@ func (s *Service) processClaimedTask(task *model.Task) error { task.Error = "任务已取消" task.CompletedAt = ptr(time.Now()) _ = s.repo.Save(task) - _ = s.MarkBillingUncertain(task.BillingOrderID, "任务取消时上游费用状态不明确") + if channelSlotFailedBeforeRequest { + _ = s.RefundBilling(task.BillingOrderID, "等待渠道槽位期间取消，上游请求未发出") + } else { + _ = s.MarkBillingUncertain(task.BillingOrderID, "任务取消时上游费用状态不明确") + } _ = s.markSessionFailed(*task, "会话任务已取消。") _ = s.log(task.UserID, task.ID, "warn", "任务已取消", "") return nil @@ -722,7 +734,7 @@ func (s *Service) processClaimedTask(task *model.Task) error { task.Error = taskFailureMessage(err) task.CompletedAt = ptr(time.Now()) _ = s.repo.Save(task) - if providerSucceeded || s.BillingFailureRequiresReview(task.BillingOrderID, task.ID, err) { + if providerSucceeded || (!channelSlotFailedBeforeRequest && s.BillingFailureRequiresReview(task.BillingOrderID, task.ID, err)) { _ = s.MarkBillingUncertain(task.BillingOrderID, task.Error) } else { _ = s.RefundBilling(task.BillingOrderID, task.Error) diff --git a/docker-compose.deploy.yml b/docker-compose.deploy.yml index 1406d8042..2a4bdfef3 100644 --- a/docker-compose.deploy.yml +++ b/docker-compose.deploy.yml @@ -34,6 +34,8 @@ services: CANVAS_DATABASE_DRIVER: postgres DATABASE_URL: ${DATABASE_URL:?请先运行一键部署脚本生成数据库连接} REDIS_URL: redis://redis:6379/0 + CANVAS_WORKER_CONCURRENCY: ${CANVAS_WORKER_CONCURRENCY:-3} + CANVAS_CHANNEL_CONCURRENCY: ${CANVAS_CHANNEL_CONCURRENCY:-3} CANVAS_REGISTRATION_ENABLED: ${CANVAS_REGISTRATION_ENABLED:-false} CANVAS_ALLOW_PRIVATE_UPSTREAMS: ${CANVAS_ALLOW_PRIVATE_UPSTREAMS:-false} CANVAS_ALLOWED_PRIVATE_UPSTREAM_HOSTS: ${CANVAS_ALLOWED_PRIVATE_UPSTREAM_HOSTS:-} diff --git a/docker-compose.local.yml b/docker-compose.local.yml index e72b30a84..3adf639aa 100644 --- a/docker-compose.local.yml +++ b/docker-compose.local.yml @@ -12,6 +12,8 @@ services: CANVAS_DATABASE_DRIVER: ${CANVAS_DATABASE_DRIVER:-sqlite} DATABASE_URL: ${DATABASE_URL:-} REDIS_URL: ${REDIS_URL:-} + CANVAS_WORKER_CONCURRENCY: ${CANVAS_WORKER_CONCURRENCY:-3} + CANVAS_CHANNEL_CONCURRENCY: ${CANVAS_CHANNEL_CONCURRENCY:-3} CANVAS_REGISTRATION_ENABLED: ${CANVAS_REGISTRATION_ENABLED:-false} CANVAS_ALLOW_PRIVATE_UPSTREAMS: ${CANVAS_ALLOW_PRIVATE_UPSTREAMS:-false} CANVAS_ALLOWED_PRIVATE_UPSTREAM_HOSTS: ${CANVAS_ALLOWED_PRIVATE_UPSTREAM_HOSTS:-} diff --git a/docker-compose.server.yml b/docker-compose.server.yml index 4ed4e2a5f..5c276229c 100644 --- a/docker-compose.server.yml +++ b/docker-compose.server.yml @@ -42,6 +42,8 @@ services: CANVAS_DATABASE_DRIVER: ${CANVAS_DATABASE_DRIVER:-postgres} DATABASE_URL: ${DATABASE_URL:-} REDIS_URL: ${REDIS_URL:-redis://redis:6379/0} + CANVAS_WORKER_CONCURRENCY: ${CANVAS_WORKER_CONCURRENCY:-3} + CANVAS_CHANNEL_CONCURRENCY: ${CANVAS_CHANNEL_CONCURRENCY:-3} CANVAS_REGISTRATION_ENABLED: ${CANVAS_REGISTRATION_ENABLED:-false} CANVAS_ALLOW_PRIVATE_UPSTREAMS: ${CANVAS_ALLOW_PRIVATE_UPSTREAMS:-false} CANVAS_ALLOWED_PRIVATE_UPSTREAM_HOSTS: ${CANVAS_ALLOWED_PRIVATE_UPSTREAM_HOSTS:-} diff --git a/docker-compose.yml b/docker-compose.yml index 9ebd1af5e..4777a0837 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -12,6 +12,8 @@ services: CANVAS_DATABASE_DRIVER: ${CANVAS_DATABASE_DRIVER:-sqlite} DATABASE_URL: ${DATABASE_URL:-} REDIS_URL: ${REDIS_URL:-} + CANVAS_WORKER_CONCURRENCY: ${CANVAS_WORKER_CONCURRENCY:-3} + CANVAS_CHANNEL_CONCURRENCY: ${CANVAS_CHANNEL_CONCURRENCY:-3} CANVAS_REGISTRATION_ENABLED: ${CANVAS_REGISTRATION_ENABLED:-false} CANVAS_ALLOW_PRIVATE_UPSTREAMS: ${CANVAS_ALLOW_PRIVATE_UPSTREAMS:-false} CANVAS_ALLOWED_PRIVATE_UPSTREAM_HOSTS: ${CANVAS_ALLOWED_PRIVATE_UPSTREAM_HOSTS:-} diff --git a/web/src/pages/admin/channels/channels-page.tsx b/web/src/pages/admin/channels/channels-page.tsx index 6b7f22750..18bedb7a8 100644 --- a/web/src/pages/admin/channels/channels-page.tsx +++ b/web/src/pages/admin/channels/channels-page.tsx @@ -1,4 +1,4 @@ -import { App, Button, Drawer, Form, Input, Select, Switch, Table, Tag } from "antd"; +import { App, Button, Drawer, Form, Input, InputNumber, Select, Switch, Table, Tag } from "antd"; import type { ColumnsType } from "antd/es/table"; import { Pencil, Plus, Power, Search } from "lucide-react"; import { useEffect, useRef, useState } from "react"; @@ -14,7 +14,7 @@ import { AdminPageFrame } from "../components/admin-shell"; import { AdminRowActions, AdminTableEmpty, AdminTableSkeleton, configuredSecretText } from "../components/admin-ui"; import { ChannelModelManager } from "../components/channel-model-manager"; -type ChannelFormValues = { name: string; baseUrl: string; apiKey?: string; interfaceType: ChannelInterfaceType; enabled?: boolean }; +type ChannelFormValues = { name: string; baseUrl: string; apiKey?: string; interfaceType: ChannelInterfaceType; useGlobalConcurrency?: boolean; concurrencyLimit?: number; enabled?: boolean }; const interfaceTypeOptions = [ { label: "文本", options: [{ label: "Chat Completions", value: "chat-completion" }, { label: "OpenAI Responses", value: "openai-response" }] }, @@ -41,6 +41,7 @@ export default function ChannelsPage() { const [managingChannel, setManagingChannel] = useState(null); const requestSequence = useRef(0); const [form] = Form.useForm(); + const useGlobalConcurrency = Form.useWatch("useGlobalConcurrency", form) !== false; const hasFilters = Boolean(keyword || interfaceType !== "all" || status !== "all"); const updateUrl = (patch: Record, replace = false) => { @@ -85,7 +86,7 @@ export default function ChannelsPage() { const openDrawer = (channel?: ModelChannel) => { setEditingChannel(channel || null); form.resetFields(); - form.setFieldsValue(channel ? { name: channel.name, baseUrl: channel.baseUrl, apiKey: "", interfaceType: channel.interfaceType || "newapi", enabled: channel.enabled !== false } : { name: "", baseUrl: "", apiKey: "", interfaceType: "newapi", enabled: true }); + form.setFieldsValue(channel ? { name: channel.name, baseUrl: channel.baseUrl, apiKey: "", interfaceType: channel.interfaceType || "newapi", useGlobalConcurrency: !channel.concurrencyLimit, concurrencyLimit: channel.concurrencyLimit || undefined, enabled: channel.enabled !== false } : { name: "", baseUrl: "", apiKey: "", interfaceType: "newapi", useGlobalConcurrency: true, concurrencyLimit: undefined, enabled: true }); setDrawerOpen(true); }; @@ -106,7 +107,7 @@ export default function ChannelsPage() { } setSaving(true); try { - const payload = { name: values.name.trim(), baseUrl: values.baseUrl.trim(), apiKey: values.apiKey?.trim() || "", interfaceType: values.interfaceType, enabled: values.enabled !== false }; + const payload = { name: values.name.trim(), baseUrl: values.baseUrl.trim(), apiKey: values.apiKey?.trim() || "", interfaceType: values.interfaceType, useGlobalConcurrency: values.useGlobalConcurrency !== false, concurrencyLimit: values.useGlobalConcurrency === false ? values.concurrencyLimit : undefined, enabled: values.enabled !== false }; await (editingChannel ? updateAdminChannel(editingChannel.id, payload) : createAdminChannel(payload)); await syncChannels(); setDrawerOpen(false); @@ -136,6 +137,7 @@ export default function ChannelsPage() { { title: "渠道", dataIndex: "name", render: (_, channel) =>
{channel.name}
{channel.baseUrl}
}, { title: "接口类型", dataIndex: "interfaceType", width: 160, render: (value: ChannelInterfaceType) => {interfaceTypeLabel(value)} }, { title: "模型", dataIndex: "models", width: 100, render: (models: string[]) => `${models?.length || 0} 个` }, + { title: "最大并发", dataIndex: "concurrencyLimit", width: 120, render: (value: number) => value > 0 ? value : 跟随系统 }, { title: "密钥", dataIndex: "hasApiKey", width: 100, render: (configured) => {configured ? "已配置" : "未配置"} }, { title: "状态", dataIndex: "enabled", width: 100, render: (enabled) => {enabled !== false ? "已启用" : "已停用"} }, { title: "操作", width: 160, fixed: "right", align: "right", render: (_, channel) => setManagingChannel(channel) }} actions={[{ key: "edit", label: "编辑渠道", icon: , onClick: () => openDrawer(channel) }, { key: "toggle", label: channel.enabled !== false ? "停用渠道" : "启用渠道", icon: , danger: channel.enabled !== false, confirm: { title: channel.enabled !== false ? "停用这个系统渠道？" : "启用这个系统渠道？", description: channel.enabled !== false ? "停用后新任务不会再使用该渠道，历史账单和调用记录继续保留。" : "启用后，配置完整的模型会重新进入系统可用模型集合。", okText: channel.enabled !== false ? "确认停用" : "确认启用" }, onClick: () => toggleChannel(channel) }]} /> }, @@ -153,7 +155,7 @@ export default function ChannelsPage() { { const current = String(form.getFieldValue("baseUrl") || "").trim(); if (!current || current === defaultBaseUrlForChannelInterface()) form.setFieldValue("baseUrl", defaultBaseUrlForChannelInterface(value)); }} /> + + diff --git a/web/src/pages/admin/components/admin-shell.tsx b/web/src/pages/admin/components/admin-shell.tsx index 9279eec7b..066ca9ec3 100644 --- a/web/src/pages/admin/components/admin-shell.tsx +++ b/web/src/pages/admin/components/admin-shell.tsx @@ -1,5 +1,5 @@ import { Tooltip } from "antd"; -import { BarChart3, BellRing, Coins, FileClock, HardDrive, Home, Mail, MessageSquareText, PanelLeftClose, PanelLeftOpen, RadioTower, ShieldCheck, TicketCheck, UsersRound } from "lucide-react"; +import { BarChart3, BellRing, Coins, FileClock, Gauge, HardDrive, Home, Mail, MessageSquareText, PanelLeftClose, PanelLeftOpen, RadioTower, ShieldCheck, TicketCheck, UsersRound } from "lucide-react"; import { useState, type ReactNode } from "react"; import { NavLink, Outlet } from "react-router"; @@ -40,6 +40,7 @@ const adminNavigation: Array<{ label: string; items: AdminNavigationItem[] }> = { label: "系统配置", items: [ + { path: "/admin/settings/concurrency", label: "任务并发", description: "Worker 与渠道请求调度", icon: , secondary: true }, { path: "/admin/settings/access", label: "登录与注册", description: "注册策略与 Linux.do", icon: , secondary: true }, { path: "/admin/settings/email", label: "邮件服务", description: "注册验证码 SMTP", icon: , secondary: true }, { path: "/admin/settings/storage", label: "存储服务", description: "OSS 与资源存储", icon: , secondary: true }, diff --git a/web/src/pages/admin/components/api-log-detail-drawer.tsx b/web/src/pages/admin/components/api-log-detail-drawer.tsx index a7e69b711..fb623126b 100644 --- a/web/src/pages/admin/components/api-log-detail-drawer.tsx +++ b/web/src/pages/admin/components/api-log-detail-drawer.tsx @@ -33,6 +33,7 @@ export function ApiLogDetailDrawer({ logId, onClose }: { logId: string | null; o ["方法与路径", `${log.method} ${log.path}`], ["HTTP 状态", String(log.statusCode || "--")], ["耗时", `${log.durationMs} ms`], + ["渠道并发上限", log.concurrencyLimit ? String(log.concurrencyLimit) : "--"], ["Token", log.usageAvailable ? `${log.inputTokens} 输入 / ${log.outputTokens} 输出 / ${log.cachedTokens} 缓存` : "未返回"], ["错误码", log.errorCode || "--"], ["错误详情", log.error || "--"], diff --git a/web/src/pages/admin/settings/concurrency-settings-page.tsx b/web/src/pages/admin/settings/concurrency-settings-page.tsx new file mode 100644 index 000000000..2a7c80d63 --- /dev/null +++ b/web/src/pages/admin/settings/concurrency-settings-page.tsx @@ -0,0 +1,73 @@ +import { App, Button, Form, InputNumber, Tag } from "antd"; +import { Gauge, RadioTower, Workflow } from "lucide-react"; +import { useEffect, useMemo, useState } from "react"; + +import { getAdminRuntimeConcurrencySetting, updateAdminRuntimeConcurrencySetting, type RuntimeConcurrencySetting } from "@/services/api/auth"; +import { useAdminContext } from "../admin-context"; +import { AdminPageFrame } from "../components/admin-shell"; +import { SettingsSectionCard } from "../components/admin-ui"; + +type ConcurrencyFormValues = Pick; + +export default function ConcurrencySettingsPage() { + const { message } = App.useApp(); + const { references } = useAdminContext(); + const [setting, setSetting] = useState(null); + const [loading, setLoading] = useState(true); + const [saving, setSaving] = useState(false); + const [form] = Form.useForm(); + const userNameById = useMemo(() => new Map(references.users.map((user) => [user.id, user.displayName || user.username])), [references.users]); + + useEffect(() => { + void getAdminRuntimeConcurrencySetting() + .then(({ setting: value }) => { + setSetting(value); + form.setFieldsValue(value); + }) + .catch((error) => message.error(error instanceof Error ? error.message : "读取任务并发配置失败")) + .finally(() => setLoading(false)); + }, [form, message]); + + const save = async () => { + const values = await form.validateFields(); + setSaving(true); + try { + const result = await updateAdminRuntimeConcurrencySetting(values); + setSetting(result.setting); + form.setFieldsValue(result.setting); + message.success("任务并发配置已生效"); + } catch (error) { + message.error(error instanceof Error ? error.message : "保存任务并发配置失败"); + } finally { + setSaving(false); + } + }; + + return ( + +
+ } + title="运行时并发" + description="控制任务消费速度和未单独配置渠道的默认请求并发。" + status={热更新} + footer={<>
{setting?.updatedAt ? `上次更新：${formatTime(setting.updatedAt)}${setting.updatedBy ? ` · ${userNameById.get(setting.updatedBy) || setting.updatedBy}` : ""}` : "当前使用环境变量默认值"}
} + > +
+
+ } /> + } /> +
+
+ +
任务可以继续进入队列；达到渠道上限时会等待空闲槽位，不会因暂时满载直接失败。单独配置的渠道上限优先于这里的全局值。
+
+ + ); +} + +function concurrencyRules(label: string) { + return [{ required: true, message: `请填写${label}` }, { type: "number" as const, min: 1, max: 100, message: `${label}必须是 1-100 的整数` }]; +} + +function formatTime(value?: string) { return value ? new Date(value).toLocaleString("zh-CN", { hour12: false }) : "--"; } diff --git a/web/src/router.tsx b/web/src/router.tsx index 68396231b..2bd8429ae 100644 --- a/web/src/router.tsx +++ b/web/src/router.tsx @@ -7,6 +7,7 @@ import { AccessSettingsPage, AnalyticsPage, AnnouncementsPage, CreditOperationsP import ChannelsPage from "@/pages/admin/channels/channels-page"; import LogsPage from "@/pages/admin/logs/logs-page"; import RedemptionCodesPage from "@/pages/admin/redemption-codes/redemption-codes-page"; +import ConcurrencySettingsPage from "@/pages/admin/settings/concurrency-settings-page"; import StorageSettingsPage from "@/pages/admin/settings/storage-settings-page"; import StoryboardPromptsPage from "@/pages/admin/storyboard-prompts/storyboard-prompts-page"; import UsersPage from "@/pages/admin/users/users-page"; @@ -61,7 +62,8 @@ export const router = createBrowserRouter([ { path: "credit-operations", element: }, { path: "redemption-codes", element: }, { path: "logs", element: }, - { path: "settings", element: }, + { path: "settings", element: }, + { path: "settings/concurrency", element: }, { path: "settings/access", element: }, { path: "settings/email", element: }, { path: "settings/storage", element: }, diff --git a/web/src/services/api/auth.ts b/web/src/services/api/auth.ts index b54ca1fc0..f183ba511 100644 --- a/web/src/services/api/auth.ts +++ b/web/src/services/api/auth.ts @@ -61,6 +61,7 @@ export type ApiCallLog = { currency?: string; errorCode?: string; error?: string; + concurrencyLimit: number; upstreamUrl: string; createdAt: string; }; @@ -196,6 +197,14 @@ export type AdminOSSSetting = { updatedAt?: string; }; +export type RuntimeConcurrencySetting = { + workerConcurrency: number; + channelConcurrency: number; + updatedBy?: string; + createdAt?: string; + updatedAt?: string; +}; + type BackendEnvelope = { code: number; data: T; msg: string }; async function request(promise: Promise<{ data: BackendEnvelope }>) { @@ -284,11 +293,11 @@ export function listAdminChannels(params: AdminListParams = {}) { return request<{ channels: ModelChannel[]; total: number; page: number; limit: number }>(api.get("/admin/channels", { params })); } -export function createAdminChannel(input: Partial) { +export function createAdminChannel(input: Partial & { useGlobalConcurrency?: boolean }) { return request<{ channel: ModelChannel }>(api.post("/admin/channels", input)); } -export function updateAdminChannel(id: string, input: Partial) { +export function updateAdminChannel(id: string, input: Partial & { useGlobalConcurrency?: boolean }) { return request<{ channel: ModelChannel }>(api.patch(`/admin/channels/${encodeURIComponent(id)}`, input)); } @@ -320,6 +329,14 @@ export function updateAdminOSSSetting(input: Partial) { return request<{ setting: AdminOSSSetting }>(api.patch("/admin/settings/oss", input)); } +export function getAdminRuntimeConcurrencySetting() { + return request<{ setting: RuntimeConcurrencySetting }>(api.get("/admin/settings/concurrency")); +} + +export function updateAdminRuntimeConcurrencySetting(input: Pick) { + return request<{ setting: RuntimeConcurrencySetting }>(api.patch("/admin/settings/concurrency", input)); +} + export function listAdminApiLogs(params: AdminListParams = {}) { return request<{ logs: ApiCallLog[]; total: number; page: number; limit: number }>(api.get("/admin/api-logs", { params })); } diff --git a/web/src/stores/use-config-store.ts b/web/src/stores/use-config-store.ts index 28de5555b..5c51d4c4b 100644 --- a/web/src/stores/use-config-store.ts +++ b/web/src/stores/use-config-store.ts @@ -20,6 +20,7 @@ export type ModelChannel = { scope?: "system" | "user"; enabled?: boolean; hasApiKey?: boolean; + concurrencyLimit?: number; modelCosts?: Array<{ model: string; capability: ModelCapability;