fix(symlinkguard): deliver alerts beside the scan, not inside it
Watch flushed the alert queue in the scan goroutine. With a hung ntfy (accepts, never answers until the 10 s timeout) and a full queue, one flush took ~6 notify timeouts, so a planted symlink stayed readable for several intervals instead of at most one. Delivery now runs in its own goroutine that owns the queue. The scan hands alerts over on a bounded channel without blocking; an alert that does not fit is counted in the never-dropped summary, so the "delivered or counted" property holds. Repro test (interval 50 ms, notifier latency 40 ms, full queue): worst link lifetime 243 ms before, under 2 intervals after. Closes #164 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -18,6 +18,8 @@ import (
|
|||||||
"sort"
|
"sort"
|
||||||
"strconv"
|
"strconv"
|
||||||
"strings"
|
"strings"
|
||||||
|
"sync"
|
||||||
|
"sync/atomic"
|
||||||
"time"
|
"time"
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -133,13 +135,25 @@ func Scan(brainDir string) (Result, error) {
|
|||||||
// queue until Notify succeeds (see alertQueue): an outage delays them, and
|
// 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
|
// 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.
|
// a summary that is never dropped, so every incident is delivered or counted.
|
||||||
|
//
|
||||||
|
// Delivery runs in its own goroutine (hyperguild#164): a hung notifier, with
|
||||||
|
// a full queue, blocked one flush for several notify timeouts, and while the
|
||||||
|
// scan waited on it a planted link stayed readable for several intervals.
|
||||||
|
// The scan hands alerts over without blocking; any that do not fit are
|
||||||
|
// counted in the summary.
|
||||||
func Watch(ctx context.Context, brainDir string, interval time.Duration, n Notifier) {
|
func Watch(ctx context.Context, brainDir string, interval time.Duration, n Notifier) {
|
||||||
if interval <= 0 {
|
if interval <= 0 {
|
||||||
interval = DefaultInterval
|
interval = DefaultInterval
|
||||||
}
|
}
|
||||||
t := time.NewTicker(interval)
|
t := time.NewTicker(interval)
|
||||||
defer t.Stop()
|
defer t.Stop()
|
||||||
q := &alertQueue{}
|
alerts := make(chan string, maxPending)
|
||||||
|
var overflow atomic.Int64
|
||||||
|
var wg sync.WaitGroup
|
||||||
|
wg.Add(1)
|
||||||
|
go func() { defer wg.Done(); deliver(ctx, interval, n, alerts, &overflow) }()
|
||||||
|
defer wg.Wait()
|
||||||
|
q := alertSender{alerts: alerts, overflow: &overflow}
|
||||||
reported := ""
|
reported := ""
|
||||||
for {
|
for {
|
||||||
select {
|
select {
|
||||||
@@ -168,8 +182,50 @@ func Watch(ctx context.Context, brainDir string, interval time.Duration, n Notif
|
|||||||
". Readers may follow symlinks out of the brain; fix them.")
|
". Readers may follow symlinks out of the brain; fix them.")
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// alertSender hands alerts to deliver without ever blocking the scan; an
|
||||||
|
// alert that does not fit is counted in overflow, which deliver adds to the
|
||||||
|
// summary.
|
||||||
|
type alertSender struct {
|
||||||
|
alerts chan<- string
|
||||||
|
overflow *atomic.Int64
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s alertSender) push(msg string) {
|
||||||
|
select {
|
||||||
|
case s.alerts <- msg:
|
||||||
|
default:
|
||||||
|
s.overflow.Add(1)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// deliver owns the alertQueue: it queues alerts as they arrive and flushes
|
||||||
|
// every interval until ctx ends.
|
||||||
|
func deliver(ctx context.Context, interval time.Duration, n Notifier, alerts <-chan string, overflow *atomic.Int64) {
|
||||||
|
t := time.NewTicker(interval)
|
||||||
|
defer t.Stop()
|
||||||
|
q := &alertQueue{}
|
||||||
|
for {
|
||||||
|
select {
|
||||||
|
case <-ctx.Done():
|
||||||
|
return
|
||||||
|
case msg := <-alerts:
|
||||||
|
q.push(msg)
|
||||||
|
case <-t.C:
|
||||||
|
for drained := false; !drained; {
|
||||||
|
select {
|
||||||
|
case msg := <-alerts:
|
||||||
|
q.push(msg)
|
||||||
|
default:
|
||||||
|
drained = true
|
||||||
|
}
|
||||||
|
}
|
||||||
|
q.suppressed += int(overflow.Swap(0))
|
||||||
q.flush(ctx, n)
|
q.flush(ctx, n)
|
||||||
}
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// alertQueue holds alerts until Notify accepts them. It is bounded (at most
|
// alertQueue holds alerts until Notify accepts them. It is bounded (at most
|
||||||
|
|||||||
@@ -496,3 +496,42 @@ func TestWatchDropsAPoisonMessageAndCountsIt(t *testing.T) {
|
|||||||
return false
|
return false
|
||||||
}, 3*time.Second, 5*time.Millisecond, "the dropped poison alert is counted in a summary")
|
}, 3*time.Second, 5*time.Millisecond, "the dropped poison alert is counted in a summary")
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// hungNotifier answers every Notify only after latency, with an outage error:
|
||||||
|
// an ntfy that accepts connections but never responds until its timeout.
|
||||||
|
type hungNotifier struct{ latency time.Duration }
|
||||||
|
|
||||||
|
func (h hungNotifier) Notify(ctx context.Context, _ string) error {
|
||||||
|
select {
|
||||||
|
case <-time.After(h.latency):
|
||||||
|
case <-ctx.Done():
|
||||||
|
}
|
||||||
|
return errors.New("ntfy timeout")
|
||||||
|
}
|
||||||
|
|
||||||
|
// hyperguild#164: alerts were flushed in the scan goroutine, so with a hung
|
||||||
|
// ntfy and a full queue one flush (6 x latency) delayed the next scan and a
|
||||||
|
// planted link stayed for ~5 intervals. Delivery now runs beside the scan.
|
||||||
|
func TestWatchLinkLifetimeStaysUnderTwoIntervalsWithAHungNotifier(t *testing.T) {
|
||||||
|
const interval = 50 * time.Millisecond
|
||||||
|
brain := t.TempDir()
|
||||||
|
ctx, cancel := context.WithCancel(context.Background())
|
||||||
|
defer cancel()
|
||||||
|
go Watch(ctx, brain, interval, hungNotifier{latency: 40 * time.Millisecond})
|
||||||
|
|
||||||
|
// Fill the queue: one removal alert per tick, more than maxPending.
|
||||||
|
for i := 0; i < maxPending+3; i++ {
|
||||||
|
link := filepath.Join(brain, "fill"+strconv.Itoa(i))
|
||||||
|
require.NoError(t, os.Symlink(t.TempDir(), link))
|
||||||
|
require.Eventually(t, func() bool { _, err := os.Lstat(link); return os.IsNotExist(err) }, 5*time.Second, time.Millisecond)
|
||||||
|
}
|
||||||
|
var worst time.Duration
|
||||||
|
for i := 0; i < 8; i++ {
|
||||||
|
link := filepath.Join(brain, "probe"+strconv.Itoa(i))
|
||||||
|
require.NoError(t, os.Symlink(t.TempDir(), link))
|
||||||
|
planted := time.Now()
|
||||||
|
require.Eventually(t, func() bool { _, err := os.Lstat(link); return os.IsNotExist(err) }, 5*time.Second, time.Millisecond)
|
||||||
|
worst = max(worst, time.Since(planted))
|
||||||
|
}
|
||||||
|
assert.LessOrEqual(t, worst, 2*interval, "worst link lifetime %v with a hung notifier", worst)
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user