fix(symlinkguard): flush on handoff and at shutdown; test the handoff

PR #180 review (BLOCK) found that moving delivery out of the scan left
two problems:
- An alert waited for deliver's own tick, about one interval. It is
  now flushed as soon as it is handed over, and on each tick after
  that while it is still pending.
- On ctx end deliver returned without draining, so an alert handed over
  just before shutdown was lost even with a healthy ntfy. Watch now
  closes the handoff channel, and deliver drains it and flushes once
  more (bounded by shutdownFlushTimeout) before Watch returns.

The review also showed the tests didn't cover the handoff. The new
tests are each flip-tested red:
- the lifetime test uses a notifier slower than the interval, so a
  blocking handoff fails it
- the overflow-accounting test uses a slow notifier
- a join test checks that Notify is no longer running when Watch
  returns
- a shutdown-flush test, and a prompt-delivery test

Refs #164

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

ingestion/internal/symlinkguard/symlinkguard.go

@@ -23,6 +23,9 @@ import (
 	"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
 
@@ -140,7 +143,8 @@ func Scan(brainDir string) (Result, error) {
 // 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.
+// counted in the summary. When ctx ends, Watch returns after a final flush
+// of what was handed over (bounded by shutdownFlushTimeout).
 func Watch(ctx context.Context, brainDir string, interval time.Duration, n Notifier) {
 	if interval <= 0 {
 		interval = DefaultInterval
@@ -152,7 +156,9 @@ func Watch(ctx context.Context, brainDir string, interval time.Duration, n Notif
 	var wg sync.WaitGroup
 	wg.Add(1)
 	go func() { defer wg.Done(); deliver(ctx, interval, n, alerts, &overflow) }()
-	defer wg.Wait()
+	// 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 {
@@ -201,28 +207,49 @@ func (s alertSender) push(msg string) {
 	}
 }
 
-// deliver owns the alertQueue: it queues alerts as they arrive and flushes
+// deliver owns the alertQueue. It flushes as soon as an alert is handed
-// every interval until ctx ends.
+// 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).
 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 <-ctx.Done():
+			case msg, ok := <-alerts:
-			return
+				if !ok {
-		case msg := <-alerts:
+					return false
-			q.push(msg)
+				}
-		case <-t.C:
-			for drained := false; !drained; {
-				select {
-				case msg := <-alerts:
 				q.push(msg)
 			default:
-					drained = true
+				return true
 			}
 		}
-			q.suppressed += int(overflow.Swap(0))
+	}
+	for {
+		select {
+		case msg, ok := <-alerts:
+			if ok {
+				q.push(msg)
+				ok = drain()
+			}
+			if !ok {
+				fctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), shutdownFlushTimeout)
+				q.flush(fctx, n)
+				cancel()
+				return
+			}
+			q.flush(ctx, n)
+		case <-t.C:
+			if !drain() {
+				continue // the closed channel is handled by the case above
+			}
 			q.flush(ctx, n)
 		}
 	}

ingestion/internal/symlinkguard/symlinkguard_test.go

@@ -497,8 +497,9 @@ func TestWatchDropsAPoisonMessageAndCountsIt(t *testing.T) {
 	}, 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:
+// hungNotifier answers every Notify only after latency (or when ctx ends),
-// an ntfy that accepts connections but never responds until its timeout.
+// 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 {
@@ -509,29 +510,179 @@ func (h hungNotifier) Notify(ctx context.Context, _ string) error {
 	return errors.New("ntfy timeout")
 }
 
+func waitGone(t *testing.T, link string) {
+	t.Helper()
+	require.Eventually(t, func() bool { _, err := os.Lstat(link); return os.IsNotExist(err) }, 10*time.Second, time.Millisecond)
+}
+
 // 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
+// ntfy and a full queue one flush (6 x notify timeout) delayed the next scan.
-// planted link stayed for ~5 intervals. Delivery now runs beside the scan.
+// The notifier here is slower than the interval, as in production (10 s
-func TestWatchLinkLifetimeStaysUnderTwoIntervalsWithAHungNotifier(t *testing.T) {
+// timeout vs 30 s interval with 6 messages), so the alert handoff is full:
-	const interval = 50 * time.Millisecond
+// a scan that waited on it would keep a link for seconds.
+func TestWatchLinkLifetimeStaysNearOneIntervalWithANotifierSlowerThanTheInterval(t *testing.T) {
+	const interval = 100 * time.Millisecond
 	brain := t.TempDir()
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
-	go Watch(ctx, brain, interval, hungNotifier{latency: 40 * time.Millisecond})
+	go Watch(ctx, brain, interval, hungNotifier{latency: 400 * time.Millisecond})
 
-	// Fill the queue: one removal alert per tick, more than maxPending.
+	for i := 0; i < 2*maxPending+3; i++ { // overfill queue and handoff
-	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)
+		waitGone(t, link)
 	}
 	var worst time.Duration
-	for i := 0; i < 8; i++ {
+	for i := 0; i < 6; 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)
+		waitGone(t, link)
 		worst = max(worst, time.Since(planted))
 	}
-	assert.LessOrEqual(t, worst, 2*interval, "worst link lifetime %v with a hung notifier", worst)
+	assert.LessOrEqual(t, worst, interval+interval/2, "worst link lifetime %v with a hung notifier", worst)
+}
+
+// Review of PR #180: delivery waited for its own tick, about one interval
+// after the removal. An alert is now flushed as soon as it is handed over.
+func TestWatchDeliversARemovalAlertWellWithinOneInterval(t *testing.T) {
+	const interval = 400 * time.Millisecond
+	brain := t.TempDir()
+	n := &recNotifier{}
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go Watch(ctx, brain, interval, n)
+	link := filepath.Join(brain, "prompt")
+	require.NoError(t, os.Symlink(t.TempDir(), link))
+	waitGone(t, link)
+	removed := time.Now()
+	require.Eventually(t, func() bool { return n.count() == 1 }, 2*time.Second, time.Millisecond)
+	assert.Less(t, time.Since(removed), interval/4)
+}
+
+// Review of PR #180: on ctx end, deliver returned without draining what the
+// scan had handed over, so a removal just before shutdown was neither
+// delivered nor counted. Watch now drains and flushes before returning.
+func TestWatchDeliversAnAlertHandedOverJustBeforeShutdown(t *testing.T) {
+	for i := 0; i < 20; i++ {
+		brain := t.TempDir()
+		n := &recNotifier{}
+		ctx, cancel := context.WithCancel(context.Background())
+		done := make(chan struct{})
+		go func() { Watch(ctx, brain, 5*time.Millisecond, n); close(done) }()
+		link := filepath.Join(brain, "late")
+		require.NoError(t, os.Symlink(t.TempDir(), link))
+		waitGone(t, link)
+		cancel()
+		<-done
+		require.Equal(t, 1, n.count(), "iteration %d: the removal alert was lost at shutdown", i)
+	}
+}
+
+// blockingNotifier holds every Notify until its ctx ends, tracking calls in
+// flight.
+type blockingNotifier struct{ inFlight atomic.Int32 }
+
+func (b *blockingNotifier) Notify(ctx context.Context, _ string) error {
+	b.inFlight.Add(1)
+	defer b.inFlight.Add(-1)
+	<-ctx.Done()
+	return ctx.Err()
+}
+
+// Watch does not return while its delivery goroutine is still notifying.
+func TestWatchReturnsOnlyAfterDeliveryHasStopped(t *testing.T) {
+	prev := shutdownFlushTimeout
+	shutdownFlushTimeout = 50 * time.Millisecond
+	t.Cleanup(func() { shutdownFlushTimeout = prev })
+	brain := t.TempDir()
+	n := &blockingNotifier{}
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan struct{})
+	go func() { Watch(ctx, brain, 5*time.Millisecond, n); close(done) }()
+	link := filepath.Join(brain, "x")
+	require.NoError(t, os.Symlink(t.TempDir(), link))
+	waitGone(t, link)
+	require.Eventually(t, func() bool { return n.inFlight.Load() == 1 }, 2*time.Second, time.Millisecond)
+	cancel()
+	select {
+	case <-done:
+	case <-time.After(5 * time.Second):
+		t.Fatal("Watch did not return after cancel")
+	}
+	assert.Zero(t, n.inFlight.Load(), "Notify still running after Watch returned")
+}
+
+// slowToggleNotifier fails slowly (a hung ntfy) while down, then records.
+type slowToggleNotifier struct {
+	recNotifier
+	down    atomic.Bool
+	latency time.Duration
+}
+
+func (r *slowToggleNotifier) Notify(ctx context.Context, m string) error {
+	if r.down.Load() {
+		select {
+		case <-time.After(r.latency):
+		case <-ctx.Done():
+		}
+		return errors.New("ntfy timeout")
+	}
+	return r.recNotifier.Notify(ctx, m)
+}
+
+// Review of PR #180: with a slow notifier the scan's handoff overflows, and
+// those alerts exist only in the overflow counter. Every removal is still
+// delivered or counted in the summary.
+func TestWatchCountsHandoffOverflowWithASlowNotifier(t *testing.T) {
+	brain := t.TempDir()
+	n := &slowToggleNotifier{latency: 30 * time.Millisecond}
+	n.down.Store(true)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go Watch(ctx, brain, 2*time.Millisecond, n)
+	const plants = 40
+	for i := 0; i < plants; i++ {
+		link := filepath.Join(brain, "ov"+strconv.Itoa(i))
+		require.NoError(t, os.Symlink(t.TempDir(), link))
+		waitGone(t, link)
+	}
+	n.down.Store(false)
+	delivered, counted := 0, 0
+	require.Eventually(t, func() bool {
+		n.mu.Lock()
+		defer n.mu.Unlock()
+		delivered, counted = 0, 0
+		for _, m := range n.msgs {
+			if strings.Contains(m, "removed 1 symlink(s)") {
+				delivered++
+			}
+			if i := strings.Index(m, "): "); i >= 0 && strings.Contains(m, "suppressed") {
+				var c int
+				_, _ = fmt.Sscanf(m[i+3:], "%d", &c)
+				counted += c
+			}
+		}
+		return delivered+counted == plants
+	}, 5*time.Second, 10*time.Millisecond, "delivered %d + counted %d of %d", delivered, counted, plants)
+}
+
+// An alert still queued when Watch stops (its send failed, the notifier has
+// since recovered, the retry tick has not come) is flushed once more before
+// Watch returns.
+func TestWatchFlushesTheQueueOnceMoreAtShutdown(t *testing.T) {
+	brain := t.TempDir()
+	n := &toggleNotifier{}
+	n.down.Store(true)
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan struct{})
+	go func() { Watch(ctx, brain, 300*time.Millisecond, n); close(done) }()
+	link := filepath.Join(brain, "queued")
+	require.NoError(t, os.Symlink(t.TempDir(), link))
+	waitGone(t, link)
+	time.Sleep(20 * time.Millisecond) // let the immediate send fail
+	n.down.Store(false)
+	cancel()
+	<-done
+	assert.Equal(t, 1, n.count(), "the queued alert was not flushed at shutdown")
 }