From 104b4802569fc8f1a9a6601152a3638ae4ccff29 Mon Sep 17 00:00:00 2001 From: jedarden Date: Mon, 4 May 2026 00:28:21 -0400 Subject: [PATCH] fix(timeline): replace unsafe drop-oldest with drop-new on queue overflow The drop-oldest path (drain + re-send) was not goroutine-safe: multiple concurrent EventBus delivery goroutines could each drain one slot and then all block waiting to re-send, causing inFlight.Wait() in Close() to deadlock. Drop-new is atomic via the select/default pattern and never blocks. Co-Authored-By: Claude Sonnet 4.6 --- mothership/internal/timeline/timeline.go | 31 ++++++------------------ 1 file changed, 8 insertions(+), 23 deletions(-) diff --git a/mothership/internal/timeline/timeline.go b/mothership/internal/timeline/timeline.go index f7ecf66..c19df45 100644 --- a/mothership/internal/timeline/timeline.go +++ b/mothership/internal/timeline/timeline.go @@ -71,8 +71,8 @@ func New(db *sql.DB) *Storage { return s } -// enqueue adds an event to the queue, dropping oldest if full. -// This is called from the EventBus subscriber callback and must never block. +// enqueue adds an event to the queue, dropping the incoming event if full. +// This is called from the EventBus subscriber goroutine and must never block. func (s *Storage) enqueue(e eventbus.Event) { select { case s.queue <- Event{ @@ -86,34 +86,19 @@ func (s *Storage) enqueue(e eventbus.Event) { }: // Event queued successfully default: - // Queue is full, drop oldest event + // Queue is full — drop this event and increment the counter. + // Drop-oldest via drain+send is not safe here: multiple goroutines + // can concurrently drain a slot and then block on the re-send. s.mu.Lock() s.dropped++ - s.mu.Unlock() - - // Warn about overflow at most once per minute - now := time.Now() - s.mu.Lock() - shouldWarn := now.Sub(s.lastWarn) > time.Minute + shouldWarn := time.Since(s.lastWarn) > time.Minute if shouldWarn { - s.lastWarn = now + s.lastWarn = time.Now() } s.mu.Unlock() if shouldWarn { - log.Printf("[WARN] Timeline storage queue overflow (dropped oldest event, %d total dropped)", s.dropped) - } - - // Drop oldest by receiving and discarding one, then sending the new one - <-s.queue - s.queue <- Event{ - Type: e.Type, - TimestampMs: e.TimestampMs, - Zone: e.Zone, - Person: e.Person, - BlobID: e.BlobID, - Detail: e.Detail, - Severity: e.Severity, + log.Printf("[WARN] Timeline storage queue overflow (dropped event, %d total dropped)", s.dropped) } } }