feat: add channel affinity exclusive lock feature - #4048
Closed
rty813 wants to merge 1 commit into
Closed
Conversation
Allow per-channel "affinity exclusive" setting. When enabled, a token bound to a channel via affinity binding will exclusively occupy it — other tokens cannot use the channel until the lock expires (same TTL as the affinity binding). Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Contributor
WalkthroughThis PR introduces an "exclusive channel affinity" feature allowing tokens to exclusively occupy channels when enabled. Requests to exclusively-locked channels are rejected with a 403 error, and lock acquisition is attempted via Redis Lua script (with mutex fallback) before recording affinity. Changes
Sequence Diagram(s)sequenceDiagram
participant Client
participant Distributor as Middleware:<br/>Distributor
participant AffinityService as Service:<br/>Affinity
participant ChannelSelect as Service:<br/>Channel Selection
participant DB as Redis/Database
Client->>Distributor: Request with specific channel_id
Distributor->>AffinityService: IsChannelAffinityExclusiveAvailable(channel_id, token_id)
AffinityService->>DB: Check exclusive lock holder
DB-->>AffinityService: Lock holder (if locked)
alt Locked by different token
AffinityService-->>Distributor: false
Distributor-->>Client: 403 Forbidden (channel_exclusive_locked)
else Available or locked by current token
AffinityService-->>Distributor: true
Distributor->>ChannelSelect: GetRandomSatisfiedChannel with affinity_exclusive filter
ChannelSelect-->>Distributor: Channel
Distributor->>Client: 200 OK (with selected channel)
end
sequenceDiagram
participant Distributor as Middleware:<br/>Distributor
participant AffinityService as Service:<br/>Channel Affinity
participant Redis
participant Cache as In-Memory Cache
Distributor->>AffinityService: RecordChannelAffinity(channel_id, token_id)
alt Channel has AffinityExclusive enabled
AffinityService->>Redis: Lua Script: SET exclusive lock (atomic)
Redis-->>AffinityService: Lock acquired or already held
alt Lock acquired
AffinityService->>Cache: Set affinity binding mapping
AffinityService->>Cache: Record normal affinity entry
AffinityService-->>Distributor: Success
else Lock not acquired (held by other token)
AffinityService-->>Distributor: Skip recording (lock conflict)
end
else AffinityExclusive not enabled
AffinityService->>Cache: Cleanup exclusive bindings for key
AffinityService->>Cache: Maybe release exclusive lock
AffinityService->>Cache: Record normal affinity entry
AffinityService-->>Distributor: Success
end
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Possibly related PRs
Suggested reviewers
🚥 Pre-merge checks | ✅ 2 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Actionable comments posted: 6
🧹 Nitpick comments (1)
service/channel_affinity_exclusive_test.go (1)
12-86: Add a concurrent selection/recording case for the exclusivity guarantee.These tests start after the lock already exists, so they won't catch the race where two requests both pass
IsChannelAffinityExclusiveAvailable()beforeRecordChannelAffinity()writes the lock. A small concurrent test around the full selection/record flow would pin down the behavior this feature needs.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@service/channel_affinity_exclusive_test.go` around lines 12 - 86, Add a concurrent test that races the full select+record flow to ensure exclusivity: create a new test (mirroring setup in TestChannelAffinityExclusiveLockRejectsOtherToken) that disables Redis, clears caches, then spawn N goroutines which all concurrently call IsChannelAffinityExclusiveAvailable(channelID, tokenCandidate) and when available immediately call the public RecordChannelAffinity or TrySetChannelAffinityExclusiveLock path used in production; use sync.WaitGroup and a start channel to align starts, increment an atomic counter for successful records, and assert exactly one goroutine succeeded and the stored holder (GetChannelAffinityExclusiveLockHolder) matches that winner; ensure cleanup restores common.RedisEnabled and clears caches.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@controller/channel_affinity_cache.go`:
- Around line 96-103: After parsing channelIDStr with strconv.Atoi into
channelID, add a validation that channelID is a positive integer (>0); if
channelID <= 0 return the same HTTP 400 JSON response (success: false, message:
"channel_id 必须是数字") and stop processing. Update the handler around the
channelID, channelIDStr and strconv.Atoi logic so negative values and zero are
rejected before the value is used downstream.
In `@service/channel_affinity.go`:
- Around line 1144-1157: The current code acquires the exclusive affinity lock
inside RecordChannelAffinity after c.Next(), which is too late; instead call
TrySetChannelAffinityExclusiveLock for the chosen channel/token during the
selection/distribution phase (before the request is executed) and populate
ChannelAffinityExclusiveBinding there so the request is reserved; remove or skip
the post-hoc lock attempt in RecordChannelAffinity to avoid race wins, and
ensure you release the lock if distribution fails (or on timeout) by invoking
the corresponding release/cleanup path when distribution/selection returns an
error or decision not to use the binding.
- Around line 518-531: ClearChannelAffinityExclusiveCacheByChannelID currently
returns bindingDeleted > 0 when cache.DeleteMany errors, which can report
success even though the exclusive lock wasn't removed; change the behavior so
that if getChannelAffinityExclusiveCache().DeleteMany(...) returns an error you
do NOT report success—handle the error path by returning false (or otherwise
indicating failure) and optionally log the error; update the error branch in
ClearChannelAffinityExclusiveCacheByChannelID (referencing
getChannelAffinityExclusiveCache, DeleteMany, and
clearChannelAffinityExclusiveBindingCacheByChannelID) so only a confirmed
DeleteMany result that includes a true deleted value leads to returning true.
- Around line 1160-1173: When
setChannelAffinityExclusiveBinding(affinityCacheKey, newBinding, ttl) fails,
roll back the prior writes by removing the cache entry that was set via
cache.SetWithTTL(cacheKey, channelID, ttl) and releasing the exclusive lock if
owned via deleteChannelAffinityExclusiveLockIfOwned(channelID, tokenID); perform
these rollback calls inside the same shouldBindExclusive branch before returning
(log but ignore rollback errors so failure path still returns cleanly).
In `@service/channel_select.go`:
- Line 131: Do not ignore the error returned by model.GetRandomSatisfiedChannel;
replace the blank identifier with an error variable (e.g., err) when calling
GetRandomSatisfiedChannel with autoGroup, param.ModelName, priorityRetry,
filters, and if err is non-nil handle/propagate it the same way as the existing
error path for GetRandomSatisfiedChannel (do not treat it as "no available
channel" or silently fall back) so real failures are not masked.
In `@web/src/i18n/locales/zh-TW.json`:
- Line 2987: The translation entry with key
"开启后，通过亲和性绑定到此渠道的令牌将独占该渠道，其他令牌无法使用此渠道，过期时间同亲和性绑定的 TTL" uses the term "頻道" but
the project standard uses "管道"; update the corresponding value string to replace
all instances of "頻道" with "管道" (so the value becomes
"開啟後，透過親和性綁定到此管道的令牌將獨佔該管道，其他令牌無法使用此管道，過期時間同親和性綁定的 TTL") and scan nearby locale
entries for other occurrences of "頻道" to make the same replacement for
consistency.
---
Nitpick comments:
In `@service/channel_affinity_exclusive_test.go`:
- Around line 12-86: Add a concurrent test that races the full select+record
flow to ensure exclusivity: create a new test (mirroring setup in
TestChannelAffinityExclusiveLockRejectsOtherToken) that disables Redis, clears
caches, then spawn N goroutines which all concurrently call
IsChannelAffinityExclusiveAvailable(channelID, tokenCandidate) and when
available immediately call the public RecordChannelAffinity or
TrySetChannelAffinityExclusiveLock path used in production; use sync.WaitGroup
and a start channel to align starts, increment an atomic counter for successful
records, and assert exactly one goroutine succeeded and the stored holder
(GetChannelAffinityExclusiveLockHolder) matches that winner; ensure cleanup
restores common.RedisEnabled and clears caches.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 2fb46d8c-b79e-4787-accd-36aba6e10cb1
📒 Files selected for processing (21)
controller/channel_affinity_cache.godto/channel_settings.goi18n/keys.goi18n/locales/en.yamli18n/locales/zh-CN.yamli18n/locales/zh-TW.yamlmiddleware/distributor.gomodel/channel_cache.gomodel/channel_cache_test.gorouter/api-router.goservice/channel_affinity.goservice/channel_affinity_exclusive_test.goservice/channel_select.goweb/src/components/table/channels/modals/EditChannelModal.jsxweb/src/i18n/locales/en.jsonweb/src/i18n/locales/fr.jsonweb/src/i18n/locales/ja.jsonweb/src/i18n/locales/ru.jsonweb/src/i18n/locales/vi.jsonweb/src/i18n/locales/zh-CN.jsonweb/src/i18n/locales/zh-TW.json
Comment on lines
+96
to
+103
| channelID, err := strconv.Atoi(channelIDStr) | ||
| if err != nil { | ||
| c.JSON(http.StatusBadRequest, gin.H{ | ||
| "success": false, | ||
| "message": "channel_id 必须是数字", | ||
| }) | ||
| return | ||
| } |
Contributor
There was a problem hiding this comment.
Validate channel_id as a positive integer.
strconv.Atoi accepts 0 and negative values; those should be rejected here to avoid passing invalid channel IDs downstream.
Suggested fix
channelID, err := strconv.Atoi(channelIDStr)
- if err != nil {
+ if err != nil || channelID <= 0 {
c.JSON(http.StatusBadRequest, gin.H{
"success": false,
- "message": "channel_id 必须是数字",
+ "message": "channel_id 必须是正整数",
})
return
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| channelID, err := strconv.Atoi(channelIDStr) | |
| if err != nil { | |
| c.JSON(http.StatusBadRequest, gin.H{ | |
| "success": false, | |
| "message": "channel_id 必须是数字", | |
| }) | |
| return | |
| } | |
| channelID, err := strconv.Atoi(channelIDStr) | |
| if err != nil || channelID <= 0 { | |
| c.JSON(http.StatusBadRequest, gin.H{ | |
| "success": false, | |
| "message": "channel_id 必须是正整数", | |
| }) | |
| return | |
| } |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@controller/channel_affinity_cache.go` around lines 96 - 103, After parsing
channelIDStr with strconv.Atoi into channelID, add a validation that channelID
is a positive integer (>0); if channelID <= 0 return the same HTTP 400 JSON
response (success: false, message: "channel_id 必须是数字") and stop processing.
Update the handler around the channelID, channelIDStr and strconv.Atoi logic so
negative values and zero are rejected before the value is used downstream.
Comment on lines
+518
to
+531
| func ClearChannelAffinityExclusiveCacheByChannelID(channelID int) bool { | ||
| cache := getChannelAffinityExclusiveCache() | ||
| key := strconv.Itoa(channelID) | ||
| bindingDeleted := clearChannelAffinityExclusiveBindingCacheByChannelID(channelID) | ||
| result, err := cache.DeleteMany([]string{key}) | ||
| if err != nil { | ||
| return bindingDeleted > 0 | ||
| } | ||
| for _, deleted := range result { | ||
| if deleted { | ||
| return true | ||
| } | ||
| } | ||
| return bindingDeleted > 0 |
Contributor
There was a problem hiding this comment.
Don't report a successful clear when the lock delete failed.
If DeleteMany() errors, returning bindingDeleted > 0 lets the admin clear endpoint claim success while the exclusive lock can still block traffic. Because the binding is removed first, that stale lock also becomes hard to release before TTL expiry.
🩹 Possible direction
func ClearChannelAffinityExclusiveCacheByChannelID(channelID int) bool {
cache := getChannelAffinityExclusiveCache()
key := strconv.Itoa(channelID)
bindingDeleted := clearChannelAffinityExclusiveBindingCacheByChannelID(channelID)
result, err := cache.DeleteMany([]string{key})
if err != nil {
- return bindingDeleted > 0
+ common.SysError(fmt.Sprintf("channel affinity exclusive cache delete failed: channel=%d, err=%v", channelID, err))
+ return false
}🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@service/channel_affinity.go` around lines 518 - 531,
ClearChannelAffinityExclusiveCacheByChannelID currently returns bindingDeleted >
0 when cache.DeleteMany errors, which can report success even though the
exclusive lock wasn't removed; change the behavior so that if
getChannelAffinityExclusiveCache().DeleteMany(...) returns an error you do NOT
report success—handle the error path by returning false (or otherwise indicating
failure) and optionally log the error; update the error branch in
ClearChannelAffinityExclusiveCacheByChannelID (referencing
getChannelAffinityExclusiveCache, DeleteMany, and
clearChannelAffinityExclusiveBindingCacheByChannelID) so only a confirmed
DeleteMany result that includes a true deleted value leads to returning true.
Comment on lines
+1144
to
+1157
| tokenID := c.GetInt(string(constant.ContextKeyTokenId)) | ||
| shouldBindExclusive := false | ||
| newBinding := ChannelAffinityExclusiveBinding{} | ||
| ch, err := model.CacheGetChannel(channelID) | ||
| if err == nil && ch != nil && ch.GetOtherSettings().AffinityExclusive && tokenID > 0 { | ||
| shouldBindExclusive = true | ||
| newBinding = ChannelAffinityExclusiveBinding{ | ||
| ChannelID: channelID, | ||
| TokenID: tokenID, | ||
| } | ||
| if !TrySetChannelAffinityExclusiveLock(channelID, tokenID, ttl) { | ||
| common.SysLog(fmt.Sprintf("channel affinity exclusive lock skipped: channel=%d, token=%d", channelID, tokenID)) | ||
| return | ||
| } |
Contributor
There was a problem hiding this comment.
Reserve the exclusive lock before the request runs.
middleware/distributor.go only calls RecordChannelAffinity() after c.Next(), so these lines acquire the lock after the channel has already been used. Two concurrent requests can both pass the pre-check and both hit the same exclusive channel; only the first finisher wins the post-hoc lock. Move reservation into selection/distribution and release it on failure.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@service/channel_affinity.go` around lines 1144 - 1157, The current code
acquires the exclusive affinity lock inside RecordChannelAffinity after
c.Next(), which is too late; instead call TrySetChannelAffinityExclusiveLock for
the chosen channel/token during the selection/distribution phase (before the
request is executed) and populate ChannelAffinityExclusiveBinding there so the
request is reserved; remove or skip the post-hoc lock attempt in
RecordChannelAffinity to avoid race wins, and ensure you release the lock if
distribution fails (or on timeout) by invoking the corresponding release/cleanup
path when distribution/selection returns an error or decision not to use the
binding.
Comment on lines
1160
to
+1173
| cache := getChannelAffinityCache() | ||
| if err := cache.SetWithTTL(cacheKey, channelID, time.Duration(ttlSeconds)*time.Second); err != nil { | ||
| if err := cache.SetWithTTL(cacheKey, channelID, ttl); err != nil { | ||
| common.SysError(fmt.Sprintf("channel affinity cache set failed: key=%s, err=%v", cacheKey, err)) | ||
| if shouldBindExclusive && (!oldBindingFound || oldBinding != newBinding) { | ||
| _ = deleteChannelAffinityExclusiveLockIfOwned(channelID, tokenID) | ||
| } | ||
| return | ||
| } | ||
|
|
||
| if shouldBindExclusive { | ||
| if err := setChannelAffinityExclusiveBinding(affinityCacheKey, newBinding, ttl); err != nil { | ||
| common.SysError(fmt.Sprintf("channel affinity exclusive binding set failed: affinity_key=%s, channel=%d, token=%d, err=%v", affinityCacheKey, channelID, tokenID, err)) | ||
| return | ||
| } |
Contributor
There was a problem hiding this comment.
Roll back the affinity entry when binding persistence fails.
By this point the affinity cache entry and the lock are already written. If setChannelAffinityExclusiveBinding() fails, the code returns with no binding record, so later rule-name clears and rebinds cannot discover and release that lock early. Roll back both writes here.
🩹 Possible rollback
if shouldBindExclusive {
if err := setChannelAffinityExclusiveBinding(affinityCacheKey, newBinding, ttl); err != nil {
common.SysError(fmt.Sprintf("channel affinity exclusive binding set failed: affinity_key=%s, channel=%d, token=%d, err=%v", affinityCacheKey, channelID, tokenID, err))
+ _, _ = cache.DeleteMany([]string{cacheKey})
+ _ = deleteChannelAffinityExclusiveLockIfOwned(channelID, tokenID)
return
}🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@service/channel_affinity.go` around lines 1160 - 1173, When
setChannelAffinityExclusiveBinding(affinityCacheKey, newBinding, ttl) fails,
roll back the prior writes by removing the cache entry that was set via
cache.SetWithTTL(cacheKey, channelID, ttl) and releasing the exclusive lock if
owned via deleteChannelAffinityExclusiveLockIfOwned(channelID, tokenID); perform
these rollback calls inside the same shouldBindExclusive branch before returning
(log but ignore rollback errors so failure path still returns cleanly).
| logger.LogDebug(param.Ctx, "Auto selecting group: %s, priorityRetry: %d", autoGroup, priorityRetry) | ||
|
|
||
| channel, _ = model.GetRandomSatisfiedChannel(autoGroup, param.ModelName, priorityRetry) | ||
| channel, _ = model.GetRandomSatisfiedChannel(autoGroup, param.ModelName, priorityRetry, filters...) |
Contributor
There was a problem hiding this comment.
自动分组路径不要吞掉渠道选择错误。
Line 131 当前忽略了 GetRandomSatisfiedChannel 的错误返回，可能把真实故障误判为“当前组无可用渠道”，导致错误被静默掩盖并进入错误的降级路径。建议与 Line 169 一样显式处理错误。
🔧 Suggested fix
- channel, _ = model.GetRandomSatisfiedChannel(autoGroup, param.ModelName, priorityRetry, filters...)
+ channel, err = model.GetRandomSatisfiedChannel(autoGroup, param.ModelName, priorityRetry, filters...)
+ if err != nil {
+ return nil, selectGroup, err
+ }
if channel == nil {📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| channel, _ = model.GetRandomSatisfiedChannel(autoGroup, param.ModelName, priorityRetry, filters...) | |
| channel, err = model.GetRandomSatisfiedChannel(autoGroup, param.ModelName, priorityRetry, filters...) | |
| if err != nil { | |
| return nil, selectGroup, err | |
| } | |
| if channel == nil { |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@service/channel_select.go` at line 131, Do not ignore the error returned by
model.GetRandomSatisfiedChannel; replace the blank identifier with an error
variable (e.g., err) when calling GetRandomSatisfiedChannel with autoGroup,
param.ModelName, priorityRetry, filters, and if err is non-nil handle/propagate
it the same way as the existing error path for GetRandomSatisfiedChannel (do not
treat it as "no available channel" or silently fall back) so real failures are
not masked.
| "无法读取剪贴板": "無法讀取剪貼簿" | ||
| "无法读取剪贴板": "無法讀取剪貼簿", | ||
| "亲和性独占": "親和性獨佔", | ||
| "开启后，通过亲和性绑定到此渠道的令牌将独占该渠道，其他令牌无法使用此渠道，过期时间同亲和性绑定的 TTL": "開啟後，透過親和性綁定到此頻道的令牌將獨佔該頻道，其他令牌無法使用此頻道，過期時間同親和性綁定的 TTL" |
Contributor
There was a problem hiding this comment.
统一术语为“管道”，避免繁中界面文案不一致
Line 2987 使用了“頻道”，但此语言包整体对 渠道 基本使用“管道”，建议统一。
✏️ Suggested change
- "开启后，通过亲和性绑定到此渠道的令牌将独占该渠道，其他令牌无法使用此渠道，过期时间同亲和性绑定的 TTL": "開啟後，透過親和性綁定到此頻道的令牌將獨佔該頻道，其他令牌無法使用此頻道，過期時間同親和性綁定的 TTL"
+ "开启后，通过亲和性绑定到此渠道的令牌将独占该渠道，其他令牌无法使用此渠道，过期时间同亲和性绑定的 TTL": "開啟後，透過親和性綁定到此管道的令牌將獨佔該管道，其他令牌無法使用此管道，過期時間同親和性綁定的 TTL"🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@web/src/i18n/locales/zh-TW.json` at line 2987, The translation entry with key
"开启后，通过亲和性绑定到此渠道的令牌将独占该渠道，其他令牌无法使用此渠道，过期时间同亲和性绑定的 TTL" uses the term "頻道" but
the project standard uses "管道"; update the corresponding value string to replace
all instances of "頻道" with "管道" (so the value becomes
"開啟後，透過親和性綁定到此管道的令牌將獨佔該管道，其他令牌無法使用此管道，過期時間同親和性綁定的 TTL") and scan nearby locale
entries for other occurrences of "頻道" to make the same replacement for
consistency.
Collaborator
|
没有用pr模版，一个commit，21个文件更改，近千行代码变更，没有关联issue沟通，我也没看懂是在做什么，还标记的 🤖 Generated with Claude Code ，而且按照他生成的Test Plan也没做。 
这让人从何看起
|
Author
|
抱歉，我先关闭 |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Changes
dto/channel_settings.go:ChannelOtherSettings新增AffinityExclusive字段service/channel_affinity.go: 独占锁缓存、TrySetChannelAffinityExclusiveLock、IsChannelAffinityExclusiveAvailable等核心逻辑model/channel_cache.go:GetRandomSatisfiedChannel支持ChannelFilter过滤service/channel_select.go: 随机选择路径传递独占过滤器middleware/distributor.go: 亲和性路径 + 指定渠道路径检查独占锁controller/channel_affinity_cache.go+router/api-router.go: 管理后台独占锁查看/清理接口EditChannelModal.jsx: 前端高级设置添加"亲和性独占"开关Test plan
go build ./...编译通过cd web && bun run build前端构建通过🤖 Generated with Claude Code
Summary by CodeRabbit
New Features
Tests