From c73f7ffaa9f79f1364be095e41b9ee10f1889b92 Mon Sep 17 00:00:00 2001 From: wangjia <809946525@qq.com> Date: Fri, 10 Jul 2026 17:25:21 +0800 Subject: [PATCH] =?UTF-8?q?feat(v2):=20webhook=20=E6=8A=95=E9=80=92?= =?UTF-8?q?=E7=A1=AC=E5=8C=96=E2=80=94=E2=80=94=E6=8C=87=E6=95=B0=E9=80=80?= =?UTF-8?q?=E9=81=BF=20+=20=E6=9C=80=E5=A4=A7=E6=AC=A1=E6=95=B0=E6=AD=BB?= =?UTF-8?q?=E4=BF=A1=20+=20=E5=91=8A=E8=AD=A6=E9=92=A9=E5=AD=90(=E6=9B=BF?= =?UTF-8?q?=E8=A3=B8=2060s=20=E7=8C=9B=E6=95=B2)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/model/webhook_delivery.go | 5 ++ internal/store/webhook.go | 46 +++++++++++++- internal/store/webhook_test.go | 33 ++++++++++ internal/webhook/notifier.go | 97 ++++++++++++++++++++++++++---- internal/webhook/notifier_test.go | 62 ++++++++++++++++++- 5 files changed, 227 insertions(+), 16 deletions(-) diff --git a/internal/model/webhook_delivery.go b/internal/model/webhook_delivery.go index 56ebe91..1a8bbdc 100644 --- a/internal/model/webhook_delivery.go +++ b/internal/model/webhook_delivery.go @@ -1,5 +1,7 @@ package model +import "time" + // WebhookDelivery 是 pay→业务方 webhook 的 outbox(v2)。unique(out_trade_no,event_type,refund_id) // 保证同一订单同一事件(同一退款单)只入队一次(幂等);后台 Notifier 扫 Delivered=false 重试兜底。 type WebhookDelivery struct { @@ -12,4 +14,7 @@ type WebhookDelivery struct { Delivered bool `gorm:"index;default:false" json:"delivered"` Attempts int `json:"attempts"` LastError string `gorm:"size:255" json:"last_error,omitempty"` + + Dead bool `gorm:"index;default:false" json:"dead"` // 达最大次数放弃投递(死信),需人工/对账介入 + NextAttemptAt *time.Time `gorm:"index" json:"next_attempt_at,omitempty"` // 指数退避的下次可投时刻;nil=立即可投 } diff --git a/internal/store/webhook.go b/internal/store/webhook.go index f253898..429089b 100644 --- a/internal/store/webhook.go +++ b/internal/store/webhook.go @@ -2,6 +2,7 @@ package store import ( "fmt" + "time" "unicode/utf8" "gorm.io/gorm" @@ -67,7 +68,8 @@ func truncateUTF8(s string, n int) string { } // MarkFailed increments attempts and records the last error, leaving the row -// undelivered for the next retry sweep. +// undelivered for the next retry sweep. Superseded in production by +// ScheduleRetry/MarkDead (退避感知);kept for existing callers/tests. func (s *WebhookStore) MarkFailed(id uint64, errMsg string) error { errMsg = truncateUTF8(errMsg, 255) if err := s.db.Model(&model.WebhookDelivery{}).Where("id = ?", id). @@ -79,3 +81,45 @@ func (s *WebhookStore) MarkFailed(id uint64, errMsg string) error { } return nil } + +// ListDeliverable 取「未投递、未死信、且退避到点(next_attempt_at NULL 或 <= now)」的行。 +func (s *WebhookStore) ListDeliverable(now time.Time, limit int) ([]WebhookDeliveryRow, error) { + if limit <= 0 || limit > 200 { + limit = 50 + } + var out []WebhookDeliveryRow + if err := s.db. + Where("delivered = ? AND dead = ? AND (next_attempt_at IS NULL OR next_attempt_at <= ?)", false, false, now). + Order("id ASC").Limit(limit).Find(&out).Error; err != nil { + return nil, fmt.Errorf("store.ListDeliverable: %w", err) + } + return out, nil +} + +// ScheduleRetry 记一次失败并排下次重试:attempts+1、last_error、next_attempt_at=nextAt。 +func (s *WebhookStore) ScheduleRetry(id uint64, errMsg string, nextAt time.Time) error { + errMsg = truncateUTF8(errMsg, 255) + if err := s.db.Model(&model.WebhookDelivery{}).Where("id = ?", id). + Updates(map[string]any{ + "attempts": gorm.Expr("attempts + 1"), + "last_error": errMsg, + "next_attempt_at": nextAt, + }).Error; err != nil { + return fmt.Errorf("store.ScheduleRetry: %w", err) + } + return nil +} + +// MarkDead 达最大次数后放弃:attempts+1、dead=true、last_error。行留库供人工/对账排查。 +func (s *WebhookStore) MarkDead(id uint64, errMsg string) error { + errMsg = truncateUTF8(errMsg, 255) + if err := s.db.Model(&model.WebhookDelivery{}).Where("id = ?", id). + Updates(map[string]any{ + "attempts": gorm.Expr("attempts + 1"), + "dead": true, + "last_error": errMsg, + }).Error; err != nil { + return fmt.Errorf("store.MarkDead: %w", err) + } + return nil +} diff --git a/internal/store/webhook_test.go b/internal/store/webhook_test.go index 3c9cbef..caa115a 100644 --- a/internal/store/webhook_test.go +++ b/internal/store/webhook_test.go @@ -2,6 +2,7 @@ package store_test import ( "testing" + "time" "unicode/utf8" "github.com/wangjia/pay/internal/model" @@ -104,3 +105,35 @@ func TestEnqueueDeliveryRefundIDUnique(t *testing.T) { t.Fatalf("undelivered rows = %d want 3: %+v", len(rows), rows) } } + +func TestWebhookScheduleRetryAndDead(t *testing.T) { + ws := store.NewWebhookStore(model.OpenTestDB(t)) + now := time.Date(2026, 7, 10, 12, 0, 0, 0, time.UTC) + _ = ws.EnqueueDelivery("PAY-1", "pangolin", "payment.succeeded", "", `{"x":1}`) + + // 刚入队:next_attempt_at NULL → 立即可投。 + rows, _ := ws.ListDeliverable(now, 10) + if len(rows) != 1 { + t.Fatalf("新单应可投, got %d", len(rows)) + } + id := rows[0].ID + + // 排下一次重试到 now+30s:此刻不可投,过点可投。 + if err := ws.ScheduleRetry(id, "http 500", now.Add(30*time.Second)); err != nil { + t.Fatalf("schedule: %v", err) + } + if r, _ := ws.ListDeliverable(now, 10); len(r) != 0 { + t.Fatalf("退避窗内不应可投, got %d", len(r)) + } + if r, _ := ws.ListDeliverable(now.Add(31*time.Second), 10); len(r) != 1 || r[0].Attempts != 1 { + t.Fatalf("过退避点应可投且 attempts=1, got %+v", r) + } + + // 标死信:不再出现在可投集。 + if err := ws.MarkDead(id, "gave up"); err != nil { + t.Fatalf("markdead: %v", err) + } + if r, _ := ws.ListDeliverable(now.Add(time.Hour), 10); len(r) != 0 { + t.Fatalf("死信不应可投, got %d", len(r)) + } +} diff --git a/internal/webhook/notifier.go b/internal/webhook/notifier.go index 3b948fa..11b1b15 100644 --- a/internal/webhook/notifier.go +++ b/internal/webhook/notifier.go @@ -31,15 +31,85 @@ type BizConfigFunc func(system string) (config.BizSystemConfig, bool) type OrderPaidFunc func(outTradeNo string) (bool, error) type Notifier struct { - deliveries *store.WebhookStore - bizConfig BizConfigFunc - orderPaid OrderPaidFunc - client *http.Client + deliveries *store.WebhookStore + bizConfig BizConfigFunc + orderPaid OrderPaidFunc + client *http.Client + now func() time.Time + maxAttempts int + baseBackoff time.Duration + maxBackoff time.Duration + alert func(d *store.WebhookDeliveryRow, reason string) } -func NewNotifier(ws *store.WebhookStore, bizConfig BizConfigFunc, orderPaid OrderPaidFunc) *Notifier { - return &Notifier{deliveries: ws, bizConfig: bizConfig, orderPaid: orderPaid, - client: &http.Client{Timeout: 10 * time.Second}} +// Option customizes a Notifier's retry/backoff/alerting behavior (functional options). +type Option func(*Notifier) + +// WithClock overrides the time source (tests inject a fake clock to drive backoff windows). +func WithClock(f func() time.Time) Option { return func(n *Notifier) { n.now = f } } + +// WithMaxAttempts sets how many failed attempts before a delivery is marked dead. +func WithMaxAttempts(m int) Option { return func(n *Notifier) { n.maxAttempts = m } } + +// WithBaseBackoff sets the base duration for exponential backoff (attempt 1). +func WithBaseBackoff(d time.Duration) Option { return func(n *Notifier) { n.baseBackoff = d } } + +// WithMaxBackoff caps the exponential backoff duration. +func WithMaxBackoff(d time.Duration) Option { return func(n *Notifier) { n.maxBackoff = d } } + +// WithAlerter overrides the dead-letter alert hook (default logs). +func WithAlerter(a func(d *store.WebhookDeliveryRow, reason string)) Option { + return func(n *Notifier) { n.alert = a } +} + +// NewNotifier builds a Notifier. The original 3-arg call form keeps compiling +// (opts is variadic); pass Option values to customize clock/backoff/alerting. +func NewNotifier(ws *store.WebhookStore, bizConfig BizConfigFunc, orderPaid OrderPaidFunc, opts ...Option) *Notifier { + n := &Notifier{ + deliveries: ws, bizConfig: bizConfig, orderPaid: orderPaid, + client: &http.Client{Timeout: 10 * time.Second}, + now: time.Now, + maxAttempts: 12, // ~ 覆盖数小时退避后放弃(见 backoffFor 封顶) + baseBackoff: 30 * time.Second, // 首次失败退避基 + maxBackoff: time.Hour, // 单次退避封顶 + } + for _, o := range opts { + o(n) + } + if n.alert == nil { + n.alert = func(d *store.WebhookDeliveryRow, reason string) { + log.Printf("[webhook][死信] out_trade_no=%s biz=%s event=%s attempts=%d 放弃投递: %s", + d.OutTradeNo, d.BizSystem, d.EventType, d.Attempts, reason) + } + } + return n +} + +// backoffFor 计算第 attempts 次失败后的退避:base·2^(attempts-1),封顶 maxBackoff。 +// attempts≥1;移位用循环倍增且封顶,防溢出。 +func (n *Notifier) backoffFor(attempts int) time.Duration { + d := n.baseBackoff + for i := 1; i < attempts; i++ { + d *= 2 + if d >= n.maxBackoff { + return n.maxBackoff + } + } + if d > n.maxBackoff { + return n.maxBackoff + } + return d +} + +// fail 统一失败分流:未达上限→退避重排;达上限→死信+告警。 +func (n *Notifier) fail(d *store.WebhookDeliveryRow, msg string) { + attempts := d.Attempts + 1 // 本次即将记的失败次数 + if attempts >= n.maxAttempts { + _ = n.deliveries.MarkDead(d.ID, msg) + n.alert(d, msg) + return + } + _ = n.deliveries.ScheduleRetry(d.ID, msg, n.now().Add(n.backoffFor(attempts))) } // Enqueue implements gateway.WebhookEnqueuer: serialize the domain payload and @@ -53,9 +123,10 @@ func (n *Notifier) Enqueue(outTradeNo, bizSystem, eventType, refundID string, da return n.deliveries.EnqueueDelivery(outTradeNo, bizSystem, eventType, refundID, string(body)) } -// DeliverPending flushes undelivered rows; returns how many succeeded this pass. +// DeliverPending flushes deliverable rows (未投递、未死信、退避到点); returns how +// many succeeded this pass. func (n *Notifier) DeliverPending(limit int) (int, error) { - rows, err := n.deliveries.ListUndelivered(limit) + rows, err := n.deliveries.ListDeliverable(n.now(), limit) if err != nil { return 0, err } @@ -80,7 +151,7 @@ func (n *Notifier) deliverOne(d *store.WebhookDeliveryRow) bool { } cfg, found := n.bizConfig(d.BizSystem) if !found || cfg.CallbackURL == "" { - _ = n.deliveries.MarkFailed(d.ID, "biz system not configured") + n.fail(d, "biz system not configured") return false } ts := strconv.FormatInt(time.Now().Unix(), 10) @@ -89,7 +160,7 @@ func (n *Notifier) deliverOne(d *store.WebhookDeliveryRow) bool { req, err := http.NewRequest(http.MethodPost, cfg.CallbackURL, bytes.NewReader([]byte(d.Payload))) if err != nil { - _ = n.deliveries.MarkFailed(d.ID, err.Error()) + n.fail(d, err.Error()) return false } req.Header.Set("Content-Type", "application/json") @@ -101,7 +172,7 @@ func (n *Notifier) deliverOne(d *store.WebhookDeliveryRow) bool { resp, err := n.client.Do(req) if err != nil { - _ = n.deliveries.MarkFailed(d.ID, err.Error()) + n.fail(d, err.Error()) return false } rb, _ := io.ReadAll(io.LimitReader(resp.Body, 4096)) @@ -111,7 +182,7 @@ func (n *Notifier) deliverOne(d *store.WebhookDeliveryRow) bool { _ = n.deliveries.MarkDelivered(d.ID) return true } - _ = n.deliveries.MarkFailed(d.ID, fmt.Sprintf("http %d: %s", resp.StatusCode, truncate(string(rb), 120))) + n.fail(d, fmt.Sprintf("http %d: %s", resp.StatusCode, truncate(string(rb), 120))) return false } diff --git a/internal/webhook/notifier_test.go b/internal/webhook/notifier_test.go index abcdb39..fb5ebaf 100644 --- a/internal/webhook/notifier_test.go +++ b/internal/webhook/notifier_test.go @@ -6,6 +6,7 @@ import ( "net/http" "net/http/httptest" "testing" + "time" "github.com/wangjia/pay/config" "github.com/wangjia/pay/internal/model" @@ -72,6 +73,60 @@ func TestNotifierDeliversSignedEvent(t *testing.T) { } } +// 业务方持续 500:每次失败按指数退避重排;时钟推进后才重投;达上限标死信 + 告警。 +func TestNotifierBackoffThenDeadWithAlert(t *testing.T) { + var hits int + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + hits++ + w.WriteHeader(http.StatusInternalServerError) + })) + defer srv.Close() + + clk := time.Date(2026, 7, 10, 12, 0, 0, 0, time.UTC) + nowFn := func() time.Time { return clk } + var alerted []string + ws := store.NewWebhookStore(model.OpenTestDB(t)) + n := webhook.NewNotifier(ws, + func(string) (config.BizSystemConfig, bool) { + return config.BizSystemConfig{CallbackURL: srv.URL, Secret: "x"}, true + }, + func(string) (bool, error) { return true, nil }, + webhook.WithClock(nowFn), + webhook.WithBaseBackoff(time.Second), + webhook.WithMaxBackoff(4*time.Second), + webhook.WithMaxAttempts(3), + webhook.WithAlerter(func(d *store.WebhookDeliveryRow, reason string) { alerted = append(alerted, d.OutTradeNo) }), + ) + _ = n.Enqueue("PAY-D", "pangolin", "payment.succeeded", "", map[string]any{"event_type": "payment.succeeded"}) + + // 尝试 1:失败 → attempts=1,退避到 +1s。 + if sent, _ := n.DeliverPending(10); sent != 0 || hits != 1 { + t.Fatalf("try1 sent=%d hits=%d", sent, hits) + } + // 退避窗内不投。 + if sent, _ := n.DeliverPending(10); sent != 0 || hits != 1 { + t.Fatalf("退避窗内不应再敲, hits=%d", hits) + } + // 推进越过退避;尝试 2 失败 → attempts=2,退避到 +2s。 + clk = clk.Add(2 * time.Second) + if sent, _ := n.DeliverPending(10); sent != 0 || hits != 2 { + t.Fatalf("try2 hits=%d", hits) + } + // 推进;尝试 3 失败 → attempts 达 maxAttempts(3)→ 死信 + 告警。 + clk = clk.Add(4 * time.Second) + if sent, _ := n.DeliverPending(10); sent != 0 || hits != 3 { + t.Fatalf("try3 hits=%d", hits) + } + if len(alerted) != 1 || alerted[0] != "PAY-D" { + t.Fatalf("死信应触发告警一次, got %+v", alerted) + } + // 已死信:无论时钟怎么走都不再投。 + clk = clk.Add(time.Hour) + if sent, _ := n.DeliverPending(10); sent != 0 || hits != 3 { + t.Fatalf("死信后不应再投, hits=%d", hits) + } +} + func TestNotifierRetriesOnFailure(t *testing.T) { var hits int srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { @@ -80,20 +135,23 @@ func TestNotifierRetriesOnFailure(t *testing.T) { })) defer srv.Close() + clk := time.Date(2026, 7, 10, 12, 0, 0, 0, time.UTC) ws := store.NewWebhookStore(model.OpenTestDB(t)) n := webhook.NewNotifier(ws, func(string) (config.BizSystemConfig, bool) { return config.BizSystemConfig{CallbackURL: srv.URL, Secret: "x"}, true - }, func(string) (bool, error) { return true, nil }) + }, func(string) (bool, error) { return true, nil }, + webhook.WithClock(func() time.Time { return clk }), + webhook.WithBaseBackoff(time.Second)) _ = n.Enqueue("PAY-3", "pangolin", "payment.succeeded", "", map[string]any{"event_type": "payment.succeeded"}) if sent, _ := n.DeliverPending(10); sent != 0 { t.Fatalf("失败不应算投递成功, got %d", sent) } - // 仍待投递,可被下一轮重试兜底 pend, _ := ws.ListUndelivered(10) if len(pend) != 1 || pend[0].Attempts != 1 { t.Fatalf("失败后应留队重试, got %+v", pend) } + clk = clk.Add(2 * time.Second) // 越过退避窗 if _, _ = n.DeliverPending(10); hits < 2 { t.Fatalf("应重试第二次, hits=%d", hits) }