From 01de1634528c04361f78c14e991462ad1d227be2 Mon Sep 17 00:00:00 2001 From: Mathias Date: Sat, 3 Oct 2026 18:24:13 +0200 Subject: [PATCH] fix(ingestion): bound the tripwire's alerts; test removal guards; deflake race tests PR #160 review: - The planter chooses the paths, and one alert listed them all, while the queue retried oldest-first forever: a message ntfy refuses (too large) blocked every later alert (2000 links: 0 alerts delivered), and an outage queued one message per removal, unbounded. Alerts now name at most 10 paths and count the rest; the queue holds at most 5, further alerts become one suppression summary; a message failing 5 times is dropped so it cannot block the rest. - removeSymlinks' Lstat re-check had no test: a direct test passes a regular file, a path through an in-brain dir link, a ../ escape and a missing path, and asserts all fail with nothing deleted. A Watch test covers dir and in-brain targets. root.Remove (not RemoveAll) is commented: it refuses a dir swapped in after the Lstat. - Every queued alert is delivered after an outage (truncating the queue after the first send is now red). - hyperguild#162: race tests did not wait for their swapper goroutines before t.TempDir cleanup ("directory not empty" in CI run 8912). Each swapper is now stopped and awaited in a t.Cleanup. Fixes #162 Refs #157 Co-Authored-By: Claude Opus 5.5 --- ingestion/internal/api/raw_race_test.go | 28 +-- .../internal/symlinkguard/symlinkguard.go | 95 ++++++++-- .../symlinkguard/symlinkguard_test.go | 179 +++++++++++++++++- ingestion/internal/watcher/watcher_test.go | 6 + 4 files changed, 277 insertions(+), 31 deletions(-) diff --git a/ingestion/internal/api/raw_race_test.go b/ingestion/internal/api/raw_race_test.go index 1c7b2bd..999c3e9 100644 --- a/ingestion/internal/api/raw_race_test.go +++ b/ingestion/internal/api/raw_race_test.go @@ -8,6 +8,7 @@ import ( "os" "path/filepath" "strings" + "sync" "sync/atomic" "syscall" "testing" @@ -25,6 +26,18 @@ import ( const raceSentinel = "SENTINEL-153-RACE" +// startSwapper runs fn until the test ends, and waits for it to exit before +// t.TempDir's cleanup removes the directory it writes into (hyperguild#162: +// an unawaited swapper raced the cleanup, "directory not empty"). +func startSwapper(t *testing.T, fn func(stop *atomic.Bool)) { + t.Helper() + var stop atomic.Bool + var wg sync.WaitGroup + wg.Add(1) + go func() { defer wg.Done(); fn(&stop) }() + t.Cleanup(func() { stop.Store(true); wg.Wait() }) +} + func swap(raw, outside string, fifo bool, stop *atomic.Bool) { tgt := filepath.Join(raw, "r.md") for i := 0; !stop.Load(); i++ { @@ -76,9 +89,7 @@ func readers(brain, raw string) string { func TestRawReadersLeakNothingUnderASymlinkSwapRace(t *testing.T) { brain, raw, outside := raceBrain(t) - var stop atomic.Bool - go swap(raw, outside, false, &stop) - defer stop.Store(true) + startSwapper(t, func(stop *atomic.Bool) { swap(raw, outside, false, stop) }) leaks, iters := 0, 0 for deadline := time.Now().Add(1500 * time.Millisecond); time.Now().Before(deadline); iters++ { leaks += strings.Count(readers(brain, raw), raceSentinel) @@ -91,8 +102,7 @@ func TestRawReadersLeakNothingUnderASymlinkSwapRace(t *testing.T) { func TestRawReadersDoNotHangUnderAFIFOSwapRace(t *testing.T) { brain, raw, outside := raceBrain(t) - var stop atomic.Bool - go swap(raw, outside, true, &stop) + startSwapper(t, func(stop *atomic.Bool) { swap(raw, outside, true, stop) }) var iters atomic.Int64 done := make(chan struct{}) go func() { @@ -104,9 +114,7 @@ func TestRawReadersDoNotHangUnderAFIFOSwapRace(t *testing.T) { }() select { case <-done: - stop.Store(true) case <-time.After(6 * time.Second): - stop.Store(true) // Unblock the stuck open so the test process can exit. if f, err := os.OpenFile(filepath.Join(raw, "r.md"), os.O_WRONLY|syscall.O_NONBLOCK, 0); err == nil { _ = f.Close() @@ -147,8 +155,7 @@ func TestDocmarkExtractionLeaksNothingUnderASymlinkSwapRace(t *testing.T) { if err := os.WriteFile(tgt, []byte("inside docx bytes"), 0o644); err != nil { t.Fatal(err) } - var stop atomic.Bool - go func() { + startSwapper(t, func(stop *atomic.Bool) { for i := 0; !stop.Load(); i++ { tmp := filepath.Join(raw, ".swapdocx") _ = os.Remove(tmp) @@ -159,8 +166,7 @@ func TestDocmarkExtractionLeaksNothingUnderASymlinkSwapRace(t *testing.T) { } _ = os.Rename(tmp, tgt) } - }() - defer stop.Store(true) + }) iters := 0 for deadline := time.Now().Add(1500 * time.Millisecond); time.Now().Before(deadline); iters++ { _, _ = api.ReadIngestFile(brain, tgt) diff --git a/ingestion/internal/symlinkguard/symlinkguard.go b/ingestion/internal/symlinkguard/symlinkguard.go index 08074de..cdf9272 100644 --- a/ingestion/internal/symlinkguard/symlinkguard.go +++ b/ingestion/internal/symlinkguard/symlinkguard.go @@ -23,6 +23,14 @@ import ( // DefaultInterval is used when Watch is given a non-positive interval. const DefaultInterval = 30 * time.Second +// Alert bounds (PR #160 review): the planter chooses the paths, so neither an +// alert's size nor the queue may grow with what was planted. +const ( + maxPathsPerAlert = 10 // further paths are counted, not listed + maxPending = 5 // queued alerts; the rest become one summary + maxAttempts = 5 // a message failing this often is dropped +) + // Notifier is the out-of-band alert sink (audit.Notifier satisfies it). type Notifier interface { Notify(ctx context.Context, msg string) error @@ -52,13 +60,22 @@ func (r Result) String() string { return strings.Join(parts, "; ") } -// quoted joins paths with each one strconv-quoted. +// quoted joins at most maxPathsPerAlert paths, each strconv-quoted, and +// counts the rest. func (Result) quoted(xs []string) string { - out := make([]string, len(xs)) - for i, x := range xs { + shown := xs + if len(shown) > maxPathsPerAlert { + shown = shown[:maxPathsPerAlert] + } + out := make([]string, len(shown)) + for i, x := range shown { out[i] = strconv.Quote(x) } - return strings.Join(out, ", ") + s := strings.Join(out, ", ") + if extra := len(xs) - len(shown); extra > 0 { + s += fmt.Sprintf(" and %d more", extra) + } + return s } // Scan walks the brain dir and reports every symlink in it and every @@ -121,7 +138,7 @@ func Watch(ctx context.Context, brainDir string, interval time.Duration, n Notif } t := time.NewTicker(interval) defer t.Stop() - var pending []string + q := &alertQueue{} reported := "" for { select { @@ -136,10 +153,9 @@ func Watch(ctx context.Context, brainDir string, interval time.Duration, n Notif } else { removed, failed := removeSymlinks(brainDir, r.Symlinks) if len(removed) > 0 { - msg := fmt.Sprintf("brain symlink tripwire (hyperguild#157): removed %d symlink(s) from the brain dir: %s", - len(removed), Result{Symlinks: removed}.quoted(removed)) - slog.Error("brain symlink tripwire removed symlinks", "paths", removed) - pending = append(pending, msg) + slog.Error("brain symlink tripwire removed symlinks", "count", len(removed)) + q.push(fmt.Sprintf("brain symlink tripwire (hyperguild#157): removed %d symlink(s) from the brain dir: %s", + len(removed), Result{}.quoted(removed))) } key = Result{Symlinks: failed, Unreadable: r.Unreadable}.String() } @@ -147,21 +163,59 @@ func Watch(ctx context.Context, brainDir string, interval time.Duration, n Notif reported = key if key != "" { slog.Error("brain symlink tripwire", "finding", key) - pending = append(pending, "brain symlink tripwire (hyperguild#157): "+key+ + q.push("brain symlink tripwire (hyperguild#157): " + key + ". Readers may follow symlinks out of the brain; fix them.") } } - if n == nil { - pending = nil - continue - } - for len(pending) > 0 { - if err := n.Notify(ctx, pending[0]); err != nil { - slog.Warn("symlink tripwire notify failed; will retry", "err", err) - break + q.flush(ctx, n) + } +} + +// alertQueue holds alerts until Notify accepts them, bounded: at most +// maxPending queued, further ones counted into one summary, and a message +// that fails maxAttempts times dropped so it cannot block the rest. +type alertQueue struct { + items []queued + suppressed int +} + +type queued struct { + msg string + attempts int +} + +func (q *alertQueue) push(msg string) { + if len(q.items) >= maxPending { + q.suppressed++ + return + } + q.items = append(q.items, queued{msg: msg}) +} + +func (q *alertQueue) flush(ctx context.Context, n Notifier) { + if n == nil { + q.items, q.suppressed = nil, 0 + return + } + for len(q.items) > 0 { + head := &q.items[0] + if err := n.Notify(ctx, head.msg); err != nil { + head.attempts++ + if head.attempts >= maxAttempts { + slog.Error("symlink tripwire alert dropped after repeated failures", "attempts", head.attempts, "err", err) + q.items = q.items[1:] + continue } - pending = pending[1:] + slog.Warn("symlink tripwire notify failed; will retry", "err", err) + return } + q.items = q.items[1:] + } + if q.suppressed > 0 { + q.items = append(q.items, queued{msg: fmt.Sprintf( + "brain symlink tripwire (hyperguild#157): %d further alert(s) were suppressed while the notifier was failing; check the ingestion logs", q.suppressed)}) + q.suppressed = 0 + q.flush(ctx, n) } } @@ -187,6 +241,9 @@ func removeSymlinks(brainDir string, rels []string) (removed, failed []string) { failed = append(failed, rel) continue } + // root.Remove, not os.RemoveAll: if the link is swapped for a + // non-empty directory after the Lstat, Remove refuses rather than + // deleting it recursively, and it cannot act outside the root. if err := root.Remove(rel); err != nil { failed = append(failed, rel) continue diff --git a/ingestion/internal/symlinkguard/symlinkguard_test.go b/ingestion/internal/symlinkguard/symlinkguard_test.go index c3ef264..19efab5 100644 --- a/ingestion/internal/symlinkguard/symlinkguard_test.go +++ b/ingestion/internal/symlinkguard/symlinkguard_test.go @@ -5,7 +5,10 @@ import ( "errors" "os" "path/filepath" + "strconv" + "strings" "sync" + "sync/atomic" "testing" "time" @@ -191,7 +194,7 @@ func TestWatchReportsASetAgainAfterItWasCleared(t *testing.T) { // 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 TestWatchRemovesSymlinksButNeverTheirTargetsOrRegularFiles(t *testing.T) { +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)) @@ -226,3 +229,177 @@ func TestWatchRemovesSymlinksButNeverTheirTargetsOrRegularFiles(t *testing.T) { 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 +} + +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 errors.New("413") + } + 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 TestWatchDeliversEveryQueuedAlertAfterAnOutage(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)) + } +} diff --git a/ingestion/internal/watcher/watcher_test.go b/ingestion/internal/watcher/watcher_test.go index eaf1dba..899b2bf 100644 --- a/ingestion/internal/watcher/watcher_test.go +++ b/ingestion/internal/watcher/watcher_test.go @@ -8,6 +8,7 @@ import ( "os" "path/filepath" "strings" + "sync" "sync/atomic" "syscall" "testing" @@ -343,7 +344,12 @@ func TestProcessDir_DoesNotHangUnderAFIFOSwapRace(t *testing.T) { brainDir := setupBrainDir(t) raw := filepath.Join(brainDir, "raw") var stop atomic.Bool + var wg sync.WaitGroup + wg.Add(1) + // Awaited before t.TempDir's cleanup (hyperguild#162). + t.Cleanup(func() { stop.Store(true); wg.Wait() }) go func() { + defer wg.Done() for i := 0; !stop.Load(); i++ { tmp := filepath.Join(raw, ".swap") _ = os.Remove(tmp)