feat: add channel RPM and concurrency admission control - #6583
feat: add channel RPM and concurrency admission control#6583faithleysath wants to merge 1 commit into
Conversation
WalkthroughThis change adds channel-level concurrency and rolling RPM limits. It introduces Redis and memory admission leases, capacity-aware selection, middleware and relay lifecycle handling, structured 429 errors, validation, documentation, localization, and channel configuration UI controls. ChangesChannel admission control
Estimated code review effort: 5 (Critical) | ~120 minutes Possibly related PRs
Sequence Diagram(s)sequenceDiagram
participant Client
participant Distributor
participant ChannelSelector
participant AdmissionService
participant Relay
participant ChannelProvider
Client->>Distributor: submit request
Distributor->>ChannelSelector: resolve channel
ChannelSelector->>AdmissionService: acquire lease
AdmissionService-->>ChannelSelector: lease or capacity error
Distributor->>Relay: forward request
Relay->>ChannelProvider: submit upstream request
ChannelProvider-->>Relay: response or error
Relay->>AdmissionService: commit or release lease
Relay-->>Client: return response
Poem
🚥 Pre-merge checks | ✅ 3 | ❌ 2❌ Failed checks (2 warnings)
✅ Passed checks (3 passed)
✨ Finishing Touches 💡 1🛠️ Fix failing CI checks 💡
🧪 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}
There was a problem hiding this comment.
Actionable comments posted: 2
🧹 Nitpick comments (8)
service/channel_admission_test.go (1)
424-432: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueMove this test next to the code it protects.
TestPickWeightedChannelCandidatePreservesZeroWeightSemanticsasserts behavior ofmodel.PickWeightedChannelCandidate. A test in themodelpackage keeps ownership clear.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@service/channel_admission_test.go` around lines 424 - 432, Move TestPickWeightedChannelCandidatePreservesZeroWeightSemantics from the service/channel_admission tests into the model package’s test file or test location alongside PickWeightedChannelCandidate. Preserve the test assertions and inputs unchanged so it continues verifying zero-weight selection semantics under the model package.service/channel_admission.go (2)
355-362: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueExtract the RPM window pruning into one helper.
acquireMemoryandsnapshotcontain the same pruning loop. The duplication makes future window changes error prone.memoryChannelAdmissionStatecan own the logic.♻️ Proposed refactor
// Call with state.mu held. func (s *memoryChannelAdmissionState) pruneRPM(cutoff time.Time) { firstActive := 0 for firstActive < len(s.rpmAdmissions) && !s.rpmAdmissions[firstActive].startedAt.After(cutoff) { firstActive++ } if firstActive > 0 { s.rpmAdmissions = append([]memoryRPMAdmission(nil), s.rpmAdmissions[firstActive:]...) } }state.mu.Lock() defer state.mu.Unlock() - cutoff := now.Add(-m.rpmWindow) - firstActiveRPM := 0 - for firstActiveRPM < len(state.rpmAdmissions) && !state.rpmAdmissions[firstActiveRPM].startedAt.After(cutoff) { - firstActiveRPM++ - } - if firstActiveRPM > 0 { - state.rpmAdmissions = append([]memoryRPMAdmission(nil), state.rpmAdmissions[firstActiveRPM:]...) - } + state.pruneRPM(now.Add(-m.rpmWindow))Also applies to: 458-465
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@service/channel_admission.go` around lines 355 - 362, Extract the duplicated RPM admission pruning loop from acquireMemory and snapshot into a memoryChannelAdmissionState method named pruneRPM, called with state.mu held. Have the helper accept the cutoff time, perform the existing removal and slice-copy behavior, and replace both inline loops with calls to it.
153-208: 🩺 Stability & Availability | 🔵 Trivial | 💤 Low value
Releaseperforms a Redis round trip while holdingl.mu.
Commituses the same mutex. A slow Redis release can block a concurrentCommitfor up tochannelAdmissionBackendTimeout. Consider capturing the release parameters under the lock, marking the lease released, and running the Redis script after unlocking.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@service/channel_admission.go` around lines 153 - 208, Refactor ChannelAdmissionLease.Release so it captures the release state and parameters while holding l.mu, marks the lease released, then unlocks before executing channelAdmissionReleaseScript.Run. Preserve the existing no-op and memory-mode behavior, and ensure Redis errors are still returned without holding l.mu so Commit is not blocked by the backend round trip.controller/relay.go (1)
366-379: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value
channelCapacityAPIErroralso writes a response header.The name states that the function builds an error. It also sets
Retry-Afteron the response.RelayTaskcalls it at Line 601 only to build aTaskError, so the header write there is implicit. Consider returning the computed seconds and setting the header at the call sites, or rename the function to state both effects.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@controller/relay.go` around lines 366 - 379, Update channelCapacityAPIError so error construction and Retry-After response mutation are no longer implicitly combined: return the computed retry-after seconds (or otherwise expose it) and set the header explicitly at each caller, including RelayTask, while preserving the existing rounding and minimum-one-second behavior.controller/channel_admission_test.go (1)
23-53: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAdd a case for the limited-channel branch of
getChannel.This test covers the snapshot path when no limits exist and no lease exists. The branch at
controller/relay.goLines 327-339 is not covered: settings declare a limit, the context holds no lease, sogetChannelloads the channel and acquires admission. A case there would protect the 429 capacity result for a fixed channel.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@controller/channel_admission_test.go` around lines 23 - 53, Extend TestGetInitialUnlimitedChannelUsesContextSnapshot with a separate limited-channel case, configuring the channel settings with a limit while leaving the context without a lease. Exercise getChannel and verify it loads the fixed channel, attempts admission, and returns the expected 429 capacity error; preserve the existing unlimited snapshot assertions.middleware/distributor.go (2)
44-52: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winRead the specific channel ID with a typed helper.
Line 51 uses an unchecked type assertion
channelID.(string). If any producer stores this context key as anint, the middleware panics on a live request path. Usecommon.GetContextKeyStringor check the assertion.🛡️ Proposed guard
if specificChannel { - id, parseErr := strconv.Atoi(channelID.(string)) - if parseErr != nil { + rawChannelID, isString := channelID.(string) + if !isString { + abortWithOpenAiMessage(c, http.StatusBadRequest, i18n.T(c, i18n.MsgDistributorInvalidChannelId)) + return + } + id, parseErr := strconv.Atoi(rawChannelID) + if parseErr != nil {🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@middleware/distributor.go` around lines 44 - 52, Update the specific-channel parsing branch in the distributor middleware to retrieve the context value through common.GetContextKeyString, or safely validate its type before conversion, instead of asserting channelID as string. Preserve the existing parse-error handling and channel-selection flow for valid values.
164-172: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valuePass the raw retry duration and reuse one rounding helper.
Line 167 converts
RetryAfterto whole seconds, thenabortWithChannelCapacityrounds up again. The rounding logic at lines 209-212 also duplicatesChannelCapacityError.RetryAfterSecondsinservice/channel_select.go. Pass the duration through and keep one implementation of the rounding rule.♻️ Proposed simplification
- abortWithChannelCapacity(c, time.Duration(capacityErr.RetryAfterSeconds())*time.Second) + abortWithChannelCapacity(c, capacityErr.RetryAfter)func abortWithChannelCapacity(c *gin.Context, retryAfter time.Duration) { - retryAfterSeconds := int64((retryAfter + time.Second - 1) / time.Second) - if retryAfterSeconds < 1 { - retryAfterSeconds = 1 - } + retryAfterSeconds := (&service.ChannelCapacityError{RetryAfter: retryAfter}).RetryAfterSeconds()Also applies to: 208-213
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@middleware/distributor.go` around lines 164 - 172, Pass the raw retry duration from the selection error to abortWithChannelCapacity instead of converting RetryAfterSeconds() at the call site. Consolidate the duration-to-seconds rounding into one shared implementation, and update abortWithChannelCapacity and ChannelCapacityError.RetryAfterSeconds to reuse it without applying rounding twice.middleware/channel_admission_test.go (1)
185-204: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAssert the response status for each exit path.
The subtest only checks that concurrency returns to 0. A regression that rejects the request in the distributor also yields concurrency 0, so the test would pass while the handler never runs. Add an expected status per table row, or assert that the handler was reached.
💚 Proposed strengthening
tests := []struct { name string exit string + status int context func() (context.Context, context.CancelFunc) }{ - {name: "success"}, - {name: "upstream error", exit: "error"}, + {name: "success", status: http.StatusNoContent}, + {name: "upstream error", exit: "error", status: http.StatusBadGateway},snapshot, snapshotErr := service.GetChannelAdmissionSnapshot(context.Background(), channel) require.NoError(t, snapshotErr) assert.Equal(t, 0, snapshot.CurrentConcurrency) + if test.status != 0 { + assert.Equal(t, test.status, response.Code) + }As per coding guidelines: "Backend tests must protect real behavior, API contracts, billing/accounting invariants, compatibility, or regression paths; prefer deterministic table tests with explicit inputs and exact outputs".
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@middleware/channel_admission_test.go` around lines 185 - 204, The table-driven subtests in the channel admission test must verify the HTTP outcome as well as concurrency cleanup. Add an expected status value to each test case, then assert response.Code against it after router.ServeHTTP, preserving the existing snapshot assertion to cover admission accounting.Source: Coding guidelines
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@controller/relay.go`:
- Around line 600-604: Update shouldRetryTaskRelay to return false for the
channel-capacity error code before the generic http.StatusTooManyRequests retry
condition. Preserve retries for other 429 errors and keep the existing task
relay behavior unchanged.
In `@middleware/distributor.go`:
- Around line 37-42: Update the retry channel-switch logic in
Distribute/getChannel so the currently held distributor lease is released via
service.GetChannelAdmissionLease(c) before replacing it with a newly selected
channel lease. Skip the release when retries continue on the same locked
channel, and ensure the defer only releases the final active lease without
leaving the original reservation held.
---
Nitpick comments:
In `@controller/channel_admission_test.go`:
- Around line 23-53: Extend TestGetInitialUnlimitedChannelUsesContextSnapshot
with a separate limited-channel case, configuring the channel settings with a
limit while leaving the context without a lease. Exercise getChannel and verify
it loads the fixed channel, attempts admission, and returns the expected 429
capacity error; preserve the existing unlimited snapshot assertions.
In `@controller/relay.go`:
- Around line 366-379: Update channelCapacityAPIError so error construction and
Retry-After response mutation are no longer implicitly combined: return the
computed retry-after seconds (or otherwise expose it) and set the header
explicitly at each caller, including RelayTask, while preserving the existing
rounding and minimum-one-second behavior.
In `@middleware/channel_admission_test.go`:
- Around line 185-204: The table-driven subtests in the channel admission test
must verify the HTTP outcome as well as concurrency cleanup. Add an expected
status value to each test case, then assert response.Code against it after
router.ServeHTTP, preserving the existing snapshot assertion to cover admission
accounting.
In `@middleware/distributor.go`:
- Around line 44-52: Update the specific-channel parsing branch in the
distributor middleware to retrieve the context value through
common.GetContextKeyString, or safely validate its type before conversion,
instead of asserting channelID as string. Preserve the existing parse-error
handling and channel-selection flow for valid values.
- Around line 164-172: Pass the raw retry duration from the selection error to
abortWithChannelCapacity instead of converting RetryAfterSeconds() at the call
site. Consolidate the duration-to-seconds rounding into one shared
implementation, and update abortWithChannelCapacity and
ChannelCapacityError.RetryAfterSeconds to reuse it without applying rounding
twice.
In `@service/channel_admission_test.go`:
- Around line 424-432: Move
TestPickWeightedChannelCandidatePreservesZeroWeightSemantics from the
service/channel_admission tests into the model package’s test file or test
location alongside PickWeightedChannelCandidate. Preserve the test assertions
and inputs unchanged so it continues verifying zero-weight selection semantics
under the model package.
In `@service/channel_admission.go`:
- Around line 355-362: Extract the duplicated RPM admission pruning loop from
acquireMemory and snapshot into a memoryChannelAdmissionState method named
pruneRPM, called with state.mu held. Have the helper accept the cutoff time,
perform the existing removal and slice-copy behavior, and replace both inline
loops with calls to it.
- Around line 153-208: Refactor ChannelAdmissionLease.Release so it captures the
release state and parameters while holding l.mu, marks the lease released, then
unlocks before executing channelAdmissionReleaseScript.Run. Preserve the
existing no-op and memory-mode behavior, and ensure Redis errors are still
returned without holding l.mu so Commit is not blocked by the backend round
trip.
🪄 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 Plus
Run ID: f4dc9209-8c4c-4ef5-bc1b-53b70394f951
📒 Files selected for processing (36)
controller/channel_admission_test.gocontroller/relay.godocs/channel/other_setting.mdi18n/keys.goi18n/locales/en.yamli18n/locales/zh-CN.yamli18n/locales/zh-TW.yamlmiddleware/channel_admission_test.gomiddleware/distributor.gomodel/ability.gomodel/channel.gomodel/channel_cache.gomodel/channel_settings_test.gorelaykit/dto/channel_settings.gorelaykit/dto/channel_settings_test.gorelaykit/types/error.goservice/channel_admission.goservice/channel_admission_test.goservice/channel_select.goservice/lua/channel_admission_acquire.luaservice/lua/channel_admission_release.luaservice/lua/channel_admission_renew.luaservice/lua/channel_admission_snapshot.luaweb/src/features/channels/components/drawers/channel-mutate-drawer.tsxweb/src/features/channels/lib/__tests__/channel-form-admission-limits.test.tsweb/src/features/channels/lib/channel-form-errors.tsweb/src/features/channels/lib/channel-form.tsweb/src/features/channels/types.tsweb/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-TW.jsonweb/src/i18n/locales/zh.jsonweb/src/i18n/static-keys.ts
91d44cd to
aedf377
Compare
There was a problem hiding this comment.
🧹 Nitpick comments (2)
middleware/channel_admission_test.go (1)
166-169: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueAvoid
requireon the parenttinside the route handler.The handler closure captures the outer
t. The handler executes while a subtest fromt.Runis running on its own goroutine. Ifleaseis nil,require.NotNilcallsFailNowon the parenttfrom a goroutine that does not own it. Go documentsFailNowas valid only in the goroutine that runs that test, so failures report against the wrong test and the subtest does not stop cleanly.Use a non-fatal check plus an early return, or capture the active subtest
tin a variable that eacht.Runbody assigns.♻️ Proposed change: assert and return instead of require
router.POST("/v1/chat/completions", func(c *gin.Context) { lease := service.GetChannelAdmissionLease(c) - require.NotNil(t, lease) + if !assert.NotNil(t, lease) { + c.Status(http.StatusInternalServerError) + return + } lease.Commit()🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@middleware/channel_admission_test.go` around lines 166 - 169, Update the route handler around service.GetChannelAdmissionLease to avoid calling require.NotNil with the parent test object; use a non-fatal assertion and return immediately when lease is nil, otherwise preserve the existing lease.Commit flow.controller/channel_admission_test.go (1)
132-143: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winIsolate the retry discriminator.
The capacity case sets both
CodeandLocalError: true. The upstream case sets neither. IfshouldRetryTaskRelayskips retry for everyLocalError, this test passes even when the function ignoresErrorCodeChannelCapacityExhausted. Add a case withLocalError: trueand an unrelated code, or a case with the capacity code andLocalError: false, to pin which field drives the decision.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@controller/channel_admission_test.go` around lines 132 - 143, Update the shouldRetryTaskRelay test cases to isolate the retry discriminator: add coverage for an unrelated code with LocalError true, or the capacity code with LocalError false, and assert the expected retry result so the test specifically verifies handling of ErrorCodeChannelCapacityExhausted rather than only LocalError.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Nitpick comments:
In `@controller/channel_admission_test.go`:
- Around line 132-143: Update the shouldRetryTaskRelay test cases to isolate the
retry discriminator: add coverage for an unrelated code with LocalError true, or
the capacity code with LocalError false, and assert the expected retry result so
the test specifically verifies handling of ErrorCodeChannelCapacityExhausted
rather than only LocalError.
In `@middleware/channel_admission_test.go`:
- Around line 166-169: Update the route handler around
service.GetChannelAdmissionLease to avoid calling require.NotNil with the parent
test object; use a non-fatal assertion and return immediately when lease is nil,
otherwise preserve the existing lease.Commit flow.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: ebf40293-e44a-4746-b5e7-0142765b293e
📒 Files selected for processing (37)
controller/channel_admission_test.gocontroller/relay.godocs/channel/other_setting.mdi18n/keys.goi18n/locales/en.yamli18n/locales/zh-CN.yamli18n/locales/zh-TW.yamlmiddleware/channel_admission_test.gomiddleware/distributor.gomodel/ability.gomodel/channel.gomodel/channel_cache.gomodel/channel_cache_test.gomodel/channel_settings_test.gorelaykit/dto/channel_settings.gorelaykit/dto/channel_settings_test.gorelaykit/types/error.goservice/channel_admission.goservice/channel_admission_test.goservice/channel_select.goservice/lua/channel_admission_acquire.luaservice/lua/channel_admission_release.luaservice/lua/channel_admission_renew.luaservice/lua/channel_admission_snapshot.luaweb/src/features/channels/components/drawers/channel-mutate-drawer.tsxweb/src/features/channels/lib/__tests__/channel-form-admission-limits.test.tsweb/src/features/channels/lib/channel-form-errors.tsweb/src/features/channels/lib/channel-form.tsweb/src/features/channels/types.tsweb/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-TW.jsonweb/src/i18n/locales/zh.jsonweb/src/i18n/static-keys.ts
🚧 Files skipped from review as they are similar to previous changes (34)
- web/src/features/channels/lib/channel-form-errors.ts
- web/src/features/channels/types.ts
- relaykit/types/error.go
- web/src/features/channels/lib/tests/channel-form-admission-limits.test.ts
- web/src/i18n/locales/ja.json
- web/src/i18n/static-keys.ts
- web/src/features/channels/lib/channel-form.ts
- web/src/features/channels/components/drawers/channel-mutate-drawer.tsx
- i18n/locales/en.yaml
- web/src/i18n/locales/ru.json
- web/src/i18n/locales/zh-TW.json
- web/src/i18n/locales/vi.json
- web/src/i18n/locales/en.json
- web/src/i18n/locales/zh.json
- service/lua/channel_admission_renew.lua
- i18n/locales/zh-TW.yaml
- service/lua/channel_admission_acquire.lua
- model/channel.go
- service/lua/channel_admission_release.lua
- i18n/locales/zh-CN.yaml
- model/channel_settings_test.go
- i18n/keys.go
- relaykit/dto/channel_settings_test.go
- docs/channel/other_setting.md
- web/src/i18n/locales/fr.json
- service/channel_select.go
- model/ability.go
- service/channel_admission_test.go
- middleware/distributor.go
- relaykit/dto/channel_settings.go
- model/channel_cache.go
- controller/relay.go
- service/lua/channel_admission_snapshot.lua
- service/channel_admission.go
|
Review follow-up is in
I verified the distributor-lease finding separately. The initial distributor lease is the first relay attempt's active lease and is released before retry selection; early exits and panics are covered by defers. The locked task path also releases a mismatched initial lease explicitly, so no second channel-switch release was added. I also kept The three OIDC locale lines reported as out of scope are value-preserving relocations caused by the repository's mandatory Revalidated with |
aedf377 to
21077fd
Compare
Rebase updateRebased Conflict resolution preserves the token-specific Auto-group snapshot for both normal selection and affinity selection, while keeping channel admission before upstream I/O. Validation on the rebased head:
The full frontend |
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (2)
service/channel_select.go (2)
164-166: 🎯 Functional Correctness | 🔵 Trivial | ⚡ Quick winAdvance the auto-group state when a group yields no selection.
Line 145 handles an empty tier list by advancing
ContextKeyAutoGroupIndex, resettingContextKeyAutoGroupRetryIndex, and callingparam.SetRetry(0). Line 164 handles anilselection from a non-empty tier list by only callingcontinue. The two paths leave different auto-group state. In thenilcase the loop moves to the next group, but the stored group index still points at the exhausted group, so a later selection call restarts at that group.Make both exhaustion paths update the same state.
♻️ Proposed consistency fix
if selection == nil { + common.SetContextKey(param.Ctx, constant.ContextKeyAutoGroupIndex, groupIndex+1) + common.SetContextKey(param.Ctx, constant.ContextKeyAutoGroupRetryIndex, 0) + param.SetRetry(0) continue }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@service/channel_select.go` around lines 164 - 166, Update the nil-selection branch in the group-selection loop near selection handling to apply the same auto-group state advancement as the empty-tier-list path: advance ContextKeyAutoGroupIndex, reset ContextKeyAutoGroupRetryIndex, and call param.SetRetry(0) before continuing. Keep the existing behavior for non-nil selections unchanged.
112-112: 🩺 Stability & Availability | 🔵 Trivial | 💤 Low valuePass the request context through channel selection.
SelectChannelWithAdmissioncurrently passes*gin.ContextintoselectAdmittedChannel, andAcquireChannelAdmission(ctx, ...)then uses the Gin context for Redis acquire/snapshot calls. Usec.Request.Context()in the caller for consistency with the specific-channel and affinity paths. If this value must come fromRetryParam, extract/wrap the request context with aRequest != nilguard instead of storing*gin.Contextdirectly.Also applies to: 152-152
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@service/channel_select.go` at line 112, Update SelectChannelWithAdmission and the related call at the second selection site to pass c.Request.Context() into selectAdmittedChannel rather than *gin.Context, matching the specific-channel and affinity paths. If the context is obtained through RetryParam, derive it from Request with a nil guard and avoid storing the Gin context directly.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@service/channel_select.go`:
- Around line 220-235: Update the candidate loop around
PickWeightedChannelCandidate and AcquireChannelAdmission so invalid or nil
candidates are removed and skipped rather than terminating selection. For
admission errors, record the error while continuing through remaining
candidates, and return the recorded error only if no candidate is ultimately
admitted; preserve immediate return for an allowed admission.
---
Nitpick comments:
In `@service/channel_select.go`:
- Around line 164-166: Update the nil-selection branch in the group-selection
loop near selection handling to apply the same auto-group state advancement as
the empty-tier-list path: advance ContextKeyAutoGroupIndex, reset
ContextKeyAutoGroupRetryIndex, and call param.SetRetry(0) before continuing.
Keep the existing behavior for non-nil selections unchanged.
- Line 112: Update SelectChannelWithAdmission and the related call at the second
selection site to pass c.Request.Context() into selectAdmittedChannel rather
than *gin.Context, matching the specific-channel and affinity paths. If the
context is obtained through RetryParam, derive it from Request with a nil guard
and avoid storing the Gin context directly.
🪄 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 Plus
Run ID: 85403d46-2a01-49bb-9b19-50f830204453
📒 Files selected for processing (37)
controller/channel_admission_test.gocontroller/relay.godocs/channel/other_setting.mdi18n/keys.goi18n/locales/en.yamli18n/locales/zh-CN.yamli18n/locales/zh-TW.yamlmiddleware/channel_admission_test.gomiddleware/distributor.gomodel/ability.gomodel/channel.gomodel/channel_cache.gomodel/channel_cache_test.gomodel/channel_settings_test.gorelaykit/dto/channel_settings.gorelaykit/dto/channel_settings_test.gorelaykit/types/error.goservice/channel_admission.goservice/channel_admission_test.goservice/channel_select.goservice/lua/channel_admission_acquire.luaservice/lua/channel_admission_release.luaservice/lua/channel_admission_renew.luaservice/lua/channel_admission_snapshot.luaweb/src/features/channels/components/drawers/channel-mutate-drawer.tsxweb/src/features/channels/lib/__tests__/channel-form-admission-limits.test.tsweb/src/features/channels/lib/channel-form-errors.tsweb/src/features/channels/lib/channel-form.tsweb/src/features/channels/types.tsweb/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-TW.jsonweb/src/i18n/locales/zh.jsonweb/src/i18n/static-keys.ts
🚧 Files skipped from review as they are similar to previous changes (28)
- i18n/keys.go
- relaykit/types/error.go
- model/channel.go
- service/lua/channel_admission_renew.lua
- i18n/locales/zh-TW.yaml
- i18n/locales/en.yaml
- relaykit/dto/channel_settings_test.go
- service/lua/channel_admission_acquire.lua
- model/channel_cache_test.go
- web/src/i18n/static-keys.ts
- service/lua/channel_admission_snapshot.lua
- model/channel_settings_test.go
- web/src/features/channels/lib/channel-form-errors.ts
- service/lua/channel_admission_release.lua
- i18n/locales/zh-CN.yaml
- web/src/features/channels/types.ts
- web/src/features/channels/components/drawers/channel-mutate-drawer.tsx
- model/ability.go
- middleware/distributor.go
- middleware/channel_admission_test.go
- controller/channel_admission_test.go
- docs/channel/other_setting.md
- web/src/features/channels/lib/channel-form.ts
- model/channel_cache.go
- controller/relay.go
- service/channel_admission_test.go
- service/channel_admission.go
- relaykit/dto/channel_settings.go
| for len(candidates) > 0 { | ||
| candidate, candidateIndex := model.PickWeightedChannelCandidate(candidates) | ||
| if candidateIndex < 0 || candidate.Channel == nil { | ||
| break | ||
| } | ||
| logger.LogDebug(param.Ctx, "Auto selecting group: %s, priorityRetry: %d", autoGroup, priorityRetry) | ||
|
|
||
| channel, _ = model.GetRandomSatisfiedChannel(autoGroup, param.ModelName, priorityRetry, param.RequestPath) | ||
| if channel == nil { | ||
| // Current group has no available channel for this model, try next group | ||
| // 当前分组没有该模型的可用渠道，尝试下一个分组 | ||
| logger.LogDebug(param.Ctx, "No available channel in group %s for model %s at priorityRetry %d, trying next group", autoGroup, param.ModelName, priorityRetry) | ||
| // 重置状态以尝试下一个分组 | ||
| common.SetContextKey(param.Ctx, constant.ContextKeyAutoGroupIndex, i+1) | ||
| common.SetContextKey(param.Ctx, constant.ContextKeyAutoGroupRetryIndex, 0) | ||
| // Reset retry counter so outer loop can continue for next group | ||
| // 重置重试计数器，以便外层循环可以为下一个分组继续 | ||
| param.SetRetry(0) | ||
| continue | ||
| candidates = append(candidates[:candidateIndex], candidates[candidateIndex+1:]...) | ||
|
|
||
| lease, decision, err := AcquireChannelAdmission(ctx, candidate.Channel) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("acquire channel #%d admission: %w", candidate.Channel.Id, err) | ||
| } | ||
| common.SetContextKey(param.Ctx, constant.ContextKeyAutoGroup, autoGroup) | ||
| selectGroup = autoGroup | ||
| logger.LogDebug(param.Ctx, "Auto selected group: %s", autoGroup) | ||
|
|
||
| // Prepare state for next retry | ||
| // 为下一次重试准备状态 | ||
| if crossGroupRetry && priorityRetry >= common.RetryTimes { | ||
| // Current group has exhausted all retries, prepare to switch to next group | ||
| // This request still uses current group, but next retry will use next group | ||
| // 当前分组已用完所有重试次数，准备切换到下一个分组 | ||
| // 本次请求仍使用当前分组，但下次重试将使用下一个分组 | ||
| logger.LogDebug(param.Ctx, "Current group %s retries exhausted (priorityRetry=%d >= RetryTimes=%d), preparing switch to next group for next retry", autoGroup, priorityRetry, common.RetryTimes) | ||
| common.SetContextKey(param.Ctx, constant.ContextKeyAutoGroupIndex, i+1) | ||
| // Reset retry counter so outer loop can continue for next group | ||
| // 重置重试计数器，以便外层循环可以为下一个分组继续 | ||
| param.SetRetry(0) | ||
| param.ResetRetryNextTry() | ||
| } else { | ||
| // Stay in current group, save current state | ||
| // 保持在当前分组，保存当前状态 | ||
| common.SetContextKey(param.Ctx, constant.ContextKeyAutoGroupIndex, i) | ||
| if decision.Allowed { | ||
| return &ChannelSelection{Channel: candidate.Channel, Group: group, Lease: lease}, nil | ||
| } | ||
| break | ||
| } | ||
| } else { | ||
| channel, err = model.GetRandomSatisfiedChannel(param.TokenGroup, param.ModelName, param.GetRetry(), param.RequestPath) | ||
| if err != nil { | ||
| return nil, param.TokenGroup, err | ||
| capacityErr.addDecision(decision) | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
One bad candidate aborts more than itself.
Two paths in this loop stop more work than the failing candidate requires:
- Line 222: if
candidate.Channelis nil, the code breaks out of the inner loop. That abandons every remaining candidate in the tier. Remove the candidate and continue instead. - Line 228: if
AcquireChannelAdmissionreturns an error, the whole selection fails.AcquireChannelAdmissionreturns an error whensettings.ValidateAdmissionLimits()fails, so one channel with an invalid storedmax_concurrencyorrpm_limitblocks routing for every other channel in the group. Skip that candidate, record the error, and return it only when no candidate is admitted.
🛡️ Proposed fix
capacityErr := &ChannelCapacityError{}
+ var lastAcquireErr error
for tierIndex := startTier; tierIndex < len(tiers); tierIndex++ {
candidates := append([]model.ChannelCandidate(nil), tiers[tierIndex].Candidates...)
for len(candidates) > 0 {
candidate, candidateIndex := model.PickWeightedChannelCandidate(candidates)
- if candidateIndex < 0 || candidate.Channel == nil {
+ if candidateIndex < 0 {
break
}
candidates = append(candidates[:candidateIndex], candidates[candidateIndex+1:]...)
+ if candidate.Channel == nil {
+ continue
+ }
lease, decision, err := AcquireChannelAdmission(ctx, candidate.Channel)
if err != nil {
- return nil, fmt.Errorf("acquire channel #%d admission: %w", candidate.Channel.Id, err)
+ lastAcquireErr = fmt.Errorf("acquire channel #%d admission: %w", candidate.Channel.Id, err)
+ logger.LogWarn(ctx, lastAcquireErr.Error())
+ continue
}
if decision.Allowed {
return &ChannelSelection{Channel: candidate.Channel, Group: group, Lease: lease}, nil
}
capacityErr.addDecision(decision)
}
}
if capacityErr.ConcurrencyRejects > 0 || capacityErr.RPMRejects > 0 {
return nil, capacityErr
}
+ if lastAcquireErr != nil {
+ return nil, lastAcquireErr
+ }
return nil, nil🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@service/channel_select.go` around lines 220 - 235, Update the candidate loop
around PickWeightedChannelCandidate and AcquireChannelAdmission so invalid or
nil candidates are removed and skipped rather than terminating selection. For
admission errors, record the error while continuing through remaining
candidates, and return the recorded error only if no candidate is ultimately
admitted; preserve immediate return for an allowed admission.
CI result after rebaseThe rebased head
The direct failures are in upstream commit |
1 participant
Important
📝 变更描述 / Description
为渠道增加可选的
max_concurrency与rpm_limit准入限制，并在请求发往上游前完成容量预留。0或缺省保持不限流，字段复用现有ChannelSettingsJSON，不需要数据库迁移，也不会改变已有渠道行为。429 channel_capacity_exhausted与Retry-After，不会静默改道或覆盖原亲和绑定。Commit，成功、失败、取消或重试均不回退。并发槽覆盖非流式、流式、WebSocket 与任务提交的上游生命周期，并在所有退出路径释放。本期刻意不实现请求排队；队列取消、超时和队头阻塞语义应作为独立需求处理。
🚀 变更类型 / Type of change
🔗 关联任务 / Related Issue
✅ 提交前检查项 / Checklist
Bug fix，关联的是既有功能请求 功能请求：渠道速率限制与并发控制功能 #1730。📸 运行证明 / Proof of Work
通过：
改动文件的聚焦 oxlint 检查通过。全仓
bun run lint仍会报告当前main中既有、且不位于本 PR 改动文件内的诊断。浏览器手动验证覆盖桌面布局、390px 移动布局以及俄语长文案；两个数值控件无重叠或溢出，页面无 console/page error。
回归测试覆盖同级改选、优先级降级、auto group、硬亲和、固定渠道、本地错误分类、配置从不限到有限及升降上限、Redis 双客户端全局并发/RPM、原子组合边界、内存 fallback、竞态、重复释放、租约过期、续租、多 Key 聚合，以及成功/错误/取消/超时/流式结束/panic 的释放路径。
Summary by CodeRabbit