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)") +}