coalesce updates to reduce intermediate updates (#7328)

* coalesce updates to reduce redundant processing in subscription handler

Signed-off-by: Huabing Zhao <[email protected]>

* retain order

Signed-off-by: Huabing Zhao <[email protected]>

* keep intermediate delete updates

Signed-off-by: Huabing Zhao <[email protected]>

* minor wording

Signed-off-by: Huabing Zhao <[email protected]>

* treat delete as normal operations

Signed-off-by: Huabing Zhao <[email protected]>

* retain the original order of the last updates for each key

Signed-off-by: Huabing Zhao <[email protected]>

* address comments

Signed-off-by: Huabing Zhao <[email protected]>

* fix test

Signed-off-by: Huabing Zhao <[email protected]>

---------

Signed-off-by: Huabing Zhao <[email protected]>

Author: zhaohuabing

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)
+			}
 		})
 	}
 }