shanshanzhong147
af22430101
Build docker and publish / build (20.15.1) (pull_request) Failing after 8m15s
- 将 OrderStatusClaimed 从 4 改为 6，与 OrderStatusFailed(4) 区分 - releaseClaim 改为返回 error，失败时不再静默吞掉 - claimAndGetOrder 对 status=claimed 返回可重试错误而非静默跳过 - ProcessTask 区分 claimed stuck 错误（触发重试）和其他非 paid 状态（跳过） - finalizeCouponAndOrder 使用直接 DB 更新 claimed→finished，绕过 UpdateOrderStatus 的 status<target 守卫（6 > 5 无法通过该条件） 并显式删除 Redis 缓存避免缓存脏读 - 新增 StuckOrderRecoveryLogic：每 10 分钟扫描超时 claimed 订单， 重置为 paid 并重新入队激活任务 Co-authored-by: multica-agent <github@multica.ai>
87 lines
2.5 KiB
Go
87 lines
2.5 KiB
Go
package orderLogic
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"time"
|
|
|
|
"github.com/hibiken/asynq"
|
|
"github.com/perfect-panel/server/internal/model/order"
|
|
"github.com/perfect-panel/server/internal/svc"
|
|
"github.com/perfect-panel/server/pkg/logger"
|
|
queueTypes "github.com/perfect-panel/server/queue/types"
|
|
)
|
|
|
|
const stuckClaimTimeout = 10 * time.Minute
|
|
|
|
// StuckOrderRecoveryLogic scans orders stuck in claimed(6) status and resets them to paid(2)
|
|
// so that asynq retry can re-claim and process them.
|
|
type StuckOrderRecoveryLogic struct {
|
|
svc *svc.ServiceContext
|
|
}
|
|
|
|
func NewStuckOrderRecoveryLogic(svc *svc.ServiceContext) *StuckOrderRecoveryLogic {
|
|
return &StuckOrderRecoveryLogic{svc: svc}
|
|
}
|
|
|
|
func (l *StuckOrderRecoveryLogic) ProcessTask(ctx context.Context, _ *asynq.Task) error {
|
|
cutoff := time.Now().Add(-stuckClaimTimeout)
|
|
|
|
var stuckOrders []order.Order
|
|
if err := l.svc.DB.WithContext(ctx).
|
|
Model(&order.Order{}).
|
|
Where("status = ? AND updated_at < ?", OrderStatusClaimed, cutoff).
|
|
Find(&stuckOrders).Error; err != nil {
|
|
logger.WithContext(ctx).Error("[StuckOrderRecovery] Failed to query stuck orders",
|
|
logger.Field("error", err.Error()),
|
|
)
|
|
return err
|
|
}
|
|
|
|
if len(stuckOrders) == 0 {
|
|
return nil
|
|
}
|
|
|
|
orderNos := make([]string, 0, len(stuckOrders))
|
|
for i := range stuckOrders {
|
|
orderNos = append(orderNos, stuckOrders[i].OrderNo)
|
|
}
|
|
|
|
logger.WithContext(ctx).Error("[StuckOrderRecovery] Found stuck claimed orders, resetting to paid",
|
|
logger.Field("count", len(stuckOrders)),
|
|
logger.Field("order_nos", orderNos),
|
|
)
|
|
|
|
result := l.svc.DB.WithContext(ctx).
|
|
Model(&order.Order{}).
|
|
Where("status = ? AND updated_at < ?", OrderStatusClaimed, cutoff).
|
|
Update("status", OrderStatusPaid)
|
|
if result.Error != nil {
|
|
logger.WithContext(ctx).Error("[StuckOrderRecovery] Failed to reset stuck orders",
|
|
logger.Field("error", result.Error.Error()),
|
|
)
|
|
return result.Error
|
|
}
|
|
|
|
for i := range stuckOrders {
|
|
ord := &stuckOrders[i]
|
|
payload, err := json.Marshal(queueTypes.ForthwithActivateOrderPayload{OrderNo: ord.OrderNo})
|
|
if err != nil {
|
|
logger.WithContext(ctx).Error("[StuckOrderRecovery] Failed to marshal payload",
|
|
logger.Field("order_no", ord.OrderNo),
|
|
logger.Field("error", err.Error()),
|
|
)
|
|
continue
|
|
}
|
|
task := asynq.NewTask(queueTypes.ForthwithActivateOrder, payload, asynq.MaxRetry(5))
|
|
if _, err := l.svc.Queue.EnqueueContext(ctx, task); err != nil {
|
|
logger.WithContext(ctx).Error("[StuckOrderRecovery] Failed to re-enqueue order",
|
|
logger.Field("order_no", ord.OrderNo),
|
|
logger.Field("error", err.Error()),
|
|
)
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|