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

Watch flushed the alert queue in the scan goroutine. With a hung ntfy
(accepts, never answers until the 10 s timeout) and a full queue, one
flush took ~6 notify timeouts, so a planted symlink stayed readable for
several intervals instead of at most one.

Delivery now runs in its own goroutine that owns the queue. The scan
hands alerts over on a bounded channel without blocking; an alert that
does not fit is counted in the never-dropped summary, so the
"delivered or counted" property holds.

Repro test (interval 50 ms, notifier latency 40 ms, full queue): worst
link lifetime 243 ms before, under 2 intervals after.

Closes #164

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

ingestion/internal/symlinkguard/symlinkguard.go

@@ -18,6 +18,8 @@ import (
 	"sort"
 	"strconv"
 	"strings"
+	"sync"
+	"sync/atomic"
 	"time"
 )
 
@@ -133,13 +135,25 @@ 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.
 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) }()
+	defer wg.Wait()
+	q := alertSender{alerts: alerts, overflow: &overflow}
 	reported := ""
 	for {
 		select {
@@ -168,7 +182,49 @@ 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 queues alerts as they arrive and flushes
+// every interval until ctx ends.
+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{}
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case msg := <-alerts:
+			q.push(msg)
+		case <-t.C:
+			for drained := false; !drained; {
+				select {
+				case msg := <-alerts:
+					q.push(msg)
+				default:
+					drained = true
+				}
+			}
+			q.suppressed += int(overflow.Swap(0))
+			q.flush(ctx, n)
+		}
 	}
 }
 

ingestion/internal/symlinkguard/symlinkguard_test.go

@@ -496,3 +496,42 @@ func TestWatchDropsAPoisonMessageAndCountsIt(t *testing.T) {
 		return false
 	}, 3*time.Second, 5*time.Millisecond, "the dropped poison alert is counted in a summary")
 }
+
+// hungNotifier answers every Notify only after latency, with an outage error:
+// an ntfy that accepts connections but never responds until its timeout.
+type hungNotifier struct{ latency time.Duration }
+
+func (h hungNotifier) Notify(ctx context.Context, _ string) error {
+	select {
+	case <-time.After(h.latency):
+	case <-ctx.Done():
+	}
+	return errors.New("ntfy timeout")
+}
+
+// hyperguild#164: alerts were flushed in the scan goroutine, so with a hung
+// ntfy and a full queue one flush (6 x latency) delayed the next scan and a
+// planted link stayed for ~5 intervals. Delivery now runs beside the scan.
+func TestWatchLinkLifetimeStaysUnderTwoIntervalsWithAHungNotifier(t *testing.T) {
+	const interval = 50 * time.Millisecond
+	brain := t.TempDir()
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go Watch(ctx, brain, interval, hungNotifier{latency: 40 * time.Millisecond})
+
+	// Fill the queue: one removal alert per tick, more than maxPending.
+	for i := 0; i < maxPending+3; i++ {
+		link := filepath.Join(brain, "fill"+strconv.Itoa(i))
+		require.NoError(t, os.Symlink(t.TempDir(), link))
+		require.Eventually(t, func() bool { _, err := os.Lstat(link); return os.IsNotExist(err) }, 5*time.Second, time.Millisecond)
+	}
+	var worst time.Duration
+	for i := 0; i < 8; i++ {
+		link := filepath.Join(brain, "probe"+strconv.Itoa(i))
+		require.NoError(t, os.Symlink(t.TempDir(), link))
+		planted := time.Now()
+		require.Eventually(t, func() bool { _, err := os.Lstat(link); return os.IsNotExist(err) }, 5*time.Second, time.Millisecond)
+		worst = max(worst, time.Since(planted))
+	}
+	assert.LessOrEqual(t, worst, 2*interval, "worst link lifetime %v with a hung notifier", worst)
+}