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 <noreply@anthropic.com>
This commit is contained in:
@@ -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)")
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user