From 52d4bfeeab2e5c84f52bba4b217cac74a6074c3a Mon Sep 17 00:00:00 2001 From: Mathias Date: Mon, 5 Oct 2026 09:45:15 +0200 Subject: [PATCH] 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 --- .../internal/symlinkguard/symlinkguard.go | 59 ++++-- .../symlinkguard/symlinkguard_test.go | 177 ++++++++++++++++-- 2 files changed, 207 insertions(+), 29 deletions(-) diff --git a/ingestion/internal/symlinkguard/symlinkguard.go b/ingestion/internal/symlinkguard/symlinkguard.go index b38e2e7..6aaa480 100644 --- a/ingestion/internal/symlinkguard/symlinkguard.go +++ b/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) } } diff --git a/ingestion/internal/symlinkguard/symlinkguard_test.go b/ingestion/internal/symlinkguard/symlinkguard_test.go index 721b1b0..56fba71 100644 --- a/ingestion/internal/symlinkguard/symlinkguard_test.go +++ b/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") }