feat(ingestion): runtime symlink tripwire removes what it finds, 30s scans (#157) (#160)

Co-authored-by: Mathias <mthbqv@gmail.com>

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
+}