diff --git a/constant/context_key.go b/constant/context_key.go index b856bc3dda14..52898045f9c1 100644 --- a/constant/context_key.go +++ b/constant/context_key.go @@ -72,4 +72,17 @@ const ( // fallback in authHelper (finishAdminAudit) skips its record to avoid // duplicate entries. ContextKeyAuditLogged ContextKey = "audit_logged" + + // ContextKeyChannelLimitGate carries the *service.GateHandle acquired by + // SelectChannelWithLimits so the retry loop / distributor can release the + // per-channel (and eventually per-key) concurrency slots after the + // upstream call completes — regardless of success or failure. + ContextKeyChannelLimitGate ContextKey = "channel_limit_gate" + + // ContextKeyChannelPreSelectedKeyIdx carries a key index that the channel-limit + // orchestrator (service.SelectChannelWithLimits) pre-selected for a multi-key + // channel after checking per-key cooldown and concurrency. SetupContextForSelectedChannel + // reads it to avoid re-running GetNextEnabledKey on a key the orchestrator + // already deemed unhealthy. + ContextKeyChannelPreSelectedKeyIdx ContextKey = "channel_pre_selected_key_idx" ) diff --git a/controller/channel_upstream_update.go b/controller/channel_upstream_update.go index 122a9f6bf9e0..549e730ea758 100644 --- a/controller/channel_upstream_update.go +++ b/controller/channel_upstream_update.go @@ -273,7 +273,7 @@ func fetchChannelUpstreamModelIDs(channel *model.Channel) ([]string, error) { } if channel.Type == constant.ChannelTypeGemini { - key, _, apiErr := channel.GetNextEnabledKey() + key, _, apiErr := channel.GetNextEnabledKey(nil) if apiErr != nil { return nil, fmt.Errorf("获取渠道密钥失败: %w", apiErr) } @@ -311,7 +311,7 @@ func fetchChannelUpstreamModelIDs(channel *model.Channel) ([]string, error) { url = fmt.Sprintf("%s/v1/models", baseURL) } - key, _, apiErr := channel.GetNextEnabledKey() + key, _, apiErr := channel.GetNextEnabledKey(nil) if apiErr != nil { return nil, fmt.Errorf("获取渠道密钥失败: %w", apiErr) } diff --git a/controller/ratio_sync.go b/controller/ratio_sync.go index 1f57bcc245f3..ba2295deb890 100644 --- a/controller/ratio_sync.go +++ b/controller/ratio_sync.go @@ -265,7 +265,7 @@ func FetchUpstreamRatios(c *gin.Context) { ch <- upstreamResult{Name: uniqueName, Err: "failed to get channel key: " + err.Error()} return } - key, _, apiErr := dbCh.GetNextEnabledKey() + key, _, apiErr := dbCh.GetNextEnabledKey(nil) if apiErr != nil { ch <- upstreamResult{Name: uniqueName, Err: "failed to get enabled channel key: " + apiErr.Error()} return diff --git a/controller/relay.go b/controller/relay.go index 6e91ccb60506..5a2c317c04c0 100644 --- a/controller/relay.go +++ b/controller/relay.go @@ -206,6 +206,9 @@ func Relay(c *gin.Context, relayFormat types.RelayFormat) { } else { newAPIError = types.NewErrorWithStatusCode(bodyErr, types.ErrorCodeReadRequestBodyFailed, http.StatusBadRequest, types.ErrOptionWithSkipRetry()) } + // Release the gate installed by the prior getChannel call so this + // early break does not leak the per-channel semaphore slot. + releaseLimitGate(c) break } c.Request.Body = io.NopCloser(bodyStorage) @@ -223,15 +226,21 @@ func Relay(c *gin.Context, relayFormat types.RelayFormat) { if newAPIError == nil { relayInfo.LastError = nil + releaseLimitGate(c) return } newAPIError = service.NormalizeViolationFeeError(newAPIError) relayInfo.LastError = newAPIError + // free the slot before retrying on another channel. + releaseLimitGate(c) processChannelError(c, *types.NewChannelError(channel.Id, channel.Type, channel.Name, channel.ChannelInfo.IsMultiKey, common.GetContextKeyString(c, constant.ContextKeyChannelKey), channel.GetAutoBan()), newAPIError) if !shouldRetry(c, newAPIError, common.RetryTimes-retryParam.GetRetry()) { + // not retrying: release the slot held by this iteration's handle so + // we don't leak the semaphore across the rest of the process lifetime. + releaseLimitGate(c) break } } @@ -261,6 +270,22 @@ func addUsedChannel(c *gin.Context, channelId int) { c.Set("use_channel", useChannel) } +// releaseLimitGate frees the per-channel (and per-key) concurrency slots held +// by the currently-selected channel's GateHandle, then clears the context key +// so a subsequent acquire in the next retry iteration starts fresh. Safe to +// call when no handle is stashed. Also clears the pre-selected key index +// stashed by the limit orchestrator so a later retry on a different channel +// does not pick up a stale idx from context. +func releaseLimitGate(c *gin.Context) { + if v, ok := common.GetContextKey(c, constant.ContextKeyChannelLimitGate); ok { + if h, ok := v.(*service.GateHandle); ok { + h.Release() + } + common.SetContextKey(c, constant.ContextKeyChannelLimitGate, nil) + } + common.SetContextKey(c, constant.ContextKeyChannelPreSelectedKeyIdx, nil) +} + func fastTokenCountMetaForPricing(request dto.Request) *types.TokenCountMeta { if request == nil { return &types.TokenCountMeta{} @@ -304,7 +329,7 @@ func getChannel(c *gin.Context, info *relaycommon.RelayInfo, retryParam *service AutoBan: &autoBanInt, }, nil } - channel, selectGroup, err := service.CacheGetRandomSatisfiedChannel(retryParam) + channel, selectGroup, handle, err := service.SelectChannelWithLimits(retryParam) info.PriceData.GroupRatioInfo = helper.HandleGroupRatio(c, info) @@ -315,8 +340,24 @@ func getChannel(c *gin.Context, info *relaycommon.RelayInfo, retryParam *service return nil, types.NewError(fmt.Errorf("分组 %s 下模型 %s 的可用渠道不存在（retry）", selectGroup, info.OriginModelName), types.ErrorCodeGetChannelFailed, types.ErrOptionWithSkipRetry()) } + // carry the gate handle so the loop can release it after the upstream call. + // A previous iteration may have stashed a handle that has not been released + // yet (e.g. early break from the loop); free it before swapping in the new + // one to avoid leaking semaphore slots across iterations. + if handle != nil { + if prev, ok := common.GetContextKey(c, constant.ContextKeyChannelLimitGate); ok { + if ph, ok := prev.(*service.GateHandle); ok && ph != handle { + ph.Release() + } + } + common.SetContextKey(c, constant.ContextKeyChannelLimitGate, handle) + } + newAPIError := middleware.SetupContextForSelectedChannel(c, channel, info.OriginModelName) if newAPIError != nil { + if handle != nil { + handle.Release() + } return nil, newAPIError } return channel, nil @@ -544,15 +585,24 @@ func RelayTask(c *gin.Context) { } else { taskErr = service.TaskErrorWrapperLocal(bodyErr, "read_request_body_failed", http.StatusBadRequest) } + // Release the gate installed by the prior getChannel call so this + // early break does not leak the per-channel semaphore slot. + releaseLimitGate(c) break } c.Request.Body = io.NopCloser(bodyStorage) result, taskErr = relay.RelayTaskSubmit(c, relayInfo) if taskErr == nil { + // submit succeeded -> channel is healthy; release the gate. + // Task accounting settles later when the upstream finishes the async job. + releaseLimitGate(c) break } + // release before processing error so this attempt's failure is + // observed without holding a slot. + releaseLimitGate(c) if !taskErr.LocalError { processChannelError(c, *types.NewChannelError(channel.Id, channel.Type, channel.Name, channel.ChannelInfo.IsMultiKey, @@ -561,6 +611,9 @@ func RelayTask(c *gin.Context) { } if !shouldRetryTaskRelay(c, channel.Id, taskErr, common.RetryTimes-retryParam.GetRetry()) { + // not retrying: release the slot held by this iteration's handle so + // we don't leak the semaphore across the rest of the process lifetime. + releaseLimitGate(c) break } } diff --git a/docs/superpowers/specs/2026-07-15-glm-affinity-concurrency-routing-design.md b/docs/superpowers/specs/2026-07-15-glm-affinity-concurrency-routing-design.md new file mode 100644 index 000000000000..5992db0a14c8 --- /dev/null +++ b/docs/superpowers/specs/2026-07-15-glm-affinity-concurrency-routing-design.md @@ -0,0 +1,305 @@ +# GLM Affinity 与并发感知渠道调度设计 + +## 1. 背景 + +同一组内配置多个 GLM 账号，每个账号对应一个渠道。所有 GLM 渠道具有相同的： + +- 优先级； +- 权重； +- 最大并发数； +- 五小时 Token 总额度。 + +系统需要在尽可能保持上游缓存命中的同时，均衡利用所有账号的并发容量，并在渠道暂时繁忙时继续满足用户请求。 + +现有用户级速率限制已经独立存在，并由管理员控制开关。本次改动不修改用户级速率限制的配置、执行顺序或行为，只调整渠道选择、Affinity 回退和渠道并发占位。 + +## 2. 目标 + +1. Affinity 命中时优先使用原绑定渠道，以提高上游缓存命中率。 +2. Affinity 请求与普通请求都必须占用渠道并发位置，不能绕过渠道并发上限。 +3. Affinity 渠道并发已满时，本次请求临时选择其他渠道，但不迁移原 Affinity。 +4. 无 Affinity 请求以及临时回退请求，在当前最高可用优先级内选择实时并发最少的渠道。 +5. 同一优先级全部满载时，继续尝试下一优先级，而不是直接返回无可用渠道。 +6. Token 耗尽、账号失效或持续熔断时，允许在新渠道成功后迁移 Affinity。 +7. 所有并发位置必须在成功、失败、超时、流结束和客户端断开时可靠释放。 + +## 3. 非目标 + +第一阶段不包含： + +- 修改或新增用户级 RPM、TPM、并发限制； +- 精确统计 GLM 五小时滚动 Token 窗口； +- 集群级全局并发计数； +- 为单个 Affinity key 设置独立并发上限； +- 根据历史 Affinity 绑定数量预留渠道； +- 修改渠道优先级、权重配置语义。 + +并发量仅作为第一阶段 Token 消耗速度的近似指标。未来可以在此设计上增加渠道 Token 预算感知。 + +## 4. 设计原则 + +### 4.1 Affinity 是软绑定 + +Affinity 指定首选渠道，但不能绕过渠道可用性和并发上限。 + +- 渠道可用且有并发位置：使用原绑定渠道； +- 渠道并发已满：本次临时回退； +- Token 耗尽、账号失效或持续熔断：允许迁移绑定； +- 临时回退成功：不修改原绑定。 + +### 4.2 相同容量使用最少连接 + +所有 GLM 渠道容量相同，因此普通权重随机不再提供额外价值。当前优先级内选择实时并发最少的渠道。 + +若多个渠道当前并发相同，在最低并发集合中随机选择，避免固定顺序偏向。 + +未来若渠道最大并发不同，可将选择分数扩展为： + +```text +(当前并发 + 1) / 最大并发 +``` + +本阶段不需要启用该扩展。 + +### 4.3 选择与占位必须原子衔接 + +读取当前并发后，其他请求可能抢先占满渠道。因此选中渠道后必须非阻塞地原子获取并发位置： + +- 获取成功：正式使用该渠道； +- 获取失败：将该渠道加入本次请求排除集合并重新选择。 + +## 5. 请求执行顺序 + +### 5.1 总流程 + +```text +请求进入 + ↓ +执行现有用户级速率限制（本次不修改） + ↓ +解析模型、分组、请求路径和 Affinity key + ↓ +查询 Affinity + ├─ 命中 + │ ↓ + │ 校验渠道状态、分组、模型和请求路径 + │ ↓ + │ 尝试获取绑定渠道的并发位置 + │ ├─ 成功：使用绑定渠道 + │ └─ 失败：标记临时并发回退，进入最少连接选择 + │ + └─ 未命中 + ↓ + 进入最少连接选择 +``` + +### 5.2 最少连接选择 + +```text +1. 获取满足分组、模型和请求路径的启用渠道 +2. 排除本次请求已经失败或抢占失败的渠道 +3. 取当前最高优先级 +4. 排除并发已满、冷却或不可用渠道 +5. 找到当前并发最少的渠道集合 +6. 在同分集合内随机选择一个渠道 +7. 原子获取渠道并发位置 + ├─ 成功：返回渠道及 GateHandle + └─ 失败：排除该渠道并回到步骤 3 +8. 当前优先级没有可用渠道时尝试下一优先级 +9. 所有优先级耗尽后返回无可用渠道 +``` + +循环上限不能使用固定的 8 次尝试。应根据本次候选渠道数量限定，保证渠道数超过 8 时仍能遍历全部候选，同时防止无限循环。 + +### 5.3 请求完成 + +并发位置覆盖完整上游请求生命周期： + +```text +获取 GateHandle + ↓ +发起上游请求 + ↓ +成功 / 失败 / 超时 / 流结束 / 客户端断开 + ↓ +统一释放 GateHandle +``` + +流式请求必须在响应流结束或断开后释放，不能在收到响应头时提前释放。 + +## 6. Affinity 更新策略 + +为每次选择记录明确策略： + +```text +keep 保持原绑定 +create 首次成功后创建绑定 +migrate 成功后迁移到新渠道 +clear 清除失效绑定 +``` + +建议在请求上下文记录以下选择原因： + +```text +affinity_hit +affinity_concurrency_fallback +no_affinity_least_connections +retry_after_token_exhausted +retry_after_channel_failure +``` + +更新规则： + +| 场景 | 当前请求处理 | Affinity 处理 | +|---|---|---| +| Affinity 命中且占位成功 | 使用原渠道 | `keep` | +| Affinity 渠道并发已满 | 临时选择其他渠道 | `keep` | +| 临时渠道成功 | 返回成功 | `keep`，禁止被 `SwitchOnSuccess` 改写 | +| 首次无绑定请求成功 | 返回成功 | `create` | +| 明确 Token 耗尽后切换成功 | 使用新渠道 | `migrate` | +| 账号失效或持续熔断后切换成功 | 使用新渠道 | `migrate` | +| 所有尝试失败 | 返回错误 | 不写入 | +| 缓存渠道已禁用且配置不保留 | 进入普通选择 | `clear` | + +现有 `SwitchOnSuccess` 不能再无条件根据最终 `channel_id` 改写 Affinity。只有策略为 `create` 或 `migrate` 且请求最终成功时才允许写入新渠道。 + +## 7. 429 与渠道错误处理 + +第一阶段不实现精确 Token 窗口，只根据 GLM 返回的明确错误码或可稳定识别的错误信息判断 Token 耗尽。 + +| 错误类型 | 是否重试其他渠道 | 是否迁移 Affinity | +|---|---|---| +| 本地渠道并发已满 | 是 | 否 | +| 无法确认原因的短暂 429 | 按现有重试配置 | 否 | +| 明确 Token 耗尽 | 是 | 新渠道成功后迁移 | +| 短暂超时或 5xx | 按现有重试配置 | 默认不迁移 | +| 鉴权失败、账号失效 | 是，并按现有规则禁用渠道 | 新渠道成功后迁移 | +| 持续失败达到熔断条件 | 是，渠道进入冷却 | 新渠道成功后迁移 | + +默认 Affinity 规则中的 `SkipRetryOnFailure` 仍可控制一般错误是否重试，但明确 Token 耗尽、账号失效和持续熔断需要有可审计的迁移路径。不能把所有 429 都视为 Token 耗尽。 + +## 8. 并发状态与多实例语义 + +继续使用进程内信号量实现渠道并发限制： + +```text +channel: +``` + +Affinity 和非 Affinity 请求共享同一个 key。 + +该限制为单进程限制。多实例部署时，每个实例都拥有独立上限。例如渠道上限为 10，部署 3 个实例时，理论最大总并发为 30。这是现有并发 gate 的既有语义，本阶段不改为 Redis 全局计数。 + +需要提供安全的只读当前并发方法，供最少连接选择使用。读取值只用于排序，最终正确性仍由原子 `TryAcquireConcurrency` 保证。 + +## 9. 与现有代码的衔接 + +主要涉及： + +- `middleware/distributor.go` + - Affinity 命中后不再直接绕过并发限制； + - 记录临时回退和 Affinity 更新策略。 +- `service/channel_limit_select.go` + - 支持首选 Affinity 渠道占位； + - 普通路径改为同优先级内最少连接； + - 当前优先级全部满后继续下一优先级； + - 尝试上限按候选数量计算。 +- `service/channel_gate.go` + - 提供安全读取当前占用数的接口； + - 保持非阻塞原子获取和统一释放。 +- `service/channel_affinity.go` + - 只有 `create`、`migrate` 才写入新绑定； + - 临时并发回退保持原绑定。 +- `controller/relay.go` + - 重试过程传递回退原因和 Affinity 更新策略； + - 所有结束路径释放 gate。 + +具体实现应复用现有 `GateHandle`、`RetryParam.ExcludeIDs` 和渠道缓存，不引入新的分布式依赖。 + +## 10. 可观测性 + +管理员日志的 `admin_info` 建议增加： + +```json +{ + "channel_selection": { + "reason": "affinity_concurrency_fallback", + "affinity_channel_id": 12, + "selected_channel_id": 18, + "affinity_update_policy": "keep", + "selected_channel_concurrency": 3, + "channel_max_concurrency": 10 + } +} +``` + +不得向普通用户暴露渠道 ID、账号信息或内部负载。 + +至少记录以下指标或可聚合日志： + +- Affinity 命中次数； +- Affinity 因并发满临时回退次数； +- 无 Affinity 最少连接选择次数； +- 原子抢占竞争失败次数； +- 全部渠道并发满次数； +- Affinity 创建、保持、迁移和清除次数。 + +## 11. 测试设计 + +### 11.1 单元测试 + +1. 三个同优先级、同上限渠道，选择当前并发最少者。 +2. 多个最低并发渠道同分时，只从同分集合选择。 +3. 最低并发渠道在占位时发生竞争，排除后选择下一渠道。 +4. 当前优先级全部满时降到下一优先级。 +5. 所有优先级全部满时返回无可用渠道。 +6. Affinity 命中后成功占用渠道 gate。 +7. Affinity 渠道满时临时回退且更新策略为 `keep`。 +8. 临时回退渠道成功后不改写原 Affinity。 +9. Token 耗尽迁移成功后更新 Affinity。 +10. GateHandle 在成功、错误和重复释放场景均安全。 + +### 11.2 并发回归测试 + +使用确定性的并发测试验证： + +- 任一渠道进行中的请求数不超过 `MaxConcurrency`； +- 多个相同容量渠道在持续请求下并发差值不长期超过 1，允许瞬时竞争偏差； +- 并发抢占失败不会泄漏槽位； +- 请求结束后所有测试 gate 恢复为空。 + +测试不得依赖 sleep 或时间性能断言，应通过同步屏障和显式 channel 控制并发阶段。 + +### 11.3 集成验证 + +1. 配置至少三个相同 GLM 渠道和相同最大并发。 +2. 使用不同 Affinity key 建立绑定。 +3. 将一个绑定渠道占满，确认该 key 临时切换且原绑定不变。 +4. 释放原渠道后，确认后续请求重新命中原绑定。 +5. 模拟明确 Token 耗尽，确认新渠道成功后绑定迁移。 +6. 验证用户级速率限制开关和行为在改动前后完全一致。 + +## 12. 验收标准 + +1. Affinity 请求计入渠道并发，无法突破单进程渠道上限。 +2. 同配置 GLM 渠道中，非 Affinity 请求优先进入当前并发最少的渠道。 +3. 同分选择不存在固定首渠道偏向。 +4. Affinity 渠道并发满时请求可以临时回退，且原绑定保持不变。 +5. 释放并发后，相同 Affinity 可以重新回到原渠道。 +6. 最高优先级全部满时可以使用下一优先级渠道。 +7. Token 耗尽、账号失效或持续熔断时，新渠道成功后可以迁移绑定。 +8. 普通错误和临时并发回退不会导致 Affinity 漂移。 +9. 用户级速率限制相关代码、配置和开关没有行为变化。 +10. 相关 Go 测试通过，并发测试无 gate 泄漏。 + +## 13. 后续扩展 + +第一阶段上线并积累数据后，可独立增加： + +- GLM 五小时滚动 Token 消耗桶； +- 根据剩余 Token 排除或降权渠道； +- 区分并发 429、RPM/TPM 429 和 Token 耗尽 429； +- 分布式全局并发计数； +- 渠道级冷却和熔断状态可视化。 + +这些扩展不改变本设计的核心边界：Affinity 优先、资源限制不可绕过、临时回退不迁移、永久不可用才迁移。 diff --git a/dto/channel_settings.go b/dto/channel_settings.go index dbcfd3181ae9..57dbdd8e9da8 100644 --- a/dto/channel_settings.go +++ b/dto/channel_settings.go @@ -17,6 +17,9 @@ type ChannelSettings struct { PassThroughBodyEnabled bool `json:"pass_through_body_enabled,omitempty"` SystemPrompt string `json:"system_prompt,omitempty"` SystemPromptOverride bool `json:"system_prompt_override,omitempty"` + + // MaxConcurrency is the per-channel concurrency cap. 0 means "unlimited". + MaxConcurrency int `json:"max_concurrency,omitempty"` } type VertexKeyType string diff --git a/middleware/distributor.go b/middleware/distributor.go index 4234011c9f7c..ded9a58f165f 100644 --- a/middleware/distributor.go +++ b/middleware/distributor.go @@ -101,44 +101,60 @@ func Distribute() func(c *gin.Context) { } } + var affinityExcludeIDs []int if preferredChannelID, found := service.GetPreferredChannelByAffinity(c, modelRequest.Model, usingGroup); found { - affinityUsable := false + affinityCandidateValid := false preferred, err := model.CacheGetChannel(preferredChannelID) if err == nil && preferred != nil && preferred.Status == common.ChannelStatusEnabled && channelSupportsRequestPath(preferred, c.Request.URL.Path, modelRequest.Model) { + preferredGroup := "" if usingGroup == "auto" { userGroup := common.GetContextKeyString(c, constant.ContextKeyUserGroup) autoGroups := service.GetUserAutoGroup(userGroup) for _, g := range autoGroups { if model.IsChannelEnabledForGroupModel(g, modelRequest.Model, preferred.Id) { - selectGroup = g - common.SetContextKey(c, constant.ContextKeyAutoGroup, g) - channel = preferred - affinityUsable = true - service.MarkChannelAffinityUsed(c, g, preferred.Id) + preferredGroup = g break } } } else if model.IsChannelEnabledForGroupModel(usingGroup, modelRequest.Model, preferred.Id) { - channel = preferred - selectGroup = usingGroup - affinityUsable = true - service.MarkChannelAffinityUsed(c, usingGroup, preferred.Id) + preferredGroup = usingGroup + } + if preferredGroup != "" { + affinityCandidateValid = true + handle, acquired := service.AcquireChannelWithLimits(c, preferred) + if acquired { + channel = preferred + selectGroup = preferredGroup + common.SetContextKey(c, constant.ContextKeyChannelLimitGate, handle) + if usingGroup == "auto" { + common.SetContextKey(c, constant.ContextKeyAutoGroup, preferredGroup) + } + service.MarkChannelAffinityUsed(c, preferredGroup, preferred.Id) + } else { + affinityExcludeIDs = append(affinityExcludeIDs, preferred.Id) + } } } - if !affinityUsable && !service.ShouldKeepChannelAffinityOnChannelDisabled() { + if !affinityCandidateValid && !service.ShouldKeepChannelAffinityOnChannelDisabled() { service.ClearCurrentChannelAffinityCache(c) } } if channel == nil { - channel, selectGroup, err = service.CacheGetRandomSatisfiedChannel(&service.RetryParam{ + rp := &service.RetryParam{ Ctx: c, ModelName: modelRequest.Model, TokenGroup: usingGroup, RequestPath: c.Request.URL.Path, Retry: common.GetPointer(0), - }) + ExcludeIDs: affinityExcludeIDs, + } + var handle *service.GateHandle + channel, selectGroup, handle, err = service.SelectChannelWithLimits(rp) + if handle != nil { + common.SetContextKey(c, constant.ContextKeyChannelLimitGate, handle) + } if err != nil { showGroup := usingGroup if usingGroup == "auto" { @@ -440,6 +456,29 @@ func getTaskOriginModelName(c *gin.Context) string { return "" } +// selectChannelKeyFromContext picks the active key for `channel`. If the +// channel-limit orchestrator (service.SelectChannelWithLimits) recorded a +// per-key pre-selection on the request context (because that key passed +// per-key concurrency gating), honor it verbatim — re-running +// GetNextEnabledKey could otherwise re-pick a previously rejected key. +// Falls back to GetNextEnabledKey(nil) when no pre-selection was recorded +// or the pre-selected index is out of range. +func selectChannelKeyFromContext(c *gin.Context, channel *model.Channel) (string, int, error) { + if channel.ChannelInfo.IsMultiKey { + if pre, ok := common.GetContextKey(c, constant.ContextKeyChannelPreSelectedKeyIdx); ok { + if idx, ok := pre.(int); ok && idx >= 0 && idx < len(channel.GetKeys()) { + keys := channel.GetKeys() + return keys[idx], idx, nil + } + } + } + key, index, apiErr := channel.GetNextEnabledKey(nil) + if apiErr != nil { + return key, index, apiErr + } + return key, index, nil +} + func SetupContextForSelectedChannel(c *gin.Context, channel *model.Channel, modelName string) *types.NewAPIError { c.Set("original_model", modelName) // for retry if channel == nil { @@ -465,9 +504,9 @@ func SetupContextForSelectedChannel(c *gin.Context, channel *model.Channel, mode common.SetContextKey(c, constant.ContextKeyChannelModelMapping, channel.GetModelMapping()) common.SetContextKey(c, constant.ContextKeyChannelStatusCodeMapping, channel.GetStatusCodeMapping()) - key, index, newAPIError := channel.GetNextEnabledKey() + key, index, newAPIError := selectChannelKeyFromContext(c, channel) if newAPIError != nil { - return newAPIError + return newAPIError.(*types.NewAPIError) } if channel.ChannelInfo.IsMultiKey { common.SetContextKey(c, constant.ContextKeyChannelIsMultiKey, true) diff --git a/middleware/distributor_test.go b/middleware/distributor_test.go new file mode 100644 index 000000000000..850e7985187f --- /dev/null +++ b/middleware/distributor_test.go @@ -0,0 +1,80 @@ +package middleware + +import ( + "testing" + + "github.com/QuantumNous/new-api/common" + "github.com/QuantumNous/new-api/constant" + "github.com/QuantumNous/new-api/model" + "github.com/gin-gonic/gin" + "github.com/stretchr/testify/require" +) + +// selectChannelKeyFromContext honors a pre-selected index recorded on the +// request context by service.SelectChannelWithLimits; otherwise it falls +// back to channel.GetNextEnabledKey(nil). + +func newMultiKeyChannel() *model.Channel { + return &model.Channel{ + Id: 9501, + Key: "key-a\nkey-b\nkey-c", + ChannelInfo: model.ChannelInfo{IsMultiKey: true}, + } +} + +func newEmptyContext(t *testing.T) *gin.Context { + t.Helper() + gin.SetMode(gin.TestMode) + c, _ := gin.CreateTestContext(nil) + c.Set("__dummy", 0) // ensure Keys map is initialized + return c +} + +func TestSelectChannelKeyFromContext_PreSelectedIdxHonored(t *testing.T) { + c := newEmptyContext(t) + common.SetContextKey(c, constant.ContextKeyChannelPreSelectedKeyIdx, 1) + ch := newMultiKeyChannel() + + key, idx, err := selectChannelKeyFromContext(c, ch) + require.NoError(t, err) + require.Equal(t, 1, idx) + require.Equal(t, "key-b", key) +} + +func TestSelectChannelKeyFromContext_OutOfRangeFallsBack(t *testing.T) { + c := newEmptyContext(t) + // An out-of-range pre-selected index must NOT panic or return empty; the + // helper must fall back to GetNextEnabledKey(nil). + common.SetContextKey(c, constant.ContextKeyChannelPreSelectedKeyIdx, 99) + ch := newMultiKeyChannel() + + key, idx, err := selectChannelKeyFromContext(c, ch) + require.NoError(t, err) + require.GreaterOrEqual(t, idx, 0) + require.Less(t, idx, len(ch.GetKeys())) + require.Contains(t, []string{"key-a", "key-b", "key-c"}, key) +} + +func TestSelectChannelKeyFromContext_NonMultiKeyIgnoresContext(t *testing.T) { + c := newEmptyContext(t) + // On a non-multi-key channel the pre-selection is meaningless; the helper + // must still return channel.Key / 0. + common.SetContextKey(c, constant.ContextKeyChannelPreSelectedKeyIdx, 5) + ch := &model.Channel{Id: 9502, Key: "single-key"} + + key, idx, err := selectChannelKeyFromContext(c, ch) + require.NoError(t, err) + require.Equal(t, 0, idx) + require.Equal(t, "single-key", key) +} + +func TestSelectChannelKeyFromContext_NoContextFallsBack(t *testing.T) { + c := newEmptyContext(t) + ch := newMultiKeyChannel() + + key, idx, err := selectChannelKeyFromContext(c, ch) + require.NoError(t, err) + require.GreaterOrEqual(t, idx, 0) + require.Less(t, idx, len(ch.GetKeys())) + require.Contains(t, []string{"key-a", "key-b", "key-c"}, key) +} diff --git a/model/ability.go b/model/ability.go index e67b28301e02..ae21a9734e79 100644 --- a/model/ability.go +++ b/model/ability.go @@ -105,7 +105,7 @@ func getChannelQuery(group string, model string, retry int) (*gorm.DB, error) { return channelQuery, nil } -func GetChannel(group string, model string, retry int, requestPath string) (*Channel, error) { +func GetChannel(group string, model string, retry int, requestPath string, excludeIDs []int) (*Channel, error) { var abilities []Ability var err error = nil @@ -122,6 +122,7 @@ func GetChannel(group string, model string, retry int, requestPath string) (*Cha return nil, err } abilities = filterAbilitiesByRequestPathAndModel(abilities, requestPath, model) + abilities = excludeAbilities(abilities, excludeIDs) channel := Channel{} if len(abilities) > 0 { // Randomly choose one @@ -146,6 +147,29 @@ func GetChannel(group string, model string, retry int, requestPath string) (*Cha return &channel, err } +func getSatisfiedChannelsFromDB(group string, modelName string, requestPath string, excludeIDs []int) ([]*Channel, error) { + var abilities []Ability + err := DB.Where(&Ability{Group: group, Model: modelName, Enabled: true}). + Order("priority DESC"). + Order("weight DESC"). + Find(&abilities).Error + if err != nil { + return nil, err + } + abilities = filterAbilitiesByRequestPathAndModel(abilities, requestPath, modelName) + abilities = excludeAbilities(abilities, excludeIDs) + + channels := make([]*Channel, 0, len(abilities)) + for _, ability := range abilities { + var channel Channel + if err := DB.First(&channel, "id = ?", ability.ChannelId).Error; err != nil { + return nil, err + } + channels = append(channels, &channel) + } + return channels, nil +} + // filterAbilitiesByRequestPathAndModel restricts candidates by request path and // model for the DB (non-memory-cache) selection path. Only Advanced Custom // (type 58) channels are path-checked: kept only when one of their routes matches @@ -389,3 +413,25 @@ func FixAbility() (int, int, error) { InitChannelCache() return successCount, failCount, nil } + +// excludeAbilities returns a new slice with any Ability whose ChannelId is in +// excludeIDs removed. If excludeIDs is empty it returns abilities unchanged. +// Allocates a fresh slice so the caller does not need to worry about the +// input slice's lifetime — matches filterAbilitiesByRequestPath's +// non-mutating contract. +func excludeAbilities(abilities []Ability, excludeIDs []int) []Ability { + if len(excludeIDs) == 0 { + return abilities + } + excluded := make(map[int]bool, len(excludeIDs)) + for _, id := range excludeIDs { + excluded[id] = true + } + out := make([]Ability, 0, len(abilities)) + for _, a := range abilities { + if !excluded[a.ChannelId] { + out = append(out, a) + } + } + return out +} diff --git a/model/channel.go b/model/channel.go index dbd1deaef4d4..47fa4afe3440 100644 --- a/model/channel.go +++ b/model/channel.go @@ -196,7 +196,12 @@ func (channel *Channel) GetKeys() []string { return keys } -func (channel *Channel) GetNextEnabledKey() (string, int, *types.NewAPIError) { +// GetNextEnabledKey returns the next usable (string, index) pair for this channel. +// excludeKeyIdx lists key indices that should be skipped (e.g. cooled or currently +// over its per-key concurrency limit). Pass nil when no exclusions apply. +// +// Non-multi-key channels ignore the exclusion map and return (channel.Key, 0). +func (channel *Channel) GetNextEnabledKey(excludeKeyIdx map[int]bool) (string, int, *types.NewAPIError) { // If not in multi-key mode, return the original key string directly. if !channel.ChannelInfo.IsMultiKey { return channel.Key, 0, nil @@ -225,10 +230,10 @@ func (channel *Channel) GetNextEnabledKey() (string, int, *types.NewAPIError) { return common.ChannelStatusEnabled } - // Collect indexes of enabled keys + // Collect indexes of enabled, non-excluded keys enabledIdx := make([]int, 0, len(keys)) for i := range keys { - if getStatus(i) == common.ChannelStatusEnabled { + if getStatus(i) == common.ChannelStatusEnabled && !excludeKeyIdx[i] { enabledIdx = append(enabledIdx, i) } } diff --git a/model/channel_cache.go b/model/channel_cache.go index 81923017d79c..f8c17e055ea1 100644 --- a/model/channel_cache.go +++ b/model/channel_cache.go @@ -111,10 +111,10 @@ func SyncChannelCache(frequency int) { } } -func GetRandomSatisfiedChannel(group string, model string, retry int, requestPath string) (*Channel, error) { +func GetRandomSatisfiedChannel(group string, model string, retry int, requestPath string, excludeIDs []int) (*Channel, error) { // if memory cache is disabled, get channel directly from database if !common.MemoryCacheEnabled { - return GetChannel(group, model, retry, requestPath) + return GetChannel(group, model, retry, requestPath, excludeIDs) } channelSyncLock.RLock() @@ -129,6 +129,8 @@ func GetRandomSatisfiedChannel(group string, model string, retry int, requestPat channels = filterChannelsByRequestPathAndModel(group2model2channels[group][normalizedModel], requestPath, model) } + channels = excludeChannelIDs(channels, excludeIDs) + if len(channels) == 0 { return nil, nil } @@ -208,6 +210,35 @@ func GetRandomSatisfiedChannel(group string, model string, retry int, requestPat return nil, errors.New("channel not found") } +// GetSatisfiedChannels returns every enabled channel that can serve the +// request, ordered by descending priority. The returned slice is independent +// from the cached channel ID slices and is safe for the caller to reorder. +func GetSatisfiedChannels(group string, modelName string, requestPath string, excludeIDs []int) ([]*Channel, error) { + if !common.MemoryCacheEnabled { + return getSatisfiedChannelsFromDB(group, modelName, requestPath, excludeIDs) + } + + channelSyncLock.RLock() + defer channelSyncLock.RUnlock() + + channelIDs := filterChannelsByRequestPathAndModel(group2model2channels[group][modelName], requestPath, modelName) + if len(channelIDs) == 0 { + normalizedModel := ratio_setting.FormatMatchingModelName(modelName) + channelIDs = filterChannelsByRequestPathAndModel(group2model2channels[group][normalizedModel], requestPath, modelName) + } + channelIDs = excludeChannelIDs(channelIDs, excludeIDs) + + channels := make([]*Channel, 0, len(channelIDs)) + for _, channelID := range channelIDs { + channel, ok := channelsIDM[channelID] + if !ok { + return nil, fmt.Errorf("channel #%d does not exist", channelID) + } + channels = append(channels, channel) + } + return channels, nil +} + // filterChannelsByRequestPathAndModel restricts candidates by request path and // model. Only Advanced Custom (type 58) channels are path-checked: they are kept // only when one of their configured routes matches requestPath and model. All @@ -327,3 +358,25 @@ func CacheUpdateChannel(channel *Channel) { channelSyncLock.Unlock() InvalidatePricingCache() } + +// excludeChannelIDs returns a new slice containing ids with any in exclude +// removed. If exclude is empty it returns ids unchanged. Allocates a fresh +// slice so the caller may continue to read the cached group2model2channels +// slice without seeing this filter's writes — matches filterChannelsByRequestPath's +// non-mutating contract. +func excludeChannelIDs(ids []int, exclude []int) []int { + if len(exclude) == 0 { + return ids + } + excluded := make(map[int]bool, len(exclude)) + for _, id := range exclude { + excluded[id] = true + } + out := make([]int, 0, len(ids)) + for _, id := range ids { + if !excluded[id] { + out = append(out, id) + } + } + return out +} diff --git a/model/channel_cache_exclude_test.go b/model/channel_cache_exclude_test.go new file mode 100644 index 000000000000..99c679966814 --- /dev/null +++ b/model/channel_cache_exclude_test.go @@ -0,0 +1,20 @@ +package model + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestExcludeChannelIDs(t *testing.T) { + require.Equal(t, []int{1, 2, 3}, excludeChannelIDs([]int{1, 2, 3}, nil)) + assert.Equal(t, []int{1, 3}, excludeChannelIDs([]int{1, 2, 3}, []int{2})) + assert.Empty(t, excludeChannelIDs([]int{1, 2}, []int{1, 2, 3})) +} + +func TestExcludeAbilities(t *testing.T) { + a := []Ability{{ChannelId: 1}, {ChannelId: 2}, {ChannelId: 3}} + got := excludeAbilities(a, []int{2}) + assert.Equal(t, []int{1, 3}, []int{got[0].ChannelId, got[1].ChannelId}) +} diff --git a/model/channel_candidates_test.go b/model/channel_candidates_test.go new file mode 100644 index 000000000000..c03b18f986c7 --- /dev/null +++ b/model/channel_candidates_test.go @@ -0,0 +1,31 @@ +package model + +import ( + "testing" + + "github.com/QuantumNous/new-api/common" + "github.com/QuantumNous/new-api/constant" + "github.com/stretchr/testify/require" +) + +func TestGetSatisfiedChannelsOrdersByPriorityAndExcludesIDs(t *testing.T) { + clearPreferredOwnerTables(t) + t.Cleanup(func() { + clearPreferredOwnerTables(t) + InitChannelCache() + }) + + originalMemoryCacheEnabled := common.MemoryCacheEnabled + common.MemoryCacheEnabled = false + t.Cleanup(func() { common.MemoryCacheEnabled = originalMemoryCacheEnabled }) + + insertPreferredOwnerCandidate(t, 9401, "glm-4", "default", constant.ChannelTypeOpenAI, 10, 100, common.ChannelStatusEnabled, true) + insertPreferredOwnerCandidate(t, 9402, "glm-4", "default", constant.ChannelTypeOpenAI, 20, 100, common.ChannelStatusEnabled, true) + insertPreferredOwnerCandidate(t, 9403, "glm-4", "default", constant.ChannelTypeOpenAI, 10, 100, common.ChannelStatusEnabled, true) + + channels, err := GetSatisfiedChannels("default", "glm-4", "", []int{9401}) + require.NoError(t, err) + require.Len(t, channels, 2) + require.Equal(t, 9402, channels[0].Id) + require.Equal(t, 9403, channels[1].Id) +} diff --git a/relay/common/relay_info.go b/relay/common/relay_info.go index 9f460ce5c6a7..8d6e8987392d 100644 --- a/relay/common/relay_info.go +++ b/relay/common/relay_info.go @@ -149,10 +149,10 @@ type RelayInfo struct { IsClaudeBetaQuery bool // /v1/messages?beta=true IsChannelTest bool // channel test request RetryIndex int - LastError *types.NewAPIError - RuntimeHeadersOverride map[string]interface{} - UseRuntimeHeadersOverride bool - ParamOverrideAudit []string + LastError *types.NewAPIError + RuntimeHeadersOverride map[string]interface{} + UseRuntimeHeadersOverride bool + ParamOverrideAudit []string // UpstreamRequestBodySize is the byte size of the marshaled upstream request // body. It is set when the body is wrapped in a BodyStorage (see diff --git a/relay/relay_task.go b/relay/relay_task.go index fb384d18937a..f485154c9796 100644 --- a/relay/relay_task.go +++ b/relay/relay_task.go @@ -90,7 +90,7 @@ func ResolveOriginTask(c *gin.Context, info *relaycommon.RelayInfo) *dto.TaskErr info.LockedChannel = ch if originTask.ChannelId != info.ChannelId { - key, _, newAPIError := ch.GetNextEnabledKey() + key, _, newAPIError := ch.GetNextEnabledKey(nil) if newAPIError != nil { return service.TaskErrorWrapper(newAPIError, "channel_no_available_key", newAPIError.StatusCode) } diff --git a/service/channel_affinity.go b/service/channel_affinity.go index 96ec13e248cc..de047e2919af 100644 --- a/service/channel_affinity.go +++ b/service/channel_affinity.go @@ -25,11 +25,19 @@ const ( ginKeyChannelAffinityMeta = "channel_affinity_meta" ginKeyChannelAffinityLogInfo = "channel_affinity_log_info" ginKeyChannelAffinitySkipRetry = "channel_affinity_skip_retry_on_failure" + ginKeyChannelAffinityPolicy = "channel_affinity_update_policy" channelAffinityCacheNamespace = "new-api:channel_affinity:v1" channelAffinityUsageCacheStatsNamespace = "new-api:channel_affinity_usage_cache_stats:v1" ) +const ( + ChannelAffinityPolicyKeep = "keep" + ChannelAffinityPolicyCreate = "create" + ChannelAffinityPolicyMigrate = "migrate" + ChannelAffinityPolicyClear = "clear" +) + var ( channelAffinityCacheOnce sync.Once channelAffinityCache *cachex.HybridCache[int] @@ -353,6 +361,14 @@ func setChannelAffinityContext(c *gin.Context, meta channelAffinityMeta) { c.Set(ginKeyChannelAffinityCacheKey, meta.CacheKey) c.Set(ginKeyChannelAffinityTTLSeconds, meta.TTLSeconds) c.Set(ginKeyChannelAffinityMeta, meta) + c.Set(ginKeyChannelAffinityPolicy, ChannelAffinityPolicyCreate) +} + +func SetChannelAffinityUpdatePolicy(c *gin.Context, policy string) { + if c == nil { + return + } + c.Set(ginKeyChannelAffinityPolicy, policy) } func getChannelAffinityContext(c *gin.Context) (string, int, bool) { @@ -616,6 +632,7 @@ func GetPreferredChannelByAffinity(c *gin.Context, modelName string, usingGroup return 0, false } if found { + SetChannelAffinityUpdatePolicy(c, ChannelAffinityPolicyKeep) return channelID, true } return 0, false @@ -711,13 +728,20 @@ func AppendChannelAffinityAdminInfo(c *gin.Context, adminInfo map[string]interfa } func RecordChannelAffinity(c *gin.Context, channelID int) { - if channelID <= 0 { + if c == nil || channelID <= 0 { return } setting := operation_setting.GetChannelAffinitySetting() if setting == nil || !setting.Enabled { return } + policy := c.GetString(ginKeyChannelAffinityPolicy) + if policy == ChannelAffinityPolicyKeep || policy == ChannelAffinityPolicyClear { + return + } + if policy != ChannelAffinityPolicyCreate && policy != ChannelAffinityPolicyMigrate { + return + } if setting.SwitchOnSuccess && c != nil { if successChannelID := c.GetInt("channel_id"); successChannelID > 0 { channelID = successChannelID diff --git a/service/channel_affinity_template_test.go b/service/channel_affinity_template_test.go index fb703a24e720..5c80a5e9a559 100644 --- a/service/channel_affinity_template_test.go +++ b/service/channel_affinity_template_test.go @@ -263,6 +263,34 @@ func TestClearCurrentChannelAffinityCache(t *testing.T) { require.False(t, ShouldSkipRetryAfterChannelAffinityFailure(ctx)) } +func TestRecordChannelAffinityHonorsUpdatePolicy(t *testing.T) { + cacheKeySuffix := fmt.Sprintf("codex cli trace:default:update-policy-%d", time.Now().UnixNano()) + cacheKeyFull := channelAffinityCacheNamespace + ":" + cacheKeySuffix + cache := getChannelAffinityCache() + require.NoError(t, cache.SetWithTTL(cacheKeySuffix, 111, time.Minute)) + t.Cleanup(func() { + _, _ = cache.DeleteMany([]string{cacheKeySuffix}) + }) + + ctx := buildChannelAffinityTemplateContextForTest(channelAffinityMeta{ + CacheKey: cacheKeyFull, + TTLSeconds: 60, + }) + SetChannelAffinityUpdatePolicy(ctx, ChannelAffinityPolicyKeep) + RecordChannelAffinity(ctx, 222) + channelID, found, err := cache.Get(cacheKeySuffix) + require.NoError(t, err) + require.True(t, found) + require.Equal(t, 111, channelID) + + SetChannelAffinityUpdatePolicy(ctx, ChannelAffinityPolicyMigrate) + RecordChannelAffinity(ctx, 222) + channelID, found, err = cache.Get(cacheKeySuffix) + require.NoError(t, err) + require.True(t, found) + require.Equal(t, 222, channelID) +} + func TestChannelAffinityHitCodexTemplatePassHeadersEffective(t *testing.T) { gin.SetMode(gin.TestMode) diff --git a/service/channel_gate.go b/service/channel_gate.go new file mode 100644 index 000000000000..d8776ddc726a --- /dev/null +++ b/service/channel_gate.go @@ -0,0 +1,50 @@ +package service + +import "sync" + +// concurrencyGates holds buffered channels used as counting semaphores, keyed by +// gateKey ("channel:" or "channel::key:"). In-memory per-process by +// design: the concurrency path must not pay a Redis RTT. Multi-node deployments +// get MaxConcurrency per node (documented in the spec). +var concurrencyGates sync.Map // map[string]chan struct{} + +// GetConcurrencyStatus returns the current occupancy and capacity of a tracked +// gate. Unknown and unlimited gates are not tracked and report (0, 0). +func GetConcurrencyStatus(gateKey string) (used int, max int) { + v, ok := concurrencyGates.Load(gateKey) + if !ok { + return 0, 0 + } + gate := v.(chan struct{}) + return len(gate), cap(gate) +} + +// TryAcquireConcurrency acquires one slot non-blockingly. max<=0 means unlimited +// and always succeeds without tracking. Returns false if the gate is full. +func TryAcquireConcurrency(gateKey string, max int) bool { + if max <= 0 { + return true + } + v, _ := concurrencyGates.LoadOrStore(gateKey, make(chan struct{}, max)) + gate := v.(chan struct{}) + select { + case gate <- struct{}{}: + return true + default: + return false + } +} + +// ReleaseConcurrency releases one slot. Safe to call when max<=0 was used (no-op). +func ReleaseConcurrency(gateKey string) { + v, ok := concurrencyGates.Load(gateKey) + if !ok { + return + } + gate := v.(chan struct{}) + select { + case <-gate: + default: + // already empty; ignore to avoid blocking on over-release + } +} diff --git a/service/channel_gate_test.go b/service/channel_gate_test.go new file mode 100644 index 000000000000..19a1b3221e5b --- /dev/null +++ b/service/channel_gate_test.go @@ -0,0 +1,77 @@ +package service + +import ( + "sync" + "sync/atomic" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestTryAcquireConcurrency_ZeroMaxIsUnlimited(t *testing.T) { + key := "test:unlimited" + assert.True(t, TryAcquireConcurrency(key, 0)) + assert.True(t, TryAcquireConcurrency(key, 0)) + // no release needed; nothing tracked +} + +func TestTryAcquireConcurrency_RespectsMax(t *testing.T) { + key := "test:cap2" + assert.True(t, TryAcquireConcurrency(key, 2)) + assert.True(t, TryAcquireConcurrency(key, 2)) + assert.False(t, TryAcquireConcurrency(key, 2), "third acquire must fail") + ReleaseConcurrency(key) + assert.True(t, TryAcquireConcurrency(key, 2), "acquire after release must succeed") + ReleaseConcurrency(key) + ReleaseConcurrency(key) +} + +func TestTryAcquireConcurrency_ConcurrentInFlightNeverExceedsMax(t *testing.T) { + const max = 5 + const goroutines = 50 + key := "test:race" + var inFlight, peak int32 + var wg sync.WaitGroup + for i := 0; i < goroutines; i++ { + wg.Add(1) + go func() { + defer wg.Done() + for j := 0; j < 20; j++ { + if TryAcquireConcurrency(key, max) { + n := addInFlight(&inFlight, 1) + recordPeak(&peak, n) + addInFlight(&inFlight, -1) + ReleaseConcurrency(key) + } + } + }() + } + wg.Wait() + assert.LessOrEqual(t, int(peak), max, "in-flight must never exceed max") +} + +func TestGetConcurrencyStatus(t *testing.T) { + const key = "test:concurrency-status" + + used, maxConcurrency := GetConcurrencyStatus(key) + assert.Equal(t, 0, used) + assert.Equal(t, 0, maxConcurrency) + + require.True(t, TryAcquireConcurrency(key, 2)) + t.Cleanup(func() { ReleaseConcurrency(key) }) + + used, maxConcurrency = GetConcurrencyStatus(key) + assert.Equal(t, 1, used) + assert.Equal(t, 2, maxConcurrency) +} + +func addInFlight(p *int32, delta int32) int32 { return atomic.AddInt32(p, delta) } +func recordPeak(peak *int32, n int32) { + for { + cur := atomic.LoadInt32(peak) + if n <= cur || atomic.CompareAndSwapInt32(peak, cur, n) { + return + } + } +} diff --git a/service/channel_limit.go b/service/channel_limit.go new file mode 100644 index 000000000000..21c71a8a9a72 --- /dev/null +++ b/service/channel_limit.go @@ -0,0 +1,26 @@ +package service + +import ( + "github.com/QuantumNous/new-api/model" + "github.com/QuantumNous/new-api/setting/operation_setting" +) + +// ChannelLimits is the effective limit set for a channel after applying the +// global toggle. MaxConcurrency of 0 means "unlimited". +type ChannelLimits struct { + Enabled bool + MaxConcurrency int +} + +// GetChannelLimits returns the effective limit set for a channel. +func GetChannelLimits(channel *model.Channel) ChannelLimits { + g := operation_setting.GetChannelLimitSetting() + if !g.Enabled { + return ChannelLimits{Enabled: false} + } + s := channel.GetSetting() + return ChannelLimits{ + Enabled: true, + MaxConcurrency: s.MaxConcurrency, + } +} diff --git a/service/channel_limit_select.go b/service/channel_limit_select.go new file mode 100644 index 000000000000..c30dbadb59cf --- /dev/null +++ b/service/channel_limit_select.go @@ -0,0 +1,253 @@ +package service + +import ( + "errors" + "fmt" + "math" + "math/rand" + + "github.com/QuantumNous/new-api/common" + "github.com/QuantumNous/new-api/constant" + "github.com/QuantumNous/new-api/model" + "github.com/gin-gonic/gin" +) + +func leastLoadedChannels(channels []*model.Channel) []*model.Channel { + minimumUsed := math.MaxInt + leastLoaded := make([]*model.Channel, 0, len(channels)) + for _, channel := range channels { + lim := GetChannelLimits(channel) + used, _ := GetConcurrencyStatus(channelGateKey(channel.Id, -1)) + if lim.Enabled && lim.MaxConcurrency > 0 && used >= lim.MaxConcurrency { + continue + } + if used < minimumUsed { + minimumUsed = used + leastLoaded = leastLoaded[:0] + } + if used == minimumUsed { + leastLoaded = append(leastLoaded, channel) + } + } + return leastLoaded +} + +// GateHandle tracks the semaphore slots acquired for one request so they can be +// released together when the request completes (success or failure). +type GateHandle struct { + GateKeys []string +} + +// Release frees every acquired semaphore slot. +func (h *GateHandle) Release() { + if h == nil { + return + } + for _, key := range h.GateKeys { + ReleaseConcurrency(key) + } + h.GateKeys = nil +} + +func (h *GateHandle) acquire(key string, max int) bool { + if !TryAcquireConcurrency(key, max) { + return false + } + h.GateKeys = append(h.GateKeys, key) + return true +} + +// channelGateKey builds the semaphore key. keyIdx<0 means per-channel. +func channelGateKey(channelID, keyIdx int) string { + if keyIdx < 0 { + return fmt.Sprintf("channel:%d", channelID) + } + return fmt.Sprintf("channel:%d:key:%d", channelID, keyIdx) +} + +// evaluateChannelForLimits atomically acquires the requested channel slot. +// It returns true when the gate is full and adds the channel to excluded. +func evaluateChannelForLimits(channelID, keyIdx int, lim ChannelLimits, excluded map[int]bool) bool { + if !lim.Enabled || lim.MaxConcurrency <= 0 { + return false + } + if !TryAcquireConcurrency(channelGateKey(channelID, keyIdx), lim.MaxConcurrency) { + excluded[channelID] = true + return true + } + return false +} + +func selectHealthyKey(channel *model.Channel, lim ChannelLimits) (int, bool) { + keys := channel.GetKeys() + excluded := map[int]bool{} + for attempt := 0; attempt < len(keys); attempt++ { + _, idx, err := channel.GetNextEnabledKey(excluded) + if err != nil { + return 0, false + } + if lim.MaxConcurrency > 0 && !TryAcquireConcurrency(channelGateKey(channel.Id, idx), lim.MaxConcurrency) { + excluded[idx] = true + continue + } + ReleaseConcurrency(channelGateKey(channel.Id, idx)) + return idx, true + } + return 0, false +} + +// AcquireChannelWithLimits tries to reserve the concurrency slots for a +// specific channel, such as an Affinity hit. A false result means the caller +// should temporarily fall back to normal channel selection. +func AcquireChannelWithLimits(ctx *gin.Context, channel *model.Channel) (*GateHandle, bool) { + handle := &GateHandle{} + lim := GetChannelLimits(channel) + if lim.Enabled && lim.MaxConcurrency > 0 && !handle.acquire(channelGateKey(channel.Id, -1), lim.MaxConcurrency) { + return handle, false + } + + if !channel.ChannelInfo.IsMultiKey || !lim.Enabled { + return handle, true + } + keyIdx, ok := selectHealthyKey(channel, lim) + if !ok { + handle.Release() + return handle, false + } + if ctx != nil { + common.SetContextKey(ctx, constant.ContextKeyChannelPreSelectedKeyIdx, keyIdx) + } + if lim.MaxConcurrency > 0 && !handle.acquire(channelGateKey(channel.Id, keyIdx), lim.MaxConcurrency) { + handle.Release() + if ctx != nil { + common.SetContextKey(ctx, constant.ContextKeyChannelPreSelectedKeyIdx, nil) + } + return handle, false + } + return handle, true +} + +// SelectChannelWithLimits chooses the least-loaded eligible channel in the +// selected priority tier and atomically acquires its concurrency slots. When a +// whole tier is full it falls through to the next lower priority. +func SelectChannelWithLimits(param *RetryParam) (*model.Channel, string, *GateHandle, error) { + excluded := make(map[int]bool, len(param.ExcludeIDs)) + for _, id := range param.ExcludeIDs { + excluded[id] = true + } + handle := &GateHandle{} + + groups := []string{param.TokenGroup} + startGroupIndex := 0 + if param.TokenGroup == "auto" { + userGroup := common.GetContextKeyString(param.Ctx, constant.ContextKeyUserGroup) + groups = GetUserAutoGroup(userGroup) + if len(groups) == 0 { + return nil, param.TokenGroup, handle, errors.New("auto groups is not enabled") + } + if index, exists := common.GetContextKey(param.Ctx, constant.ContextKeyAutoGroupIndex); exists { + if value, ok := index.(int); ok && value >= 0 && value < len(groups) { + startGroupIndex = value + } + } + } + + for groupIndex := startGroupIndex; groupIndex < len(groups); groupIndex++ { + selectGroup := groups[groupIndex] + param.ExcludeIDs = mapKeys(excluded) + candidates, err := model.GetSatisfiedChannels(selectGroup, param.ModelName, param.RequestPath, param.ExcludeIDs) + if err != nil { + return nil, selectGroup, handle, err + } + if len(candidates) == 0 { + continue + } + + priorityTiers := make([][]*model.Channel, 0) + for _, candidate := range candidates { + if len(priorityTiers) == 0 || priorityTiers[len(priorityTiers)-1][0].GetPriority() != candidate.GetPriority() { + priorityTiers = append(priorityTiers, []*model.Channel{candidate}) + continue + } + priorityTiers[len(priorityTiers)-1] = append(priorityTiers[len(priorityTiers)-1], candidate) + } + + startPriority := param.GetRetry() + if groupIndex > startGroupIndex { + startPriority = 0 + } + if startPriority >= len(priorityTiers) { + startPriority = len(priorityTiers) - 1 + } + for priorityIndex := startPriority; priorityIndex < len(priorityTiers); priorityIndex++ { + tier := priorityTiers[priorityIndex] + for len(tier) > 0 { + leastLoaded := leastLoadedChannels(tier) + if len(leastLoaded) == 0 { + break + } + + channel := leastLoaded[rand.Intn(len(leastLoaded))] + lim := GetChannelLimits(channel) + if evaluateChannelForLimits(channel.Id, -1, lim, excluded) { + tier = excludeChannels(tier, channel.Id) + continue + } + + if channel.ChannelInfo.IsMultiKey && lim.Enabled { + keyIdx, ok := selectHealthyKey(channel, lim) + if !ok { + if lim.MaxConcurrency > 0 { + ReleaseConcurrency(channelGateKey(channel.Id, -1)) + } + excluded[channel.Id] = true + tier = excludeChannels(tier, channel.Id) + continue + } + common.SetContextKey(param.Ctx, constant.ContextKeyChannelPreSelectedKeyIdx, keyIdx) + if lim.MaxConcurrency > 0 { + keyGate := channelGateKey(channel.Id, keyIdx) + if !handle.acquire(keyGate, lim.MaxConcurrency) { + ReleaseConcurrency(channelGateKey(channel.Id, -1)) + excluded[channel.Id] = true + tier = excludeChannels(tier, channel.Id) + common.SetContextKey(param.Ctx, constant.ContextKeyChannelPreSelectedKeyIdx, nil) + continue + } + } + } + if lim.Enabled && lim.MaxConcurrency > 0 { + handle.GateKeys = append(handle.GateKeys, channelGateKey(channel.Id, -1)) + } + if param.TokenGroup == "auto" { + common.SetContextKey(param.Ctx, constant.ContextKeyAutoGroup, selectGroup) + common.SetContextKey(param.Ctx, constant.ContextKeyAutoGroupIndex, groupIndex) + } + param.ExcludeIDs = mapKeys(excluded) + return channel, selectGroup, handle, nil + } + } + if param.TokenGroup == "auto" { + common.SetContextKey(param.Ctx, constant.ContextKeyAutoGroupIndex, groupIndex+1) + } + } + return nil, param.TokenGroup, handle, nil +} + +func excludeChannels(channels []*model.Channel, channelID int) []*model.Channel { + filtered := make([]*model.Channel, 0, len(channels)-1) + for _, channel := range channels { + if channel.Id != channelID { + filtered = append(filtered, channel) + } + } + return filtered +} + +func mapKeys(m map[int]bool) []int { + out := make([]int, 0, len(m)) + for key := range m { + out = append(out, key) + } + return out +} diff --git a/service/channel_limit_select_test.go b/service/channel_limit_select_test.go new file mode 100644 index 000000000000..cefac8dbfe3d --- /dev/null +++ b/service/channel_limit_select_test.go @@ -0,0 +1,192 @@ +package service + +import ( + "fmt" + "testing" + + "github.com/QuantumNous/new-api/common" + "github.com/QuantumNous/new-api/model" + "github.com/QuantumNous/new-api/setting/operation_setting" + "github.com/gin-gonic/gin" + "github.com/stretchr/testify/require" +) + +// TestSelectChannelWithLimits_GateHandleReleaseIsSafe exercises the +// GateHandle.Release contract that the relay loop depends on: the handle must +// be safe to Release even when no slot was ever acquired (nil receiver, empty +// GateKeys), and Release must hand the slot back so the gate is reusable. We +// drive the gate helpers directly to avoid spinning up the channel cache (the +// orchestrator's full loop still needs integration coverage elsewhere). +func TestSelectChannelWithLimits_GateHandleReleaseIsSafe(t *testing.T) { + // nil receiver must not panic + var h *GateHandle // nil + h.Release() + require.Nil(t, h) + + // empty GateKeys slice — Release is a no-op + h = &GateHandle{} + h.Release() + require.Nil(t, h.GateKeys) + + // acquire one slot via TryAcquireConcurrency, record it on a handle, + // Release it, and assert the slot is reusable by a fresh acquire. + const key = "test:safe-release" + require.True(t, TryAcquireConcurrency(key, 1), "first acquire must succeed") + + h = &GateHandle{GateKeys: []string{key}} + require.False(t, TryAcquireConcurrency(key, 1), "gate must be full before Release") + + h.Release() + // After release, a fresh acquire must succeed. + require.True(t, TryAcquireConcurrency(key, 1), "slot must be reusable after Release") + ReleaseConcurrency(key) +} + +func TestSelectChannelWithLimits_FailedConcurrencyAcquireSkipsAndExcludes(t *testing.T) { + // fill the per-channel semaphore to capacity + const ch = 9302 + gate := channelGateKey(ch, -1) + require.True(t, TryAcquireConcurrency(gate, 1)) + t.Cleanup(func() { ReleaseConcurrency(gate) }) + + excluded := map[int]bool{} + lim := ChannelLimits{Enabled: true, MaxConcurrency: 1} + skip := evaluateChannelForLimits(ch, -1, lim, excluded) + require.True(t, skip) + require.True(t, excluded[ch]) +} + +func TestSelectChannelWithLimits_DisabledLimitAllowsEverything(t *testing.T) { + excluded := map[int]bool{} + lim := ChannelLimits{Enabled: false} + skip := evaluateChannelForLimits(9303, -1, lim, excluded) + require.False(t, skip) +} + +func TestLeastLoadedChannelsReturnsOnlyMinimumOccupancy(t *testing.T) { + setting := `{"max_concurrency":3}` + channels := []*model.Channel{ + {Id: 9411, Setting: common.GetPointer(setting)}, + {Id: 9412, Setting: common.GetPointer(setting)}, + {Id: 9413, Setting: common.GetPointer(setting)}, + } + require.True(t, TryAcquireConcurrency(channelGateKey(9411, -1), 3)) + require.True(t, TryAcquireConcurrency(channelGateKey(9413, -1), 3)) + t.Cleanup(func() { + ReleaseConcurrency(channelGateKey(9411, -1)) + ReleaseConcurrency(channelGateKey(9413, -1)) + }) + + leastLoaded := leastLoadedChannels(channels) + require.Len(t, leastLoaded, 1) + require.Equal(t, 9412, leastLoaded[0].Id) +} + +func TestLeastLoadedChannelsExcludesFullChannels(t *testing.T) { + setting := `{"max_concurrency":1}` + channels := []*model.Channel{ + {Id: 9421, Setting: common.GetPointer(setting)}, + {Id: 9422, Setting: common.GetPointer(setting)}, + } + require.True(t, TryAcquireConcurrency(channelGateKey(9421, -1), 1)) + t.Cleanup(func() { ReleaseConcurrency(channelGateKey(9421, -1)) }) + + leastLoaded := leastLoadedChannels(channels) + require.Len(t, leastLoaded, 1) + require.Equal(t, 9422, leastLoaded[0].Id) +} + +func insertChannelSelectionCandidate(t *testing.T, channelID int, priority int64, maxConcurrency int) { + t.Helper() + setting := fmt.Sprintf(`{"max_concurrency":%d}`, maxConcurrency) + require.NoError(t, model.DB.Create(&model.Channel{ + Id: channelID, + Name: fmt.Sprintf("channel-%d", channelID), + Key: fmt.Sprintf("key-%d", channelID), + Status: common.ChannelStatusEnabled, + Priority: &priority, + Setting: common.GetPointer(setting), + }).Error) + require.NoError(t, model.DB.Create(&model.Ability{ + Group: "glm-routing-test", + Model: "glm-4", + ChannelId: channelID, + Enabled: true, + Priority: &priority, + Weight: 100, + }).Error) +} + +func prepareChannelSelectionTest(t *testing.T, channelIDs ...int) { + t.Helper() + require.NoError(t, model.DB.AutoMigrate(&model.Ability{})) + originalMemoryCacheEnabled := common.MemoryCacheEnabled + common.MemoryCacheEnabled = false + originalLimitEnabled := operation_setting.GetChannelLimitSetting().Enabled + operation_setting.GetChannelLimitSetting().Enabled = true + t.Cleanup(func() { + model.DB.Where("channel_id IN ?", channelIDs).Delete(&model.Ability{}) + model.DB.Where("id IN ?", channelIDs).Delete(&model.Channel{}) + common.MemoryCacheEnabled = originalMemoryCacheEnabled + operation_setting.GetChannelLimitSetting().Enabled = originalLimitEnabled + }) +} + +func TestSelectChannelWithLimits_SelectsLeastLoadedChannel(t *testing.T) { + prepareChannelSelectionTest(t, 9431, 9432, 9433) + insertChannelSelectionCandidate(t, 9431, 10, 2) + insertChannelSelectionCandidate(t, 9432, 10, 2) + insertChannelSelectionCandidate(t, 9433, 10, 2) + require.True(t, TryAcquireConcurrency(channelGateKey(9431, -1), 2)) + require.True(t, TryAcquireConcurrency(channelGateKey(9433, -1), 2)) + t.Cleanup(func() { + ReleaseConcurrency(channelGateKey(9431, -1)) + ReleaseConcurrency(channelGateKey(9433, -1)) + }) + + channel, _, handle, err := SelectChannelWithLimits(&RetryParam{ + Ctx: &gin.Context{}, + TokenGroup: "glm-routing-test", + ModelName: "glm-4", + }) + require.NoError(t, err) + require.NotNil(t, channel) + require.Equal(t, 9432, channel.Id) + handle.Release() +} + +func TestSelectChannelWithLimits_FallsThroughWhenTopPriorityIsFull(t *testing.T) { + prepareChannelSelectionTest(t, 9441, 9442) + insertChannelSelectionCandidate(t, 9441, 10, 1) + insertChannelSelectionCandidate(t, 9442, 5, 1) + require.True(t, TryAcquireConcurrency(channelGateKey(9441, -1), 1)) + t.Cleanup(func() { ReleaseConcurrency(channelGateKey(9441, -1)) }) + + channel, _, handle, err := SelectChannelWithLimits(&RetryParam{ + Ctx: &gin.Context{}, + TokenGroup: "glm-routing-test", + ModelName: "glm-4", + }) + require.NoError(t, err) + require.NotNil(t, channel) + require.Equal(t, 9442, channel.Id) + handle.Release() +} + +func TestAcquireChannelWithLimitsUsesSameChannelGate(t *testing.T) { + setting := `{"max_concurrency":1}` + channel := &model.Channel{Id: 9451, Setting: common.GetPointer(setting)} + + handle, acquired := AcquireChannelWithLimits(&gin.Context{}, channel) + require.True(t, acquired) + require.NotNil(t, handle) + + secondHandle, acquired := AcquireChannelWithLimits(&gin.Context{}, channel) + require.False(t, acquired) + require.NotNil(t, secondHandle) + + handle.Release() + thirdHandle, acquired := AcquireChannelWithLimits(&gin.Context{}, channel) + require.True(t, acquired) + thirdHandle.Release() +} diff --git a/service/channel_limit_test.go b/service/channel_limit_test.go new file mode 100644 index 000000000000..7285efaefcc0 --- /dev/null +++ b/service/channel_limit_test.go @@ -0,0 +1,35 @@ +package service + +import ( + "testing" + + "github.com/QuantumNous/new-api/common" + "github.com/QuantumNous/new-api/model" + "github.com/QuantumNous/new-api/setting/operation_setting" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestGetChannelLimits_DisabledGlobal(t *testing.T) { + orig := operation_setting.GetChannelLimitSetting().Enabled + operation_setting.GetChannelLimitSetting().Enabled = false + t.Cleanup(func() { operation_setting.GetChannelLimitSetting().Enabled = orig }) + + ch := &model.Channel{} + l := GetChannelLimits(ch) + assert.False(t, l.Enabled) +} + +func TestGetChannelLimits_ChannelValueOverridesDefault(t *testing.T) { + g := operation_setting.GetChannelLimitSetting() + orig := g.Enabled + t.Cleanup(func() { g.Enabled = orig }) + g.Enabled = true + + settingJSON := `{"max_concurrency":5}` + ch := &model.Channel{Id: 1, Setting: common.GetPointer(settingJSON)} + + l := GetChannelLimits(ch) + require.True(t, l.Enabled) + assert.Equal(t, 5, l.MaxConcurrency) +} diff --git a/service/channel_select.go b/service/channel_select.go index 24c4e252bfb3..73601c5848ac 100644 --- a/service/channel_select.go +++ b/service/channel_select.go @@ -17,6 +17,7 @@ type RetryParam struct { ModelName string RequestPath string Retry *int + ExcludeIDs []int resetNextTry bool } @@ -116,7 +117,7 @@ func CacheGetRandomSatisfiedChannel(param *RetryParam) (*model.Channel, string, } logger.LogDebug(param.Ctx, "Auto selecting group: %s, priorityRetry: %d", autoGroup, priorityRetry) - channel, _ = model.GetRandomSatisfiedChannel(autoGroup, param.ModelName, priorityRetry, param.RequestPath) + channel, _ = model.GetRandomSatisfiedChannel(autoGroup, param.ModelName, priorityRetry, param.RequestPath, param.ExcludeIDs) if channel == nil { // Current group has no available channel for this model, try next group // 当前分组没有该模型的可用渠道，尝试下一个分组 @@ -154,7 +155,7 @@ func CacheGetRandomSatisfiedChannel(param *RetryParam) (*model.Channel, string, break } } else { - channel, err = model.GetRandomSatisfiedChannel(param.TokenGroup, param.ModelName, param.GetRetry(), param.RequestPath) + channel, err = model.GetRandomSatisfiedChannel(param.TokenGroup, param.ModelName, param.GetRetry(), param.RequestPath, param.ExcludeIDs) if err != nil { return nil, param.TokenGroup, err } diff --git a/setting/operation_setting/channel_limit_setting.go b/setting/operation_setting/channel_limit_setting.go new file mode 100644 index 000000000000..e71e57799114 --- /dev/null +++ b/setting/operation_setting/channel_limit_setting.go @@ -0,0 +1,24 @@ +package operation_setting + +import ( + "github.com/QuantumNous/new-api/setting/config" +) + +// ChannelLimitSetting holds the global toggle for per-channel concurrency control. +// Registered with config.GlobalConfig so it auto-loads/saves from the options +// table under key "channel_limit_setting.enabled". +type ChannelLimitSetting struct { + Enabled bool `json:"enabled"` +} + +var channelLimitSetting = ChannelLimitSetting{ + Enabled: true, +} + +func init() { + config.GlobalConfig.Register("channel_limit_setting", &channelLimitSetting) +} + +func GetChannelLimitSetting() *ChannelLimitSetting { + return &channelLimitSetting +} diff --git a/web/default/src/features/channels/components/drawers/channel-mutate-drawer.tsx b/web/default/src/features/channels/components/drawers/channel-mutate-drawer.tsx index b3aa0672983f..f375b3fd8e6a 100644 --- a/web/default/src/features/channels/components/drawers/channel-mutate-drawer.tsx +++ b/web/default/src/features/channels/components/drawers/channel-mutate-drawer.tsx @@ -3658,6 +3658,29 @@ export function ChannelMutateDrawer({ )} /> + ( + + {t('Max Concurrency')} + + + field.onChange(Number(e.target.value)) + } + /> + + + {t(FIELD_DESCRIPTIONS.MAX_CONCURRENCY)} + + + + )} + /> { const disableParsed = parseHttpStatusCodeRules( @@ -127,6 +130,7 @@ type RoutingReliabilitySectionProps = { 'monitor_setting.auto_test_channel_enabled': boolean 'monitor_setting.auto_test_channel_minutes': number 'monitor_setting.channel_test_mode': ChannelTestMode + 'channel_limit_setting.enabled': boolean } } @@ -145,6 +149,7 @@ type NormalizedRoutingReliabilityValues = { 'monitor_setting.auto_test_channel_enabled': boolean 'monitor_setting.auto_test_channel_minutes': number 'monitor_setting.channel_test_mode': ChannelTestMode + 'channel_limit_setting.enabled': boolean } function normalizeChannelTestMode(value?: string): ChannelTestMode { @@ -172,6 +177,9 @@ const buildFormDefaults = ( defaults['monitor_setting.channel_test_mode'] ), }, + channel_limit_setting: { + enabled: defaults['channel_limit_setting.enabled'] ?? true, + }, }) const normalizeDefaults = ( @@ -197,6 +205,7 @@ const normalizeDefaults = ( 'monitor_setting.channel_test_mode': normalizeChannelTestMode( defaults['monitor_setting.channel_test_mode'] ), + 'channel_limit_setting.enabled': defaults['channel_limit_setting.enabled'], }) const normalizeFormValues = ( @@ -220,6 +229,7 @@ const normalizeFormValues = ( 'monitor_setting.auto_test_channel_minutes': values.monitor_setting.auto_test_channel_minutes, 'monitor_setting.channel_test_mode': values.monitor_setting.channel_test_mode, + 'channel_limit_setting.enabled': values.channel_limit_setting.enabled, }) export function RoutingReliabilitySection({ @@ -480,6 +490,38 @@ export function RoutingReliabilitySection({ +
+
+

{t('Channel Limits')}

+
+
+ ( + + + {t('Channel Limit Enabled')} + + {t( + 'Master switch for per-channel concurrency limits' + )} + + + + + + + )} + /> +
+
+ + +

{t('Auto-disable rules')}

diff --git a/web/default/src/features/system-settings/models/section-registry.tsx b/web/default/src/features/system-settings/models/section-registry.tsx index c21046202ce4..88dc9eeaa305 100644 --- a/web/default/src/features/system-settings/models/section-registry.tsx +++ b/web/default/src/features/system-settings/models/section-registry.tsx @@ -85,6 +85,8 @@ const MODELS_SECTIONS = [ settings['monitor_setting.auto_test_channel_minutes'], 'monitor_setting.channel_test_mode': settings['monitor_setting.channel_test_mode'], + 'channel_limit_setting.enabled': + settings['channel_limit_setting.enabled'], }} /> ), diff --git a/web/default/src/features/system-settings/types.ts b/web/default/src/features/system-settings/types.ts index 11c51f08adc3..7fc6c6f6c8b4 100644 --- a/web/default/src/features/system-settings/types.ts +++ b/web/default/src/features/system-settings/types.ts @@ -235,6 +235,7 @@ export type ModelSettings = { 'monitor_setting.auto_test_channel_enabled': boolean 'monitor_setting.auto_test_channel_minutes': number 'monitor_setting.channel_test_mode': 'scheduled_all' | 'passive_recovery' + 'channel_limit_setting.enabled': boolean 'channel_affinity_setting.enabled': boolean 'channel_affinity_setting.switch_on_success': boolean 'channel_affinity_setting.keep_on_channel_disabled': boolean diff --git a/web/default/src/i18n/locales/en.json b/web/default/src/i18n/locales/en.json index 9eba821e391d..929d7c4d65ed 100644 --- a/web/default/src/i18n/locales/en.json +++ b/web/default/src/i18n/locales/en.json @@ -2025,7 +2025,7 @@ "footer.columns.related.links.oneApi": "One API", "footer.columns.related.title": "Related Projects", "footer.defaultCopyright": "All rights reserved.", - "footer.new\u0061pi.projectAttributionSuffix": "All rights reserved. Designed and developed by the project contributors.", + "footer.newapi.projectAttributionSuffix": "All rights reserved. Designed and developed by the project contributors.", "For channels added after May 10, 2025, no need to remove \".\" from model names during deployment": "For channels added after May 10, 2025, no need to remove \".\" from model names during deployment", "For private deployments, format: https://fastgpt.run/api/openapi": "For private deployments, format: https://fastgpt.run/api/openapi", "Force a syntactically valid JSON response": "Force a syntactically valid JSON response", @@ -5178,6 +5178,11 @@ "Zero retention": "Zero retention", "Zhipu": "Zhipu", "Zhipu V4": "Zhipu V4", - "Zoom": "Zoom" + "Zoom": "Zoom", + "Channel Limit Enabled": "Channel Limit Enabled", + "Channel Limits": "Channel Limits", + "Master switch for per-channel concurrency limits": "Master switch for per-channel concurrency limits", + "Max Concurrency": "Max Concurrency", + "Max concurrent requests to this channel (0 = unlimited)": "Max concurrent requests to this channel (0 = unlimited)" } } diff --git a/web/default/src/i18n/locales/fr.json b/web/default/src/i18n/locales/fr.json index f4014a48f7f2..942d77457cf0 100644 --- a/web/default/src/i18n/locales/fr.json +++ b/web/default/src/i18n/locales/fr.json @@ -2025,7 +2025,7 @@ "footer.columns.related.links.oneApi": "One API", "footer.columns.related.title": "Projets liés", "footer.defaultCopyright": "Tous droits réservés.", - "footer.new\u0061pi.projectAttributionSuffix": "Tous droits réservés. Conçu et développé par les contributeurs du projet.", + "footer.newapi.projectAttributionSuffix": "Tous droits réservés. Conçu et développé par les contributeurs du projet.", "For channels added after May 10, 2025, no need to remove \".\" from model names during deployment": "Pour les canaux ajoutés après le 10 mai 2025, pas besoin de supprimer \".\" des noms de modèles lors du déploiement", "For private deployments, format: https://fastgpt.run/api/openapi": "Pour les déploiements privés, format : https://fastgpt.run/api/openapi", "Force a syntactically valid JSON response": "Imposer une réponse JSON syntaxiquement valide", @@ -5178,6 +5178,11 @@ "Zero retention": "Aucune rétention", "Zhipu": "Zhipu", "Zhipu V4": "Zhipu V4", - "Zoom": "Zoom" + "Zoom": "Zoom", + "Channel Limit Enabled": "Channel Limit Enabled", + "Channel Limits": "Channel Limits", + "Master switch for per-channel concurrency limits": "Interrupteur principal pour les limites de concurrence par canal", + "Max Concurrency": "Max Concurrency", + "Max concurrent requests to this channel (0 = unlimited)": "Max concurrent requests to this channel (0 = unlimited)" } } diff --git a/web/default/src/i18n/locales/ja.json b/web/default/src/i18n/locales/ja.json index 527f49b9f25e..a7234985754e 100644 --- a/web/default/src/i18n/locales/ja.json +++ b/web/default/src/i18n/locales/ja.json @@ -2025,7 +2025,7 @@ "footer.columns.related.links.oneApi": "1つのAPI", "footer.columns.related.title": "関連プロジェクト", "footer.defaultCopyright": "すべての権利を留保します。", - "footer.new\u0061pi.projectAttributionSuffix": "すべての権利を留保します。プロジェクトコントリビューターにより設計・開発されています。", + "footer.newapi.projectAttributionSuffix": "すべての権利を留保します。プロジェクトコントリビューターにより設計・開発されています。", "For channels added after May 10, 2025, no need to remove \".\" from model names during deployment": "2025 年 5 月 10 日以降に追加されたチャネルの場合、デプロイ時にモデル名から「.」を削除する必要はありません", "For private deployments, format: https://fastgpt.run/api/openapi": "プライベートデプロイメントの場合、形式: https://fastgpt.run/api/openapi", "Force a syntactically valid JSON response": "構文的に有効な JSON 応答を強制", @@ -5178,6 +5178,11 @@ "Zero retention": "データ保持なし", "Zhipu": "Zhipu", "Zhipu V4": "Zhipu V 4", - "Zoom": "ズーム" + "Zoom": "ズーム", + "Channel Limit Enabled": "Channel Limit Enabled", + "Channel Limits": "Channel Limits", + "Master switch for per-channel concurrency limits": "チャネルごとの同時実行制限のメインスイッチ", + "Max Concurrency": "Max Concurrency", + "Max concurrent requests to this channel (0 = unlimited)": "Max concurrent requests to this channel (0 = unlimited)" } } diff --git a/web/default/src/i18n/locales/ru.json b/web/default/src/i18n/locales/ru.json index 0c424ad55de2..bb46cd98d6d2 100644 --- a/web/default/src/i18n/locales/ru.json +++ b/web/default/src/i18n/locales/ru.json @@ -2025,7 +2025,7 @@ "footer.columns.related.links.oneApi": "Один API", "footer.columns.related.title": "Связанные проекты", "footer.defaultCopyright": "Все права защищены.", - "footer.new\u0061pi.projectAttributionSuffix": "Все права защищены. Разработано участниками проекта.", + "footer.newapi.projectAttributionSuffix": "Все права защищены. Разработано участниками проекта.", "For channels added after May 10, 2025, no need to remove \".\" from model names during deployment": "Для каналов, добавленных после 10 мая 2025 г., не нужно удалять \".\" из имён моделей при развёртывании", "For private deployments, format: https://fastgpt.run/api/openapi": "Для частных развертываний, формат: https://fastgpt.run/api/openapi", "Force a syntactically valid JSON response": "Принудительно возвращать синтаксически корректный JSON", @@ -5178,6 +5178,11 @@ "Zero retention": "Без хранения данных", "Zhipu": "Zhipu", "Zhipu V4": "Zhipu V4", - "Zoom": "Zoom" + "Zoom": "Zoom", + "Channel Limit Enabled": "Channel Limit Enabled", + "Channel Limits": "Channel Limits", + "Master switch for per-channel concurrency limits": "Главный переключатель ограничений параллелизма канала", + "Max Concurrency": "Max Concurrency", + "Max concurrent requests to this channel (0 = unlimited)": "Max concurrent requests to this channel (0 = unlimited)" } } diff --git a/web/default/src/i18n/locales/vi.json b/web/default/src/i18n/locales/vi.json index 17fc695a8471..7499955ca020 100644 --- a/web/default/src/i18n/locales/vi.json +++ b/web/default/src/i18n/locales/vi.json @@ -2025,7 +2025,7 @@ "footer.columns.related.links.oneApi": "One API", "footer.columns.related.title": "Các Dự Án Liên Quan", "footer.defaultCopyright": "Bản quyền được bảo lưu.", - "footer.new\u0061pi.projectAttributionSuffix": "Bản quyền được bảo lưu. Được thiết kế và phát triển bởi các cộng tác viên dự án.", + "footer.newapi.projectAttributionSuffix": "Bản quyền được bảo lưu. Được thiết kế và phát triển bởi các cộng tác viên dự án.", "For channels added after May 10, 2025, no need to remove \".\" from model names during deployment": "Đối với các kênh được thêm sau ngày 10 tháng 5 năm 2025, không cần loại bỏ \".\" khỏi tên mô hình trong quá trình triển khai", "For private deployments, format: https://fastgpt.run/api/openapi": "Đối với các triển khai riêng tư, định dạng: https://fastgpt.run/api/openapi", "Force a syntactically valid JSON response": "Buộc phản hồi JSON hợp lệ về cú pháp", @@ -5178,6 +5178,11 @@ "Zero retention": "Không lưu dữ liệu", "Zhipu": "Zhipu", "Zhipu V4": "Zhipu V4", - "Zoom": "Zoom" + "Zoom": "Zoom", + "Channel Limit Enabled": "Channel Limit Enabled", + "Channel Limits": "Channel Limits", + "Master switch for per-channel concurrency limits": "Công tắc chính cho giới hạn đồng thời mỗi kênh", + "Max Concurrency": "Max Concurrency", + "Max concurrent requests to this channel (0 = unlimited)": "Max concurrent requests to this channel (0 = unlimited)" } } diff --git a/web/default/src/i18n/locales/zh.json b/web/default/src/i18n/locales/zh.json index 45368da4da94..7d5ea2b65aff 100644 --- a/web/default/src/i18n/locales/zh.json +++ b/web/default/src/i18n/locales/zh.json @@ -2025,7 +2025,7 @@ "footer.columns.related.links.oneApi": "One API", "footer.columns.related.title": "相关项目", "footer.defaultCopyright": "版权所有。", - "footer.new\u0061pi.projectAttributionSuffix": "版权所有，由项目贡献者设计与开发。", + "footer.newapi.projectAttributionSuffix": "版权所有，由项目贡献者设计与开发。", "For channels added after May 10, 2025, no need to remove \".\" from model names during deployment": "对于 2025 年 5 月 10 日之后添加的渠道，在部署时无需从模型名称中移除 \".\"", "For private deployments, format: https://fastgpt.run/api/openapi": "对于私有部署，格式为：https://fastgpt.run/api/openapi", "Force a syntactically valid JSON response": "强制返回语法合法的 JSON", @@ -5178,6 +5178,11 @@ "Zero retention": "零数据保留", "Zhipu": "智谱", "Zhipu V4": "智谱 V4", - "Zoom": "缩放" + "Zoom": "缩放", + "Channel Limit Enabled": "启用渠道限流", + "Channel Limits": "渠道限流", + "Master switch for per-channel concurrency limits": "每个渠道并发限制的总开关", + "Max Concurrency": "最大并发", + "Max concurrent requests to this channel (0 = unlimited)": "该渠道的最大并发请求数（0=不限）" } }