feat(notify): generic handling of 429 in retrier

notify/notify_test.go

@@ -540,6 +540,115 @@ func TestRetryStageWithContextCanceled(t *testing.T) {
 	require.NotNil(t, resctx)
 }
 
+func TestRetryStageHonorsRetryAfter(t *testing.T) {
+	attempts := 0
+	i := Integration{
+		name: "test",
+		notifier: notifierFunc(func(ctx context.Context, alerts ...*types.Alert) (bool, error) {
+			attempts++
+			if attempts < 4 {
+				err := NewErrorWithReason(RateLimitedReason, errors.New("received 429 Too Many Requests"))
+				err.RetryAfter = 10 * time.Millisecond
+				return true, err
+			}
+			return false, nil
+		}),
+		rs: sendResolved(false),
+	}
+	r := NewRetryStage(i, "", NewMetrics(prometheus.NewRegistry(), featurecontrol.NoopFlags{}), eventrecorder.NopRecorder())
+
+	alerts := []*types.Alert{{
+		Alert: model.Alert{
+			EndsAt: time.Now().Add(time.Hour),
+		},
+	}}
+
+	// The default exponential backoff starts at 500ms after the first immediate
+	// attempt, so 4 attempts can only complete within this timeout when
+	// Retry-After is actually honored.
+	ctx, cancel := context.WithTimeout(context.Background(), 400*time.Millisecond)
+	defer cancel()
+	ctx = WithFiringAlerts(ctx, []uint64{0})
+
+	start := time.Now()
+	_, _, err := r.Exec(ctx, promslog.NewNopLogger(), alerts...)
+	elapsed := time.Since(start)
+	require.NoError(t, err)
+	require.Equal(t, 4, attempts)
+	require.GreaterOrEqual(t, elapsed, 30*time.Millisecond)
+	require.Less(t, elapsed, 350*time.Millisecond)
+}
+
+func TestRetryStageRecalculatesBackoffAfterRetryAfter(t *testing.T) {
+	attempts := 0
+	i := Integration{
+		name: "test",
+		notifier: notifierFunc(func(ctx context.Context, alerts ...*types.Alert) (bool, error) {
+			attempts++
+			switch attempts {
+			case 1:
+				err := NewErrorWithReason(RateLimitedReason, errors.New("received 429 Too Many Requests"))
+				err.RetryAfter = 10 * time.Millisecond
+				return true, err
+			case 2:
+				return true, errors.New("temporary failure")
+			default:
+				return false, nil
+			}
+		}),
+		rs: sendResolved(false),
+	}
+	r := NewRetryStage(i, "", NewMetrics(prometheus.NewRegistry(), featurecontrol.NoopFlags{}), eventrecorder.NopRecorder())
+
+	alerts := []*types.Alert{{
+		Alert: model.Alert{
+			EndsAt: time.Now().Add(time.Hour),
+		},
+	}}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond)
+	defer cancel()
+	ctx = WithFiringAlerts(ctx, []uint64{0})
+
+	_, _, err := r.Exec(ctx, promslog.NewNopLogger(), alerts...)
+	require.Error(t, err)
+	require.Contains(t, err.Error(), "notify retry canceled after 2 attempts")
+	require.Equal(t, 2, attempts)
+}
+
+func TestRetryStageWithoutRetryAfterUsesExponentialBackoff(t *testing.T) {
+	attempts := 0
+	i := Integration{
+		name: "test",
+		notifier: notifierFunc(func(ctx context.Context, alerts ...*types.Alert) (bool, error) {
+			attempts++
+			if attempts < 4 {
+				return true, NewErrorWithReason(RateLimitedReason, errors.New("received 429 Too Many Requests"))
+			}
+			return false, nil
+		}),
+		rs: sendResolved(false),
+	}
+	r := NewRetryStage(i, "", NewMetrics(prometheus.NewRegistry(), featurecontrol.NoopFlags{}), eventrecorder.NopRecorder())
+
+	alerts := []*types.Alert{{
+		Alert: model.Alert{
+			EndsAt: time.Now().Add(time.Hour),
+		},
+	}}
+
+	// Without Retry-After we should follow the default backoff, whose first
+	// interval after the initial attempt is far larger than this timeout.
+	ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond)
+	defer cancel()
+	ctx = WithFiringAlerts(ctx, []uint64{0})
+
+	_, _, err := r.Exec(ctx, promslog.NewNopLogger(), alerts...)
+	require.Error(t, err)
+	require.Contains(t, err.Error(), "notify retry canceled after 1 attempts")
+	require.Equal(t, 1, attempts)
+}
+
 func TestRetryStageNoResolved(t *testing.T) {
 	sent := []*types.Alert{}
 	i := Integration{

notify/retry_stage.go

@@ -112,8 +112,18 @@ func (r RetryStage) exec(ctx context.Context, l *slog.Logger, alerts ...*alert.A
 	// the ticker retries indefinitely until the context is canceled.
 	b := backoff.NewExponentialBackOff()
 
-	tick := backoff.NewTicker(b)
-	defer tick.Stop()
+	stopTimer := func(timer *time.Timer) {
+		if !timer.Stop() {
+			select {
+			case <-timer.C:
+			default:
+			}
+		}
+	}
+
+	// Fire immediately for the first attempt.
+	attemptTimer := time.NewTimer(0)
+	defer stopTimer(attemptTimer)
 
 	var (
 		i    = 0
@@ -147,7 +157,7 @@ func (r RetryStage) exec(ctx context.Context, l *slog.Logger, alerts ...*alert.A
 		}
 
 		select {
-		case <-tick.C:
+		case <-attemptTimer.C:
 			now := time.Now()
 			retry, err := r.integration.Notify(ctx, sent...)
 			i++
@@ -160,12 +170,35 @@ func (r RetryStage) exec(ctx context.Context, l *slog.Logger, alerts ...*alert.A
 					return ctx, alerts, fmt.Errorf("%s/%s: notify retry canceled due to unrecoverable error after %d attempts: %w", r.groupName, r.integration.String(), i, err)
 				}
 				if ctx.Err() == nil {
-					if iErr == nil || err.Error() != iErr.Error() {
-						// Log the error if the context isn't done and the error isn't the same as before.
-						l.Warn("Notify attempt failed, will retry later", "attempts", i, "err", err)
+					nextDelay := b.NextBackOff()
+
+					// Defensive: NextBackOff only returns Stop when MaxElapsedTime > 0,
+					// which we don't set, but guard against future config changes.
+					if nextDelay == backoff.Stop {
+						return ctx, nil, fmt.Errorf("%s/%s: notify retry stopped after %d attempts: %w", r.groupName, r.integration.String(), i, err)
+					}
+
+					var e *ErrorWithReason
+					if errors.As(err, &e) && e.Reason == RateLimitedReason && e.RetryAfter > 0 {
+						nextDelay = e.RetryAfter
+						l.Warn("Notify attempt failed, honoring Retry-After", "attempts", i, "retry_after", e.RetryAfter, "err", err)
+					} else {
+						// Subtract the attempt duration so the next attempt fires at
+						// approximately attempt_start + backoff, matching the behavior
+						// of the previous backoff.Ticker (which started counting from
+						// when the tick was consumed, not when the attempt finished).
+						nextDelay -= dur
+						if nextDelay < 0 {
+							nextDelay = 0
+						}
+						if iErr == nil || err.Error() != iErr.Error() {
+							// Log if context isn't done and the error differs from last time.
+							l.Warn("Notify attempt failed, will retry later", "attempts", i, "err", err)
+						}
 					}
-					// Save this error to be able to return the last seen error by an
-					// integration upon context timeout.
+
+					attemptTimer.Reset(nextDelay)
+					// Save the error to return the last seen error on context timeout.
 					iErr = err
 				}
 			} else {

notify/util.go

@@ -23,7 +23,9 @@ import (
 	"net/http"
 	"net/url"
 	"slices"
+	"strconv"
 	"strings"
+	"time"
 
 	commoncfg "github.com/prometheus/common/config"
 	"github.com/prometheus/common/version"
@@ -239,6 +241,28 @@ type Retrier struct {
 	RetryCodes []int
 }
 
+// parseRetryAfter parses the Retry-After header value, which can be either
+// a delay in seconds (integer) or an HTTP-date. Returns zero if absent or unparseable.
+func parseRetryAfter(h http.Header) time.Duration {
+	val := h.Get("Retry-After")
+	if val == "" {
+		return 0
+	}
+	// Try integer seconds first.
+	if secs, err := strconv.Atoi(val); err == nil {
+		return time.Duration(secs) * time.Second
+	}
+	// Try HTTP-date format.
+	if t, err := http.ParseTime(val); err == nil {
+		d := time.Until(t)
+		if d < 0 {
+			return 0
+		}
+		return d
+	}
+	return 0
+}
+
 // Check returns a boolean indicating whether the request should be retried
 // and an optional error if the request has failed. If body is not nil, it will
 // be included in the error message.
@@ -264,10 +288,62 @@ func (r *Retrier) Check(statusCode int, body io.Reader) (bool, error) {
 	return retry, errors.New(s)
 }
 
+// CheckResponse returns a boolean indicating whether the request should be
+// retried and an optional ErrorWithReason if the request has failed.
+// Unlike Check, it accepts the full *http.Response so it can parse the
+// Retry-After header on 429 responses and attach it to the returned error.
+func (r *Retrier) CheckResponse(resp *http.Response) (bool, error) {
+	if resp == nil {
+		return false, NewErrorWithReason(DefaultReason, errors.New("nil HTTP response"))
+	}
+
+	// 2xx responses are always successful.
+	if resp.StatusCode/100 == 2 {
+		return false, nil
+	}
+
+	s := fmt.Sprintf("unexpected status code %v", resp.StatusCode)
+	var details string
+	if r.CustomDetailsFunc != nil {
+		details = r.CustomDetailsFunc(resp.StatusCode, resp.Body)
+	} else {
+		details = readAll(resp.Body)
+	}
+	if details != "" {
+		s = fmt.Sprintf("%s: %s", s, details)
+	}
+
+	// Codes in RetryCodes are retriable regardless of class, except 429
+	// which is handled separately below to attach Retry-After.
+	if slices.Contains(r.RetryCodes, resp.StatusCode) && resp.StatusCode != http.StatusTooManyRequests {
+		return true, NewErrorWithReason(GetFailureReasonFromStatusCode(resp.StatusCode), errors.New(s))
+	}
+
+	if resp.StatusCode == http.StatusTooManyRequests {
+		e := NewErrorWithReason(RateLimitedReason, errors.New(s))
+		if d := parseRetryAfter(resp.Header); d > 0 {
+			e.RetryAfter = d
+		}
+		return true, e
+	}
+
+	if resp.StatusCode/100 == 4 {
+		return false, NewErrorWithReason(GetFailureReasonFromStatusCode(resp.StatusCode), errors.New(s))
+	}
+
+	// 5xx responses are always retried.
+	if resp.StatusCode/100 == 5 {
+		return true, NewErrorWithReason(ServerErrorReason, errors.New(s))
+	}
+
+	return false, NewErrorWithReason(GetFailureReasonFromStatusCode(resp.StatusCode), errors.New(s))
+}
+
 type ErrorWithReason struct {
 	Err error
 
-	Reason Reason
+	Reason     Reason
+	RetryAfter time.Duration
 }
 
 func NewErrorWithReason(reason Reason, err error) *ErrorWithReason {