test(search): hold a writer on swapped-in FIFOs; await the swapper

PR #161 round 2: ReadRegular bypassing readOpened (non-blocking open,
no handle check) stayed green, because the search FIFO race used FIFOs
with no writer, which read EOF at once. The swapper now opens each FIFO
O_RDWR and keeps every writer open until it exits, as any local process
could, so a reader that skips the handle check parks forever; the bypass
flip is red. The swapper is stopped and awaited in a t.Cleanup before
t.TempDir removal (hyperguild#162).

Refs #157

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

ingestion/internal/api/raw_race_test.go

@@ -8,7 +8,6 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
-	"sync"
 	"sync/atomic"
 	"syscall"
 	"testing"
@@ -26,18 +25,6 @@ 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++ {
@@ -89,7 +76,9 @@ func readers(brain, raw string) string {
 
 func TestRawReadersLeakNothingUnderASymlinkSwapRace(t *testing.T) {
 	brain, raw, outside := raceBrain(t)
-	startSwapper(t, func(stop *atomic.Bool) { swap(raw, outside, false, stop) })
+	var stop atomic.Bool
+	go swap(raw, outside, false, &stop)
+	defer stop.Store(true)
 	leaks, iters := 0, 0
 	for deadline := time.Now().Add(1500 * time.Millisecond); time.Now().Before(deadline); iters++ {
 		leaks += strings.Count(readers(brain, raw), raceSentinel)
@@ -102,7 +91,8 @@ func TestRawReadersLeakNothingUnderASymlinkSwapRace(t *testing.T) {
 
 func TestRawReadersDoNotHangUnderAFIFOSwapRace(t *testing.T) {
 	brain, raw, outside := raceBrain(t)
-	startSwapper(t, func(stop *atomic.Bool) { swap(raw, outside, true, stop) })
+	var stop atomic.Bool
+	go swap(raw, outside, true, &stop)
 	var iters atomic.Int64
 	done := make(chan struct{})
 	go func() {
@@ -114,7 +104,9 @@ 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()
@@ -155,7 +147,8 @@ func TestDocmarkExtractionLeaksNothingUnderASymlinkSwapRace(t *testing.T) {
 	if err := os.WriteFile(tgt, []byte("inside docx bytes"), 0o644); err != nil {
 		t.Fatal(err)
 	}
-	startSwapper(t, func(stop *atomic.Bool) {
+	var stop atomic.Bool
+	go func() {
 		for i := 0; !stop.Load(); i++ {
 			tmp := filepath.Join(raw, ".swapdocx")
 			_ = os.Remove(tmp)
@@ -166,7 +159,8 @@ 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)

ingestion/internal/audit/ntfy.go

@@ -49,19 +49,7 @@ func (n *NtfyNotifier) Notify(ctx context.Context, msg string) error {
 	}
 	defer func() { _ = resp.Body.Close() }()
 	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
-		return &RejectedError{Status: resp.StatusCode}
+		return fmt.Errorf("ntfy notify: status %d", 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
-}

ingestion/internal/audit/ntfy_rejected_test.go

@@ -1,27 +0,0 @@
-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)
-	}
-}

ingestion/internal/symlinkguard/symlinkguard.go

@@ -9,7 +9,6 @@ package symlinkguard
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"io/fs"
 	"log/slog"
@@ -22,14 +21,7 @@ import (
 )
 
 // DefaultInterval is used when Watch is given a non-positive interval.
-const DefaultInterval = 30 * time.Second
+const DefaultInterval = 5 * time.Minute
-
-// 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 {
@@ -49,7 +41,13 @@ 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 := r.quoted
+	q := func(xs []string) string {
+		out := make([]string, len(xs))
+		for i, x := range xs {
+			out[i] = strconv.Quote(x)
+		}
+		return strings.Join(out, ", ")
+	}
 	var parts []string
 	if len(r.Symlinks) > 0 {
 		parts = append(parts, fmt.Sprintf("%d symlink(s): %s", len(r.Symlinks), q(r.Symlinks)))
@@ -60,24 +58,6 @@ 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
@@ -126,20 +106,15 @@ func Scan(brainDir string) (Result, error) {
 }
 
 // Watch rescans every interval (DefaultInterval if non-positive) until ctx
-// ends. Every symlink found is removed (the link, never its target:
+// ends, and notifies once per distinct non-clean result. A result counts as
-// hyperguild#157 decision), so one is readable for at most one interval.
+// reported only after Notify succeeds, so an alert lost to a notifier outage
-// Each removal is an alert; a persistent finding (a link it could not remove,
+// is retried on the next tick.
-// 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 {
@@ -148,117 +123,31 @@ func Watch(ctx context.Context, brainDir string, interval time.Duration, n Notif
 		case <-t.C:
 		}
 		r, err := Scan(brainDir)
-		key := ""
+		var key, msg string
-		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 {
-			reported = key
-			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.")
-			}
-		}
-		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 err != nil:
-		case permanent(err):
+			key = "error: " + err.Error()
-			slog.Error("symlink tripwire alert rejected permanently; counted in the summary", "err", err)
+			msg = "brain symlink tripwire (hyperguild#157): cannot scan the brain dir: " + err.Error()
-			q.suppressed++
+		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."
+		}
+		if key == reported {
+			continue
+		}
+		slog.Error("brain symlink tripwire", "finding", key)
+		if n == nil {
+			reported = key
+			continue
+		}
+		if err := n.Notify(ctx, msg); err != nil {
 			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
+		reported = key
-		// 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
 }

ingestion/internal/symlinkguard/symlinkguard_test.go

@@ -3,13 +3,9 @@ package symlinkguard
 import (
 	"context"
 	"errors"
-	"fmt"
 	"os"
 	"path/filepath"
-	"strconv"
-	"strings"
 	"sync"
-	"sync/atomic"
 	"testing"
 	"time"
 
@@ -186,313 +182,8 @@ 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)
-	// The tripwire removed it; planting it again is a new event.
+	require.NoError(t, os.Remove(link))
-	require.Eventually(t, func() bool { _, err := os.Lstat(link); return os.IsNotExist(err) }, 2*time.Second, 10*time.Millisecond)
+	time.Sleep(60 * 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")
-}

ingestion/internal/watcher/watcher_test.go

@@ -8,7 +8,6 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
-	"sync"
 	"sync/atomic"
 	"syscall"
 	"testing"
@@ -344,12 +343,7 @@ 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)