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 <noreply@anthropic.com>

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

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

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,27 +201,33 @@ 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)})
+		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
-		q.flush(ctx, n)
 		}
 	}
+}
+
+// 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

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