mathiasandClaude Opus 5.5
4422dd1f69
Three Watch tests flaked under -race and CPU load (3/8 of 8 parallel x10 runs on main): - QueueIsBounded slept 100ms before looking for the summary; it now waits for the summary, then checks the bound. - DeliversEveryQueuedAlert asserted FIFO order, but the outage ends asynchronously, possibly mid-flush, so a later item can send first. It now asserts each removal is delivered exactly once (order is not what it is named for). - AccountsForEveryRemoval slept 50ms after the first summary, counted only "removed 1" and took the last summary's count. It now sums every "removed N" and every suppression count and waits until they account for all plants. Stress: 0/8 failures in three rounds of 8 parallel x -test.count=10 under -race. Flips still red: unbounded queue, dropped retries, suppressions not counted. Fixes #193. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
787 lines
30 KiB
Go
787 lines
30 KiB
Go
package symlinkguard
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"os"
|
|
"log/slog"
|
|
"path/filepath"
|
|
"strconv"
|
|
"strings"
|
|
"sync"
|
|
"sync/atomic"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/stretchr/testify/assert"
|
|
"github.com/stretchr/testify/require"
|
|
)
|
|
|
|
// hyperguild#157: the brain dir is a Syncthing sendreceive folder and
|
|
// Syncthing syncs symlinks; several readers follow them out of the brain.
|
|
// Until every reader goes through an os.Root, a symlink anywhere in the brain,
|
|
// or a directory the scan cannot read (which could hide one), is an incident.
|
|
|
|
func TestScanReportsSymlinksAtEveryDepthIncludingNestedGit(t *testing.T) {
|
|
brain := t.TempDir()
|
|
outside := t.TempDir()
|
|
require.NoError(t, os.MkdirAll(filepath.Join(brain, "wiki", "w", "facts", ".git"), 0o755))
|
|
require.NoError(t, os.MkdirAll(filepath.Join(brain, ".git", "objects"), 0o755))
|
|
require.NoError(t, os.Symlink(filepath.Join(outside, "f"), filepath.Join(brain, "wiki", "w", "facts", "leak.md")))
|
|
require.NoError(t, os.Symlink(outside, filepath.Join(brain, "wiki", "dirlink")))
|
|
// PR #159 review: only the ROOT .git is skipped; a nested .git is not
|
|
// special to the brain readers, which walk into it.
|
|
require.NoError(t, os.Symlink(filepath.Join(outside, "g"), filepath.Join(brain, "wiki", "w", "facts", ".git", "leak.md")))
|
|
require.NoError(t, os.Symlink(outside, filepath.Join(brain, ".git", "objects", "alt")))
|
|
|
|
r, err := Scan(brain)
|
|
require.NoError(t, err)
|
|
assert.ElementsMatch(t, []string{"wiki/w/facts/leak.md", "wiki/dirlink", "wiki/w/facts/.git/leak.md"}, r.Symlinks)
|
|
assert.Empty(t, r.Unreadable)
|
|
}
|
|
|
|
// PR #159 review: WalkDir stopped at the first unreadable directory and the
|
|
// startup check passed, while search (which keeps walking) served a planted
|
|
// symlink. Scan now records the unreadable dir and keeps going.
|
|
func TestScanReportsUnreadableDirsAndKeepsWalkingPastThem(t *testing.T) {
|
|
if os.Geteuid() == 0 {
|
|
t.Skip("root reads mode-000 directories")
|
|
}
|
|
brain := t.TempDir()
|
|
locked := filepath.Join(brain, "aaa")
|
|
require.NoError(t, os.MkdirAll(locked, 0o755))
|
|
require.NoError(t, os.Chmod(locked, 0o000))
|
|
t.Cleanup(func() { _ = os.Chmod(locked, 0o755) })
|
|
require.NoError(t, os.Symlink(t.TempDir(), filepath.Join(brain, "zzz-planted")))
|
|
|
|
r, err := Scan(brain)
|
|
require.NoError(t, err)
|
|
assert.Equal(t, []string{"aaa"}, r.Unreadable)
|
|
assert.Equal(t, []string{"zzz-planted"}, r.Symlinks, "the walk continues past the unreadable dir")
|
|
}
|
|
|
|
func TestScanResolvesASymlinkedRootAndReportsSymlinkedEntriesInsideIt(t *testing.T) {
|
|
real := t.TempDir()
|
|
link := filepath.Join(t.TempDir(), "brain")
|
|
require.NoError(t, os.Symlink(real, link))
|
|
require.NoError(t, os.WriteFile(filepath.Join(real, "n.md"), []byte("x"), 0o644))
|
|
require.NoError(t, os.Symlink(t.TempDir(), filepath.Join(real, "inner")))
|
|
r, err := Scan(link)
|
|
require.NoError(t, err)
|
|
assert.Equal(t, []string{"inner"}, r.Symlinks)
|
|
}
|
|
|
|
func TestScanOfAMissingRootIsErrNotExist(t *testing.T) {
|
|
_, err := Scan(filepath.Join(t.TempDir(), "nope"))
|
|
assert.True(t, errors.Is(err, os.ErrNotExist), "got %v", err)
|
|
}
|
|
|
|
type recNotifier struct {
|
|
mu sync.Mutex
|
|
msgs []string
|
|
fail int // fail the first N calls
|
|
}
|
|
|
|
func (r *recNotifier) Notify(_ context.Context, m string) error {
|
|
r.mu.Lock()
|
|
defer r.mu.Unlock()
|
|
r.msgs = append(r.msgs, m)
|
|
if r.fail > 0 {
|
|
r.fail--
|
|
return errors.New("ntfy down")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (r *recNotifier) count() int { r.mu.Lock(); defer r.mu.Unlock(); return len(r.msgs) }
|
|
|
|
func TestWatchNotifiesOncePerDistinctSetAndAgainForANewSet(t *testing.T) {
|
|
brain := t.TempDir()
|
|
n := &recNotifier{}
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
go Watch(ctx, brain, 10*time.Millisecond, n)
|
|
|
|
time.Sleep(50 * time.Millisecond)
|
|
assert.Zero(t, n.count(), "a clean brain raises nothing")
|
|
require.NoError(t, os.Symlink(t.TempDir(), filepath.Join(brain, "first")))
|
|
require.Eventually(t, func() bool { return n.count() == 1 }, 2*time.Second, 10*time.Millisecond)
|
|
time.Sleep(60 * time.Millisecond)
|
|
assert.Equal(t, 1, n.count(), "the same set is reported once, not every tick")
|
|
require.NoError(t, os.Symlink(t.TempDir(), filepath.Join(brain, "second")))
|
|
require.Eventually(t, func() bool { return n.count() == 2 }, 2*time.Second, 10*time.Millisecond)
|
|
}
|
|
|
|
// PR #159 review: last was recorded before Notify, so an alert lost to an
|
|
// ntfy outage was never retried while the symlink stayed.
|
|
func TestWatchRetriesAnAlertThatFailedToSend(t *testing.T) {
|
|
brain := t.TempDir()
|
|
require.NoError(t, os.Symlink(t.TempDir(), filepath.Join(brain, "planted")))
|
|
n := &recNotifier{fail: 2}
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
go Watch(ctx, brain, 10*time.Millisecond, n)
|
|
require.Eventually(t, func() bool { return n.count() == 3 }, 2*time.Second, 10*time.Millisecond)
|
|
time.Sleep(60 * time.Millisecond)
|
|
assert.Equal(t, 3, n.count(), "after one successful send, no more")
|
|
}
|
|
|
|
func TestWatchAlertsOnAnUnreadableDir(t *testing.T) {
|
|
if os.Geteuid() == 0 {
|
|
t.Skip("root reads mode-000 directories")
|
|
}
|
|
brain := t.TempDir()
|
|
locked := filepath.Join(brain, "locked")
|
|
require.NoError(t, os.MkdirAll(locked, 0o755))
|
|
require.NoError(t, os.Chmod(locked, 0o000))
|
|
t.Cleanup(func() { _ = os.Chmod(locked, 0o755) })
|
|
n := &recNotifier{}
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
go Watch(ctx, brain, 10*time.Millisecond, n)
|
|
require.Eventually(t, func() bool { return n.count() == 1 }, 2*time.Second, 10*time.Millisecond)
|
|
assert.Contains(t, n.msgs[0], "locked")
|
|
}
|
|
|
|
// PR #159 review: a non-positive interval panicked time.NewTicker inside the
|
|
// goroutine and crashlooped the server.
|
|
func TestWatchToleratesANonPositiveInterval(t *testing.T) {
|
|
brain := t.TempDir()
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
done := make(chan struct{})
|
|
go func() { Watch(ctx, brain, 0, nil); close(done) }()
|
|
cancel()
|
|
select {
|
|
case <-done:
|
|
case <-time.After(2 * time.Second):
|
|
t.Fatal("Watch did not return on cancel")
|
|
}
|
|
}
|
|
|
|
// PR #159 round 2: a brain root that can be entered but not listed (mode
|
|
// 0100) made WalkDir fail at the root; the error was discarded and Scan
|
|
// reported clean, while search (which goes straight into wiki/) still served
|
|
// a planted symlink. A root that cannot be listed is now an error.
|
|
func TestScanOfAnUnlistableRootIsAnError(t *testing.T) {
|
|
if os.Geteuid() == 0 {
|
|
t.Skip("root lists mode-0100 directories")
|
|
}
|
|
brain := t.TempDir()
|
|
require.NoError(t, os.MkdirAll(filepath.Join(brain, "wiki"), 0o755))
|
|
require.NoError(t, os.Symlink(t.TempDir(), filepath.Join(brain, "wiki", "leak.md")))
|
|
require.NoError(t, os.Chmod(brain, 0o100))
|
|
t.Cleanup(func() { _ = os.Chmod(brain, 0o755) })
|
|
_, err := Scan(brain)
|
|
require.Error(t, err)
|
|
assert.False(t, errors.Is(err, os.ErrNotExist), "an unlistable root is not a missing root")
|
|
}
|
|
|
|
// The same set that disappears and comes back is reported again.
|
|
func TestWatchReportsASetAgainAfterItWasCleared(t *testing.T) {
|
|
brain := t.TempDir()
|
|
link := filepath.Join(brain, "planted")
|
|
n := &recNotifier{}
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
go Watch(ctx, brain, 10*time.Millisecond, n)
|
|
require.NoError(t, os.Symlink(t.TempDir(), link))
|
|
require.Eventually(t, func() bool { return n.count() == 1 }, 2*time.Second, 10*time.Millisecond)
|
|
// The tripwire removed it; planting it again is a new event.
|
|
require.Eventually(t, func() bool { _, err := os.Lstat(link); return os.IsNotExist(err) }, 2*time.Second, 10*time.Millisecond)
|
|
require.NoError(t, os.Symlink(t.TempDir(), link))
|
|
require.Eventually(t, func() bool { return n.count() == 2 }, 2*time.Second, 10*time.Millisecond)
|
|
}
|
|
|
|
// hyperguild#157 decision: the runtime tripwire removes a symlink it finds
|
|
// (the link, never its target), so it is readable for at most one interval
|
|
// instead of until a human acts, and the alert says what was removed.
|
|
func TestWatchRemovesAnOutsideFileLinkButNotItsTargetOrRegularFiles(t *testing.T) {
|
|
brain := t.TempDir()
|
|
outside := filepath.Join(t.TempDir(), "target.md")
|
|
require.NoError(t, os.WriteFile(outside, []byte("keep me"), 0o644))
|
|
require.NoError(t, os.MkdirAll(filepath.Join(brain, "wiki"), 0o755))
|
|
link := filepath.Join(brain, "wiki", "leak.md")
|
|
require.NoError(t, os.Symlink(outside, link))
|
|
regular := filepath.Join(brain, "wiki", "note.md")
|
|
require.NoError(t, os.WriteFile(regular, []byte("note"), 0o644))
|
|
|
|
n := &recNotifier{}
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
go Watch(ctx, brain, 10*time.Millisecond, n)
|
|
|
|
require.Eventually(t, func() bool {
|
|
_, err := os.Lstat(link)
|
|
return os.IsNotExist(err)
|
|
}, 2*time.Second, 10*time.Millisecond, "the symlink is removed")
|
|
require.Eventually(t, func() bool { return n.count() >= 1 }, 2*time.Second, 10*time.Millisecond)
|
|
b, err := os.ReadFile(outside)
|
|
require.NoError(t, err)
|
|
assert.Equal(t, "keep me", string(b), "the target is untouched")
|
|
b, err = os.ReadFile(regular)
|
|
require.NoError(t, err)
|
|
assert.Equal(t, "note", string(b), "regular files are untouched")
|
|
n.mu.Lock()
|
|
defer n.mu.Unlock()
|
|
assert.Contains(t, n.msgs[0], "removed")
|
|
assert.Contains(t, n.msgs[0], "wiki/leak.md")
|
|
}
|
|
|
|
func TestDefaultIntervalIsThirtySeconds(t *testing.T) {
|
|
assert.Equal(t, 30*time.Second, DefaultInterval)
|
|
}
|
|
|
|
// PR #160 review: the Lstat re-check and the os.Root confinement in
|
|
// removeSymlinks had no test. Anything that is not a symlink inside the brain
|
|
// is reported failed and nothing is deleted.
|
|
func TestRemoveSymlinksRefusesNonSymlinksEscapesAndPathsThroughLinks(t *testing.T) {
|
|
brain := t.TempDir()
|
|
require.NoError(t, os.MkdirAll(filepath.Join(brain, "wiki", "real"), 0o755))
|
|
regular := filepath.Join(brain, "wiki", "note.md")
|
|
require.NoError(t, os.WriteFile(regular, []byte("note"), 0o644))
|
|
inLinked := filepath.Join(brain, "wiki", "real", "x.md")
|
|
require.NoError(t, os.WriteFile(inLinked, []byte("x"), 0o644))
|
|
require.NoError(t, os.Symlink(filepath.Join(brain, "wiki", "real"), filepath.Join(brain, "wiki", "linkdir")))
|
|
outside := filepath.Join(filepath.Dir(brain), "outside-157.md")
|
|
require.NoError(t, os.WriteFile(outside, []byte("o"), 0o644))
|
|
t.Cleanup(func() { _ = os.Remove(outside) })
|
|
|
|
removed, failed := removeSymlinks(brain, []string{
|
|
"wiki/note.md", "wiki/linkdir/x.md", "../outside-157.md", "wiki/missing.md",
|
|
})
|
|
assert.Empty(t, removed)
|
|
assert.ElementsMatch(t, []string{"wiki/note.md", "wiki/linkdir/x.md", "../outside-157.md", "wiki/missing.md"}, failed)
|
|
for _, p := range []string{regular, inLinked, outside} {
|
|
_, err := os.Stat(p)
|
|
assert.NoError(t, err, "%s must not be deleted", p)
|
|
}
|
|
}
|
|
|
|
// Dir targets and in-brain targets survive too, not only an outside file.
|
|
func TestWatchRemovesDirAndInBrainLinksButNotTheirTargets(t *testing.T) {
|
|
brain := t.TempDir()
|
|
outsideDir := t.TempDir()
|
|
require.NoError(t, os.WriteFile(filepath.Join(outsideDir, "keep.md"), []byte("keep"), 0o644))
|
|
require.NoError(t, os.MkdirAll(filepath.Join(brain, "wiki"), 0o755))
|
|
inBrain := filepath.Join(brain, "wiki", "target.md")
|
|
require.NoError(t, os.WriteFile(inBrain, []byte("t"), 0o644))
|
|
require.NoError(t, os.Symlink(outsideDir, filepath.Join(brain, "wiki", "dirlink")))
|
|
require.NoError(t, os.Symlink(inBrain, filepath.Join(brain, "wiki", "inlink.md")))
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
go Watch(ctx, brain, 10*time.Millisecond, nil)
|
|
require.Eventually(t, func() bool {
|
|
r, err := Scan(brain)
|
|
return err == nil && len(r.Symlinks) == 0
|
|
}, 2*time.Second, 10*time.Millisecond)
|
|
for _, p := range []string{filepath.Join(outsideDir, "keep.md"), inBrain} {
|
|
_, err := os.Stat(p)
|
|
assert.NoError(t, err, "%s must survive", p)
|
|
}
|
|
}
|
|
|
|
// PR #160 review: the planter chooses the paths. An alert names at most
|
|
// maxPathsPerAlert of them and counts the rest, so its size stays bounded.
|
|
func TestWatchAlertNamesBoundedPathsAndCountsTheRest(t *testing.T) {
|
|
brain := t.TempDir()
|
|
for i := 0; i < 50; i++ {
|
|
require.NoError(t, os.Symlink(t.TempDir(), filepath.Join(brain, strings.Repeat("n", 100)+strconv.Itoa(i))))
|
|
}
|
|
n := &recNotifier{}
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
go Watch(ctx, brain, 10*time.Millisecond, n)
|
|
require.Eventually(t, func() bool { return n.count() >= 1 }, 2*time.Second, 10*time.Millisecond)
|
|
n.mu.Lock()
|
|
msg := n.msgs[0]
|
|
n.mu.Unlock()
|
|
assert.Contains(t, msg, "removed 50 symlink(s)")
|
|
assert.Contains(t, msg, "and 40 more")
|
|
assert.Less(t, len(msg), 2500, "alert size is bounded regardless of how many links were planted")
|
|
}
|
|
|
|
// rejectNotifier refuses every message containing a marker, forever.
|
|
type rejectNotifier struct {
|
|
recNotifier
|
|
reject string
|
|
}
|
|
|
|
// permanentErr is a rejection the notifier says can never succeed (like ntfy 413).
|
|
type permanentErr struct{}
|
|
|
|
func (permanentErr) Error() string { return "413 payload too large" }
|
|
func (permanentErr) Permanent() bool { return true }
|
|
|
|
func (r *rejectNotifier) Notify(ctx context.Context, m string) error {
|
|
if strings.Contains(m, r.reject) {
|
|
r.mu.Lock()
|
|
r.msgs = append(r.msgs, "REJECTED")
|
|
r.mu.Unlock()
|
|
return permanentErr{}
|
|
}
|
|
return r.recNotifier.Notify(ctx, m)
|
|
}
|
|
|
|
// PR #160 review: one message the server will never accept blocked every
|
|
// later alert forever (head-of-line). It is dropped after maxAttempts.
|
|
func TestWatchDropsAMessageThatNeverSendsSoLaterAlertsFlow(t *testing.T) {
|
|
brain := t.TempDir()
|
|
n := &rejectNotifier{reject: "POISON"}
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
go Watch(ctx, brain, 10*time.Millisecond, n)
|
|
require.NoError(t, os.Symlink(t.TempDir(), filepath.Join(brain, "POISON")))
|
|
time.Sleep(30 * time.Millisecond)
|
|
require.NoError(t, os.Symlink(t.TempDir(), filepath.Join(brain, "later")))
|
|
require.Eventually(t, func() bool {
|
|
n.mu.Lock()
|
|
defer n.mu.Unlock()
|
|
for _, m := range n.msgs {
|
|
if strings.Contains(m, "later") {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}, 3*time.Second, 10*time.Millisecond, "a later alert is delivered despite a poison message")
|
|
}
|
|
|
|
// 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")
|
|
}
|
|
return r.recNotifier.Notify(ctx, m)
|
|
}
|
|
|
|
// PR #160 review: an outage queued one message per removal, unbounded. The
|
|
// queue holds at most maxPending messages; the rest become one summary.
|
|
func TestWatchQueueIsBoundedDuringAnOutage(t *testing.T) {
|
|
brain := t.TempDir()
|
|
n := &toggleNotifier{}
|
|
n.down.Store(true)
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
go Watch(ctx, brain, 5*time.Millisecond, n)
|
|
for i := 0; i < 30; i++ {
|
|
link := filepath.Join(brain, "p"+strconv.Itoa(i))
|
|
require.NoError(t, os.Symlink(t.TempDir(), link))
|
|
require.Eventually(t, func() bool { _, err := os.Lstat(link); return os.IsNotExist(err) }, 2*time.Second, 2*time.Millisecond)
|
|
}
|
|
n.down.Store(false)
|
|
// Synchronise on the summary being delivered (hyperguild#193: a fixed
|
|
// sleep flaked under load); once it is, the queue has been flushed.
|
|
require.Eventually(t, func() bool {
|
|
n.mu.Lock()
|
|
defer n.mu.Unlock()
|
|
for _, m := range n.msgs {
|
|
if strings.Contains(m, "suppressed") {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}, 10*time.Second, 5*time.Millisecond, "suppressed alerts are counted in a summary")
|
|
assert.LessOrEqual(t, n.count(), maxPending+1, "queued alerts plus at most one suppression summary")
|
|
}
|
|
|
|
// Within the bound, every alert queued during an outage is delivered after
|
|
// it, each naming its own removal (PR #160 review: truncating the queue after
|
|
// the first send stayed green).
|
|
func TestWatchDeliversEveryQueuedAlertAfterAShortOutage(t *testing.T) {
|
|
brain := t.TempDir()
|
|
n := &toggleNotifier{}
|
|
n.down.Store(true)
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
go Watch(ctx, brain, 5*time.Millisecond, n)
|
|
for _, name := range []string{"qa", "qb", "qc"} {
|
|
link := filepath.Join(brain, name)
|
|
require.NoError(t, os.Symlink(t.TempDir(), link))
|
|
require.Eventually(t, func() bool { _, err := os.Lstat(link); return os.IsNotExist(err) }, 2*time.Second, 2*time.Millisecond)
|
|
}
|
|
n.down.Store(false)
|
|
require.Eventually(t, func() bool { return n.count() >= 3 }, 10*time.Second, 5*time.Millisecond)
|
|
n.mu.Lock()
|
|
defer n.mu.Unlock()
|
|
// Each removal delivered exactly once, in any order: the outage ends
|
|
// asynchronously, possibly mid-flush, so a later item can send before an
|
|
// earlier one (hyperguild#193). Order is not what this test is about.
|
|
assert.Len(t, n.msgs, 3)
|
|
for _, name := range []string{"qa", "qb", "qc"} {
|
|
hits := 0
|
|
for _, m := range n.msgs {
|
|
if strings.Contains(m, strconv.Quote(name)) {
|
|
hits++
|
|
}
|
|
}
|
|
assert.Equal(t, 1, hits, "%s delivered exactly once", name)
|
|
}
|
|
}
|
|
|
|
// PR #160 round 2: dropping after N failures could not tell a poison message
|
|
// from an outage, so a 15-minute outage silently lost the only incident. A
|
|
// message is only dropped when it keeps failing while others send; through
|
|
// an outage nothing is dropped, however long.
|
|
func TestWatchLosesNoAlertThroughALongOutage(t *testing.T) {
|
|
brain := t.TempDir()
|
|
n := &toggleNotifier{}
|
|
n.down.Store(true)
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
go Watch(ctx, brain, 2*time.Millisecond, n)
|
|
link := filepath.Join(brain, "only-incident")
|
|
require.NoError(t, os.Symlink(t.TempDir(), link))
|
|
require.Eventually(t, func() bool { _, err := os.Lstat(link); return os.IsNotExist(err) }, 2*time.Second, 2*time.Millisecond)
|
|
time.Sleep(200 * time.Millisecond) // ~100 ticks of outage, far past any attempt limit
|
|
n.down.Store(false)
|
|
require.Eventually(t, func() bool { return n.count() >= 1 }, 2*time.Second, 5*time.Millisecond)
|
|
n.mu.Lock()
|
|
defer n.mu.Unlock()
|
|
assert.Contains(t, n.msgs[0], strconv.Quote("only-incident"))
|
|
}
|
|
|
|
// Every removal is either delivered or counted in the suppression summary:
|
|
// none disappears from the record (round 2: 74 of 300 were in neither).
|
|
func TestWatchAccountsForEveryRemovalDuringAnOutage(t *testing.T) {
|
|
brain := t.TempDir()
|
|
n := &toggleNotifier{}
|
|
n.down.Store(true)
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
go Watch(ctx, brain, 2*time.Millisecond, n)
|
|
const plants = 25
|
|
for i := 0; i < plants; i++ {
|
|
link := filepath.Join(brain, "acct"+strconv.Itoa(i))
|
|
require.NoError(t, os.Symlink(t.TempDir(), link))
|
|
require.Eventually(t, func() bool { _, err := os.Lstat(link); return os.IsNotExist(err) }, 2*time.Second, 1*time.Millisecond)
|
|
}
|
|
n.down.Store(false)
|
|
// Synchronise on the record being complete, not on a fixed sleep after
|
|
// the first summary (hyperguild#193): sum every "removed N" alert and
|
|
// every suppression count until they account for all plants.
|
|
account := func() (delivered, counted int) {
|
|
n.mu.Lock()
|
|
defer n.mu.Unlock()
|
|
for _, m := range n.msgs {
|
|
var k int
|
|
if i := strings.Index(m, "removed "); i >= 0 && strings.Contains(m, "symlink(s)") {
|
|
if _, err := fmt.Sscanf(m[i+len("removed "):], "%d", &k); err == nil {
|
|
delivered += k
|
|
}
|
|
}
|
|
if i := strings.Index(m, "): "); i >= 0 && strings.Contains(m, "suppressed") {
|
|
if _, err := fmt.Sscanf(m[i+3:], "%d", &k); err == nil {
|
|
counted += k
|
|
}
|
|
}
|
|
}
|
|
return delivered, counted
|
|
}
|
|
require.Eventually(t, func() bool { d, c := account(); return d+c >= plants }, 10*time.Second, 5*time.Millisecond)
|
|
d, c := account()
|
|
assert.Equal(t, plants, d+c, "delivered %d + counted %d must equal %d removals", d, c, plants)
|
|
}
|
|
|
|
// A poison message is still dropped (others send while it fails), and the
|
|
// drop is counted in the summary.
|
|
func TestWatchDropsAPoisonMessageAndCountsIt(t *testing.T) {
|
|
brain := t.TempDir()
|
|
n := &rejectNotifier{reject: "POISON"}
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
go Watch(ctx, brain, 5*time.Millisecond, n)
|
|
require.NoError(t, os.Symlink(t.TempDir(), filepath.Join(brain, "POISON")))
|
|
time.Sleep(20 * time.Millisecond)
|
|
require.NoError(t, os.Symlink(t.TempDir(), filepath.Join(brain, "fine")))
|
|
require.Eventually(t, func() bool {
|
|
n.mu.Lock()
|
|
defer n.mu.Unlock()
|
|
for _, m := range n.msgs {
|
|
if strings.Contains(m, "1 further alert(s)") {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}, 3*time.Second, 5*time.Millisecond, "the dropped poison alert is counted in a summary")
|
|
}
|
|
|
|
// 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 {
|
|
select {
|
|
case <-time.After(h.latency):
|
|
case <-ctx.Done():
|
|
}
|
|
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 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: 400 * time.Millisecond})
|
|
|
|
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))
|
|
waitGone(t, link)
|
|
}
|
|
var worst time.Duration
|
|
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()
|
|
waitGone(t, link)
|
|
worst = max(worst, time.Since(planted))
|
|
}
|
|
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, 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)
|
|
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")
|
|
}
|
|
|
|
// 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)")
|
|
}
|