From 3067841f1431ca1e2d58b452201b24fb6cb7939d Mon Sep 17 00:00:00 2001 From: Benoit Sigoure Date: Mon, 17 Aug 2026 15:42:53 +0000 Subject: [PATCH] watcher: deliver events to consumers in order, without drops Events were fanned out to each consumer in a fresh goroutine per event (go c.Send(payload)), so two events emitted back to back could be delivered in reverse order. Instance deletion emits update(status= deleted) followed by delete; when a consumer observed them reversed, the late update re-inserted an already-deleted instance into event- driven caches, permanently leaking it (observed in prod: 819 phantom instances after 9 days, dashboard reporting 895 instances vs 77 real). Consumer channels were also buffered at 1 with a 1s send timeout that silently dropped events during bursts, with only a debug log. Replace the per-event goroutine with a synchronous enqueue into an ordered per-consumer queue, drained by a dedicated dispatch goroutine. Send never blocks on a slow consumer and never drops events; a warning is logged if a consumer's queue depth keeps growing. The dispatch loop now owns closing the messages channel, so a send on a closed channel cannot happen. Add regression tests asserting in-order delivery and no drops under burst with a slow consumer. --- database/watcher/consumer.go | 72 +++++++++++---- database/watcher/ordering_test.go | 141 ++++++++++++++++++++++++++++++ database/watcher/watcher.go | 8 +- 3 files changed, 203 insertions(+), 18 deletions(-) create mode 100644 database/watcher/ordering_test.go diff --git a/database/watcher/consumer.go b/database/watcher/consumer.go index ed0967e92..7b4bf1d52 100644 --- a/database/watcher/consumer.go +++ b/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: + } } } diff --git a/database/watcher/ordering_test.go b/database/watcher/ordering_test.go new file mode 100644 index 000000000..a3b25b47e --- /dev/null +++ b/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) + } +} diff --git a/database/watcher/watcher.go b/database/watcher/watcher.go index 804dec70e..36c6934bb 100644 --- a/database/watcher/watcher.go +++ b/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 }