fix(delivery): 修复 DLC 正式下单重复提交与发货并发冲突问题
1. DLC 正式下单接口禁用传输层自动重试:/public/users/orders 属于非幂等状态变更接口,发生频控或抖动时由上层回调与状态同步接管,严禁盲目原单号重试导致上游抛出“该子订单已处理,请勿重复提交”; 2. 扩充队列防重与已有记录识别:识别上游“该订单已有发货记录/已处理”错误,重新建队时自动追加独立重试单号(_xxxx)并透传至正式下单,避免单号污染冲突; 3. 上游回调支持重试单号兼容:HandleShipNotify 在精准匹配未命中时,自动按下划线前缀匹配主订单号(FO..._xxxx -> FO...),保障重试单回调正常交付; 4. 加强提交并发锁防护:在 35 秒活跃执行窗口内对 in_progress 阶段实行严格并发互斥,避免上游 Webhook 提前推回失败状态导致后续重试穿透产生并发打架;提单失败时显式重置 provider_order_stage 释放锁。
This commit is contained in:
@@ -675,7 +675,20 @@ func (s *DeliveryService) claimDeliverySubmission(merchantID, apiClientID uint,
|
||||
}
|
||||
now := time.Now()
|
||||
stage := resultDataString(order.ResultData, "provider_order_stage")
|
||||
// ship_failed 可立即重试;仅对仍处于 paid 等提交中态拦截并发双发
|
||||
startedAtRaw := resultDataString(order.ResultData, "submit_started_at")
|
||||
|
||||
// 1. 如果已有协程正在活跃提交中(在 35 秒执行窗口内且处于 claimed/queue_created 阶段),
|
||||
// 即使已被外部 Webhook 临时推回 ship_failed,也绝对不允许并发重入发货打架
|
||||
if deliverySubmissionInProgress(stage) && startedAtRaw != "" {
|
||||
if startedAt, parseErr := time.Parse(timeutil.APITimeLayout, startedAtRaw); parseErr == nil {
|
||||
if now.Sub(startedAt) < 35*time.Second {
|
||||
updated = order
|
||||
return nil
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 2. 正常并发防双发:仅对 paid 等非最终失败态订单检查 10 分钟未超时
|
||||
if orderStatus != model.OrderStatusShipFailed &&
|
||||
deliverySubmissionInProgress(stage) &&
|
||||
!deliverySubmissionStale(&order, now) {
|
||||
@@ -736,8 +749,13 @@ func (s *DeliveryService) markDeliverySubmissionFailed(order *model.FulfillmentO
|
||||
if fresh, err := s.fulfillment.GetOrder(order.MerchantID, order.OrderNo); err == nil {
|
||||
latest = fresh
|
||||
}
|
||||
// 确保失败时 stage 绝不能继续保持在 in_progress(claimed / queue_created),否则阻碍重试
|
||||
releaseStage := stage
|
||||
if deliverySubmissionInProgress(releaseStage) {
|
||||
releaseStage = deliveryStageCreateOrder
|
||||
}
|
||||
resultData := mergeResultData(latest.ResultData, map[string]interface{}{
|
||||
"provider_order_stage": stage,
|
||||
"provider_order_stage": releaseStage,
|
||||
"fail_stage": stage,
|
||||
"fail_reason": reason,
|
||||
"fail_at": timeutil.FormatAPITime(time.Now()),
|
||||
|
||||
@@ -77,10 +77,21 @@ func (s *FulfillmentService) HandleShipNotify(in ShipNotifyInput) (*ShipNotifyRe
|
||||
Preload("MerchantProduct.Product").
|
||||
Where("order_no = ?", in.OrderNo).
|
||||
First(&order).Error; err != nil {
|
||||
if errors.Is(err, gorm.ErrRecordNotFound) {
|
||||
return errors.New("订单不存在")
|
||||
if errors.Is(err, gorm.ErrRecordNotFound) && strings.Contains(in.OrderNo, "_") {
|
||||
baseOrderNo := strings.SplitN(in.OrderNo, "_", 2)[0]
|
||||
if err2 := tx.Clauses(clause.Locking{Strength: "UPDATE"}).
|
||||
Preload("MerchantProduct.Product").
|
||||
Where("order_no = ?", baseOrderNo).
|
||||
First(&order).Error; err2 == nil {
|
||||
err = nil
|
||||
}
|
||||
}
|
||||
if err != nil {
|
||||
if errors.Is(err, gorm.ErrRecordNotFound) {
|
||||
return errors.New("订单不存在")
|
||||
}
|
||||
return err
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
now := time.Now()
|
||||
|
||||
@@ -219,16 +219,24 @@ func (p *DLCProvider) Submit(ctx context.Context, order *model.FulfillmentOrder,
|
||||
partial := map[string]interface{}{
|
||||
"provider": ProviderDLC,
|
||||
}
|
||||
effectiveOrderSN := order.OrderNo
|
||||
if savedSN := resultDataString(order.ResultData, "order_sn"); savedSN != "" {
|
||||
effectiveOrderSN = savedSN
|
||||
}
|
||||
|
||||
if queueOrderID == "" {
|
||||
createdID, err := p.createOrderQueue(goodID, order.OrderNo, goodsNo)
|
||||
createdID, sn, err := p.createOrderQueue(goodID, order.OrderNo, goodsNo)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("创建发货队列失败: %w", err)
|
||||
}
|
||||
queueOrderID = createdID
|
||||
effectiveOrderSN = sn
|
||||
partial["queue_order_id"] = queueOrderID
|
||||
partial["order_sn"] = effectiveOrderSN
|
||||
partial["provider_order_stage"] = deliveryStageQueueCreated
|
||||
} else {
|
||||
partial["queue_order_id"] = queueOrderID
|
||||
partial["order_sn"] = effectiveOrderSN
|
||||
partial["provider_order_stage"] = deliveryStageQueueCreated
|
||||
}
|
||||
|
||||
@@ -236,12 +244,14 @@ func (p *DLCProvider) Submit(ctx context.Context, order *model.FulfillmentOrder,
|
||||
if isQueueDeadError(err) {
|
||||
// 旧队列已失效/不可操作,清除失效队列并重新创建
|
||||
delete(partial, "queue_order_id")
|
||||
newQueueID, createErr := p.createOrderQueue(goodID, order.OrderNo, goodsNo)
|
||||
newQueueID, newSN, createErr := p.createOrderQueue(goodID, order.OrderNo, goodsNo)
|
||||
if createErr != nil {
|
||||
return &ProviderSubmitResult{PatchResultData: partial}, fmt.Errorf("重置发货队列失败: %w (原错误: %v)", createErr, err)
|
||||
}
|
||||
queueOrderID = newQueueID
|
||||
effectiveOrderSN = newSN
|
||||
partial["queue_order_id"] = queueOrderID
|
||||
partial["order_sn"] = effectiveOrderSN
|
||||
if patchErr := p.patchOrderQueue(queueOrderID, gameAccount, bindUUID, goodsNo, goodsName); patchErr != nil {
|
||||
return &ProviderSubmitResult{PatchResultData: partial}, fmt.Errorf("补充发货账号失败: %w", patchErr)
|
||||
}
|
||||
@@ -250,17 +260,20 @@ func (p *DLCProvider) Submit(ctx context.Context, order *model.FulfillmentOrder,
|
||||
}
|
||||
}
|
||||
|
||||
upstreamOrder, err := p.createUpstreamOrder(queueOrderID, gameAccount, goodID, order.OrderNo)
|
||||
upstreamOrder, err := p.createUpstreamOrder(queueOrderID, gameAccount, goodID, effectiveOrderSN)
|
||||
if err != nil {
|
||||
if isQueueDeadError(err) {
|
||||
// 创建正式单阶段出现订单不可操作,尝试重建队列重试一次
|
||||
// 创建正式单阶段出现订单不可操作或重复提交,尝试换新重试单号重建队列重试一次
|
||||
delete(partial, "queue_order_id")
|
||||
newQueueID, createErr := p.createOrderQueue(goodID, order.OrderNo, goodsNo)
|
||||
retryBaseSN := fmt.Sprintf("%s_%x", strings.SplitN(order.OrderNo, "_", 2)[0], time.Now().UnixNano()%0xffff)
|
||||
newQueueID, newSN, createErr := p.createOrderQueue(goodID, retryBaseSN, goodsNo)
|
||||
if createErr == nil {
|
||||
queueOrderID = newQueueID
|
||||
effectiveOrderSN = newSN
|
||||
partial["queue_order_id"] = queueOrderID
|
||||
partial["order_sn"] = effectiveOrderSN
|
||||
if patchErr := p.patchOrderQueue(queueOrderID, gameAccount, bindUUID, goodsNo, goodsName); patchErr == nil {
|
||||
if retriedOrder, retriedErr := p.createUpstreamOrder(queueOrderID, gameAccount, goodID, order.OrderNo); retriedErr == nil {
|
||||
if retriedOrder, retriedErr := p.createUpstreamOrder(queueOrderID, gameAccount, goodID, effectiveOrderSN); retriedErr == nil {
|
||||
upstreamOrder = retriedOrder
|
||||
err = nil
|
||||
} else {
|
||||
@@ -380,14 +393,15 @@ func (p *DLCProvider) accountBound(bindUUID, goodID string) (map[string]interfac
|
||||
return account, err
|
||||
}
|
||||
|
||||
func (p *DLCProvider) createOrderQueue(goodID, orderNo, goodsNo string) (string, error) {
|
||||
func (p *DLCProvider) createOrderQueue(goodID, orderNo, goodsNo string) (string, string, error) {
|
||||
var out struct {
|
||||
Orders []map[string]interface{} `json:"orders"`
|
||||
}
|
||||
usedSN := orderNo
|
||||
payload := map[string]interface{}{
|
||||
"good_id": goodID,
|
||||
"quantity": 1,
|
||||
"order_sn": orderNo,
|
||||
"order_sn": usedSN,
|
||||
"h5_prefix": p.channel,
|
||||
}
|
||||
if goodsNo != "" {
|
||||
@@ -395,10 +409,14 @@ func (p *DLCProvider) createOrderQueue(goodID, orderNo, goodsNo string) (string,
|
||||
}
|
||||
err := p.signProxy("/public/users/orders-queue", "POST", payload, &out)
|
||||
if err != nil {
|
||||
if strings.Contains(err.Error(), "正在处理中") || strings.Contains(err.Error(), "请勿重复提交") {
|
||||
// DLC 已存在此 order_sn 的队列,追加随机后缀重新创建队列
|
||||
retrySN := fmt.Sprintf("%s_%x", orderNo, time.Now().UnixNano()%0xffff)
|
||||
payload["order_sn"] = retrySN
|
||||
if strings.Contains(err.Error(), "正在处理中") ||
|
||||
strings.Contains(err.Error(), "请勿重复提交") ||
|
||||
strings.Contains(err.Error(), "已有发货记录") ||
|
||||
strings.Contains(err.Error(), "已处理") {
|
||||
// DLC 已存在此 order_sn 的队列/发货记录,追加随机后缀重新创建队列
|
||||
retrySN := fmt.Sprintf("%s_%x", strings.SplitN(orderNo, "_", 2)[0], time.Now().UnixNano()%0xffff)
|
||||
usedSN = retrySN
|
||||
payload["order_sn"] = usedSN
|
||||
out.Orders = nil
|
||||
if retryErr := p.signProxy("/public/users/orders-queue", "POST", payload, &out); retryErr == nil && len(out.Orders) > 0 {
|
||||
err = nil
|
||||
@@ -406,16 +424,16 @@ func (p *DLCProvider) createOrderQueue(goodID, orderNo, goodsNo string) (string,
|
||||
}
|
||||
}
|
||||
if err != nil {
|
||||
return "", err
|
||||
return "", "", err
|
||||
}
|
||||
if len(out.Orders) == 0 {
|
||||
return "", errors.New("发货服务未返回队列订单")
|
||||
return "", "", errors.New("发货服务未返回队列订单")
|
||||
}
|
||||
orderID := firstNonEmpty(stringFromMap(out.Orders[0], "_id"), stringFromMap(out.Orders[0], "id"))
|
||||
if orderID == "" {
|
||||
return "", errors.New("发货服务未返回队列订单 ID")
|
||||
return "", "", errors.New("发货服务未返回队列订单 ID")
|
||||
}
|
||||
return orderID, nil
|
||||
return orderID, usedSN, nil
|
||||
}
|
||||
|
||||
func (p *DLCProvider) patchOrderQueue(orderID, gameAccount, bindUUID, goodsNo, goodsName string) error {
|
||||
@@ -468,7 +486,9 @@ func isQueueDeadError(err error) bool {
|
||||
strings.Contains(msg, "订单不存在") ||
|
||||
strings.Contains(msg, "队列已失效") ||
|
||||
strings.Contains(msg, "正在处理中") ||
|
||||
strings.Contains(msg, "请勿重复提交")
|
||||
strings.Contains(msg, "请勿重复提交") ||
|
||||
strings.Contains(msg, "已有发货记录") ||
|
||||
strings.Contains(msg, "已处理")
|
||||
}
|
||||
|
||||
func isRateLimitError(err error) bool {
|
||||
@@ -510,8 +530,8 @@ func (p *DLCProvider) signProxy(path, method string, data interface{}, out inter
|
||||
}
|
||||
|
||||
maxRetries := 2
|
||||
if path == "/public/games/account-bound" {
|
||||
maxRetries = 0 // 轮询绑定接口不触发重试睡眠,避免堆积阻塞
|
||||
if path == "/public/games/account-bound" || path == "/public/users/orders" {
|
||||
maxRetries = 0 // 轮询绑定接口与非幂等正式下单接口不触发传输层重试,避免堆积阻塞或单号重复冲突
|
||||
}
|
||||
var lastErr error
|
||||
|
||||
|
||||
@@ -22,6 +22,8 @@ func TestIsQueueDeadError(t *testing.T) {
|
||||
{errors.New("队列已失效"), true},
|
||||
{errors.New("该订单正在处理中,请勿重复提交"), true},
|
||||
{errors.New("创建发货队列失败: 该订单正在处理中,请勿重复提交"), true},
|
||||
{errors.New("创建发货队列失败: 该订单已有发货记录,请续发未完成项或重试失败项"), true},
|
||||
{errors.New("创建上游正式订单失败: 该子订单已处理,请勿重复提交"), true},
|
||||
{errors.New("账号尚未绑定"), false},
|
||||
{errors.New("请求过于频繁"), false},
|
||||
}
|
||||
@@ -106,8 +108,8 @@ func TestSignProxyRetriesOnRateLimitAndSucceeds(t *testing.T) {
|
||||
"code": 0,
|
||||
"message": "success",
|
||||
"data": map[string]interface{}{
|
||||
"gameAccount": map[string]interface{}{
|
||||
"game_account": "12345678",
|
||||
"good": map[string]interface{}{
|
||||
"good_id": "test-good",
|
||||
},
|
||||
},
|
||||
})
|
||||
@@ -118,10 +120,10 @@ func TestSignProxyRetriesOnRateLimitAndSucceeds(t *testing.T) {
|
||||
p := NewDLCProvider(server.URL, "dlc", client)
|
||||
|
||||
var out struct {
|
||||
GameAccount map[string]interface{} `json:"gameAccount"`
|
||||
Good map[string]interface{} `json:"good"`
|
||||
}
|
||||
err := p.signProxy("/public/games/account-bound", "POST", map[string]interface{}{
|
||||
"bind_uuid": "test-uuid",
|
||||
err := p.signProxy("/public/goods/detail", "POST", map[string]interface{}{
|
||||
"good_id": "test-good",
|
||||
}, &out)
|
||||
|
||||
if err != nil {
|
||||
@@ -133,7 +135,41 @@ func TestSignProxyRetriesOnRateLimitAndSucceeds(t *testing.T) {
|
||||
t.Fatalf("预期总共调用 2 次(1次失败 + 1次重试成功),实际调用: %d", finalCount)
|
||||
}
|
||||
|
||||
if out.GameAccount == nil || out.GameAccount["game_account"] != "12345678" {
|
||||
t.Fatalf("预期返回账号信息,实际: %v", out.GameAccount)
|
||||
if out.Good == nil || out.Good["good_id"] != "test-good" {
|
||||
t.Fatalf("预期返回商品信息,实际: %v", out.Good)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSignProxyDoesNotRetryOnFormalOrderCreation(t *testing.T) {
|
||||
var callCount int32
|
||||
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
atomic.AddInt32(&callCount, 1)
|
||||
w.WriteHeader(http.StatusBadRequest)
|
||||
_ = json.NewEncoder(w).Encode(map[string]interface{}{
|
||||
"code": 1,
|
||||
"message": "请求过于频繁,请稍后再试",
|
||||
})
|
||||
}))
|
||||
defer server.Close()
|
||||
|
||||
client := &http.Client{Timeout: 5 * time.Second}
|
||||
p := NewDLCProvider(server.URL, "dlc", client)
|
||||
|
||||
var out struct {
|
||||
Order map[string]interface{} `json:"order"`
|
||||
}
|
||||
err := p.signProxy("/public/users/orders", "POST", map[string]interface{}{
|
||||
"order_sn": "FO123456",
|
||||
}, &out)
|
||||
|
||||
if err == nil {
|
||||
t.Fatalf("预期返回错误,但返回成功")
|
||||
}
|
||||
|
||||
finalCount := atomic.LoadInt32(&callCount)
|
||||
if finalCount != 1 {
|
||||
t.Fatalf("正式下单接口预期只调用 1 次且不触发重试,实际调用: %d", finalCount)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user