test(e2e): verify delivery retry limits

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

Author: kahirokunn

test/e2e/delivery_backoff_test.go

@@ -0,0 +1,312 @@
+//go:build e2e
+// +build e2e
+
+/*
+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 e2e
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"os"
+	"sort"
+	"strconv"
+	"sync"
+	"testing"
+	"time"
+
+	cetest "github.com/cloudevents/sdk-go/v2/test"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/utils/ptr"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+	"knative.dev/pkg/system"
+
+	natssv1alpha1 "knative.dev/eventing-natss/pkg/apis/messaging/v1alpha1"
+	natssclient "knative.dev/eventing-natss/pkg/client/injection/client"
+	eventingduckv1 "knative.dev/eventing/pkg/apis/duck/v1"
+	eventingv1 "knative.dev/eventing/pkg/apis/eventing/v1"
+	messagingv1 "knative.dev/eventing/pkg/apis/messaging/v1"
+	eventingclient "knative.dev/eventing/pkg/client/injection/client"
+	"knative.dev/reconciler-test/pkg/environment"
+	"knative.dev/reconciler-test/pkg/eventshub"
+	"knative.dev/reconciler-test/pkg/feature"
+	"knative.dev/reconciler-test/pkg/k8s"
+	"knative.dev/reconciler-test/pkg/knative"
+)
+
+const deliveryFeaturesE2E = "EVENTING_DELIVERY_FEATURES_E2E"
+
+var (
+	brokerGVR       = schema.GroupVersionResource{Group: "eventing.knative.dev", Version: "v1", Resource: "brokers"}
+	triggerGVR      = schema.GroupVersionResource{Group: "eventing.knative.dev", Version: "v1", Resource: "triggers"}
+	subscriptionGVR = schema.GroupVersionResource{Group: "messaging.knative.dev", Version: "v1", Resource: "subscriptions"}
+	natsChannelGVR  = schema.GroupVersionResource{Group: "messaging.knative.dev", Version: "v1alpha1", Resource: "natsjetstreamchannels"}
+	serviceGVR      = schema.GroupVersionResource{Group: "", Version: "v1", Resource: "services"}
+)
+
+type deliveryRoute string
+
+const (
+	brokerRoute  deliveryRoute = "broker"
+	channelRoute deliveryRoute = "channel"
+)
+
+type deliveryScenario struct {
+	name              string
+	retries           int32
+	responseCode      int
+	retryAfterSeconds int
+	retryAfterMax     *string
+	expectedIntervals []time.Duration
+}
+
+// TestDeliveryBackoffTiming exercises the public Knative resources and the
+// NATS data planes. It only runs against the source-built Eventing matrix leg,
+// because released Eventing versions do not yet contain DeliverySpec.BackoffMax.
+func TestDeliveryBackoffTiming(t *testing.T) {
+	if os.Getenv(deliveryFeaturesE2E) != "true" {
+		t.Skipf("set %s=true to run delivery feature tests", deliveryFeaturesE2E)
+	}
+
+	retryAfterMax := "PT2S"
+	scenarios := []deliveryScenario{
+		{
+			name:              "backoff maximum",
+			retries:           4,
+			responseCode:      http.StatusServiceUnavailable,
+			expectedIntervals: []time.Duration{time.Second, 2 * time.Second, 2 * time.Second, 2 * time.Second},
+		},
+		{
+			name:              "retry after maximum",
+			retries:           2,
+			responseCode:      http.StatusTooManyRequests,
+			retryAfterSeconds: 5,
+			retryAfterMax:     &retryAfterMax,
+			expectedIntervals: []time.Duration{2 * time.Second, 2 * time.Second},
+		},
+	}
+
+	for _, route := range []deliveryRoute{brokerRoute, channelRoute} {
+		for _, scenario := range scenarios {
+			t.Run(string(route)+"/"+scenario.name, func(t *testing.T) {
+				ctx, env := global.Environment(
+					knative.WithKnativeNamespace(system.Namespace()),
+					knative.WithLoggingConfig,
+					knative.WithObservabilityConfig,
+					k8s.WithEventListener,
+				)
+				defer env.Finish()
+				env.Test(ctx, t, deliveryTimingFeature(route, scenario))
+			})
+		}
+	}
+}
+
+func deliveryTimingFeature(route deliveryRoute, scenario deliveryScenario) *feature.Feature {
+	const (
+		receiverName = "delivery-receiver"
+		senderName   = "delivery-sender"
+	)
+	event := cetest.FullEvent()
+	receiverOptions := []eventshub.EventsHubOption{
+		eventshub.StartReceiver,
+		eventshub.DropFirstN(uint(scenario.retries)),
+		eventshub.DropEventsResponseCode(scenario.responseCode),
+	}
+	if scenario.retryAfterSeconds > 0 {
+		receiverOptions = append(receiverOptions, eventshub.DropEventsResponseHeaders(map[string]string{
+			"Retry-After": strconv.Itoa(scenario.retryAfterSeconds),
+		}))
+	}
+
+	f := feature.NewFeatureNamed(fmt.Sprintf("%s delivery %s", route, scenario.name))
+	f.Setup("install receiver", eventshub.Install(receiverName, receiverOptions...))
+	f.Requirement("receiver is addressable", k8s.IsAddressable(serviceGVR, receiverName, time.Second, time.Minute))
+
+	delivery := deliverySpec(scenario)
+	var target schema.GroupVersionResource
+	var targetName string
+	switch route {
+	case brokerRoute:
+		target = brokerGVR
+		targetName = "delivery-broker"
+		f.Setup("install broker and trigger", installBrokerRoute(targetName, "delivery-trigger", receiverName, delivery))
+		f.Requirement("broker is ready", k8s.IsReady(brokerGVR, targetName, time.Second, 3*time.Minute))
+		f.Requirement("trigger is ready", k8s.IsReady(triggerGVR, "delivery-trigger", time.Second, 3*time.Minute))
+	case channelRoute:
+		target = natsChannelGVR
+		targetName = "delivery-channel"
+		f.Setup("install channel and subscription", installChannelRoute(targetName, "delivery-subscription", receiverName, delivery))
+		f.Requirement("channel is ready", k8s.IsReady(natsChannelGVR, targetName, time.Second, 3*time.Minute))
+		f.Requirement("subscription is ready", k8s.IsReady(subscriptionGVR, "delivery-subscription", time.Second, 3*time.Minute))
+	default:
+		panic(fmt.Sprintf("unsupported delivery route %q", route))
+	}
+
+	f.Assert("send event", eventshub.Install(
+		senderName,
+		eventshub.StartSenderToResource(target, targetName),
+		eventshub.InputEvent(event),
+	))
+	f.Assert("receiver rejects configured deliveries", func(ctx context.Context, t feature.T) {
+		eventshub.StoreFromContext(ctx, receiverName).AssertExact(
+			ctx, t, int(scenario.retries), eventWithKind(event.ID(), eventshub.EventRejected))
+	})
+	f.Assert("receiver accepts the final delivery", func(ctx context.Context, t feature.T) {
+		eventshub.StoreFromContext(ctx, receiverName).AssertExact(
+			ctx, t, 1, eventWithKind(event.ID(), eventshub.EventReceived))
+	})
+	f.Assert("deliveries follow configured intervals", func(ctx context.Context, t feature.T) {
+		eventshub.StoreFromContext(ctx, receiverName).AssertExact(
+			ctx, t, int(scenario.retries)+1, deliveriesFollowIntervals(event.ID(), scenario.expectedIntervals))
+	})
+
+	return f
+}
+
+func deliverySpec(scenario deliveryScenario) *eventingduckv1.DeliverySpec {
+	policy := eventingduckv1.BackoffPolicyExponential
+	return &eventingduckv1.DeliverySpec{
+		Retry:         ptr.To(scenario.retries),
+		BackoffPolicy: &policy,
+		BackoffDelay:  ptr.To("PT1S"),
+		BackoffMax:    ptr.To("PT2S"),
+		RetryAfterMax: scenario.retryAfterMax,
+	}
+}
+
+func installBrokerRoute(brokerName, triggerName, receiverName string, delivery *eventingduckv1.DeliverySpec) feature.StepFn {
+	return func(ctx context.Context, t feature.T) {
+		namespace := environment.FromContext(ctx).Namespace()
+		client := eventingclient.Get(ctx).EventingV1()
+		_, err := client.Brokers(namespace).Create(ctx, &eventingv1.Broker{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      brokerName,
+				Namespace: namespace,
+				Annotations: map[string]string{
+					"eventing.knative.dev/broker.class": "NatsJetStreamBroker",
+				},
+			},
+		}, metav1.CreateOptions{})
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		_, err = client.Triggers(namespace).Create(ctx, &eventingv1.Trigger{
+			ObjectMeta: metav1.ObjectMeta{Name: triggerName, Namespace: namespace},
+			Spec: eventingv1.TriggerSpec{
+				Broker: brokerName,
+				Subscriber: duckv1.Destination{Ref: &duckv1.KReference{
+					APIVersion: "v1",
+					Kind:       "Service",
+					Name:       receiverName,
+				}},
+				Delivery: delivery,
+			},
+		}, metav1.CreateOptions{})
+		if err != nil {
+			t.Fatal(err)
+		}
+	}
+}
+
+func installChannelRoute(channelName, subscriptionName, receiverName string, delivery *eventingduckv1.DeliverySpec) feature.StepFn {
+	return func(ctx context.Context, t feature.T) {
+		namespace := environment.FromContext(ctx).Namespace()
+		_, err := natssclient.Get(ctx).MessagingV1alpha1().NatsJetStreamChannels(namespace).Create(ctx, &natssv1alpha1.NatsJetStreamChannel{
+			ObjectMeta: metav1.ObjectMeta{Name: channelName, Namespace: namespace},
+			Spec: natssv1alpha1.NatsJetStreamChannelSpec{ChannelableSpec: eventingduckv1.ChannelableSpec{
+				Delivery: delivery,
+			}},
+		}, metav1.CreateOptions{})
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		_, err = eventingclient.Get(ctx).MessagingV1().Subscriptions(namespace).Create(ctx, &messagingv1.Subscription{
+			ObjectMeta: metav1.ObjectMeta{Name: subscriptionName, Namespace: namespace},
+			Spec: messagingv1.SubscriptionSpec{
+				Channel: duckv1.KReference{
+					APIVersion: natssv1alpha1.SchemeGroupVersion.String(),
+					Kind:       "NatsJetStreamChannel",
+					Name:       channelName,
+				},
+				Subscriber: &duckv1.Destination{Ref: &duckv1.KReference{
+					APIVersion: "v1",
+					Kind:       "Service",
+					Name:       receiverName,
+				}},
+			},
+		}, metav1.CreateOptions{})
+		if err != nil {
+			t.Fatal(err)
+		}
+	}
+}
+
+func eventWithKind(id string, kind eventshub.EventKind) eventshub.EventInfoMatcher {
+	return func(info eventshub.EventInfo) error {
+		if info.Event == nil || info.Event.ID() != id {
+			return fmt.Errorf("received a different event")
+		}
+		if info.Kind != kind {
+			return fmt.Errorf("event kind %q, want %q", info.Kind, kind)
+		}
+		return nil
+	}
+}
+
+func deliveriesFollowIntervals(id string, expected []time.Duration) eventshub.EventInfoMatcher {
+	type deliveryKey struct {
+		kind     eventshub.EventKind
+		sequence uint64
+	}
+
+	var mu sync.Mutex
+	seen := make(map[deliveryKey]eventshub.EventInfo, len(expected)+1)
+
+	return func(info eventshub.EventInfo) error {
+		if info.Event == nil || info.Event.ID() != id {
+			return fmt.Errorf("received a different event")
+		}
+
+		mu.Lock()
+		defer mu.Unlock()
+		seen[deliveryKey{kind: info.Kind, sequence: info.Sequence}] = info
+		if len(seen) < len(expected)+1 {
+			return nil
+		}
+
+		deliveries := make([]eventshub.EventInfo, 0, len(seen))
+		for _, delivery := range seen {
+			deliveries = append(deliveries, delivery)
+		}
+		sort.Slice(deliveries, func(i, j int) bool {
+			return deliveries[i].Time.Before(deliveries[j].Time)
+		})
+
+		for i, wait := range expected {
+			actual := deliveries[i+1].Time.Sub(deliveries[i].Time)
+			if actual < wait-500*time.Millisecond || actual > wait+3*time.Second {
+				return fmt.Errorf("delivery %d waited %s, expected %s", i+2, actual, wait)
+			}
+		}
+		return nil
+	}
+}