keep intermediate delete updates

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

Author: zhaohuabing

internal/message/watchutil.go

@@ -106,6 +106,13 @@ 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 deleted keys are treated distinctly from updated keys.
+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.
@@ -115,15 +122,16 @@ func coalesceUpdates[K comparable, V any](runner string, updates []watchable.Upd
 	}
 
 	result := make([]watchable.Update[K, V], 0, len(updates))
-	indexByKey := make(map[K]int, len(updates))
+	indexByKey := make(map[updateKey[K]]int, len(updates))
 
 	for i := len(updates) - 1; i >= 0; i-- {
 		update := updates[i]
-		if _, ok := indexByKey[update.Key]; ok {
+		key := updateKey[K]{Key: update.Key, Delete: update.Delete}
+		if _, ok := indexByKey[key]; ok {
 			continue
 		}
 
-		indexByKey[update.Key] = len(result)
+		indexByKey[key] = len(result)
 		result = append(result, update)
 	}
 

internal/message/watchutil_internal_test.go

@@ -12,7 +12,7 @@ import (
 	"github.com/telepresenceio/watchable"
 )
 
-func TestMergeUpdates(t *testing.T) {
+func TestCoalesceUpdates(t *testing.T) {
 	t.Parallel()
 
 	tests := []struct {
@@ -26,18 +26,51 @@ func TestMergeUpdates(t *testing.T) {
 			expected: []watchable.Update[string, int]{},
 		},
 		{
-			name: "latest update per key delete state wins",
+			name: "simple updates without repeats",
 			input: []watchable.Update[string, int]{
 				{Key: "foo", Value: 1},
-				{Key: "bar", Delete: true, Value: 10},
-				{Key: "baz", Value: 5},
-				{Key: "bar", Delete: true, Value: 11},
-				{Key: "foo", Value: 2},
+				{Key: "bar", Value: 2},
+				{Key: "baz", Value: 3},
 			},
 			expected: []watchable.Update[string, int]{
-				{Key: "baz", Value: 5},
-				{Key: "bar", Delete: true, Value: 11},
-				{Key: "foo", Value: 2},
+				{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", 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: "keep intermediate deletes",
+			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},
+			},
+			expected: []watchable.Update[string, int]{
+				{Key: "bar", Delete: true, Value: 2},
+				{Key: "baz", Value: 3},
+				{Key: "foo", Value: 5},
+				{Key: "baz", Delete: true, Value: 6},
+				{Key: "bar", Value: 6},
 			},
 		},
 	}

internal/message/watchutil_test.go

@@ -89,43 +89,10 @@ func TestHandleSubscriptionAlreadyInitialized(t *testing.T) {
 			}
 		},
 	)
-	assert.LessOrEqual(t, storeCalls, 2) // Can be coalesced TODO handle delete
+	assert.Equal(t, 2, storeCalls)
 	assert.Equal(t, 1, deleteCalls)
 }
 
-func TestHandleSubscriptionCoalescesUpdates(t *testing.T) {
-	snapshotC := make(chan watchable.Snapshot[string, any], 2)
-	snapshotC <- watchable.Snapshot[string, any]{State: map[string]any{}}
-	snapshotC <- watchable.Snapshot[string, any]{
-		State: map[string]any{},
-		Updates: []watchable.Update[string, any]{
-			{Key: "foo", Value: "v1"},
-			{Key: "foo", Value: "v2"},
-			{Key: "foo", Delete: true},
-			{Key: "foo", Value: "v3"},
-			{Key: "bar", Value: "v1"},
-			{Key: "bar", Value: "v2"},
-		},
-	}
-	close(snapshotC)
-
-	var observed []watchable.Update[string, any]
-	message.HandleSubscription[string, any](
-		message.Metadata{Runner: "demo", Message: "demo"},
-		snapshotC,
-		func(update message.Update[string, any], errChans chan error) {
-			observed = append(observed, watchable.Update[string, any](update))
-		},
-	)
-
-	assert.Len(t, observed, 2)
-	assert.Equal(t, "foo", observed[0].Key)
-	assert.Equal(t, "v3", observed[0].Value)
-	assert.False(t, observed[0].Delete)
-	assert.Equal(t, "bar", observed[1].Key)
-	assert.Equal(t, "v2", observed[1].Value)
-}
-
 func TestControllerResourceUpdate(t *testing.T) {
 	tests := []struct {
 		desc      string