mathiasandClaude Opus 5.5
b7a10485af
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 <noreply@anthropic.com>
538 lines
21 KiB
Go
538 lines
21 KiB
Go
package symlinkguard
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"os"
|
|
"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.
|
|
type toggleNotifier struct {
|
|
recNotifier
|
|
down atomic.Bool
|
|
}
|
|
|
|
func (r *toggleNotifier) Notify(ctx context.Context, m string) error {
|
|
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)
|
|
require.Eventually(t, func() bool { return n.count() >= 1 }, 2*time.Second, 5*time.Millisecond)
|
|
time.Sleep(100 * time.Millisecond)
|
|
assert.LessOrEqual(t, n.count(), maxPending+1, "queued alerts plus at most one suppression summary")
|
|
n.mu.Lock()
|
|
defer n.mu.Unlock()
|
|
found := false
|
|
for _, m := range n.msgs {
|
|
if strings.Contains(m, "suppressed") {
|
|
found = true
|
|
}
|
|
}
|
|
assert.True(t, found, "suppressed alerts are counted in a 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 }, 2*time.Second, 5*time.Millisecond)
|
|
n.mu.Lock()
|
|
defer n.mu.Unlock()
|
|
for i, name := range []string{"qa", "qb", "qc"} {
|
|
assert.Contains(t, n.msgs[i], strconv.Quote(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)
|
|
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
|
|
}, 2*time.Second, 5*time.Millisecond)
|
|
time.Sleep(50 * time.Millisecond)
|
|
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") {
|
|
_, _ = fmt.Sscanf(m[i+3:], "%d", &counted)
|
|
}
|
|
}
|
|
assert.Equal(t, plants, delivered+counted, "delivered %d + counted %d must equal %d removals", delivered, counted, 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, 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)
|
|
}
|