diff --git a/internal/gateway/gateway_test.go b/internal/gateway/gateway_test.go index b32646f..590f34f 100644 --- a/internal/gateway/gateway_test.go +++ b/internal/gateway/gateway_test.go @@ -35,7 +35,11 @@ func (stubResolver) Resolve(sku, currency string) (int64, string, string, error) type spyEnqueuer struct { calls []map[string]any - failNext bool // 置 true 模拟 outbox 入队失败(settle 崩溃窗口测试用) + failNext bool // 置 true 模拟 outbox 入队失败(settle 崩溃窗口测试用) + seen map[string]bool // 镜像真实 WebhookStore.EnqueueDelivery 的 (out_trade_no,event_type, + // refund_id) 唯一键 ON CONFLICT DO NOTHING:重复 key 静默 no-op(不追加 calls,也不算失败)。 + // 修复"重投补入队自愈"后,业务代码会在 duplicate/重投分支也调用 Enqueue,若 spy 仍是无脑 + // 计数器就会把 outbox 天然幂等的重复行误判成"多发了一次 webhook",这里镜像真实幂等语义。 } func (s *spyEnqueuer) Enqueue(outTradeNo, bizSystem, eventType, refundID string, data map[string]any) error { @@ -43,6 +47,14 @@ func (s *spyEnqueuer) Enqueue(outTradeNo, bizSystem, eventType, refundID string, s.failNext = false return errors.New("outbox down") } + key := outTradeNo + "|" + eventType + "|" + refundID + if s.seen == nil { + s.seen = make(map[string]bool) + } + if s.seen[key] { + return nil // 幂等 no-op,行已存在 + } + s.seen[key] = true s.calls = append(s.calls, data) return nil } diff --git a/internal/gateway/settle.go b/internal/gateway/settle.go index e5e6ba6..38faa8b 100644 --- a/internal/gateway/settle.go +++ b/internal/gateway/settle.go @@ -138,7 +138,17 @@ func (g *Gateway) HandleCallback(ctx context.Context, method string, in provider // settleRenewal 处理续费 invoice.paid(设计 §5/§4 决策记录):每期铸独立 renewal OrderV2 // (out_trade_no = 首购单号 + "-r-" + invoice id),建即 paid(续费不经收银台,无 pending 中间态)。 // 幂等靠 renewal attempt 的 (channel,provider_ref=invoice.ID) 唯一索引 + renewal order 的 -// out_trade_no 唯一索引双保险;出问题让 Stripe 重投,重投时 created=false 不重复入队,自愈。 +// out_trade_no 唯一索引双保险。 +// +// 入队纪律(与文件头 Settle 的顺序不变量同根同源,但形态不同——续费建单是"单步已 paid", +// 没有 Settle 那种"翻转严格发生在成功入队之后"的两段式可依赖):**duplicate 分支(created= +// false)也必须尝试入队**,不能像旧实现那样直接 return。理由:首次入队若瞬时失败,订单/attempt +// 已在同一事务内落为 paid(不回滚,幂等键护着),而 outbox 从未有行;Stripe 拿不到 200 会重投 +// 同一 invoice.paid,此时 created 必为 false——若 duplicate 分支不入队,该续费通知永久丢失 +// (SyncPendingAttempts 也救不了:续费 attempt 生来就是 AttemptPaid,不在 pending 轮询范围)。 +// outbox 唯一键 (out_trade_no,event_type,refund_id) 上 ON CONFLICT DO NOTHING 天然幂等:行已 +// 存在则本次 no-op,行曾丢失则本次补建——重投即自愈,任何一分支入队失败都仍返回 SettleFailed +// 交渠道再重投。 func (g *Gateway) settleRenewal(ctx context.Context, method string, ev *provider.PaidEvent) (SettleResult, error) { // channel = 触发本次回调的 method(与 markSubscriptionPastDue/recordChargeback 同式); // 订阅诞生时 Subscription.Channel 落的正是 att.Channel=method,查询须对齐,不能硬编码字面量 @@ -174,18 +184,27 @@ func (g *Gateway) settleRenewal(ctx context.Context, method string, ev *provider if _, err := g.subs.Activate(sub.SubID, &nextEnd); err != nil { return SettleFailed, err } - if !created || sub.BizSystem == "" { - return SettleDuplicate, nil // 重投 / 独立收款 + // 无论 created 与否都尝试入队(见函数注释的入队纪律);独立收款(BizSystem=="")无下游可发,跳过。 + if sub.BizSystem != "" { + if err := g.enqueueRenewed(sub, renewalNo, ev, paidAt); err != nil { + return SettleFailed, err // 已建单的事实不回滚(幂等键护着);渠道重投时补入队 + } } - if err := g.webhook.Enqueue(renewalNo, sub.BizSystem, EvtSubscriptionRenewed, "", map[string]any{ + if !created { + return SettleDuplicate, nil // 重投:订单/attempt 已在(幂等 no-op),outbox 已补齐 + } + return SettleProcessed, nil // 首过(含无业务方的独立收款,对齐 enqueuePaymentSucceeded 语义) +} + +// enqueueRenewed 组 subscription.renewed 领域 payload 并幂等入队(抽出供 settleRenewal 的 +// created/duplicate 两分支共用,避免复制两份 payload 构造)。 +func (g *Gateway) enqueueRenewed(sub *model.Subscription, renewalNo string, ev *provider.PaidEvent, paidAt time.Time) error { + return g.webhook.Enqueue(renewalNo, sub.BizSystem, EvtSubscriptionRenewed, "", map[string]any{ "event_type": EvtSubscriptionRenewed, "out_trade_no": renewalNo, "sub_id": sub.SubID, "biz_system": sub.BizSystem, "biz_ref": sub.BizRef, "product_biz_code": sub.BizCode, "amount_minor": ev.PaidAmountMinor, "currency": ev.PaidCurrency, "channel": sub.Channel, "paid_at": paidAt.Format(time.RFC3339), - }); err != nil { - return SettleFailed, err - } - return SettleProcessed, nil + }) } func (g *Gateway) recordChargeback(ctx context.Context, method string, ev *provider.PaidEvent) (SettleResult, error) { diff --git a/internal/gateway/subscription.go b/internal/gateway/subscription.go index e9c9a6f..51a8aa2 100644 --- a/internal/gateway/subscription.go +++ b/internal/gateway/subscription.go @@ -195,20 +195,24 @@ func (g *Gateway) settleSubscriptionCanceled(ctx context.Context, method string, } // finalizeCanceled 是取消的唯一落地点(主动取消 API + 入站 webhook 两路收敛于此): -// MarkCanceled 幂等翻转 canceled;仅首次真正翻转(flipped)且有业务方(BizSystem 非空) -// 才入队 subscription.canceled——重投/独立收款不重复发。 +// MarkCanceled 幂等翻转 canceled,随后(有业务方时)入队 subscription.canceled。 +// +// 入队不按 flipped 分叉(P8 Task4 审计发现的同型反纪律,同 commit 一并修):若第一次翻转 +// 成功但入队瞬时失败,调用方拿到 err 后重投——此时 MarkCanceled 已是"已取消"故 flipped= +// false,若像旧实现那样 `!flipped→return nil` 直接跳过,outbox 永远补不上这一行,通知永久 +// 丢失。改为无论 flipped 与否都尝试入队,outbox 唯一键 ON CONFLICT DO NOTHING 天然幂等: +// 行已存在则 no-op,行曾丢失则本次补建。canceled 每订阅只发生一次,不存在"同订阅多次 canceled +// 真被吞"的歧义(不同于 past_due 可能有多次不同 invoice 失败),无条件入队是安全的。 // // outbox 键取真实首购单号 sub.OutTradeNo(不用合成后缀键):该单已 paid,Notifier 投递门禁 // (orderPaid)天然放行;event_type=subscription.canceled 与该单已有的 payment.succeeded / // subscription.created 行不同 event_type,唯一键 (out_trade_no,event_type,refund_id) 不撞。 -// canceled 每订阅只发生一次,不存在"同订阅多次 canceled 被吞"的问题(不同于 past_due)。 func (g *Gateway) finalizeCanceled(sub *model.Subscription) error { - flipped, err := g.subs.MarkCanceled(sub.SubID) - if err != nil { + if _, err := g.subs.MarkCanceled(sub.SubID); err != nil { return err } - if !flipped || sub.BizSystem == "" { - return nil // 已取消(重投)/ 独立收款无业务方回调 + if sub.BizSystem == "" { + return nil // 独立收款无业务方回调 } return g.webhook.Enqueue(sub.OutTradeNo, sub.BizSystem, EvtSubscriptionCanceled, "", map[string]any{ "event_type": EvtSubscriptionCanceled, "out_trade_no": sub.OutTradeNo, "sub_id": sub.SubID, @@ -221,9 +225,15 @@ func (g *Gateway) finalizeCanceled(sub *model.Subscription) error { // 当前 active 时翻转,已 past_due/已 canceled 不动),入队 subscription.past_due 供业务方提醒 // 用户换卡。下期 invoice.paid 成功由 settleRenewal 的 Activate 自动恢复 active。 // +// 入队不按 flipped 分叉(P8 Task4 审计发现的同型反纪律,同 commit 一并修):首次翻转成功但 +// 入队瞬时失败时,重投这个同一事件会命中 flipped=false(已是 past_due),旧实现直接 return +// 会让这条 past_due 通知永久丢失。改为无论 flipped 与否都尝试入队,outbox 唯一键 ON CONFLICT +// DO NOTHING 天然幂等——重投即自愈。 +// // outbox 键取真实首购单号 sub.OutTradeNo(理由同 finalizeCanceled):Task5 最小可行接受 -// "同订阅多次 past_due 只报首次"(第二次失败的行撞 (out_trade_no,event_type) 唯一键、 -// EnqueueDelivery 幂等 no-op),Task 7 视需要再引入合成键 + Notifier 门禁旁路。 +// "同订阅多次 past_due(不同 invoice 各自失败)只报首次"这一语义限制不变——第二次真实失败的 +// 行同样撞 (out_trade_no,event_type) 唯一键、幂等 no-op,Task 7 视需要再引入合成键 + Notifier +// 门禁旁路;这里修的只是"同一事件重投也补不回入队"这一段自愈缺口,不扩大既有限制的范围。 func (g *Gateway) markSubscriptionPastDue(ctx context.Context, method string, ev *provider.PaidEvent) (SettleResult, error) { sub, err := g.subs.GetByProviderRef(method, ev.SubscriptionRef) if err != nil { @@ -236,15 +246,17 @@ func (g *Gateway) markSubscriptionPastDue(ctx context.Context, method string, ev if err != nil { return SettleFailed, err } - if !flipped || sub.BizSystem == "" { - return SettleDuplicate, nil + if sub.BizSystem != "" { + if err := g.webhook.Enqueue(sub.OutTradeNo, sub.BizSystem, EvtSubscriptionPastDue, "", map[string]any{ + "event_type": EvtSubscriptionPastDue, "out_trade_no": sub.OutTradeNo, "sub_id": sub.SubID, + "biz_system": sub.BizSystem, "biz_ref": sub.BizRef, "product_biz_code": sub.BizCode, + "failed_at": time.Now().Format(time.RFC3339), + }); err != nil { + return SettleFailed, err + } } - if err := g.webhook.Enqueue(sub.OutTradeNo, sub.BizSystem, EvtSubscriptionPastDue, "", map[string]any{ - "event_type": EvtSubscriptionPastDue, "out_trade_no": sub.OutTradeNo, "sub_id": sub.SubID, - "biz_system": sub.BizSystem, "biz_ref": sub.BizRef, "product_biz_code": sub.BizCode, - "failed_at": time.Now().Format(time.RFC3339), - }); err != nil { - return SettleFailed, err + if !flipped { + return SettleDuplicate, nil } return SettleProcessed, nil } diff --git a/internal/gateway/subscription_test.go b/internal/gateway/subscription_test.go index 5f016ba..3d57400 100644 --- a/internal/gateway/subscription_test.go +++ b/internal/gateway/subscription_test.go @@ -685,3 +685,120 @@ func TestMarkSubscriptionPastDueDuplicateInvoiceNotDoubleEnqueued(t *testing.T) t.Fatalf("webhook calls after replay = %d, want still 1: %+v", len(spy.calls), spy.calls) } } + +// TestSettleRenewalEnqueueFailureThenRetryRecovers 复现 Important 发现:settleRenewal 先 +// CreateRenewalPaid(建单已 paid,单步无 pending 中间态)后 Enqueue——首次入队失败(瞬时)后, +// invoice.paid 重投走 created=false 的 duplicate 分支必须仍尝试入队(outbox 唯一键幂等, +// 行不存在则补建),否则 subscription.renewed 永久丢失(SyncPendingAttempts 救不了:续费 +// attempt 生来就是 AttemptPaid,不在 pending 轮询范围)。 +// 断言:①首次入队失败 → SettleFailed+err,订单已建为 paid(幂等键护着,不回滚)②同一 +// invoice.paid 重投 → subscription.renewed 最终恰入队一次 ③renewal order 不因重投双铸。 +func TestSettleRenewalEnqueueFailureThenRetryRecovers(t *testing.T) { + g, fp, spy, orders, _ := newSubGateway(t) + ctx := context.Background() + activateSub(t, g, fp) + spy.calls = nil + + renewalRaw, err := json.Marshal(provider.PaidEvent{ + Kind: provider.EventSubscriptionRenewal, SubscriptionRef: "sub_new", InvoiceRef: "in_flaky", + PaidAmountMinor: 2999, PaidCurrency: "USD", Status: provider.PaidSucceeded, + }) + if err != nil { + t.Fatalf("marshal renewal event: %v", err) + } + + spy.failNext = true + result, err := g.HandleCallback(ctx, "substripe", provider.CallbackInput{Raw: renewalRaw}) + if err == nil || result != gateway.SettleFailed { + t.Fatalf("首次入队失败应 SettleFailed+err, got %v, %v", result, err) + } + allOrders, err := orders.ListOrders("pangolin", "u-1", 10) + if err != nil { + t.Fatalf("ListOrders: %v", err) + } + renewalCount := 0 + for i := range allOrders { + if strings.Contains(allOrders[i].OutTradeNo, "-r-") { + renewalCount++ + if allOrders[i].Status != model.OrderPaidV2 { + t.Fatalf("renewal order 应已建为 paid(幂等键护着,不因入队失败回滚), got %+v", allOrders[i]) + } + } + } + if renewalCount != 1 { + t.Fatalf("renewal order count after first(失败) attempt = %d, want 1", renewalCount) + } + if len(spy.calls) != 0 { + t.Fatalf("入队失败不应留下 webhook 记录, got %+v", spy.calls) + } + + // 渠道重投同一 invoice.paid(Stripe 拿不到 200 会重投):outbox 补建自愈。 + result2, err := g.HandleCallback(ctx, "substripe", provider.CallbackInput{Raw: renewalRaw}) + if err != nil { + t.Fatalf("HandleCallback retry: %v", err) + } + _ = result2 // duplicate(created=false,订单已在)——本用例只关心自愈,不断言具体 result 值 + if len(spy.calls) != 1 { + t.Fatalf("重投后 subscription.renewed 应恰入队一次(自愈), got %d: %+v", len(spy.calls), spy.calls) + } + if spy.calls[0]["event_type"] != gateway.EvtSubscriptionRenewed { + t.Fatalf("event_type = %v, want %s", spy.calls[0]["event_type"], gateway.EvtSubscriptionRenewed) + } + + allOrdersAfter, err := orders.ListOrders("pangolin", "u-1", 10) + if err != nil { + t.Fatalf("ListOrders after retry: %v", err) + } + renewalCountAfter := 0 + for i := range allOrdersAfter { + if strings.Contains(allOrdersAfter[i].OutTradeNo, "-r-") { + renewalCountAfter++ + } + } + if renewalCountAfter != 1 { + t.Fatalf("renewal order count after retry = %d, want still 1(不双铸)", renewalCountAfter) + } +} + +// TestSettleRenewalNoBizSystemStillProcessed 独立收款(无业务方回调,BizSystem=="")的续费首过 +// 应与 enqueuePaymentSucceeded 的"无业务方=跳过入队但仍 processed"语义对齐,不能误判 duplicate +// (created=true 是真正的首次成交,只是没有下游 webhook 可发)。 +func TestSettleRenewalNoBizSystemStillProcessed(t *testing.T) { + g, fp, spy, _, subs := newSubGateway(t) + ctx := context.Background() + _, err := g.CreateSubscription(ctx, gateway.CreateSubscriptionInput{ + SKU: "pro_monthly", Method: "substripe", // BizSystem/BizRef 留空 = 独立收款 + }) + if err != nil { + t.Fatalf("CreateSubscription: %v", err) + } + activateRaw, err := json.Marshal(provider.PaidEvent{ + Kind: provider.EventPayment, ProviderRef: fp.sessionRef, Status: provider.PaidSucceeded, + PaidAmountMinor: 2999, PaidCurrency: "USD", SubscriptionRef: "sub_new", + }) + if err != nil { + t.Fatalf("marshal activate event: %v", err) + } + if _, err := g.HandleCallback(ctx, "substripe", provider.CallbackInput{Raw: activateRaw}); err != nil { + t.Fatalf("activate: %v", err) + } + sub, err := subs.GetByProviderRef("substripe", "sub_new") + if err != nil || sub.BizSystem != "" { + t.Fatalf("subscription = %+v, %v, want BizSystem empty(独立收款)", sub, err) + } + + renewalRaw, err := json.Marshal(provider.PaidEvent{ + Kind: provider.EventSubscriptionRenewal, SubscriptionRef: "sub_new", InvoiceRef: "in_indep_1", + PaidAmountMinor: 2999, PaidCurrency: "USD", Status: provider.PaidSucceeded, + }) + if err != nil { + t.Fatalf("marshal renewal event: %v", err) + } + result, err := g.HandleCallback(ctx, "substripe", provider.CallbackInput{Raw: renewalRaw}) + if err != nil || result != gateway.SettleProcessed { + t.Fatalf("独立收款续费首过应 processed(非 duplicate), got %v, %v", result, err) + } + if len(spy.calls) != 0 { + t.Fatalf("独立收款不应入队 webhook, got %+v", spy.calls) + } +}