fix(v2): 取消失配收敛——渠道已取消(哨兵)→本地补齐+事件,不再500

CancelSubscription 本地 active/past_due 但渠道已先行取消(dashboard 手工 / 竞态未消费的 deleted webhook)时,stripe adapter 原样透传渠道拒绝,handler 映成 500 cancel_failed——渠道取消这一事实明明已成立。新增 provider.ErrSubAlready Canceled 哨兵,stripe adapter 识别 resource_missing / "already been canceled" 两种真实 Stripe 错误形态并 wrap;gateway.CancelSubscription 命中哨兵后走与入站 webhook 相同的 finalizeCanceled 本地收敛,两路对同一终态天然幂等。同 re-review 顺手核掉同型缺口:onSubscriptionActivated 的 `!created→return nil` 在入队 subscription.created 之前短路,首次入队失败后 Stripe 重投会因 created=false 永久跳过入队——通知永久丢失。改为无论 created 与否都无条件入队,outbox 唯一键 ON CONFLICT DO NOTHING 天然幂等自愈,与 finalizeCanceled/markSubscriptionPastDue/settleRenewal 同一写法。 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_013nMthbVEmQquxBRKb9Fj8u
2026-07-10 18:42:20 +08:00
parent ff65a06bf2
commit 02b2fcfa41
6 changed files with 270 additions and 6 deletions
@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	"fmt"
 	"strings"
 	"testing"

@@ -454,6 +455,79 @@ func TestCancelSubscriptionIdempotentNoOp(t *testing.T) {
 	}
 }

+// TestCancelSubscriptionChannelAlreadyCanceledConverges 覆盖渠道已先行取消(dashboard 手工 /
+// 竞态下未消费的 deleted webhook 抢先落地)的失配场景:本地仍 active,adapter 打渠道拿到
+// provider.ErrSubAlreadyCanceled(此处用死脚手架 fp.cancelErr 模拟 stripe adapter 已 wrap 好的
+// 哨兵错误)——CancelSubscription 不应再报错(修复前会把这个 err 原样透传,handler 会映成 500
+// cancel_failed;修复后按哨兵走本地收敛),本地翻 canceled 且恰好入队一次 subscription.canceled。
+func TestCancelSubscriptionChannelAlreadyCanceledConverges(t *testing.T) {
+	g, fp, spy, _, subs := newSubGateway(t)
+	subID, providerSubRef, orderNo := activateSub(t, g, fp)
+	spy.calls = nil
+	fp.cancelErr = fmt.Errorf("%w: stripe simulated already-canceled", provider.ErrSubAlreadyCanceled)
+
+	if err := g.CancelSubscription(context.Background(), subID); err != nil {
+		t.Fatalf("CancelSubscription(渠道已取消场景应本地收敛,不应报错): %v", err)
+	}
+	if len(fp.cancelCalls) != 1 || fp.cancelCalls[0] != providerSubRef {
+		t.Fatalf("cancelCalls = %+v, want [%s](渠道确实被打过一次,只是回了'已取消'错误)", fp.cancelCalls, providerSubRef)
+	}
+	sub, err := subs.GetBySubID(subID)
+	if err != nil {
+		t.Fatalf("GetBySubID: %v", err)
+	}
+	if sub.Status != model.SubCanceled || sub.CanceledAt == nil {
+		t.Fatalf("subscription after cancel(渠道失配场景) = %+v, want locally canceled", sub)
+	}
+	if len(spy.calls) != 1 {
+		t.Fatalf("webhook calls = %d, want 1: %+v", len(spy.calls), spy.calls)
+	}
+	c := spy.calls[0]
+	if c["event_type"] != gateway.EvtSubscriptionCanceled || c["sub_id"] != subID || c["out_trade_no"] != orderNo {
+		t.Fatalf("cancel webhook payload = %+v, want event_type=%s sub_id=%s out_trade_no=%s",
+			c, gateway.EvtSubscriptionCanceled, subID, orderNo)
+	}
+}
+
+// TestCancelSubscriptionWebhookThenAPIConverge 覆盖反向竞态:入站 customer.subscription.deleted
+// 先到(本地先翻 canceled + 发一次 webhook),随后业务方/用户侧发起的主动取消 API 才姗姗来迟。
+// 此时本地已是终态,CancelSubscription 应在打渠道之前就早退(sub.Status==canceled 分支),
+// 绝不二次调用渠道、也绝不重复入队——即便 fp.cancelErr 被设成"渠道已取消"哨兵(模拟万一实现
+// 顺序有误、真打了渠道也不该出错),结果仍应是幂等 no-op。
+func TestCancelSubscriptionWebhookThenAPIConverge(t *testing.T) {
+	g, fp, spy, _, subs := newSubGateway(t)
+	ctx := context.Background()
+	subID, providerSubRef, _ := activateSub(t, g, fp)
+	spy.calls = nil
+	fp.cancelErr = fmt.Errorf("%w: stripe simulated already-canceled", provider.ErrSubAlreadyCanceled)
+
+	raw, err := json.Marshal(provider.PaidEvent{Kind: provider.EventSubscriptionCanceled, SubscriptionRef: providerSubRef})
+	if err != nil {
+		t.Fatalf("marshal: %v", err)
+	}
+	result, err := g.HandleCallback(ctx, "substripe", provider.CallbackInput{Raw: raw})
+	if err != nil || result != gateway.SettleProcessed {
+		t.Fatalf("HandleCallback deleted(webhook 先到) = %v, %v", result, err)
+	}
+	if len(spy.calls) != 1 {
+		t.Fatalf("webhook calls after inbound deleted = %d, want 1", len(spy.calls))
+	}
+
+	if err := g.CancelSubscription(ctx, subID); err != nil {
+		t.Fatalf("CancelSubscription(webhook 已先到,应幂等 no-op 不报错): %v", err)
+	}
+	if len(fp.cancelCalls) != 0 {
+		t.Fatalf("cancelCalls = %+v, want 0(本地已终态,不应再打渠道)", fp.cancelCalls)
+	}
+	if len(spy.calls) != 1 {
+		t.Fatalf("webhook calls after API cancel(webhook 先到之后) = %d, want still 1(不重复入队)", len(spy.calls))
+	}
+	sub, err := subs.GetBySubID(subID)
+	if err != nil || sub.Status != model.SubCanceled {
+		t.Fatalf("subscription = %+v, %v", sub, err)
+	}
+}
+
 // TestInboundSubscriptionDeletedConverges 覆盖入站 customer.subscription.deleted:同订阅
 // 幂等标 canceled + 入队一次;重投(渠道 webhook 常见重投)不重复发 webhook。
 func TestInboundSubscriptionDeletedConverges(t *testing.T) {
@@ -760,6 +834,74 @@ func TestSettleRenewalEnqueueFailureThenRetryRecovers(t *testing.T) {
 	}
 }

+// TestSubscriptionActivatedEnqueueFailureThenRetryRecovers 镜像
+// TestSettleRenewalEnqueueFailureThenRetryRecovers,覆盖同型缺口(T5 re-review 发现):
+// onSubscriptionActivated 先 subs.Create(先于入队幂等诞生订阅行)后 Enqueue subscription.created
+// ——首次入队失败(瞬时)后,checkout.session.completed 重投走 created=false 的分支必须仍尝试
+// 入队(outbox 唯一键幂等,行不存在则补建),否则 subscription.created 永久丢失(旧实现
+// `!created→return nil` 直接跳过,行永远补不上)。
+//
+// 用 spy.failOnEventType 只让 subscription.created 这一次入队失败(而非 Settle 内更早的
+// payment.succeeded)——同一 HandleCallback 里先后两次 Enqueue,只想复现"第二次失败"这个窗口。
+//
+// 断言:①首次:payment.succeeded 已入队(Settle 已 processed,订单已 paid),订阅行已幂等
+// 诞生 active(Create 先于 Enqueue),但 subscription.created 入队失败 → HandleCallback 整体
+// SettleFailed+err ②同一事件重投 → subscription.created 最终恰入队一次(payment.succeeded
+// 不因重投重发,outbox 唯一键幂等)③重投不双铸订阅行(SubID 幂等派生)。
+func TestSubscriptionActivatedEnqueueFailureThenRetryRecovers(t *testing.T) {
+	g, fp, spy, orders, subs := newSubGateway(t)
+	ctx := context.Background()
+	res, err := g.CreateSubscription(ctx, gateway.CreateSubscriptionInput{
+		SKU: "pro_monthly", Method: "substripe", BizSystem: "pangolin", BizRef: "u-1",
+	})
+	if err != nil {
+		t.Fatalf("CreateSubscription: %v", err)
+	}
+	raw, 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)
+	}
+
+	spy.failOnEventType = gateway.EvtSubscriptionCreated
+	result, err := g.HandleCallback(ctx, "substripe", provider.CallbackInput{Raw: raw})
+	if err == nil || result != gateway.SettleFailed {
+		t.Fatalf("subscription.created 入队失败应 SettleFailed+err, got %v, %v", result, err)
+	}
+	o, err := orders.GetOrder(res.OrderNo)
+	if err != nil || o.Status != model.OrderPaidV2 {
+		t.Fatalf("order after first(失败) attempt = %+v, %v, want paid(payment.succeeded 已先成功入队+翻转)", o, err)
+	}
+	subID := "SUB-" + res.OrderNo
+	sub, err := subs.GetBySubID(subID)
+	if err != nil || sub.Status != model.SubActive {
+		t.Fatalf("subscription after first(失败) attempt = %+v, %v, want already active(Create 先于 Enqueue,幂等诞生)", sub, err)
+	}
+	if len(spy.calls) != 1 || spy.calls[0]["event_type"] != "payment.succeeded" {
+		t.Fatalf("首次 calls = %+v, want 仅 payment.succeeded 一条(subscription.created 那次入队失败,未记入)", spy.calls)
+	}
+
+	// Stripe 拿不到 200 会重投同一 checkout.session.completed。
+	result2, err := g.HandleCallback(ctx, "substripe", provider.CallbackInput{Raw: raw})
+	if err != nil {
+		t.Fatalf("HandleCallback retry: %v", err)
+	}
+	_ = result2 // duplicate(created=false,订单/订阅已在)——本用例只关心自愈,不断言具体 result 值
+	if len(spy.calls) != 2 {
+		t.Fatalf("重投后 calls = %d, want 2(payment.succeeded 幂等 no-op 不重发 + subscription.created 补建恰一次): %+v", len(spy.calls), spy.calls)
+	}
+	if spy.calls[1]["event_type"] != gateway.EvtSubscriptionCreated || spy.calls[1]["sub_id"] != subID {
+		t.Fatalf("payload = %+v, want event_type=%s sub_id=%s", spy.calls[1], gateway.EvtSubscriptionCreated, subID)
+	}
+
+	subAfter, err := subs.GetBySubID(subID)
+	if err != nil || subAfter.Status != model.SubActive {
+		t.Fatalf("subscription after retry = %+v, %v", subAfter, err)
+	}
+}
+
 // TestSettleRenewalNoBizSystemStillProcessed 独立收款(无业务方回调,BizSystem=="")的续费首过
 // 应与 enqueuePaymentSucceeded 的"无业务方=跳过入队但仍 processed"语义对齐,不能误判 duplicate
 // (created=true 是真正的首次成交,只是没有下游 webhook 可发)。