fix: correct deduplication of watch events

This is extracted from #633 as a separate fix.

The idea is that if we merge/deduplicate watch events, we should do this
by immutable attributes of the resource: namespace, type, id.

The previous implementation might split two events on e.g. finalizers
being empty, so two events which report finalizers empty/finalizers
non-empty might be delivered in the wrong order: first empty, next
non-empty.

It might not trigger a real bug due to the way controllers work, but
still is not consistent and correct.

Signed-off-by: Andrey Smirnov <[email protected]>

Author: smira

pkg/controller/runtime/internal/reduced/reduced.go

@@ -7,22 +7,42 @@ package reduced
 
 import "github.com/cosi-project/runtime/pkg/resource"
 
-// Metadata is _comparable_, so that it can be a map key.
+// Metadata reduces resource metadata for deduplication.
+//
+// It consists of two parts:
+// - a comparable Key which is used for deduplication.
+// - a Value which is reduced for duplicate keys to the last observed value.
 type Metadata struct {
-	Namespace       resource.Namespace
-	Typ             resource.Type
-	ID              resource.ID
+	Key
+	Value
+}
+
+// Key is a comparable representation of deduplication entry.
+type Key struct {
+	Namespace resource.Namespace
+	Typ       resource.Type
+	ID        resource.ID
+}
+
+// Value is a reduced representation of resource metadata.
+type Value struct {
+	Labels          *resource.Labels
 	Phase           resource.Phase
 	FinalizersEmpty bool
 }
 
 // NewMetadata creates a new reduced Metadata from a resource.Metadata.
 func NewMetadata(md *resource.Metadata) Metadata {
 	return Metadata{
-		Namespace:       md.Namespace(),
-		Typ:             md.Type(),
-		ID:              md.ID(),
-		Phase:           md.Phase(),
-		FinalizersEmpty: md.Finalizers().Empty(),
+		Key: Key{
+			Namespace: md.Namespace(),
+			Typ:       md.Type(),
+			ID:        md.ID(),
+		},
+		Value: Value{
+			Phase:           md.Phase(),
+			FinalizersEmpty: md.Finalizers().Empty(),
+			Labels:          md.Labels(),
+		},
 	}
 }

pkg/controller/runtime/runtime.go

@@ -266,13 +266,18 @@ func (runtime *Runtime) watch(resourceNamespace resource.Namespace, resourceType
 	return runtime.state.WatchKindAggregated(runtime.runCtx, kind, runtime.watchCh, state.WithBootstrapBookmark(true))
 }
 
-type dedup map[reduced.Metadata]struct{}
+type dedup map[reduced.Key]reduced.Value
 
 func (d dedup) takeOne() reduced.Metadata {
 	for k := range d {
+		md := reduced.Metadata{
+			Key:   k,
+			Value: d[k],
+		}
+
 		delete(d, k)
 
-		return k
+		return md
 	}
 
 	panic("dedup is empty")
@@ -353,7 +358,8 @@ eventLoop:
 			}
 		}
 
-		m[reduced.NewMetadata(e.Resource.Metadata())] = struct{}{}
+		reducedMD := reduced.NewMetadata(e.Resource.Metadata())
+		m[reducedMD.Key] = reducedMD.Value
 	}
 
 	return true