treat delete as normal operations

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

Author: zhaohuabing

internal/message/watchutil.go

@@ -106,13 +106,6 @@ func HandleSubscription[K comparable, V any](
 	}
 }
 
-// updateKey is a helper struct used to uniquely identify updates by their key and deletion status.
-// This is to ensure that delete operations are treated distinctly from update operations.
-type updateKey[K comparable] struct {
-	Key    K
-	Delete bool
-}
-
 // 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.
@@ -122,25 +115,22 @@ func coalesceUpdates[K comparable, V any](runner string, updates []watchable.Upd
 	}
 
 	result := make([]watchable.Update[K, V], 0, len(updates))
-	indexByKey := make(map[updateKey[K]]int, len(updates))
+	indexByKey := make(map[K]int, len(updates))
 
-	for i := len(updates) - 1; i >= 0; i-- {
-		update := updates[i]
-		key := updateKey[K]{Key: update.Key, Delete: update.Delete}
-		if _, ok := indexByKey[key]; ok {
+	for _, update := range updates {
+		if idx, ok := indexByKey[update.Key]; ok {
+			// Keep the latest update for this key.
+			result[idx] = update
 			continue
 		}
 
-		indexByKey[key] = len(result)
+		indexByKey[update.Key] = len(result)
 		result = append(result, update)
 	}
 
-	// Reverse the result slice to restore the original order
-	for left, right := 0, len(result)-1; left < right; left, right = left+1, right-1 {
-		result[left], result[right] = result[right], result[left]
+	if len(result) != len(updates) {
+		logger.WithValues("runner", runner).Info("coalesced updates", "count", len(result), "before", len(updates))
 	}
 
-	logger.WithValues("runner", runner).Info("coalesced updates", "count", len(result), "before", len(updates))
-
 	return result
 }

internal/message/watchutil_internal_test.go

@@ -40,37 +40,19 @@ func TestCoalesceUpdates(t *testing.T) {
 		},
 		{
 			name: "latest update per key wins",
-			input: []watchable.Update[string, int]{
-				{Key: "foo", Value: 1},
-				{Key: "bar", Value: 2},
-				{Key: "baz", Value: 3},
-				{Key: "bar", Value: 4},
-				{Key: "foo", Value: 5},
-				{Key: "baz", Value: 6},
-			},
-			expected: []watchable.Update[string, int]{
-				{Key: "bar", Value: 4},
-				{Key: "foo", Value: 5},
-				{Key: "baz", Value: 6},
-			},
-		},
-		{
-			name: "do not coalesce delete and update for the same key",
 			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: 6},
+				{Key: "bar", Value: 7},
 			},
 			expected: []watchable.Update[string, int]{
-				{Key: "bar", Delete: true, Value: 2},
-				{Key: "baz", Value: 3},
 				{Key: "foo", Value: 5},
+				{Key: "bar", Value: 7},
 				{Key: "baz", Delete: true, Value: 6},
-				{Key: "bar", Value: 6},
 			},
 		},
 	}