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/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 2dc11f0..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" @@ -21,7 +22,14 @@ import ( ) // DefaultInterval is used when Watch is given a non-positive interval. -const DefaultInterval = 5 * time.Minute +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 +) // Notifier is the out-of-band alert sink (audit.Notifier satisfies it). type Notifier interface { @@ -41,13 +49,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 +60,24 @@ func (r Result) String() string { return strings.Join(parts, "; ") } +// quoted joins at most maxPathsPerAlert paths, each strconv-quoted, and +// counts the rest. +func (Result) quoted(xs []string) string { + shown := xs + if len(shown) > maxPathsPerAlert { + shown = shown[:maxPathsPerAlert] + } + out := make([]string, len(shown)) + for i, x := range shown { + out[i] = strconv.Quote(x) + } + 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 // 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 +126,20 @@ 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 (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 } t := time.NewTicker(interval) defer t.Stop() + q := &alertQueue{} reported := "" for { select { @@ -123,31 +148,117 @@ 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 { + 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() } - if key == reported { - continue - } - slog.Error("brain symlink tripwire", "finding", key) - if n == nil { + if key != reported { reported = key - continue + if key != "" { + slog.Error("brain symlink tripwire", "finding", key) + q.push("brain symlink tripwire (hyperguild#157): " + key + + ". Readers may follow symlinks out of the brain; fix them.") + } } - if err := n.Notify(ctx, msg); err != nil { - slog.Warn("symlink tripwire notify failed; will retry", "err", err) - continue - } - reported = key + q.flush(ctx, n) } } + +// 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 []string + suppressed int +} + +func (q *alertQueue) push(msg string) { + if len(q.items) >= maxPending { + q.suppressed++ + return + } + q.items = append(q.items, msg) +} + +func (q *alertQueue) flush(ctx context.Context, n Notifier) { + if n == nil { + q.items, q.suppressed = nil, 0 + return + } + 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) + kept = append(kept, msg) + } + } + q.items = kept + if q.suppressed > 0 { + 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. +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 + } + // 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 + } + 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..36fe13d 100644 --- a/ingestion/internal/symlinkguard/symlinkguard_test.go +++ b/ingestion/internal/symlinkguard/symlinkguard_test.go @@ -3,9 +3,13 @@ package symlinkguard import ( "context" "errors" + "fmt" "os" "path/filepath" + "strconv" + "strings" "sync" + "sync/atomic" "testing" "time" @@ -182,8 +186,313 @@ 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 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") +} 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)