From f6b11eb9ace7e8fa7f8b3e89062c9551ec0bcb3b Mon Sep 17 00:00:00 2001 From: break Date: Mon, 10 Aug 2026 17:37:55 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .env | 4 -- internal/model/dto/polling_concurrency_dto.go | 2 +- internal/model/polling.go | 2 +- .../service/polling/concurrency_service.go | 21 ++++++--- internal/task/polling_base.go | 13 +++++- ..._polling_concurrency_task_configs.down.sql | 7 +++ ...ix_polling_concurrency_task_configs.up.sql | 9 ++++ .../.openspec.yaml | 2 + .../design.md | 45 +++++++++++++++++++ .../proposal.md | 26 +++++++++++ .../specs/polling-operations/spec.md | 36 +++++++++++++++ .../tasks.md | 17 +++++++ openspec/specs/polling-operations/spec.md | 26 ++++++++++- 13 files changed, 197 insertions(+), 13 deletions(-) create mode 100644 migrations/000206_fix_polling_concurrency_task_configs.down.sql create mode 100644 migrations/000206_fix_polling_concurrency_task_configs.up.sql create mode 100644 openspec/changes/archive/2026-08-10-fix-polling-concurrency-task-configs/.openspec.yaml create mode 100644 openspec/changes/archive/2026-08-10-fix-polling-concurrency-task-configs/design.md create mode 100644 openspec/changes/archive/2026-08-10-fix-polling-concurrency-task-configs/proposal.md create mode 100644 openspec/changes/archive/2026-08-10-fix-polling-concurrency-task-configs/specs/polling-operations/spec.md create mode 100644 openspec/changes/archive/2026-08-10-fix-polling-concurrency-task-configs/tasks.md diff --git a/.env b/.env index 31b4342..590b569 100644 --- a/.env +++ b/.env @@ -5,10 +5,6 @@ DB_USER=erp_pgsql DB_PASSWORD=erp_2025 DB_NAME=junhong_cmp_test DB_SSLMODE=disable -GOOGLE_GEMINI_BASE_URL="http://45.155.220.179:8317" # 根据实际填写你服务器的ip地址或者域名 -GEMINI_API_KEY="sk-VoNbvr6aGpjvZX64rvhrwowrZrCgtGuX9oxykIy8F1DBg" -GOOGLE_GENAI_USE_GCA="true" -GEMINI_MODEL="gemini-3-pro-preview" # 如果你有gemini3权限可以填： gemini-3-pro-preview # 七月迭代：Worker、企微 Adapter 与旧审批入口切换 JUNHONG_WORKER_ROLE=all diff --git a/internal/model/dto/polling_concurrency_dto.go b/internal/model/dto/polling_concurrency_dto.go index fc2f798..ecf1a3a 100644 --- a/internal/model/dto/polling_concurrency_dto.go +++ b/internal/model/dto/polling_concurrency_dto.go @@ -8,7 +8,7 @@ type GetPollingConcurrencyReq struct { // UpdatePollingConcurrencyReq 更新轮询并发配置请求 type UpdatePollingConcurrencyReq struct { TaskType string `path:"task_type" description:"任务类型" required:"true"` - MaxConcurrency int `json:"max_concurrency" validate:"required,min=1,max=1000" description:"最大并发数（1-1000）"` + MaxConcurrency int `json:"max_concurrency" validate:"required,min=1" description:"最大并发数（正整数）"` } // PollingConcurrencyResp 轮询并发配置响应 diff --git a/internal/model/polling.go b/internal/model/polling.go index 8a44185..46223da 100644 --- a/internal/model/polling.go +++ b/internal/model/polling.go @@ -33,7 +33,7 @@ func (PollingConfig) TableName() string { // PollingConcurrencyConfig 并发控制配置表 type PollingConcurrencyConfig struct { ID uint `gorm:"column:id;primaryKey;autoIncrement" json:"id"` - TaskType string `gorm:"column:task_type;type:varchar(50);uniqueIndex;not null;comment:任务类型：realname/carddata/package/stop_start" json:"task_type"` + TaskType string `gorm:"column:task_type;type:varchar(50);uniqueIndex;not null;comment:任务类型：realname/carddata/package/protect/card_status" json:"task_type"` MaxConcurrency int `gorm:"column:max_concurrency;not null;default:50;comment:最大并发数" json:"max_concurrency"` Description string `gorm:"column:description;type:text;comment:配置说明" json:"description"` CreatedAt time.Time `gorm:"column:created_at;not null;default:CURRENT_TIMESTAMP;comment:创建时间" json:"created_at"` diff --git a/internal/service/polling/concurrency_service.go b/internal/service/polling/concurrency_service.go index 6c438e1..45c4ef2 100644 --- a/internal/service/polling/concurrency_service.go +++ b/internal/service/polling/concurrency_service.go @@ -69,6 +69,9 @@ func (s *ConcurrencyService) List(ctx context.Context) ([]*ConcurrencyStatus, er if err != nil && err != redis.Nil { current = 0 } + if current < 0 { + current = 0 + } status.Current = current status.Available = int64(cfg.MaxConcurrency) - current @@ -104,6 +107,9 @@ func (s *ConcurrencyService) GetByTaskType(ctx context.Context, taskType string) if err != nil && err != redis.Nil { current = 0 } + if current < 0 { + current = 0 + } status.Current = current status.Available = int64(cfg.MaxConcurrency) - current @@ -120,8 +126,8 @@ func (s *ConcurrencyService) GetByTaskType(ctx context.Context, taskType string) // UpdateMaxConcurrency 更新最大并发数 func (s *ConcurrencyService) UpdateMaxConcurrency(ctx context.Context, taskType string, maxConcurrency int, updatedBy uint) error { // 验证参数 - if maxConcurrency < 1 || maxConcurrency > 1000 { - return errors.New(errors.CodeInvalidParam, "并发数必须在 1-1000 之间") + if maxConcurrency < 1 { + return errors.New(errors.CodeInvalidParam, "并发数必须为正整数") } // 验证任务类型存在 @@ -248,6 +254,7 @@ func (s *ConcurrencyService) InitFromDB(ctx context.Context) error { continue } } + _ = s.redis.Del(ctx, constants.RedisPollingConcurrencyConfigKey("stop_start")).Err() return nil } @@ -269,12 +276,16 @@ func pollingConcurrencyCurrentKey(taskType string) string { // getTaskTypeName 获取任务类型的中文名称 func (s *ConcurrencyService) getTaskTypeName(taskType string) string { switch taskType { - case constants.TaskTypePollingRealname: + case "realname": return "实名检查" - case constants.TaskTypePollingCarddata: + case "carddata": return "流量检查" - case constants.TaskTypePollingPackage: + case "package": return "套餐检查" + case "protect": + return "保护期检查" + case "card_status": + return "卡状态检查" default: return taskType } diff --git a/internal/task/polling_base.go b/internal/task/polling_base.go index 88b7897..213be76 100644 --- a/internal/task/polling_base.go +++ b/internal/task/polling_base.go @@ -29,6 +29,17 @@ var acquireConcurrencyScript = redis.NewScript(` return current `) +var releaseConcurrencyScript = redis.NewScript(` + local current = tonumber(redis.call('GET', KEYS[1]) or '0') or 0 + if current <= 0 then + if redis.call('EXISTS', KEYS[1]) == 1 then + redis.call('SET', KEYS[1], 0, 'KEEPTTL') + end + return 0 + end + return redis.call('DECR', KEYS[1]) +`) + const pollingFallbackOperationTimeout = 5 * time.Second // pollingFallbackContext 创建轮询兜底操作使用的独立短超时上下文。 @@ -101,7 +112,7 @@ func (b *PollingBase) releaseConcurrency(_ context.Context, taskType string) { defer cancel() currentKey := constants.RedisPollingConcurrencyCurrentKey(taskType) - if err := b.redis.Decr(ctx, currentKey).Err(); err != nil { + if err := releaseConcurrencyScript.Run(ctx, b.redis, []string{currentKey}).Err(); err != nil { b.logger.Warn("释放并发计数失败", zap.String("task_type", taskType), zap.Error(err)) } } diff --git a/migrations/000206_fix_polling_concurrency_task_configs.down.sql b/migrations/000206_fix_polling_concurrency_task_configs.down.sql new file mode 100644 index 0000000..565b015 --- /dev/null +++ b/migrations/000206_fix_polling_concurrency_task_configs.down.sql @@ -0,0 +1,7 @@ +-- 回滚至旧的四项配置集合。 +DELETE FROM tb_polling_concurrency_config +WHERE task_type IN ('protect', 'card_status'); + +INSERT INTO tb_polling_concurrency_config (task_type, max_concurrency, description) +VALUES ('stop_start', 50, '停复机检查最大并发数') +ON CONFLICT (task_type) DO NOTHING; diff --git a/migrations/000206_fix_polling_concurrency_task_configs.up.sql b/migrations/000206_fix_polling_concurrency_task_configs.up.sql new file mode 100644 index 0000000..d5f4dce --- /dev/null +++ b/migrations/000206_fix_polling_concurrency_task_configs.up.sql @@ -0,0 +1,9 @@ +-- 轮询并发配置与实际执行的五类短任务类型保持一致。 +INSERT INTO tb_polling_concurrency_config (task_type, max_concurrency, description) +VALUES + ('protect', 300, '保护期检查最大并发数'), + ('card_status', 300, '卡状态检查最大并发数') +ON CONFLICT (task_type) DO NOTHING; + +DELETE FROM tb_polling_concurrency_config +WHERE task_type = 'stop_start'; diff --git a/openspec/changes/archive/2026-08-10-fix-polling-concurrency-task-configs/.openspec.yaml b/openspec/changes/archive/2026-08-10-fix-polling-concurrency-task-configs/.openspec.yaml new file mode 100644 index 0000000..d7bc011 --- /dev/null +++ b/openspec/changes/archive/2026-08-10-fix-polling-concurrency-task-configs/.openspec.yaml @@ -0,0 +1,2 @@ +schema: spec-driven +created: 2026-08-10 diff --git a/openspec/changes/archive/2026-08-10-fix-polling-concurrency-task-configs/design.md b/openspec/changes/archive/2026-08-10-fix-polling-concurrency-task-configs/design.md new file mode 100644 index 0000000..10c706f --- /dev/null +++ b/openspec/changes/archive/2026-08-10-fix-polling-concurrency-task-configs/design.md @@ -0,0 +1,45 @@ +## Context + +现有配置表保存短任务类型，Worker 使用 `polling:` 作为当前计数键；配置缓存仍使用短类型。当前数据缺少两个已运行的任务、保留一个无执行器任务，且管理服务的展示名称匹配了错误的完整任务类型。详见 proposal.md 与轮询运营 delta spec。 + +## Goals / Non-Goals + +**Goals:** +- 让数据库配置、Redis 缓存与五个 Worker 任务类型保持一致。 +- 保持 `protect` 与 `card_status` 当前回退生效的 300 上限不变。 +- 在共享释放点消除负信号量。 + +**Non-Goals:** +- 不调整既有 `realname`、`carddata`、`package` 的最大并发数。 +- 不依据资产数量自动计算或提高并发，也不更改队列、Worker 数量或第三方请求策略。 +- 不修改历史迁移。 + +## Decisions + +### 以数据迁移校正配置集合 +新增成对迁移使用幂等插入补齐 `protect`、`card_status`，初始值均为 300；删除 `stop_start`。300 是两类任务当前 Redis 配置缺失时 Worker 已采用的回退值，迁移后可观测、可维护而不改变即时压力。回滚恢复遗留 `stop_start` 的默认记录并移除两项新增记录。 + +备选方案是把两项直接设为 5000；这会把实际请求上限从当前回退值提高，超出本次配置一致性修复范围。 + +### 以短任务类型作为管理面唯一标识 +管理配置表、路由参数与 Redis 配置缓存继续使用短类型；仅当前占用计数键在服务内部转换为完整任务类型。名称映射按短类型覆盖全部五项，避免接口将内部键回显给运营者。 + +### 释放脚本钳制最小值 +共享 `releaseConcurrency` 改用 Lua 原子操作：仅当当前值大于零时递减，否则将计数保持或归零。这样重置、TTL 到期与重复释放均不会产生负数，且不需在每个 Handler 添加分支。 + +### 移除无效上限校验 +最大并发仍必须为正整数，移除与已有 5000 配置冲突的 1000 上限。数据库的整数类型继续承担存储边界；本变更不引入新的容量策略。 + +## Risks / Trade-offs + +- [回滚无法保留历史 `stop_start` 的人工修改值] → 该任务没有执行器，回滚仅恢复默认遗留记录；上线前记录现有值。 +- [Redis 中可能残留 `stop_start` 缓存键] → 部署初始化仅同步有效数据库配置；实施时显式删除该遗留缓存键。 +- [释放与获取并发] → 释放脚本是单键原子操作，不改变获取脚本或任务重入队语义。 + +## Migration Plan + +1. 发布包含新迁移与服务/Worker 修复的版本。 +2. 执行迁移，补齐两项 300 配置并删除 `stop_start`。 +3. 服务启动同步五项配置到 Redis，并清理遗留 `stop_start` 缓存键。 +4. 通过管理接口确认五项任务、中文名称和非负计数；在重置后完成一个运行中任务，确认计数不低于零。 +5. 若必须回滚，先停止新版本，再执行 down migration，恢复旧版本并重新同步其配置缓存。 diff --git a/openspec/changes/archive/2026-08-10-fix-polling-concurrency-task-configs/proposal.md b/openspec/changes/archive/2026-08-10-fix-polling-concurrency-task-configs/proposal.md new file mode 100644 index 0000000..aa3bf90 --- /dev/null +++ b/openspec/changes/archive/2026-08-10-fix-polling-concurrency-task-configs/proposal.md @@ -0,0 +1,26 @@ +## Why + +轮询并发配置表与实际执行的五类轮询任务不一致：`protect`、`card_status` 没有可管理配置，遗留的 `stop_start` 又不再被执行器使用。管理接口还会显示错误名称、拒绝维护已有的 5000 并发配置，且重置或过期后的任务释放可能把计数减为负数。 + +## What Changes + +- 将轮询并发配置的受支持任务集合统一为 `realname`、`carddata`、`package`、`protect`、`card_status`。 +- 以新迁移补齐 `protect` 与 `card_status` 的 300 并发配置，移除遗留 `stop_start` 配置。 +- 让并发管理接口展示正确的中文任务名称，并允许维护现有的正整数并发上限。 +- 让并发信号量释放在计数已不存在或不为正时保持为零，避免管理接口与限流状态出现负数。 + +## Capabilities + +### New Capabilities + +- 无。 + +### Modified Capabilities + +- `polling-operations`: 轮询并发配置、状态展示与信号量释放的可观察行为。 + +## Impact + +- 新增一组成对数据库迁移，修改 `tb_polling_concurrency_config` 的初始化数据。 +- 影响轮询并发管理服务、共享任务限流器、管理接口返回值与接口文档。 +- 不新增依赖，不变更既有三个任务的并发配置值。 diff --git a/openspec/changes/archive/2026-08-10-fix-polling-concurrency-task-configs/specs/polling-operations/spec.md b/openspec/changes/archive/2026-08-10-fix-polling-concurrency-task-configs/specs/polling-operations/spec.md new file mode 100644 index 0000000..5c23ce9 --- /dev/null +++ b/openspec/changes/archive/2026-08-10-fix-polling-concurrency-task-configs/specs/polling-operations/spec.md @@ -0,0 +1,36 @@ +## ADDED Requirements + +### Requirement: 轮询并发配置覆盖实际任务 +系统 SHALL 为 `realname`、`carddata`、`package`、`protect` 与 `card_status` 五类实际轮询任务各维护一项可查询、可更新的并发配置；`stop_start` 不得作为轮询并发配置返回或接受维护。新增的 `protect` 与 `card_status` 初始最大并发数 MUST 为 300。 + +#### Scenario: 查询完整配置集合 +- **WHEN** 授权操作者查询轮询并发配置列表 +- **THEN** 返回上述五类任务且不含 `stop_start` + +#### Scenario: 维护高于一千的既有并发配置 +- **WHEN** 授权操作者为已配置轮询任务提交大于 1000 的正整数最大并发数 +- **THEN** 系统保存该值并使后续轮询限流读取该值 + +### Requirement: 轮询任务类型名称可读 +系统 SHALL 在轮询并发状态中返回与任务类型一致的中文名称：实名检查、流量检查、套餐检查、保护期检查或卡状态检查。 + +#### Scenario: 查询卡状态并发配置 +- **WHEN** 授权操作者查询 `card_status` 的并发状态 +- **THEN** 返回的 `task_type_name` 为“卡状态检查” + +## MODIFIED Requirements + +### Requirement: 轮询并发计数可观测 +系统 SHALL 在轮询并发配置列表和详情中返回与实际限流器相同任务类型的当前计数、可用并发和使用率；重置操作 MUST 重置该同一计数。当前计数、可用并发与使用率 MUST 不因释放过期或已重置的计数而呈现负值。 + +#### Scenario: 查询运行中的任务计数 +- **WHEN** 某轮询任务正在占用并发配额 +- **THEN** 查询该任务类型的并发状态返回非零当前计数，并据此计算可用并发和使用率 + +#### Scenario: 重置任务计数 +- **WHEN** 授权操作者重置某轮询任务类型的并发计数 +- **THEN** 后续状态查询返回该任务类型的当前计数为零，且不影响其他任务类型的计数 + +#### Scenario: 重置后的任务完成 +- **WHEN** 某任务在其并发计数已重置或过期后完成 +- **THEN** 该任务类型的当前计数保持为零而不变为负数 diff --git a/openspec/changes/archive/2026-08-10-fix-polling-concurrency-task-configs/tasks.md b/openspec/changes/archive/2026-08-10-fix-polling-concurrency-task-configs/tasks.md new file mode 100644 index 0000000..cd189b6 --- /dev/null +++ b/openspec/changes/archive/2026-08-10-fix-polling-concurrency-task-configs/tasks.md @@ -0,0 +1,17 @@ +## 1. 配置数据迁移 + +- [x] 1.1 新增 `000206` 成对迁移：幂等写入 `protect`、`card_status` 两项 300 并发配置，并删除 `stop_start`；down 迁移恢复旧配置集合。 +- [x] 1.2 更新并发配置模型的任务类型说明，反映五项实际受支持的短任务类型。 + +## 2. 并发管理与限流 + +- [x] 2.1 修正并发管理服务的任务中文名称映射，覆盖全部五个短任务类型。 +- [x] 2.2 移除与现有数据冲突的 1000 上限，保留最大并发必须为正整数的校验。 +- [x] 2.3 在配置初始化流程清理 `stop_start` 的 Redis 配置缓存。 +- [x] 2.4 将共享并发释放改为原子非负释放，确保重置、过期或重复释放后计数不小于零。 + +## 3. 文档与验证 + +- [x] 3.1 更新接口生成文档，确保轮询并发管理接口描述与响应一致。 +- [x] 3.2 运行 gofmt、`go build ./cmd/api ./cmd/worker`、`go run cmd/gendocs/main.go`、`openspec validate --all` 与 `./scripts/context-health.sh`，记录结果。 +- [x] 3.3 在隔离数据库执行迁移并验证配置列表只含五项、两项新增值为 300；验证重置后释放不会产生负数，并验证 down 迁移可执行。 diff --git a/openspec/specs/polling-operations/spec.md b/openspec/specs/polling-operations/spec.md index bbc05db..35d18de 100644 --- a/openspec/specs/polling-operations/spec.md +++ b/openspec/specs/polling-operations/spec.md @@ -28,7 +28,7 @@ ### Requirement: 轮询并发计数可观测 -系统 SHALL 在轮询并发配置列表和详情中返回与实际限流器相同任务类型的当前计数、可用并发和使用率；重置操作 MUST 重置该同一计数。 +系统 SHALL 在轮询并发配置列表和详情中返回与实际限流器相同任务类型的当前计数、可用并发和使用率；重置操作 MUST 重置该同一计数。当前计数、可用并发与使用率 MUST 不因释放过期或已重置的计数而呈现负值。 #### Scenario: 查询运行中的任务计数 - **WHEN** 某轮询任务正在占用并发配额 @@ -38,6 +38,30 @@ - **WHEN** 授权操作者重置某轮询任务类型的并发计数 - **THEN** 后续状态查询返回该任务类型的当前计数为零，且不影响其他任务类型的计数 +#### Scenario: 重置后的任务完成 +- **WHEN** 某任务在其并发计数已重置或过期后完成 +- **THEN** 该任务类型的当前计数保持为零而不变为负数 + +### Requirement: 轮询并发配置覆盖实际任务 + +系统 SHALL 为 `realname`、`carddata`、`package`、`protect` 与 `card_status` 五类实际轮询任务各维护一项可查询、可更新的并发配置；`stop_start` 不得作为轮询并发配置返回或接受维护。新增的 `protect` 与 `card_status` 初始最大并发数 MUST 为 300。 + +#### Scenario: 查询完整配置集合 +- **WHEN** 授权操作者查询轮询并发配置列表 +- **THEN** 返回上述五类任务且不含 `stop_start` + +#### Scenario: 维护高于一千的既有并发配置 +- **WHEN** 授权操作者为已配置轮询任务提交大于 1000 的正整数最大并发数 +- **THEN** 系统保存该值并使后续轮询限流读取该值 + +### Requirement: 轮询任务类型名称可读 + +系统 SHALL 在轮询并发状态中返回与任务类型一致的中文名称：实名检查、流量检查、套餐检查、保护期检查或卡状态检查。 + +#### Scenario: 查询卡状态并发配置 +- **WHEN** 授权操作者查询 `card_status` 的并发状态 +- **THEN** 返回的 `task_type_name` 为“卡状态检查” + ## 可达操作索引 本节只用于入口导航，不是行为 Requirement；业务义务以上述 Requirements 为准。