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
-// every interval until ctx ends.
+// 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).
 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 <-ctx.Done():
-			return
-		case msg := <-alerts:
-			q.push(msg)
+		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:
-			for drained := false; !drained; {
-				select {
-				case msg := <-alerts:
-					q.push(msg)
-				default:
-					drained = true
-				}
+			if !drain() {
+				continue // the closed channel is handled by the case above
 			}
-			q.suppressed += int(overflow.Swap(0))
 			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:
-// an ntfy that accepts connections but never responds until its timeout.
+// hungNotifier answers every Notify only after latency (or when ctx ends),
+// 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
-// planted link stayed for ~5 intervals. Delivery now runs beside the scan.
-func TestWatchLinkLifetimeStaysUnderTwoIntervalsWithAHungNotifier(t *testing.T) {
-	const interval = 50 * time.Millisecond
+// ntfy and a full queue one flush (6 x notify timeout) delayed the next scan.
+// The notifier here is slower than the interval, as in production (10 s
+// timeout vs 30 s interval with 6 messages), so the alert handoff is full:
+// 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 < maxPending+3; i++ {
+	for i := 0; i < 2*maxPending+3; i++ { // overfill queue and handoff
 		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")
 }