Merge branch 'main' into auto-getn

Author: codefromthecrypt

go.mod

@@ -15,9 +15,9 @@ require (
 	github.com/docker/cli v28.5.1+incompatible
 	github.com/docker/docker v28.3.3+incompatible
 	github.com/dominikbraun/graph v0.23.0
-	github.com/envoyproxy/go-control-plane v0.13.5-0.20250929230642-07d3df27ff4f
+	github.com/envoyproxy/go-control-plane v0.13.5-0.20251022160057-de4316c523b7
 	github.com/envoyproxy/go-control-plane/contrib v1.32.5-0.20250430092421-68a532e11403
-	github.com/envoyproxy/go-control-plane/envoy v1.35.1-0.20250929230642-07d3df27ff4f
+	github.com/envoyproxy/go-control-plane/envoy v1.35.1-0.20251022160057-de4316c523b7
 	github.com/envoyproxy/go-control-plane/ratelimit v0.1.1-0.20250805143705-d51f8590a549
 	github.com/envoyproxy/ratelimit v1.4.1-0.20230427142404-e2a87f41d3a7
 	github.com/evanphx/json-patch v5.9.11+incompatible

go.sum

@@ -173,12 +173,12 @@ github.com/emicklei/go-restful/v3 v3.13.0/go.mod h1:6n3XBCmQQb25CM2LCACGz8ukIrRr
 github.com/envoyproxy/go-control-plane v0.9.0/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
 github.com/envoyproxy/go-control-plane v0.9.1-0.20191026205805-5f8ba28d4473/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
 github.com/envoyproxy/go-control-plane v0.9.4/go.mod h1:6rpuAdCZL397s3pYoYcLgu1mIlRU8Am5FuJP05cCM98=
-github.com/envoyproxy/go-control-plane v0.13.5-0.20250929230642-07d3df27ff4f h1:36vvJBe/wXWfD7qrTb1WnbPVPMxNFDfEygztH8wgebw=
-github.com/envoyproxy/go-control-plane v0.13.5-0.20250929230642-07d3df27ff4f/go.mod h1:PTY7yDlLxB4bW7rEOO7e79uTDr9yXzpuI1QGIDfxEzc=
+github.com/envoyproxy/go-control-plane v0.13.5-0.20251022160057-de4316c523b7 h1:JsOVlgacLOjgYWGb4V2E/d/c/o4+HEvYjr0+KnL4j4o=
+github.com/envoyproxy/go-control-plane v0.13.5-0.20251022160057-de4316c523b7/go.mod h1:Alz8LEClvR7xKsrq3qzoc4N0guvVNSS8KmSChGYr9hs=
 github.com/envoyproxy/go-control-plane/contrib v1.32.5-0.20250430092421-68a532e11403 h1:5wPocL1bGYhA4TtKZwcdVI5fsXo1JatkbcxPBcFQswc=
 github.com/envoyproxy/go-control-plane/contrib v1.32.5-0.20250430092421-68a532e11403/go.mod h1:Xkwx/TGvEKRCL2mitdiuQWOD1ECvfM5krWWVo2vI2Zk=
-github.com/envoyproxy/go-control-plane/envoy v1.35.1-0.20250929230642-07d3df27ff4f h1:4efYrIQgVRwCmwCveby6ck+VpxqzibdOL1Out1rJqqc=
-github.com/envoyproxy/go-control-plane/envoy v1.35.1-0.20250929230642-07d3df27ff4f/go.mod h1:2LcmvJoXsDSrsGZIxGM0Gah9ykiwTn/kgjyQdnNH8Jc=
+github.com/envoyproxy/go-control-plane/envoy v1.35.1-0.20251022160057-de4316c523b7 h1:Q4zISHdb9brRNvzmQl8Bwvap7GAtGrQjHyyw9OYmkFk=
+github.com/envoyproxy/go-control-plane/envoy v1.35.1-0.20251022160057-de4316c523b7/go.mod h1:ty89S1YCCVruQAm9OtKeEkQLTb+Lkz0k8v9W0Oxsv98=
 github.com/envoyproxy/go-control-plane/ratelimit v0.1.1-0.20250805143705-d51f8590a549 h1:5K0vH5H4dtCIO8+w/yq6vDaMcGn9RoPrHfmPAFAztwU=
 github.com/envoyproxy/go-control-plane/ratelimit v0.1.1-0.20250805143705-d51f8590a549/go.mod h1:KxtyvDAPIEkqUUvF9ooo5gSGVOtQ08wUTnQe5LsJC6c=
 github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c=

internal/message/watchutil.go

@@ -99,8 +99,41 @@ func HandleSubscription[K comparable, V any](
 	}
 	for snapshot := range subscription {
 		watchableDepth.With(meta.LabelValues()...).Record(float64(len(subscription)))
-		for _, update := range snapshot.Updates {
+
+		for _, update := range coalesceUpdates(meta.Runner, snapshot.Updates) {
 			handleWithCrashRecovery(handle, Update[K, V](update), meta, errChans)
 		}
 	}
 }
+
+// coalesceUpdates merges multiple updates for the same key into a single update,
+// preserving the latest state for each key.
+// This helps reduce redundant processing and ensures that only the most recent update per key is handled.
+func coalesceUpdates[K comparable, V any](runner string, updates []watchable.Update[K, V]) []watchable.Update[K, V] {
+	if len(updates) <= 1 {
+		return updates
+	}
+
+	seen := make(map[K]struct{}, len(updates))
+	write := len(updates) - 1
+
+	for read := len(updates) - 1; read >= 0; read-- {
+		update := updates[read]
+		if _, ok := seen[update.Key]; ok {
+			continue
+		}
+		seen[update.Key] = struct{}{}
+		updates[write] = update
+		write--
+	}
+
+	result := updates[write+1:]
+	if len(result) != len(updates) {
+		logger.WithValues("runner", runner).Info(
+			"coalesced updates",
+			"count", len(result),
+			"before", len(updates),
+		)
+	}
+	return result
+}

internal/message/watchutil_internal_test.go

@@ -0,0 +1,68 @@
+// Copyright Envoy Gateway Authors
+// SPDX-License-Identifier: Apache-2.0
+// The full text of the Apache license is available in the LICENSE file at
+// the root of the repo.
+
+package message
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"github.com/telepresenceio/watchable"
+)
+
+func TestCoalesceUpdates(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name     string
+		input    []watchable.Update[string, int]
+		expected []watchable.Update[string, int]
+	}{
+		{
+			name:     "empty input returns nil",
+			input:    []watchable.Update[string, int]{},
+			expected: []watchable.Update[string, int]{},
+		},
+		{
+			name: "simple updates without repeats",
+			input: []watchable.Update[string, int]{
+				{Key: "foo", Value: 1},
+				{Key: "bar", Value: 2},
+				{Key: "baz", Value: 3},
+			},
+			expected: []watchable.Update[string, int]{
+				{Key: "foo", Value: 1},
+				{Key: "bar", Value: 2},
+				{Key: "baz", Value: 3},
+			},
+		},
+		{
+			name: "latest update per key wins",
+			input: []watchable.Update[string, int]{
+				{Key: "foo", Value: 1},
+				{Key: "bar", Delete: true, Value: 2},
+				{Key: "baz", Value: 3},
+				{Key: "bar", Value: 4},
+				{Key: "foo", Value: 5},
+				{Key: "baz", Delete: true, Value: 6},
+				{Key: "bar", Value: 7},
+			},
+			expected: []watchable.Update[string, int]{
+				{Key: "foo", Value: 5},
+				{Key: "baz", Delete: true, Value: 6},
+				{Key: "bar", Value: 7},
+			},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			actual := coalesceUpdates("test-runner", tc.input)
+			require.Equal(t, tc.expected, actual)
+		})
+	}
+}

internal/message/watchutil_test.go

@@ -89,7 +89,7 @@ func TestHandleSubscriptionAlreadyInitialized(t *testing.T) {
 			}
 		},
 	)
-	assert.Equal(t, 2, storeCalls)
+	assert.LessOrEqual(t, 2, storeCalls) // updates can be coalesced
 	assert.Equal(t, 1, deleteCalls)
 }
 
@@ -268,7 +268,11 @@ func TestControllerResourceUpdate(t *testing.T) {
 					m.GatewayAPIResources.Close()
 				}
 			})
-			assert.Equal(t, tc.updates, updates)
+			if tc.updates > 1 {
+				assert.LessOrEqual(t, updates, tc.updates) // Updates can be coalesced
+			} else {
+				assert.Equal(t, 1, updates)
+			}
 		})
 	}
 }

internal/xds/extensions/extensions.gen.go

@@ -76,6 +76,7 @@ import (
 	_ "github.com/envoyproxy/go-control-plane/envoy/extensions/clusters/dns/v3"
 	_ "github.com/envoyproxy/go-control-plane/envoy/extensions/clusters/dynamic_forward_proxy/v3"
 	_ "github.com/envoyproxy/go-control-plane/envoy/extensions/clusters/redis/v3"
+	_ "github.com/envoyproxy/go-control-plane/envoy/extensions/clusters/reverse_connection/v3"
 	_ "github.com/envoyproxy/go-control-plane/envoy/extensions/common/async_files/v3"
 	_ "github.com/envoyproxy/go-control-plane/envoy/extensions/common/aws/v3"
 	_ "github.com/envoyproxy/go-control-plane/envoy/extensions/common/dynamic_forward_proxy/v3"
@@ -140,6 +141,7 @@ import (
 	_ "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/http/kill_request/v3"
 	_ "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/http/local_ratelimit/v3"
 	_ "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/http/lua/v3"
+	_ "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/http/mcp/v3"
 	_ "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/http/oauth2/v3"
 	_ "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/http/on_demand/v3"
 	_ "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/http/original_src/v3"
@@ -224,6 +226,7 @@ import (
 	_ "github.com/envoyproxy/go-control-plane/envoy/extensions/http/custom_response/local_response_policy/v3"
 	_ "github.com/envoyproxy/go-control-plane/envoy/extensions/http/custom_response/redirect_policy/v3"
 	_ "github.com/envoyproxy/go-control-plane/envoy/extensions/http/early_header_mutation/header_mutation/v3"
+	_ "github.com/envoyproxy/go-control-plane/envoy/extensions/http/ext_proc/processing_request_modifiers/mapped_attribute_builder/v3"
 	_ "github.com/envoyproxy/go-control-plane/envoy/extensions/http/ext_proc/response_processors/save_processing_response/v3"
 	_ "github.com/envoyproxy/go-control-plane/envoy/extensions/http/header_formatters/preserve_case/v3"
 	_ "github.com/envoyproxy/go-control-plane/envoy/extensions/http/header_validators/envoy_default/v3"
@@ -253,6 +256,7 @@ import (
 	_ "github.com/envoyproxy/go-control-plane/envoy/extensions/matching/common_inputs/environment_variable/v3"
 	_ "github.com/envoyproxy/go-control-plane/envoy/extensions/matching/common_inputs/network/v3"
 	_ "github.com/envoyproxy/go-control-plane/envoy/extensions/matching/common_inputs/ssl/v3"
+	_ "github.com/envoyproxy/go-control-plane/envoy/extensions/matching/common_inputs/stats/v3"
 	_ "github.com/envoyproxy/go-control-plane/envoy/extensions/matching/input_matchers/consistent_hashing/v3"
 	_ "github.com/envoyproxy/go-control-plane/envoy/extensions/matching/input_matchers/ip/v3"
 	_ "github.com/envoyproxy/go-control-plane/envoy/extensions/matching/input_matchers/metadata/v3"

internal/xds/translator/extauth.go

@@ -134,26 +134,32 @@ func extAuthConfig(extAuth *ir.ExtAuth) (*extauthv3.ExtAuthz, error) {
 		timeout = durationpb.New(extAuth.Timeout.Duration)
 	}
 
+	var rp *corev3.RetryPolicy
+	// Set the retry policy if it exists.
+	if extAuth.Traffic != nil && extAuth.Traffic.Retry != nil {
+		var err error
+		rp, err = buildNonRouteRetryPolicy(extAuth.Traffic.Retry)
+		if err != nil {
+			return nil, fmt.Errorf("build retry policy for http service: %w", err)
+		}
+	}
+
 	if extAuth.HTTP != nil {
+		hs := httpService(extAuth.HTTP, timeout)
+		hs.RetryPolicy = rp
+
 		config.Services = &extauthv3.ExtAuthz_HttpService{
-			HttpService: httpService(extAuth.HTTP, timeout),
+			HttpService: hs,
 		}
-		// Retry policy is not supported for HTTP service.
 	} else if extAuth.GRPC != nil {
 		service := &corev3.GrpcService{
 			TargetSpecifier: &corev3.GrpcService_EnvoyGrpc_{
 				EnvoyGrpc: grpcService(extAuth.GRPC),
 			},
 			Timeout: timeout,
 		}
-		// Set the retry policy if it exists.
-		if extAuth.Traffic != nil && extAuth.Traffic.Retry != nil {
-			rp, err := buildNonRouteRetryPolicy(extAuth.Traffic.Retry)
-			if err != nil {
-				return nil, fmt.Errorf("build retry policy for gRPC service: %w", err)
-			}
-			service.RetryPolicy = rp
-		}
+		service.RetryPolicy = rp
+
 		config.Services = &extauthv3.ExtAuthz_GrpcService{
 			GrpcService: service,
 		}