(null)
async function loadGroups() {
try {
@@ -104,6 +106,13 @@ async function loadGroups() {
} catch { /* ignore */ }
}
+async function loadPaymentConfig() {
+ try {
+ const res = await adminPaymentAPI.getConfig()
+ paymentConfig.value = res.data
+ } catch { /* preview only */ }
+}
+
function getGroup(id: number): AdminGroup | undefined {
return groups.value.find(g => g.id === id)
}
@@ -181,6 +190,7 @@ async function handleDeletePlan() {
onMounted(() => {
loadGroups()
+ loadPaymentConfig()
loadPlans()
})
diff --git a/frontend/src/views/admin/orders/PlanEditDialog.vue b/frontend/src/views/admin/orders/PlanEditDialog.vue
index acc70bef16..92b6574734 100644
--- a/frontend/src/views/admin/orders/PlanEditDialog.vue
+++ b/frontend/src/views/admin/orders/PlanEditDialog.vue
@@ -35,7 +35,16 @@
-
+
+
+
+
+ {{ t('payment.admin.subscriptionCnyPayPreview', { amount: subscriptionCnyPreview.amount }) }}
+
+ {{ t('payment.admin.subscriptionCnyPayPreviewWithFee', { feeRate: subscriptionCnyPreview.feeRate, total: subscriptionCnyPreview.total }) }}
+
+
+
@@ -81,7 +90,9 @@ import { ref, reactive, computed, watch } from 'vue'
import { useI18n } from 'vue-i18n'
import { useAppStore } from '@/stores/app'
import { adminPaymentAPI } from '@/api/admin/payment'
+import type { AdminPaymentConfig } from '@/api/admin/payment'
import { extractApiErrorMessage } from '@/utils/apiError'
+import { formatPaymentAmount } from '@/components/payment/currency'
import type { SubscriptionPlan } from '@/types/payment'
import type { AdminGroup } from '@/types'
import BaseDialog from '@/components/common/BaseDialog.vue'
@@ -94,6 +105,7 @@ const props = defineProps<{
show: boolean
plan: SubscriptionPlan | null
groups: AdminGroup[]
+ paymentConfig?: AdminPaymentConfig | null
}>()
const emit = defineEmits<{
@@ -129,6 +141,31 @@ const selectedGroupInfo = computed(() => {
return props.groups.find(g => g.id === planForm.group_id) || null
})
+function roundCnyAmount(value: number): number {
+ return Math.round(value * 100) / 100
+}
+
+function ceilCnyAmount(value: number): number {
+ return Math.ceil(value * 100) / 100
+}
+
+const subscriptionCnyPreview = computed(() => {
+ const price = Number(planForm.price) || 0
+ const rate = Number(props.paymentConfig?.subscription_usd_to_cny_rate) || 0
+ if (price <= 0 || rate <= 0) return null
+
+ const amount = roundCnyAmount(price * rate)
+ const feeRate = Number(props.paymentConfig?.recharge_fee_rate) || 0
+ const fee = feeRate > 0 ? ceilCnyAmount((amount * feeRate) / 100) : 0
+ const total = feeRate > 0 ? roundCnyAmount(amount + fee) : amount
+
+ return {
+ amount: formatPaymentAmount(amount, 'CNY'),
+ feeRate,
+ total: formatPaymentAmount(total, 'CNY'),
+ }
+})
+
// Reset form when dialog opens
watch(() => props.show, (visible) => {
if (!visible) return
diff --git a/frontend/src/views/admin/orders/__tests__/PlanEditDialog.spec.ts b/frontend/src/views/admin/orders/__tests__/PlanEditDialog.spec.ts
new file mode 100644
index 0000000000..9c31e7c176
--- /dev/null
+++ b/frontend/src/views/admin/orders/__tests__/PlanEditDialog.spec.ts
@@ -0,0 +1,77 @@
+import { describe, expect, it, vi } from 'vitest'
+import { mount } from '@vue/test-utils'
+import PlanEditDialog from '../PlanEditDialog.vue'
+
+vi.mock('vue-i18n', () => ({
+ useI18n: () => ({
+ t: (key: string, params?: Record
) => {
+ if (key === 'payment.admin.subscriptionCnyPayPreview') return `preview ${params?.amount}`
+ if (key === 'payment.admin.subscriptionCnyPayPreviewWithFee') return `fee ${params?.feeRate} ${params?.total}`
+ return key
+ },
+ }),
+}))
+
+vi.mock('@/stores/app', () => ({
+ useAppStore: () => ({
+ showError: vi.fn(),
+ showSuccess: vi.fn(),
+ }),
+}))
+
+vi.mock('@/api/admin/payment', () => ({
+ adminPaymentAPI: {
+ createPlan: vi.fn(),
+ updatePlan: vi.fn(),
+ },
+}))
+
+function mountDialog(paymentConfig: Record | null) {
+ return mount(PlanEditDialog, {
+ props: {
+ show: true,
+ plan: null,
+ groups: [],
+ paymentConfig,
+ },
+ global: {
+ stubs: {
+ BaseDialog: {
+ props: ['show'],
+ template: '
',
+ },
+ Select: true,
+ Icon: true,
+ GroupBadge: true,
+ },
+ },
+ })
+}
+
+describe('PlanEditDialog subscription CNY payment preview', () => {
+ it('shows CNY channel charge using the configured subscription rate and fee', async () => {
+ const wrapper = mountDialog({
+ subscription_usd_to_cny_rate: 7.15,
+ recharge_fee_rate: 2.5,
+ })
+
+ await wrapper.find('input[type="number"]').setValue('9.99')
+
+ expect(wrapper.text()).toContain('preview')
+ expect(wrapper.text()).toContain('¥71.43')
+ expect(wrapper.text()).toContain('fee 2.5')
+ expect(wrapper.text()).toContain('¥73.22')
+ })
+
+ it('hides the preview when the subscription rate is not configured', async () => {
+ const wrapper = mountDialog({
+ subscription_usd_to_cny_rate: 0,
+ recharge_fee_rate: 2.5,
+ })
+
+ await wrapper.find('input[type="number"]').setValue('9.99')
+
+ expect(wrapper.text()).not.toContain('preview')
+ expect(wrapper.text()).not.toContain('¥71.43')
+ })
+})
From 6b4c3c6ed18eae39aba5c7ed774215b1ef34db61 Mon Sep 17 00:00:00 2001
From: Heatherm Huang
Date: Mon, 6 Jul 2026 18:23:38 +0800
Subject: [PATCH 06/19] test: cover Grok video media rate billing
---
.../openai_gateway_record_usage_test.go | 45 +++++++++++++++++++
1 file changed, 45 insertions(+)
diff --git a/backend/internal/service/openai_gateway_record_usage_test.go b/backend/internal/service/openai_gateway_record_usage_test.go
index 8323035ffe..5546190d14 100644
--- a/backend/internal/service/openai_gateway_record_usage_test.go
+++ b/backend/internal/service/openai_gateway_record_usage_test.go
@@ -1803,6 +1803,51 @@ func TestOpenAIGatewayServiceRecordUsage_ImageIndependentMultiplierUsesImageRate
require.Equal(t, string(BillingModeImage), *usageRepo.lastLog.BillingMode)
}
+func TestOpenAIGatewayServiceRecordUsage_GrokVideoUsesImageRateMultiplier(t *testing.T) {
+ mediaPrice2K := 0.4
+ groupID := int64(126)
+
+ usageRepo := &openAIRecordUsageLogRepoStub{inserted: true}
+ svc := newOpenAIRecordUsageServiceForTest(usageRepo, &openAIRecordUsageUserRepoStub{}, &openAIRecordUsageSubRepoStub{}, nil)
+
+ err := svc.RecordUsage(context.Background(), &OpenAIRecordUsageInput{
+ Result: &OpenAIForwardResult{
+ RequestID: "video-request-123",
+ ResponseID: "video-request-123",
+ Model: "grok-imagine-video-1.5",
+ BillingModel: "grok-imagine-video-1.5",
+ ImageCount: 1,
+ ImageSize: ImageBillingSize2K,
+ Duration: time.Second,
+ },
+ APIKey: &APIKey{
+ ID: 10126,
+ GroupID: i64p(groupID),
+ Group: &Group{
+ ID: groupID,
+ Platform: PlatformGrok,
+ RateMultiplier: 0.15,
+ ImageRateIndependent: true,
+ ImageRateMultiplier: 0.5,
+ ImagePrice2K: &mediaPrice2K,
+ },
+ },
+ User: &User{ID: 20126},
+ Account: &Account{ID: 30126, Platform: PlatformGrok},
+ })
+
+ require.NoError(t, err)
+ require.NotNil(t, usageRepo.lastLog)
+ require.Equal(t, "grok-imagine-video-1.5", usageRepo.lastLog.Model)
+ require.Equal(t, 1, usageRepo.lastLog.ImageCount)
+ require.Equal(t, ImageBillingSize2K, *usageRepo.lastLog.ImageSize)
+ require.InDelta(t, 0.4, usageRepo.lastLog.TotalCost, 1e-12)
+ require.InDelta(t, 0.2, usageRepo.lastLog.ActualCost, 1e-12)
+ require.InDelta(t, 0.5, usageRepo.lastLog.RateMultiplier, 1e-12)
+ require.NotNil(t, usageRepo.lastLog.BillingMode)
+ require.Equal(t, string(BillingModeImage), *usageRepo.lastLog.BillingMode)
+}
+
func TestOpenAIGatewayServiceRecordUsage_ChannelImageBillingUsesImageCountAndSharedMultiplier(t *testing.T) {
groupID := int64(123)
usageRepo := &openAIRecordUsageLogRepoStub{inserted: true}
From 95717a7e6261ac0cbbb596400a4e7de2ec04d281 Mon Sep 17 00:00:00 2001
From: Heatherm Huang
Date: Mon, 6 Jul 2026 18:30:32 +0800
Subject: [PATCH 07/19] test: clarify Grok video billing test name
---
backend/internal/service/openai_gateway_record_usage_test.go | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/backend/internal/service/openai_gateway_record_usage_test.go b/backend/internal/service/openai_gateway_record_usage_test.go
index 5546190d14..733cc8ae68 100644
--- a/backend/internal/service/openai_gateway_record_usage_test.go
+++ b/backend/internal/service/openai_gateway_record_usage_test.go
@@ -1803,7 +1803,7 @@ func TestOpenAIGatewayServiceRecordUsage_ImageIndependentMultiplierUsesImageRate
require.Equal(t, string(BillingModeImage), *usageRepo.lastLog.BillingMode)
}
-func TestOpenAIGatewayServiceRecordUsage_GrokVideoUsesImageRateMultiplier(t *testing.T) {
+func TestGrokVideoMediaBillingUsesImageRateMultiplier(t *testing.T) {
mediaPrice2K := 0.4
groupID := int64(126)
From de31e57333de894ead45520a70d60c5cfa33c3fb Mon Sep 17 00:00:00 2001
From: Heatherm Huang
Date: Mon, 6 Jul 2026 18:32:29 +0800
Subject: [PATCH 08/19] test: document Grok video media billing unit
---
backend/internal/service/grok_media.go | 1 +
.../internal/service/openai_gateway_record_usage_test.go | 7 ++++---
2 files changed, 5 insertions(+), 3 deletions(-)
diff --git a/backend/internal/service/grok_media.go b/backend/internal/service/grok_media.go
index 8942404eaa..f8b7ff3e83 100644
--- a/backend/internal/service/grok_media.go
+++ b/backend/internal/service/grok_media.go
@@ -483,6 +483,7 @@ func grokMediaUsageFromResponse(endpoint GrokMediaEndpoint, requestInfo GrokMedi
meta.ImageOutputSizes = collectOpenAIResponseImageOutputSizesFromJSONBytes(responseBody)
case GrokMediaEndpointVideosGenerations:
meta.ResponseID = extractGrokMediaVideoRequestID(responseBody)
+ // Video generation is one billable media unit; the legacy usage schema stores it in ImageCount.
meta.ImageCount = 1
meta.ImageSize = requestInfo.SizeTier
meta.ImageInputSize = requestInfo.Size
diff --git a/backend/internal/service/openai_gateway_record_usage_test.go b/backend/internal/service/openai_gateway_record_usage_test.go
index 733cc8ae68..697d89e81c 100644
--- a/backend/internal/service/openai_gateway_record_usage_test.go
+++ b/backend/internal/service/openai_gateway_record_usage_test.go
@@ -1816,9 +1816,10 @@ func TestGrokVideoMediaBillingUsesImageRateMultiplier(t *testing.T) {
ResponseID: "video-request-123",
Model: "grok-imagine-video-1.5",
BillingModel: "grok-imagine-video-1.5",
- ImageCount: 1,
- ImageSize: ImageBillingSize2K,
- Duration: time.Second,
+ // The usage schema has no separate video count; video generation is billed as one media unit.
+ ImageCount: 1,
+ ImageSize: ImageBillingSize2K,
+ Duration: time.Second,
},
APIKey: &APIKey{
ID: 10126,
From 4b321142bbbef8c46ee5ea8abf932418e04f064a Mon Sep 17 00:00:00 2001
From: DaydreamCoding <22166516+DaydreamCoding@users.noreply.github.com>
Date: Fri, 26 Jun 2026 18:01:31 +0800
Subject: [PATCH 09/19] =?UTF-8?q?fix(codex-detect):=20codex=20=E7=89=88?=
=?UTF-8?q?=E6=9C=AC=E9=97=A8=E6=8B=92=E7=BB=9D=E8=BF=94=E5=9B=9E=E5=B8=A6?=
=?UTF-8?q?=E7=89=88=E6=9C=AC=E5=8F=B7=E7=9A=84=E5=B7=AE=E5=BC=82=E5=8C=96?=
=?UTF-8?q?=E6=96=87=E6=A1=88?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
codex_cli_only 的 403 出口此前对所有拒绝原因硬编码同一句
"This account only allows Codex official clients"，但能走到版本门
（判定链第 5 步）的请求其实已命中官方 UA/originator，仅版本不符，
再回「只允许官方客户端」会误导。
- CodexClientRestrictionDetectionResult 增 DetectedVersion/MinCodexVersion/
MaxCodexVersion，仅在 VersionTooLow/VersionTooHigh 分支填充
- 新增 CodexClientRestrictionMessage：too_low/too_high 给出带实际版本号与
边界的提示；undetectable/未命中/黑名单/缺指纹 仍用通用兜底句（不泄露门控细节）
- Forward 唯一 403 出口改用该映射函数
Co-Authored-By: Claude Opus 4.8 (1M context)
---
.../openai_client_restriction_detector.go | 48 ++++++++++++++-
...openai_client_restriction_detector_test.go | 60 +++++++++++++++++++
.../service/openai_gateway_service.go | 2 +-
...nai_gateway_service_codex_cli_only_test.go | 46 ++++++++++++++
4 files changed, 153 insertions(+), 3 deletions(-)
diff --git a/backend/internal/service/openai_client_restriction_detector.go b/backend/internal/service/openai_client_restriction_detector.go
index abca88ce66..8a8097c879 100644
--- a/backend/internal/service/openai_client_restriction_detector.go
+++ b/backend/internal/service/openai_client_restriction_detector.go
@@ -1,6 +1,7 @@
package service
import (
+ "fmt"
"net/http"
"github.com/Wei-Shaw/sub2api/internal/config"
@@ -8,6 +9,11 @@ import (
"github.com/gin-gonic/gin"
)
+// CodexOfficialClientsOnlyMessage 是 codex_cli_only 拒绝时面向客户端的通用兜底文案。
+// 仅当拒绝原因不是「可解析版本但越界」（VersionTooLow/VersionTooHigh）时使用：
+// 未命中官方/黑名单/缺指纹/版本无法识别都沿用这句（避免向伪装客户端泄露门控细节）。
+const CodexOfficialClientsOnlyMessage = "This account only allows Codex official clients"
+
const (
// CodexClientRestrictionReasonDisabled 表示账号未开启 codex_cli_only。
CodexClientRestrictionReasonDisabled = "codex_cli_only_disabled"
@@ -51,6 +57,13 @@ type CodexClientRestrictionDetectionResult struct {
Enabled bool
Matched bool
Reason string
+ // DetectedVersion 是从官方 UA 解析出的 Codex 引擎版本；仅在版本门拒绝
+ // (VersionTooLow / VersionTooHigh) 时填充，供面向客户端的差异化文案使用。
+ DetectedVersion string
+ // MinCodexVersion 是触发 VersionTooLow 时的最低要求版本（来自策略快照）。
+ MinCodexVersion string
+ // MaxCodexVersion 是触发 VersionTooHigh 时的最高允许版本（来自策略快照）。
+ MaxCodexVersion string
}
// CodexClientRestrictionDetector 定义 codex_cli_only 统一检测入口。
@@ -127,10 +140,22 @@ func (d *OpenAICodexClientRestrictionDetector) Detect(c *gin.Context, account *A
return CodexClientRestrictionDetectionResult{Enabled: true, Matched: false, Reason: CodexClientRestrictionReasonVersionUndetectable}
}
if policy.MinCodexVersion != "" && CompareVersions(ver, policy.MinCodexVersion) < 0 {
- return CodexClientRestrictionDetectionResult{Enabled: true, Matched: false, Reason: CodexClientRestrictionReasonVersionTooLow}
+ return CodexClientRestrictionDetectionResult{
+ Enabled: true,
+ Matched: false,
+ Reason: CodexClientRestrictionReasonVersionTooLow,
+ DetectedVersion: ver,
+ MinCodexVersion: policy.MinCodexVersion,
+ }
}
if policy.MaxCodexVersion != "" && CompareVersions(ver, policy.MaxCodexVersion) > 0 {
- return CodexClientRestrictionDetectionResult{Enabled: true, Matched: false, Reason: CodexClientRestrictionReasonVersionTooHigh}
+ return CodexClientRestrictionDetectionResult{
+ Enabled: true,
+ Matched: false,
+ Reason: CodexClientRestrictionReasonVersionTooHigh,
+ DetectedVersion: ver,
+ MaxCodexVersion: policy.MaxCodexVersion,
+ }
}
}
@@ -145,3 +170,22 @@ func (d *OpenAICodexClientRestrictionDetector) Detect(c *gin.Context, account *A
return CodexClientRestrictionDetectionResult{Enabled: true, Matched: true, Reason: reason}
}
+
+// CodexClientRestrictionMessage 把检测结果映射为面向客户端的 403 文案。
+// 仅版本越界（VersionTooLow/VersionTooHigh）给出带实际版本号与边界的差异化提示——
+// 这类请求其实已被识别为官方 Codex（命中官方 UA/originator），再回「只允许官方客户端」会误导；
+// 其余拒绝原因统一沿用通用兜底句，不暴露门控细节。
+func CodexClientRestrictionMessage(r CodexClientRestrictionDetectionResult) string {
+ switch r.Reason {
+ case CodexClientRestrictionReasonVersionTooLow:
+ return fmt.Sprintf(
+ "Your Codex version (%s) is below the minimum required version (%s). Please update Codex.",
+ r.DetectedVersion, r.MinCodexVersion)
+ case CodexClientRestrictionReasonVersionTooHigh:
+ return fmt.Sprintf(
+ "Your Codex version (%s) exceeds the maximum allowed version (%s). Please downgrade Codex to %s or lower.",
+ r.DetectedVersion, r.MaxCodexVersion, r.MaxCodexVersion)
+ default:
+ return CodexOfficialClientsOnlyMessage
+ }
+}
diff --git a/backend/internal/service/openai_client_restriction_detector_test.go b/backend/internal/service/openai_client_restriction_detector_test.go
index 291c79f6bf..6c79432ae2 100644
--- a/backend/internal/service/openai_client_restriction_detector_test.go
+++ b/backend/internal/service/openai_client_restriction_detector_test.go
@@ -284,6 +284,66 @@ func TestDetect_V3_AppServerAndSkipAndVersionScope(t *testing.T) {
})
}
+func TestDetect_VersionGateCarriesVersionFields(t *testing.T) {
+ gin.SetMode(gin.TestMode)
+ d := NewOpenAICodexClientRestrictionDetector(nil)
+ acc := func() *Account {
+ return &Account{Platform: PlatformOpenAI, Type: AccountTypeOAuth, Extra: map[string]any{"codex_cli_only": true}}
+ }
+
+ t.Run("版本太低：携带 DetectedVersion + MinCodexVersion", func(t *testing.T) {
+ c := newCodexDetectorTestContext("codex_cli_rs/0.39.0 (x)", "")
+ r := d.Detect(c, acc(), CodexRestrictionPolicy{MinCodexVersion: "0.42.0"}, nil)
+ require.False(t, r.Matched)
+ require.Equal(t, CodexClientRestrictionReasonVersionTooLow, r.Reason)
+ require.Equal(t, "0.39.0", r.DetectedVersion)
+ require.Equal(t, "0.42.0", r.MinCodexVersion)
+ })
+
+ t.Run("版本太高：携带 DetectedVersion + MaxCodexVersion", func(t *testing.T) {
+ c := newCodexDetectorTestContext("codex_cli_rs/0.45.0 (x)", "")
+ r := d.Detect(c, acc(), CodexRestrictionPolicy{MaxCodexVersion: "0.42.0"}, nil)
+ require.False(t, r.Matched)
+ require.Equal(t, CodexClientRestrictionReasonVersionTooHigh, r.Reason)
+ require.Equal(t, "0.45.0", r.DetectedVersion)
+ require.Equal(t, "0.42.0", r.MaxCodexVersion)
+ })
+}
+
+func TestCodexClientRestrictionMessage(t *testing.T) {
+ t.Run("版本太低：带实际版本与最低要求", func(t *testing.T) {
+ msg := CodexClientRestrictionMessage(CodexClientRestrictionDetectionResult{
+ Reason: CodexClientRestrictionReasonVersionTooLow,
+ DetectedVersion: "0.39.0",
+ MinCodexVersion: "0.42.0",
+ })
+ require.Equal(t, "Your Codex version (0.39.0) is below the minimum required version (0.42.0). Please update Codex.", msg)
+ })
+
+ t.Run("版本太高：带实际版本与最高允许", func(t *testing.T) {
+ msg := CodexClientRestrictionMessage(CodexClientRestrictionDetectionResult{
+ Reason: CodexClientRestrictionReasonVersionTooHigh,
+ DetectedVersion: "0.45.0",
+ MaxCodexVersion: "0.42.0",
+ })
+ require.Equal(t, "Your Codex version (0.45.0) exceeds the maximum allowed version (0.42.0). Please downgrade Codex to 0.42.0 or lower.", msg)
+ })
+
+ t.Run("无法识别版本：保持原通用句", func(t *testing.T) {
+ msg := CodexClientRestrictionMessage(CodexClientRestrictionDetectionResult{
+ Reason: CodexClientRestrictionReasonVersionUndetectable,
+ })
+ require.Equal(t, "This account only allows Codex official clients", msg)
+ })
+
+ t.Run("未命中官方：保持原通用句", func(t *testing.T) {
+ msg := CodexClientRestrictionMessage(CodexClientRestrictionDetectionResult{
+ Reason: CodexClientRestrictionReasonNotMatchedUA,
+ })
+ require.Equal(t, "This account only allows Codex official clients", msg)
+ })
+}
+
func TestDetect_EngineFingerprintSignals(t *testing.T) {
gin.SetMode(gin.TestMode)
det := NewOpenAICodexClientRestrictionDetector(&config.Config{})
diff --git a/backend/internal/service/openai_gateway_service.go b/backend/internal/service/openai_gateway_service.go
index dd3d48aed0..5f76dc66f6 100644
--- a/backend/internal/service/openai_gateway_service.go
+++ b/backend/internal/service/openai_gateway_service.go
@@ -2617,7 +2617,7 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco
c.JSON(http.StatusForbidden, gin.H{
"error": gin.H{
"type": "forbidden_error",
- "message": "This account only allows Codex official clients",
+ "message": CodexClientRestrictionMessage(restrictionResult),
},
})
return nil, errors.New("codex_cli_only restriction: only codex official clients are allowed")
diff --git a/backend/internal/service/openai_gateway_service_codex_cli_only_test.go b/backend/internal/service/openai_gateway_service_codex_cli_only_test.go
index 23a1750021..7eb133c125 100644
--- a/backend/internal/service/openai_gateway_service_codex_cli_only_test.go
+++ b/backend/internal/service/openai_gateway_service_codex_cli_only_test.go
@@ -59,6 +59,52 @@ func TestOpenAIGatewayService_GetCodexClientRestrictionDetector(t *testing.T) {
})
}
+func TestOpenAIGatewayService_Forward_VersionGateMessage(t *testing.T) {
+ gin.SetMode(gin.TestMode)
+
+ newCtx := func() (*httptest.ResponseRecorder, *gin.Context) {
+ rec := httptest.NewRecorder()
+ c, _ := gin.CreateTestContext(rec)
+ c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", bytes.NewReader(nil))
+ return rec, c
+ }
+ account := func() *Account {
+ return &Account{Platform: PlatformOpenAI, Type: AccountTypeOAuth, Extra: map[string]any{"codex_cli_only": true}}
+ }
+ body := []byte(`{"model":"gpt-5.1-codex"}`)
+
+ t.Run("版本太低：返回带版本号的差异化文案", func(t *testing.T) {
+ rec, c := newCtx()
+ svc := &OpenAIGatewayService{codexDetector: &stubCodexRestrictionDetector{result: CodexClientRestrictionDetectionResult{
+ Enabled: true,
+ Matched: false,
+ Reason: CodexClientRestrictionReasonVersionTooLow,
+ DetectedVersion: "0.39.0",
+ MinCodexVersion: "0.42.0",
+ }}}
+
+ _, err := svc.Forward(context.Background(), c, account(), body)
+ require.Error(t, err)
+ require.Equal(t, http.StatusForbidden, rec.Code)
+ require.Contains(t, rec.Body.String(), "Your Codex version (0.39.0) is below the minimum required version (0.42.0)")
+ require.NotContains(t, rec.Body.String(), "This account only allows Codex official clients")
+ })
+
+ t.Run("未命中官方：仍返回通用兜底文案", func(t *testing.T) {
+ rec, c := newCtx()
+ svc := &OpenAIGatewayService{codexDetector: &stubCodexRestrictionDetector{result: CodexClientRestrictionDetectionResult{
+ Enabled: true,
+ Matched: false,
+ Reason: CodexClientRestrictionReasonNotMatchedUA,
+ }}}
+
+ _, err := svc.Forward(context.Background(), c, account(), body)
+ require.Error(t, err)
+ require.Equal(t, http.StatusForbidden, rec.Code)
+ require.Contains(t, rec.Body.String(), "This account only allows Codex official clients")
+ })
+}
+
func TestGetAPIKeyIDFromContext(t *testing.T) {
gin.SetMode(gin.TestMode)
From 1c0ccb477a11656fe42e8f3a0e3b0d0d23dc5d1c Mon Sep 17 00:00:00 2001
From: cat
Date: Mon, 6 Jul 2026 19:09:26 +0800
Subject: [PATCH 10/19] fix: add missing Codex CLI headers for OAuth account
test
The testOpenAIAccountConnection function was sending requests to
chatgpt.com/backend-api/codex/responses without the required Codex CLI
identification headers, causing 401 Unauthorized for OAuth/PAT accounts.
The gateway Forward path and the compact test path both set these headers:
- Originator: codex_cli_rs
- User-Agent: codex_cli_rs/
- OpenAI-Beta: responses=experimental
This patch adds the same headers to the normal (non-compact) OAuth test
path. API key accounts and other platforms are unaffected since the
headers are inside the isOAuth branch.
---
backend/internal/service/account_test_service.go | 3 +++
1 file changed, 3 insertions(+)
diff --git a/backend/internal/service/account_test_service.go b/backend/internal/service/account_test_service.go
index 80a862b971..a67a699e9f 100644
--- a/backend/internal/service/account_test_service.go
+++ b/backend/internal/service/account_test_service.go
@@ -600,6 +600,9 @@ func (s *AccountTestService) testOpenAIAccountConnection(c *gin.Context, account
if isOAuth {
req.Host = "chatgpt.com"
req.Header.Set("accept", "text/event-stream")
+ req.Header.Set("OpenAI-Beta", "responses=experimental")
+ req.Header.Set("Originator", "codex_cli_rs")
+ req.Header.Set("User-Agent", codexCLIUserAgent)
setOpenAIChatGPTAccountHeaders(req.Header, credentialAccount)
}
From 5089c303e1c056ebc038921f385c3dd5301516c7 Mon Sep 17 00:00:00 2001
From: jjaw
Date: Mon, 6 Jul 2026 20:27:13 +0800
Subject: [PATCH 11/19] =?UTF-8?q?=E4=BC=98=E5=8C=96=20Redis=20SCAN=20?=
=?UTF-8?q?=E6=B8=85=E7=90=86=E6=9E=B6=E6=9E=84?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
---
REDIS_SCAN_ARCHITECTURE_OPTIMIZATION.md | 568 ++++++++++++++++++
.../internal/repository/concurrency_cache.go | 487 +++++++++++----
.../concurrency_cache_integration_test.go | 134 ++++-
.../repository/user_msg_queue_cache.go | 205 ++++---
.../user_msg_queue_cache_integration_test.go | 128 ++++
.../internal/service/concurrency_service.go | 2 +-
.../service/user_msg_queue_service.go | 26 +-
.../service/user_msg_queue_service_test.go | 54 ++
8 files changed, 1383 insertions(+), 221 deletions(-)
create mode 100644 REDIS_SCAN_ARCHITECTURE_OPTIMIZATION.md
create mode 100644 backend/internal/repository/user_msg_queue_cache_integration_test.go
create mode 100644 backend/internal/service/user_msg_queue_service_test.go
diff --git a/REDIS_SCAN_ARCHITECTURE_OPTIMIZATION.md b/REDIS_SCAN_ARCHITECTURE_OPTIMIZATION.md
new file mode 100644
index 0000000000..7d95419a39
--- /dev/null
+++ b/REDIS_SCAN_ARCHITECTURE_OPTIMIZATION.md
@@ -0,0 +1,568 @@
+# Redis SCAN 架构优化执行文档
+
+本文是架构级执行文档，只覆盖三个目标：
+
+1. 账号并发活跃负载查询不再通过 Redis keyspace `SCAN` 发现账号。
+2. 账号/用户并发槽过期清理、启动遗留槽清理不再通过 Redis keyspace `SCAN` 发现 key。
+3. 用户消息队列孤儿锁清理不再通过 Redis keyspace `SCAN` 发现 lock key。
+
+不覆盖旁路录制、OpenAI failover、日志量、业务限流配置调参。不要把本文扩展成短期止血方案。
+
+## 成功标准
+
+实现完成后必须同时满足：
+
+- `backend/internal/repository/concurrency_cache.go` 中不得再调用 `c.rdb.Scan(...)`。
+- `backend/internal/repository/user_msg_queue_cache.go` 中不得再调用 `c.rdb.Scan(...)`。
+- `backend/internal/service/user_msg_queue_service.go` 中不得再出现 `ScanLockKeys` 接口调用。
+- `GetActiveAccountLoadMap` 只读显式维护的 Redis 索引，不扫描 Redis keyspace。
+- `CleanupExpiredAccountSlotKeys` 只处理显式索引中的候选账号，不扫描 Redis keyspace。
+- `CleanupStaleProcessSlots` 不扫描 Redis keyspace；它必须基于显式索引清理，或只依赖 TTL/score 自然过期。
+- UMQ cleanup worker 只读 `umq:lock:index` 候选项，不扫描 `umq:{*}:lock`。
+- 主业务并发限制仍以原账号/用户 slot key 为准，不能以索引为准。索引只能用于发现候选对象、监控和清理。
+
+执行完必须用下面命令确认生产代码没有遗留扫描：
+
+```powershell
+rg -n "\.Scan\(" backend/internal/repository/concurrency_cache.go backend/internal/repository/user_msg_queue_cache.go backend/internal/service/user_msg_queue_service.go
+rg -n "ScanLockKeys|scanAccountIDsByPrefix|cleanupSlotsByPattern|deleteKeysByPattern|umqScanPattern" backend/internal/repository/concurrency_cache.go backend/internal/repository/user_msg_queue_cache.go backend/internal/service/user_msg_queue_service.go
+```
+
+第一条必须无输出。第二条必须无生产函数残留；测试文件不在此检查范围。
+
+## 不可违反的约束
+
+- 不要用 `KEYS` 替代 `SCAN`。
+- 不要把全量 Redis keyspace 扫描移动到另一个函数、goroutine、启动流程或管理接口里。
+- 不要在请求路径、后台 worker、启动流程中做 Redis keyspace pattern enumeration。
+- 不要在 Redis Lua 脚本里同时操作“全局索引 key”和“账号/用户局部 key”。项目代码当前有 Redis Cluster 兼容要求，这种写法会在 Cluster 下触发 CROSSSLOT。
+- 索引更新失败不得改变主业务 acquire/release 的成功结果。索引是 best-effort discovery structure，不是并发正确性的来源。
+- 不能因为索引缺失而拒绝用户请求。索引缺失最多影响 Ops 实时视图和后台提前清理；原 slot/wait key 的 TTL 必须保证最终自愈。
+
+## 新增 Redis Key
+
+### 并发索引
+
+在 `backend/internal/repository/concurrency_cache.go` 增加常量：
+
+```go
+const (
+ accountActiveIndexKey = "concurrency:account:active_index" // ZSET member=accountID, score=expireAtUnixSeconds
+ userActiveIndexKey = "concurrency:user:active_index" // ZSET member=userID, score=expireAtUnixSeconds
+
+ activeIndexCleanupBatchSize = 1000
+ activeIndexPipelineChunkSize = 500
+)
+```
+
+语义：
+
+- `accountActiveIndexKey` 记录“可能有账号槽位或账号等待计数”的账号 ID。
+- `userActiveIndexKey` 记录“可能有用户槽位或用户等待计数”的用户 ID。
+- ZSET score 是候选对象的保守过期时间，单位为 Unix 秒。
+- member 必须是十进制 ID 字符串，不要存完整 Redis key。
+- 索引允许短暂 stale；读索引后必须二次查询真实 slot/wait key。
+
+score 规则：
+
+- 成功获取账号槽位：score = Redis 当前秒 + `slotTTLSeconds`。
+- 成功增加账号等待计数：score = Redis 当前秒 + `waitQueueTTLSeconds`。
+- 成功获取用户槽位：score = Redis 当前秒 + `slotTTLSeconds`。
+- 成功增加用户等待计数：score = Redis 当前秒 + `waitQueueTTLSeconds`。
+- release/decrement 后如果真实 slot count 和 wait count 都为 0，则从索引 `ZREM`。
+- release/decrement 后如果仍有 slot 或 wait，则重新 `ZADD` 一个新的保守过期时间。
+
+### UMQ 锁索引
+
+在 `backend/internal/repository/user_msg_queue_cache.go` 增加常量：
+
+```go
+const (
+ umqLockIndexKey = "umq:lock:index" // ZSET member=accountID, score=lockExpireAtUnixMs
+ umqLockIndexCleanupBatchSize = 1000
+)
+```
+
+语义：
+
+- `umqLockIndexKey` 记录“可能存在 UMQ lock”的账号 ID。
+- ZSET score 是 lock 的预计过期时间，单位为 Unix 毫秒。
+- member 必须是十进制 accountID 字符串。
+- 索引只用于 cleanup worker 找候选 lock。锁是否存在、是否孤儿，必须再查 `umq:{accountID}:lock`。
+
+## 第一部分：并发活跃索引
+
+修改文件：`backend/internal/repository/concurrency_cache.go`。
+
+### 1.1 增加 Redis 时间 helper
+
+新增 helper，所有索引 score 使用 Redis server time，不用本机时间：
+
+```go
+func (c *concurrencyCache) redisUnixSeconds(ctx context.Context) (int64, error) {
+ now, err := c.rdb.Time(ctx).Result()
+ if err != nil {
+ return 0, fmt.Errorf("redis TIME: %w", err)
+ }
+ return now.Unix(), nil
+}
+```
+
+不要在 Lua 脚本里写全局索引，避免 CROSSSLOT。
+
+### 1.2 增加索引 touch/remove/refresh helper
+
+新增以下 helper。名字可以微调，但行为不能改。
+
+```go
+func (c *concurrencyCache) touchAccountActiveIndex(ctx context.Context, accountID int64, ttlSeconds int) {
+ c.touchActiveIndex(ctx, accountActiveIndexKey, accountID, ttlSeconds)
+}
+
+func (c *concurrencyCache) touchUserActiveIndex(ctx context.Context, userID int64, ttlSeconds int) {
+ c.touchActiveIndex(ctx, userActiveIndexKey, userID, ttlSeconds)
+}
+
+func (c *concurrencyCache) touchActiveIndex(ctx context.Context, indexKey string, id int64, ttlSeconds int) {
+ if c == nil || c.rdb == nil || id <= 0 || ttlSeconds <= 0 {
+ return
+ }
+ now, err := c.redisUnixSeconds(ctx)
+ if err != nil {
+ return
+ }
+ _ = c.rdb.ZAdd(ctx, indexKey, redis.Z{
+ Score: float64(now + int64(ttlSeconds)),
+ Member: strconv.FormatInt(id, 10),
+ }).Err()
+}
+```
+
+索引维护是 best-effort，所以 helper 内部吞掉错误。不要把索引错误返回给 acquire/release 调用方。
+
+再新增 refresh helper：
+
+```go
+func (c *concurrencyCache) refreshAccountActiveIndex(ctx context.Context, accountID int64) {
+ // 真实状态以 accountSlotKey(accountID) 和 accountWaitKey(accountID) 为准。
+ // 先清理该账号 slot 中过期成员，再读 ZCARD 和 GET wait。
+ // 如果 slotCount == 0 && waitCount <= 0：ZREM accountActiveIndexKey accountID。
+ // 否则：ZADD accountActiveIndexKey accountID，score = now + maxRelevantTTL。
+}
+
+func (c *concurrencyCache) refreshUserActiveIndex(ctx context.Context, userID int64) {
+ // 真实状态以 userSlotKey(userID) 和 waitQueueKey(userID) 为准。
+ // 行为同 refreshAccountActiveIndex。
+}
+```
+
+实现要求：
+
+- `refresh*` 必须 best-effort，不能向 release/decrement 返回索引错误。
+- `waitCount` 读取 `redis.Nil` 时按 0 处理。
+- `waitCount < 0` 必须按 0 处理。
+- `slotCount > 0` 时 score 至少延长 `slotTTLSeconds`。
+- `waitCount > 0` 时 score 至少延长 `waitQueueTTLSeconds`。
+- 两者都存在时使用更大的 TTL。
+
+### 1.3 修改账号写路径
+
+修改 `AcquireAccountSlot`：
+
+```go
+result, err := acquireScript.Run(...).Int()
+if err != nil { return false, err }
+if result == 1 {
+ c.touchAccountActiveIndex(ctx, accountID, c.slotTTLSeconds)
+}
+return result == 1, nil
+```
+
+修改 `ReleaseAccountSlot`：
+
+```go
+if err := c.rdb.ZRem(ctx, key, requestID).Err(); err != nil {
+ return err
+}
+c.refreshAccountActiveIndex(ctx, accountID)
+return nil
+```
+
+修改 `IncrementAccountWaitCount`：
+
+```go
+result, err := incrementAccountWaitScript.Run(...).Int()
+if err != nil { return false, err }
+if result == 1 {
+ c.touchAccountActiveIndex(ctx, accountID, c.waitQueueTTLSeconds)
+}
+return result == 1, nil
+```
+
+修改 `DecrementAccountWaitCount`：
+
+```go
+_, err := decrementWaitScript.Run(...).Result()
+if err == nil {
+ c.refreshAccountActiveIndex(ctx, accountID)
+}
+return err
+```
+
+### 1.4 修改用户写路径
+
+同账号路径，修改：
+
+- `AcquireUserSlot`
+- `ReleaseUserSlot`
+- `IncrementWaitCount`
+- `DecrementWaitCount`
+
+用户索引使用 `userActiveIndexKey`。
+
+### 1.5 重写 GetActiveAccountLoadMap
+
+删除 `scanAccountIDsByPrefix` 和 `parseAccountIDFromPrefixedKey` 的生产调用。`GetActiveAccountLoadMap` 必须改成：
+
+1. 获取 Redis 当前秒。
+2. `ZRemRangeByScore(accountActiveIndexKey, "-inf", strconv.FormatInt(now, 10))` 删除过期候选。
+3. `ZRangeByScore(accountActiveIndexKey, &redis.ZRangeBy{Min: strconv.FormatInt(now+1, 10), Max: "+inf"})` 获取候选账号 ID。
+4. 解析 member 为 `int64`，非法 member 记录到待删除列表。
+5. 分块 pipeline，块大小 `activeIndexPipelineChunkSize`。
+6. 对每个候选账号执行：
+ - `ZRemRangeByScore(accountSlotKey(id), "-inf", cutoffUnixSeconds)`
+ - `ZCard(accountSlotKey(id))`
+ - `Get(accountWaitKey(id))`
+7. 构造结果时只返回 `currentConcurrency > 0 || waitingCount > 0` 的账号。
+8. 对真实状态为空或 member 非法的账号执行 `ZREM accountActiveIndexKey member`。
+9. 对真实状态仍活跃但 index score 已接近过期的账号，调用 `touchAccountActiveIndex` 刷新。
+
+禁止：
+
+- 禁止再扫 `concurrency:account:*`。
+- 禁止再扫 `wait:account:*`。
+- 禁止用索引里的 score 直接判断并发数。
+
+## 第二部分：并发槽清理和启动清理
+
+修改文件：`backend/internal/repository/concurrency_cache.go`。
+
+### 2.1 重写 CleanupExpiredAccountSlotKeys
+
+当前实现调用 `cleanupExpiredSlotKeysByPattern(ctx, accountSlotKeyPrefix+"*")`，必须删除。
+
+新行为：
+
+1. 获取 Redis 当前秒 `now`。
+2. 从 `accountActiveIndexKey` 读取过期候选：
+
+```go
+ids, err := c.rdb.ZRangeByScore(ctx, accountActiveIndexKey, &redis.ZRangeBy{
+ Min: "-inf",
+ Max: strconv.FormatInt(now, 10),
+ Count: activeIndexCleanupBatchSize,
+}).Result()
+```
+
+3. 对每个候选账号清理该账号 slot 过期成员并读真实状态。
+4. 如果真实 `slotCount == 0 && waitCount <= 0`，从 `accountActiveIndexKey` 删除该账号。
+5. 如果真实仍活跃，刷新 `accountActiveIndexKey` score。
+6. 不需要处理不在索引中的账号；其 slot key 自身有 `EXPIRE`，并且 acquire/get-load 会惰性清理过期成员。
+
+这个函数不再表示“遍历所有账号槽位 key”，而是“处理索引中到期的账号候选”。保留原函数名是为了少改接口。
+
+### 2.2 重写 CleanupStaleProcessSlots
+
+当前实现会扫描：
+
+- `concurrency:account:*`
+- `concurrency:user:*`
+- `wait:account:*`
+- `concurrency:wait:*`
+
+必须去掉这些扫描。
+
+新行为必须基于索引：
+
+1. 从 `accountActiveIndexKey` 读取所有未过期候选账号。
+2. 对每个账号：
+ - 对 `accountSlotKey(id)` 运行“单 key 清理脚本”，删除 requestID 前缀不是当前 `activeRequestPrefix` 的成员。
+ - 删除 `accountWaitKey(id)`，因为等待者属于旧进程，重启后不能继续等待。
+ - 调用 `refreshAccountActiveIndex(ctx, id)`。
+3. 从 `userActiveIndexKey` 读取所有未过期候选用户。
+4. 对每个用户：
+ - 对 `userSlotKey(id)` 运行同一个“单 key 清理脚本”。
+ - 删除 `waitQueueKey(id)`。
+ - 调用 `refreshUserActiveIndex(ctx, id)`。
+
+新增单 key Lua 脚本，替代当前 `startupCleanupScript` 的多 key 版本：
+
+```lua
+local key = KEYS[1]
+local activePrefix = ARGV[1]
+local slotTTL = tonumber(ARGV[2])
+local removed = 0
+local members = redis.call('ZRANGE', key, 0, -1)
+for _, member in ipairs(members) do
+ if string.sub(member, 1, string.len(activePrefix)) ~= activePrefix then
+ removed = removed + redis.call('ZREM', key, member)
+ end
+end
+if redis.call('ZCARD', key) == 0 then
+ redis.call('DEL', key)
+else
+ redis.call('EXPIRE', key, slotTTL)
+end
+return removed
+```
+
+该脚本只接受一个 slot key，避免 Redis Cluster CROSSSLOT。
+
+如果索引不存在或为空：
+
+- `CleanupStaleProcessSlots` 直接返回 nil。
+- 不要 fallback 到 `SCAN`。
+- 旧版本遗留 key 依赖 Redis TTL 自然过期。不要在 app 启动时做兼容性 keyspace backfill。
+
+### 2.3 删除旧扫描函数
+
+删除以下生产函数：
+
+- `scanAccountIDsByPrefix`
+- `parseAccountIDFromPrefixedKey`，如果没有其他生产调用
+- `cleanupExpiredSlotKeysByPattern`
+- `cleanupSlotsByPattern`
+- `deleteKeysByPattern`
+
+如果测试需要解析 key，测试内自建 helper，不要保留生产 helper。
+
+## 第三部分：UMQ 锁索引
+
+修改文件：
+
+- `backend/internal/repository/user_msg_queue_cache.go`
+- `backend/internal/service/user_msg_queue_service.go`
+
+### 3.1 修改 service 接口
+
+在 `backend/internal/service/user_msg_queue_service.go` 的 `UserMsgQueueCache` 接口中删除：
+
+```go
+ScanLockKeys(ctx context.Context, maxCount int) ([]int64, error)
+ForceReleaseLock(ctx context.Context, accountID int64) error
+```
+
+替换为：
+
+```go
+ReconcileExpiredLockCandidates(ctx context.Context, maxCount int) (cleaned int, err error)
+```
+
+原因：cleanup worker 不应该知道 Redis lock key 的枚举方式，也不应该先枚举再逐个 `ForceReleaseLock`。候选读取、PTTL 校验、索引刷新应该封装在 cache 层。
+
+### 3.2 修改 acquireLockScript 返回值
+
+当前脚本只返回 0/1。改成返回数组：
+
+```lua
+redis.replicate_commands()
+local cur = redis.call('GET', KEYS[1])
+local ttl = tonumber(ARGV[2])
+if cur == ARGV[1] then
+ redis.call('PEXPIRE', KEYS[1], ttl)
+ local t = redis.call('TIME')
+ local ms = tonumber(t[1])*1000 + math.floor(tonumber(t[2])/1000)
+ return {1, ms + ttl}
+end
+if cur ~= false then
+ return {0, 0}
+end
+redis.call('SET', KEYS[1], ARGV[1], 'PX', ttl)
+local t = redis.call('TIME')
+local ms = tonumber(t[1])*1000 + math.floor(tonumber(t[2])/1000)
+return {1, ms + ttl}
+```
+
+Go 侧解析：
+
+- 第一个元素是 acquired，1 表示拿到锁。
+- 第二个元素是 Redis 时间计算出的 `expireAtUnixMs`。
+- acquired 为 1 时，best-effort 写 `ZADD umqLockIndexKey expireAtMs accountID`。
+- `ZADD` 失败不能让 `AcquireLock` 返回失败。
+
+### 3.3 修改 ReleaseLock
+
+`ReleaseLock` 主逻辑保持原子释放锁和写 last key。
+
+释放成功时：
+
+```go
+if result == 1 {
+ _ = c.rdb.ZRem(ctx, umqLockIndexKey, strconv.FormatInt(accountID, 10)).Err()
+}
+```
+
+释放失败时不要删除索引。失败可能是 requestID 不匹配或 lock 已过期；cleanup worker 会处理 stale index。
+
+### 3.4 新增 reconcile 脚本
+
+删除 `forceReleaseLockScript` 的外部使用。新增脚本：
+
+```lua
+local pttl = redis.call('PTTL', KEYS[1])
+if pttl == -2 then
+ return {-2, 0}
+end
+if pttl == -1 then
+ redis.call('DEL', KEYS[1])
+ return {-1, 0}
+end
+return {1, pttl}
+```
+
+返回语义：
+
+- `-2`：lock key 不存在。Go 侧 `ZREM umqLockIndexKey accountID`。
+- `-1`：lock key 存在但无 TTL，脚本已删除。Go 侧 `ZREM umqLockIndexKey accountID`，cleaned++。
+- `1`：lock key 仍有 TTL。Go 侧用 Redis 当前毫秒 + pttl 刷新 `umqLockIndexKey` score。
+
+### 3.5 实现 ReconcileExpiredLockCandidates
+
+实现步骤：
+
+1. 用 `c.rdb.Time(ctx)` 获取 Redis 当前毫秒 `nowMs`。
+2. 从 `umqLockIndexKey` 取到期候选：
+
+```go
+members, err := c.rdb.ZRangeByScore(ctx, umqLockIndexKey, &redis.ZRangeBy{
+ Min: "-inf",
+ Max: strconv.FormatInt(nowMs, 10),
+ Count: int64(maxCount),
+}).Result()
+```
+
+3. 逐个解析 accountID。非法 member 直接 `ZREM`。
+4. 对合法 accountID 运行 reconcile 脚本，key 为 `umqLockKey(accountID)`。
+5. 根据返回值删除索引、刷新索引或累计 cleaned。
+6. 函数返回 cleaned 数。
+
+禁止：
+
+- 禁止 fallback 到 `SCAN umq:{*}:lock`。
+- 禁止用 `KEYS umq:*`。
+- 禁止 cleanup worker 自己解析 lock key。
+
+### 3.6 修改 StartCleanupWorker
+
+当前 worker 先 `ScanLockKeys` 再逐个 `ForceReleaseLock`。改成：
+
+```go
+cleaned, err := s.cache.ReconcileExpiredLockCandidates(ctx, 1000)
+if err != nil {
+ logger.LegacyPrintf("service.umq", "Cleanup reconcile failed: %v", err)
+ return
+}
+if cleaned > 0 {
+ logger.LegacyPrintf("service.umq", "Cleanup completed: released %d orphaned locks", cleaned)
+}
+```
+
+worker 不再知道扫描、PTTL、索引等细节。
+
+### 3.7 删除旧 UMQ 扫描函数
+
+删除：
+
+- `umqScanPattern`
+- `ScanLockKeys`
+- `ForceReleaseLock`，如果无生产调用
+
+如果测试仍需要强造 PTTL == -1 的 key，只在测试里直接写 Redis。
+
+## 测试要求
+
+### 并发缓存测试
+
+新增或修改 `backend/internal/repository/concurrency_cache_*_test.go`。
+
+必须覆盖：
+
+1. `AcquireAccountSlot` 成功后 `GetActiveAccountLoadMap` 能看到该账号。
+2. `ReleaseAccountSlot` 后 `GetActiveAccountLoadMap` 不再返回该账号。
+3. `IncrementAccountWaitCount` 成功后 `GetActiveAccountLoadMap` 能看到 waiting count。
+4. `DecrementAccountWaitCount` 后如果无 slot，则索引被移除。
+5. `CleanupExpiredAccountSlotKeys` 不依赖 keyspace scan：测试里只创建索引成员和对应 slot key，然后确认会清理；再创建未索引 slot key，确认不会被该函数主动发现。
+6. `CleanupStaleProcessSlots` 只处理索引中的 account/user，删除旧 request prefix 成员，保留当前 prefix 成员，删除 account/user wait key。
+7. 索引中存在非法 member 时，`GetActiveAccountLoadMap` 不报错，并移除非法 member。
+
+### UMQ 测试
+
+新增或修改 `backend/internal/repository/user_msg_queue_cache*_test.go` 和 `backend/internal/service/user_msg_queue_service*_test.go`。
+
+必须覆盖：
+
+1. `AcquireLock` 成功后写入 `umq:lock:index`，score 大于 Redis 当前毫秒。
+2. `ReleaseLock` 成功后删除 `umq:lock:index` member。
+3. lock 已自然过期时，`ReconcileExpiredLockCandidates` 删除 stale index member。
+4. lock 仍有 TTL 但 index score 到期时，`ReconcileExpiredLockCandidates` 刷新 index score，不删除 lock。
+5. lock 存在且 `PTTL == -1` 时，`ReconcileExpiredLockCandidates` 删除 lock，删除 index member，并返回 cleaned=1。
+6. index 中非法 member 不导致错误，并被删除。
+7. `StartCleanupWorker` 调用 `ReconcileExpiredLockCandidates`，不再调用 `ScanLockKeys` 或 `ForceReleaseLock`。
+
+### 禁止项测试
+
+实现完成后运行：
+
+```powershell
+rg -n "\.Scan\(" backend/internal/repository/concurrency_cache.go backend/internal/repository/user_msg_queue_cache.go backend/internal/service/user_msg_queue_service.go
+rg -n "ScanLockKeys|umqScanPattern|scanAccountIDsByPrefix|cleanupExpiredSlotKeysByPattern|cleanupSlotsByPattern|deleteKeysByPattern" backend/internal/repository/concurrency_cache.go backend/internal/repository/user_msg_queue_cache.go backend/internal/service/user_msg_queue_service.go
+```
+
+上述命令必须无输出。
+
+再运行相关测试。按项目约定，编译很慢时先把代码复制到 WSL 文件系统再跑：
+
+```bash
+cd backend
+go test ./internal/repository ./internal/service
+```
+
+如果全量包太慢，至少先跑：
+
+```bash
+cd backend
+go test ./internal/repository -run 'Concurrency|UserMsgQueue|Redis'
+go test ./internal/service -run 'Concurrency|UserMessageQueue'
+```
+
+## 迁移和兼容
+
+不要在应用启动时扫描旧 key 回填索引。
+
+原因：
+
+- 这会把问题从运行期 `SCAN` 搬到启动期 `SCAN`。
+- 生产实例重启时 Redis 已经高 CPU，启动扫描会放大抖动。
+- 并发 slot key 和 wait key 都有 TTL，新版本写路径会为新流量维护索引，旧 key 可自然过期。
+
+兼容策略：
+
+- 新版本上线后，新请求会逐步填充 `concurrency:*:active_index` 和 `umq:lock:index`。
+- 旧并发 slot key 没有索引时，不影响并发限制本身；对应账号下一次 acquire/get-load 会清理自己的 slot。
+- 旧 UMQ lock 如果有 TTL，会自然过期。
+- 极少数历史 `PTTL == -1` UMQ lock 且没有 index 的情况，不由应用自动发现。需要人工离线维护时，单独写一次性脚本，维护窗口运行，不要放进服务启动或后台 worker。
+
+## 代码审查检查表
+
+提交前逐项确认：
+
+- [ ] 没有新增 `KEYS`。
+- [ ] 没有新增生产路径 `SCAN`。
+- [ ] 没有在 Lua 脚本中同时操作全局索引 key 和账号/用户局部 key。
+- [ ] 索引维护失败不会让 acquire/release/decrement 的主结果失败。
+- [ ] `GetActiveAccountLoadMap` 对 stale index、非法 member、Redis nil 都能正常返回。
+- [ ] `CleanupExpiredAccountSlotKeys` 不再遍历 keyspace。
+- [ ] `CleanupStaleProcessSlots` 不再遍历 keyspace。
+- [ ] UMQ cleanup worker 不再知道 lock key pattern。
+- [ ] 所有旧扫描 helper 已删除或仅存在于测试文件。
+- [ ] 新测试覆盖成功路径、stale index、非法 member、PTTL -1、自然过期。
diff --git a/backend/internal/repository/concurrency_cache.go b/backend/internal/repository/concurrency_cache.go
index b530c30b90..8cdb5015de 100644
--- a/backend/internal/repository/concurrency_cache.go
+++ b/backend/internal/repository/concurrency_cache.go
@@ -34,6 +34,15 @@ const (
// 默认槽位过期时间（分钟），可通过配置覆盖
defaultSlotTTLMinutes = 15
+
+ // 活跃索引用来替代后台任务全量 SCAN 槽位键。
+ // member 是账号/用户 ID，score 是“预计仍需关注到”的 Redis Unix 秒时间戳。
+ accountActiveIndexKey = "concurrency:account:active_index" // ZSET member=accountID, score=expireAtUnixSeconds
+ userActiveIndexKey = "concurrency:user:active_index" // ZSET member=userID, score=expireAtUnixSeconds
+
+ // 后台清理只按批处理索引候选，避免单次任务占用 Redis 太久。
+ activeIndexCleanupBatchSize = 1000
+ activeIndexPipelineChunkSize = 500
)
var (
@@ -174,50 +183,24 @@ var (
return 1
`)
- // cleanupExpiredSlotKeysScript 批量清理实际存在的账号槽位键，避免后台任务从数据库加载全量账号。
- // KEYS = 有序集合键列表，ARGV[1] = TTL（秒）。
- cleanupExpiredSlotKeysScript = redis.NewScript(`
- -- Redis 3.2-4.x compat: opt into effects replication so redis.call('TIME')
- -- replicates correctly. No-op on Redis 5.0+ (effects replication is default).
- redis.replicate_commands()
- local ttl = tonumber(ARGV[1])
- local timeResult = redis.call('TIME')
- local now = tonumber(timeResult[1])
- local expireBefore = now - ttl
- local removed = 0
- for i = 1, #KEYS do
- local key = KEYS[i]
- removed = removed + redis.call('ZREMRANGEBYSCORE', key, '-inf', expireBefore)
- if redis.call('ZCARD', key) == 0 then
- redis.call('DEL', key)
- else
- redis.call('EXPIRE', key, ttl)
- end
- end
- return removed
- `)
-
- // startupCleanupScript 清理非当前进程前缀的槽位成员。
- // KEYS 是有序集合键列表，ARGV[1] 是当前进程前缀，ARGV[2] 是槽位 TTL。
- // 遍历每个 KEYS[i]，移除前缀不匹配的成员，清空后删 key，否则刷新 EXPIRE。
- startupCleanupScript = redis.NewScript(`
+ // startupCleanupSlotScript 清理单个槽位 key 中非当前进程前缀的成员，避免 Redis Cluster CROSSSLOT。
+ // KEYS[1] 是有序集合键，ARGV[1] 是当前进程前缀，ARGV[2] 是槽位 TTL。
+ startupCleanupSlotScript = redis.NewScript(`
+ local key = KEYS[1]
local activePrefix = ARGV[1]
local slotTTL = tonumber(ARGV[2])
local removed = 0
- for i = 1, #KEYS do
- local key = KEYS[i]
- local members = redis.call('ZRANGE', key, 0, -1)
- for _, member in ipairs(members) do
- if string.sub(member, 1, string.len(activePrefix)) ~= activePrefix then
- removed = removed + redis.call('ZREM', key, member)
- end
- end
- if redis.call('ZCARD', key) == 0 then
- redis.call('DEL', key)
- else
- redis.call('EXPIRE', key, slotTTL)
+ local members = redis.call('ZRANGE', key, 0, -1)
+ for _, member in ipairs(members) do
+ if string.sub(member, 1, string.len(activePrefix)) ~= activePrefix then
+ removed = removed + redis.call('ZREM', key, member)
end
end
+ if redis.call('ZCARD', key) == 0 then
+ redis.call('DEL', key)
+ else
+ redis.call('EXPIRE', key, slotTTL)
+ end
return removed
`)
)
@@ -262,6 +245,198 @@ func accountWaitKey(accountID int64) string {
return fmt.Sprintf("%s%d", accountWaitKeyPrefix, accountID)
}
+// redisUnixSeconds 统一使用 Redis 服务器时间，避免多实例本地时钟漂移导致索引提前/延后过期。
+func (c *concurrencyCache) redisUnixSeconds(ctx context.Context) (int64, error) {
+ now, err := c.rdb.Time(ctx).Result()
+ if err != nil {
+ return 0, fmt.Errorf("redis TIME: %w", err)
+ }
+ return now.Unix(), nil
+}
+
+func (c *concurrencyCache) touchAccountActiveIndex(ctx context.Context, accountID int64, ttlSeconds int) {
+ c.touchActiveIndex(ctx, accountActiveIndexKey, accountID, ttlSeconds)
+}
+
+func (c *concurrencyCache) touchUserActiveIndex(ctx context.Context, userID int64, ttlSeconds int) {
+ c.touchActiveIndex(ctx, userActiveIndexKey, userID, ttlSeconds)
+}
+
+// touchActiveIndex 是写路径上的轻量标记：主操作已成功时，尽力把 ID 放入活跃索引。
+// 索引失败不影响并发槽位/等待队列本身，后续释放或清理会再次校正。
+func (c *concurrencyCache) touchActiveIndex(ctx context.Context, indexKey string, id int64, ttlSeconds int) {
+ if c == nil || c.rdb == nil || id <= 0 || ttlSeconds <= 0 {
+ return
+ }
+ now, err := c.redisUnixSeconds(ctx)
+ if err != nil {
+ return
+ }
+ _ = c.rdb.ZAdd(ctx, indexKey, redis.Z{
+ Score: float64(now + int64(ttlSeconds)),
+ Member: strconv.FormatInt(id, 10),
+ }).Err()
+}
+
+func (c *concurrencyCache) refreshAccountActiveIndex(ctx context.Context, accountID int64) {
+ c.refreshActiveIndex(ctx, accountActiveIndexKey, accountID, accountSlotKey(accountID), accountWaitKey(accountID))
+}
+
+func (c *concurrencyCache) refreshUserActiveIndex(ctx context.Context, userID int64) {
+ c.refreshActiveIndex(ctx, userActiveIndexKey, userID, userSlotKey(userID), waitQueueKey(userID))
+}
+
+// refreshActiveIndex 以 Redis 中的真实槽位/等待数为准重建索引状态。
+// 释放槽位、等待计数减少、清理过期成员后都会调用它，防止索引残留。
+func (c *concurrencyCache) refreshActiveIndex(ctx context.Context, indexKey string, id int64, slotKey, waitKey string) {
+ if c == nil || c.rdb == nil || id <= 0 {
+ return
+ }
+ now, err := c.redisUnixSeconds(ctx)
+ if err != nil {
+ return
+ }
+
+ load, err := c.readActiveLoadForKey(ctx, id, slotKey, waitKey, now)
+ if err != nil {
+ return
+ }
+ member := strconv.FormatInt(id, 10)
+ if load.slotCount == 0 && load.waitCount <= 0 {
+ _ = c.rdb.ZRem(ctx, indexKey, member).Err()
+ return
+ }
+
+ ttlSeconds := c.activeIndexTTL(load.slotCount, load.waitCount)
+ if ttlSeconds <= 0 {
+ return
+ }
+ _ = c.rdb.ZAdd(ctx, indexKey, redis.Z{
+ Score: float64(now + int64(ttlSeconds)),
+ Member: member,
+ }).Err()
+}
+
+type activeIndexLoad struct {
+ id int64
+ member string
+ slotCount int
+ waitCount int
+}
+
+// activeIndexTTL 取槽位 TTL 与等待队列 TTL 中仍然需要关注的较大值。
+// 只要并发槽位或等待计数还有负载，就保留索引；两者都为 0 时调用方会删除索引。
+func (c *concurrencyCache) activeIndexTTL(slotCount int, waitCount int) int {
+ ttlSeconds := 0
+ if slotCount > 0 {
+ ttlSeconds = c.slotTTLSeconds
+ }
+ if waitCount > 0 && c.waitQueueTTLSeconds > ttlSeconds {
+ ttlSeconds = c.waitQueueTTLSeconds
+ }
+ return ttlSeconds
+}
+
+// readActiveLoadForKey 读取单个 ID 的当前负载，并顺手清理该槽位集合中的过期成员。
+func (c *concurrencyCache) readActiveLoadForKey(ctx context.Context, id int64, slotKey, waitKey string, now int64) (activeIndexLoad, error) {
+ cutoffTime := now - int64(c.slotTTLSeconds)
+ pipe := c.rdb.Pipeline()
+ pipe.ZRemRangeByScore(ctx, slotKey, "-inf", strconv.FormatInt(cutoffTime, 10))
+ zcardCmd := pipe.ZCard(ctx, slotKey)
+ getCmd := pipe.Get(ctx, waitKey)
+ if _, err := pipe.Exec(ctx); err != nil && !errors.Is(err, redis.Nil) {
+ return activeIndexLoad{}, fmt.Errorf("pipeline exec: %w", err)
+ }
+
+ waitCount := 0
+ if v, err := getCmd.Int(); err == nil && v > 0 {
+ waitCount = v
+ }
+ return activeIndexLoad{
+ id: id,
+ member: strconv.FormatInt(id, 10),
+ slotCount: int(zcardCmd.Val()),
+ waitCount: waitCount,
+ }, nil
+}
+
+// readAccountIndexLoads 批量读取账号索引候选的真实负载。
+// 分块 Pipeline 可以减少 Redis 往返，同时避免一次 Pipeline 塞入过多命令。
+func (c *concurrencyCache) readAccountIndexLoads(ctx context.Context, members []string, now int64) ([]activeIndexLoad, []string, error) {
+ loads := make([]activeIndexLoad, 0, len(members))
+ staleMembers := make([]string, 0)
+ candidates := make([]activeIndexLoad, 0, len(members))
+ for _, member := range members {
+ id, err := strconv.ParseInt(member, 10, 64)
+ if err != nil || id <= 0 {
+ staleMembers = append(staleMembers, member)
+ continue
+ }
+ candidates = append(candidates, activeIndexLoad{id: id, member: member})
+ }
+
+ cutoffTime := now - int64(c.slotTTLSeconds)
+ for start := 0; start < len(candidates); start += activeIndexPipelineChunkSize {
+ end := start + activeIndexPipelineChunkSize
+ if end > len(candidates) {
+ end = len(candidates)
+ }
+ chunk := candidates[start:end]
+
+ pipe := c.rdb.Pipeline()
+ type accountCmd struct {
+ activeIndexLoad
+ zcardCmd *redis.IntCmd
+ getCmd *redis.StringCmd
+ }
+ cmds := make([]accountCmd, 0, len(chunk))
+ for _, candidate := range chunk {
+ slotKey := accountSlotKey(candidate.id)
+ waitKey := accountWaitKey(candidate.id)
+ pipe.ZRemRangeByScore(ctx, slotKey, "-inf", strconv.FormatInt(cutoffTime, 10))
+ cmds = append(cmds, accountCmd{
+ activeIndexLoad: candidate,
+ zcardCmd: pipe.ZCard(ctx, slotKey),
+ getCmd: pipe.Get(ctx, waitKey),
+ })
+ }
+ if _, err := pipe.Exec(ctx); err != nil && !errors.Is(err, redis.Nil) {
+ return nil, nil, fmt.Errorf("pipeline exec: %w", err)
+ }
+ for _, cmd := range cmds {
+ waitCount := 0
+ if v, err := cmd.getCmd.Int(); err == nil && v > 0 {
+ waitCount = v
+ }
+ loads = append(loads, activeIndexLoad{
+ id: cmd.id,
+ member: cmd.member,
+ slotCount: int(cmd.zcardCmd.Val()),
+ waitCount: waitCount,
+ })
+ }
+ }
+
+ return loads, staleMembers, nil
+}
+
+// removeActiveIndexMembers 清理无效 member；这是辅助索引的维护动作，调用方无需因为失败中断主流程。
+func (c *concurrencyCache) removeActiveIndexMembers(ctx context.Context, indexKey string, members []string) {
+ if len(members) == 0 {
+ return
+ }
+ args := make([]interface{}, 0, len(members))
+ for _, member := range members {
+ args = append(args, member)
+ }
+ _ = c.rdb.ZRem(ctx, indexKey, args...).Err()
+}
+
+// touchActiveIndexForLoad 根据已读取的真实负载刷新索引过期时间。
+func (c *concurrencyCache) touchActiveIndexForLoad(ctx context.Context, indexKey string, load activeIndexLoad) {
+ c.touchActiveIndex(ctx, indexKey, load.id, c.activeIndexTTL(load.slotCount, load.waitCount))
+}
+
// Account slot operations
func (c *concurrencyCache) AcquireAccountSlot(ctx context.Context, accountID int64, maxConcurrency int, requestID string) (bool, error) {
@@ -271,12 +446,21 @@ func (c *concurrencyCache) AcquireAccountSlot(ctx context.Context, accountID int
if err != nil {
return false, err
}
+ if result == 1 {
+ // 成功占槽后标记活跃账号，后台清理即可从索引定位候选账号。
+ c.touchAccountActiveIndex(ctx, accountID, c.slotTTLSeconds)
+ }
return result == 1, nil
}
func (c *concurrencyCache) ReleaseAccountSlot(ctx context.Context, accountID int64, requestID string) error {
key := accountSlotKey(accountID)
- return c.rdb.ZRem(ctx, key, requestID).Err()
+ if err := c.rdb.ZRem(ctx, key, requestID).Err(); err != nil {
+ return err
+ }
+ // 释放后用真实负载刷新索引；若没有槽位和等待计数，会移除索引 member。
+ c.refreshAccountActiveIndex(ctx, accountID)
+ return nil
}
func (c *concurrencyCache) GetAccountConcurrency(ctx context.Context, accountID int64) (int, error) {
@@ -335,12 +519,21 @@ func (c *concurrencyCache) AcquireUserSlot(ctx context.Context, userID int64, ma
if err != nil {
return false, err
}
+ if result == 1 {
+ // 成功占槽后标记活跃用户，避免启动清理依赖全量 SCAN。
+ c.touchUserActiveIndex(ctx, userID, c.slotTTLSeconds)
+ }
return result == 1, nil
}
func (c *concurrencyCache) ReleaseUserSlot(ctx context.Context, userID int64, requestID string) error {
key := userSlotKey(userID)
- return c.rdb.ZRem(ctx, key, requestID).Err()
+ if err := c.rdb.ZRem(ctx, key, requestID).Err(); err != nil {
+ return err
+ }
+ // 释放后按 Redis 中剩余负载修正索引状态。
+ c.refreshUserActiveIndex(ctx, userID)
+ return nil
}
func (c *concurrencyCache) GetUserConcurrency(ctx context.Context, userID int64) (int, error) {
@@ -361,12 +554,20 @@ func (c *concurrencyCache) IncrementWaitCount(ctx context.Context, userID int64,
if err != nil {
return false, err
}
+ if result == 1 {
+ // 等待队列也会让用户保持“活跃”，否则槽位为 0 时后台任务可能漏看等待计数。
+ c.touchUserActiveIndex(ctx, userID, c.waitQueueTTLSeconds)
+ }
return result == 1, nil
}
func (c *concurrencyCache) DecrementWaitCount(ctx context.Context, userID int64) error {
key := waitQueueKey(userID)
_, err := decrementWaitScript.Run(ctx, c.rdb, []string{key}).Result()
+ if err == nil {
+ // 等待数减少后重新判断是否还需要保留索引。
+ c.refreshUserActiveIndex(ctx, userID)
+ }
return err
}
@@ -378,12 +579,20 @@ func (c *concurrencyCache) IncrementAccountWaitCount(ctx context.Context, accoun
if err != nil {
return false, err
}
+ if result == 1 {
+ // 账号级等待队列同样写入账号活跃索引，供负载查询和清理任务使用。
+ c.touchAccountActiveIndex(ctx, accountID, c.waitQueueTTLSeconds)
+ }
return result == 1, nil
}
func (c *concurrencyCache) DecrementAccountWaitCount(ctx context.Context, accountID int64) error {
key := accountWaitKey(accountID)
_, err := decrementWaitScript.Run(ctx, c.rdb, []string{key}).Result()
+ if err == nil {
+ // 等待计数归零后索引需要同步删除，避免后台任务反复处理空账号。
+ c.refreshAccountActiveIndex(ctx, accountID)
+ }
return err
}
@@ -523,101 +732,151 @@ func (c *concurrencyCache) GetUsersLoadBatch(ctx context.Context, users []servic
func (c *concurrencyCache) CleanupExpiredAccountSlots(ctx context.Context, accountID int64) error {
key := accountSlotKey(accountID)
_, err := cleanupExpiredSlotsScript.Run(ctx, c.rdb, []string{key}, c.slotTTLSeconds).Result()
+ if err == nil {
+ // 单账号清理后同步索引，保持后台批量清理的候选集准确。
+ c.refreshAccountActiveIndex(ctx, accountID)
+ }
return err
}
-func (c *concurrencyCache) CleanupExpiredAccountSlotKeys(ctx context.Context) error {
- return c.cleanupExpiredSlotKeysByPattern(ctx, accountSlotKeyPrefix+"*")
+// GetActiveAccountLoadMap 只读取活跃账号索引中的账号负载。
+// 这是给热路径使用的轻量视图，避免为获取全局账号负载而扫描所有槽位键。
+func (c *concurrencyCache) GetActiveAccountLoadMap(ctx context.Context) (map[int64]*service.AccountLoadInfo, error) {
+ now, err := c.redisUnixSeconds(ctx)
+ if err != nil {
+ return nil, err
+ }
+ if err := c.rdb.ZRemRangeByScore(ctx, accountActiveIndexKey, "-inf", strconv.FormatInt(now, 10)).Err(); err != nil {
+ return nil, fmt.Errorf("cleanup account active index: %w", err)
+ }
+ members, err := c.rdb.ZRangeByScore(ctx, accountActiveIndexKey, &redis.ZRangeBy{
+ Min: strconv.FormatInt(now+1, 10),
+ Max: "+inf",
+ }).Result()
+ if err != nil {
+ return nil, fmt.Errorf("read account active index: %w", err)
+ }
+
+ loads, staleMembers, err := c.readAccountIndexLoads(ctx, members, now)
+ if err != nil {
+ return nil, err
+ }
+
+ loadMap := make(map[int64]*service.AccountLoadInfo, len(loads))
+ for _, load := range loads {
+ if load.slotCount == 0 && load.waitCount <= 0 {
+ // 索引候选已没有实际负载，删除 member 而不是返回空负载。
+ staleMembers = append(staleMembers, load.member)
+ continue
+ }
+ loadMap[load.id] = &service.AccountLoadInfo{
+ AccountID: load.id,
+ CurrentConcurrency: load.slotCount,
+ WaitingCount: load.waitCount,
+ }
+ c.touchActiveIndexForLoad(ctx, accountActiveIndexKey, load)
+ }
+ c.removeActiveIndexMembers(ctx, accountActiveIndexKey, staleMembers)
+ return loadMap, nil
}
+// CleanupExpiredAccountSlotKeys 只处理索引中过期的账号候选。
+// 若候选仍有真实负载，则刷新索引；若没有负载，则移除索引 member。
+func (c *concurrencyCache) CleanupExpiredAccountSlotKeys(ctx context.Context) error {
+ now, err := c.redisUnixSeconds(ctx)
+ if err != nil {
+ return err
+ }
+ members, err := c.rdb.ZRangeByScore(ctx, accountActiveIndexKey, &redis.ZRangeBy{
+ Min: "-inf",
+ Max: strconv.FormatInt(now, 10),
+ Count: activeIndexCleanupBatchSize,
+ }).Result()
+ if err != nil {
+ return fmt.Errorf("read expired account active index: %w", err)
+ }
+
+ loads, staleMembers, err := c.readAccountIndexLoads(ctx, members, now)
+ if err != nil {
+ return err
+ }
+ for _, load := range loads {
+ if load.slotCount == 0 && load.waitCount <= 0 {
+ // 真实槽位和等待数都为空，说明这个索引 member 已经完成使命。
+ staleMembers = append(staleMembers, load.member)
+ continue
+ }
+ c.touchActiveIndexForLoad(ctx, accountActiveIndexKey, load)
+ }
+ c.removeActiveIndexMembers(ctx, accountActiveIndexKey, staleMembers)
+ return nil
+}
+
+// CleanupStaleProcessSlots 启动时清理非当前进程前缀的槽位。
+// 清理范围来自活跃索引，避免在 Redis 上 SCAN 全部 concurrency:* 键。
func (c *concurrencyCache) CleanupStaleProcessSlots(ctx context.Context, activeRequestPrefix string) error {
if activeRequestPrefix == "" {
return nil
}
-
- // 1. 清理有序集合中非当前进程前缀的成员
- slotPatterns := []string{accountSlotKeyPrefix + "*", userSlotKeyPrefix + "*"}
- for _, pattern := range slotPatterns {
- if err := c.cleanupSlotsByPattern(ctx, pattern, activeRequestPrefix); err != nil {
- return err
- }
+ now, err := c.redisUnixSeconds(ctx)
+ if err != nil {
+ return err
}
- // 2. 删除所有等待队列计数器（重启后计数器失效）
- waitPatterns := []string{accountWaitKeyPrefix + "*", waitQueueKeyPrefix + "*"}
- for _, pattern := range waitPatterns {
- if err := c.deleteKeysByPattern(ctx, pattern); err != nil {
- return err
- }
+ accountMembers, err := c.activeIndexMembers(ctx, accountActiveIndexKey, now)
+ if err != nil {
+ return err
+ }
+ if err := c.cleanupStaleProcessSlotsForIndex(ctx, accountActiveIndexKey, accountMembers, activeRequestPrefix, accountSlotKey, accountWaitKey, c.refreshAccountActiveIndex); err != nil {
+ return err
}
- return nil
+ userMembers, err := c.activeIndexMembers(ctx, userActiveIndexKey, now)
+ if err != nil {
+ return err
+ }
+ return c.cleanupStaleProcessSlotsForIndex(ctx, userActiveIndexKey, userMembers, activeRequestPrefix, userSlotKey, waitQueueKey, c.refreshUserActiveIndex)
}
-// cleanupExpiredSlotKeysByPattern 扫描实际存在的账号槽位键并批量清理过期成员。
-func (c *concurrencyCache) cleanupExpiredSlotKeysByPattern(ctx context.Context, pattern string) error {
- const scanCount = 200
- var cursor uint64
- for {
- keys, nextCursor, err := c.rdb.Scan(ctx, cursor, pattern, scanCount).Result()
- if err != nil {
- return fmt.Errorf("scan %s: %w", pattern, err)
- }
- if len(keys) > 0 {
- _, err := cleanupExpiredSlotKeysScript.Run(ctx, c.rdb, keys, c.slotTTLSeconds).Result()
- if err != nil {
- return fmt.Errorf("cleanup expired slots %s: %w", pattern, err)
- }
- }
- cursor = nextCursor
- if cursor == 0 {
- break
- }
+// activeIndexMembers 只返回当前仍未过期的索引 member；过期 member 由对应清理任务处理。
+func (c *concurrencyCache) activeIndexMembers(ctx context.Context, indexKey string, now int64) ([]string, error) {
+ members, err := c.rdb.ZRangeByScore(ctx, indexKey, &redis.ZRangeBy{
+ Min: strconv.FormatInt(now+1, 10),
+ Max: "+inf",
+ }).Result()
+ if err != nil {
+ return nil, fmt.Errorf("read active index %s: %w", indexKey, err)
}
- return nil
+ return members, nil
}
-// cleanupSlotsByPattern 扫描匹配 pattern 的有序集合键，批量调用 Lua 脚本清理非当前进程成员。
-func (c *concurrencyCache) cleanupSlotsByPattern(ctx context.Context, pattern, activePrefix string) error {
- const scanCount = 200
- var cursor uint64
- for {
- keys, nextCursor, err := c.rdb.Scan(ctx, cursor, pattern, scanCount).Result()
- if err != nil {
- return fmt.Errorf("scan %s: %w", pattern, err)
+// cleanupStaleProcessSlotsForIndex 逐个处理索引中的账号/用户。
+// Lua 脚本一次只碰一个槽位 key，兼容 Redis Cluster，随后删除重启后已失效的等待计数。
+func (c *concurrencyCache) cleanupStaleProcessSlotsForIndex(
+ ctx context.Context,
+ indexKey string,
+ members []string,
+ activeRequestPrefix string,
+ slotKeyForID func(int64) string,
+ waitKeyForID func(int64) string,
+ refreshIndex func(context.Context, int64),
+) error {
+ staleMembers := make([]string, 0)
+ for _, member := range members {
+ id, err := strconv.ParseInt(member, 10, 64)
+ if err != nil || id <= 0 {
+ staleMembers = append(staleMembers, member)
+ continue
}
- if len(keys) > 0 {
- _, err := startupCleanupScript.Run(ctx, c.rdb, keys, activePrefix, c.slotTTLSeconds).Result()
- if err != nil {
- return fmt.Errorf("cleanup slots %s: %w", pattern, err)
- }
- }
- cursor = nextCursor
- if cursor == 0 {
- break
- }
- }
- return nil
-}
-// deleteKeysByPattern 扫描匹配 pattern 的键并删除。
-func (c *concurrencyCache) deleteKeysByPattern(ctx context.Context, pattern string) error {
- const scanCount = 200
- var cursor uint64
- for {
- keys, nextCursor, err := c.rdb.Scan(ctx, cursor, pattern, scanCount).Result()
- if err != nil {
- return fmt.Errorf("scan %s: %w", pattern, err)
+ if _, err := startupCleanupSlotScript.Run(ctx, c.rdb, []string{slotKeyForID(id)}, activeRequestPrefix, c.slotTTLSeconds).Result(); err != nil {
+ return fmt.Errorf("cleanup stale process slots %s: %w", slotKeyForID(id), err)
}
- if len(keys) > 0 {
- if err := c.rdb.Del(ctx, keys...).Err(); err != nil {
- return fmt.Errorf("del %s: %w", pattern, err)
- }
- }
- cursor = nextCursor
- if cursor == 0 {
- break
+ if err := c.rdb.Del(ctx, waitKeyForID(id)).Err(); err != nil {
+ return fmt.Errorf("delete stale wait key %s: %w", waitKeyForID(id), err)
}
+ refreshIndex(ctx, id)
}
+ c.removeActiveIndexMembers(ctx, indexKey, staleMembers)
return nil
}
diff --git a/backend/internal/repository/concurrency_cache_integration_test.go b/backend/internal/repository/concurrency_cache_integration_test.go
index 4aa172b252..aa18ba16cb 100644
--- a/backend/internal/repository/concurrency_cache_integration_test.go
+++ b/backend/internal/repository/concurrency_cache_integration_test.go
@@ -5,6 +5,7 @@ package repository
import (
"errors"
"fmt"
+ "strconv"
"testing"
"time"
@@ -22,7 +23,8 @@ var testSlotTTL = time.Duration(testSlotTTLMinutes) * time.Minute
type ConcurrencyCacheSuite struct {
IntegrationRedisSuite
- cache service.ConcurrencyCache
+ cache service.ConcurrencyCache
+ rawCache *concurrencyCache
}
func TestConcurrencyCacheSuite(t *testing.T) {
@@ -31,7 +33,8 @@ func TestConcurrencyCacheSuite(t *testing.T) {
func (s *ConcurrencyCacheSuite) SetupTest() {
s.IntegrationRedisSuite.SetupTest()
- s.cache = NewConcurrencyCache(s.rdb, testSlotTTLMinutes, int(testSlotTTL.Seconds()))
+ s.rawCache = NewConcurrencyCache(s.rdb, testSlotTTLMinutes, int(testSlotTTL.Seconds())).(*concurrencyCache)
+ s.cache = s.rawCache
}
func (s *ConcurrencyCacheSuite) TestAccountSlot_AcquireAndRelease() {
@@ -61,6 +64,61 @@ func (s *ConcurrencyCacheSuite) TestAccountSlot_AcquireAndRelease() {
require.Equal(s.T(), 1, cur, "expected 1 after release")
}
+func (s *ConcurrencyCacheSuite) TestActiveAccountLoadMap_AcquireAndRelease() {
+ accountID := int64(610)
+ reqID := "active-load-req"
+
+ ok, err := s.cache.AcquireAccountSlot(s.ctx, accountID, 2, reqID)
+ require.NoError(s.T(), err)
+ require.True(s.T(), ok)
+
+ loadMap, err := s.rawCache.GetActiveAccountLoadMap(s.ctx)
+ require.NoError(s.T(), err)
+ require.Contains(s.T(), loadMap, accountID)
+ require.Equal(s.T(), 1, loadMap[accountID].CurrentConcurrency)
+
+ require.NoError(s.T(), s.cache.ReleaseAccountSlot(s.ctx, accountID, reqID))
+
+ loadMap, err = s.rawCache.GetActiveAccountLoadMap(s.ctx)
+ require.NoError(s.T(), err)
+ require.NotContains(s.T(), loadMap, accountID)
+}
+
+func (s *ConcurrencyCacheSuite) TestActiveAccountLoadMap_AccountWaitIndexLifecycle() {
+ accountID := int64(611)
+
+ ok, err := s.cache.IncrementAccountWaitCount(s.ctx, accountID, 2)
+ require.NoError(s.T(), err)
+ require.True(s.T(), ok)
+
+ loadMap, err := s.rawCache.GetActiveAccountLoadMap(s.ctx)
+ require.NoError(s.T(), err)
+ require.Contains(s.T(), loadMap, accountID)
+ require.Equal(s.T(), 1, loadMap[accountID].WaitingCount)
+
+ require.NoError(s.T(), s.cache.DecrementAccountWaitCount(s.ctx, accountID))
+
+ loadMap, err = s.rawCache.GetActiveAccountLoadMap(s.ctx)
+ require.NoError(s.T(), err)
+ require.NotContains(s.T(), loadMap, accountID)
+}
+
+func (s *ConcurrencyCacheSuite) TestActiveAccountLoadMap_RemovesInvalidIndexMember() {
+ now, err := s.rawCache.redisUnixSeconds(s.ctx)
+ require.NoError(s.T(), err)
+ require.NoError(s.T(), s.rdb.ZAdd(s.ctx, accountActiveIndexKey, redis.Z{
+ Score: float64(now + 60),
+ Member: "not-an-account-id",
+ }).Err())
+
+ loadMap, err := s.rawCache.GetActiveAccountLoadMap(s.ctx)
+ require.NoError(s.T(), err)
+ require.Empty(s.T(), loadMap)
+
+ _, err = s.rdb.ZScore(s.ctx, accountActiveIndexKey, "not-an-account-id").Result()
+ require.ErrorIs(s.T(), err, redis.Nil)
+}
+
func (s *ConcurrencyCacheSuite) TestAccountSlot_TTL() {
accountID := int64(11)
reqID := "req_ttl_test"
@@ -254,12 +312,16 @@ func (s *ConcurrencyCacheSuite) TestAccountWaitQueue_IncrementAndDecrement() {
func (s *ConcurrencyCacheSuite) TestCleanupStaleProcessSlots() {
accountID := int64(901)
userID := int64(902)
+ unindexedAccountID := int64(1901)
accountKey := fmt.Sprintf("%s%d", accountSlotKeyPrefix, accountID)
userKey := fmt.Sprintf("%s%d", userSlotKeyPrefix, userID)
+ unindexedAccountKey := fmt.Sprintf("%s%d", accountSlotKeyPrefix, unindexedAccountID)
userWaitKey := fmt.Sprintf("%s%d", waitQueueKeyPrefix, userID)
accountWaitKey := fmt.Sprintf("%s%d", accountWaitKeyPrefix, accountID)
+ unindexedAccountWaitKey := fmt.Sprintf("%s%d", accountWaitKeyPrefix, unindexedAccountID)
- now := time.Now().Unix()
+ now, err := s.rawCache.redisUnixSeconds(s.ctx)
+ require.NoError(s.T(), err)
require.NoError(s.T(), s.rdb.ZAdd(s.ctx, accountKey,
redis.Z{Score: float64(now), Member: "oldproc-1"},
redis.Z{Score: float64(now), Member: "keep-1"},
@@ -268,8 +330,20 @@ func (s *ConcurrencyCacheSuite) TestCleanupStaleProcessSlots() {
redis.Z{Score: float64(now), Member: "oldproc-2"},
redis.Z{Score: float64(now), Member: "keep-2"},
).Err())
+ require.NoError(s.T(), s.rdb.ZAdd(s.ctx, unindexedAccountKey,
+ redis.Z{Score: float64(now), Member: "oldproc-unindexed"},
+ ).Err())
require.NoError(s.T(), s.rdb.Set(s.ctx, userWaitKey, 3, time.Minute).Err())
require.NoError(s.T(), s.rdb.Set(s.ctx, accountWaitKey, 2, time.Minute).Err())
+ require.NoError(s.T(), s.rdb.Set(s.ctx, unindexedAccountWaitKey, 2, time.Minute).Err())
+ require.NoError(s.T(), s.rdb.ZAdd(s.ctx, accountActiveIndexKey, redis.Z{
+ Score: float64(now + 60),
+ Member: strconv.FormatInt(accountID, 10),
+ }).Err())
+ require.NoError(s.T(), s.rdb.ZAdd(s.ctx, userActiveIndexKey, redis.Z{
+ Score: float64(now + 60),
+ Member: strconv.FormatInt(userID, 10),
+ }).Err())
require.NoError(s.T(), s.cache.CleanupStaleProcessSlots(s.ctx, "keep-"))
@@ -286,6 +360,12 @@ func (s *ConcurrencyCacheSuite) TestCleanupStaleProcessSlots() {
_, err = s.rdb.Get(s.ctx, accountWaitKey).Result()
require.True(s.T(), errors.Is(err, redis.Nil))
+
+ unindexedMembers, err := s.rdb.ZRange(s.ctx, unindexedAccountKey, 0, -1).Result()
+ require.NoError(s.T(), err)
+ require.Equal(s.T(), []string{"oldproc-unindexed"}, unindexedMembers)
+ _, err = s.rdb.Get(s.ctx, unindexedAccountWaitKey).Result()
+ require.NoError(s.T(), err)
}
func (s *ConcurrencyCacheSuite) TestGetAccountConcurrency_Missing() {
@@ -436,11 +516,13 @@ func (s *ConcurrencyCacheSuite) TestCleanupExpiredAccountSlots_NoExpired() {
}
func (s *ConcurrencyCacheSuite) TestCleanupExpiredAccountSlotKeys() {
- now := time.Now().Unix()
+ now, err := s.rawCache.redisUnixSeconds(s.ctx)
+ require.NoError(s.T(), err)
expiredTime := now - int64(testSlotTTL.Seconds()) - 10
accountKeyWithFresh := fmt.Sprintf("%s%d", accountSlotKeyPrefix, 301)
accountKeyExpiredOnly := fmt.Sprintf("%s%d", accountSlotKeyPrefix, 302)
userKey := fmt.Sprintf("%s%d", userSlotKeyPrefix, 303)
+ unindexedAccountKey := fmt.Sprintf("%s%d", accountSlotKeyPrefix, 304)
require.NoError(s.T(), s.rdb.ZAdd(s.ctx, accountKeyWithFresh,
redis.Z{Score: float64(expiredTime), Member: "expired"},
@@ -452,6 +534,13 @@ func (s *ConcurrencyCacheSuite) TestCleanupExpiredAccountSlotKeys() {
require.NoError(s.T(), s.rdb.ZAdd(s.ctx, userKey,
redis.Z{Score: float64(expiredTime), Member: "user-expired"},
).Err())
+ require.NoError(s.T(), s.rdb.ZAdd(s.ctx, unindexedAccountKey,
+ redis.Z{Score: float64(expiredTime), Member: "unindexed-expired"},
+ ).Err())
+ require.NoError(s.T(), s.rdb.ZAdd(s.ctx, accountActiveIndexKey,
+ redis.Z{Score: float64(now), Member: "301"},
+ redis.Z{Score: float64(now), Member: "302"},
+ ).Err())
require.NoError(s.T(), s.cache.CleanupExpiredAccountSlotKeys(s.ctx))
@@ -466,6 +555,16 @@ func (s *ConcurrencyCacheSuite) TestCleanupExpiredAccountSlotKeys() {
userMembers, err := s.rdb.ZRange(s.ctx, userKey, 0, -1).Result()
require.NoError(s.T(), err)
require.Equal(s.T(), []string{"user-expired"}, userMembers)
+
+ unindexedMembers, err := s.rdb.ZRange(s.ctx, unindexedAccountKey, 0, -1).Result()
+ require.NoError(s.T(), err)
+ require.Equal(s.T(), []string{"unindexed-expired"}, unindexedMembers)
+
+ score, err := s.rdb.ZScore(s.ctx, accountActiveIndexKey, "301").Result()
+ require.NoError(s.T(), err)
+ require.Greater(s.T(), int64(score), now)
+ _, err = s.rdb.ZScore(s.ctx, accountActiveIndexKey, "302").Result()
+ require.ErrorIs(s.T(), err, redis.Nil)
}
func (s *ConcurrencyCacheSuite) TestCleanupStaleProcessSlots_RemovesOldPrefixesAndWaitCounters() {
@@ -476,19 +575,28 @@ func (s *ConcurrencyCacheSuite) TestCleanupStaleProcessSlots_RemovesOldPrefixesA
userWaitKey := fmt.Sprintf("%s%d", waitQueueKeyPrefix, userID)
accountWaitKey := fmt.Sprintf("%s%d", accountWaitKeyPrefix, accountID)
- now := float64(time.Now().Unix())
+ now, err := s.rawCache.redisUnixSeconds(s.ctx)
+ require.NoError(s.T(), err)
require.NoError(s.T(), s.rdb.ZAdd(s.ctx, accountSlotKey,
- redis.Z{Score: now, Member: "oldproc-1"},
- redis.Z{Score: now, Member: "activeproc-1"},
+ redis.Z{Score: float64(now), Member: "oldproc-1"},
+ redis.Z{Score: float64(now), Member: "activeproc-1"},
).Err())
require.NoError(s.T(), s.rdb.Expire(s.ctx, accountSlotKey, testSlotTTL).Err())
require.NoError(s.T(), s.rdb.ZAdd(s.ctx, userSlotKey,
- redis.Z{Score: now, Member: "oldproc-2"},
- redis.Z{Score: now, Member: "activeproc-2"},
+ redis.Z{Score: float64(now), Member: "oldproc-2"},
+ redis.Z{Score: float64(now), Member: "activeproc-2"},
).Err())
require.NoError(s.T(), s.rdb.Expire(s.ctx, userSlotKey, testSlotTTL).Err())
require.NoError(s.T(), s.rdb.Set(s.ctx, userWaitKey, 3, testSlotTTL).Err())
require.NoError(s.T(), s.rdb.Set(s.ctx, accountWaitKey, 2, testSlotTTL).Err())
+ require.NoError(s.T(), s.rdb.ZAdd(s.ctx, accountActiveIndexKey, redis.Z{
+ Score: float64(now + 60),
+ Member: strconv.FormatInt(accountID, 10),
+ }).Err())
+ require.NoError(s.T(), s.rdb.ZAdd(s.ctx, userActiveIndexKey, redis.Z{
+ Score: float64(now + 60),
+ Member: strconv.FormatInt(userID, 10),
+ }).Err())
require.NoError(s.T(), s.cache.CleanupStaleProcessSlots(s.ctx, "activeproc-"))
@@ -509,8 +617,14 @@ func (s *ConcurrencyCacheSuite) TestCleanupStaleProcessSlots_RemovesOldPrefixesA
func (s *ConcurrencyCacheSuite) TestCleanupStaleProcessSlots_DeletesEmptySlotKeys() {
accountID := int64(903)
accountSlotKey := fmt.Sprintf("%s%d", accountSlotKeyPrefix, accountID)
- require.NoError(s.T(), s.rdb.ZAdd(s.ctx, accountSlotKey, redis.Z{Score: float64(time.Now().Unix()), Member: "oldproc-1"}).Err())
+ now, err := s.rawCache.redisUnixSeconds(s.ctx)
+ require.NoError(s.T(), err)
+ require.NoError(s.T(), s.rdb.ZAdd(s.ctx, accountSlotKey, redis.Z{Score: float64(now), Member: "oldproc-1"}).Err())
require.NoError(s.T(), s.rdb.Expire(s.ctx, accountSlotKey, testSlotTTL).Err())
+ require.NoError(s.T(), s.rdb.ZAdd(s.ctx, accountActiveIndexKey, redis.Z{
+ Score: float64(now + 60),
+ Member: strconv.FormatInt(accountID, 10),
+ }).Err())
require.NoError(s.T(), s.cache.CleanupStaleProcessSlots(s.ctx, "activeproc-"))
diff --git a/backend/internal/repository/user_msg_queue_cache.go b/backend/internal/repository/user_msg_queue_cache.go
index 2b5b96bdc3..ca52d6eb91 100644
--- a/backend/internal/repository/user_msg_queue_cache.go
+++ b/backend/internal/repository/user_msg_queue_cache.go
@@ -5,8 +5,6 @@ import (
"errors"
"fmt"
"strconv"
- "strings"
- "time"
"github.com/Wei-Shaw/sub2api/internal/service"
"github.com/redis/go-redis/v9"
@@ -18,18 +16,30 @@ const (
umqKeyPrefix = "umq:"
umqLockSuffix = ":lock" // STRING (requestID), PX lockTtlMs
umqLastSuffix = ":last" // STRING (毫秒时间戳), EX 60s
+
+ // 锁索引用来替代后台清理对 umq:*:lock 的全量 SCAN。
+ // member 是 accountID，score 是锁预计过期的 Redis Unix 毫秒时间戳。
+ umqLockIndexKey = "umq:lock:index" // ZSET member=accountID, score=lockExpireAtUnixMs
+ umqLockIndexCleanupBatchSize = 1000
)
// Lua 脚本：原子获取串行锁（SET NX PX + 重入安全）
+// 返回 {是否获取成功, 锁预计过期时间毫秒}，让 Go 侧用同一 Redis 时间源更新索引。
var acquireLockScript = redis.NewScript(`
+redis.replicate_commands()
local cur = redis.call('GET', KEYS[1])
+local ttl = tonumber(ARGV[2])
if cur == ARGV[1] then
- redis.call('PEXPIRE', KEYS[1], tonumber(ARGV[2]))
- return 1
+ redis.call('PEXPIRE', KEYS[1], ttl)
+ local t = redis.call('TIME')
+ local ms = tonumber(t[1])*1000 + math.floor(tonumber(t[2])/1000)
+ return {1, ms + ttl}
end
-if cur ~= false then return 0 end
-redis.call('SET', KEYS[1], ARGV[1], 'PX', tonumber(ARGV[2]))
-return 1
+if cur ~= false then return {0, 0} end
+redis.call('SET', KEYS[1], ARGV[1], 'PX', ttl)
+local t = redis.call('TIME')
+local ms = tonumber(t[1])*1000 + math.floor(tonumber(t[2])/1000)
+return {1, ms + ttl}
`)
// Lua 脚本：原子释放锁 + 记录完成时间（使用 Redis TIME 避免时钟偏差）
@@ -48,14 +58,18 @@ end
return 0
`)
-// Lua 脚本：原子清理孤儿锁（仅在 PTTL == -1 时删除，避免 TOCTOU 竞态误删合法锁）
-var forceReleaseLockScript = redis.NewScript(`
+// Lua 脚本：校验锁 TTL 状态，PTTL == -1 时原子删除异常锁。
+// 返回状态: -2=锁不存在，-1=无 TTL 的异常锁已删除，1=锁仍存活并返回剩余 PTTL。
+var reconcileLockScript = redis.NewScript(`
local pttl = redis.call('PTTL', KEYS[1])
+if pttl == -2 then
+ return {-2, 0}
+end
if pttl == -1 then
redis.call('DEL', KEYS[1])
- return 1
+ return {-1, 0}
end
-return 0
+return {1, pttl}
`)
type userMsgQueueCache struct {
@@ -77,22 +91,33 @@ func umqLastKey(accountID int64) string {
return umqKeyPrefix + "{" + strconv.FormatInt(accountID, 10) + "}" + umqLastSuffix
}
-// umqScanPattern 用于 SCAN 扫描锁 key
-func umqScanPattern() string {
- return umqKeyPrefix + "{*}" + umqLockSuffix
-}
-
// AcquireLock 尝试获取账号级串行锁
+// 成功后尽力写入锁索引，后台清理只需要看“到期候选”而不是扫描所有锁 key。
func (c *userMsgQueueCache) AcquireLock(ctx context.Context, accountID int64, requestID string, lockTtlMs int) (bool, error) {
key := umqLockKey(accountID)
- result, err := acquireLockScript.Run(ctx, c.rdb, []string{key}, requestID, lockTtlMs).Int()
+ result, err := acquireLockScript.Run(ctx, c.rdb, []string{key}, requestID, lockTtlMs).Result()
if err != nil {
return false, fmt.Errorf("umq acquire lock: %w", err)
}
- return result == 1, nil
+ acquired, err := redisScriptInt64At(result, 0)
+ if err != nil {
+ return false, fmt.Errorf("umq parse acquire lock result: %w", err)
+ }
+ expireAtMs, err := redisScriptInt64At(result, 1)
+ if err != nil {
+ return false, fmt.Errorf("umq parse acquire lock expire: %w", err)
+ }
+ if acquired == 1 {
+ _ = c.rdb.ZAdd(ctx, umqLockIndexKey, redis.Z{
+ Score: float64(expireAtMs),
+ Member: strconv.FormatInt(accountID, 10),
+ }).Err()
+ }
+ return acquired == 1, nil
}
// ReleaseLock 释放锁并记录完成时间
+// 只有 requestID 匹配时才删除锁索引，避免误删其他请求重入后写入的新锁。
func (c *userMsgQueueCache) ReleaseLock(ctx context.Context, accountID int64, requestID string) (bool, error) {
lockKey := umqLockKey(accountID)
lastKey := umqLastKey(accountID)
@@ -100,6 +125,9 @@ func (c *userMsgQueueCache) ReleaseLock(ctx context.Context, accountID int64, re
if err != nil {
return false, fmt.Errorf("umq release lock: %w", err)
}
+ if result == 1 {
+ _ = c.rdb.ZRem(ctx, umqLockIndexKey, strconv.FormatInt(accountID, 10)).Err()
+ }
return result == 1, nil
}
@@ -120,65 +148,6 @@ func (c *userMsgQueueCache) GetLastCompletedMs(ctx context.Context, accountID in
return ms, nil
}
-// ForceReleaseLock 原子清理孤儿锁（仅在 PTTL == -1 时删除，防止 TOCTOU 竞态误删合法锁）
-func (c *userMsgQueueCache) ForceReleaseLock(ctx context.Context, accountID int64) error {
- key := umqLockKey(accountID)
- _, err := forceReleaseLockScript.Run(ctx, c.rdb, []string{key}).Result()
- if err != nil && !errors.Is(err, redis.Nil) {
- return fmt.Errorf("umq force release lock: %w", err)
- }
- return nil
-}
-
-// ScanLockKeys 扫描所有锁 key，仅返回 PTTL == -1（无过期时间）的孤儿锁 accountID 列表
-// 正常的锁都有 PX 过期时间，PTTL == -1 表示异常状态（如 Redis 故障恢复后丢失 TTL）
-func (c *userMsgQueueCache) ScanLockKeys(ctx context.Context, maxCount int) ([]int64, error) {
- var accountIDs []int64
- var cursor uint64
- pattern := umqScanPattern()
-
- for {
- keys, nextCursor, err := c.rdb.Scan(ctx, cursor, pattern, 100).Result()
- if err != nil {
- return nil, fmt.Errorf("umq scan lock keys: %w", err)
- }
- for _, key := range keys {
- // 检查 PTTL：只清理 PTTL == -1（无过期时间）的异常锁
- pttl, err := c.rdb.PTTL(ctx, key).Result()
- if err != nil {
- continue
- }
- // PTTL 返回值：-2 = key 不存在，-1 = 无过期时间，>0 = 剩余毫秒
- // go-redis 对哨兵值 -1/-2 不乘精度系数，直接返回 time.Duration(-1)/-2
- // 只删除 -1（无过期时间的异常锁），跳过正常持有的锁
- if pttl != time.Duration(-1) {
- continue
- }
-
- // 从 key 中提取 accountID: umq:{123}:lock → 提取 {} 内的数字
- openBrace := strings.IndexByte(key, '{')
- closeBrace := strings.IndexByte(key, '}')
- if openBrace < 0 || closeBrace <= openBrace+1 {
- continue
- }
- idStr := key[openBrace+1 : closeBrace]
- id, err := strconv.ParseInt(idStr, 10, 64)
- if err != nil {
- continue
- }
- accountIDs = append(accountIDs, id)
- if len(accountIDs) >= maxCount {
- return accountIDs, nil
- }
- }
- cursor = nextCursor
- if cursor == 0 {
- break
- }
- }
- return accountIDs, nil
-}
-
// GetCurrentTimeMs 通过 Redis TIME 命令获取当前服务器时间（毫秒），确保与锁记录的时间源一致
func (c *userMsgQueueCache) GetCurrentTimeMs(ctx context.Context) (int64, error) {
t, err := c.rdb.Time(ctx).Result()
@@ -187,3 +156,85 @@ func (c *userMsgQueueCache) GetCurrentTimeMs(ctx context.Context) (int64, error)
}
return t.UnixMilli(), nil
}
+
+// ReconcileExpiredLockCandidates 只处理索引里已经到期的候选锁。
+// 候选到期不等于锁一定失效：可能是续租后索引滞后，所以必须再用 PTTL 二次确认。
+func (c *userMsgQueueCache) ReconcileExpiredLockCandidates(ctx context.Context, maxCount int) (int, error) {
+ if maxCount <= 0 {
+ maxCount = umqLockIndexCleanupBatchSize
+ }
+ nowMs, err := c.GetCurrentTimeMs(ctx)
+ if err != nil {
+ return 0, err
+ }
+ members, err := c.rdb.ZRangeByScore(ctx, umqLockIndexKey, &redis.ZRangeBy{
+ Min: "-inf",
+ Max: strconv.FormatInt(nowMs, 10),
+ Count: int64(maxCount),
+ }).Result()
+ if err != nil {
+ return 0, fmt.Errorf("umq read lock index: %w", err)
+ }
+
+ cleaned := 0
+ for _, member := range members {
+ accountID, err := strconv.ParseInt(member, 10, 64)
+ if err != nil || accountID <= 0 {
+ _ = c.rdb.ZRem(ctx, umqLockIndexKey, member).Err()
+ continue
+ }
+
+ result, err := reconcileLockScript.Run(ctx, c.rdb, []string{umqLockKey(accountID)}).Result()
+ if err != nil && !errors.Is(err, redis.Nil) {
+ return cleaned, fmt.Errorf("umq reconcile lock: %w", err)
+ }
+ status, err := redisScriptInt64At(result, 0)
+ if err != nil {
+ return cleaned, fmt.Errorf("umq parse reconcile status: %w", err)
+ }
+ pttl, err := redisScriptInt64At(result, 1)
+ if err != nil {
+ return cleaned, fmt.Errorf("umq parse reconcile pttl: %w", err)
+ }
+
+ switch status {
+ case -2:
+ // 锁自然过期或已释放，只需移除索引残留。
+ _ = c.rdb.ZRem(ctx, umqLockIndexKey, member).Err()
+ case -1:
+ // 无 TTL 的锁会永久阻塞队列，Lua 已原子删除它，这里统计一次清理。
+ _ = c.rdb.ZRem(ctx, umqLockIndexKey, member).Err()
+ cleaned++
+ case 1:
+ // 锁仍存活，说明索引过期时间滞后；按剩余 PTTL 重新排期。
+ _ = c.rdb.ZAdd(ctx, umqLockIndexKey, redis.Z{
+ Score: float64(nowMs + pttl),
+ Member: member,
+ }).Err()
+ }
+ }
+ return cleaned, nil
+}
+
+// redisScriptInt64At 兼容 go-redis 对 Lua 数组元素的不同返回类型。
+func redisScriptInt64At(result interface{}, index int) (int64, error) {
+ values, ok := result.([]interface{})
+ if !ok {
+ return 0, fmt.Errorf("expected redis script array, got %T", result)
+ }
+ if index < 0 || index >= len(values) {
+ return 0, fmt.Errorf("redis script array missing index %d", index)
+ }
+ switch v := values[index].(type) {
+ case int64:
+ return v, nil
+ case int:
+ return int64(v), nil
+ case string:
+ return strconv.ParseInt(v, 10, 64)
+ case []byte:
+ return strconv.ParseInt(string(v), 10, 64)
+ default:
+ return 0, fmt.Errorf("unexpected redis script value %T", v)
+ }
+}
diff --git a/backend/internal/repository/user_msg_queue_cache_integration_test.go b/backend/internal/repository/user_msg_queue_cache_integration_test.go
new file mode 100644
index 0000000000..c61b658357
--- /dev/null
+++ b/backend/internal/repository/user_msg_queue_cache_integration_test.go
@@ -0,0 +1,128 @@
+//go:build integration
+
+package repository
+
+import (
+ "errors"
+ "testing"
+ "time"
+
+ "github.com/redis/go-redis/v9"
+ "github.com/stretchr/testify/require"
+ "github.com/stretchr/testify/suite"
+)
+
+type UserMsgQueueCacheSuite struct {
+ IntegrationRedisSuite
+ cache *userMsgQueueCache
+}
+
+func TestUserMsgQueueCacheSuite(t *testing.T) {
+ suite.Run(t, new(UserMsgQueueCacheSuite))
+}
+
+func (s *UserMsgQueueCacheSuite) SetupTest() {
+ s.IntegrationRedisSuite.SetupTest()
+ s.cache = NewUserMsgQueueCache(s.rdb).(*userMsgQueueCache)
+}
+
+func (s *UserMsgQueueCacheSuite) TestAcquireLockWritesIndexAndReleaseRemovesIt() {
+ accountID := int64(701)
+ nowMs, err := s.cache.GetCurrentTimeMs(s.ctx)
+ require.NoError(s.T(), err)
+
+ acquired, err := s.cache.AcquireLock(s.ctx, accountID, "req-701", 10_000)
+ require.NoError(s.T(), err)
+ require.True(s.T(), acquired)
+
+ score, err := s.rdb.ZScore(s.ctx, umqLockIndexKey, "701").Result()
+ require.NoError(s.T(), err)
+ require.Greater(s.T(), int64(score), nowMs)
+
+ released, err := s.cache.ReleaseLock(s.ctx, accountID, "req-701")
+ require.NoError(s.T(), err)
+ require.True(s.T(), released)
+
+ _, err = s.rdb.ZScore(s.ctx, umqLockIndexKey, "701").Result()
+ require.ErrorIs(s.T(), err, redis.Nil)
+}
+
+func (s *UserMsgQueueCacheSuite) TestReconcileExpiredLockCandidatesRemovesNaturallyExpiredLockIndex() {
+ accountID := int64(702)
+ acquired, err := s.cache.AcquireLock(s.ctx, accountID, "req-702", 20)
+ require.NoError(s.T(), err)
+ require.True(s.T(), acquired)
+
+ score, err := s.rdb.ZScore(s.ctx, umqLockIndexKey, "702").Result()
+ require.NoError(s.T(), err)
+ require.Eventually(s.T(), func() bool {
+ nowMs, err := s.cache.GetCurrentTimeMs(s.ctx)
+ return err == nil && nowMs >= int64(score)
+ }, time.Second, 10*time.Millisecond)
+
+ cleaned, err := s.cache.ReconcileExpiredLockCandidates(s.ctx, 1000)
+ require.NoError(s.T(), err)
+ require.Equal(s.T(), 0, cleaned)
+
+ _, err = s.rdb.ZScore(s.ctx, umqLockIndexKey, "702").Result()
+ require.ErrorIs(s.T(), err, redis.Nil)
+}
+
+func (s *UserMsgQueueCacheSuite) TestReconcileExpiredLockCandidatesRefreshesLiveLockIndex() {
+ accountID := int64(703)
+ nowMs, err := s.cache.GetCurrentTimeMs(s.ctx)
+ require.NoError(s.T(), err)
+ require.NoError(s.T(), s.rdb.Set(s.ctx, umqLockKey(accountID), "req-703", time.Minute).Err())
+ require.NoError(s.T(), s.rdb.ZAdd(s.ctx, umqLockIndexKey, redis.Z{
+ Score: float64(nowMs - 1),
+ Member: "703",
+ }).Err())
+
+ cleaned, err := s.cache.ReconcileExpiredLockCandidates(s.ctx, 1000)
+ require.NoError(s.T(), err)
+ require.Equal(s.T(), 0, cleaned)
+
+ score, err := s.rdb.ZScore(s.ctx, umqLockIndexKey, "703").Result()
+ require.NoError(s.T(), err)
+ require.Greater(s.T(), int64(score), nowMs)
+ exists, err := s.rdb.Exists(s.ctx, umqLockKey(accountID)).Result()
+ require.NoError(s.T(), err)
+ require.EqualValues(s.T(), 1, exists)
+}
+
+func (s *UserMsgQueueCacheSuite) TestReconcileExpiredLockCandidatesDeletesNoTTLLock() {
+ accountID := int64(704)
+ nowMs, err := s.cache.GetCurrentTimeMs(s.ctx)
+ require.NoError(s.T(), err)
+ require.NoError(s.T(), s.rdb.Set(s.ctx, umqLockKey(accountID), "req-704", 0).Err())
+ require.NoError(s.T(), s.rdb.ZAdd(s.ctx, umqLockIndexKey, redis.Z{
+ Score: float64(nowMs),
+ Member: "704",
+ }).Err())
+
+ cleaned, err := s.cache.ReconcileExpiredLockCandidates(s.ctx, 1000)
+ require.NoError(s.T(), err)
+ require.Equal(s.T(), 1, cleaned)
+
+ exists, err := s.rdb.Exists(s.ctx, umqLockKey(accountID)).Result()
+ require.NoError(s.T(), err)
+ require.EqualValues(s.T(), 0, exists)
+ _, err = s.rdb.ZScore(s.ctx, umqLockIndexKey, "704").Result()
+ require.ErrorIs(s.T(), err, redis.Nil)
+}
+
+func (s *UserMsgQueueCacheSuite) TestReconcileExpiredLockCandidatesRemovesInvalidMember() {
+ nowMs, err := s.cache.GetCurrentTimeMs(s.ctx)
+ require.NoError(s.T(), err)
+ require.NoError(s.T(), s.rdb.ZAdd(s.ctx, umqLockIndexKey, redis.Z{
+ Score: float64(nowMs),
+ Member: "not-an-account-id",
+ }).Err())
+
+ cleaned, err := s.cache.ReconcileExpiredLockCandidates(s.ctx, 1000)
+ require.NoError(s.T(), err)
+ require.Equal(s.T(), 0, cleaned)
+
+ _, err = s.rdb.ZScore(s.ctx, umqLockIndexKey, "not-an-account-id").Result()
+ require.True(s.T(), errors.Is(err, redis.Nil))
+}
diff --git a/backend/internal/service/concurrency_service.go b/backend/internal/service/concurrency_service.go
index 50e0077b94..10f47f5b74 100644
--- a/backend/internal/service/concurrency_service.go
+++ b/backend/internal/service/concurrency_service.go
@@ -37,7 +37,7 @@ type ConcurrencyCache interface {
ReleaseUserSlot(ctx context.Context, userID int64, requestID string) error
GetUserConcurrency(ctx context.Context, userID int64) (int, error)
- // 等待队列计数（只在首次创建时设置 TTL）
+ // 等待队列计数（每次入队都会刷新 TTL，避免长时间排队时计数提前过期）
IncrementWaitCount(ctx context.Context, userID int64, maxWait int) (bool, error)
DecrementWaitCount(ctx context.Context, userID int64) error
diff --git a/backend/internal/service/user_msg_queue_service.go b/backend/internal/service/user_msg_queue_service.go
index f3f105ac93..72027d387a 100644
--- a/backend/internal/service/user_msg_queue_service.go
+++ b/backend/internal/service/user_msg_queue_service.go
@@ -25,10 +25,8 @@ type UserMsgQueueCache interface {
GetLastCompletedMs(ctx context.Context, accountID int64) (int64, error)
// GetCurrentTimeMs 获取 Redis 服务器当前时间（毫秒），与 ReleaseLock 记录的时间源一致
GetCurrentTimeMs(ctx context.Context) (int64, error)
- // ForceReleaseLock 强制释放锁（孤儿锁清理）
- ForceReleaseLock(ctx context.Context, accountID int64) error
- // ScanLockKeys 扫描 PTTL == -1 的孤儿锁 key，返回 accountID 列表
- ScanLockKeys(ctx context.Context, maxCount int) ([]int64, error)
+ // ReconcileExpiredLockCandidates 处理锁索引中的到期候选，按真实 PTTL 清理或刷新索引
+ ReconcileExpiredLockCandidates(ctx context.Context, maxCount int) (cleaned int, err error)
}
// QueueLockResult 锁获取结果
@@ -246,8 +244,8 @@ func (s *UserMessageQueueService) CalculateRPMAwareDelay(ctx context.Context, ac
return applyJitter(baseDelay, 0.15)
}
-// StartCleanupWorker 启动孤儿锁清理 worker
-// 定期 SCAN umq:*:lock 并清理 PTTL == -1 的异常锁（PTTL 检查在 cache.ScanLockKeys 内完成）
+// StartCleanupWorker 启动孤儿锁清理 worker。
+// worker 只处理锁索引中的到期候选，真正删除前由 cache 层再次校验锁 PTTL。
func (s *UserMessageQueueService) StartCleanupWorker(interval time.Duration) {
if s == nil || s.cache == nil || interval <= 0 {
return
@@ -257,23 +255,13 @@ func (s *UserMessageQueueService) StartCleanupWorker(interval time.Duration) {
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
- accountIDs, err := s.cache.ScanLockKeys(ctx, 1000)
+ // 每轮限制处理数量，避免清理任务在大量过期候选时长时间占用 Redis。
+ cleaned, err := s.cache.ReconcileExpiredLockCandidates(ctx, 1000)
if err != nil {
- logger.LegacyPrintf("service.umq", "Cleanup scan failed: %v", err)
+ logger.LegacyPrintf("service.umq", "Cleanup reconcile failed: %v", err)
return
}
- cleaned := 0
- for _, accountID := range accountIDs {
- cleanCtx, cleanCancel := context.WithTimeout(context.Background(), 2*time.Second)
- if err := s.cache.ForceReleaseLock(cleanCtx, accountID); err != nil {
- logger.LegacyPrintf("service.umq", "Cleanup force release failed for account %d: %v", accountID, err)
- } else {
- cleaned++
- }
- cleanCancel()
- }
-
if cleaned > 0 {
logger.LegacyPrintf("service.umq", "Cleanup completed: released %d orphaned locks", cleaned)
}
diff --git a/backend/internal/service/user_msg_queue_service_test.go b/backend/internal/service/user_msg_queue_service_test.go
new file mode 100644
index 0000000000..f554269b88
--- /dev/null
+++ b/backend/internal/service/user_msg_queue_service_test.go
@@ -0,0 +1,54 @@
+//go:build unit
+
+package service
+
+import (
+ "context"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/require"
+)
+
+type cleanupWorkerUserMsgQueueCache struct {
+ reconcileCalls atomic.Int64
+ maxCount atomic.Int64
+}
+
+var _ UserMsgQueueCache = (*cleanupWorkerUserMsgQueueCache)(nil)
+
+func (c *cleanupWorkerUserMsgQueueCache) AcquireLock(context.Context, int64, string, int) (bool, error) {
+ return true, nil
+}
+
+func (c *cleanupWorkerUserMsgQueueCache) ReleaseLock(context.Context, int64, string) (bool, error) {
+ return true, nil
+}
+
+func (c *cleanupWorkerUserMsgQueueCache) GetLastCompletedMs(context.Context, int64) (int64, error) {
+ return 0, nil
+}
+
+func (c *cleanupWorkerUserMsgQueueCache) GetCurrentTimeMs(context.Context) (int64, error) {
+ return time.Now().UnixMilli(), nil
+}
+
+func (c *cleanupWorkerUserMsgQueueCache) ReconcileExpiredLockCandidates(_ context.Context, maxCount int) (int, error) {
+ c.reconcileCalls.Add(1)
+ c.maxCount.Store(int64(maxCount))
+ return 1, nil
+}
+
+func TestStartCleanupWorker_ReconcilesExpiredLockCandidates(t *testing.T) {
+ cache := &cleanupWorkerUserMsgQueueCache{}
+ svc := NewUserMessageQueueService(cache, nil, nil)
+ defer svc.Stop()
+
+ svc.StartCleanupWorker(time.Millisecond)
+
+ require.Eventually(t, func() bool {
+ return cache.reconcileCalls.Load() > 0
+ }, time.Second, 10*time.Millisecond)
+ require.EqualValues(t, 1000, cache.maxCount.Load())
+}
From cb151e36ebc73bf366aad3250baf0ac30d3eb05b Mon Sep 17 00:00:00 2001
From: cat
Date: Mon, 6 Jul 2026 20:45:00 +0800
Subject: [PATCH 12/19] fix: respect custom User-Agent in OAuth account test
Match the image OAuth test path and gateway behavior: check for a
custom user_agent credential first, fall back to codexCLIUserAgent
only when no custom UA is configured.
---
backend/internal/service/account_test_service.go | 6 +++++-
1 file changed, 5 insertions(+), 1 deletion(-)
diff --git a/backend/internal/service/account_test_service.go b/backend/internal/service/account_test_service.go
index a67a699e9f..fd29b324b1 100644
--- a/backend/internal/service/account_test_service.go
+++ b/backend/internal/service/account_test_service.go
@@ -602,7 +602,11 @@ func (s *AccountTestService) testOpenAIAccountConnection(c *gin.Context, account
req.Header.Set("accept", "text/event-stream")
req.Header.Set("OpenAI-Beta", "responses=experimental")
req.Header.Set("Originator", "codex_cli_rs")
- req.Header.Set("User-Agent", codexCLIUserAgent)
+ if customUA := strings.TrimSpace(credentialAccount.GetOpenAIUserAgent()); customUA != "" {
+ req.Header.Set("User-Agent", customUA)
+ } else {
+ req.Header.Set("User-Agent", codexCLIUserAgent)
+ }
setOpenAIChatGPTAccountHeaders(req.Header, credentialAccount)
}
From 7c2a828c5648838f2e7353e2ab15e910980e40b8 Mon Sep 17 00:00:00 2001
From: qqqy <26496027+qqqianye@users.noreply.github.com>
Date: Tue, 7 Jul 2026 01:18:40 +0800
Subject: [PATCH 13/19] fix(frontend): add compact probe mode to admin account
test modal
---
.../admin/account/AccountTestModal.vue | 41 +++++++++++++++++--
.../__tests__/AccountTestModal.spec.ts | 34 +++++++++++++++
2 files changed, 71 insertions(+), 4 deletions(-)
diff --git a/frontend/src/components/admin/account/AccountTestModal.vue b/frontend/src/components/admin/account/AccountTestModal.vue
index ca9d06ee1e..0a0e3dd9ae 100644
--- a/frontend/src/components/admin/account/AccountTestModal.vue
+++ b/frontend/src/components/admin/account/AccountTestModal.vue
@@ -55,6 +55,17 @@
/>
+
+
+
+