From 4ba9d20b4e244ded88d708ebafffdeb7b12fd679 Mon Sep 17 00:00:00 2001 From: Mathias Date: Sat, 3 Oct 2026 18:30:46 +0200 Subject: [PATCH] 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") +}