feat: deduplicate event writer events for spec/statusupdate

Signed-off-by: Jonathan West <jgwest@gmail.com>

Author: jgwest

internal/event/event_writer.go

@@ -2,6 +2,8 @@ package event
 
 import (
 	"context"
+	"math"
+	"math/rand/v2"
 	"sync"
 	"time"
 
@@ -17,7 +19,7 @@ import (
 
 const (
 	// maxEventRetries is the maximum number of times an event will be retried before giving up.
-	maxEventRetries = 12
+	maxEventRetries = math.MaxInt
 )
 
 type streamWriter interface {
@@ -113,9 +115,10 @@ func (ew *EventWriter) Add(ev *cloudevents.Event) {
 
 	defaultBackoff := wait.Backoff{
 		Steps:    maxEventRetries,
-		Duration: 1 * time.Second,
-		Factor:   1.5,
-		Jitter:   0.1,
+		Duration: 5 * time.Second,
+		Factor:   4,
+		Jitter:   1,
+		Cap:      2 * time.Minute, // never wait longer than 2 minutes
 	}
 
 	if resID == "" {
@@ -171,7 +174,7 @@ func (ew *EventWriter) Get(resID string) *eventMessage {
 
 	// Then check unsent queue
 	if eq, exists := ew.unsentEvents[resID]; exists {
-		return eq.get()
+		return eq.peek()
 	}
 	return nil
 }
@@ -196,13 +199,18 @@ func (ew *EventWriter) Remove(ev *cloudevents.Event) {
 		}
 	}
 
+	// JGW-TODO:
+	// - A) Should Remove remove from both sentEvents/unsentEvents? (I mean, it seems likely it will be in one or the other, but we could check both?)
+	// and
+	// - B) In the unsent case, why does remove only remove from the front of the queue?
+
 	// If not in sent events, check unsent queue
 	eq, exists := ew.unsentEvents[resourceID]
 	if !exists {
 		return
 	}
 
-	front := eq.get()
+	front := eq.peek()
 	if front == nil {
 		return
 	}
@@ -248,6 +256,11 @@ func (ew *EventWriter) SendWaitingEvents(ctx context.Context) {
 			}
 			ew.mu.RUnlock()
 
+			// Shuffle so no resource is systematically starved
+			rand.Shuffle(len(resourceIDs), func(i, j int) {
+				resourceIDs[i], resourceIDs[j] = resourceIDs[j], resourceIDs[i]
+			})
+
 			for _, resourceID := range resourceIDs {
 				ew.sendEvent(resourceID)
 			}
@@ -484,26 +497,63 @@ func (eq *eventQueue) add(ev *eventMessage) {
 	eq.mu.Lock()
 	defer eq.mu.Unlock()
 
-	if len(eq.items) > 0 {
-		tail := eq.items[len(eq.items)-1]
-		tail.mu.Lock()
+	eq.items = append(eq.items, ev)
 
-		// Replace an older event with a newer one of the same type
-		if ev.event.Type() == tail.event.Type() {
-			tail.event = ev.event
-			tail.backoff = ev.backoff
-			tail.retryAfter = ev.retryAfter
-			tail.mu.Unlock()
-			return
+	deduplicateEventMessageItems(&eq.items)
+
+	// if len(eq.items) > 0 {
+	// 	tail := eq.items[len(eq.items)-1]
+	// 	tail.mu.Lock()
+
+	// 	// Replace an older event with a newer one of the same type
+	// 	if ev.event.Type() == tail.event.Type() {
+	// 		tail.event = ev.event
+	// 		tail.backoff = ev.backoff
+	// 		tail.retryAfter = ev.retryAfter
+	// 		tail.mu.Unlock()
+	// 		return
+	// 	}
+	// 	tail.mu.Unlock()
+	// }
+
+	// eq.items = append(eq.items, ev)
+}
+
+// deduplicateEventMessageItems
+// - Ensure you own the lock on the items parameter (e.g. via eq.mu.Lock()) before calling this function
+func deduplicateEventMessageItems(items *[]*eventMessage) {
+
+	// key: Type() of event
+	// value: (not used)
+	haveWeSeenMsgWithType := make(map[string]bool, 0)
+
+	// Work backwards through the list:
+	// - Items at the end of the list are 'fresher', items are the beginning of the list are more stale
+	// - We thus remove items early in the list in favour of those that are later in the list
+	for idx := len(*items) - 1; idx >= 0; idx-- {
+		item := (*items)[idx]
+
+		myType := item.event.Type()
+
+		// No de-duplication of events we can't guarantee are safe to de-duplicate
+		if myType != StatusUpdate.String() && myType != SpecUpdate.String() {
+			continue
+		}
+
+		// De-duplicate statusupdate and specupdate, as we know they are safe to de-duplicate
+		if _, typePreviouslySeen := haveWeSeenMsgWithType[myType]; typePreviouslySeen {
+			// Stale duplicate: a fresher same-type entry was retained when scanning from the tail.
+			*items = append((*items)[:idx], (*items)[idx+1:]...)
+		} else {
+			// This is the first type we have seen the type, so add it to the map and continue without removal
+			haveWeSeenMsgWithType[myType] = true
 		}
-		tail.mu.Unlock()
 	}
 
-	eq.items = append(eq.items, ev)
 }
 
-// get the first item from the queue.
-func (eq *eventQueue) get() *eventMessage {
+// peek the first item from the queue.
+func (eq *eventQueue) peek() *eventMessage {
 	eq.mu.RLock()
 	defer eq.mu.RUnlock()
 	if len(eq.items) == 0 {

internal/event/event_writer_test.go

@@ -286,29 +286,29 @@ func TestEventWriter(t *testing.T) {
 		require.Equal(t, 1, sentMsg.retryCount)
 	})
 
-	t.Run("should give up after max retries", func(t *testing.T) {
-		fs := &fakeStream{}
-		evSender := NewEventWriter("test", fs)
-
-		ev := es.ApplicationEvent(Create, app1)
-		resID := createResourceID(app1.ObjectMeta)
-		evSender.Add(ev)
-
-		// Send the event
-		evSender.sendEvent(resID)
-		sentMsg := evSender.sentEvents[resID]
-		require.NotNil(t, sentMsg)
-
-		// Exhaust retries
-		for i := 0; i <= maxEventRetries; i++ {
-			pastTime := time.Now().Add(-1 * time.Second)
-			sentMsg.retryAfter = &pastTime
-			evSender.retrySentEvent(resID, sentMsg)
-		}
-
-		// After max retries, event should be removed from sentEvents
-		require.NotContains(t, evSender.sentEvents, resID)
-	})
+	// t.Run("should give up after max retries", func(t *testing.T) {
+	// 	fs := &fakeStream{}
+	// 	evSender := NewEventWriter("test", fs)
+
+	// 	ev := es.ApplicationEvent(Create, app1)
+	// 	resID := createResourceID(app1.ObjectMeta)
+	// 	evSender.Add(ev)
+
+	// 	// Send the event
+	// 	evSender.sendEvent(resID)
+	// 	sentMsg := evSender.sentEvents[resID]
+	// 	require.NotNil(t, sentMsg)
+
+	// 	// Exhaust retries
+	// 	for i := 0; i <= maxEventRetries; i++ {
+	// 		pastTime := time.Now().Add(-1 * time.Second)
+	// 		sentMsg.retryAfter = &pastTime
+	// 		evSender.retrySentEvent(resID, sentMsg)
+	// 	}
+
+	// 	// After max retries, event should be removed from sentEvents
+	// 	require.NotContains(t, evSender.sentEvents, resID)
+	// })
 
 	t.Run("should not send ACK events to sentEvents", func(t *testing.T) {
 		fs := &fakeStream{}
@@ -477,6 +477,107 @@ func TestEventWriter(t *testing.T) {
 	})
 }
 
+func TestDeduplicateEventMessageItems(t *testing.T) {
+	es := NewEventSource("dedupe-test")
+	app := &v1alpha1.Application{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "app",
+			Namespace: "ns",
+			UID:       "uid-a",
+		},
+	}
+
+	eventMsg := func(evType EventType, resourceVersion string) *eventMessage {
+		app.ResourceVersion = resourceVersion
+		return &eventMessage{event: es.ApplicationEvent(evType, app)}
+	}
+
+	t.Run("empty slice", func(t *testing.T) {
+		items := []*eventMessage{}
+		deduplicateEventMessageItems(&items)
+		require.Empty(t, items)
+	})
+
+	t.Run("single spec update unchanged", func(t *testing.T) {
+		ev := eventMsg(SpecUpdate, "1")
+		items := []*eventMessage{ev}
+		deduplicateEventMessageItems(&items)
+		require.Len(t, items, 1)
+		require.Equal(t, EventID(ev.event), EventID(items[0].event))
+	})
+
+	t.Run("multiple spec updates keep only freshest", func(t *testing.T) {
+		items := []*eventMessage{
+			eventMsg(SpecUpdate, "1"),
+			eventMsg(SpecUpdate, "2"),
+			eventMsg(SpecUpdate, "3"),
+		}
+		deduplicateEventMessageItems(&items)
+		require.Len(t, items, 1)
+		require.Contains(t, EventID(items[0].event), "_3")
+	})
+
+	t.Run("multiple status updates keep only freshest", func(t *testing.T) {
+		items := []*eventMessage{
+			eventMsg(StatusUpdate, "10"),
+			eventMsg(StatusUpdate, "20"),
+		}
+		deduplicateEventMessageItems(&items)
+		require.Len(t, items, 1)
+		require.Contains(t, EventID(items[0].event), "_20")
+	})
+
+	t.Run("spec and status updates both retained", func(t *testing.T) {
+		items := []*eventMessage{
+			eventMsg(SpecUpdate, "1"),
+			eventMsg(StatusUpdate, "2"),
+		}
+		deduplicateEventMessageItems(&items)
+		require.Len(t, items, 2)
+		require.Equal(t, SpecUpdate.String(), items[0].event.Type())
+		require.Equal(t, StatusUpdate.String(), items[1].event.Type())
+	})
+
+	t.Run("interleaved create leaves creates alone but dedupes spec updates", func(t *testing.T) {
+		items := []*eventMessage{
+			eventMsg(SpecUpdate, "1"),
+			eventMsg(Create, "2"),
+			eventMsg(SpecUpdate, "3"),
+		}
+		deduplicateEventMessageItems(&items)
+		require.Len(t, items, 2)
+		require.Equal(t, Create.String(), items[0].event.Type())
+		require.Equal(t, SpecUpdate.String(), items[1].event.Type())
+		require.Contains(t, EventID(items[1].event), "_3")
+	})
+
+	t.Run("does not dedupe event types other than spec or status update", func(t *testing.T) {
+		items := []*eventMessage{
+			eventMsg(Create, "1"),
+			eventMsg(Create, "2"),
+		}
+		deduplicateEventMessageItems(&items)
+		require.Len(t, items, 2)
+		require.Contains(t, EventID(items[0].event), "_1")
+		require.Contains(t, EventID(items[1].event), "_2")
+	})
+
+	t.Run("keeps latest of each deduped type when interleaved", func(t *testing.T) {
+		items := []*eventMessage{
+			eventMsg(SpecUpdate, "1"),
+			eventMsg(SpecUpdate, "2"),
+			eventMsg(StatusUpdate, "3"),
+			eventMsg(SpecUpdate, "4"),
+		}
+		deduplicateEventMessageItems(&items)
+		require.Len(t, items, 2)
+		require.Equal(t, StatusUpdate.String(), items[0].event.Type())
+		require.Contains(t, EventID(items[0].event), "_3")
+		require.Equal(t, SpecUpdate.String(), items[1].event.Type())
+		require.Contains(t, EventID(items[1].event), "_4")
+	})
+}
+
 type fakeStream struct {
 	mu     sync.RWMutex
 	events map[string][]string