diff --git a/CHANGELOG.md b/CHANGELOG.md index f89a924ad..a26fa1529 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,8 @@ ## Unreleased +- 管理后台新增统一资源与请求策略配置，支持默认重置、自用模式、即时生效和用户配额用量查看。 + ## v1.0.4 - 系统后台支持热更新 Worker 与全局渠道并发，各系统渠道可跟随全局值或单独覆盖；渠道满载时请求等待空闲槽位，不再直接失败。 diff --git a/README.md b/README.md index d3e1dcdab..4635c846c 100644 --- a/README.md +++ b/README.md @@ -120,7 +120,7 @@ bun run dev 打开 `http://localhost:3000`，注册首个管理员账号，再在系统设置中配置渠道和模型。 -任务 Worker 和全局渠道并发可在“系统配置 → 任务并发”中热更新；系统渠道可选择跟随全局值，或单独设置 `1-100` 的最大并发数。未保存后台配置时分别回退到 `CANVAS_WORKER_CONCURRENCY` 和 `CANVAS_CHANNEL_CONCURRENCY`，两者默认均为 `3`。渠道槽位暂满时任务会等待，不会直接标记失败。 +资源配额、Worker/渠道/账号任务并发、业务频控、任务超时和渠道中转策略可在“系统配置 → 资源与策略”中统一热更新，支持重置和自用模式；系统渠道可选择跟随全局值，或单独设置 `1-999` 的最大并发数。未保存后台配置时 Worker 和全局渠道并发分别回退到 `CANVAS_WORKER_CONCURRENCY` 和 `CANVAS_CHANNEL_CONCURRENCY`，两者默认均为 `3`。渠道槽位暂满时任务会等待，不会直接标记失败。 Docker 一体化运行： @@ -133,7 +133,7 @@ docker compose -f docker-compose.local.yml up -d --build - 用户自定义 AI API Key 保存在浏览器本地；创建异步任务时会提交给自部署后端并加密入队，生产环境必须使用 HTTPS。 - 画布和素材登录后同步到后端，本地 `localForage` 继续承担缓存和降级存储。 - 媒体资源在启用 OSS 时保存到私有 OSS，否则保存到后端数据目录；删除业务记录不会自动清理 OSS 对象。 -- 用户主动上传的单个文件必须小于 50MB，每个账号按 UTC 自然日累计必须小于 200MB；AI 生成结果不计入主动上传额度。 +- 用户主动上传、Agent 会话附件和 AI 生成资源的单文件上限、账号容量及 UTC 日上传总量由后台“资源与策略”统一维护，默认分别为 50MB、32MB、64MB、2GB 和 200MB；管理员可按可信部署需要调整，单文件业务上限最高 999MB，Nginx 请求体硬上限为 1024MB。 ## 公网部署安全 diff --git a/backend/internal/handler/auth.go b/backend/internal/handler/auth.go index 06df6e3af..db176eb9b 100644 --- a/backend/internal/handler/auth.go +++ b/backend/internal/handler/auth.go @@ -4,6 +4,7 @@ import ( "bytes" "encoding/json" "errors" + "fmt" "io" "net/http" "net/url" @@ -33,7 +34,8 @@ func RegisterAuthRoutes(r *gin.RouterGroup, svc *service.Service) { fail(c, http.StatusBadRequest, err) return } - if !enforceRateLimit(c, "register:"+c.ClientIP(), 5, time.Hour) { + policy, available := loadRuntimePolicy(c, svc) + if !available || !enforceRateLimit(c, "register:"+c.ClientIP(), policy.Request.RegisterPerHour, time.Hour) { return } result, err := svc.Register(req) @@ -53,7 +55,8 @@ func RegisterAuthRoutes(r *gin.RouterGroup, svc *service.Service) { fail(c, http.StatusBadRequest, err) return } - if !enforceRateLimit(c, "email-code:"+c.ClientIP(), 10, time.Hour) { + policy, available := loadRuntimePolicy(c, svc) + if !available || !enforceRateLimit(c, "email-code:"+c.ClientIP(), policy.Request.EmailCodePerHour, time.Hour) { return } if err := svc.SendRegistrationEmailCode(req.Email); err != nil { @@ -69,10 +72,11 @@ func RegisterAuthRoutes(r *gin.RouterGroup, svc *service.Service) { fail(c, http.StatusBadRequest, err) return } - if !enforceRateLimit(c, "login-ip:"+c.ClientIP(), 50, 10*time.Minute) { + policy, available := loadRuntimePolicy(c, svc) + if !available || !enforceRateLimit(c, "login-ip:"+c.ClientIP(), policy.Request.LoginIPPerTenMinutes, 10*time.Minute) { return } - if !enforceRateLimit(c, "login:"+c.ClientIP()+":"+strings.ToLower(strings.TrimSpace(req.Username)), 10, 10*time.Minute) { + if !enforceRateLimit(c, "login:"+c.ClientIP()+":"+strings.ToLower(strings.TrimSpace(req.Username)), policy.Request.LoginAccountPerTenMinutes, 10*time.Minute) { return } result, err := svc.Login(req) @@ -112,7 +116,12 @@ func RegisterAuthRoutes(r *gin.RouterGroup, svc *service.Service) { return } channels, _ := svc.PublicSystemChannels() - ok(c, gin.H{"user": publicUser, "systemChannels": channels}) + limits, err := svc.PublicRuntimeLimits() + if err != nil { + failService(c, err) + return + } + ok(c, gin.H{"user": publicUser, "systemChannels": channels, "runtimeLimits": limits}) }) r.GET("/channels/system", func(c *gin.Context) { if _, err := currentUser(c, svc); err != nil { @@ -439,31 +448,58 @@ func RegisterAdminRoutes(r *gin.RouterGroup, svc *service.Service) { } ok(c, gin.H{"setting": setting}) }) - r.GET("/admin/settings/concurrency", func(c *gin.Context) { + r.GET("/admin/settings/runtime-policy", func(c *gin.Context) { user, err := currentUser(c, svc) if err != nil { failService(c, err) return } - setting, err := svc.AdminRuntimeConcurrencySetting(user) + setting, err := svc.AdminRuntimePolicySetting(user) if err != nil { failService(c, err) return } ok(c, gin.H{"setting": setting}) }) - r.PATCH("/admin/settings/concurrency", func(c *gin.Context) { + r.GET("/admin/settings/runtime-policy/self-use", func(c *gin.Context) { user, err := currentUser(c, svc) if err != nil { failService(c, err) return } - var req service.RuntimeConcurrencySettingRequest + setting, err := svc.AdminSelfUseRuntimePolicy(user) + if err != nil { + failService(c, err) + return + } + ok(c, gin.H{"setting": setting}) + }) + r.PUT("/admin/settings/runtime-policy", func(c *gin.Context) { + user, err := currentUser(c, svc) + if err != nil { + failService(c, err) + return + } + c.Request.Body = http.MaxBytesReader(c.Writer, c.Request.Body, 64<<10) + var req service.RuntimePolicySetting if err := c.ShouldBindJSON(&req); err != nil { fail(c, http.StatusBadRequest, err) return } - setting, err := svc.UpdateRuntimeConcurrencySetting(user, req) + setting, err := svc.UpdateRuntimePolicySetting(user, req) + if err != nil { + failService(c, err) + return + } + ok(c, gin.H{"setting": setting}) + }) + r.DELETE("/admin/settings/runtime-policy", func(c *gin.Context) { + user, err := currentUser(c, svc) + if err != nil { + failService(c, err) + return + } + setting, err := svc.ResetRuntimePolicySetting(user) if err != nil { failService(c, err) return @@ -536,14 +572,15 @@ func RegisterSystemProxyRoutes(r *gin.RouterGroup, svc *service.Service) { func proxySystemRequest(c *gin.Context, svc *service.Service, user *model.User, channel *model.ModelChannel) { startedAt := time.Now() - if !enforceRateLimit(c, "system-proxy:"+user.ID, 120, time.Minute) { + policy, available := loadRuntimePolicy(c, svc) + if !available || !enforceRateLimit(c, "system-proxy:"+user.ID, policy.Request.SystemRelayPerMinute, time.Minute) { return } path := c.Param("path") if path == "" { path = "/" } - c.Request.Body = http.MaxBytesReader(c.Writer, c.Request.Body, maxSystemProxyBodyBytes) + c.Request.Body = http.MaxBytesReader(c.Writer, c.Request.Body, policy.Request.SystemRelayRequestMB<<20) body, err := io.ReadAll(c.Request.Body) if err != nil { fail(c, http.StatusBadRequest, err) @@ -633,7 +670,8 @@ func proxySystemRequest(c *gin.Context, svc *service.Service, user *model.User, if resp.StatusCode < 200 || resp.StatusCode >= 300 { status = model.ApiCallStatusFailed } - responseBody, readErr := io.ReadAll(io.LimitReader(resp.Body, (128<<20)+1)) + responseLimit := policy.Request.SystemRelayResponseMB << 20 + responseBody, readErr := io.ReadAll(io.LimitReader(resp.Body, responseLimit+1)) if readErr != nil { status = model.ApiCallStatusFailed errorText = readErr.Error() @@ -642,9 +680,9 @@ func proxySystemRequest(c *gin.Context, svc *service.Service, user *model.User, fail(c, http.StatusBadGateway, errors.New("系统渠道响应读取失败")) return } - if len(responseBody) > 128<<20 { + if int64(len(responseBody)) > responseLimit { _ = svc.MarkBillingUncertain(billingOrderID, "上游已响应但响应体超过限制，费用状态待核对") - fail(c, http.StatusBadGateway, errors.New("系统渠道响应超过 128MB 限制")) + fail(c, http.StatusBadGateway, fmt.Errorf("系统渠道响应超过 %dMB 限制", policy.Request.SystemRelayResponseMB)) return } if status == model.ApiCallStatusSucceeded { diff --git a/backend/internal/handler/custom_proxy.go b/backend/internal/handler/custom_proxy.go index 831ab4d14..6055d6e47 100644 --- a/backend/internal/handler/custom_proxy.go +++ b/backend/internal/handler/custom_proxy.go @@ -8,7 +8,6 @@ import ( "mime" "net/http" "strings" - "sync" "time" "infinite-canvas/backend/internal/service" @@ -16,23 +15,9 @@ import ( "github.com/gin-gonic/gin" ) -const ( - maxCustomRelayBodyBytes int64 = 32 << 20 - maxCustomRelayModelResponseBytes int64 = 4 << 20 - maxCustomRelayJSONResponseBytes int64 = 16 << 20 - maxCustomRelayErrorResponseBytes int64 = 64 << 10 - maxCustomRelayStreamBytes int64 = 32 << 20 - maxCustomRelayConcurrency = 4 - customRelayHTTPTimeout = 10 * time.Minute -) +const maxCustomRelayErrorResponseBytes int64 = 64 << 10 -var ( - customRelayClient = service.CustomRelayHTTPClient - customRelayActive = struct { - sync.Mutex - users map[string]int - }{users: map[string]int{}} -) +var customRelayClient = service.CustomRelayHTTPClient func RegisterCustomRelayRoutes(r *gin.RouterGroup, svc *service.Service) { r.Any("/ai/custom", func(c *gin.Context) { @@ -41,19 +26,26 @@ func RegisterCustomRelayRoutes(r *gin.RouterGroup, svc *service.Service) { failService(c, err) return } - if !enforceRateLimit(c, "custom-relay:"+user.ID, 120, time.Minute) { + policy, available := loadRuntimePolicy(c, svc) + if !available || !enforceRateLimit(c, "custom-relay:"+user.ID, policy.Request.CustomRelayPerMinute, time.Minute) { return } - if !acquireCustomRelaySlot(user.ID) { + ttl := time.Duration(policy.Request.CustomRelayTimeoutMinutes+1) * time.Minute + release, acquired, err := svc.AcquireCustomRelaySlot(c.Request.Context(), user.ID, policy.Request.CustomRelayConcurrency, ttl) + if err != nil { + fail(c, http.StatusServiceUnavailable, errors.New("自定义渠道并发协调服务不可用")) + return + } + if !acquired { fail(c, http.StatusTooManyRequests, errors.New("自定义渠道并发请求过多，请等待已有请求完成")) return } - defer releaseCustomRelaySlot(user.ID) - proxyCustomRelayRequest(c) + defer release() + proxyCustomRelayRequest(c, policy.Request) }) } -func proxyCustomRelayRequest(c *gin.Context) { +func proxyCustomRelayRequest(c *gin.Context, policy service.RuntimeRequestPolicy) { target, err := service.ValidateCustomRelayURL(c.GetHeader("X-Canvas-Upstream-URL")) if err != nil { failService(c, err) @@ -72,16 +64,17 @@ func proxyCustomRelayRequest(c *gin.Context) { fail(c, http.StatusUnauthorized, err) return } - if c.Request.ContentLength > maxCustomRelayBodyBytes { - fail(c, http.StatusRequestEntityTooLarge, errors.New("自定义渠道请求不能超过 32MB")) + requestLimit := policy.CustomRelayRequestMB << 20 + if c.Request.ContentLength > requestLimit { + fail(c, http.StatusRequestEntityTooLarge, errors.New("自定义渠道请求超过配置上限")) return } - c.Request.Body = http.MaxBytesReader(c.Writer, c.Request.Body, maxCustomRelayBodyBytes) + c.Request.Body = http.MaxBytesReader(c.Writer, c.Request.Body, requestLimit) body, err := io.ReadAll(c.Request.Body) if err != nil { var maxBytesError *http.MaxBytesError if errors.As(err, &maxBytesError) { - fail(c, http.StatusRequestEntityTooLarge, errors.New("自定义渠道请求不能超过 32MB")) + fail(c, http.StatusRequestEntityTooLarge, errors.New("自定义渠道请求超过配置上限")) return } fail(c, http.StatusBadRequest, errors.New("读取自定义渠道请求失败")) @@ -111,16 +104,16 @@ func proxyCustomRelayRequest(c *gin.Context) { upstreamReq.Header.Set("Authorization", "Bearer "+apiKey) } - resp, err := customRelayClient(customRelayHTTPTimeout).Do(upstreamReq) + resp, err := customRelayClient(time.Duration(policy.CustomRelayTimeoutMinutes) * time.Minute).Do(upstreamReq) if err != nil { fail(c, http.StatusBadGateway, errors.New("自定义渠道上游连接失败")) return } defer resp.Body.Close() - writeCustomRelayResponse(c, resp, apiKey, c.Request.Method == http.MethodGet) + writeCustomRelayResponse(c, resp, apiKey, policy.CustomRelayResponseMB<<20) } -func writeCustomRelayResponse(c *gin.Context, resp *http.Response, apiKey string, modelList bool) { +func writeCustomRelayResponse(c *gin.Context, resp *http.Response, apiKey string, responseLimit int64) { c.Header("Cache-Control", "no-store") c.Header("X-Content-Type-Options", "nosniff") mediaType, _, _ := mime.ParseMediaType(resp.Header.Get("Content-Type")) @@ -133,17 +126,14 @@ func writeCustomRelayResponse(c *gin.Context, resp *http.Response, apiKey string c.Header("X-Accel-Buffering", "no") c.Status(resp.StatusCode) c.Writer.WriteHeaderNow() - copyCustomRelayStream(c, resp.Body, apiKey) + copyCustomRelayStream(c, resp.Body, apiKey, responseLimit) return } if mediaType != "application/json" && !strings.HasSuffix(mediaType, "+json") { fail(c, http.StatusBadGateway, errors.New("自定义渠道上游返回了不支持的内容类型")) return } - limit := maxCustomRelayJSONResponseBytes - if modelList { - limit = maxCustomRelayModelResponseBytes - } + limit := responseLimit body, err := readLimitedRelayBody(resp.Body, limit) if err != nil || !json.Valid(body) { fail(c, http.StatusBadGateway, errors.New("自定义渠道上游返回无效或过大的 JSON")) @@ -167,14 +157,14 @@ func writeCustomRelayError(c *gin.Context, resp *http.Response, apiKey string, m fail(c, resp.StatusCode, errors.New("自定义渠道上游请求失败")) } -func copyCustomRelayStream(c *gin.Context, source io.Reader, apiKey string) { +func copyCustomRelayStream(c *gin.Context, source io.Reader, apiKey string, maxBytes int64) { redactor := newRelayStreamRedactor(apiKey) buffer := make([]byte, 32<<10) var written int64 - for written < maxCustomRelayStreamBytes { + for written < maxBytes { read, err := source.Read(buffer) if read > 0 { - remaining := maxCustomRelayStreamBytes - written + remaining := maxBytes - written if int64(read) > remaining { read = int(remaining) } @@ -217,26 +207,6 @@ func customRelayAPIKey(value string) (string, error) { return apiKey, nil } -func acquireCustomRelaySlot(userID string) bool { - customRelayActive.Lock() - defer customRelayActive.Unlock() - if customRelayActive.users[userID] >= maxCustomRelayConcurrency { - return false - } - customRelayActive.users[userID]++ - return true -} - -func releaseCustomRelaySlot(userID string) { - customRelayActive.Lock() - defer customRelayActive.Unlock() - if customRelayActive.users[userID] <= 1 { - delete(customRelayActive.users, userID) - return - } - customRelayActive.users[userID]-- -} - func redactRelaySecret(body []byte, apiKey string) []byte { if apiKey == "" { return body diff --git a/backend/internal/handler/custom_proxy_test.go b/backend/internal/handler/custom_proxy_test.go index b6f6e620d..d856f7dc6 100644 --- a/backend/internal/handler/custom_proxy_test.go +++ b/backend/internal/handler/custom_proxy_test.go @@ -10,6 +10,8 @@ import ( "testing" "time" + "infinite-canvas/backend/internal/service" + "github.com/gin-gonic/gin" ) @@ -44,7 +46,7 @@ func TestCustomRelayForwardsOpenAIRequestWithoutBrowserHeaders(t *testing.T) { context, _ := gin.CreateTestContext(response) context.Request = request - proxyCustomRelayRequest(context) + proxyCustomRelayRequest(context, defaultCustomRelayTestPolicy()) if response.Code != http.StatusOK { t.Fatalf("status = %d, body = %s", response.Code, response.Body.String()) } @@ -82,7 +84,7 @@ func TestCustomRelayConvertsGeminiAuthentication(t *testing.T) { context, _ := gin.CreateTestContext(response) context.Request = request - proxyCustomRelayRequest(context) + proxyCustomRelayRequest(context, defaultCustomRelayTestPolicy()) if response.Code != http.StatusOK { t.Fatalf("status = %d, body = %s", response.Code, response.Body.String()) } @@ -106,7 +108,7 @@ func TestCustomRelayStreamsBeforeUpstreamCompletes(t *testing.T) { proxy := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { context, _ := gin.CreateTestContext(w) context.Request = r - proxyCustomRelayRequest(context) + proxyCustomRelayRequest(context, defaultCustomRelayTestPolicy()) })) defer proxy.Close() request, _ := http.NewRequest(http.MethodPost, proxy.URL, strings.NewReader(`{"model":"test"}`)) @@ -157,7 +159,7 @@ func TestCustomRelayRejectsOversizedDeclaredBodyBeforeConnecting(t *testing.T) { t.Cleanup(func() { customRelayClient = previous }) request := httptest.NewRequest(http.MethodPost, "/api/ai/custom", strings.NewReader(`{"model":"test"}`)) - request.ContentLength = maxCustomRelayBodyBytes + 1 + request.ContentLength = (defaultCustomRelayTestPolicy().CustomRelayRequestMB << 20) + 1 request.Header.Set("Authorization", "Bearer test-key") request.Header.Set("Content-Type", "application/json") request.Header.Set("X-Canvas-Upstream-URL", "https://127.0.0.1/v1/responses") @@ -166,7 +168,7 @@ func TestCustomRelayRejectsOversizedDeclaredBodyBeforeConnecting(t *testing.T) { context, _ := gin.CreateTestContext(response) context.Request = request - proxyCustomRelayRequest(context) + proxyCustomRelayRequest(context, defaultCustomRelayTestPolicy()) if response.Code != http.StatusRequestEntityTooLarge { t.Fatalf("status = %d, body = %s", response.Code, response.Body.String()) } @@ -175,28 +177,15 @@ func TestCustomRelayRejectsOversizedDeclaredBodyBeforeConnecting(t *testing.T) { } } -func TestCustomRelayConcurrencyLimitReleasesSlots(t *testing.T) { - const userID = "relay-limit-test-user" - for index := 0; index < maxCustomRelayConcurrency; index++ { - if !acquireCustomRelaySlot(userID) { - t.Fatalf("slot %d should be available", index) - } - } - if acquireCustomRelaySlot(userID) { - t.Fatal("request above the concurrency limit should be rejected") - } - for index := 0; index < maxCustomRelayConcurrency; index++ { - releaseCustomRelaySlot(userID) - } - if !acquireCustomRelaySlot(userID) { - t.Fatal("released slot should be reusable") - } - releaseCustomRelaySlot(userID) -} - func useCustomRelayTestClient(t *testing.T, client *http.Client) { t.Helper() previous := customRelayClient customRelayClient = func(time.Duration) *http.Client { return client } t.Cleanup(func() { customRelayClient = previous }) } + +func defaultCustomRelayTestPolicy() service.RuntimeRequestPolicy { + return service.RuntimeRequestPolicy{ + CustomRelayRequestMB: 32, CustomRelayResponseMB: 32, CustomRelayTimeoutMinutes: 10, + } +} diff --git a/backend/internal/handler/routes.go b/backend/internal/handler/routes.go index c501604a3..6841910be 100644 --- a/backend/internal/handler/routes.go +++ b/backend/internal/handler/routes.go @@ -17,7 +17,8 @@ func RegisterTaskRoutes(r *gin.RouterGroup, svc *service.Service) { failService(c, err) return } - if !enforceRateLimit(c, "tasks:"+user.ID, 30, time.Minute) { + policy, available := loadRuntimePolicy(c, svc) + if !available || !enforceRateLimit(c, "tasks:"+user.ID, policy.Request.TaskCreatePerMinute, time.Minute) { return } c.Request.Body = http.MaxBytesReader(c.Writer, c.Request.Body, 16<<20) @@ -112,7 +113,8 @@ func RegisterSessionRoutes(r *gin.RouterGroup, svc *service.Service) { failService(c, err) return } - if !enforceRateLimit(c, "sessions:"+user.ID, 20, time.Minute) { + policy, available := loadRuntimePolicy(c, svc) + if !available || !enforceRateLimit(c, "sessions:"+user.ID, policy.Request.SessionCreatePerMinute, time.Minute) { return } c.Request.Body = http.MaxBytesReader(c.Writer, c.Request.Body, 16<<20) @@ -147,10 +149,11 @@ func RegisterSessionRoutes(r *gin.RouterGroup, svc *service.Service) { failService(c, err) return } - if !enforceRateLimit(c, "session-files:"+user.ID, 30, time.Minute) { + policy, available := loadRuntimePolicy(c, svc) + if !available || !enforceRateLimit(c, "session-files:"+user.ID, policy.Request.SessionFilePerMinute, time.Minute) { return } - c.Request.Body = http.MaxBytesReader(c.Writer, c.Request.Body, service.MaxSessionUploadBytes+(1<<20)) + c.Request.Body = http.MaxBytesReader(c.Writer, c.Request.Body, (policy.Resource.SessionUploadMB<<20)+(1<<20)) file, err := c.FormFile("file") if err != nil { fail(c, http.StatusBadRequest, err) diff --git a/backend/internal/handler/security.go b/backend/internal/handler/security.go index e725d7b5b..080a9c83d 100644 --- a/backend/internal/handler/security.go +++ b/backend/internal/handler/security.go @@ -20,8 +20,6 @@ import ( "github.com/gin-gonic/gin" ) -const maxSystemProxyBodyBytes int64 = 64 << 20 - var ( runtimeService *service.Service geminiGeneratePath = regexp.MustCompile(`^/models/([^/:]+):(generateContent|streamGenerateContent)$`) @@ -120,6 +118,15 @@ func enforceRateLimit(c *gin.Context, key string, limit int, window time.Duratio return false } +func loadRuntimePolicy(c *gin.Context, svc *service.Service) (service.RuntimePolicySetting, bool) { + policy, err := svc.RuntimePolicy() + if err != nil { + fail(c, http.StatusServiceUnavailable, errors.New("读取运行时策略失败："+err.Error())) + return service.RuntimePolicySetting{}, false + } + return policy, true +} + func authorizeSystemProxy(channel *model.ModelChannel, method string, requestPath string, contentType string, body []byte) error { requestPath, err := normalizedProxyPath(requestPath) if err != nil { diff --git a/backend/internal/handler/user_data.go b/backend/internal/handler/user_data.go index d3aed32c3..7d2936ca6 100644 --- a/backend/internal/handler/user_data.go +++ b/backend/internal/handler/user_data.go @@ -68,10 +68,11 @@ func RegisterUserDataRoutes(r *gin.RouterGroup, svc *service.Service) { failService(c, err) return } - if !enforceRateLimit(c, "resources-upload:"+user.ID, 30, time.Minute) { + policy, available := loadRuntimePolicy(c, svc) + if !available || !enforceRateLimit(c, "resources-upload:"+user.ID, policy.Request.ResourceUploadPerMinute, time.Minute) { return } - c.Request.Body = http.MaxBytesReader(c.Writer, c.Request.Body, service.MaxResourceUploadBytes+(1<<20)) + c.Request.Body = http.MaxBytesReader(c.Writer, c.Request.Body, (policy.Resource.ResourceUploadMB<<20)+(1<<20)) file, err := c.FormFile("file") if err != nil { fail(c, http.StatusBadRequest, err) @@ -93,7 +94,8 @@ func RegisterUserDataRoutes(r *gin.RouterGroup, svc *service.Service) { failService(c, err) return } - if !enforceRateLimit(c, "resources-import:"+user.ID, 30, time.Minute) { + policy, available := loadRuntimePolicy(c, svc) + if !available || !enforceRateLimit(c, "resources-import:"+user.ID, policy.Request.ResourceImportPerMinute, time.Minute) { return } c.Request.Body = http.MaxBytesReader(c.Writer, c.Request.Body, 64<<10) @@ -247,7 +249,8 @@ func RegisterUserDataRoutes(r *gin.RouterGroup, svc *service.Service) { failService(c, err) return } - if !enforceRateLimit(c, "assets-write:"+user.ID, 120, time.Minute) { + policy, available := loadRuntimePolicy(c, svc) + if !available || !enforceRateLimit(c, "assets-write:"+user.ID, policy.Request.AssetWritePerMinute, time.Minute) { return } c.Request.Body = http.MaxBytesReader(c.Writer, c.Request.Body, 5<<20) @@ -316,7 +319,8 @@ func RegisterUserDataRoutes(r *gin.RouterGroup, svc *service.Service) { failService(c, err) return } - if !enforceRateLimit(c, "canvas-write:"+user.ID, 120, time.Minute) { + policy, available := loadRuntimePolicy(c, svc) + if !available || !enforceRateLimit(c, "canvas-write:"+user.ID, policy.Request.CanvasWritePerMinute, time.Minute) { return } c.Request.Body = http.MaxBytesReader(c.Writer, c.Request.Body, 5<<20) diff --git a/backend/internal/repository/finance.go b/backend/internal/repository/finance.go index 631418907..62dd43e03 100644 --- a/backend/internal/repository/finance.go +++ b/backend/internal/repository/finance.go @@ -144,9 +144,9 @@ func (r *Repository) CreditLedgerReferenceExists(referenceKey string) (bool, err return count > 0, err } -func (r *Repository) CreateTaskWithCreditReservation(task *model.Task, order *model.BillingOrder) error { +func (r *Repository) CreateTaskWithCreditReservation(task *model.Task, order *model.BillingOrder, activeTaskLimit int) error { return r.db.Transaction(func(tx *gorm.DB) error { - if err := enforceActiveTaskLimit(tx, task.UserID); err != nil { + if err := enforceActiveTaskLimit(tx, task.UserID, activeTaskLimit); err != nil { return err } if err := reserveBillingOrder(tx, order); err != nil { @@ -156,19 +156,19 @@ func (r *Repository) CreateTaskWithCreditReservation(task *model.Task, order *mo }) } -func (r *Repository) CreateTaskWithActiveLimit(task *model.Task) error { +func (r *Repository) CreateTaskWithActiveLimit(task *model.Task, activeTaskLimit int) error { return r.db.Transaction(func(tx *gorm.DB) error { - if err := enforceActiveTaskLimit(tx, task.UserID); err != nil { + if err := enforceActiveTaskLimit(tx, task.UserID, activeTaskLimit); err != nil { return err } return tx.Create(task).Error }) } -func (r *Repository) RetryTaskWithBilling(userID string, taskID string, order *model.BillingOrder) (*model.Task, error) { +func (r *Repository) RetryTaskWithBilling(userID string, taskID string, order *model.BillingOrder, activeTaskLimit int) (*model.Task, error) { var task model.Task err := r.db.Transaction(func(tx *gorm.DB) error { - if err := enforceActiveTaskLimit(tx, userID); err != nil { + if err := enforceActiveTaskLimit(tx, userID, activeTaskLimit); err != nil { return err } if order != nil { @@ -197,12 +197,12 @@ func (r *Repository) RetryTaskWithBilling(userID string, taskID string, order *m return &task, err } -func enforceActiveTaskLimit(tx *gorm.DB, userID string) error { +func enforceActiveTaskLimit(tx *gorm.DB, userID string, activeTaskLimit int) error { var count int64 if err := tx.Model(&model.Task{}).Where("user_id = ? AND status IN ?", userID, []model.TaskStatus{model.TaskStatusQueued, model.TaskStatusRunning}).Count(&count).Error; err != nil { return err } - if count >= 5 { + if count >= int64(activeTaskLimit) { return ErrActiveTaskLimit } return nil diff --git a/backend/internal/repository/repository.go b/backend/internal/repository/repository.go index f2830c48b..236579c14 100644 --- a/backend/internal/repository/repository.go +++ b/backend/internal/repository/repository.go @@ -18,15 +18,15 @@ type Repository struct { } type UserStorageUsage struct { - AssetCount int64 - AssetBytes int64 - CanvasCount int64 - CanvasBytes int64 - SessionCount int64 - SessionBytes int64 - TaskCount int64 - TaskBytes int64 - APICallCount int64 + AssetCount int64 `json:"assetCount"` + AssetBytes int64 `json:"assetBytes"` + CanvasCount int64 `json:"canvasCount"` + CanvasBytes int64 `json:"canvasBytes"` + SessionCount int64 `json:"sessionCount"` + SessionBytes int64 `json:"sessionBytes"` + TaskCount int64 `json:"taskCount"` + TaskBytes int64 `json:"taskBytes"` + APICallCount int64 `json:"apiCallCount"` } func New(db *gorm.DB) *Repository { @@ -532,6 +532,10 @@ func (r *Repository) SaveSystemSetting(setting *model.SystemSetting) error { return r.db.Save(setting).Error } +func (r *Repository) DeleteSystemSetting(key string) error { + return r.db.Delete(&model.SystemSetting{}, "key = ?", key).Error +} + func (r *Repository) LatestUserOSSSetting(userID string) (*model.UserOSSSetting, error) { var setting model.UserOSSSetting if err := r.db.Where("user_id = ?", userID).Order("created_at desc, id desc").First(&setting).Error; err != nil { diff --git a/backend/internal/service/admin.go b/backend/internal/service/admin.go index 3e444529b..a3f22158e 100644 --- a/backend/internal/service/admin.go +++ b/backend/internal/service/admin.go @@ -487,6 +487,10 @@ func (s *Service) LogAPICall(log model.ApiCallLog) error { return err } } + policy, err := s.RuntimePolicy() + if err != nil { + return err + } s.storageMu.Lock() defer s.storageMu.Unlock() usage, err := s.repo.UserStorageUsage(log.UserID) @@ -494,7 +498,7 @@ func (s *Service) LogAPICall(log model.ApiCallLog) error { return err } incomingBytes := int64(len(log.Path) + len(log.Model) + len(log.ProviderRequestID) + len(log.ErrorCode) + len(log.Error) + len(log.UpstreamURL)) - if err := validateAPICallLogQuota(usage, incomingBytes); err != nil { + if err := validateAPICallLogQuotaWithPolicy(usage, incomingBytes, policy.Resource); err != nil { return err } return s.repo.Create(&log) @@ -537,7 +541,7 @@ func channelFromRequest(req ChannelRequest, channel model.ModelChannel) (model.M channel.ConcurrencyLimit = 0 } else if req.ConcurrencyLimit != nil { if *req.ConcurrencyLimit < minChannelConcurrencyLimit || *req.ConcurrencyLimit > maxChannelConcurrencyLimit { - return channel, BadAuthRequest("最大并发数必须是 1-100 的整数") + return channel, BadAuthRequest("最大并发数必须是 1-999 的整数") } channel.ConcurrencyLimit = *req.ConcurrencyLimit } else if req.UseGlobalConcurrency != nil { diff --git a/backend/internal/service/admin_audit.go b/backend/internal/service/admin_audit.go index d9be598ea..454075d0b 100644 --- a/backend/internal/service/admin_audit.go +++ b/backend/internal/service/admin_audit.go @@ -10,9 +10,13 @@ import ( ) type AdminUserDetail struct { - User model.User `json:"user"` - Account model.CreditAccount `json:"account"` - Counts repository.AdminUserCounts `json:"counts"` + User model.User `json:"user"` + Account model.CreditAccount `json:"account"` + Counts repository.AdminUserCounts `json:"counts"` + StorageUsage repository.UserStorageUsage `json:"storageUsage"` + StoredFileBytes int64 `json:"storedFileBytes"` + DailyUploadBytes int64 `json:"dailyUploadBytes"` + Quota RuntimeResourcePolicy `json:"quota"` } type AdminTaskPage struct { @@ -63,7 +67,26 @@ func (s *Service) AdminUserDetail(actor *model.User, userID string) (*AdminUserD if err != nil { return nil, err } - return &AdminUserDetail{User: *user, Account: *account, Counts: counts}, nil + usage, err := s.repo.UserStorageUsage(user.ID) + if err != nil { + return nil, err + } + storedFileBytes, err := s.repo.UserStoredFileBytes(user.ID) + if err != nil { + return nil, err + } + dailyUploadBytes, err := s.repo.DailyUploadBytes(user.ID, time.Now().UTC().Format("2006-01-02")) + if err != nil { + return nil, err + } + policy, err := s.RuntimePolicy() + if err != nil { + return nil, err + } + return &AdminUserDetail{ + User: *user, Account: *account, Counts: counts, StorageUsage: usage, + StoredFileBytes: storedFileBytes, DailyUploadBytes: dailyUploadBytes, Quota: policy.Resource, + }, nil } func (s *Service) AdminUserLedger(actor *model.User, userID string, entryType string, page int, limit int) (*WalletSummary, error) { diff --git a/backend/internal/service/admin_channel_test.go b/backend/internal/service/admin_channel_test.go index 1d1bd5a03..cd08810ec 100644 --- a/backend/internal/service/admin_channel_test.go +++ b/backend/internal/service/admin_channel_test.go @@ -76,7 +76,7 @@ func TestChannelFromRequestRejectsUnknownInterfaceType(t *testing.T) { } func TestChannelFromRequestRejectsInvalidConcurrencyLimit(t *testing.T) { - for _, limit := range []int{0, 101} { + for _, limit := range []int{0, 1000} { _, 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) @@ -87,7 +87,7 @@ func TestChannelFromRequestRejectsInvalidConcurrencyLimit(t *testing.T) { func TestRuntimeConcurrencyUsesEnvironmentFallback(t *testing.T) { t.Setenv("CANVAS_CHANNEL_CONCURRENCY", "7") t.Setenv("CANVAS_WORKER_CONCURRENCY", "9") - setting := runtimeConcurrencyFromEnvironment() + setting := defaultRuntimePolicy().Task if setting.ChannelConcurrency != 7 || setting.WorkerConcurrency != 9 { t.Fatalf("runtimeConcurrencyFromEnvironment() = %#v", setting) } diff --git a/backend/internal/service/concurrency_settings.go b/backend/internal/service/concurrency_settings.go deleted file mode 100644 index f805a7e57..000000000 --- a/backend/internal/service/concurrency_settings.go +++ /dev/null @@ -1,122 +0,0 @@ -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 index 0f70b61eb..038709df0 100644 --- a/backend/internal/service/concurrency_settings_test.go +++ b/backend/internal/service/concurrency_settings_test.go @@ -28,11 +28,14 @@ func TestRuntimeConcurrencySettingPersistsAndChannelOverrideWins(t *testing.T) { } 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}) + policy := defaultRuntimePolicy() + policy.Task.WorkerConcurrency = 8 + policy.Task.ChannelConcurrency = 6 + setting, err := svc.UpdateRuntimePolicySetting(actor, policy) if err != nil { t.Fatal(err) } - if setting.WorkerConcurrency != 8 || setting.ChannelConcurrency != 6 { + if setting.Task.WorkerConcurrency != 8 || setting.Task.ChannelConcurrency != 6 { t.Fatalf("setting = %#v", setting) } diff --git a/backend/internal/service/coordination.go b/backend/internal/service/coordination.go index 9e625275b..429faabef 100644 --- a/backend/internal/service/coordination.go +++ b/backend/internal/service/coordination.go @@ -20,7 +20,7 @@ type localRateEntry struct { const ( minChannelConcurrencyLimit = 1 - maxChannelConcurrencyLimit = 100 + maxChannelConcurrencyLimit = maxRuntimeConcurrency defaultChannelConcurrencyValue = 3 ) @@ -189,7 +189,7 @@ func (c *runtimeCoordinator) circuitOpen(ctx context.Context, channelID string) return count > 0, err } -func (c *runtimeCoordinator) recordChannelResult(ctx context.Context, channelID string, failed bool) { +func (c *runtimeCoordinator) recordChannelResult(ctx context.Context, channelID string, failed bool, failureLimit int, openDuration time.Duration) { if c.redis == nil || strings.TrimSpace(channelID) == "" { return } @@ -204,8 +204,8 @@ func (c *runtimeCoordinator) recordChannelResult(ctx context.Context, channelID return } _ = c.redis.Expire(ctx, failureKey, time.Minute).Err() - if count >= int64(envInt("CANVAS_CHANNEL_CIRCUIT_FAILURES", 5)) { - _ = c.redis.Set(ctx, openKey, "1", time.Duration(envInt("CANVAS_CHANNEL_CIRCUIT_SECONDS", 60))*time.Second).Err() + if count >= int64(failureLimit) { + _ = c.redis.Set(ctx, openKey, "1", openDuration).Err() } } @@ -243,7 +243,7 @@ func (s *Service) AcquireChannelSlot(ctx context.Context, channelID string, fall } if channel.ConcurrencyLimit > 0 { if channel.ConcurrencyLimit < minChannelConcurrencyLimit || channel.ConcurrencyLimit > maxChannelConcurrencyLimit { - return nil, limit, channelSlotError{scope: scope, limit: limit, err: errors.New("渠道并发配置超出 1-100 范围")} + return nil, limit, channelSlotError{scope: scope, limit: limit, err: errors.New("渠道并发配置超出 1-999 范围")} } limit = channel.ConcurrencyLimit } @@ -273,3 +273,22 @@ func (s *Service) AllowRequest(ctx context.Context, key string, limit int, windo } return s.coordinator.allow(ctx, key, limit, window) } + +func (s *Service) AcquireCustomRelaySlot(ctx context.Context, userID string, limit int, ttl time.Duration) (func(), bool, error) { + if s.coordinator == nil { + return nil, false, errors.New("运行时协调器未初始化") + } + return s.coordinator.acquire(ctx, "custom-relay:"+userID, limit, ttl) +} + +func (s *Service) RecordChannelResult(ctx context.Context, channelID string, failed bool) error { + policy, err := s.RuntimePolicy() + if err != nil { + return err + } + if s.coordinator == nil { + return errors.New("运行时协调器未初始化") + } + s.coordinator.recordChannelResult(ctx, channelID, failed, policy.Request.ChannelCircuitFailureCount, time.Duration(policy.Request.ChannelCircuitOpenSeconds)*time.Second) + return nil +} diff --git a/backend/internal/service/provider.go b/backend/internal/service/provider.go index a611e10fe..390998de2 100644 --- a/backend/internal/service/provider.go +++ b/backend/internal/service/provider.go @@ -243,12 +243,17 @@ func (s *Service) hydrateProviderMedia(userID string, media *providerMedia, requ return fmt.Errorf("读取任务参考资源失败：%w", err) } defer body.Close() - data, err := io.ReadAll(io.LimitReader(body, maxTaskResourceBytes+1)) + policy, err := s.RuntimePolicy() if err != nil { return err } - if int64(len(data)) > maxTaskResourceBytes { - return errors.New("任务参考资源超过 512MB") + resourceLimit := megabytes(policy.Resource.ResourceUploadMB) + data, err := io.ReadAll(io.LimitReader(body, resourceLimit+1)) + if err != nil { + return err + } + if int64(len(data)) > resourceLimit { + return fmt.Errorf("任务参考资源超过 %dMB", policy.Resource.ResourceUploadMB) } mimeType := normalizedMediaMimeType(firstNonEmpty(media.MimeType, resource.MimeType), data) media.DataURL = dataURL(mimeType, data) @@ -630,7 +635,7 @@ func runVideoTask(ctx context.Context, input canvasGenerationInput) (map[string] if id == "" { return nil, errors.New("视频接口没有返回任务 ID") } - for deadline := time.Now().Add(videoPollTimeout); time.Now().Before(deadline); { + for deadline := providerPollingDeadline(ctx); time.Now().Before(deadline); { var state map[string]interface{} if err := getJSON(ctx, input.Config, "/videos/"+id, &state); err != nil { return nil, err @@ -871,7 +876,7 @@ func runNewAPIChannel1VideoTask(ctx context.Context, input canvasGenerationInput if id == "" { return nil, errors.New("NewAPI 渠道 1 没有返回任务 ID") } - for deadline := time.Now().Add(videoPollTimeout); time.Now().Before(deadline); { + for deadline := providerPollingDeadline(ctx); time.Now().Before(deadline); { var state map[string]interface{} if err := getJSON(ctx, input.Config, "/videos/"+id, &state); err != nil { return nil, err @@ -1076,7 +1081,7 @@ func runSeedanceVideosTask(ctx context.Context, input canvasGenerationInput) (ma if id == "" { return nil, errors.New("Seedance 接口没有返回任务 ID") } - for deadline := time.Now().Add(videoPollTimeout); time.Now().Before(deadline); { + for deadline := providerPollingDeadline(ctx); time.Now().Before(deadline); { var state map[string]interface{} if err := getJSON(ctx, input.Config, "/videos/"+id, &state); err != nil { return nil, err @@ -1138,7 +1143,7 @@ func runSeedanceAgentPlanVideoTask(ctx context.Context, input canvasGenerationIn if id == "" { return nil, errors.New("Seedance 接口没有返回任务 ID") } - for deadline := time.Now().Add(videoPollTimeout); time.Now().Before(deadline); { + for deadline := providerPollingDeadline(ctx); time.Now().Before(deadline); { var state map[string]interface{} if err := getJSON(ctx, input.Config, "/contents/generations/tasks/"+id, &state); err != nil { return nil, err @@ -1259,12 +1264,26 @@ func doJSON(req *http.Request, target interface{}) error { func doBinary(req *http.Request) ([]byte, string, error) { startedAt := time.Now() + requestTimeout := providerHTTPTimeout + if deadline, ok := req.Context().Deadline(); ok { + if remaining := time.Until(deadline); remaining > 0 { + requestTimeout = remaining + } + } var release func() var coordinator *runtimeCoordinator + var runtimeService *Service + responseLimit := maxProviderResponseBytes channelID := "" if metadata, ok := req.Context().Value(providerAnalyticsKey{}).(providerAnalyticsContext); ok && metadata.Service != nil { + runtimeService = metadata.Service coordinator = metadata.Service.coordinator channelID = metadata.ChannelID + policy, err := metadata.Service.RuntimePolicy() + if err != nil { + return nil, "", fmt.Errorf("读取生成资源限制失败：%w", err) + } + responseLimit = megabytes(policy.Resource.GeneratedFileMB) open, err := coordinator.circuitOpen(req.Context(), channelID) if err != nil { return nil, "", fmt.Errorf("读取渠道熔断状态失败：%w", err) @@ -1277,7 +1296,7 @@ func doBinary(req *http.Request) ([]byte, string, error) { slotID = "custom:" + strings.ToLower(req.URL.Host) } var concurrencyLimit int - release, concurrencyLimit, err = metadata.Service.AcquireChannelSlot(req.Context(), channelID, slotID, providerHTTPTimeout+time.Minute) + release, concurrencyLimit, err = metadata.Service.AcquireChannelSlot(req.Context(), channelID, slotID, requestTimeout+time.Minute) metadata.ConcurrencyLimit = concurrencyLimit req = req.WithContext(context.WithValue(req.Context(), providerAnalyticsKey{}, metadata)) if err != nil { @@ -1290,47 +1309,54 @@ func doBinary(req *http.Request) ([]byte, string, error) { recordProviderRequest(req, startedAt, 0, nil, err) return nil, "", err } - client := OutboundHTTPClient(providerHTTPTimeout) + client := OutboundHTTPClient(requestTimeout) resp, err := client.Do(req) if err != nil { - if coordinator != nil { - coordinator.recordChannelResult(req.Context(), channelID, !errors.Is(err, context.Canceled)) + if runtimeService != nil { + _ = runtimeService.RecordChannelResult(req.Context(), channelID, !errors.Is(err, context.Canceled)) } recordProviderRequest(req, startedAt, 0, nil, err) return nil, "", err } defer resp.Body.Close() - if resp.ContentLength > maxProviderResponseBytes { - err = errors.New("上游响应超过 64MB 限制") + if resp.ContentLength > responseLimit { + err = fmt.Errorf("上游响应超过 %s 限制", formatStorageLimit(responseLimit)) recordProviderRequest(req, startedAt, resp.StatusCode, nil, err) return nil, "", err } - data, err := io.ReadAll(io.LimitReader(resp.Body, maxProviderResponseBytes+1)) + data, err := io.ReadAll(io.LimitReader(resp.Body, responseLimit+1)) if err != nil { recordProviderRequest(req, startedAt, resp.StatusCode, nil, err) return nil, "", err } - if int64(len(data)) > maxProviderResponseBytes { - err = errors.New("上游响应超过 64MB 限制") + if int64(len(data)) > responseLimit { + err = fmt.Errorf("上游响应超过 %s 限制", formatStorageLimit(responseLimit)) recordProviderRequest(req, startedAt, resp.StatusCode, nil, err) return nil, "", err } mimeType := resp.Header.Get("Content-Type") if resp.StatusCode < 200 || resp.StatusCode >= 300 { - if coordinator != nil { - coordinator.recordChannelResult(req.Context(), channelID, resp.StatusCode >= 500) + if runtimeService != nil { + _ = runtimeService.RecordChannelResult(req.Context(), channelID, resp.StatusCode >= 500) } httpErr := providerHTTPError{StatusCode: resp.StatusCode, Status: resp.Status, Body: string(data)} recordProviderRequest(req, startedAt, resp.StatusCode, data, httpErr) return nil, "", httpErr } recordProviderRequest(req, startedAt, resp.StatusCode, data, nil) - if coordinator != nil { - coordinator.recordChannelResult(req.Context(), channelID, false) + if runtimeService != nil { + _ = runtimeService.RecordChannelResult(req.Context(), channelID, false) } return data, mimeType, nil } +func providerPollingDeadline(ctx context.Context) time.Time { + if deadline, ok := ctx.Deadline(); ok { + return deadline + } + return time.Now().Add(videoPollTimeout) +} + func recordProviderRequest(req *http.Request, startedAt time.Time, statusCode int, responseBody []byte, requestErr error) { metadata, ok := req.Context().Value(providerAnalyticsKey{}).(providerAnalyticsContext) if !ok || metadata.Service == nil { diff --git a/backend/internal/service/resource.go b/backend/internal/service/resource.go index 7b4c70c5b..715cab6ff 100644 --- a/backend/internal/service/resource.go +++ b/backend/internal/service/resource.go @@ -27,8 +27,6 @@ import ( "infinite-canvas/backend/internal/model" ) -const maxImportedResourceBytes int64 = MaxResourceUploadBytes -const maxTaskResourceBytes int64 = 512 << 20 const providerResourceURLTTL = 4 * time.Hour const directResourceURLTTL = 5 * time.Minute @@ -125,7 +123,11 @@ func detectUploadedMimeType(file multipart.File, fileName string, declared strin } func (s *Service) ImportResourceURL(userID string, rawURL string, kind string, width int, height int, durationMs int64) (*model.Resource, error) { - payload, err := downloadRemoteResource(rawURL) + policy, err := s.RuntimePolicy() + if err != nil { + return nil, err + } + payload, err := downloadRemoteResource(rawURL, megabytes(policy.Resource.ResourceUploadMB)) if err != nil { return nil, err } @@ -294,7 +296,7 @@ func (s *Service) persistGeneratedMediaValueMode(userID string, value interface{ return item, nil case map[string]interface{}: if raw := inlineMediaValue(item); raw != "" { - mimeType, data, err := decodeDataURL(raw) + mimeType, data, err := s.decodeDataURL(raw) if err != nil && !skipInvalidDataURL { return nil, err } @@ -361,7 +363,7 @@ func inlineMediaValue(item map[string]interface{}) string { return "" } -func decodeDataURL(value string) (string, []byte, error) { +func (s *Service) decodeDataURL(value string) (string, []byte, error) { header, encoded, ok := strings.Cut(value, ",") if !ok || !strings.HasPrefix(header, "data:") || !strings.HasSuffix(strings.ToLower(header), ";base64") { return "", nil, fmt.Errorf("%w：格式无效", errInvalidGeneratedDataURL) @@ -371,8 +373,12 @@ func decodeDataURL(value string) (string, []byte, error) { if err != nil { return "", nil, fmt.Errorf("%w：base64 解码失败：%v", errInvalidGeneratedDataURL, err) } - if int64(len(data)) > maxTaskResourceBytes { - return "", nil, errors.New("单个生成资源超过 512MB") + policy, err := s.RuntimePolicy() + if err != nil { + return "", nil, err + } + if int64(len(data)) > megabytes(policy.Resource.GeneratedFileMB) { + return "", nil, fmt.Errorf("单个生成资源超过 %dMB", policy.Resource.GeneratedFileMB) } return mimeType, data, nil } @@ -398,7 +404,7 @@ type remoteResourcePayload struct { data []byte } -func downloadRemoteResource(rawURL string) (remoteResourcePayload, error) { +func downloadRemoteResource(rawURL string, maxBytes int64) (remoteResourcePayload, error) { parsed, err := validateRemoteResourceURL(rawURL) if err != nil { return remoteResourcePayload{}, err @@ -416,12 +422,12 @@ func downloadRemoteResource(rawURL string) (remoteResourcePayload, error) { if resp.StatusCode < 200 || resp.StatusCode >= 300 { return remoteResourcePayload{}, fmt.Errorf("远程资源下载失败：%s", resp.Status) } - data, err := io.ReadAll(io.LimitReader(resp.Body, maxImportedResourceBytes)) + data, err := io.ReadAll(io.LimitReader(resp.Body, maxBytes)) if err != nil { return remoteResourcePayload{}, err } - if int64(len(data)) >= maxImportedResourceBytes { - return remoteResourcePayload{}, BadAuthRequest("远程资源必须小于 50MB") + if int64(len(data)) >= maxBytes { + return remoteResourcePayload{}, BadAuthRequest(fmt.Sprintf("远程资源必须小于 %s", formatStorageLimit(maxBytes))) } mimeType := strings.TrimSpace(resp.Header.Get("Content-Type")) if idx := strings.Index(mimeType, ";"); idx >= 0 { diff --git a/backend/internal/service/resource_test.go b/backend/internal/service/resource_test.go index 3b44c7140..528c46c79 100644 --- a/backend/internal/service/resource_test.go +++ b/backend/internal/service/resource_test.go @@ -216,7 +216,7 @@ func TestPersistGeneratedMediaAppliesStoredFileQuota(t *testing.T) { ID: "existing", UserID: "user-1", Status: model.ResourceStatusReady, - Size: MaxUserStoredFileBytes - 1, + Size: gigabytes(defaultRuntimePolicy().Resource.StoredFileGB) - 1, }); err != nil { t.Fatal(err) } diff --git a/backend/internal/service/runtime_policy.go b/backend/internal/service/runtime_policy.go new file mode 100644 index 000000000..1230eecce --- /dev/null +++ b/backend/internal/service/runtime_policy.go @@ -0,0 +1,380 @@ +package service + +import ( + "encoding/json" + "errors" + "fmt" + "strings" + "time" + + "infinite-canvas/backend/internal/model" + + "gorm.io/gorm" +) + +const runtimePolicySettingKey = "runtime_policy" + +const ( + maxRuntimeUploadMB int64 = 999 + maxRuntimeStorageGB int64 = 999 + maxRuntimeDataMB int64 = 999_999 + maxRuntimeCount int64 = 999_999_999 + maxRuntimeRate = 999_999 + maxRuntimeConcurrency = 999 + maxRuntimeTimeoutMinutes = 9_999 +) + +type RuntimeResourcePolicy struct { + ResourceUploadMB int64 `json:"resourceUploadMB"` + SessionUploadMB int64 `json:"sessionUploadMB"` + GeneratedFileMB int64 `json:"generatedFileMB"` + DailyUploadMB int64 `json:"dailyUploadMB"` + StoredFileGB int64 `json:"storedFileGB"` + StructuredDataMB int64 `json:"structuredDataMB"` + TaskDataGB int64 `json:"taskDataGB"` + AssetCount int64 `json:"assetCount"` + CanvasCount int64 `json:"canvasCount"` + SessionCount int64 `json:"sessionCount"` + TaskCount int64 `json:"taskCount"` + APICallLogCount int64 `json:"apiCallLogCount"` +} + +type RuntimeTaskPolicy struct { + WorkerConcurrency int `json:"workerConcurrency"` + ChannelConcurrency int `json:"channelConcurrency"` + ActiveTaskLimit int `json:"activeTaskLimit"` + ImageTimeoutMinutes int `json:"imageTimeoutMinutes"` + TextTimeoutMinutes int `json:"textTimeoutMinutes"` + AudioTimeoutMinutes int `json:"audioTimeoutMinutes"` + VideoTimeoutMinutes int `json:"videoTimeoutMinutes"` + StoryboardTimeoutMinutes int `json:"storyboardTimeoutMinutes"` + DefaultTimeoutMinutes int `json:"defaultTimeoutMinutes"` +} + +type RuntimeRequestPolicy struct { + TaskCreatePerMinute int `json:"taskCreatePerMinute"` + SessionCreatePerMinute int `json:"sessionCreatePerMinute"` + ResourceUploadPerMinute int `json:"resourceUploadPerMinute"` + ResourceImportPerMinute int `json:"resourceImportPerMinute"` + SessionFilePerMinute int `json:"sessionFilePerMinute"` + AssetWritePerMinute int `json:"assetWritePerMinute"` + CanvasWritePerMinute int `json:"canvasWritePerMinute"` + RegisterPerHour int `json:"registerPerHour"` + EmailCodePerHour int `json:"emailCodePerHour"` + LoginIPPerTenMinutes int `json:"loginIPPerTenMinutes"` + LoginAccountPerTenMinutes int `json:"loginAccountPerTenMinutes"` + SystemRelayPerMinute int `json:"systemRelayPerMinute"` + CustomRelayPerMinute int `json:"customRelayPerMinute"` + CustomRelayConcurrency int `json:"customRelayConcurrency"` + CustomRelayRequestMB int64 `json:"customRelayRequestMB"` + CustomRelayResponseMB int64 `json:"customRelayResponseMB"` + CustomRelayTimeoutMinutes int `json:"customRelayTimeoutMinutes"` + SystemRelayRequestMB int64 `json:"systemRelayRequestMB"` + SystemRelayResponseMB int64 `json:"systemRelayResponseMB"` + ChannelCircuitFailureCount int `json:"channelCircuitFailureCount"` + ChannelCircuitOpenSeconds int `json:"channelCircuitOpenSeconds"` +} + +type RuntimePolicySetting struct { + Resource RuntimeResourcePolicy `json:"resource"` + Task RuntimeTaskPolicy `json:"task"` + Request RuntimeRequestPolicy `json:"request"` +} + +type PublicRuntimePolicySetting struct { + RuntimePolicySetting + Configured bool `json:"configured"` + UpdatedBy string `json:"updatedBy"` + CreatedAt time.Time `json:"createdAt"` + UpdatedAt time.Time `json:"updatedAt"` +} + +type PublicRuntimeLimits struct { + ActiveTaskLimit int `json:"activeTaskLimit"` + ResourceUploadMB int64 `json:"resourceUploadMB"` + SessionUploadMB int64 `json:"sessionUploadMB"` +} + +func defaultRuntimePolicy() RuntimePolicySetting { + return RuntimePolicySetting{ + Resource: RuntimeResourcePolicy{ + ResourceUploadMB: 50, + SessionUploadMB: 32, + GeneratedFileMB: 64, + DailyUploadMB: 200, + StoredFileGB: 2, + StructuredDataMB: 256, + TaskDataGB: 1, + AssetCount: 2_000, + CanvasCount: 1_000, + SessionCount: 1_000, + TaskCount: 20_000, + APICallLogCount: 100_000, + }, + Task: RuntimeTaskPolicy{ + WorkerConcurrency: effectiveChannelConcurrencyLimit(envInt("CANVAS_WORKER_CONCURRENCY", taskWorkerConcurrency)), + ChannelConcurrency: defaultChannelConcurrencyLimit(), + ActiveTaskLimit: 5, + ImageTimeoutMinutes: 8, + TextTimeoutMinutes: 8, + AudioTimeoutMinutes: 8, + VideoTimeoutMinutes: 30, + StoryboardTimeoutMinutes: 12, + DefaultTimeoutMinutes: 10, + }, + Request: RuntimeRequestPolicy{ + TaskCreatePerMinute: 30, + SessionCreatePerMinute: 20, + ResourceUploadPerMinute: 30, + ResourceImportPerMinute: 30, + SessionFilePerMinute: 30, + AssetWritePerMinute: 120, + CanvasWritePerMinute: 120, + RegisterPerHour: 5, + EmailCodePerHour: 10, + LoginIPPerTenMinutes: 50, + LoginAccountPerTenMinutes: 10, + SystemRelayPerMinute: 120, + CustomRelayPerMinute: 120, + CustomRelayConcurrency: 4, + CustomRelayRequestMB: 32, + CustomRelayResponseMB: 32, + CustomRelayTimeoutMinutes: 10, + SystemRelayRequestMB: 64, + SystemRelayResponseMB: 128, + ChannelCircuitFailureCount: min(envInt("CANVAS_CHANNEL_CIRCUIT_FAILURES", 5), maxRuntimeConcurrency), + ChannelCircuitOpenSeconds: min(envInt("CANVAS_CHANNEL_CIRCUIT_SECONDS", 60), 86_400), + }, + } +} + +func selfUseRuntimePolicy() RuntimePolicySetting { + value := defaultRuntimePolicy() + value.Resource = RuntimeResourcePolicy{ + ResourceUploadMB: maxRuntimeUploadMB, SessionUploadMB: maxRuntimeUploadMB, GeneratedFileMB: maxRuntimeUploadMB, + DailyUploadMB: maxRuntimeDataMB, StoredFileGB: maxRuntimeStorageGB, StructuredDataMB: maxRuntimeDataMB, + TaskDataGB: maxRuntimeStorageGB, AssetCount: maxRuntimeCount, CanvasCount: maxRuntimeCount, + SessionCount: maxRuntimeCount, TaskCount: maxRuntimeCount, APICallLogCount: maxRuntimeCount, + } + value.Task = RuntimeTaskPolicy{ + WorkerConcurrency: maxRuntimeConcurrency, ChannelConcurrency: maxRuntimeConcurrency, ActiveTaskLimit: maxRuntimeConcurrency, + ImageTimeoutMinutes: maxRuntimeTimeoutMinutes, TextTimeoutMinutes: maxRuntimeTimeoutMinutes, + AudioTimeoutMinutes: maxRuntimeTimeoutMinutes, VideoTimeoutMinutes: maxRuntimeTimeoutMinutes, + StoryboardTimeoutMinutes: maxRuntimeTimeoutMinutes, DefaultTimeoutMinutes: maxRuntimeTimeoutMinutes, + } + value.Request = RuntimeRequestPolicy{ + TaskCreatePerMinute: maxRuntimeRate, SessionCreatePerMinute: maxRuntimeRate, + ResourceUploadPerMinute: maxRuntimeRate, ResourceImportPerMinute: maxRuntimeRate, + SessionFilePerMinute: maxRuntimeRate, AssetWritePerMinute: maxRuntimeRate, CanvasWritePerMinute: maxRuntimeRate, + RegisterPerHour: maxRuntimeRate, EmailCodePerHour: maxRuntimeRate, + LoginIPPerTenMinutes: maxRuntimeRate, LoginAccountPerTenMinutes: maxRuntimeRate, + SystemRelayPerMinute: maxRuntimeRate, CustomRelayPerMinute: maxRuntimeRate, + CustomRelayConcurrency: maxRuntimeConcurrency, CustomRelayRequestMB: maxRuntimeUploadMB, + CustomRelayResponseMB: maxRuntimeUploadMB, CustomRelayTimeoutMinutes: maxRuntimeTimeoutMinutes, + SystemRelayRequestMB: maxRuntimeUploadMB, SystemRelayResponseMB: maxRuntimeUploadMB, + ChannelCircuitFailureCount: maxRuntimeConcurrency, ChannelCircuitOpenSeconds: 1, + } + return value +} + +func (s *Service) RuntimePolicy() (RuntimePolicySetting, error) { + _, value, err := s.readRuntimePolicy() + return value, err +} + +func (s *Service) runtimeConcurrencySetting() (RuntimeTaskPolicy, error) { + policy, err := s.RuntimePolicy() + return policy.Task, err +} + +func (s *Service) PublicRuntimeLimits() (*PublicRuntimeLimits, error) { + policy, err := s.RuntimePolicy() + if err != nil { + return nil, err + } + return &PublicRuntimeLimits{ + ActiveTaskLimit: policy.Task.ActiveTaskLimit, ResourceUploadMB: policy.Resource.ResourceUploadMB, + SessionUploadMB: policy.Resource.SessionUploadMB, + }, nil +} + +func (s *Service) AdminRuntimePolicySetting(actor *model.User) (*PublicRuntimePolicySetting, error) { + if err := s.RequireAdmin(actor); err != nil { + return nil, err + } + setting, value, err := s.readRuntimePolicy() + if err != nil { + return nil, err + } + return publicRuntimePolicy(setting, value), nil +} + +func (s *Service) AdminSelfUseRuntimePolicy(actor *model.User) (*PublicRuntimePolicySetting, error) { + if err := s.RequireAdmin(actor); err != nil { + return nil, err + } + return publicRuntimePolicy(nil, selfUseRuntimePolicy()), nil +} + +func (s *Service) UpdateRuntimePolicySetting(actor *model.User, value RuntimePolicySetting) (*PublicRuntimePolicySetting, error) { + if err := s.RequireAdmin(actor); err != nil { + return nil, err + } + if err := validateRuntimePolicy(value); err != nil { + return nil, err + } + current, before, err := s.readRuntimePolicy() + if err != nil { + return nil, err + } + encoded, err := json.Marshal(value) + if err != nil { + return nil, err + } + setting := model.SystemSetting{Key: runtimePolicySettingKey, 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_policy.update", "system_setting", runtimePolicySettingKey, "更新资源与请求策略", map[string]any{"before": before, "after": value}); err != nil { + return nil, err + } + return publicRuntimePolicy(&setting, value), nil +} + +func (s *Service) ResetRuntimePolicySetting(actor *model.User) (*PublicRuntimePolicySetting, error) { + if err := s.RequireAdmin(actor); err != nil { + return nil, err + } + _, before, err := s.readRuntimePolicy() + if err != nil { + return nil, err + } + if err := s.repo.DeleteSystemSetting(runtimePolicySettingKey); err != nil { + return nil, err + } + after := defaultRuntimePolicy() + if err := s.appendAdminAudit(actor, "runtime_policy.reset", "system_setting", runtimePolicySettingKey, "重置资源与请求策略", map[string]any{"before": before, "after": after}); err != nil { + return nil, err + } + return publicRuntimePolicy(nil, after), nil +} + +func (s *Service) readRuntimePolicy() (*model.SystemSetting, RuntimePolicySetting, error) { + setting, err := s.repo.SystemSetting(runtimePolicySettingKey) + if errors.Is(err, gorm.ErrRecordNotFound) { + value := defaultRuntimePolicy() + return nil, value, validateRuntimePolicy(value) + } + if err != nil { + return nil, RuntimePolicySetting{}, err + } + value := RuntimePolicySetting{} + if strings.TrimSpace(setting.ValueJSON) == "" || json.Unmarshal([]byte(setting.ValueJSON), &value) != nil { + return nil, RuntimePolicySetting{}, errors.New("资源与请求策略配置格式无效") + } + if err := validateRuntimePolicy(value); err != nil { + return nil, RuntimePolicySetting{}, err + } + return setting, value, nil +} + +func validateRuntimePolicy(value RuntimePolicySetting) error { + resource := value.Resource + for label, item := range map[string]int64{ + "普通资源单文件": resource.ResourceUploadMB, "Agent 会话附件": resource.SessionUploadMB, + "单个生成资源": resource.GeneratedFileMB, + } { + if item < 1 || item > maxRuntimeUploadMB { + return BadAuthRequest(fmt.Sprintf("%s必须是 1-%d MB 的整数", label, maxRuntimeUploadMB)) + } + } + if resource.DailyUploadMB < 1 || resource.DailyUploadMB > maxRuntimeDataMB { + return BadAuthRequest(fmt.Sprintf("每日上传量必须是 1-%d MB 的整数", maxRuntimeDataMB)) + } + if resource.StoredFileGB < 1 || resource.StoredFileGB > maxRuntimeStorageGB || resource.TaskDataGB < 1 || resource.TaskDataGB > maxRuntimeStorageGB { + return BadAuthRequest(fmt.Sprintf("账号文件与任务数据容量必须是 1-%d GB 的整数", maxRuntimeStorageGB)) + } + if resource.StructuredDataMB < 1 || resource.StructuredDataMB > maxRuntimeDataMB { + return BadAuthRequest(fmt.Sprintf("结构化数据容量必须是 1-%d MB 的整数", maxRuntimeDataMB)) + } + storedMB := resource.StoredFileGB * 1024 + if resource.ResourceUploadMB > storedMB || resource.SessionUploadMB > storedMB || resource.GeneratedFileMB > storedMB { + return BadAuthRequest("单文件上限不能大于账号文件总容量") + } + for label, item := range map[string]int64{ + "素材数量": resource.AssetCount, "画布数量": resource.CanvasCount, "Agent 会话数量": resource.SessionCount, + "任务历史数量": resource.TaskCount, "请求日志数量": resource.APICallLogCount, + } { + if item < 1 || item > maxRuntimeCount { + return BadAuthRequest(fmt.Sprintf("%s必须是 1-%d 的整数", label, maxRuntimeCount)) + } + } + task := value.Task + for label, item := range map[string]int{ + "Worker 并发数": task.WorkerConcurrency, "全局渠道并发数": task.ChannelConcurrency, "活动任务上限": task.ActiveTaskLimit, + } { + if item < 1 || item > maxRuntimeConcurrency { + return BadAuthRequest(fmt.Sprintf("%s必须是 1-%d 的整数", label, maxRuntimeConcurrency)) + } + } + for label, item := range map[string]int{ + "图片任务超时": task.ImageTimeoutMinutes, "文本任务超时": task.TextTimeoutMinutes, + "音频任务超时": task.AudioTimeoutMinutes, "视频任务超时": task.VideoTimeoutMinutes, + "分镜任务超时": task.StoryboardTimeoutMinutes, "默认任务超时": task.DefaultTimeoutMinutes, + } { + if item < 1 || item > maxRuntimeTimeoutMinutes { + return BadAuthRequest(fmt.Sprintf("%s必须是 1-%d 分钟的整数", label, maxRuntimeTimeoutMinutes)) + } + } + request := value.Request + for label, item := range map[string]int{ + "任务创建频控": request.TaskCreatePerMinute, "会话创建频控": request.SessionCreatePerMinute, + "资源上传频控": request.ResourceUploadPerMinute, "资源导入频控": request.ResourceImportPerMinute, + "会话附件频控": request.SessionFilePerMinute, "素材写入频控": request.AssetWritePerMinute, + "画布写入频控": request.CanvasWritePerMinute, "注册频控": request.RegisterPerHour, + "验证码频控": request.EmailCodePerHour, "登录 IP 频控": request.LoginIPPerTenMinutes, + "登录账号频控": request.LoginAccountPerTenMinutes, "系统渠道频控": request.SystemRelayPerMinute, + "自定义渠道频控": request.CustomRelayPerMinute, + } { + if item < 1 || item > maxRuntimeRate { + return BadAuthRequest(fmt.Sprintf("%s必须是 1-%d 的整数", label, maxRuntimeRate)) + } + } + if request.CustomRelayConcurrency < 1 || request.CustomRelayConcurrency > maxRuntimeConcurrency { + return BadAuthRequest(fmt.Sprintf("自定义渠道并发必须是 1-%d 的整数", maxRuntimeConcurrency)) + } + for label, item := range map[string]int64{ + "自定义渠道请求体": request.CustomRelayRequestMB, "自定义渠道响应体": request.CustomRelayResponseMB, + "系统渠道请求体": request.SystemRelayRequestMB, "系统渠道响应体": request.SystemRelayResponseMB, + } { + if item < 1 || item > maxRuntimeUploadMB { + return BadAuthRequest(fmt.Sprintf("%s必须是 1-%d MB 的整数", label, maxRuntimeUploadMB)) + } + } + if request.CustomRelayTimeoutMinutes < 1 || request.CustomRelayTimeoutMinutes > maxRuntimeTimeoutMinutes { + return BadAuthRequest(fmt.Sprintf("自定义渠道超时必须是 1-%d 分钟的整数", maxRuntimeTimeoutMinutes)) + } + if request.ChannelCircuitFailureCount < 1 || request.ChannelCircuitFailureCount > maxRuntimeConcurrency { + return BadAuthRequest(fmt.Sprintf("渠道熔断失败次数必须是 1-%d 的整数", maxRuntimeConcurrency)) + } + if request.ChannelCircuitOpenSeconds < 1 || request.ChannelCircuitOpenSeconds > 86_400 { + return BadAuthRequest("渠道熔断时长必须是 1-86400 秒的整数") + } + return nil +} + +func publicRuntimePolicy(setting *model.SystemSetting, value RuntimePolicySetting) *PublicRuntimePolicySetting { + result := &PublicRuntimePolicySetting{RuntimePolicySetting: value, Configured: setting != nil} + if setting != nil { + result.UpdatedBy = setting.UpdatedBy + result.CreatedAt = setting.CreatedAt + result.UpdatedAt = setting.UpdatedAt + } + return result +} + +func megabytes(value int64) int64 { return value << 20 } +func gigabytes(value int64) int64 { return value << 30 } diff --git a/backend/internal/service/runtime_policy_test.go b/backend/internal/service/runtime_policy_test.go new file mode 100644 index 000000000..ae34d3f0b --- /dev/null +++ b/backend/internal/service/runtime_policy_test.go @@ -0,0 +1,65 @@ +package service + +import ( + "testing" + + "infinite-canvas/backend/internal/model" + "infinite-canvas/backend/internal/repository" + + "gorm.io/driver/sqlite" + "gorm.io/gorm" +) + +func TestRuntimePolicyDefaultsAndSelfUseModeValidate(t *testing.T) { + if err := validateRuntimePolicy(defaultRuntimePolicy()); err != nil { + t.Fatalf("default runtime policy error = %v", err) + } + selfUse := selfUseRuntimePolicy() + if err := validateRuntimePolicy(selfUse); err != nil { + t.Fatalf("self-use runtime policy error = %v", err) + } + if selfUse.Task.WorkerConcurrency != 999 || selfUse.Resource.ResourceUploadMB != 999 { + t.Fatalf("self-use maxima = worker %d, upload %d", selfUse.Task.WorkerConcurrency, selfUse.Resource.ResourceUploadMB) + } +} + +func TestRuntimePolicyRejectsSingleFileAboveAccountCapacity(t *testing.T) { + policy := defaultRuntimePolicy() + policy.Resource.StoredFileGB = 1 + policy.Resource.ResourceUploadMB = 999 + if err := validateRuntimePolicy(policy); err != nil { + t.Fatalf("999MB should fit in 1GB: %v", err) + } + policy.Resource.StoredFileGB = 0 + if err := validateRuntimePolicy(policy); err == nil { + t.Fatal("zero account capacity should be rejected") + } +} + +func TestRuntimePolicySaveAndResetTakeEffectImmediately(t *testing.T) { + db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{}) + if err != nil { + t.Fatal(err) + } + if err := db.AutoMigrate(&model.SystemSetting{}, &model.AdminAuditEvent{}); err != nil { + t.Fatal(err) + } + svc := New(repository.New(db), t.TempDir()) + actor := &model.User{ID: "admin", Role: model.UserRoleAdmin} + policy := defaultRuntimePolicy() + policy.Task.ActiveTaskLimit = 17 + if _, err := svc.UpdateRuntimePolicySetting(actor, policy); err != nil { + t.Fatal(err) + } + effective, err := svc.RuntimePolicy() + if err != nil || effective.Task.ActiveTaskLimit != 17 { + t.Fatalf("effective active task limit = %d, error = %v", effective.Task.ActiveTaskLimit, err) + } + if _, err := svc.ResetRuntimePolicySetting(actor); err != nil { + t.Fatal(err) + } + effective, err = svc.RuntimePolicy() + if err != nil || effective.Task.ActiveTaskLimit != 5 { + t.Fatalf("reset active task limit = %d, error = %v", effective.Task.ActiveTaskLimit, err) + } +} diff --git a/backend/internal/service/service.go b/backend/internal/service/service.go index f38d5f577..be76161d6 100644 --- a/backend/internal/service/service.go +++ b/backend/internal/service/service.go @@ -37,15 +37,6 @@ type Service struct { const taskWorkerConcurrency = 3 const taskLogPayloadLimit = 4000 -const MaxSessionUploadBytes int64 = 32 << 20 -const ( - imageTaskTimeout = 8 * time.Minute - textTaskTimeout = 8 * time.Minute - audioTaskTimeout = 8 * time.Minute - videoTaskTimeout = 30 * time.Minute - storyboardTaskTimeout = 12 * time.Minute - defaultTaskTimeout = 10 * time.Minute -) type CreateSessionRequest struct { ProjectID string `json:"projectId"` @@ -202,6 +193,10 @@ func (s *Service) CreateSession(userID string, req CreateSessionRequest) (*Sessi compactedSnapshot := compactPersistedValue(req.CanvasSnapshot) snapshotJSON, _ := json.Marshal(compactedSnapshot) session := model.Session{ID: newID(), UserID: userID, ProjectID: req.ProjectID, Status: model.SessionStatusActive, Prompt: prompt, CanvasSnapshotJSON: string(snapshotJSON)} + policy, err := s.RuntimePolicy() + if err != nil { + return nil, err + } s.storageMu.Lock() usage, err := s.repo.UserStorageUsage(userID) if err != nil { @@ -209,7 +204,7 @@ func (s *Service) CreateSession(userID string, req CreateSessionRequest) (*Sessi return nil, err } incomingBytes := int64(len([]byte(prompt))*2 + len(snapshotJSON)) - if err := validateStructuredStorageQuota(usage, "session", true, incomingBytes); err != nil { + if err := validateStructuredStorageQuotaWithPolicy(usage, "session", true, incomingBytes, policy.Resource); err != nil { s.storageMu.Unlock() return nil, err } @@ -279,12 +274,16 @@ func (s *Service) CreateTask(userID string, req CreateTaskRequest) (*model.Task, if containsInlineMediaDataURL(normalizedInput) { return nil, BadAuthRequest("任务输入不能包含内嵌媒体，请先上传到资源存储") } + policy, err := s.RuntimePolicy() + if err != nil { + return nil, err + } activeTasks, err := s.repo.ActiveTaskCountForUser(userID) if err != nil { return nil, err } - if activeTasks >= 5 { - return nil, BadAuthRequest("同时排队或运行的任务最多 5 个，请等待已有任务完成") + if activeTasks >= int64(policy.Task.ActiveTaskLimit) { + return nil, BadAuthRequest(fmt.Sprintf("同时排队或运行的任务最多 %d 个，请等待已有任务完成", policy.Task.ActiveTaskLimit)) } taskType := req.Type if taskType == "" { @@ -303,9 +302,9 @@ func (s *Service) CreateTask(userID string, req CreateTaskRequest) (*model.Task, if billingOrder != nil { task.BillingOrderID = billingOrder.ID } - err = s.createTaskWithinStorageQuota(&task, billingOrder) + err = s.createTaskWithinStorageQuota(&task, billingOrder, policy) if errors.Is(err, repository.ErrActiveTaskLimit) { - return nil, BadAuthRequest("同时排队或运行的任务最多 5 个，请等待已有任务完成") + return nil, BadAuthRequest(fmt.Sprintf("同时排队或运行的任务最多 %d 个，请等待已有任务完成", policy.Task.ActiveTaskLimit)) } if errors.Is(err, repository.ErrInsufficientCredits) { return nil, BadAuthRequest("积分不足，请先使用兑换码充值") @@ -407,12 +406,16 @@ func (s *Service) RetryTask(userID string, id string) (*model.Task, error) { if err != nil { return nil, err } - task, err = s.repo.RetryTaskWithBilling(userID, task.ID, billingOrder) + policy, err := s.RuntimePolicy() + if err != nil { + return nil, err + } + task, err = s.repo.RetryTaskWithBilling(userID, task.ID, billingOrder, policy.Task.ActiveTaskLimit) if errors.Is(err, repository.ErrInsufficientCredits) { return nil, BadAuthRequest("积分不足，请先使用兑换码充值") } if errors.Is(err, repository.ErrActiveTaskLimit) { - return nil, BadAuthRequest("同时排队或运行的任务最多 5 个，请等待已有任务完成") + return nil, BadAuthRequest(fmt.Sprintf("同时排队或运行的任务最多 %d 个，请等待已有任务完成", policy.Task.ActiveTaskLimit)) } if errors.Is(err, repository.ErrTaskNotRetryable) { return nil, BadAuthRequest("任务已被其他请求重新入队，请勿重复重试") @@ -592,10 +595,15 @@ func publicTaskInputJSON(raw string) string { } func (s *Service) StoreUpload(userID string, sessionID string, header *multipart.FileHeader) (*model.SessionFile, error) { - if header == nil || header.Size > MaxSessionUploadBytes { - return nil, BadAuthRequest("会话文件不能超过 32MB") + policy, err := s.RuntimePolicy() + if err != nil { + return nil, err } - day, err := s.reserveUserUploadQuota(userID, header.Size) + maxBytes := megabytes(policy.Resource.SessionUploadMB) + if header == nil || header.Size > maxBytes { + return nil, BadAuthRequest(fmt.Sprintf("会话文件不能超过 %dMB", policy.Resource.SessionUploadMB)) + } + day, err := s.reserveSessionUploadQuota(userID, header.Size) if err != nil { return nil, err } @@ -625,7 +633,7 @@ func (s *Service) StoreUpload(userID string, sessionID string, header *multipart if err != nil { return nil, err } - size, err := io.Copy(dst, io.LimitReader(file, MaxSessionUploadBytes+1)) + size, err := io.Copy(dst, io.LimitReader(file, maxBytes+1)) closeErr := dst.Close() if err != nil { _ = os.Remove(path) @@ -635,9 +643,9 @@ func (s *Service) StoreUpload(userID string, sessionID string, header *multipart _ = os.Remove(path) return nil, closeErr } - if size > MaxSessionUploadBytes { + if size > maxBytes { _ = os.Remove(path) - return nil, BadAuthRequest("会话文件不能超过 32MB") + return nil, BadAuthRequest(fmt.Sprintf("会话文件不能超过 %dMB", policy.Resource.SessionUploadMB)) } item := model.SessionFile{ID: newID(), UserID: userID, SessionID: sessionID, FileName: header.Filename, MimeType: header.Header.Get("Content-Type"), Path: path, Size: size} if err := s.repo.Create(&item); err != nil { @@ -659,7 +667,11 @@ func (s *Service) ProcessNextTask() error { func (s *Service) processClaimedTask(task *model.Task) error { _ = s.log(task.UserID, task.ID, "info", "后端任务开始处理", "") - ctx, cancel := context.WithTimeout(context.Background(), taskExecutionTimeout(task.Type)) + policy, err := s.RuntimePolicy() + if err != nil { + return err + } + ctx, cancel := context.WithTimeout(context.Background(), taskExecutionTimeoutWithPolicy(task.Type, policy.Task)) defer cancel() leaseDone := make(chan struct{}) leaseLost := make(chan error, 1) @@ -784,20 +796,20 @@ func taskFailureMessage(err error) string { return truncateRunes(err.Error(), 2_000) } -func taskExecutionTimeout(taskType string) time.Duration { +func taskExecutionTimeoutWithPolicy(taskType string, policy RuntimeTaskPolicy) time.Duration { switch { case taskType == "agent_storyboard" || taskType == "agent_storyboard_rows": - return storyboardTaskTimeout + return time.Duration(policy.StoryboardTimeoutMinutes) * time.Minute case strings.HasPrefix(taskType, "canvas_video") || strings.HasPrefix(taskType, "video_"): - return videoTaskTimeout + return time.Duration(policy.VideoTimeoutMinutes) * time.Minute case strings.HasPrefix(taskType, "canvas_image"): - return imageTaskTimeout + return time.Duration(policy.ImageTimeoutMinutes) * time.Minute case strings.HasPrefix(taskType, "canvas_audio"): - return audioTaskTimeout + return time.Duration(policy.AudioTimeoutMinutes) * time.Minute case strings.HasPrefix(taskType, "canvas_text"): - return textTaskTimeout + return time.Duration(policy.TextTimeoutMinutes) * time.Minute default: - return defaultTaskTimeout + return time.Duration(policy.DefaultTimeoutMinutes) * time.Minute } } diff --git a/backend/internal/service/storage_quota.go b/backend/internal/service/storage_quota.go index d407cc129..19f6a33d7 100644 --- a/backend/internal/service/storage_quota.go +++ b/backend/internal/service/storage_quota.go @@ -1,89 +1,80 @@ package service import ( + "fmt" "time" "infinite-canvas/backend/internal/model" "infinite-canvas/backend/internal/repository" ) -const ( - MaxUserStructuredDataBytes int64 = 256 << 20 - MaxUserTaskDataBytes int64 = 1 << 30 - MaxUserAssetCount = 2_000 - MaxUserCanvasCount = 1_000 - MaxUserSessionCount = 1_000 - MaxUserTaskCount = 20_000 - MaxUserAPICallLogCount = 100_000 -) - func structuredBytes(usage repository.UserStorageUsage) int64 { return usage.AssetBytes + usage.CanvasBytes + usage.SessionBytes } -func validateStructuredStorageQuota(usage repository.UserStorageUsage, kind string, creating bool, deltaBytes int64) error { - if structuredBytes(usage)+deltaBytes > MaxUserStructuredDataBytes { - return BadAuthRequest("账号画布、素材和会话数据已达到 256MB 上限，请先删除不需要的内容") +func validateStructuredStorageQuotaWithPolicy(usage repository.UserStorageUsage, kind string, creating bool, deltaBytes int64, policy RuntimeResourcePolicy) error { + if structuredBytes(usage)+deltaBytes > megabytes(policy.StructuredDataMB) { + return BadAuthRequest(fmt.Sprintf("账号画布、素材和会话数据已达到 %dMB 上限，请先删除不需要的内容", policy.StructuredDataMB)) } if !creating { return nil } switch kind { case "asset": - if usage.AssetCount >= MaxUserAssetCount { - return BadAuthRequest("账号素材数量已达到 2000 个上限") + if usage.AssetCount >= policy.AssetCount { + return BadAuthRequest(fmt.Sprintf("账号素材数量已达到 %d 个上限", policy.AssetCount)) } case "canvas": - if usage.CanvasCount >= MaxUserCanvasCount { - return BadAuthRequest("账号画布数量已达到 1000 个上限") + if usage.CanvasCount >= policy.CanvasCount { + return BadAuthRequest(fmt.Sprintf("账号画布数量已达到 %d 个上限", policy.CanvasCount)) } case "session": - if usage.SessionCount >= MaxUserSessionCount { - return BadAuthRequest("账号 Agent 会话数量已达到 1000 个上限") + if usage.SessionCount >= policy.SessionCount { + return BadAuthRequest(fmt.Sprintf("账号 Agent 会话数量已达到 %d 个上限", policy.SessionCount)) } } return nil } -func validateTaskStorageQuota(usage repository.UserStorageUsage, incomingBytes int64) error { - if usage.TaskCount >= MaxUserTaskCount { - return BadAuthRequest("账号任务历史已达到 20000 条上限，请联系管理员归档") +func validateTaskStorageQuotaWithPolicy(usage repository.UserStorageUsage, incomingBytes int64, policy RuntimeResourcePolicy) error { + if usage.TaskCount >= policy.TaskCount { + return BadAuthRequest(fmt.Sprintf("账号任务历史已达到 %d 条上限，请联系管理员归档", policy.TaskCount)) } - return validateTaskDataGrowthQuota(usage, incomingBytes) + return validateTaskDataGrowthQuotaWithPolicy(usage, incomingBytes, policy) } -func validateTaskDataGrowthQuota(usage repository.UserStorageUsage, incomingBytes int64) error { - if usage.TaskBytes+incomingBytes > MaxUserTaskDataBytes { - return BadAuthRequest("账号任务历史数据已达到 1GB 上限，请联系管理员归档") +func validateTaskDataGrowthQuotaWithPolicy(usage repository.UserStorageUsage, incomingBytes int64, policy RuntimeResourcePolicy) error { + if usage.TaskBytes+incomingBytes > gigabytes(policy.TaskDataGB) { + return BadAuthRequest(fmt.Sprintf("账号任务历史数据已达到 %dGB 上限，请联系管理员归档", policy.TaskDataGB)) } return nil } -func validateAPICallLogQuota(usage repository.UserStorageUsage, incomingBytes int64) error { - if usage.APICallCount >= MaxUserAPICallLogCount { - return BadAuthRequest("账号上游请求日志已达到 100000 条上限，请联系管理员归档") +func validateAPICallLogQuotaWithPolicy(usage repository.UserStorageUsage, incomingBytes int64, policy RuntimeResourcePolicy) error { + if usage.APICallCount >= policy.APICallLogCount { + return BadAuthRequest(fmt.Sprintf("账号上游请求日志已达到 %d 条上限，请联系管理员归档", policy.APICallLogCount)) } - return validateTaskDataGrowthQuota(usage, incomingBytes) + return validateTaskDataGrowthQuotaWithPolicy(usage, incomingBytes, policy) } -func validateStructuredReplacementQuota(usage repository.UserStorageUsage, kind string, count int, bytes int64) error { +func validateStructuredReplacementQuotaWithPolicy(usage repository.UserStorageUsage, kind string, count int, bytes int64, policy RuntimeResourcePolicy) error { deltaBytes := bytes switch kind { case "asset": - if count > MaxUserAssetCount { - return BadAuthRequest("账号素材数量不能超过 2000 个") + if int64(count) > policy.AssetCount { + return BadAuthRequest(fmt.Sprintf("账号素材数量不能超过 %d 个", policy.AssetCount)) } deltaBytes -= usage.AssetBytes case "canvas": - if count > MaxUserCanvasCount { - return BadAuthRequest("账号画布数量不能超过 1000 个") + if int64(count) > policy.CanvasCount { + return BadAuthRequest(fmt.Sprintf("账号画布数量不能超过 %d 个", policy.CanvasCount)) } deltaBytes -= usage.CanvasBytes } - return validateStructuredStorageQuota(usage, kind, false, deltaBytes) + return validateStructuredStorageQuotaWithPolicy(usage, kind, false, deltaBytes, policy) } -func (s *Service) createTaskWithinStorageQuota(task *model.Task, billingOrder *model.BillingOrder) error { +func (s *Service) createTaskWithinStorageQuota(task *model.Task, billingOrder *model.BillingOrder, policy RuntimePolicySetting) error { s.storageMu.Lock() defer s.storageMu.Unlock() usage, err := s.repo.UserStorageUsage(task.UserID) @@ -91,17 +82,21 @@ func (s *Service) createTaskWithinStorageQuota(task *model.Task, billingOrder *m return err } incomingBytes := int64(len([]byte(task.Prompt)) + len([]byte(task.InputJSON)) + len([]byte(task.Error))) - if err := validateTaskStorageQuota(usage, incomingBytes); err != nil { + if err := validateTaskStorageQuotaWithPolicy(usage, incomingBytes, policy.Resource); err != nil { return err } if billingOrder != nil { - return s.repo.CreateTaskWithCreditReservation(task, billingOrder) + return s.repo.CreateTaskWithCreditReservation(task, billingOrder, policy.Task.ActiveTaskLimit) } - return s.repo.CreateTaskWithActiveLimit(task) + return s.repo.CreateTaskWithActiveLimit(task, policy.Task.ActiveTaskLimit) } // 任务完成会同时扩张任务历史和 Agent 会话数据，必须在同一临界区核算并原子写入。 func (s *Service) saveTaskCompletionWithinStorageQuota(task *model.Task, resultJSON []byte, opsJSON []byte, hasCanvasOps bool) error { + policy, err := s.RuntimePolicy() + if err != nil { + return err + } s.storageMu.Lock() defer s.storageMu.Unlock() @@ -139,10 +134,10 @@ func (s *Service) saveTaskCompletionWithinStorageQuota(task *model.Task, resultJ for index := range results { taskDelta += int64(len(results[index].URL) + len(results[index].Payload)) } - if err := validateTaskDataGrowthQuota(usage, taskDelta); err != nil { + if err := validateTaskDataGrowthQuotaWithPolicy(usage, taskDelta, policy.Resource); err != nil { return err } - if err := validateStructuredStorageQuota(usage, "session", false, structuredDelta); err != nil { + if err := validateStructuredStorageQuotaWithPolicy(usage, "session", false, structuredDelta, policy.Resource); err != nil { return err } diff --git a/backend/internal/service/storage_quota_test.go b/backend/internal/service/storage_quota_test.go index c606a0c4e..e1c2e137d 100644 --- a/backend/internal/service/storage_quota_test.go +++ b/backend/internal/service/storage_quota_test.go @@ -11,30 +11,33 @@ import ( ) func TestValidateStructuredStorageQuotaRejectsBytesAndCounts(t *testing.T) { - usage := repository.UserStorageUsage{AssetBytes: MaxUserStructuredDataBytes - 8, AssetCount: MaxUserAssetCount} - if err := validateStructuredStorageQuota(usage, "asset", false, 9); err == nil { + policy := defaultRuntimePolicy().Resource + usage := repository.UserStorageUsage{AssetBytes: megabytes(policy.StructuredDataMB) - 8, AssetCount: policy.AssetCount} + if err := validateStructuredStorageQuotaWithPolicy(usage, "asset", false, 9, policy); err == nil { t.Fatal("validateStructuredStorageQuota() byte error = nil") } - if err := validateStructuredStorageQuota(usage, "asset", true, 0); err == nil { + if err := validateStructuredStorageQuotaWithPolicy(usage, "asset", true, 0, policy); err == nil { t.Fatal("validateStructuredStorageQuota() count error = nil") } } func TestValidateStructuredStorageQuotaAllowsReplacementThatShrinksData(t *testing.T) { - usage := repository.UserStorageUsage{AssetBytes: MaxUserStructuredDataBytes, AssetCount: MaxUserAssetCount} - if err := validateStructuredStorageQuota(usage, "asset", false, -1); err != nil { + policy := defaultRuntimePolicy().Resource + usage := repository.UserStorageUsage{AssetBytes: megabytes(policy.StructuredDataMB), AssetCount: policy.AssetCount} + if err := validateStructuredStorageQuotaWithPolicy(usage, "asset", false, -1, policy); err != nil { t.Fatalf("validateStructuredStorageQuota() error = %v", err) } } func TestValidateTaskStorageQuotaRejectsHistoryGrowth(t *testing.T) { - if err := validateTaskStorageQuota(repository.UserStorageUsage{TaskCount: MaxUserTaskCount}, 0); err == nil { + policy := defaultRuntimePolicy().Resource + if err := validateTaskStorageQuotaWithPolicy(repository.UserStorageUsage{TaskCount: policy.TaskCount}, 0, policy); err == nil { t.Fatal("validateTaskStorageQuota() count error = nil") } - if err := validateTaskStorageQuota(repository.UserStorageUsage{TaskBytes: MaxUserTaskDataBytes}, 1); err == nil { + if err := validateTaskStorageQuotaWithPolicy(repository.UserStorageUsage{TaskBytes: gigabytes(policy.TaskDataGB)}, 1, policy); err == nil { t.Fatal("validateTaskStorageQuota() byte error = nil") } - if err := validateAPICallLogQuota(repository.UserStorageUsage{APICallCount: MaxUserAPICallLogCount}, 0); err == nil { + if err := validateAPICallLogQuotaWithPolicy(repository.UserStorageUsage{APICallCount: policy.APICallLogCount}, 0, policy); err == nil { t.Fatal("validateAPICallLogQuota() count error = nil") } } @@ -44,7 +47,7 @@ func TestUserStorageUsageCountsPersistedPayloads(t *testing.T) { if err != nil { t.Fatal(err) } - if err := db.AutoMigrate(&model.Asset{}, &model.CanvasProject{}, &model.Session{}, &model.Message{}, &model.Task{}, &model.TaskLog{}, &model.Result{}, &model.ApiCallLog{}); err != nil { + if err := db.AutoMigrate(&model.SystemSetting{}, &model.Asset{}, &model.CanvasProject{}, &model.Session{}, &model.Message{}, &model.Task{}, &model.TaskLog{}, &model.Result{}, &model.ApiCallLog{}); err != nil { t.Fatal(err) } items := []any{ @@ -76,7 +79,7 @@ func TestSaveTaskCompletionPersistsRelatedRowsTogether(t *testing.T) { if err != nil { t.Fatal(err) } - if err := db.AutoMigrate(&model.Asset{}, &model.CanvasProject{}, &model.Session{}, &model.Message{}, &model.Task{}, &model.TaskLog{}, &model.Result{}, &model.ApiCallLog{}); err != nil { + if err := db.AutoMigrate(&model.SystemSetting{}, &model.Asset{}, &model.CanvasProject{}, &model.Session{}, &model.Message{}, &model.Task{}, &model.TaskLog{}, &model.Result{}, &model.ApiCallLog{}); err != nil { t.Fatal(err) } session := model.Session{ID: "session-1", UserID: "user-1", Status: model.SessionStatusActive} diff --git a/backend/internal/service/task_security_test.go b/backend/internal/service/task_security_test.go index 9262b874f..517d4dfe9 100644 --- a/backend/internal/service/task_security_test.go +++ b/backend/internal/service/task_security_test.go @@ -81,7 +81,7 @@ func TestCreateSessionRemovesDraftWhenTaskCreationFails(t *testing.T) { if err != nil { t.Fatal(err) } - if err := db.AutoMigrate(&model.Asset{}, &model.CanvasProject{}, &model.Session{}, &model.Message{}, &model.Task{}, &model.TaskLog{}, &model.Result{}, &model.ApiCallLog{}); err != nil { + if err := db.AutoMigrate(&model.SystemSetting{}, &model.Asset{}, &model.CanvasProject{}, &model.Session{}, &model.Message{}, &model.Task{}, &model.TaskLog{}, &model.Result{}, &model.ApiCallLog{}); err != nil { t.Fatal(err) } for range 5 { diff --git a/backend/internal/service/upload_quota.go b/backend/internal/service/upload_quota.go index 39f67cbd7..5995c56c8 100644 --- a/backend/internal/service/upload_quota.go +++ b/backend/internal/service/upload_quota.go @@ -2,26 +2,39 @@ package service import ( "errors" + "fmt" "log" "time" "infinite-canvas/backend/internal/repository" ) -const MaxResourceUploadBytes int64 = 50 << 20 -const MaxDailyUserUploadBytes int64 = 200 << 20 -const MaxUserStoredFileBytes int64 = 2 << 30 - // 上传额度在写文件或 OSS 前原子预留，避免并发请求同时通过日限额检查。 func (s *Service) reserveUserUploadQuota(userID string, size int64) (string, error) { - return s.reserveUserStoredFileQuota(userID, size, MaxResourceUploadBytes, "单个上传文件必须小于 50MB") + policy, err := s.RuntimePolicy() + if err != nil { + return "", err + } + return s.reserveUserStoredFileQuota(userID, size, megabytes(policy.Resource.ResourceUploadMB), megabytes(policy.Resource.DailyUploadMB), gigabytes(policy.Resource.StoredFileGB), fmt.Sprintf("单个上传文件必须小于 %dMB", policy.Resource.ResourceUploadMB)) +} + +func (s *Service) reserveSessionUploadQuota(userID string, size int64) (string, error) { + policy, err := s.RuntimePolicy() + if err != nil { + return "", err + } + return s.reserveUserStoredFileQuota(userID, size, megabytes(policy.Resource.SessionUploadMB)+1, megabytes(policy.Resource.DailyUploadMB), gigabytes(policy.Resource.StoredFileGB), fmt.Sprintf("会话文件不能超过 %dMB", policy.Resource.SessionUploadMB)) } func (s *Service) reserveGeneratedResourceQuota(userID string, size int64) (string, error) { - return s.reserveUserStoredFileQuota(userID, size, maxProviderResponseBytes+1, "单个生成文件不能超过 64MB") + policy, err := s.RuntimePolicy() + if err != nil { + return "", err + } + return s.reserveUserStoredFileQuota(userID, size, megabytes(policy.Resource.GeneratedFileMB)+1, megabytes(policy.Resource.DailyUploadMB), gigabytes(policy.Resource.StoredFileGB), fmt.Sprintf("单个生成文件不能超过 %dMB", policy.Resource.GeneratedFileMB)) } -func (s *Service) reserveUserStoredFileQuota(userID string, size int64, exclusiveSingleFileLimit int64, singleFileMessage string) (string, error) { +func (s *Service) reserveUserStoredFileQuota(userID string, size int64, exclusiveSingleFileLimit int64, dailyLimit int64, storedLimit int64, singleFileMessage string) (string, error) { if size <= 0 { return "", BadAuthRequest("上传文件不能为空") } @@ -38,20 +51,27 @@ func (s *Service) reserveUserStoredFileQuota(userID string, size int64, exclusiv if s.pendingStorage == nil { s.pendingStorage = map[string]int64{} } - if storedBytes+s.pendingStorage[userID]+size >= MaxUserStoredFileBytes { - return "", BadAuthRequest("账号资源和会话附件已达到 2GB 上限，请联系管理员清理历史文件") + if storedBytes+s.pendingStorage[userID]+size >= storedLimit { + return "", BadAuthRequest(fmt.Sprintf("账号资源和会话附件已达到 %s 上限，请联系管理员清理历史文件", formatStorageLimit(storedLimit))) } s.pendingStorage[userID] += size - if err := s.repo.ReserveDailyUpload(userID, day, size, MaxDailyUserUploadBytes); err != nil { + if err := s.repo.ReserveDailyUpload(userID, day, size, dailyLimit); err != nil { s.decreasePendingStorage(userID, size) if errors.Is(err, repository.ErrDailyUploadLimitExceeded) { - return "", BadAuthRequest("每个账号 UTC 自然日上传总量必须小于 200MB") + return "", BadAuthRequest(fmt.Sprintf("每个账号 UTC 自然日上传总量必须小于 %s", formatStorageLimit(dailyLimit))) } return "", err } return day, nil } +func formatStorageLimit(value int64) string { + if value%(1<<30) == 0 { + return fmt.Sprintf("%dGB", value>>30) + } + return fmt.Sprintf("%dMB", value>>20) +} + func (s *Service) releaseUserUploadQuota(userID string, day string, size int64) { if day == "" || size <= 0 { return diff --git a/backend/internal/service/upload_quota_test.go b/backend/internal/service/upload_quota_test.go index b27064673..53d5e3f23 100644 --- a/backend/internal/service/upload_quota_test.go +++ b/backend/internal/service/upload_quota_test.go @@ -9,7 +9,7 @@ import ( func TestReserveUserUploadQuotaRejectsSingleFileAtLimit(t *testing.T) { svc := newResourceTestService(t) - _, err := svc.reserveUserUploadQuota("user-1", MaxResourceUploadBytes) + _, err := svc.reserveUserUploadQuota("user-1", megabytes(defaultRuntimePolicy().Resource.ResourceUploadMB)) if err == nil || !strings.Contains(err.Error(), "小于 50MB") { t.Fatalf("reserveUserUploadQuota() error = %v", err) } @@ -60,7 +60,7 @@ func TestCommitUserUploadQuotaKeepsDailyUsageWithoutPendingStorage(t *testing.T) func TestReserveUserUploadQuotaRejectsTotalStoredFilesAtLimit(t *testing.T) { svc := newResourceTestService(t) - if err := svc.repo.Create(&model.Resource{ID: "resource-1", UserID: "user-1", Status: model.ResourceStatusReady, Size: MaxUserStoredFileBytes - 1}); err != nil { + if err := svc.repo.Create(&model.Resource{ID: "resource-1", UserID: "user-1", Status: model.ResourceStatusReady, Size: gigabytes(defaultRuntimePolicy().Resource.StoredFileGB) - 1}); err != nil { t.Fatal(err) } if _, err := svc.reserveUserUploadQuota("user-1", 1); err == nil || !strings.Contains(err.Error(), "2GB 上限") { diff --git a/backend/internal/service/user_data.go b/backend/internal/service/user_data.go index 720b7b2af..2ea3555a9 100644 --- a/backend/internal/service/user_data.go +++ b/backend/internal/service/user_data.go @@ -52,6 +52,10 @@ func (s *Service) UpsertUserAsset(userID string, raw json.RawMessage) (UserDataS if err != nil { return UserDataSummary{}, err } + policy, err := s.RuntimePolicy() + if err != nil { + return UserDataSummary{}, err + } s.storageMu.Lock() defer s.storageMu.Unlock() existing, existingErr := s.repo.AssetForUser(userID, asset.ID) @@ -66,7 +70,7 @@ func (s *Service) UpsertUserAsset(userID string, raw json.RawMessage) (UserDataS if err != nil { return UserDataSummary{}, err } - if err := validateStructuredStorageQuota(usage, "asset", errors.Is(existingErr, gorm.ErrRecordNotFound), int64(len(raw))-existingBytes); err != nil { + if err := validateStructuredStorageQuotaWithPolicy(usage, "asset", errors.Is(existingErr, gorm.ErrRecordNotFound), int64(len(raw))-existingBytes, policy.Resource); err != nil { return UserDataSummary{}, err } if err := s.repo.UpsertAsset(&asset); err != nil { @@ -107,13 +111,17 @@ func (s *Service) ReplaceUserAssets(userID string, req AssetsSyncRequest) ([]jso assets = append(assets, item) totalBytes += int64(len(raw)) } + policy, err := s.RuntimePolicy() + if err != nil { + return nil, err + } s.storageMu.Lock() defer s.storageMu.Unlock() usage, err := s.repo.UserStorageUsage(userID) if err != nil { return nil, err } - if err := validateStructuredReplacementQuota(usage, "asset", len(assets), totalBytes); err != nil { + if err := validateStructuredReplacementQuotaWithPolicy(usage, "asset", len(assets), totalBytes, policy.Resource); err != nil { return nil, err } if err := s.repo.ReplaceAssets(userID, assets); err != nil { @@ -164,6 +172,10 @@ func (s *Service) UpsertUserCanvasProject(userID string, raw json.RawMessage) (U if err != nil { return UserDataSummary{}, err } + policy, err := s.RuntimePolicy() + if err != nil { + return UserDataSummary{}, err + } s.storageMu.Lock() defer s.storageMu.Unlock() existing, existingErr := s.repo.CanvasProjectForUser(userID, project.ID) @@ -178,7 +190,7 @@ func (s *Service) UpsertUserCanvasProject(userID string, raw json.RawMessage) (U if err != nil { return UserDataSummary{}, err } - if err := validateStructuredStorageQuota(usage, "canvas", errors.Is(existingErr, gorm.ErrRecordNotFound), int64(len(raw))-existingBytes); err != nil { + if err := validateStructuredStorageQuotaWithPolicy(usage, "canvas", errors.Is(existingErr, gorm.ErrRecordNotFound), int64(len(raw))-existingBytes, policy.Resource); err != nil { return UserDataSummary{}, err } if err := s.repo.UpsertCanvasProject(&project); err != nil { @@ -208,13 +220,17 @@ func (s *Service) ReplaceUserCanvasProjects(userID string, req CanvasProjectsSyn projects = append(projects, item) totalBytes += int64(len(raw)) } + policy, err := s.RuntimePolicy() + if err != nil { + return nil, err + } s.storageMu.Lock() defer s.storageMu.Unlock() usage, err := s.repo.UserStorageUsage(userID) if err != nil { return nil, err } - if err := validateStructuredReplacementQuota(usage, "canvas", len(projects), totalBytes); err != nil { + if err := validateStructuredReplacementQuotaWithPolicy(usage, "canvas", len(projects), totalBytes, policy.Resource); err != nil { return nil, err } if err := s.repo.ReplaceCanvasProjects(userID, projects); err != nil { diff --git a/nginx.conf b/nginx.conf index 6b11a7d37..28350f239 100644 --- a/nginx.conf +++ b/nginx.conf @@ -12,7 +12,8 @@ server { server_name _; root /usr/share/nginx/html; index index.html; - client_max_body_size 64m; + # 业务配置最高允许 999MB；网关保留少量 multipart 开销，避免保存出无法生效的上限。 + client_max_body_size 1024m; access_log /var/log/nginx/access.log canvas_no_query; add_header Referrer-Policy "no-referrer" always; gzip on; diff --git a/web/src/lib/canvas/canvas-generation-batch.ts b/web/src/lib/canvas/canvas-generation-batch.ts index d82fe8b94..22d645fe6 100644 --- a/web/src/lib/canvas/canvas-generation-batch.ts +++ b/web/src/lib/canvas/canvas-generation-batch.ts @@ -1,9 +1,9 @@ import type { CanvasGenerationBatch, CanvasGenerationBatchStatus } from "@/types/canvas"; -const TASK_CAPACITY_MESSAGE = "同时排队或运行的任务最多 5 个"; +const TASK_CAPACITY_MESSAGE = /同时排队或运行的任务最多 \d+ 个/; export function isGenerationTaskCapacityError(error: unknown) { - return error instanceof Error && error.message.includes(TASK_CAPACITY_MESSAGE); + return error instanceof Error && TASK_CAPACITY_MESSAGE.test(error.message); } export function isGenerationCostUncertainError(error: unknown) { diff --git a/web/src/lib/user-session.ts b/web/src/lib/user-session.ts index f23c3f9cc..950fca272 100644 --- a/web/src/lib/user-session.ts +++ b/web/src/lib/user-session.ts @@ -19,6 +19,7 @@ export async function applyUserSession(payload: AuthSessionPayload) { ]); const persistedConfig = scopedLocalStorage.getItem(CONFIG_STORE_KEY); useUserStore.getState().setUser(payload.user); + useUserStore.getState().setRuntimeLimits(payload.runtimeLimits); await Promise.all([useCanvasStore.persist.rehydrate(), useAssetStore.persist.rehydrate(), useConfigStore.persist.rehydrate()]); // Zustand 在目标 scope 没有快照时会保留旧内存，必须显式恢复该 scope 的空状态。 if (!persistedCanvas) useCanvasStore.setState({ projects: [] }); diff --git a/web/src/pages/admin/channels/channels-page.tsx b/web/src/pages/admin/channels/channels-page.tsx index 18bedb7a8..8ca547401 100644 --- a/web/src/pages/admin/channels/channels-page.tsx +++ b/web/src/pages/admin/channels/channels-page.tsx @@ -164,7 +164,7 @@ export default function ChannelsPage() { - + diff --git a/web/src/pages/admin/components/admin-shell.tsx b/web/src/pages/admin/components/admin-shell.tsx index 066ca9ec3..538eeead5 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, Gauge, HardDrive, Home, Mail, MessageSquareText, PanelLeftClose, PanelLeftOpen, RadioTower, ShieldCheck, TicketCheck, UsersRound } from "lucide-react"; +import { BarChart3, BellRing, Coins, FileClock, HardDrive, Home, Mail, MessageSquareText, PanelLeftClose, PanelLeftOpen, RadioTower, Settings2, ShieldCheck, TicketCheck, UsersRound } from "lucide-react"; import { useState, type ReactNode } from "react"; import { NavLink, Outlet } from "react-router"; @@ -40,7 +40,7 @@ const adminNavigation: Array<{ label: string; items: AdminNavigationItem[] }> = { label: "系统配置", items: [ - { path: "/admin/settings/concurrency", label: "任务并发", description: "Worker 与渠道请求调度", icon: , secondary: true }, + { path: "/admin/settings/runtime-policy", label: "资源与策略", description: "配额、并发、频控与超时", 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/admin-user-detail-drawer.tsx b/web/src/pages/admin/components/admin-user-detail-drawer.tsx index 28cbc4529..2e08d4ca0 100644 --- a/web/src/pages/admin/components/admin-user-detail-drawer.tsx +++ b/web/src/pages/admin/components/admin-user-detail-drawer.tsx @@ -1,5 +1,5 @@ import { useEffect, useState } from "react"; -import { App, Descriptions, Drawer, Empty, Skeleton, Table, Tabs, Tag } from "antd"; +import { App, Descriptions, Drawer, Empty, Progress, Skeleton, Table, Tabs, Tag } from "antd"; import { formatCredits } from "@/constant/credits"; import { getAdminUserDetail, listAdminUserAuditEvents, listAdminUserLedger, listAdminUserTasks, type AdminAuditEvent, type AdminUserDetail, type AdminUserTask } from "@/services/api/auth"; @@ -119,6 +119,20 @@ export function AdminUserDetailDrawer({ userId, onClose }: { userId: string | nu ))} +
+
资源与配额占用
+
+ {quotaUsageItems(detail).map((item) => ( +
+
+ {item.label} + {item.display} +
+ 0 ? Math.round(item.value / item.limit * 100) : 0)} size="small" showInfo={false} status={item.value >= item.limit ? "exception" : "normal"} /> +
+ ))} +
+
), }, @@ -195,3 +209,20 @@ export function AdminUserDetailDrawer({ userId, onClose }: { userId: string | nu function formatTime(value?: string) { return value ? new Date(value).toLocaleString("zh-CN", { hour12: false }) : "--"; } + +function quotaUsageItems(detail: AdminUserDetail) { + const structuredBytes = detail.storageUsage.assetBytes + detail.storageUsage.canvasBytes + detail.storageUsage.sessionBytes; + const bytes = (value: number) => value >= 1024 ** 3 ? `${(value / 1024 ** 3).toFixed(2)} GB` : `${(value / 1024 ** 2).toFixed(1)} MB`; + const number = (value: number) => new Intl.NumberFormat("zh-CN").format(value); + return [ + { label: "资源与附件", value: detail.storedFileBytes, limit: detail.quota.storedFileGB * 1024 ** 3, display: `${bytes(detail.storedFileBytes)} / ${detail.quota.storedFileGB} GB` }, + { label: "今日上传（UTC）", value: detail.dailyUploadBytes, limit: detail.quota.dailyUploadMB * 1024 ** 2, display: `${bytes(detail.dailyUploadBytes)} / ${detail.quota.dailyUploadMB} MB` }, + { label: "画布、素材与会话数据", value: structuredBytes, limit: detail.quota.structuredDataMB * 1024 ** 2, display: `${bytes(structuredBytes)} / ${detail.quota.structuredDataMB} MB` }, + { label: "任务与请求日志数据", value: detail.storageUsage.taskBytes, limit: detail.quota.taskDataGB * 1024 ** 3, display: `${bytes(detail.storageUsage.taskBytes)} / ${detail.quota.taskDataGB} GB` }, + { label: "素材数量", value: detail.storageUsage.assetCount, limit: detail.quota.assetCount, display: `${number(detail.storageUsage.assetCount)} / ${number(detail.quota.assetCount)}` }, + { label: "画布数量", value: detail.storageUsage.canvasCount, limit: detail.quota.canvasCount, display: `${number(detail.storageUsage.canvasCount)} / ${number(detail.quota.canvasCount)}` }, + { label: "Agent 会话数量", value: detail.storageUsage.sessionCount, limit: detail.quota.sessionCount, display: `${number(detail.storageUsage.sessionCount)} / ${number(detail.quota.sessionCount)}` }, + { label: "任务历史数量", value: detail.storageUsage.taskCount, limit: detail.quota.taskCount, display: `${number(detail.storageUsage.taskCount)} / ${number(detail.quota.taskCount)}` }, + { label: "上游请求日志数量", value: detail.storageUsage.apiCallCount, limit: detail.quota.apiCallLogCount, display: `${number(detail.storageUsage.apiCallCount)} / ${number(detail.quota.apiCallLogCount)}` }, + ]; +} diff --git a/web/src/pages/admin/settings/concurrency-settings-page.tsx b/web/src/pages/admin/settings/concurrency-settings-page.tsx deleted file mode 100644 index 2a7c80d63..000000000 --- a/web/src/pages/admin/settings/concurrency-settings-page.tsx +++ /dev/null @@ -1,73 +0,0 @@ -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/pages/admin/settings/runtime-policy-settings-page.tsx b/web/src/pages/admin/settings/runtime-policy-settings-page.tsx new file mode 100644 index 000000000..43cbcf658 --- /dev/null +++ b/web/src/pages/admin/settings/runtime-policy-settings-page.tsx @@ -0,0 +1,200 @@ +import { App, Button, Form, InputNumber, Tag } from "antd"; +import { Database, Gauge, Infinity as InfinityIcon, Network, RotateCcw, Save, ShieldCheck, TimerReset } from "lucide-react"; +import { useEffect, useMemo, useState, type ReactNode } from "react"; + +import { + getAdminRuntimePolicySetting, + getAdminSelfUseRuntimePolicy, + resetAdminRuntimePolicySetting, + updateAdminRuntimePolicySetting, + type RuntimePolicySetting, +} from "@/services/api/auth"; +import { useAdminContext } from "../admin-context"; +import { AdminPageFrame } from "../components/admin-shell"; +import { SettingsSectionCard } from "../components/admin-ui"; + +type PolicyGroup = "resource" | "task" | "request"; +type PolicyField = { + group: PolicyGroup; + name: string; + label: string; + extra: string; + unit: string; + max: number; +}; + +const resourceFields: PolicyField[] = [ + { group: "resource", name: "resourceUploadMB", label: "普通资源单文件", extra: "素材上传和远程导入的单文件业务上限。", unit: "MB", max: 999 }, + { group: "resource", name: "sessionUploadMB", label: "Agent 会话附件", extra: "单个会话附件的大小上限。", unit: "MB", max: 999 }, + { group: "resource", name: "generatedFileMB", label: "单个生成资源", extra: "上游生成响应和落库资源的单文件上限。", unit: "MB", max: 999 }, + { group: "resource", name: "dailyUploadMB", label: "每日上传总量", extra: "按 UTC 自然日累计资源与附件上传。", unit: "MB", max: 999_999 }, + { group: "resource", name: "storedFileGB", label: "账号文件总量", extra: "资源文件与 Agent 会话附件合计。", unit: "GB", max: 999 }, + { group: "resource", name: "structuredDataMB", label: "结构化数据总量", extra: "画布、素材和 Agent 会话结构化数据合计。", unit: "MB", max: 999_999 }, + { group: "resource", name: "taskDataGB", label: "任务数据总量", extra: "任务历史、结果和上游请求日志合计。", unit: "GB", max: 999 }, + { group: "resource", name: "assetCount", label: "素材数量", extra: "单账号可保存的素材记录数。", unit: "条", max: 999_999_999 }, + { group: "resource", name: "canvasCount", label: "画布数量", extra: "单账号可保存的画布数量。", unit: "个", max: 999_999_999 }, + { group: "resource", name: "sessionCount", label: "Agent 会话数量", extra: "单账号可保存的 Agent 会话数量。", unit: "个", max: 999_999_999 }, + { group: "resource", name: "taskCount", label: "任务历史数量", extra: "单账号保留的任务历史记录数。", unit: "条", max: 999_999_999 }, + { group: "resource", name: "apiCallLogCount", label: "请求日志数量", extra: "单账号保留的上游请求日志数。", unit: "条", max: 999_999_999 }, +]; + +const concurrencyFields: PolicyField[] = [ + { group: "task", name: "workerConcurrency", label: "Worker 并发", extra: "集群同时执行的后台任务数。", unit: "个", max: 999 }, + { group: "task", name: "channelConcurrency", label: "全局渠道并发", extra: "渠道选择跟随系统时采用的并发上限。", unit: "个", max: 999 }, + { group: "task", name: "activeTaskLimit", label: "账号活动任务", extra: "单账号同时排队或运行的任务数。", unit: "个", max: 999 }, +]; + +const timeoutFields: PolicyField[] = [ + { group: "task", name: "imageTimeoutMinutes", label: "图片任务超时", extra: "图片任务进入失败状态前的最长执行时间。", unit: "分钟", max: 9_999 }, + { group: "task", name: "textTimeoutMinutes", label: "文本任务超时", extra: "文本任务的最长执行时间。", unit: "分钟", max: 9_999 }, + { group: "task", name: "audioTimeoutMinutes", label: "音频任务超时", extra: "音频任务的最长执行时间。", unit: "分钟", max: 9_999 }, + { group: "task", name: "videoTimeoutMinutes", label: "视频任务超时", extra: "视频任务的最长执行时间。", unit: "分钟", max: 9_999 }, + { group: "task", name: "storyboardTimeoutMinutes", label: "分镜任务超时", extra: "Agent 分镜任务的最长执行时间。", unit: "分钟", max: 9_999 }, + { group: "task", name: "defaultTimeoutMinutes", label: "默认任务超时", extra: "未匹配专用类型时使用的最长执行时间。", unit: "分钟", max: 9_999 }, +]; + +const rateFields: PolicyField[] = [ + { group: "request", name: "taskCreatePerMinute", label: "任务创建", extra: "每账号每分钟允许创建的任务数。", unit: "次/分钟", max: 999_999 }, + { group: "request", name: "sessionCreatePerMinute", label: "会话创建", extra: "每账号每分钟允许创建的会话数。", unit: "次/分钟", max: 999_999 }, + { group: "request", name: "resourceUploadPerMinute", label: "资源上传", extra: "每账号每分钟上传资源的次数。", unit: "次/分钟", max: 999_999 }, + { group: "request", name: "resourceImportPerMinute", label: "资源导入", extra: "每账号每分钟导入远程资源的次数。", unit: "次/分钟", max: 999_999 }, + { group: "request", name: "sessionFilePerMinute", label: "会话附件", extra: "每账号每分钟上传会话附件的次数。", unit: "次/分钟", max: 999_999 }, + { group: "request", name: "assetWritePerMinute", label: "素材写入", extra: "每账号每分钟写入素材的次数。", unit: "次/分钟", max: 999_999 }, + { group: "request", name: "canvasWritePerMinute", label: "画布写入", extra: "每账号每分钟写入画布的次数。", unit: "次/分钟", max: 999_999 }, + { group: "request", name: "registerPerHour", label: "账号注册", extra: "每 IP 每小时允许注册的次数。", unit: "次/小时", max: 999_999 }, + { group: "request", name: "emailCodePerHour", label: "邮箱验证码", extra: "每 IP 每小时允许请求验证码的次数。", unit: "次/小时", max: 999_999 }, + { group: "request", name: "loginIPPerTenMinutes", label: "登录 IP", extra: "每 IP 每 10 分钟允许登录的次数。", unit: "次/10分钟", max: 999_999 }, + { group: "request", name: "loginAccountPerTenMinutes", label: "登录账号组合", extra: "同一 IP 与账号组合每 10 分钟的登录次数。", unit: "次/10分钟", max: 999_999 }, + { group: "request", name: "systemRelayPerMinute", label: "系统渠道中转", extra: "每账号每分钟使用系统渠道的请求数。", unit: "次/分钟", max: 999_999 }, + { group: "request", name: "customRelayPerMinute", label: "自定义渠道中转", extra: "每账号每分钟使用自定义渠道的请求数。", unit: "次/分钟", max: 999_999 }, +]; + +const relayFields: PolicyField[] = [ + { group: "request", name: "customRelayConcurrency", label: "自定义渠道并发", extra: "单账号同时进行的自定义渠道请求数。", unit: "个", max: 999 }, + { group: "request", name: "customRelayRequestMB", label: "自定义渠道请求体", extra: "中转到自定义上游的请求体上限。", unit: "MB", max: 999 }, + { group: "request", name: "customRelayResponseMB", label: "自定义渠道响应体", extra: "自定义上游 JSON 与流式响应的读取上限。", unit: "MB", max: 999 }, + { group: "request", name: "customRelayTimeoutMinutes", label: "自定义渠道超时", extra: "自定义渠道连接与响应的最长等待时间。", unit: "分钟", max: 9_999 }, + { group: "request", name: "systemRelayRequestMB", label: "系统渠道请求体", extra: "中转到系统渠道的请求体上限。", unit: "MB", max: 999 }, + { group: "request", name: "systemRelayResponseMB", label: "系统渠道响应体", extra: "系统渠道上游响应的读取上限。", unit: "MB", max: 999 }, + { group: "request", name: "channelCircuitFailureCount", label: "熔断失败次数", extra: "一分钟内连续失败达到该值后打开熔断。", unit: "次", max: 999 }, + { group: "request", name: "channelCircuitOpenSeconds", label: "熔断持续时间", extra: "渠道熔断打开后拒绝请求的时间。", unit: "秒", max: 86_400 }, +]; + +export default function RuntimePolicySettingsPage() { + const { message, modal } = App.useApp(); + const { references } = useAdminContext(); + const [setting, setSetting] = useState(null); + const [loading, setLoading] = useState(true); + const [saving, setSaving] = useState(false); + const [dirty, setDirty] = 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 getAdminRuntimePolicySetting() + .then(({ setting: value }) => { + setSetting(value); + form.setFieldsValue(value); + }) + .catch((error) => message.error(error instanceof Error ? error.message : "读取资源与请求策略失败")) + .finally(() => setLoading(false)); + }, [form, message]); + + useEffect(() => { + const beforeUnload = (event: BeforeUnloadEvent) => { + if (!dirty) return; + event.preventDefault(); + }; + window.addEventListener("beforeunload", beforeUnload); + return () => window.removeEventListener("beforeunload", beforeUnload); + }, [dirty]); + + const useSelfMode = async () => { + try { + const result = await getAdminSelfUseRuntimePolicy(); + form.setFieldsValue(result.setting); + setDirty(true); + message.info("已填入自用模式上限，保存后生效"); + } catch (error) { + message.error(error instanceof Error ? error.message : "读取自用模式失败"); + } + }; + + const reset = () => { + modal.confirm({ + title: "重置全部资源与请求策略？", + content: "将删除已保存的自定义策略并立即恢复系统默认值。", + okText: "重置默认值", + cancelText: "取消", + okButtonProps: { danger: true }, + onOk: async () => { + try { + const result = await resetAdminRuntimePolicySetting(); + setSetting(result.setting); + form.setFieldsValue(result.setting); + setDirty(false); + message.success("已恢复系统默认策略"); + } catch (error) { + message.error(error instanceof Error ? error.message : "重置资源与请求策略失败"); + throw error; + } + }, + }); + }; + + const save = async () => { + const values = await form.validateFields(); + setSaving(true); + try { + const result = await updateAdminRuntimePolicySetting({ resource: values.resource, task: values.task, request: values.request }); + setSetting(result.setting); + form.setFieldsValue(result.setting); + setDirty(false); + message.success("资源与请求策略已即时生效"); + } catch (error) { + message.error(error instanceof Error ? error.message : "保存资源与请求策略失败"); + } finally { + setSaving(false); + } + }; + + return ( + } + > +
setDirty(true)}> +
+ } title="资源与账号配额" description="上传、文件容量、结构化数据和历史记录上限。" fields={resourceFields} /> + } title="任务与并发" description="后台任务消费、渠道调度和单账号活动任务上限。" fields={concurrencyFields} status={热更新} /> + } title="任务超时" description="不同生成类型的最长执行时间。" fields={timeoutFields} /> + } title="业务频控" description="账号与 IP 维度的固定窗口请求限制。" fields={rateFields} /> + } title="渠道中转与熔断" description="请求体、响应体、并发、超时和上游故障保护。" fields={relayFields} /> +
+
{setting?.updatedAt ? `上次更新：${formatTime(setting.updatedAt)}${setting.updatedBy ? ` · ${userNameById.get(setting.updatedBy) || setting.updatedBy}` : ""}` : "当前使用系统默认策略"}
+ +
+
+
+ + ); +} + +function PolicySection({ icon, title, description, fields, status }: { icon: ReactNode; title: string; description: string; fields: PolicyField[]; status?: ReactNode }) { + return ( + +
+ {fields.map((field) => ( + + + + ))} +
+ + ); +} + +function formatTime(value?: string) { + return value ? new Date(value).toLocaleString("zh-CN", { hour12: false }) : "--"; +} diff --git a/web/src/pages/canvas/use-canvas-generation-batches.ts b/web/src/pages/canvas/use-canvas-generation-batches.ts index 399d6062e..9758d4dff 100644 --- a/web/src/pages/canvas/use-canvas-generation-batches.ts +++ b/web/src/pages/canvas/use-canvas-generation-batches.ts @@ -7,12 +7,12 @@ import { buildGenerationConfig, generationTaskMetadata, resetGenerationTaskMetad import { unchangedModeratedPrompt } from "@/lib/generation-error"; import { cancelGenerationTask, listGenerationTasks } from "@/services/api/task-center"; import { useConfigStore, useEffectiveConfig } from "@/stores/use-config-store"; +import { useUserStore } from "@/stores/use-user-store"; import type { CanvasGenerationBatch, CanvasGenerationBatchItem, CanvasGenerationBatchMode, CanvasNodeData } from "@/types/canvas"; import type { CanvasNodeGenerationOptions } from "./use-canvas-generation-executor"; import type { CanvasNodeGenerationMode } from "@/components/canvas/canvas-node-prompt-panel"; -const MAX_ACTIVE_TASKS_PER_USER = 5; const SCHEDULER_INTERVAL_MS = 2_000; const MAX_BATCH_HISTORY = 20; @@ -31,6 +31,7 @@ export function useCanvasGenerationBatches({ projectId, projectLoaded, nodes, no const { message, modal } = App.useApp(); const effectiveConfig = useEffectiveConfig(); const isAiConfigReady = useConfigStore((state) => state.isAiConfigReady); + const activeTaskLimit = useUserStore((state) => state.runtimeLimits.activeTaskLimit); const schedulingRef = useRef(false); const controllersRef = useRef(new Map()); @@ -139,7 +140,7 @@ export function useCanvasGenerationBatches({ projectId, projectLoaded, nodes, no }); }, [projectId, setNodes]); - // 调度只补齐用户级 5 个活跃任务的空位，后端 3 个 worker 继续负责实际执行并发。 + // 调度只补齐后台返回的用户级活跃任务空位，最终并发仍由后端原子校验。 const scheduleWaitingItems = useCallback(async () => { if (!projectLoaded || schedulingRef.current) return; schedulingRef.current = true; @@ -154,7 +155,7 @@ export function useCanvasGenerationBatches({ projectId, projectLoaded, nodes, no const item = currentNodes.flatMap((node) => node.metadata?.generationBatches || []).flatMap((batch) => batch.items).find((candidate) => candidate.id === itemId); return item ? !nodeById.get(item.nodeId)?.metadata?.taskId : false; }).length; - let availableSlots = Math.max(0, MAX_ACTIVE_TASKS_PER_USER - activeTaskCount - pendingReservations); + let availableSlots = Math.max(0, activeTaskLimit - activeTaskCount - pendingReservations); if (!availableSlots) return; const candidates: Array<{ batch: CanvasGenerationBatch; item: CanvasGenerationBatchItem; node: CanvasNodeData }> = []; @@ -198,7 +199,7 @@ export function useCanvasGenerationBatches({ projectId, projectLoaded, nodes, no } finally { schedulingRef.current = false; } - }, [effectiveConfig, handleGenerateNode, isAiConfigReady, nodesRef, projectId, projectLoaded, reconcileBatches, updateBatch]); + }, [activeTaskLimit, effectiveConfig, handleGenerateNode, isAiConfigReady, nodesRef, projectId, projectLoaded, reconcileBatches, updateBatch]); const retryFailedBatchItems = useCallback((sourceNodeId: string, batchId: string, itemId?: string) => { const batch = findBatch(nodesRef.current, sourceNodeId, batchId); diff --git a/web/src/router.tsx b/web/src/router.tsx index 2bd8429ae..691329e59 100644 --- a/web/src/router.tsx +++ b/web/src/router.tsx @@ -7,7 +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 RuntimePolicySettingsPage from "@/pages/admin/settings/runtime-policy-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"; @@ -62,8 +62,9 @@ export const router = createBrowserRouter([ { path: "credit-operations", element: }, { path: "redemption-codes", element: }, { path: "logs", element: }, - { path: "settings", element: }, - { path: "settings/concurrency", element: }, + { path: "settings", element: }, + { path: "settings/concurrency", element: }, + { path: "settings/runtime-policy", 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 f183ba511..0ce8eeb48 100644 --- a/web/src/services/api/auth.ts +++ b/web/src/services/api/auth.ts @@ -29,6 +29,13 @@ export type AdminUser = LocalUser & { export type AuthSessionPayload = { user: LocalUser | null; systemChannels?: ModelChannel[]; + runtimeLimits?: RuntimeLimits; +}; + +export type RuntimeLimits = { + activeTaskLimit: number; + resourceUploadMB: number; + sessionUploadMB: number; }; export type ApiCallLog = { @@ -81,6 +88,13 @@ export type AdminUserDetail = { user: LocalUser; account: { userId: string; availableMicrocredits: number; reservedMicrocredits: number; version: number }; counts: { ledgerEntries: number; tasks: number; apiCalls: number; auditEvents: number }; + storageUsage: { + assetCount: number; assetBytes: number; canvasCount: number; canvasBytes: number; + sessionCount: number; sessionBytes: number; taskCount: number; taskBytes: number; apiCallCount: number; + }; + storedFileBytes: number; + dailyUploadBytes: number; + quota: RuntimeResourcePolicy; }; export type AdminUserTask = { @@ -197,9 +211,62 @@ export type AdminOSSSetting = { updatedAt?: string; }; -export type RuntimeConcurrencySetting = { +export type RuntimeResourcePolicy = { + resourceUploadMB: number; + sessionUploadMB: number; + generatedFileMB: number; + dailyUploadMB: number; + storedFileGB: number; + structuredDataMB: number; + taskDataGB: number; + assetCount: number; + canvasCount: number; + sessionCount: number; + taskCount: number; + apiCallLogCount: number; +}; + +export type RuntimeTaskPolicy = { workerConcurrency: number; channelConcurrency: number; + activeTaskLimit: number; + imageTimeoutMinutes: number; + textTimeoutMinutes: number; + audioTimeoutMinutes: number; + videoTimeoutMinutes: number; + storyboardTimeoutMinutes: number; + defaultTimeoutMinutes: number; +}; + +export type RuntimeRequestPolicy = { + taskCreatePerMinute: number; + sessionCreatePerMinute: number; + resourceUploadPerMinute: number; + resourceImportPerMinute: number; + sessionFilePerMinute: number; + assetWritePerMinute: number; + canvasWritePerMinute: number; + registerPerHour: number; + emailCodePerHour: number; + loginIPPerTenMinutes: number; + loginAccountPerTenMinutes: number; + systemRelayPerMinute: number; + customRelayPerMinute: number; + customRelayConcurrency: number; + customRelayRequestMB: number; + customRelayResponseMB: number; + customRelayTimeoutMinutes: number; + systemRelayRequestMB: number; + systemRelayResponseMB: number; + channelCircuitFailureCount: number; + channelCircuitOpenSeconds: number; +}; + +export type RuntimePolicySetting = { + resource: RuntimeResourcePolicy; + task: RuntimeTaskPolicy; + request: RuntimeRequestPolicy; + configured?: boolean; updatedBy?: string; createdAt?: string; updatedAt?: string; @@ -329,12 +396,20 @@ 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 getAdminRuntimePolicySetting() { + return request<{ setting: RuntimePolicySetting }>(api.get("/admin/settings/runtime-policy")); } -export function updateAdminRuntimeConcurrencySetting(input: Pick) { - return request<{ setting: RuntimeConcurrencySetting }>(api.patch("/admin/settings/concurrency", input)); +export function getAdminSelfUseRuntimePolicy() { + return request<{ setting: RuntimePolicySetting }>(api.get("/admin/settings/runtime-policy/self-use")); +} + +export function updateAdminRuntimePolicySetting(input: Pick) { + return request<{ setting: RuntimePolicySetting }>(api.put("/admin/settings/runtime-policy", input)); +} + +export function resetAdminRuntimePolicySetting() { + return request<{ setting: RuntimePolicySetting }>(api.delete("/admin/settings/runtime-policy")); } export function listAdminApiLogs(params: AdminListParams = {}) { diff --git a/web/src/stores/use-user-store.ts b/web/src/stores/use-user-store.ts index b05726f3b..24c3baa9a 100644 --- a/web/src/stores/use-user-store.ts +++ b/web/src/stores/use-user-store.ts @@ -16,10 +16,18 @@ export type LocalUser = { updatedAt?: string; }; +export type RuntimeLimits = { + activeTaskLimit: number; + resourceUploadMB: number; + sessionUploadMB: number; +}; + type UserStore = { hydrated: boolean; user: LocalUser | null; + runtimeLimits: RuntimeLimits; setUser: (user: LocalUser | null) => void; + setRuntimeLimits: (limits?: RuntimeLimits) => void; setHydrated: (hydrated: boolean) => void; clearSession: () => void; }; @@ -27,7 +35,9 @@ type UserStore = { export const useUserStore = create()((set) => ({ hydrated: false, user: null, + runtimeLimits: { activeTaskLimit: 5, resourceUploadMB: 50, sessionUploadMB: 32 }, setUser: (user) => set({ user }), + setRuntimeLimits: (runtimeLimits) => set({ runtimeLimits: runtimeLimits || { activeTaskLimit: 5, resourceUploadMB: 50, sessionUploadMB: 32 } }), setHydrated: (hydrated) => set({ hydrated }), - clearSession: () => set({ user: null }), + clearSession: () => set({ user: null, runtimeLimits: { activeTaskLimit: 5, resourceUploadMB: 50, sessionUploadMB: 32 } }), }));