From af0822949dfda152f70c2637047aaf538b56289e Mon Sep 17 00:00:00 2001 From: yml2213 Date: Fri, 18 Sep 2026 16:19:37 +0800 Subject: [PATCH] =?UTF-8?q?fix(delivery):=20=E4=BF=AE=E5=A4=8D=20DLC=20?= =?UTF-8?q?=E6=AD=A3=E5=BC=8F=E4=B8=8B=E5=8D=95=E9=87=8D=E5=A4=8D=E6=8F=90?= =?UTF-8?q?=E4=BA=A4=E4=B8=8E=E5=8F=91=E8=B4=A7=E5=B9=B6=E5=8F=91=E5=86=B2?= =?UTF-8?q?=E7=AA=81=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1. DLC 正式下单接口禁用传输层自动重试：/public/users/orders 属于非幂等状态变更接口，发生频控或抖动时由上层回调与状态同步接管，严禁盲目原单号重试导致上游抛出“该子订单已处理，请勿重复提交”； 2. 扩充队列防重与已有记录识别：识别上游“该订单已有发货记录/已处理”错误，重新建队时自动追加独立重试单号（_xxxx）并透传至正式下单，避免单号污染冲突； 3. 上游回调支持重试单号兼容：HandleShipNotify 在精准匹配未命中时，自动按下划线前缀匹配主订单号（FO..._xxxx -> FO...），保障重试单回调正常交付； 4. 加强提交并发锁防护：在 35 秒活跃执行窗口内对 in_progress 阶段实行严格并发互斥，避免上游 Webhook 提前推回失败状态导致后续重试穿透产生并发打架；提单失败时显式重置 provider_order_stage 释放锁。 --- backend/internal/service/delivery.go | 22 ++++++- .../service/fulfillment_shipnotify.go | 17 +++++- backend/internal/service/provider_dlc.go | 58 +++++++++++++------ backend/internal/service/provider_dlc_test.go | 50 +++++++++++++--- 4 files changed, 116 insertions(+), 31 deletions(-) diff --git a/backend/internal/service/delivery.go b/backend/internal/service/delivery.go index 9c57549..c7e34e3 100644 --- a/backend/internal/service/delivery.go +++ b/backend/internal/service/delivery.go @@ -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()), diff --git a/backend/internal/service/fulfillment_shipnotify.go b/backend/internal/service/fulfillment_shipnotify.go index 0f84832..333a6f7 100644 --- a/backend/internal/service/fulfillment_shipnotify.go +++ b/backend/internal/service/fulfillment_shipnotify.go @@ -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() diff --git a/backend/internal/service/provider_dlc.go b/backend/internal/service/provider_dlc.go index 3b26132..739a79a 100644 --- a/backend/internal/service/provider_dlc.go +++ b/backend/internal/service/provider_dlc.go @@ -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 diff --git a/backend/internal/service/provider_dlc_test.go b/backend/internal/service/provider_dlc_test.go index d23d702..ed5812c 100644 --- a/backend/internal/service/provider_dlc_test.go +++ b/backend/internal/service/provider_dlc_test.go @@ -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) + } +} +