fix(symlinkguard): test the WithoutCancel final flush; log what shutdown drops
CI / Lint / Test / Vet (pull_request) Successful in 1m5s
CI / Mirror to GitHub (pull_request) Skipped

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 <noreply@anthropic.com>
This commit is contained in:
2026-10-05 16:27:54 +02:00
co-authored by Claude Opus 5.5
parent 52d4bfeeab
commit cadc0cc02f
2 changed files with 101 additions and 5 deletions
@@ -144,7 +144,11 @@ func Scan(brainDir string) (Result, error) {
// scan waited on it a planted link stayed readable for several intervals. // 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 // 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 // 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) { func Watch(ctx context.Context, brainDir string, interval time.Duration, n Notifier) {
if interval <= 0 { if interval <= 0 {
interval = DefaultInterval interval = DefaultInterval
@@ -211,7 +215,8 @@ func (s alertSender) push(msg string) {
// over, and again every interval while any are pending (outage retries). // over, and again every interval while any are pending (outage retries).
// When alerts is closed (Watch is stopping) it drains it and flushes once // When alerts is closed (Watch is stopping) it drains it and flushes once
// more, bounded by shutdownFlushTimeout, so an alert handed over just before // 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) { func deliver(ctx context.Context, interval time.Duration, n Notifier, alerts <-chan string, overflow *atomic.Int64) {
t := time.NewTicker(interval) t := time.NewTicker(interval)
defer t.Stop() defer t.Stop()
@@ -240,9 +245,14 @@ func deliver(ctx context.Context, interval time.Duration, n Notifier, alerts <-c
ok = drain() ok = drain()
} }
if !ok { if !ok {
q.suppressed += int(overflow.Swap(0))
fctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), shutdownFlushTimeout) fctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), shutdownFlushTimeout)
q.flush(fctx, n) q.flush(fctx, n)
cancel() 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 return
} }
q.flush(ctx, n) q.flush(ctx, n)
@@ -1,10 +1,12 @@
package symlinkguard package symlinkguard
import ( import (
"bytes"
"context" "context"
"errors" "errors"
"fmt" "fmt"
"os" "os"
"log/slog"
"path/filepath" "path/filepath"
"strconv" "strconv"
"strings" "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") }, 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 { type toggleNotifier struct {
recNotifier recNotifier
down atomic.Bool down atomic.Bool
} }
func (r *toggleNotifier) Notify(ctx context.Context, m string) error { func (r *toggleNotifier) Notify(ctx context.Context, m string) error {
if err := ctx.Err(); err != nil {
return err
}
if r.down.Load() { if r.down.Load() {
return errors.New("ntfy down") 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 // 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 // since recovered, the retry tick has not come) is flushed once more before
// Watch returns. // Watch returns, on a ctx that is not the cancelled one: toggleNotifier fails
func TestWatchFlushesTheQueueOnceMoreAtShutdown(t *testing.T) { // on a done ctx, as the real ntfy notifier does.
func TestWatchFinalFlushSendsAQueuedAlertOnACtxThatOutlivesTheCancel(t *testing.T) {
brain := t.TempDir() brain := t.TempDir()
n := &toggleNotifier{} n := &toggleNotifier{}
n.down.Store(true) n.down.Store(true)
@@ -686,3 +695,80 @@ func TestWatchFlushesTheQueueOnceMoreAtShutdown(t *testing.T) {
<-done <-done
assert.Equal(t, 1, n.count(), "the queued alert was not flushed at shutdown") 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)")
}