fix: 修复佣金计算任务载荷二次序列化导致 base64 编码错误
All checks were successful
构建并部署到测试环境(无 SSH) / build-and-deploy (push) Successful in 7m11s
All checks were successful
构建并部署到测试环境(无 SSH) / build-and-deploy (push) Successful in 7m11s
问题:EnqueueTask 内部调用 sonic.Marshal,传入 []byte 时会将字节数组 base64 编码为 JSON 字符串,Handler 反序列化时类型不匹配(CommissionCalculationPayload vs string) 根因:order/service.go 的 enqueueCommissionCalculation 预先 Marshal 得到 []byte 后传给 EnqueueTask,导致载荷被二次序列化 修复:直接传入 map,由 EnqueueTask 统一序列化一次 规范同步: - pkg/queue/client.go 函数注释明确禁止传 []byte - AGENTS.md 新增「异步任务载荷规范」防止重现
This commit is contained in:
32
AGENTS.md
32
AGENTS.md
@@ -570,6 +570,38 @@ func RedisOrderCreateLockKey(carrierType string, carrierID uint) string
|
||||
| 套餐激活 | `internal/service/package/activation_service.go` → `ActivateQueuedPackage()` | 策略 2(简化版):Redis 分布式锁 |
|
||||
| 钱包扣款 | `internal/service/order/service.go` → `WalletPay()` | 策略 3:乐观锁(version 字段) |
|
||||
|
||||
### 异步任务载荷规范(Asynq)
|
||||
|
||||
**必须遵守:**
|
||||
|
||||
- ✅ 调用 `queueClient.EnqueueTask()` 时,`payload` 必须传入 **struct 或 map**
|
||||
- ❌ **禁止传入 `[]byte`**:`EnqueueTask` 内部统一调用 `sonic.Marshal`,若传入 `[]byte` 会被 base64 编码成字符串,Handler 反序列化时类型不匹配直接崩溃
|
||||
|
||||
```go
|
||||
// ✅ 正确:传 struct
|
||||
queueClient.EnqueueTask(ctx, constants.TaskTypeXxx, MyPayload{OrderID: id})
|
||||
|
||||
// ✅ 正确:传 map
|
||||
queueClient.EnqueueTask(ctx, constants.TaskTypeXxx, map[string]any{"order_id": id})
|
||||
|
||||
// ❌ 错误:预先序列化后传 []byte(二次 Marshal → base64 编码)
|
||||
payloadBytes, _ := sonic.Marshal(MyPayload{OrderID: id})
|
||||
queueClient.EnqueueTask(ctx, constants.TaskTypeXxx, payloadBytes)
|
||||
```
|
||||
|
||||
**直接用 `asynq.NewTask` 入队时**(绕过 `EnqueueTask`)才需要自己 Marshal:
|
||||
|
||||
```go
|
||||
// 直接构造 asynq.Task 时,需自行序列化
|
||||
payloadBytes, _ := sonic.Marshal(MyPayload{OrderID: id})
|
||||
task := asynq.NewTask(constants.TaskTypeXxx, payloadBytes, asynq.Queue(...))
|
||||
client.EnqueueContext(ctx, task)
|
||||
```
|
||||
|
||||
**原因**:`pkg/queue/client.go` 的 `EnqueueTask` 在内部已封装了 `sonic.Marshal`,统一处理序列化,调用方不应关心底层字节格式。
|
||||
|
||||
---
|
||||
|
||||
### 审计日志规范
|
||||
|
||||
**适用场景**:任何敏感操作(账号管理、权限变更、数据删除等)
|
||||
|
||||
@@ -20,7 +20,6 @@ import (
|
||||
"github.com/break/junhong_cmp_fiber/pkg/middleware"
|
||||
"github.com/break/junhong_cmp_fiber/pkg/queue"
|
||||
"github.com/break/junhong_cmp_fiber/pkg/wechat"
|
||||
"github.com/bytedance/sonic"
|
||||
"github.com/redis/go-redis/v9"
|
||||
"go.uber.org/zap"
|
||||
"gorm.io/gorm"
|
||||
@@ -2069,19 +2068,8 @@ func (s *Service) enqueueCommissionCalculation(ctx context.Context, orderID uint
|
||||
return
|
||||
}
|
||||
|
||||
payload := map[string]interface{}{
|
||||
"order_id": orderID,
|
||||
}
|
||||
|
||||
payloadBytes, err := sonic.Marshal(payload)
|
||||
if err != nil {
|
||||
s.logger.Error("佣金计算任务载荷序列化失败",
|
||||
zap.Uint("order_id", orderID),
|
||||
zap.Error(err))
|
||||
return
|
||||
}
|
||||
|
||||
if err := s.queueClient.EnqueueTask(ctx, constants.TaskTypeCommission, payloadBytes); err != nil {
|
||||
// 直接传 map,由 EnqueueTask 内部统一序列化一次(传 []byte 会导致 sonic.Marshal 二次 base64 编码)
|
||||
if err := s.queueClient.EnqueueTask(ctx, constants.TaskTypeCommission, map[string]any{"order_id": orderID}); err != nil {
|
||||
s.logger.Error("佣金计算任务入队失败",
|
||||
zap.Uint("order_id", orderID),
|
||||
zap.Error(err),
|
||||
|
||||
@@ -35,8 +35,10 @@ func NewClient(redisClient *redis.Client, logger *zap.Logger) *Client {
|
||||
}
|
||||
|
||||
// EnqueueTask 提交任务到队列
|
||||
// payload 必须传入 struct 或 map,禁止传入 []byte。
|
||||
// 传入 []byte 会导致 sonic.Marshal 将其 base64 编码,handler 解析时类型不匹配。
|
||||
func (c *Client) EnqueueTask(ctx context.Context, taskType string, payload interface{}, opts ...asynq.Option) error {
|
||||
// 序列化载荷
|
||||
// 内部统一序列化,调用方无需预先 Marshal
|
||||
payloadBytes, err := sonic.Marshal(payload)
|
||||
if err != nil {
|
||||
c.logger.Error("任务载荷序列化失败",
|
||||
|
||||
Reference in New Issue
Block a user