From 504feb372f5dab4ebe44cc37cbb630e2c7b9629e Mon Sep 17 00:00:00 2001 From: Mathias Date: Sat, 3 Oct 2026 18:16:43 +0200 Subject: [PATCH 1/3] feat(ingestion): the runtime symlink tripwire removes what it finds Decision (#157): the tripwire only alerted, so a symlink planted in the Syncthing-shared brain dir stayed readable until a human removed it. Watch now unlinks every symlink it finds through an os.Root on the brain dir (re-checking it is still a symlink; the target is never touched), and the default scan interval drops from 5 min to 30 s, so a planted link is readable for at most one interval. Each removal is an alert naming what was removed; a persistent finding (a link it could not remove, an unreadable dir, a scan error) alerts once per distinct state. Alerts queue until Notify succeeds, so an ntfy outage delays them without losing them, even though the next scan of the cleaned brain is clean. The startup check still refuses to start on any symlink. Refs #157 Co-Authored-By: Claude Opus 5.5 --- .../internal/symlinkguard/symlinkguard.go | 108 ++++++++++++------ .../symlinkguard/symlinkguard_test.go | 43 ++++++- 2 files changed, 117 insertions(+), 34 deletions(-) diff --git a/ingestion/internal/symlinkguard/symlinkguard.go b/ingestion/internal/symlinkguard/symlinkguard.go index 2dc11f0..08074de 100644 --- a/ingestion/internal/symlinkguard/symlinkguard.go +++ b/ingestion/internal/symlinkguard/symlinkguard.go @@ -21,7 +21,7 @@ import ( ) // DefaultInterval is used when Watch is given a non-positive interval. -const DefaultInterval = 5 * time.Minute +const DefaultInterval = 30 * time.Second // Notifier is the out-of-band alert sink (audit.Notifier satisfies it). type Notifier interface { @@ -41,13 +41,7 @@ func (r Result) Clean() bool { return len(r.Symlinks) == 0 && len(r.Unreadable) // String describes the findings for a log line or alert. Paths are quoted, so // a name containing a newline cannot blur two findings together. func (r Result) String() string { - q := func(xs []string) string { - out := make([]string, len(xs)) - for i, x := range xs { - out[i] = strconv.Quote(x) - } - return strings.Join(out, ", ") - } + q := r.quoted var parts []string if len(r.Symlinks) > 0 { parts = append(parts, fmt.Sprintf("%d symlink(s): %s", len(r.Symlinks), q(r.Symlinks))) @@ -58,6 +52,15 @@ func (r Result) String() string { return strings.Join(parts, "; ") } +// quoted joins paths with each one strconv-quoted. +func (Result) quoted(xs []string) string { + out := make([]string, len(xs)) + for i, x := range xs { + out[i] = strconv.Quote(x) + } + return strings.Join(out, ", ") +} + // Scan walks the brain dir and reports every symlink in it and every // directory it could not read, continuing past both. Only the ROOT .git is // skipped (it is brain-sync's, and no reader walks it); a nested .git is @@ -106,15 +109,19 @@ func Scan(brainDir string) (Result, error) { } // Watch rescans every interval (DefaultInterval if non-positive) until ctx -// ends, and notifies once per distinct non-clean result. A result counts as -// reported only after Notify succeeds, so an alert lost to a notifier outage -// is retried on the next tick. +// ends. Every symlink found is removed (the link, never its target: +// hyperguild#157 decision), so one is readable for at most one interval. +// Each removal is an alert; a persistent finding (a link it could not remove, +// an unreadable dir, a scan error) alerts once per distinct state. Alerts +// queue until Notify succeeds, so an outage delays them but loses none, even +// though the next scan of a cleaned brain is clean. func Watch(ctx context.Context, brainDir string, interval time.Duration, n Notifier) { if interval <= 0 { interval = DefaultInterval } t := time.NewTicker(interval) defer t.Stop() + var pending []string reported := "" for { select { @@ -123,31 +130,68 @@ func Watch(ctx context.Context, brainDir string, interval time.Duration, n Notif case <-t.C: } r, err := Scan(brainDir) - var key, msg string - switch { - case err != nil: - key = "error: " + err.Error() - msg = "brain symlink tripwire (hyperguild#157): cannot scan the brain dir: " + err.Error() - case r.Clean(): - reported = "" - continue - default: - key = r.String() - msg = "brain symlink tripwire (hyperguild#157): " + key + - ". Readers may follow symlinks out of the brain; remove them." + key := "" + if err != nil { + key = "cannot scan the brain dir: " + err.Error() + } 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) + } + key = Result{Symlinks: failed, Unreadable: r.Unreadable}.String() } - if key == reported { - continue - } - slog.Error("brain symlink tripwire", "finding", key) - if n == nil { + if key != reported { reported = key + if key != "" { + slog.Error("brain symlink tripwire", "finding", key) + pending = append(pending, "brain symlink tripwire (hyperguild#157): "+key+ + ". Readers may follow symlinks out of the brain; fix them.") + } + } + if n == nil { + pending = nil continue } - if err := n.Notify(ctx, msg); err != nil { - slog.Warn("symlink tripwire notify failed; will retry", "err", err) - 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 + } + pending = pending[1:] } - reported = key } } + +// removeSymlinks unlinks each brain-relative symlink through an os.Root on +// the brain dir, after re-checking it is still a symlink. Removing a symlink +// never touches its target. It returns what was removed and what was not. +func removeSymlinks(brainDir string, rels []string) (removed, failed []string) { + if len(rels) == 0 { + return nil, nil + } + resolved, err := filepath.EvalSymlinks(brainDir) + if err != nil { + return nil, rels + } + root, err := os.OpenRoot(resolved) + if err != nil { + return nil, rels + } + defer func() { _ = root.Close() }() + for _, rel := range rels { + fi, err := root.Lstat(rel) + if err != nil || fi.Mode()&os.ModeSymlink == 0 { + failed = append(failed, rel) + continue + } + if err := root.Remove(rel); err != nil { + failed = append(failed, rel) + continue + } + removed = append(removed, rel) + } + return removed, failed +} diff --git a/ingestion/internal/symlinkguard/symlinkguard_test.go b/ingestion/internal/symlinkguard/symlinkguard_test.go index 6b46112..c3ef264 100644 --- a/ingestion/internal/symlinkguard/symlinkguard_test.go +++ b/ingestion/internal/symlinkguard/symlinkguard_test.go @@ -182,8 +182,47 @@ func TestWatchReportsASetAgainAfterItWasCleared(t *testing.T) { 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) - require.NoError(t, os.Remove(link)) - time.Sleep(60 * 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 TestWatchRemovesSymlinksButNeverTheirTargetsOrRegularFiles(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) +} -- 2.54.0 From 01de1634528c04361f78c14e991462ad1d227be2 Mon Sep 17 00:00:00 2001 From: Mathias Date: Sat, 3 Oct 2026 18:24:13 +0200 Subject: [PATCH 2/3] 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) -- 2.54.0 From 4ba9d20b4e244ded88d708ebafffdeb7b12fd679 Mon Sep 17 00:00:00 2001 From: Mathias Date: Sat, 3 Oct 2026 18:30:46 +0200 Subject: [PATCH 3/3] fix(ingestion): tripwire alerts survive outages; drop only permanent rejections PR #160 round 2: dropping a message after 5 failures could not tell a poison message from an ntfy outage. A 15-minute outage dropped the only incident with nothing reported, and drops were not counted (74 of 300 removals were neither delivered nor counted). - audit.NtfyNotifier returns *RejectedError for non-2xx; Permanent() is true for 4xx other than 429. - The tripwire queue drops a message only on a permanent rejection and counts it; every other failure is an outage and is retried, however long. Each flush tries every queued message, so an undeliverable one cannot block the rest. - Overflow and drops go into one counter, sent as a summary retried every tick and never dropped. Every removal is delivered or counted. Tests (red first): an outage of ~100 ticks still delivers the only incident; delivered + counted equals the number of removals; a permanently rejected message is dropped and counted. The short-outage test is renamed to say it is short. The Watch doc no longer claims 'loses none' unconditionally. Refs #157 Co-Authored-By: Claude Opus 5.5 --- ingestion/internal/audit/ntfy.go | 14 ++- .../internal/audit/ntfy_rejected_test.go | 27 ++++++ .../internal/symlinkguard/symlinkguard.go | 66 +++++++------ .../symlinkguard/symlinkguard_test.go | 97 ++++++++++++++++++- 4 files changed, 173 insertions(+), 31 deletions(-) create mode 100644 ingestion/internal/audit/ntfy_rejected_test.go diff --git a/ingestion/internal/audit/ntfy.go b/ingestion/internal/audit/ntfy.go index 7aea375..9208c85 100644 --- a/ingestion/internal/audit/ntfy.go +++ b/ingestion/internal/audit/ntfy.go @@ -49,7 +49,19 @@ func (n *NtfyNotifier) Notify(ctx context.Context, msg string) error { } defer func() { _ = resp.Body.Close() }() if resp.StatusCode < 200 || resp.StatusCode >= 300 { - return fmt.Errorf("ntfy notify: status %d", resp.StatusCode) + return &RejectedError{Status: resp.StatusCode} } return nil } + +// RejectedError is a non-2xx ntfy response. +type RejectedError struct{ Status int } + +func (e *RejectedError) Error() string { return fmt.Sprintf("ntfy notify: status %d", e.Status) } + +// Permanent reports whether resending the same message cannot succeed: a 4xx +// other than 429 (rate limited). Callers may drop such a message; anything +// else (5xx, 429, transport errors) is an outage and must be retried. +func (e *RejectedError) Permanent() bool { + return e.Status >= 400 && e.Status < 500 && e.Status != http.StatusTooManyRequests +} diff --git a/ingestion/internal/audit/ntfy_rejected_test.go b/ingestion/internal/audit/ntfy_rejected_test.go new file mode 100644 index 0000000..3b8387b --- /dev/null +++ b/ingestion/internal/audit/ntfy_rejected_test.go @@ -0,0 +1,27 @@ +package audit + +import ( + "context" + "errors" + "net/http" + "net/http/httptest" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// A non-2xx ntfy response is a *RejectedError, and only a client-side refusal +// other than rate limiting is permanent: callers may drop a message on a +// permanent rejection and must retry the rest (hyperguild#157, PR #160). +func TestNotifyClassifiesRejectionsAsPermanentOrNot(t *testing.T) { + for status, permanent := range map[int]bool{413: true, 400: true, 403: true, 429: false, 500: false, 503: false} { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(status) })) + err := NewNtfyNotifier(srv.URL, "").Notify(context.Background(), "m") + srv.Close() + var rej *RejectedError + require.True(t, errors.As(err, &rej), "status %d: %v", status, err) + assert.Equal(t, status, rej.Status) + assert.Equal(t, permanent, rej.Permanent(), "status %d", status) + } +} diff --git a/ingestion/internal/symlinkguard/symlinkguard.go b/ingestion/internal/symlinkguard/symlinkguard.go index cdf9272..e8c8984 100644 --- a/ingestion/internal/symlinkguard/symlinkguard.go +++ b/ingestion/internal/symlinkguard/symlinkguard.go @@ -9,6 +9,7 @@ package symlinkguard import ( "context" + "errors" "fmt" "io/fs" "log/slog" @@ -28,7 +29,6 @@ const DefaultInterval = 30 * time.Second 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). @@ -130,8 +130,9 @@ func Scan(brainDir string) (Result, error) { // hyperguild#157 decision), so one is readable for at most one interval. // Each removal is an alert; a persistent finding (a link it could not remove, // an unreadable dir, a scan error) alerts once per distinct state. Alerts -// queue until Notify succeeds, so an outage delays them but loses none, even -// though the next scan of a cleaned brain is clean. +// queue until Notify succeeds (see alertQueue): an outage delays them, and +// any that do not fit the bounded queue, or are undeliverable, are counted in +// a summary that is never dropped, so every incident is delivered or counted. func Watch(ctx context.Context, brainDir string, interval time.Duration, n Notifier) { if interval <= 0 { interval = DefaultInterval @@ -171,25 +172,28 @@ func Watch(ctx context.Context, brainDir string, interval time.Duration, n Notif } } -// 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. +// alertQueue holds alerts until Notify accepts them. It is bounded (at most +// maxPending queued) and loses nothing from the record: every overflow and +// every dropped message increments suppressed, sent as a summary that is +// retried every tick and never dropped. +// +// A message is dropped only when the notifier rejects it permanently (an +// error with Permanent() == true, e.g. ntfy 413). Every other failure is an +// outage, and nothing is dropped however long it lasts (PR #160 round 2: +// dropping after N raw failures lost the only incident of a long outage). +// Each flush tries every queued message, so one undeliverable message cannot +// block the others. type alertQueue struct { - items []queued + items []string 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}) + q.items = append(q.items, msg) } func (q *alertQueue) flush(ctx context.Context, n Notifier) { @@ -197,28 +201,34 @@ func (q *alertQueue) flush(ctx context.Context, n Notifier) { 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 - } + kept := q.items[:0] + for _, msg := range q.items { + err := n.Notify(ctx, msg) + switch { + case err == nil: + case permanent(err): + slog.Error("symlink tripwire alert rejected permanently; counted in the summary", "err", err) + q.suppressed++ + default: slog.Warn("symlink tripwire notify failed; will retry", "err", err) - return + kept = append(kept, msg) } - q.items = q.items[1:] } + q.items = kept 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) + msg := fmt.Sprintf("brain symlink tripwire (hyperguild#157): %d further alert(s) were suppressed or undeliverable; check the ingestion logs", q.suppressed) + if err := n.Notify(ctx, msg); err == nil { + q.suppressed = 0 + } } } +// permanent reports whether err says the message itself can never be sent. +func permanent(err error) bool { + var p interface{ Permanent() bool } + return errors.As(err, &p) && p.Permanent() +} + // removeSymlinks unlinks each brain-relative symlink through an os.Root on // the brain dir, after re-checking it is still a symlink. Removing a symlink // never touches its target. It returns what was removed and what was not. diff --git a/ingestion/internal/symlinkguard/symlinkguard_test.go b/ingestion/internal/symlinkguard/symlinkguard_test.go index 19efab5..36fe13d 100644 --- a/ingestion/internal/symlinkguard/symlinkguard_test.go +++ b/ingestion/internal/symlinkguard/symlinkguard_test.go @@ -3,6 +3,7 @@ package symlinkguard import ( "context" "errors" + "fmt" "os" "path/filepath" "strconv" @@ -305,12 +306,18 @@ type rejectNotifier struct { 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 errors.New("413") + return permanentErr{} } return r.recNotifier.Notify(ctx, m) } @@ -383,7 +390,7 @@ func TestWatchQueueIsBoundedDuringAnOutage(t *testing.T) { // 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) { +func TestWatchDeliversEveryQueuedAlertAfterAShortOutage(t *testing.T) { brain := t.TempDir() n := &toggleNotifier{} n.down.Store(true) @@ -403,3 +410,89 @@ func TestWatchDeliversEveryQueuedAlertAfterAnOutage(t *testing.T) { 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") +} -- 2.54.0