feat(delivery): support maximum backoff duration

Signed-off-by: kahirokunn <okinakahiro@gmail.com>

Author: kahirokunn

docs/broker.md

@@ -157,6 +157,25 @@ spec:
     retry: 3
     backoffPolicy: exponential
     backoffDelay: PT1S
+    backoffMax: PT10M
+```
+
+`backoffMax` is useful when a subscriber may be unavailable for many retry
+attempts. In this example, the Broker filter never waits more than 10 minutes
+between normal delivery attempts. The field does not cap delays requested by a
+`Retry-After` response header; use `retryAfterMax` for those delays.
+
+Cluster operators must enable the experimental field in the Knative Eventing
+`config-features` ConfigMap before Trigger or Broker owners use it:
+
+```yaml
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: config-features
+  namespace: knative-eventing
+data:
+  delivery-backoff-max: enabled
 ```
 
 ## Configuration

pkg/broker/utils/delivery.go

@@ -29,6 +29,7 @@ func DeliveryIsSet(d *eventingduckv1.DeliverySpec) bool {
 		d.Timeout != nil ||
 		d.BackoffPolicy != nil ||
 		d.BackoffDelay != nil ||
+		d.BackoffMax != nil ||
 		d.RetryAfterMax != nil ||
 		d.Format != nil)
 }

pkg/broker/utils/delivery_test.go

@@ -27,6 +27,7 @@ import (
 
 func TestDeliveryIsSet(t *testing.T) {
 	r := int32(1)
+	backoffMax := "PT10M"
 	tests := []struct {
 		name string
 		spec *eventingduckv1.DeliverySpec
@@ -35,6 +36,7 @@ func TestDeliveryIsSet(t *testing.T) {
 		{name: "nil", spec: nil, want: false},
 		{name: "empty", spec: &eventingduckv1.DeliverySpec{}, want: false},
 		{name: "retry set", spec: &eventingduckv1.DeliverySpec{Retry: &r}, want: true},
+		{name: "backoff max set", spec: &eventingduckv1.DeliverySpec{BackoffMax: &backoffMax}, want: true},
 		{name: "dls set", spec: &eventingduckv1.DeliverySpec{DeadLetterSink: &duckv1.Destination{}}, want: true},
 	}
 	for _, tc := range tests {

pkg/channel/jetstream/dispatcher/message_dispatcher_test.go

@@ -560,15 +560,10 @@ func TestDispatchMessage(t *testing.T) {
 			var retryConfig kncloudevents.RetryConfig
 			if tc.delivery == nil {
 				retryConfig = kncloudevents.NoRetries()
-				backoffLinear := v1.BackoffPolicyLinear
-				retryConfig.BackoffPolicy = &backoffLinear
-				backoffDelay := "5s"
-				retryConfig.BackoffDelay = &backoffDelay
 			} else {
-				retryConfig = kncloudevents.RetryConfig{
-					RetryMax:      int(*tc.delivery.Retry),
-					BackoffPolicy: tc.delivery.BackoffPolicy,
-					BackoffDelay:  tc.delivery.BackoffDelay,
+				retryConfig, err = kncloudevents.RetryConfigFromDeliverySpec(*tc.delivery)
+				if err != nil {
+					t.Fatal(err)
 				}
 			}
 

pkg/channel/jetstream/utils/consumerconfig.go

@@ -17,15 +17,12 @@ limitations under the License.
 package utils
 
 import (
-	"math"
 	"time"
 
 	"github.com/nats-io/nats.go/jetstream"
 
 	"github.com/nats-io/nats.go"
-	"github.com/rickb777/date/period"
 	"knative.dev/eventing-natss/pkg/apis/messaging/v1alpha1"
-	v1 "knative.dev/eventing/pkg/apis/duck/v1"
 	"knative.dev/eventing/pkg/kncloudevents"
 )
 
@@ -111,31 +108,8 @@ func CalcRequestTimeout(numDelivered int, ackWait time.Duration) time.Duration {
 }
 
 func CalculateNakDelayForRetryNumber(attemptNum int, config *kncloudevents.RetryConfig) time.Duration {
-	backoff, backoffDelay := parseBackoffFuncAndDelay(config)
-	return backoff(attemptNum, backoffDelay)
-}
-
-type backoffFunc func(attemptNum int, delayDuration time.Duration) time.Duration
-
-func LinearBackoff(attemptNum int, delayDuration time.Duration) time.Duration {
-	return delayDuration * time.Duration(attemptNum)
-}
-
-func ExpBackoff(attemptNum int, delayDuration time.Duration) time.Duration {
-	return delayDuration * time.Duration(math.Exp2(float64(attemptNum)))
-}
-
-func parseBackoffFuncAndDelay(config *kncloudevents.RetryConfig) (backoffFunc, time.Duration) {
-	var backoff backoffFunc
-	switch *config.BackoffPolicy {
-	case v1.BackoffPolicyExponential:
-		backoff = ExpBackoff
-	case v1.BackoffPolicyLinear:
-		backoff = LinearBackoff
+	if config == nil || config.Backoff == nil {
+		return 0
 	}
-	// it should be validated at this point
-	delay, _ := period.Parse(*config.BackoffDelay)
-	backoffDelay, _ := delay.Duration()
-
-	return backoff, backoffDelay
+	return config.Backoff(attemptNum, nil)
 }

pkg/channel/jetstream/utils/consumerconfig_test.go

@@ -0,0 +1,68 @@
+/*
+Copyright 2026 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package utils
+
+import (
+	"testing"
+	"time"
+
+	eventingduckv1 "knative.dev/eventing/pkg/apis/duck/v1"
+	"knative.dev/eventing/pkg/kncloudevents"
+)
+
+func TestCalculateNakDelayForRetryNumber(t *testing.T) {
+	delay := "PT1S"
+	backoffMax := "PT10S"
+	exponential := eventingduckv1.BackoffPolicyExponential
+	linear := eventingduckv1.BackoffPolicyLinear
+
+	newConfig := func(policy eventingduckv1.BackoffPolicyType, max *string) *kncloudevents.RetryConfig {
+		t.Helper()
+		config, err := kncloudevents.RetryConfigFromDeliverySpec(eventingduckv1.DeliverySpec{
+			BackoffPolicy: &policy,
+			BackoffDelay:  &delay,
+			BackoffMax:    max,
+		})
+		if err != nil {
+			t.Fatal(err)
+		}
+		return &config
+	}
+
+	tests := []struct {
+		name    string
+		attempt int
+		config  *kncloudevents.RetryConfig
+		want    time.Duration
+	}{
+		{name: "nil config", attempt: 1, want: 0},
+		{name: "config without backoff", attempt: 1, config: &kncloudevents.RetryConfig{}, want: 0},
+		{name: "first exponential retry", attempt: 1, config: newConfig(exponential, &backoffMax), want: 2 * time.Second},
+		{name: "uncapped exponential retry", attempt: 4, config: newConfig(exponential, nil), want: 16 * time.Second},
+		{name: "capped exponential retry", attempt: 4, config: newConfig(exponential, &backoffMax), want: 10 * time.Second},
+		{name: "capped linear retry", attempt: 20, config: newConfig(linear, &backoffMax), want: 10 * time.Second},
+		{name: "huge retry stays capped", attempt: int(^uint(0) >> 1), config: newConfig(exponential, &backoffMax), want: 10 * time.Second},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := CalculateNakDelayForRetryNumber(tt.attempt, tt.config); got != tt.want {
+				t.Errorf("CalculateNakDelayForRetryNumber() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}