feat(v2): webhook 投递硬化——指数退避 + 最大次数死信 + 告警钩子(替裸 60s 猛敲)
This commit is contained in:
wangjia
@@ -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
|
||||
}
|
||||
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user