fix(symlinkguard): deliver alerts beside the scan, not inside it (#164) (#180)

Co-authored-by: Mathias <mthbqv@gmail.com>

ingestion/internal/symlinkguard/symlinkguard.go

@@ -18,9 +18,14 @@ import (
 	"sort"
 	"strconv"
 	"strings"
+	"sync"
+	"sync/atomic"
 	"time"
 )
 
+// shutdownFlushTimeout bounds the final flush when Watch stops.
+var shutdownFlushTimeout = 5 * time.Second
+
 // DefaultInterval is used when Watch is given a non-positive interval.
 const DefaultInterval = 30 * time.Second
 
@@ -133,13 +138,32 @@ func Scan(brainDir string) (Result, error) {
 // queue until Notify succeeds (see alertQueue): an outage delays them, and
 // any that do not fit the bounded queue, or are undeliverable, are counted in
 // a summary that is never dropped, so every incident is delivered or counted.
+//
+// Delivery runs in its own goroutine (hyperguild#164): a hung notifier, with
+// a full queue, blocked one flush for several notify timeouts, and while the
+// scan waited on it a planted link stayed readable for several intervals.
+// The scan hands alerts over without blocking; any that do not fit are
+// counted in the summary. When ctx ends, Watch returns after a final flush
+// of what was handed over (bounded by shutdownFlushTimeout); alerts that
+// flush cannot deliver are counted in one error log line instead. That
+// shutdown path only runs for a caller that cancels ctx: cmd/server passes
+// context.Background(), so in production the flush on handoff is what keeps
+// alerts from waiting on a restart.
 func Watch(ctx context.Context, brainDir string, interval time.Duration, n Notifier) {
 	if interval <= 0 {
 		interval = DefaultInterval
 	}
 	t := time.NewTicker(interval)
 	defer t.Stop()
-	q := &alertQueue{}
+	alerts := make(chan string, maxPending)
+	var overflow atomic.Int64
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() { defer wg.Done(); deliver(ctx, interval, n, alerts, &overflow) }()
+	// The scan is the only sender: closing alerts tells deliver to drain and
+	// flush once more, and Watch returns only after it has.
+	defer func() { close(alerts); wg.Wait() }()
+	q := alertSender{alerts: alerts, overflow: &overflow}
 	reported := ""
 	for {
 		select {
@@ -168,7 +192,76 @@ func Watch(ctx context.Context, brainDir string, interval time.Duration, n Notif
 					". Readers may follow symlinks out of the brain; fix them.")
 			}
 		}
-		q.flush(ctx, n)
+	}
+}
+
+// alertSender hands alerts to deliver without ever blocking the scan; an
+// alert that does not fit is counted in overflow, which deliver adds to the
+// summary.
+type alertSender struct {
+	alerts   chan<- string
+	overflow *atomic.Int64
+}
+
+func (s alertSender) push(msg string) {
+	select {
+	case s.alerts <- msg:
+	default:
+		s.overflow.Add(1)
+	}
+}
+
+// deliver owns the alertQueue. It flushes as soon as an alert is handed
+// over, and again every interval while any are pending (outage retries).
+// When alerts is closed (Watch is stopping) it drains it and flushes once
+// more, bounded by shutdownFlushTimeout, so an alert handed over just before
+// shutdown is not lost (PR #180 review). Whatever that flush cannot deliver
+// is logged as one error with its counts, never dropped silently.
+func deliver(ctx context.Context, interval time.Duration, n Notifier, alerts <-chan string, overflow *atomic.Int64) {
+	t := time.NewTicker(interval)
+	defer t.Stop()
+	q := &alertQueue{}
+	// drain moves everything handed over into the queue; it reports false
+	// once alerts is closed.
+	drain := func() bool {
+		defer func() { q.suppressed += int(overflow.Swap(0)) }()
+		for {
+			select {
+			case msg, ok := <-alerts:
+				if !ok {
+					return false
+				}
+				q.push(msg)
+			default:
+				return true
+			}
+		}
+	}
+	for {
+		select {
+		case msg, ok := <-alerts:
+			if ok {
+				q.push(msg)
+				ok = drain()
+			}
+			if !ok {
+				q.suppressed += int(overflow.Swap(0))
+				fctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), shutdownFlushTimeout)
+				q.flush(fctx, n)
+				cancel()
+				if len(q.items) > 0 || q.suppressed > 0 {
+					slog.Error("symlink tripwire alerts not delivered at shutdown",
+						"undelivered", len(q.items), "suppressed", q.suppressed)
+				}
+				return
+			}
+			q.flush(ctx, n)
+		case <-t.C:
+			if !drain() {
+				continue // the closed channel is handled by the case above
+			}
+			q.flush(ctx, n)
+		}
 	}
 }