From b7a10485af45e26955cec0b9b5b6bd7d03f12cdc Mon Sep 17 00:00:00 2001 From: Mathias Date: Mon, 5 Oct 2026 09:34:27 +0200 Subject: [PATCH 1/3] 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) +} -- 2.54.0 From 52d4bfeeab2e5c84f52bba4b217cac74a6074c3a Mon Sep 17 00:00:00 2001 From: Mathias Date: Mon, 5 Oct 2026 09:45:15 +0200 Subject: [PATCH 2/3] 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") } -- 2.54.0 From cadc0cc02f6ed8cbbf9af32c33f4b322e8c9631a Mon Sep 17 00:00:00 2001 From: Mathias Date: Mon, 5 Oct 2026 16:27:54 +0200 Subject: [PATCH 3/3] fix(symlinkguard): test the WithoutCancel final flush; log what shutdown drops PR #180 round-2 review (comments 15467, 15487): - toggleNotifier now fails on a done ctx, like audit.NtfyNotifier, so replacing context.WithoutCancel(ctx) with ctx in the final flush turns TestWatchFinalFlushSendsAQueuedAlertOnACtxThatOutlivesTheCancel (renamed from ...FlushesTheQueueOnceMoreAtShutdown) red. - After the final flush, any undelivered or suppressed alerts are logged as one error with both counts, never silently. - The !ok first-receive branch now swaps overflow into suppressed. - Deterministic deliver() tests for the drain-on-close path and the closed-and-empty overflow case. - Watch doc states the shutdown path only runs for a cancelling caller. Co-Authored-By: Claude Opus 5.5 --- .../internal/symlinkguard/symlinkguard.go | 14 ++- .../symlinkguard/symlinkguard_test.go | 92 ++++++++++++++++++- 2 files changed, 101 insertions(+), 5 deletions(-) diff --git a/ingestion/internal/symlinkguard/symlinkguard.go b/ingestion/internal/symlinkguard/symlinkguard.go index 6aaa480..b5bd5da 100644 --- a/ingestion/internal/symlinkguard/symlinkguard.go +++ b/ingestion/internal/symlinkguard/symlinkguard.go @@ -144,7 +144,11 @@ func Scan(brainDir string) (Result, error) { // 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). +// 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 @@ -211,7 +215,8 @@ func (s alertSender) push(msg string) { // 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). +// 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() @@ -240,9 +245,14 @@ func deliver(ctx context.Context, interval time.Duration, n Notifier, alerts <-c 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) diff --git a/ingestion/internal/symlinkguard/symlinkguard_test.go b/ingestion/internal/symlinkguard/symlinkguard_test.go index 56fba71..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") } @@ -669,8 +677,9 @@ func TestWatchCountsHandoffOverflowWithASlowNotifier(t *testing.T) { // 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) { +// 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) @@ -686,3 +695,80 @@ func TestWatchFlushesTheQueueOnceMoreAtShutdown(t *testing.T) { <-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)") +} -- 2.54.0