avoid spamming notification dispatch

johnstcn · johnstcn · commit 5a1bcb30723a · 2025-04-22T17:56:00.000+01:00
diff --git a/coderd/notifications/notifications_test.go b/coderd/notifications/notifications_test.go
@@ -2149,6 +2149,8 @@ func TestNotificationEnqueuePubsubNotify(t *testing.T) {
 		}
 	}()
 	_ = testutil.TryReceive(ctx, t, recvDone)
+	// TODO: this sometimes fails with
+	//     t.go:106: 2025-04-22 16:55:04.153 [warn]  manager: content canceled with pending updates in buffer, these messages will be sent again after lease expires  success_count=6  failure_count=0
 }
 
 type fakeHandler struct {
diff --git a/coderd/notifications/notifier.go b/coderd/notifications/notifier.go
@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"sync"
+	"sync/atomic"
 	"text/template"
 
 	"github.com/google/uuid"
@@ -134,8 +135,13 @@ func (n *notifier) run(success chan<- dispatchResult, failure chan<- dispatchRes
 		}
 	}()
 
+	// Keep track of how many notification_enqueued events seen this loop to avoid
+	// unnecessary database load.
+	var enqueueEventsThisLoop atomic.Int64
+
 	// Periodically trigger the processing loop.
 	tick := n.clock.TickerFunc(n.gracefulCtx, n.cfg.FetchInterval.Value(), func() error {
+		defer enqueueEventsThisLoop.Store(0)
 		c := make(chan struct{})
 		loopTick <- c
 		// Wait for the processing to finish before continuing. The ticker will
@@ -146,7 +152,12 @@ func (n *notifier) run(success chan<- dispatchResult, failure chan<- dispatchRes
 
 	// Also signal the processing loop when a notification is enqueued.
 	if stopListen, err := n.ps.Subscribe(EventNotificationEnqueued, func(ctx context.Context, _ []byte) {
-		n.log.Debug(n.outerCtx, "got pubsub event", slog.F("event", EventNotificationEnqueued))
+		enqueued := enqueueEventsThisLoop.Add(1)
+		skipEarlyDispatch := enqueued > 1
+		n.log.Debug(n.outerCtx, "got pubsub event", slog.F("count", enqueued), slog.F("skip_early_dispatch", skipEarlyDispatch), slog.F("event", EventNotificationEnqueued))
+		if enqueued > 1 {
+			return
+		}
 		c := make(chan struct{})
 		select {
 		case <-n.gracefulCtx.Done():

Original file line number	Diff line number	Diff line change
`@@ -2149,6 +2149,8 @@ func TestNotificationEnqueuePubsubNotify(t *testing.T) {`
`2149`	`2149`	`}`
`2150`	`2150`	`}()`
`2151`	`2151`	`_ = testutil.TryReceive(ctx, t, recvDone)`
	`2152`	`+ // TODO: this sometimes fails with`
	`2153`	`+ // t.go:106: 2025-04-22 16:55:04.153 [warn] manager: content canceled with pending updates in buffer, these messages will be sent again after lease expires success_count=6 failure_count=0`
`2152`	`2154`	`}`
`2153`	`2155`
`2154`	`2156`	`type fakeHandler struct {`