watcher: deliver events to consumers in order, without drops

database/watcher/consumer.go

@@ -18,20 +18,26 @@ import (
 	"context"
 	"log/slog"
 	"sync"
-	"time"
 
 	"github.com/cloudbase/garm/database/common"
 )
 
+// queueWarnThreshold is the pending-queue depth at which we start warning
+// that a consumer is not keeping up with the event stream. Events are never
+// dropped; this only makes a slow consumer visible.
+const queueWarnThreshold = 512
+
 type consumer struct {
 	messages chan common.ChangePayload
 	filters  []common.PayloadFilterFunc
 	id       string
 
-	mux    sync.Mutex
-	closed bool
-	quit   chan struct{}
-	ctx    context.Context
+	mux     sync.Mutex
+	cond    *sync.Cond
+	pending []common.ChangePayload
+	closed  bool
+	quit    chan struct{}
+	ctx     context.Context
 }
 
 func (w *consumer) SetFilters(filters ...common.PayloadFilterFunc) {
@@ -50,9 +56,11 @@ func (w *consumer) Close() {
 	if w.closed {
 		return
 	}
-	close(w.messages)
 	close(w.quit)
 	w.closed = true
+	// Wake the dispatch loop so it notices the closed state and closes
+	// the messages channel.
+	w.cond.Broadcast()
 }
 
 func (w *consumer) IsClosed() bool {
@@ -61,6 +69,11 @@ func (w *consumer) IsClosed() bool {
 	return w.closed
 }
 
+// Send enqueues a payload for delivery to this consumer. It never blocks on
+// the consumer and never drops events: payloads are appended to an ordered
+// queue drained by the dispatch loop. Callers invoking Send sequentially are
+// guaranteed in-order delivery, which consumers rely on (e.g. an instance
+// update followed by its delete must not be observed in reverse).
 func (w *consumer) Send(payload common.ChangePayload) {
 	w.mux.Lock()
 	defer w.mux.Unlock()
@@ -83,16 +96,41 @@ func (w *consumer) Send(payload common.ChangePayload) {
 		}
 	}
 
-	timer := time.NewTimer(1 * time.Second)
-	defer timer.Stop()
-	slog.DebugContext(w.ctx, "sending payload")
-	select {
-	case <-w.quit:
-		slog.DebugContext(w.ctx, "consumer is closed")
-	case <-w.ctx.Done():
-		slog.DebugContext(w.ctx, "consumer is closed")
-	case <-timer.C:
-		slog.DebugContext(w.ctx, "timeout trying to send payload", "payload", payload)
-	case w.messages <- payload:
+	w.pending = append(w.pending, payload)
+	if len(w.pending) >= queueWarnThreshold && len(w.pending)%queueWarnThreshold == 0 {
+		slog.WarnContext(w.ctx, "consumer is falling behind on events", "consumer_id", w.id, "pending_events", len(w.pending))
+	}
+	w.cond.Signal()
+}
+
+// dispatch drains the pending queue in order, delivering each payload to the
+// messages channel. It owns closing the messages channel: doing it here (and
+// only here) guarantees we never send on a closed channel.
+func (w *consumer) dispatch() {
+	defer close(w.messages)
+	for {
+		w.mux.Lock()
+		for len(w.pending) == 0 && !w.closed {
+			w.cond.Wait()
+		}
+		if w.closed {
+			w.mux.Unlock()
+			return
+		}
+		payload := w.pending[0]
+		w.pending = w.pending[1:]
+		if len(w.pending) == 0 {
+			// Don't pin the backing array of a previously grown queue.
+			w.pending = nil
+		}
+		w.mux.Unlock()
+
+		select {
+		case <-w.quit:
+			return
+		case <-w.ctx.Done():
+			return
+		case w.messages <- payload:
+		}
 	}
 }

database/watcher/ordering_test.go

@@ -0,0 +1,141 @@
+//go:build testing
+
+// Copyright 2025 Cloudbase Solutions SRL
+//
+//	Licensed under the Apache License, Version 2.0 (the "License"); you may
+//	not use this file except in compliance with the License. You may obtain
+//	a copy of the License at
+//
+//	     http://www.apache.org/licenses/LICENSE-2.0
+//
+//	Unless required by applicable law or agreed to in writing, software
+//	distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+//	WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+//	License for the specific language governing permissions and limitations
+//	under the License.
+package watcher_test
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/cloudbase/garm/database/common"
+	"github.com/cloudbase/garm/database/watcher"
+	"github.com/cloudbase/garm/params"
+)
+
+// setupWatcher initializes a fresh watcher for a test and returns a cleanup
+// function that tears it down.
+func setupWatcher(t *testing.T) func() {
+	t.Helper()
+	watcher.InitWatcher(context.TODO())
+	return func() {
+		if w := watcher.GetWatcher(); w != nil {
+			w.Close()
+			watcher.SetWatcher(nil)
+		}
+	}
+}
+
+// TestEventsAreDeliveredInOrder asserts that events published sequentially by
+// a producer are observed by a consumer in the exact same order. The previous
+// implementation dispatched each event to consumers in a new goroutine, which
+// allowed reordering (e.g. an instance delete observed before the update that
+// preceded it), leaking stale entries in event-driven caches.
+func TestEventsAreDeliveredInOrder(t *testing.T) {
+	defer setupWatcher(t)()
+
+	ctx := context.TODO()
+	prod, err := watcher.RegisterProducer(ctx, "order-producer")
+	require.NoError(t, err)
+	cons, err := watcher.RegisterConsumer(ctx, "order-consumer")
+	require.NoError(t, err)
+	defer cons.Close()
+
+	const numEvents = 2000
+	done := make(chan error, 1)
+	go func() {
+		for i := 0; i < numEvents; i++ {
+			op := common.UpdateOperation
+			if i%2 == 1 {
+				op = common.DeleteOperation
+			}
+			payload := common.ChangePayload{
+				EntityType: common.InstanceEntityType,
+				Operation:  op,
+				Payload: params.Instance{
+					Name: fmt.Sprintf("instance-%d", i),
+				},
+			}
+			if err := prod.Notify(payload); err != nil {
+				done <- err
+				return
+			}
+		}
+		done <- nil
+	}()
+
+	timeout := time.After(30 * time.Second)
+	for i := 0; i < numEvents; i++ {
+		select {
+		case event := <-cons.Watch():
+			instance, ok := event.Payload.(params.Instance)
+			require.True(t, ok)
+			require.Equal(t, fmt.Sprintf("instance-%d", i), instance.Name, "event %d delivered out of order", i)
+			expectedOp := common.UpdateOperation
+			if i%2 == 1 {
+				expectedOp = common.DeleteOperation
+			}
+			require.Equal(t, expectedOp, event.Operation)
+		case <-timeout:
+			t.Fatalf("timed out waiting for event %d", i)
+		}
+	}
+	require.NoError(t, <-done)
+}
+
+// TestSlowConsumerDoesNotDropEvents asserts that a consumer that is slower
+// than the producer still receives every event. The previous implementation
+// dropped events after a 1 second send timeout on a channel with a buffer of
+// one, silently losing events during bursts.
+func TestSlowConsumerDoesNotDropEvents(t *testing.T) {
+	defer setupWatcher(t)()
+
+	ctx := context.TODO()
+	prod, err := watcher.RegisterProducer(ctx, "burst-producer")
+	require.NoError(t, err)
+	cons, err := watcher.RegisterConsumer(ctx, "slow-consumer")
+	require.NoError(t, err)
+	defer cons.Close()
+
+	const numEvents = 200
+	for i := 0; i < numEvents; i++ {
+		payload := common.ChangePayload{
+			EntityType: common.InstanceEntityType,
+			Operation:  common.DeleteOperation,
+			Payload: params.Instance{
+				Name: fmt.Sprintf("instance-%d", i),
+			},
+		}
+		require.NoError(t, prod.Notify(payload))
+	}
+
+	timeout := time.After(60 * time.Second)
+	for i := 0; i < numEvents; i++ {
+		// Drain slowly relative to the burst above; every event must
+		// still arrive, in order.
+		select {
+		case event := <-cons.Watch():
+			instance, ok := event.Payload.(params.Instance)
+			require.True(t, ok)
+			require.Equal(t, fmt.Sprintf("instance-%d", i), instance.Name)
+		case <-timeout:
+			t.Fatalf("timed out waiting for event %d; events were dropped", i)
+		}
+		time.Sleep(time.Millisecond)
+	}
+}

database/watcher/watcher.go

@@ -119,8 +119,12 @@ func (w *watcher) serviceProducer(prod *producer) {
 			return
 		case payload := <-prod.messages:
 			w.mux.Lock()
+			// Send synchronously and in registration-independent but stable
+			// per-consumer order. Send only enqueues (it cannot block on a slow
+			// consumer), and calling it inline preserves event ordering for each
+			// consumer.
 			for _, c := range w.consumers {
-				go c.Send(payload)
+				c.Send(payload)
 			}
 			w.mux.Unlock()
 		}
@@ -140,7 +144,9 @@ func (w *watcher) RegisterConsumer(ctx context.Context, id string, filters ...co
 		id:       id,
 		ctx:      ctx,
 	}
+	c.cond = sync.NewCond(&c.mux)
 	w.consumers[id] = c
+	go c.dispatch()
 	go w.serviceConsumer(c)
 	return c, nil
 }