test(delivery): cover backoff maximum through triggers

pkg/kncloudevents/retries_test.go

@@ -22,10 +22,12 @@ import (
 	"net/http"
 	"strconv"
 	"testing"
+	"testing/synctest"
 	"time"
 
 	"github.com/rickb777/date/period"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 	"k8s.io/utils/pointer"
 	"knative.dev/pkg/ptr"
 
@@ -184,6 +186,93 @@ func TestRetryConfigBackoffMax(t *testing.T) {
 	}
 }
 
+func TestDoWithRetriesBackoffMax(t *testing.T) {
+	tests := []struct {
+		name          string
+		delay         string
+		backoffMax    *string
+		retryAfter    string
+		retryAfterMax *string
+		wantTimes     []time.Duration
+	}{
+		{
+			name:       "exponential backoff is capped",
+			delay:      "PT1S",
+			backoffMax: ptr.String("PT2S"),
+			wantTimes:  []time.Duration{0, time.Second, 3 * time.Second, 5 * time.Second, 7 * time.Second},
+		},
+		{
+			name:      "exponential backoff without a maximum",
+			delay:     "PT1S",
+			wantTimes: []time.Duration{0, time.Second, 3 * time.Second, 7 * time.Second, 15 * time.Second},
+		},
+		{
+			name:       "maximum below the initial delay",
+			delay:      "PT3S",
+			backoffMax: ptr.String("PT2S"),
+			wantTimes:  []time.Duration{0, 2 * time.Second, 4 * time.Second, 6 * time.Second, 8 * time.Second},
+		},
+		{
+			name:          "retry after can exceed the backoff maximum",
+			delay:         "PT1S",
+			backoffMax:    ptr.String("PT2S"),
+			retryAfter:    "3",
+			retryAfterMax: ptr.String("PT5S"),
+			wantTimes:     []time.Duration{0, 3 * time.Second, 6 * time.Second, 9 * time.Second, 12 * time.Second},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			synctest.Test(t, func(t *testing.T) {
+				policy := v1.BackoffPolicyExponential
+				config, err := RetryConfigFromDeliverySpec(v1.DeliverySpec{
+					Retry:         ptr.Int32(4),
+					BackoffPolicy: &policy,
+					BackoffDelay:  &tt.delay,
+					BackoffMax:    tt.backoffMax,
+					RetryAfterMax: tt.retryAfterMax,
+				})
+				require.NoError(t, err)
+
+				start := time.Now()
+				var sentAt []time.Duration
+				// Only HTTP responses are faked; the retry loop uses its real timers.
+				c := client{Client: http.Client{Transport: retryRoundTripperFunc(func(req *http.Request) (*http.Response, error) {
+					sentAt = append(sentAt, time.Since(start))
+					resp := &http.Response{
+						StatusCode: http.StatusNoContent,
+						Header:     make(http.Header),
+						Body:       http.NoBody,
+						Request:    req,
+					}
+					if len(sentAt) <= 4 {
+						resp.StatusCode = http.StatusServiceUnavailable
+						if tt.retryAfter != "" {
+							resp.Header.Set("Retry-After", tt.retryAfter)
+						}
+					}
+					return resp, nil
+				})}}
+				req, err := http.NewRequestWithContext(t.Context(), http.MethodPost, "http://subscriber.example.com", nil)
+				require.NoError(t, err)
+
+				resp, err := c.DoWithRetries(req, &config)
+				require.NoError(t, err)
+				defer resp.Body.Close()
+				assert.Equal(t, http.StatusNoContent, resp.StatusCode)
+				assert.Equal(t, tt.wantTimes, sentAt)
+			})
+		})
+	}
+}
+
+type retryRoundTripperFunc func(*http.Request) (*http.Response, error)
+
+func (f retryRoundTripperFunc) RoundTrip(req *http.Request) (*http.Response, error) {
+	return f(req)
+}
+
 func TestSaturatingPeriodDuration(t *testing.T) {
 	representable := period.MustParse("P292Y")
 	want, _ := representable.Duration()

pkg/reconciler/broker/trigger/trigger_test.go

@@ -363,8 +363,11 @@ func TestReconcile(t *testing.T) {
 					WithTriggerOIDCIdentityCreatedSucceededBecauseOIDCFeatureDisabled()),
 			}},
 		}, {
-			Name: "Creates subscription with retry from trigger",
+			Name: "Creates subscription with retry and backoff max from trigger",
 			Key:  testKey,
+			Ctx: feature.ToContext(context.Background(), feature.Flags{
+				feature.DeliveryBackoffMax: feature.Enabled,
+			}),
 			Objects: []runtime.Object{
 				NewBroker(brokerName, testNS,
 					WithBrokerClass(eventing.MTChannelBrokerClassValue),
@@ -378,16 +381,54 @@ func TestReconcile(t *testing.T) {
 				NewTrigger(triggerName, testNS, brokerName,
 					WithTriggerUID(triggerUID),
 					WithTriggerSubscriberURI(subscriberURI),
-					WithTriggerRetry(5, nil, nil)),
+					WithTriggerRetry(5, nil, nil),
+					withTriggerBackoffMax("PT2S")),
 			},
 			WantCreates: []runtime.Object{
-				resources.NewSubscription(ctx, makeTrigger(testNS), createTriggerChannelRef(), makeServiceURI(), makeBrokerRef(), makeDelivery(nil, ptr.Int32(5), nil, nil)),
+				resources.NewSubscription(ctx, makeTrigger(testNS), createTriggerChannelRef(), makeServiceURI(), makeBrokerRef(), makeDeliveryWithBackoffMax(nil, ptr.Int32(5), nil, nil, "PT2S")),
 			},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
 				Object: NewTrigger(triggerName, testNS, brokerName,
 					WithTriggerUID(triggerUID),
 					WithTriggerSubscriberURI(subscriberURI),
 					WithTriggerRetry(5, nil, nil),
+					withTriggerBackoffMax("PT2S"),
+					WithTriggerBrokerReady(),
+					WithTriggerDependencyReady(),
+					WithTriggerSubscriberResolvedSucceeded(),
+					WithTriggerDeadLetterSinkNotConfigured(),
+					WithTriggerSubscribedUnknown("SubscriptionNotConfigured", "Subscription has not yet been reconciled."),
+					WithTriggerStatusSubscriberURI(subscriberURI),
+					WithTriggerOIDCIdentityCreatedSucceededBecauseOIDCFeatureDisabled()),
+			}},
+		}, {
+			Name: "Creates subscription with backoff max from broker",
+			Key:  testKey,
+			Ctx: feature.ToContext(context.Background(), feature.Flags{
+				feature.DeliveryBackoffMax: feature.Enabled,
+			}),
+			Objects: []runtime.Object{
+				NewBroker(brokerName, testNS,
+					WithBrokerClass(eventing.MTChannelBrokerClassValue),
+					WithBrokerConfig(config()),
+					WithInitBrokerConditions,
+					WithBrokerReady,
+					WithChannelAddressAnnotation(triggerChannelURL),
+					WithChannelAPIVersionAnnotation(triggerChannelAPIVersion),
+					WithChannelKindAnnotation(triggerChannelKind),
+					WithChannelNameAnnotation(triggerChannelName),
+					withBrokerBackoffMax("PT2S")),
+				NewTrigger(triggerName, testNS, brokerName,
+					WithTriggerUID(triggerUID),
+					WithTriggerSubscriberURI(subscriberURI)),
+			},
+			WantCreates: []runtime.Object{
+				resources.NewSubscription(ctx, makeTrigger(testNS), createTriggerChannelRef(), makeServiceURI(), makeBrokerRef(), makeDeliveryWithBackoffMax(nil, nil, nil, nil, "PT2S")),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
+				Object: NewTrigger(triggerName, testNS, brokerName,
+					WithTriggerUID(triggerUID),
+					WithTriggerSubscriberURI(subscriberURI),
 					WithTriggerBrokerReady(),
 					WithTriggerDependencyReady(),
 					WithTriggerSubscriberResolvedSucceeded(),
@@ -1993,6 +2034,24 @@ func makeBrokerRefInDifferentNamespace() *duckv1.Destination {
 	}
 }
 
+func withBrokerBackoffMax(backoffMax string) BrokerOption {
+	return func(b *eventingv1.Broker) {
+		if b.Spec.Delivery == nil {
+			b.Spec.Delivery = new(eventingduckv1.DeliverySpec)
+		}
+		b.Spec.Delivery.BackoffMax = pointer.String(backoffMax)
+	}
+}
+
+func withTriggerBackoffMax(backoffMax string) TriggerOption {
+	return func(t *eventingv1.Trigger) {
+		if t.Spec.Delivery == nil {
+			t.Spec.Delivery = new(eventingduckv1.DeliverySpec)
+		}
+		t.Spec.Delivery.BackoffMax = pointer.String(backoffMax)
+	}
+}
+
 func makeReplyDestinationViaBrokerFilter() *duckv1.Destination {
 	return &duckv1.Destination{
 		URI: &apis.URL{
@@ -2018,6 +2077,12 @@ func makeDelivery(dls *duckv1.Destination, retry *int32, backoffPolicy *eventing
 	return ds
 }
 
+func makeDeliveryWithBackoffMax(dls *duckv1.Destination, retry *int32, backoffPolicy *eventingduckv1.BackoffPolicyType, backoffDelay *string, backoffMax string) *eventingduckv1.DeliverySpec {
+	ds := makeDelivery(dls, retry, backoffPolicy, backoffDelay)
+	ds.BackoffMax = pointer.String(backoffMax)
+	return ds
+}
+
 func makeDLSViaBrokerFilter() *eventingduckv1.DeliverySpec {
 	ds := &eventingduckv1.DeliverySpec{
 		DeadLetterSink: &duckv1.Destination{

pkg/reconciler/inmemorychannel/dispatcher/inmemorychannel_test.go

@@ -22,6 +22,7 @@ import (
 	"net/http"
 	"reflect"
 	"testing"
+	"time"
 
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/go-cmp/cmp/cmpopts"
@@ -352,12 +353,20 @@ func TestReconciler_ReconcileKind(t *testing.T) {
 	if err != nil {
 		t.Error(err)
 	}
+	previousSubscriber := subscriber1WithBackoffMax.DeepCopy()
+	previousSubscriber.Delivery.BackoffMax = ptr.String("PT2S")
+	previousSubscription, err := fanout.SubscriberSpecToFanoutConfig(*previousSubscriber)
+	if err != nil {
+		t.Fatal(err)
+	}
+	previousSubscription.Namespace = testNS
 
 	testCases := map[string]struct {
-		imc        *v1.InMemoryChannel
-		subs       []fanout.Subscription
-		wantSubs   []fanout.Subscription
-		wantResult reconciler.Event
+		imc          *v1.InMemoryChannel
+		subs         []fanout.Subscription
+		wantSubs     []fanout.Subscription
+		wantBackoffs map[int]time.Duration
+		wantResult   reconciler.Event
 	}{
 		"with no existing subscribers, 2 added": {
 			imc: NewInMemoryChannel(imcName, testNS,
@@ -542,20 +551,11 @@ func TestReconciler_ReconcileKind(t *testing.T) {
 				WithInMemoryChannelAddress(channelServiceAddress),
 				WithInMemoryChannelDLSUnknown(),
 				WithInMemoryChannelEventPoliciesReady()),
-			subs: []fanout.Subscription{{
-				Subscriber: duckv1.Addressable{
-					URL: apis.HTTP("call1"),
-				},
-				Reply: &duckv1.Addressable{
-					URL: apis.HTTP("sink2"),
-				},
-				RetryConfig: &kncloudevents.RetryConfig{
-					RetryMax:      3,
-					BackoffPolicy: &linear,
-					BackoffDelay:  ptr.String("PT1S"),
-					BackoffMax:    ptr.String("PT2S"),
-				},
-			}},
+			subs: []fanout.Subscription{*previousSubscription},
+			wantBackoffs: map[int]time.Duration{
+				3:  3 * time.Second,
+				20: 10 * time.Second,
+			},
 			wantSubs: []fanout.Subscription{{
 				Namespace: testNS,
 				Subscriber: duckv1.Addressable{
@@ -600,7 +600,7 @@ func TestReconciler_ReconcileKind(t *testing.T) {
 				handler := newFakeMultiChannelHandler()
 				if fanoutHandler != nil {
 					fanoutHandler.SetSubscriptions(context.TODO(), tc.subs)
-					handler.SetChannelHandler(channelServiceAddress.URL.String(), fanoutHandler)
+					handler.SetChannelHandler(channelServiceAddress.URL.Host, fanoutHandler)
 				}
 				r := &Reconciler{
 					multiChannelEventHandler: handler,
@@ -615,9 +615,23 @@ func TestReconciler_ReconcileKind(t *testing.T) {
 				if channelHandler == nil {
 					t.Fatalf("Did not get handler for %s", channelServiceAddress.URL.Host)
 				}
-				if diff := cmp.Diff(tc.wantSubs, channelHandler.GetSubscriptions(context.TODO()), cmpopts.IgnoreFields(kncloudevents.RetryConfig{}, "Backoff", "CheckRetry"), cmpopts.IgnoreFields(fanout.Subscription{}, "UID")); diff != "" {
+				if fanoutHandler != nil && channelHandler != fanoutHandler {
+					t.Fatal("expected the existing channel handler to be updated in place")
+				}
+				gotSubs := channelHandler.GetSubscriptions(context.TODO())
+				if diff := cmp.Diff(tc.wantSubs, gotSubs, cmpopts.IgnoreFields(kncloudevents.RetryConfig{}, "Backoff", "CheckRetry"), cmpopts.IgnoreFields(fanout.Subscription{}, "UID")); diff != "" {
 					t.Error("unexpected subs (+want/-got)", diff)
 				}
+				if tc.wantBackoffs != nil {
+					if len(gotSubs) != 1 || gotSubs[0].RetryConfig == nil || gotSubs[0].RetryConfig.Backoff == nil {
+						t.Fatal("expected one subscription with a backoff function")
+					}
+					for attempt, want := range tc.wantBackoffs {
+						if got := gotSubs[0].RetryConfig.Backoff(attempt, nil); got != want {
+							t.Errorf("Backoff(%d) = %s, want %s", attempt, got, want)
+						}
+					}
+				}
 			})
 		}
 	}

pkg/reconciler/subscription/subscription_test.go

@@ -2290,6 +2290,7 @@ func TestAllCases(t *testing.T) {
 			Ctx: feature.ToContext(context.TODO(), feature.Flags{
 				feature.DeliveryTimeout:    feature.Enabled,
 				feature.DeliveryRetryAfter: feature.Enabled,
+				feature.DeliveryBackoffMax: feature.Enabled,
 			}),
 			Objects: []runtime.Object{
 				NewSubscription("a-"+subscriptionName, testNS,
@@ -2298,6 +2299,7 @@ func TestAllCases(t *testing.T) {
 					WithSubscriptionSubscriberRef(serviceGVK, serviceName, testNS),
 					WithSubscriptionDeliverySpec(&eventingduck.DeliverySpec{
 						Timeout:       pointer.String("PT1S"),
+						BackoffMax:    pointer.String("PT2S"),
 						RetryAfterMax: pointer.String("PT2S"),
 					}),
 				),
@@ -2314,6 +2316,7 @@ func TestAllCases(t *testing.T) {
 					WithInMemoryChannelReadySubscriber("a-"+subscriptionUID),
 					WithInMemoryChannelDelivery(&eventingduck.DeliverySpec{
 						Timeout:       pointer.String("PT10S"),
+						BackoffMax:    pointer.String("PT10S"),
 						RetryAfterMax: pointer.String("PT20S"),
 					}),
 				),
@@ -2338,6 +2341,7 @@ func TestAllCases(t *testing.T) {
 					WithSubscriptionOIDCIdentityCreatedSucceededBecauseOIDCFeatureDisabled(),
 					WithSubscriptionDeliverySpec(&eventingduck.DeliverySpec{
 						Timeout:       pointer.String("PT1S"),
+						BackoffMax:    pointer.String("PT2S"),
 						RetryAfterMax: pointer.String("PT2S"),
 					}),
 				),
@@ -2349,6 +2353,7 @@ func TestAllCases(t *testing.T) {
 						SubscriberURI: serviceURI,
 						Delivery: &eventingduck.DeliverySpec{
 							Timeout:       pointer.String("PT1S"),
+							BackoffMax:    pointer.String("PT2S"),
 							RetryAfterMax: pointer.String("PT2S"),
 						},
 						Name: pointer.String("a-" + subscriptionName),

test/experimental/backoff_max_test.go

@@ -42,4 +42,5 @@ func TestBackoffMax(t *testing.T) {
 	)
 
 	env.Test(ctx, t, backoff_max.ChannelToSink())
+	env.Test(ctx, t, backoff_max.BrokerToTrigger())
 }

test/experimental/features/backoff_max/assertions.go

@@ -0,0 +1,30 @@
+/*
+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 backoff_max
+
+import (
+	cetest "github.com/cloudevents/sdk-go/v2/test"
+	"knative.dev/reconciler-test/pkg/eventshub/assert"
+	"knative.dev/reconciler-test/pkg/feature"
+)
+
+func assertRetryDelivery(f *feature.Feature, receiverName, eventID string) {
+	f.Assert("receiver rejects the first four deliveries", assert.OnStore(receiverName).
+		MatchRejectedEvent(cetest.HasId(eventID)).Exact(4))
+	f.Assert("receiver accepts the fifth delivery", assert.OnStore(receiverName).
+		MatchReceivedEvent(cetest.HasId(eventID)).Exact(1))
+}