fix(platform): 渠道并发 - 支持后台热配置并等待空闲槽位 (#34)

Co-authored-by: ddCat <ddcat666@88.com>
2026-07-23 18:35:51 +08:00
parent f26f9983bc
commit 93316dd3d7
25 changed files with 703 additions and 112 deletions
--- a/backend/internal/handler/auth.go
+++ b/backend/internal/handler/auth.go
@@ -439,6 +439,37 @@ func RegisterAdminRoutes(r *gin.RouterGroup, svc *service.Service) {
 		}
 		ok(c, gin.H{"setting": setting})
 	})
+	r.GET("/admin/settings/concurrency", func(c *gin.Context) {
+		user, err := currentUser(c, svc)
+		if err != nil {
+			failService(c, err)
+			return
+		}
+		setting, err := svc.AdminRuntimeConcurrencySetting(user)
+		if err != nil {
+			failService(c, err)
+			return
+		}
+		ok(c, gin.H{"setting": setting})
+	})
+	r.PATCH("/admin/settings/concurrency", func(c *gin.Context) {
+		user, err := currentUser(c, svc)
+		if err != nil {
+			failService(c, err)
+			return
+		}
+		var req service.RuntimeConcurrencySettingRequest
+		if err := c.ShouldBindJSON(&req); err != nil {
+			fail(c, http.StatusBadRequest, err)
+			return
+		}
+		setting, err := svc.UpdateRuntimeConcurrencySetting(user, req)
+		if err != nil {
+			failService(c, err)
+			return
+		}
+		ok(c, gin.H{"setting": setting})
+	})
 	r.GET("/admin/api-logs", func(c *gin.Context) {
 		user, err := currentUser(c, svc)
 		if err != nil {
@@ -524,12 +555,35 @@ func proxySystemRequest(c *gin.Context, svc *service.Service, user *model.User,
 	}
 	modelName := proxyRequestModel(c.GetHeader("Content-Type"), body)
 	billingOrderID := ""
+	query := c.Request.URL.Query()
+	for _, key := range []string{"key", "api_key", "access_token", "token"} {
+		query.Del(key)
+	}
+	target := strings.TrimRight(channel.BaseURL, "/") + path
+	if encodedQuery := query.Encode(); encodedQuery != "" {
+		target += "?" + encodedQuery
+	}
+	if _, err := service.ValidateOutboundURL(target); err != nil {
+		_ = svc.RefundBilling(billingOrderID, "系统渠道地址校验失败")
+		failService(c, err)
+		return
+	}
+	// 同步代理与后台任务必须共享渠道槽位，否则两条入口会共同超过供应商并发上限。
+	releaseChannel, concurrencyLimit, err := svc.AcquireChannelSlot(c.Request.Context(), channel.ID, "", 36*time.Minute)
+	if err != nil {
+		log := apiCallLog(user, channel, billingOrderID, c.Request.Method, path, target, body, model.ApiCallStatusFailed, 0, time.Since(startedAt), err.Error(), concurrencyLimit)
+		log.ErrorCode, log.Error = service.ChannelSlotFailureDetails(err)
+		logSystemProxyCall(svc, log, nil)
+		fail(c, http.StatusServiceUnavailable, err)
+		return
+	}
+	defer releaseChannel()
 	if c.Request.Method == http.MethodPost {
 		capability := "text"
 		switch channel.InterfaceType {
 		case model.ChannelInterfaceOpenAIImage:
 			capability = "image"
-		case model.ChannelInterfaceNewAPIVideo, model.ChannelInterfaceNewAPIChannel1, model.ChannelInterfaceNewAPIChannel2:
+		case model.ChannelInterfaceNewAPIVideo, model.ChannelInterfaceNewAPIChannel1, model.ChannelInterfaceNewAPIChannel2, model.ChannelInterfaceXAIVideo:
 			capability = "video"
 		}
 		order, err := svc.ReserveProxyBilling(user.ID, channel.ID, strings.TrimPrefix(modelName, "models/"), capability, c.GetHeader("X-Canvas-Scene"), c.GetHeader("X-Idempotency-Key"))
@@ -544,19 +598,6 @@ func proxySystemRequest(c *gin.Context, svc *service.Service, user *model.User,
 			return
 		}
 	}
-	query := c.Request.URL.Query()
-	for _, key := range []string{"key", "api_key", "access_token", "token"} {
-		query.Del(key)
-	}
-	target := strings.TrimRight(channel.BaseURL, "/") + path
-	if encodedQuery := query.Encode(); encodedQuery != "" {
-		target += "?" + encodedQuery
-	}
-	if _, err := service.ValidateOutboundURL(target); err != nil {
-		_ = svc.RefundBilling(billingOrderID, "系统渠道地址校验失败")
-		failService(c, err)
-		return
-	}
 	upstreamReq, err := http.NewRequestWithContext(c.Request.Context(), c.Request.Method, target, bytes.NewReader(body))
 	if err != nil {
 		_ = svc.RefundBilling(billingOrderID, "系统渠道请求构造失败")
@@ -583,7 +624,7 @@ func proxySystemRequest(c *gin.Context, svc *service.Service, user *model.User,
 		status = model.ApiCallStatusFailed
 		errorText = err.Error()
 		_ = svc.MarkBillingUncertain(billingOrderID, "系统渠道连接中断，费用状态待核对")
-		logSystemProxyCall(svc, apiCallLog(user, channel, billingOrderID, c.Request.Method, path, target, body, status, statusCode, time.Since(startedAt), errorText), nil)
+		logSystemProxyCall(svc, apiCallLog(user, channel, billingOrderID, c.Request.Method, path, target, body, status, statusCode, time.Since(startedAt), errorText, concurrencyLimit), nil)
 		fail(c, http.StatusBadGateway, errors.New("系统渠道连接失败"))
 		return
 	}
@@ -597,7 +638,7 @@ func proxySystemRequest(c *gin.Context, svc *service.Service, user *model.User,
 		status = model.ApiCallStatusFailed
 		errorText = readErr.Error()
 		_ = svc.MarkBillingUncertain(billingOrderID, "系统渠道响应读取失败，费用状态待核对")
-		logSystemProxyCall(svc, apiCallLog(user, channel, billingOrderID, c.Request.Method, path, target, body, status, statusCode, time.Since(startedAt), errorText), nil)
+		logSystemProxyCall(svc, apiCallLog(user, channel, billingOrderID, c.Request.Method, path, target, body, status, statusCode, time.Since(startedAt), errorText, concurrencyLimit), nil)
 		fail(c, http.StatusBadGateway, errors.New("系统渠道响应读取失败"))
 		return
 	}
@@ -615,7 +656,7 @@ func proxySystemRequest(c *gin.Context, svc *service.Service, user *model.User,
 	} else {
 		_ = svc.RefundBilling(billingOrderID, "上游明确返回失败")
 	}
-	logSystemProxyCall(svc, apiCallLog(user, channel, billingOrderID, c.Request.Method, path, target, body, status, statusCode, time.Since(startedAt), errorText), responseBody)
+	logSystemProxyCall(svc, apiCallLog(user, channel, billingOrderID, c.Request.Method, path, target, body, status, statusCode, time.Since(startedAt), errorText, concurrencyLimit), responseBody)
 	for _, key := range []string{"Content-Type", "Cache-Control", "Content-Disposition"} {
 		if value := resp.Header.Get(key); value != "" {
 			c.Header(key, value)
@@ -625,12 +666,12 @@ func proxySystemRequest(c *gin.Context, svc *service.Service, user *model.User,
 	c.Data(resp.StatusCode, resp.Header.Get("Content-Type"), responseBody)
 }

-func apiCallLog(user *model.User, channel *model.ModelChannel, billingOrderID string, method string, path string, target string, body []byte, status model.ApiCallStatus, statusCode int, duration time.Duration, errorText string) model.ApiCallLog {
+func apiCallLog(user *model.User, channel *model.ModelChannel, billingOrderID string, method string, path string, target string, body []byte, status model.ApiCallStatus, statusCode int, duration time.Duration, errorText string, concurrencyLimit int) model.ApiCallLog {
 	capability := "text"
 	switch channel.InterfaceType {
 	case model.ChannelInterfaceOpenAIImage:
 		capability = "image"
-	case model.ChannelInterfaceNewAPIVideo, model.ChannelInterfaceNewAPIChannel1, model.ChannelInterfaceNewAPIChannel2:
+	case model.ChannelInterfaceNewAPIVideo, model.ChannelInterfaceNewAPIChannel1, model.ChannelInterfaceNewAPIChannel2, model.ChannelInterfaceXAIVideo:
 		capability = "video"
 	}
 	requestKind := "create"
@@ -641,22 +682,23 @@ func apiCallLog(user *model.User, channel *model.ModelChannel, billingOrderID st
 		}
 	}
 	return model.ApiCallLog{
-		UserID:         user.ID,
-		ChannelID:      channel.ID,
-		BillingOrderID: billingOrderID,
-		Source:         "system-channel",
-		Capability:     capability,
-		RequestKind:    requestKind,
-		Billable:       method == http.MethodPost,
-		APIFormat:      channel.APIFormat,
-		Method:         method,
-		Path:           path,
-		Model:          readPayloadModel(body),
-		Status:         status,
-		StatusCode:     statusCode,
-		DurationMs:     duration.Milliseconds(),
-		Error:          errorText,
-		UpstreamURL:    target,
+		UserID:           user.ID,
+		ChannelID:        channel.ID,
+		BillingOrderID:   billingOrderID,
+		Source:           "system-channel",
+		Capability:       capability,
+		RequestKind:      requestKind,
+		Billable:         method == http.MethodPost,
+		APIFormat:        channel.APIFormat,
+		Method:           method,
+		Path:             path,
+		Model:            readPayloadModel(body),
+		Status:           status,
+		StatusCode:       statusCode,
+		DurationMs:       duration.Milliseconds(),
+		Error:            errorText,
+		ConcurrencyLimit: concurrencyLimit,
+		UpstreamURL:      target,
 	}
 }

--- a/backend/internal/model/models.go
+++ b/backend/internal/model/models.go
@@ -147,18 +147,19 @@ type EmailVerificationCode struct {
 }

 type ModelChannel struct {
-	ID            string               `json:"id" gorm:"primaryKey;size:36"`
-	UserID        string               `json:"userId" gorm:"index;size:36"`
-	Scope         ChannelScope         `json:"scope" gorm:"index;size:24"`
-	Enabled       bool                 `json:"enabled" gorm:"index"`
-	Name          string               `json:"name" gorm:"size:80"`
-	BaseURL       string               `json:"baseUrl"`
-	APIKey        string               `json:"-"`
-	APIFormat     string               `json:"apiFormat" gorm:"size:24"`
-	InterfaceType ChannelInterfaceType `json:"interfaceType" gorm:"size:32"`
-	ModelsJSON    string               `json:"modelsJson" gorm:"type:text"`
-	CreatedAt     time.Time            `json:"createdAt"`
-	UpdatedAt     time.Time            `json:"updatedAt"`
+	ID               string               `json:"id" gorm:"primaryKey;size:36"`
+	UserID           string               `json:"userId" gorm:"index;size:36"`
+	Scope            ChannelScope         `json:"scope" gorm:"index;size:24"`
+	Enabled          bool                 `json:"enabled" gorm:"index"`
+	Name             string               `json:"name" gorm:"size:80"`
+	BaseURL          string               `json:"baseUrl"`
+	APIKey           string               `json:"-"`
+	APIFormat        string               `json:"apiFormat" gorm:"size:24"`
+	InterfaceType    ChannelInterfaceType `json:"interfaceType" gorm:"size:32"`
+	ConcurrencyLimit int                  `json:"concurrencyLimit"`
+	ModelsJSON       string               `json:"modelsJson" gorm:"type:text"`
+	CreatedAt        time.Time            `json:"createdAt"`
+	UpdatedAt        time.Time            `json:"updatedAt"`
 }

 type ChannelModel struct {
@@ -207,6 +208,7 @@ type ApiCallLog struct {
 	Currency            string        `json:"currency" gorm:"size:12"`
 	ErrorCode           string        `json:"errorCode,omitempty" gorm:"index;size:80"`
 	Error               string        `json:"error"`
+	ConcurrencyLimit    int           `json:"concurrencyLimit"`
 	UpstreamURL         string        `json:"upstreamUrl"`
 	CreatedAt           time.Time     `json:"createdAt" gorm:"index;index:idx_api_logs_user_created,priority:2;index:idx_api_logs_channel_created,priority:2;index:idx_api_logs_model_created,priority:2;index:idx_api_logs_status_created,priority:2"`
 }
--- a/backend/internal/service/admin.go
+++ b/backend/internal/service/admin.go
@@ -75,29 +75,32 @@ type AdminReferenceData struct {
 }

 type ChannelRequest struct {
-	Name          string   `json:"name"`
-	BaseURL       string   `json:"baseUrl"`
-	APIKey        string   `json:"apiKey"`
-	InterfaceType string   `json:"interfaceType"`
-	Models        []string `json:"models"`
-	Enabled       *bool    `json:"enabled"`
+	Name                 string   `json:"name"`
+	BaseURL              string   `json:"baseUrl"`
+	APIKey               string   `json:"apiKey"`
+	InterfaceType        string   `json:"interfaceType"`
+	ConcurrencyLimit     *int     `json:"concurrencyLimit"`
+	UseGlobalConcurrency *bool    `json:"useGlobalConcurrency"`
+	Models               []string `json:"models"`
+	Enabled              *bool    `json:"enabled"`
 }

 type PublicModelChannel struct {
-	ID            string                     `json:"id"`
-	UserID        string                     `json:"userId"`
-	Scope         model.ChannelScope         `json:"scope"`
-	Enabled       bool                       `json:"enabled"`
-	Name          string                     `json:"name"`
-	BaseURL       string                     `json:"baseUrl"`
-	APIKey        string                     `json:"apiKey"`
-	APIFormat     string                     `json:"apiFormat"`
-	InterfaceType model.ChannelInterfaceType `json:"interfaceType"`
-	Models        []string                   `json:"models"`
-	ModelCosts    []PublicChannelModelPrice  `json:"modelCosts"`
-	HasAPIKey     bool                       `json:"hasApiKey"`
-	CreatedAt     time.Time                  `json:"createdAt"`
-	UpdatedAt     time.Time                  `json:"updatedAt"`
+	ID               string                     `json:"id"`
+	UserID           string                     `json:"userId"`
+	Scope            model.ChannelScope         `json:"scope"`
+	Enabled          bool                       `json:"enabled"`
+	Name             string                     `json:"name"`
+	BaseURL          string                     `json:"baseUrl"`
+	APIKey           string                     `json:"apiKey"`
+	APIFormat        string                     `json:"apiFormat"`
+	InterfaceType    model.ChannelInterfaceType `json:"interfaceType"`
+	ConcurrencyLimit int                        `json:"concurrencyLimit"`
+	Models           []string                   `json:"models"`
+	ModelCosts       []PublicChannelModelPrice  `json:"modelCosts"`
+	HasAPIKey        bool                       `json:"hasApiKey"`
+	CreatedAt        time.Time                  `json:"createdAt"`
+	UpdatedAt        time.Time                  `json:"updatedAt"`
 }

 type PublicChannelModelPrice struct {
@@ -530,6 +533,16 @@ func channelFromRequest(req ChannelRequest, channel model.ModelChannel) (model.M
 	// 系统渠道均由后端按已声明的接口类型分发，调用格式固定为 Bearer/OpenAI 兼容鉴权。
 	channel.APIFormat = "openai"
 	channel.InterfaceType = interfaceType
+	if req.UseGlobalConcurrency != nil && *req.UseGlobalConcurrency {
+		channel.ConcurrencyLimit = 0
+	} else if req.ConcurrencyLimit != nil {
+		if *req.ConcurrencyLimit < minChannelConcurrencyLimit || *req.ConcurrencyLimit > maxChannelConcurrencyLimit {
+			return channel, BadAuthRequest("最大并发数必须是 1-100 的整数")
+		}
+		channel.ConcurrencyLimit = *req.ConcurrencyLimit
+	} else if req.UseGlobalConcurrency != nil {
+		return channel, BadAuthRequest("请填写渠道最大并发数")
+	}
 	channel.ModelsJSON = string(modelsJSON)
 	if req.Enabled != nil {
 		channel.Enabled = *req.Enabled
@@ -611,20 +624,21 @@ func publicChannel(channel model.ModelChannel, admin bool, channelModels []model
 		interfaceType = inferChannelInterfaceType(models)
 	}
 	return PublicModelChannel{
-		ID:            channel.ID,
-		UserID:        channel.UserID,
-		Scope:         channel.Scope,
-		Enabled:       channel.Enabled,
-		Name:          channel.Name,
-		BaseURL:       baseURL,
-		APIKey:        apiKey,
-		APIFormat:     channel.APIFormat,
-		InterfaceType: interfaceType,
-		Models:        models,
-		ModelCosts:    modelCosts,
-		HasAPIKey:     strings.TrimSpace(channel.APIKey) != "",
-		CreatedAt:     channel.CreatedAt,
-		UpdatedAt:     channel.UpdatedAt,
+		ID:               channel.ID,
+		UserID:           channel.UserID,
+		Scope:            channel.Scope,
+		Enabled:          channel.Enabled,
+		Name:             channel.Name,
+		BaseURL:          baseURL,
+		APIKey:           apiKey,
+		APIFormat:        channel.APIFormat,
+		InterfaceType:    interfaceType,
+		ConcurrencyLimit: channel.ConcurrencyLimit,
+		Models:           models,
+		ModelCosts:       modelCosts,
+		HasAPIKey:        strings.TrimSpace(channel.APIKey) != "",
+		CreatedAt:        channel.CreatedAt,
+		UpdatedAt:        channel.UpdatedAt,
 	}
 }

--- a/backend/internal/service/admin_channel_test.go
+++ b/backend/internal/service/admin_channel_test.go
@@ -14,11 +14,12 @@ func TestChannelFromRequestStoresInterfaceType(t *testing.T) {
 	defer server.Close()

 	channel, err := channelFromRequest(ChannelRequest{
-		Name:          "NewAPI 渠道 1",
-		BaseURL:       server.URL + "/v1",
-		APIKey:        "secret",
-		InterfaceType: "newapi-channel-1",
-		Models:        []string{"seedance-2.0"},
+		Name:             "NewAPI 渠道 1",
+		BaseURL:          server.URL + "/v1",
+		APIKey:           "secret",
+		InterfaceType:    "newapi-channel-1",
+		ConcurrencyLimit: intPtr(6),
+		Models:           []string{"seedance-2.0"},
 	}, model.ModelChannel{})
 	if err != nil {
 		t.Fatalf("channelFromRequest() error = %v", err)
@@ -29,6 +30,9 @@ func TestChannelFromRequestStoresInterfaceType(t *testing.T) {
 	if channel.APIFormat != "openai" {
 		t.Fatalf("APIFormat = %q, want openai", channel.APIFormat)
 	}
+	if channel.ConcurrencyLimit != 6 {
+		t.Fatalf("ConcurrencyLimit = %d, want 6", channel.ConcurrencyLimit)
+	}
 }

 func TestChannelFromRequestStoresXAIVideoInterfaceType(t *testing.T) {
@@ -70,3 +74,29 @@ func TestChannelFromRequestRejectsUnknownInterfaceType(t *testing.T) {
 		t.Fatal("channelFromRequest() error = nil")
 	}
 }
+
+func TestChannelFromRequestRejectsInvalidConcurrencyLimit(t *testing.T) {
+	for _, limit := range []int{0, 101} {
+		_, err := channelFromRequest(ChannelRequest{Name: "Bad", BaseURL: "https://example.com/v1", InterfaceType: "newapi", ConcurrencyLimit: &limit}, model.ModelChannel{})
+		if err == nil {
+			t.Fatalf("channelFromRequest() concurrencyLimit = %d, error = nil", limit)
+		}
+	}
+}
+
+func TestRuntimeConcurrencyUsesEnvironmentFallback(t *testing.T) {
+	t.Setenv("CANVAS_CHANNEL_CONCURRENCY", "7")
+	t.Setenv("CANVAS_WORKER_CONCURRENCY", "9")
+	setting := runtimeConcurrencyFromEnvironment()
+	if setting.ChannelConcurrency != 7 || setting.WorkerConcurrency != 9 {
+		t.Fatalf("runtimeConcurrencyFromEnvironment() = %#v", setting)
+	}
+
+	useGlobal := true
+	channel, err := channelFromRequest(ChannelRequest{Name: "Global", BaseURL: "https://example.com/v1", InterfaceType: "newapi", UseGlobalConcurrency: &useGlobal}, model.ModelChannel{ConcurrencyLimit: 4})
+	if err != nil || channel.ConcurrencyLimit != 0 {
+		t.Fatalf("global concurrency channel = %#v, error = %v", channel, err)
+	}
+}
+
+func intPtr(value int) *int { return &value }
--- a/backend/internal/service/analytics.go
+++ b/backend/internal/service/analytics.go
@@ -245,9 +245,9 @@ func (s *Service) AdminAPICallLogsCSV(actor *model.User, query APICallLogQuery)
 	var buffer bytes.Buffer
 	buffer.WriteString("\xEF\xBB\xBF")
 	writer := csv.NewWriter(&buffer)
-	_ = writer.Write([]string{"时间", "用户ID", "渠道ID", "任务ID", "计费单ID", "能力", "请求阶段", "模型", "状态", "HTTP状态", "耗时毫秒", "供应商任务ID", "错误码", "错误"})
+	_ = writer.Write([]string{"时间", "用户ID", "渠道ID", "任务ID", "计费单ID", "能力", "请求阶段", "模型", "状态", "HTTP状态", "耗时毫秒", "渠道并发上限", "供应商任务ID", "错误码", "错误"})
 	for _, log := range logs {
-		_ = writer.Write([]string{log.CreatedAt.UTC().Format(time.RFC3339), log.UserID, log.ChannelID, log.TaskID, log.BillingOrderID, log.Capability, log.RequestKind, log.Model, string(log.Status), strconv.Itoa(log.StatusCode), strconv.FormatInt(log.DurationMs, 10), log.ProviderRequestID, log.ErrorCode, log.Error})
+		_ = writer.Write([]string{log.CreatedAt.UTC().Format(time.RFC3339), log.UserID, log.ChannelID, log.TaskID, log.BillingOrderID, log.Capability, log.RequestKind, log.Model, string(log.Status), strconv.Itoa(log.StatusCode), strconv.FormatInt(log.DurationMs, 10), strconv.Itoa(log.ConcurrencyLimit), log.ProviderRequestID, log.ErrorCode, log.Error})
 	}
 	writer.Flush()
 	if err := writer.Error(); err != nil {
--- a/backend/internal/service/concurrency_settings.go
+++ b/backend/internal/service/concurrency_settings.go
@@ -0,0 +1,122 @@
+package service
+
+import (
+	"encoding/json"
+	"errors"
+	"strings"
+	"time"
+
+	"infinite-canvas/backend/internal/model"
+
+	"gorm.io/gorm"
+)
+
+const runtimeConcurrencySettingKey = "runtime_concurrency"
+
+type RuntimeConcurrencySettingRequest struct {
+	WorkerConcurrency  int `json:"workerConcurrency"`
+	ChannelConcurrency int `json:"channelConcurrency"`
+}
+
+type PublicRuntimeConcurrencySetting struct {
+	WorkerConcurrency  int       `json:"workerConcurrency"`
+	ChannelConcurrency int       `json:"channelConcurrency"`
+	UpdatedBy          string    `json:"updatedBy"`
+	CreatedAt          time.Time `json:"createdAt"`
+	UpdatedAt          time.Time `json:"updatedAt"`
+}
+
+type runtimeConcurrencySettingValue struct {
+	WorkerConcurrency  int `json:"workerConcurrency"`
+	ChannelConcurrency int `json:"channelConcurrency"`
+}
+
+func (s *Service) AdminRuntimeConcurrencySetting(actor *model.User) (*PublicRuntimeConcurrencySetting, error) {
+	if err := s.RequireAdmin(actor); err != nil {
+		return nil, err
+	}
+	setting, value, err := s.readRuntimeConcurrencySetting()
+	if err != nil {
+		return nil, err
+	}
+	return publicRuntimeConcurrencySetting(setting, value), nil
+}
+
+func (s *Service) UpdateRuntimeConcurrencySetting(actor *model.User, req RuntimeConcurrencySettingRequest) (*PublicRuntimeConcurrencySetting, error) {
+	if err := s.RequireAdmin(actor); err != nil {
+		return nil, err
+	}
+	value := runtimeConcurrencySettingValue{WorkerConcurrency: req.WorkerConcurrency, ChannelConcurrency: req.ChannelConcurrency}
+	if err := validateRuntimeConcurrencySetting(value); err != nil {
+		return nil, err
+	}
+	encoded, err := json.Marshal(value)
+	if err != nil {
+		return nil, err
+	}
+	current, _, err := s.readRuntimeConcurrencySetting()
+	if err != nil {
+		return nil, err
+	}
+	setting := model.SystemSetting{Key: runtimeConcurrencySettingKey, ValueJSON: string(encoded), UpdatedBy: actor.ID}
+	if current != nil {
+		setting.CreatedAt = current.CreatedAt
+	}
+	if err := s.repo.SaveSystemSetting(&setting); err != nil {
+		return nil, err
+	}
+	if err := s.appendAdminAudit(actor, "runtime_concurrency.update", "system_setting", runtimeConcurrencySettingKey, "更新任务并发配置", value); err != nil {
+		return nil, err
+	}
+	return publicRuntimeConcurrencySetting(&setting, value), nil
+}
+
+func (s *Service) runtimeConcurrencySetting() (runtimeConcurrencySettingValue, error) {
+	_, value, err := s.readRuntimeConcurrencySetting()
+	return value, err
+}
+
+func (s *Service) readRuntimeConcurrencySetting() (*model.SystemSetting, runtimeConcurrencySettingValue, error) {
+	setting, err := s.repo.SystemSetting(runtimeConcurrencySettingKey)
+	if errors.Is(err, gorm.ErrRecordNotFound) {
+		return nil, runtimeConcurrencyFromEnvironment(), nil
+	}
+	if err != nil {
+		return nil, runtimeConcurrencySettingValue{}, err
+	}
+	value := runtimeConcurrencySettingValue{}
+	if strings.TrimSpace(setting.ValueJSON) == "" || json.Unmarshal([]byte(setting.ValueJSON), &value) != nil {
+		return nil, runtimeConcurrencySettingValue{}, errors.New("任务并发配置格式无效")
+	}
+	if err := validateRuntimeConcurrencySetting(value); err != nil {
+		return nil, runtimeConcurrencySettingValue{}, err
+	}
+	return setting, value, nil
+}
+
+func runtimeConcurrencyFromEnvironment() runtimeConcurrencySettingValue {
+	return runtimeConcurrencySettingValue{
+		WorkerConcurrency:  effectiveChannelConcurrencyLimit(envInt("CANVAS_WORKER_CONCURRENCY", taskWorkerConcurrency)),
+		ChannelConcurrency: defaultChannelConcurrencyLimit(),
+	}
+}
+
+func validateRuntimeConcurrencySetting(value runtimeConcurrencySettingValue) error {
+	if value.WorkerConcurrency < minChannelConcurrencyLimit || value.WorkerConcurrency > maxChannelConcurrencyLimit {
+		return BadAuthRequest("Worker 并发数必须是 1-100 的整数")
+	}
+	if value.ChannelConcurrency < minChannelConcurrencyLimit || value.ChannelConcurrency > maxChannelConcurrencyLimit {
+		return BadAuthRequest("全局渠道并发数必须是 1-100 的整数")
+	}
+	return nil
+}
+
+func publicRuntimeConcurrencySetting(setting *model.SystemSetting, value runtimeConcurrencySettingValue) *PublicRuntimeConcurrencySetting {
+	result := &PublicRuntimeConcurrencySetting{WorkerConcurrency: value.WorkerConcurrency, ChannelConcurrency: value.ChannelConcurrency}
+	if setting != nil {
+		result.UpdatedBy = setting.UpdatedBy
+		result.CreatedAt = setting.CreatedAt
+		result.UpdatedAt = setting.UpdatedAt
+	}
+	return result
+}
--- a/backend/internal/service/concurrency_settings_test.go
+++ b/backend/internal/service/concurrency_settings_test.go
@@ -0,0 +1,63 @@
+package service
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"infinite-canvas/backend/internal/model"
+	"infinite-canvas/backend/internal/repository"
+
+	"gorm.io/driver/sqlite"
+	"gorm.io/gorm"
+)
+
+func TestRuntimeConcurrencySettingPersistsAndChannelOverrideWins(t *testing.T) {
+	t.Setenv("REDIS_URL", "")
+	t.Setenv("CANVAS_WORKER_CONCURRENCY", "3")
+	t.Setenv("CANVAS_CHANNEL_CONCURRENCY", "3")
+	db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if sqlDB, err := db.DB(); err == nil {
+		sqlDB.SetMaxOpenConns(1)
+	}
+	if err := db.AutoMigrate(&model.SystemSetting{}, &model.AdminAuditEvent{}, &model.ModelChannel{}); err != nil {
+		t.Fatal(err)
+	}
+	svc := New(repository.New(db), t.TempDir())
+	actor := &model.User{ID: "admin-1", Role: model.UserRoleAdmin}
+	setting, err := svc.UpdateRuntimeConcurrencySetting(actor, RuntimeConcurrencySettingRequest{WorkerConcurrency: 8, ChannelConcurrency: 6})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if setting.WorkerConcurrency != 8 || setting.ChannelConcurrency != 6 {
+		t.Fatalf("setting = %#v", setting)
+	}
+
+	channel := model.ModelChannel{ID: "channel-1", Scope: model.ChannelScopeSystem, Enabled: true, ConcurrencyLimit: 0}
+	if err := db.Create(&channel).Error; err != nil {
+		t.Fatal(err)
+	}
+	release, limit, err := svc.AcquireChannelSlot(context.Background(), channel.ID, "", time.Minute)
+	if err != nil {
+		t.Fatal(err)
+	}
+	release()
+	if limit != 6 {
+		t.Fatalf("global channel limit = %d, want 6", limit)
+	}
+
+	if err := db.Model(&channel).Update("concurrency_limit", 9).Error; err != nil {
+		t.Fatal(err)
+	}
+	release, limit, err = svc.AcquireChannelSlot(context.Background(), channel.ID, "", time.Minute)
+	if err != nil {
+		t.Fatal(err)
+	}
+	release()
+	if limit != 9 {
+		t.Fatalf("overridden channel limit = %d, want 9", limit)
+	}
+}
--- a/backend/internal/service/coordination.go
+++ b/backend/internal/service/coordination.go
@@ -18,11 +18,50 @@ type localRateEntry struct {
 	count   int
 }

+const (
+	minChannelConcurrencyLimit     = 1
+	maxChannelConcurrencyLimit     = 100
+	defaultChannelConcurrencyValue = 3
+)
+
+type channelSlotError struct {
+	scope string
+	limit int
+	err   error
+}
+
+func (e channelSlotError) Error() string {
+	if errors.Is(e.err, context.DeadlineExceeded) {
+		return fmt.Sprintf("等待渠道并发槽位超时（渠道 %s，并发上限 %d）", e.scope, e.limit)
+	}
+	if errors.Is(e.err, context.Canceled) {
+		return fmt.Sprintf("等待渠道并发槽位已取消（渠道 %s，并发上限 %d）", e.scope, e.limit)
+	}
+	return fmt.Sprintf("获取渠道并发配额失败（渠道 %s，并发上限 %d）：%v", e.scope, e.limit, e.err)
+}
+
+func (e channelSlotError) Unwrap() error { return e.err }
+
+func ChannelSlotFailureDetails(err error) (string, string) {
+	var slotErr channelSlotError
+	if !errors.As(err, &slotErr) {
+		return "", ""
+	}
+	if errors.Is(slotErr, context.DeadlineExceeded) {
+		return "channel_concurrency_wait_timeout", slotErr.Error()
+	}
+	if errors.Is(slotErr, context.Canceled) {
+		return "channel_concurrency_wait_cancelled", slotErr.Error()
+	}
+	return "channel_concurrency_unavailable", slotErr.Error()
+}
+
 type runtimeCoordinator struct {
 	redis      *redis.Client
 	instanceID string
 	localMu    sync.Mutex
 	localRate  map[string]localRateEntry
+	localSlots map[string]map[string]time.Time
 }

 var fixedWindowScript = redis.NewScript(`
@@ -40,7 +79,7 @@ return 1
 `)

 func newRuntimeCoordinator(dialect string) (*runtimeCoordinator, error) {
-	coordinator := &runtimeCoordinator{instanceID: newID(), localRate: map[string]localRateEntry{}}
+	coordinator := &runtimeCoordinator{instanceID: newID(), localRate: map[string]localRateEntry{}, localSlots: map[string]map[string]time.Time{}}
 	redisURL := strings.TrimSpace(os.Getenv("REDIS_URL"))
 	if redisURL == "" {
 		if dialect == "postgres" {
@@ -84,7 +123,30 @@ func (c *runtimeCoordinator) allow(ctx context.Context, key string, limit int, w

 func (c *runtimeCoordinator) acquire(ctx context.Context, scope string, limit int, ttl time.Duration) (func(), bool, error) {
 	if c.redis == nil {
-		return func() {}, true, nil
+		c.localMu.Lock()
+		now := time.Now()
+		slots := c.localSlots[scope]
+		if slots == nil {
+			slots = map[string]time.Time{}
+			c.localSlots[scope] = slots
+		}
+		for token, expiresAt := range slots {
+			if !expiresAt.After(now) {
+				delete(slots, token)
+			}
+		}
+		if len(slots) >= limit {
+			c.localMu.Unlock()
+			return nil, false, nil
+		}
+		token := c.instanceID + ":" + newID()
+		slots[token] = now.Add(ttl)
+		c.localMu.Unlock()
+		return func() {
+			c.localMu.Lock()
+			delete(c.localSlots[scope], token)
+			c.localMu.Unlock()
+		}, true, nil
 	}
 	// 有过期分数的有序集合避免实例崩溃后永久占槽，业务数据库仍保存任务与账本真相。
 	key := "canvas:slots:" + scope
@@ -97,6 +159,28 @@ func (c *runtimeCoordinator) acquire(ctx context.Context, scope string, limit in
 	return func() { _ = c.redis.ZRem(context.Background(), key, token).Err() }, true, nil
 }

+func (c *runtimeCoordinator) acquireWithWait(ctx context.Context, scope string, limit int, ttl time.Duration) (func(), error) {
+	ticker := time.NewTicker(200 * time.Millisecond)
+	defer ticker.Stop()
+	for {
+		if err := ctx.Err(); err != nil {
+			return nil, err
+		}
+		release, acquired, err := c.acquire(ctx, scope, limit, ttl)
+		if err != nil {
+			return nil, err
+		}
+		if acquired {
+			return release, nil
+		}
+		select {
+		case <-ctx.Done():
+			return nil, ctx.Err()
+		case <-ticker.C:
+		}
+	}
+}
+
 func (c *runtimeCoordinator) circuitOpen(ctx context.Context, channelID string) (bool, error) {
 	if c.redis == nil || strings.TrimSpace(channelID) == "" {
 		return false, nil
@@ -133,6 +217,52 @@ func envInt(key string, fallback int) int {
 	return value
 }

+func defaultChannelConcurrencyLimit() int {
+	return effectiveChannelConcurrencyLimit(envInt("CANVAS_CHANNEL_CONCURRENCY", defaultChannelConcurrencyValue))
+}
+
+func effectiveChannelConcurrencyLimit(configured int) int {
+	if configured < minChannelConcurrencyLimit || configured > maxChannelConcurrencyLimit {
+		return defaultChannelConcurrencyValue
+	}
+	return configured
+}
+
+func (s *Service) AcquireChannelSlot(ctx context.Context, channelID string, fallbackScope string, ttl time.Duration) (func(), int, error) {
+	setting, err := s.runtimeConcurrencySetting()
+	limit := defaultChannelConcurrencyLimit()
+	if err != nil {
+		return nil, limit, channelSlotError{scope: firstNonEmpty(strings.TrimSpace(channelID), strings.TrimSpace(fallbackScope), "unknown"), limit: limit, err: fmt.Errorf("读取全局并发配置失败：%w", err)}
+	}
+	limit = setting.ChannelConcurrency
+	scope := strings.TrimSpace(channelID)
+	if scope != "" {
+		channel, err := s.repo.SystemChannel(scope)
+		if err != nil {
+			return nil, limit, channelSlotError{scope: scope, limit: limit, err: fmt.Errorf("读取渠道并发配置失败：%w", err)}
+		}
+		if channel.ConcurrencyLimit > 0 {
+			if channel.ConcurrencyLimit < minChannelConcurrencyLimit || channel.ConcurrencyLimit > maxChannelConcurrencyLimit {
+				return nil, limit, channelSlotError{scope: scope, limit: limit, err: errors.New("渠道并发配置超出 1-100 范围")}
+			}
+			limit = channel.ConcurrencyLimit
+		}
+	} else {
+		scope = strings.TrimSpace(fallbackScope)
+	}
+	if scope == "" {
+		return nil, limit, channelSlotError{scope: "unknown", limit: limit, err: errors.New("渠道并发范围为空")}
+	}
+	if s.coordinator == nil {
+		return nil, limit, channelSlotError{scope: scope, limit: limit, err: errors.New("运行时协调器未初始化")}
+	}
+	release, err := s.coordinator.acquireWithWait(ctx, "channel:"+scope, limit, ttl)
+	if err != nil {
+		return nil, limit, channelSlotError{scope: scope, limit: limit, err: err}
+	}
+	return release, limit, nil
+}
+
 func (s *Service) ValidateRuntime() error {
 	return s.runtimeErr
 }
--- a/backend/internal/service/coordination_test.go
+++ b/backend/internal/service/coordination_test.go
@@ -0,0 +1,54 @@
+package service
+
+import (
+	"context"
+	"testing"
+	"time"
+)
+
+func TestRuntimeCoordinatorWaitsUntilChannelSlotIsReleased(t *testing.T) {
+	coordinator := &runtimeCoordinator{instanceID: "test", localRate: map[string]localRateEntry{}, localSlots: map[string]map[string]time.Time{}}
+	releaseFirst, acquired, err := coordinator.acquire(context.Background(), "channel:one", 1, time.Minute)
+	if err != nil || !acquired {
+		t.Fatalf("first acquire = (%v, %v), want acquired", acquired, err)
+	}
+
+	result := make(chan error, 1)
+	go func() {
+		releaseSecond, waitErr := coordinator.acquireWithWait(context.Background(), "channel:one", 1, time.Minute)
+		if waitErr == nil {
+			releaseSecond()
+		}
+		result <- waitErr
+	}()
+
+	select {
+	case err := <-result:
+		t.Fatalf("second acquire returned before release: %v", err)
+	case <-time.After(50 * time.Millisecond):
+	}
+	releaseFirst()
+	select {
+	case err := <-result:
+		if err != nil {
+			t.Fatalf("second acquire after release: %v", err)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("second acquire did not resume after release")
+	}
+}
+
+func TestRuntimeCoordinatorStopsWaitingWhenContextIsCancelled(t *testing.T) {
+	coordinator := &runtimeCoordinator{instanceID: "test", localRate: map[string]localRateEntry{}, localSlots: map[string]map[string]time.Time{}}
+	release, acquired, err := coordinator.acquire(context.Background(), "channel:one", 1, time.Minute)
+	if err != nil || !acquired {
+		t.Fatalf("first acquire = (%v, %v), want acquired", acquired, err)
+	}
+	defer release()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	if _, err := coordinator.acquireWithWait(ctx, "channel:one", 1, time.Minute); err == nil {
+		t.Fatal("acquireWithWait() error = nil after cancellation")
+	}
+}
--- a/backend/internal/service/provider.go
+++ b/backend/internal/service/provider.go
@@ -97,6 +97,7 @@ type providerAnalyticsContext struct {
 	VideoSeconds      int
 	RequestKind       string
 	ProviderRequestID string
+	ConcurrencyLimit  int
 }

 func withProviderAnalytics(ctx context.Context, service *Service, task model.Task) context.Context {
@@ -1274,17 +1275,17 @@ func doBinary(req *http.Request) ([]byte, string, error) {
 		if open {
 			return nil, "", errors.New("当前渠道连续失败，已暂时熔断，请稍后重试")
 		}
-		var acquired bool
 		slotID := channelID
 		if slotID == "" {
 			slotID = "custom:" + strings.ToLower(req.URL.Host)
 		}
-		release, acquired, err = coordinator.acquire(req.Context(), "channel:"+slotID, envInt("CANVAS_CHANNEL_CONCURRENCY", 2), providerHTTPTimeout+time.Minute)
+		var concurrencyLimit int
+		release, concurrencyLimit, err = metadata.Service.AcquireChannelSlot(req.Context(), channelID, slotID, providerHTTPTimeout+time.Minute)
+		metadata.ConcurrencyLimit = concurrencyLimit
+		req = req.WithContext(context.WithValue(req.Context(), providerAnalyticsKey{}, metadata))
 		if err != nil {
-			return nil, "", fmt.Errorf("获取渠道并发配额失败：%w", err)
-		}
-		if !acquired {
-			return nil, "", errors.New("当前渠道并发已满，请稍后重试")
+			recordProviderRequest(req, startedAt, 0, nil, err)
+			return nil, "", err
 		}
 		defer release()
 	}
@@ -1356,7 +1357,13 @@ func recordProviderRequest(req *http.Request, startedAt time.Time, statusCode in
 		RequestKind: requestKind, Billable: req.Method == http.MethodPost,
 		APIFormat: "openai", Method: req.Method, Path: req.URL.Path, Model: metadata.Model,
 		Status: status, StatusCode: statusCode, DurationMs: time.Since(startedAt).Milliseconds(),
-		Error: errorText, UpstreamURL: req.URL.Scheme + "://" + req.URL.Host + req.URL.Path,
+		Error: errorText, ConcurrencyLimit: metadata.ConcurrencyLimit, UpstreamURL: req.URL.Scheme + "://" + req.URL.Host + req.URL.Path,
+	}
+	channelSlotFailure := false
+	if code, message := ChannelSlotFailureDetails(requestErr); code != "" {
+		channelSlotFailure = true
+		log.ErrorCode = code
+		log.Error = message
 	}
 	if requestKind == "create" && metadata.Capability == "video" {
 		log.VideoSeconds = metadata.VideoSeconds
@@ -1370,7 +1377,9 @@ func recordProviderRequest(req *http.Request, startedAt time.Time, statusCode in
 	}
 	metadata.Service.EnrichAPICallLog(&log, responseBody)
 	if err := metadata.Service.LogAPICall(log); err != nil {
-		_ = metadata.Service.MarkBillingUncertain(metadata.BillingOrderID, "上游调用日志写入失败，费用状态待核对")
+		if !channelSlotFailure {
+			_ = metadata.Service.MarkBillingUncertain(metadata.BillingOrderID, "上游调用日志写入失败，费用状态待核对")
+		}
 	}
 }

--- a/backend/internal/service/service.go
+++ b/backend/internal/service/service.go
@@ -160,10 +160,14 @@ func New(repo *repository.Repository, dataDir string) *Service {

 func (s *Service) StartWorker() {
 	go func() {
-		workerConcurrency := envInt("CANVAS_WORKER_CONCURRENCY", taskWorkerConcurrency)
-		slots := make(chan struct{}, workerConcurrency)
+		slots := make(chan struct{}, maxChannelConcurrencyLimit)
 		dispatch := func() {
-			for len(slots) < cap(slots) {
+			setting, err := s.runtimeConcurrencySetting()
+			if err != nil {
+				return
+			}
+			workerConcurrency := setting.WorkerConcurrency
+			for len(slots) < workerConcurrency {
 				releaseGlobal, acquired, err := s.coordinator.acquire(context.Background(), "workers", workerConcurrency, 45*time.Minute)
 				if err != nil || !acquired {
 					return
@@ -697,6 +701,10 @@ func (s *Service) processClaimedTask(task *model.Task) error {
 		result, err = s.persistGeneratedMediaResult(task.UserID, result)
 	}
 	if err != nil {
+		channelSlotFailedBeforeRequest := false
+		if code, _ := ChannelSlotFailureDetails(err); code != "" {
+			channelSlotFailedBeforeRequest = true
+		}
 		select {
 		case leaseErr := <-leaseLost:
 			_ = s.log(task.UserID, task.ID, "warn", "任务租约失效，等待其他 worker 恢复", leaseErr.Error())
@@ -709,7 +717,11 @@ func (s *Service) processClaimedTask(task *model.Task) error {
 			task.Error = "任务已取消"
 			task.CompletedAt = ptr(time.Now())
 			_ = s.repo.Save(task)
-			_ = s.MarkBillingUncertain(task.BillingOrderID, "任务取消时上游费用状态不明确")
+			if channelSlotFailedBeforeRequest {
+				_ = s.RefundBilling(task.BillingOrderID, "等待渠道槽位期间取消，上游请求未发出")
+			} else {
+				_ = s.MarkBillingUncertain(task.BillingOrderID, "任务取消时上游费用状态不明确")
+			}
 			_ = s.markSessionFailed(*task, "会话任务已取消。")
 			_ = s.log(task.UserID, task.ID, "warn", "任务已取消", "")
 			return nil
@@ -722,7 +734,7 @@ func (s *Service) processClaimedTask(task *model.Task) error {
 		task.Error = taskFailureMessage(err)
 		task.CompletedAt = ptr(time.Now())
 		_ = s.repo.Save(task)
-		if providerSucceeded || s.BillingFailureRequiresReview(task.BillingOrderID, task.ID, err) {
+		if providerSucceeded || (!channelSlotFailedBeforeRequest && s.BillingFailureRequiresReview(task.BillingOrderID, task.ID, err)) {
 			_ = s.MarkBillingUncertain(task.BillingOrderID, task.Error)
 		} else {
 			_ = s.RefundBilling(task.BillingOrderID, task.Error)