From 8c48e2e343501c161b1b2cbdaefa07ab76526e4e Mon Sep 17 00:00:00 2001 From: Mathias Date: Mon, 5 Oct 2026 14:48:16 +0000 Subject: [PATCH] fix(symlinkguard): deliver alerts beside the scan, not inside it (#164) (#180) Co-authored-by: Mathias --- .../internal/symlinkguard/symlinkguard.go | 97 +++++- .../symlinkguard/symlinkguard_test.go | 278 +++++++++++++++++- 2 files changed, 372 insertions(+), 3 deletions(-) diff --git a/ingestion/internal/symlinkguard/symlinkguard.go b/ingestion/internal/symlinkguard/symlinkguard.go index e8c8984..b5bd5da 100644 --- a/ingestion/internal/symlinkguard/symlinkguard.go +++ b/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) + } } } diff --git a/ingestion/internal/symlinkguard/symlinkguard_test.go b/ingestion/internal/symlinkguard/symlinkguard_test.go index 36fe13d..3bb1ac3 100644 --- a/ingestion/internal/symlinkguard/symlinkguard_test.go +++ b/ingestion/internal/symlinkguard/symlinkguard_test.go @@ -1,10 +1,12 @@ package symlinkguard import ( + "bytes" "context" "errors" "fmt" "os" + "log/slog" "path/filepath" "strconv" "strings" @@ -345,13 +347,19 @@ func TestWatchDropsAMessageThatNeverSendsSoLaterAlertsFlow(t *testing.T) { }, 3*time.Second, 10*time.Millisecond, "a later alert is delivered despite a poison message") } -// toggleNotifier fails while down is set. +// toggleNotifier fails while down is set. Like the real audit.NtfyNotifier +// (http.NewRequestWithContext), it also fails, without recording, when its +// ctx is already done (PR #180 round 2: notifiers that ignored ctx let a +// final flush on the cancelled ctx pass). type toggleNotifier struct { recNotifier down atomic.Bool } func (r *toggleNotifier) Notify(ctx context.Context, m string) error { + if err := ctx.Err(); err != nil { + return err + } if r.down.Load() { return errors.New("ntfy down") } @@ -496,3 +504,271 @@ 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 (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 { + select { + case <-time.After(h.latency): + case <-ctx.Done(): + } + 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 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: 400 * time.Millisecond}) + + 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)) + waitGone(t, link) + } + var worst time.Duration + 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() + waitGone(t, link) + worst = max(worst, time.Since(planted)) + } + 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, on a ctx that is not the cancelled one: toggleNotifier fails +// on a done ctx, as the real ntfy notifier does. +func TestWatchFinalFlushSendsAQueuedAlertOnACtxThatOutlivesTheCancel(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") +} + +// lockedBuffer is a goroutine-safe log sink: Watch goroutines of earlier +// tests may still log while one test captures. +type lockedBuffer struct { + mu sync.Mutex + buf bytes.Buffer +} + +func (b *lockedBuffer) Write(p []byte) (int, error) { + b.mu.Lock() + defer b.mu.Unlock() + return b.buf.Write(p) +} + +func (b *lockedBuffer) String() string { b.mu.Lock(); defer b.mu.Unlock(); return b.buf.String() } + +func captureLog(t *testing.T) *lockedBuffer { + t.Helper() + b := &lockedBuffer{} + prev := slog.Default() + slog.SetDefault(slog.New(slog.NewTextHandler(b, nil))) + t.Cleanup(func() { slog.SetDefault(prev) }) + return b +} + +// PR #180 round 2: when the final flush fails (ntfy hung at shutdown) the +// undelivered alerts vanished with only "will retry" warnings. Watch now logs +// one error with both counts. +func TestWatchLogsUndeliveredAndSuppressedCountsWhenTheFinalFlushFails(t *testing.T) { + prev := shutdownFlushTimeout + shutdownFlushTimeout = 50 * time.Millisecond + t.Cleanup(func() { shutdownFlushTimeout = prev }) + logs := captureLog(t) + brain := t.TempDir() + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { Watch(ctx, brain, 5*time.Millisecond, hungNotifier{latency: time.Hour}); close(done) }() + link := filepath.Join(brain, "lost") + require.NoError(t, os.Symlink(t.TempDir(), link)) + waitGone(t, link) + cancel() + <-done + out := logs.String() + assert.Contains(t, out, "level=ERROR") + assert.Contains(t, out, `msg="symlink tripwire alerts not delivered at shutdown"`) + assert.Contains(t, out, "undelivered=1") + assert.Contains(t, out, "suppressed=0") +} + +// deliver, given a handoff that already holds alerts and is closed, moves them +// into the queue and runs the final flush, even though the ctx is cancelled: +// the drain-on-close path, without depending on when the close lands. +func TestDeliverFlushesAlertsDrainedFromAClosedHandoffOnACancelledCtx(t *testing.T) { + alerts := make(chan string, 3) + alerts <- "a1" + alerts <- "a2" + close(alerts) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + n := &toggleNotifier{} + var overflow atomic.Int64 + deliver(ctx, time.Hour, n, alerts, &overflow) + assert.Equal(t, []string{"a1", "a2"}, n.msgs) +} + +// PR #180 round 2: when the first receive after close finds the handoff empty, +// overflow counted by the scan was never moved into the summary. +func TestDeliverCountsOverflowWhenTheHandoffIsClosedAndEmpty(t *testing.T) { + alerts := make(chan string) + close(alerts) + n := &recNotifier{} + var overflow atomic.Int64 + overflow.Store(3) + deliver(context.Background(), time.Hour, n, alerts, &overflow) + require.Len(t, n.msgs, 1) + assert.Contains(t, n.msgs[0], "3 further alert(s)") +}