From b7a10485af45e26955cec0b9b5b6bd7d03f12cdc Mon Sep 17 00:00:00 2001 From: Mathias Date: Mon, 5 Oct 2026 09:34:27 +0200 Subject: [PATCH] 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 --- .../internal/symlinkguard/symlinkguard.go | 60 ++++++++++++++++++- .../symlinkguard/symlinkguard_test.go | 39 ++++++++++++ 2 files changed, 97 insertions(+), 2 deletions(-) diff --git a/ingestion/internal/symlinkguard/symlinkguard.go b/ingestion/internal/symlinkguard/symlinkguard.go index e8c8984..b38e2e7 100644 --- a/ingestion/internal/symlinkguard/symlinkguard.go +++ b/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) + } } } diff --git a/ingestion/internal/symlinkguard/symlinkguard_test.go b/ingestion/internal/symlinkguard/symlinkguard_test.go index 36fe13d..721b1b0 100644 --- a/ingestion/internal/symlinkguard/symlinkguard_test.go +++ b/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) +}