diff --git a/ingestion/internal/symlinkguard/symlinkguard.go b/ingestion/internal/symlinkguard/symlinkguard.go index b38e2e7..6aaa480 100644 --- a/ingestion/internal/symlinkguard/symlinkguard.go +++ b/ingestion/internal/symlinkguard/symlinkguard.go @@ -23,6 +23,9 @@ import ( "time" ) +// shutdownFlushTimeout bounds the final flush when Watch stops. +var shutdownFlushTimeout = 5 * time.Second + // DefaultInterval is used when Watch is given a non-positive interval. const DefaultInterval = 30 * time.Second @@ -140,7 +143,8 @@ func Scan(brainDir string) (Result, error) { // 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. +// counted in the summary. When ctx ends, Watch returns after a final flush +// of what was handed over (bounded by shutdownFlushTimeout). func Watch(ctx context.Context, brainDir string, interval time.Duration, n Notifier) { if interval <= 0 { interval = DefaultInterval @@ -152,7 +156,9 @@ func Watch(ctx context.Context, brainDir string, interval time.Duration, n Notif var wg sync.WaitGroup wg.Add(1) go func() { defer wg.Done(); deliver(ctx, interval, n, alerts, &overflow) }() - defer wg.Wait() + // The scan is the only sender: closing alerts tells deliver to drain and + // flush once more, and Watch returns only after it has. + defer func() { close(alerts); wg.Wait() }() q := alertSender{alerts: alerts, overflow: &overflow} reported := "" for { @@ -201,28 +207,49 @@ func (s alertSender) push(msg string) { } } -// deliver owns the alertQueue: it queues alerts as they arrive and flushes -// every interval until ctx ends. +// deliver owns the alertQueue. It flushes as soon as an alert is handed +// over, and again every interval while any are pending (outage retries). +// When alerts is closed (Watch is stopping) it drains it and flushes once +// more, bounded by shutdownFlushTimeout, so an alert handed over just before +// shutdown is not lost (PR #180 review). 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{} + // drain moves everything handed over into the queue; it reports false + // once alerts is closed. + drain := func() bool { + defer func() { q.suppressed += int(overflow.Swap(0)) }() + for { + select { + case msg, ok := <-alerts: + if !ok { + return false + } + q.push(msg) + default: + return true + } + } + } for { select { - case <-ctx.Done(): - return - case msg := <-alerts: - q.push(msg) + case msg, ok := <-alerts: + if ok { + q.push(msg) + ok = drain() + } + if !ok { + fctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), shutdownFlushTimeout) + q.flush(fctx, n) + cancel() + return + } + q.flush(ctx, n) case <-t.C: - for drained := false; !drained; { - select { - case msg := <-alerts: - q.push(msg) - default: - drained = true - } + if !drain() { + continue // the closed channel is handled by the case above } - q.suppressed += int(overflow.Swap(0)) q.flush(ctx, n) } } diff --git a/ingestion/internal/symlinkguard/symlinkguard_test.go b/ingestion/internal/symlinkguard/symlinkguard_test.go index 721b1b0..56fba71 100644 --- a/ingestion/internal/symlinkguard/symlinkguard_test.go +++ b/ingestion/internal/symlinkguard/symlinkguard_test.go @@ -497,8 +497,9 @@ func TestWatchDropsAPoisonMessageAndCountsIt(t *testing.T) { }, 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. +// hungNotifier answers every Notify only after latency (or when ctx ends), +// 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 { @@ -509,29 +510,179 @@ func (h hungNotifier) Notify(ctx context.Context, _ string) error { return errors.New("ntfy timeout") } +func waitGone(t *testing.T, link string) { + t.Helper() + require.Eventually(t, func() bool { _, err := os.Lstat(link); return os.IsNotExist(err) }, 10*time.Second, time.Millisecond) +} + // 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 +// ntfy and a full queue one flush (6 x notify timeout) delayed the next scan. +// The notifier here is slower than the interval, as in production (10 s +// timeout vs 30 s interval with 6 messages), so the alert handoff is full: +// a scan that waited on it would keep a link for seconds. +func TestWatchLinkLifetimeStaysNearOneIntervalWithANotifierSlowerThanTheInterval(t *testing.T) { + const interval = 100 * time.Millisecond brain := t.TempDir() ctx, cancel := context.WithCancel(context.Background()) defer cancel() - go Watch(ctx, brain, interval, hungNotifier{latency: 40 * time.Millisecond}) + go Watch(ctx, brain, interval, hungNotifier{latency: 400 * time.Millisecond}) - // Fill the queue: one removal alert per tick, more than maxPending. - for i := 0; i < maxPending+3; i++ { + for i := 0; i < 2*maxPending+3; i++ { // overfill queue and handoff 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) + waitGone(t, link) } var worst time.Duration - for i := 0; i < 8; i++ { + for i := 0; i < 6; 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) + waitGone(t, link) worst = max(worst, time.Since(planted)) } - assert.LessOrEqual(t, worst, 2*interval, "worst link lifetime %v with a hung notifier", worst) + assert.LessOrEqual(t, worst, interval+interval/2, "worst link lifetime %v with a hung notifier", worst) +} + +// Review of PR #180: delivery waited for its own tick, about one interval +// after the removal. An alert is now flushed as soon as it is handed over. +func TestWatchDeliversARemovalAlertWellWithinOneInterval(t *testing.T) { + const interval = 400 * time.Millisecond + brain := t.TempDir() + n := &recNotifier{} + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go Watch(ctx, brain, interval, n) + link := filepath.Join(brain, "prompt") + require.NoError(t, os.Symlink(t.TempDir(), link)) + waitGone(t, link) + removed := time.Now() + require.Eventually(t, func() bool { return n.count() == 1 }, 2*time.Second, time.Millisecond) + assert.Less(t, time.Since(removed), interval/4) +} + +// Review of PR #180: on ctx end, deliver returned without draining what the +// scan had handed over, so a removal just before shutdown was neither +// delivered nor counted. Watch now drains and flushes before returning. +func TestWatchDeliversAnAlertHandedOverJustBeforeShutdown(t *testing.T) { + for i := 0; i < 20; i++ { + brain := t.TempDir() + n := &recNotifier{} + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { Watch(ctx, brain, 5*time.Millisecond, n); close(done) }() + link := filepath.Join(brain, "late") + require.NoError(t, os.Symlink(t.TempDir(), link)) + waitGone(t, link) + cancel() + <-done + require.Equal(t, 1, n.count(), "iteration %d: the removal alert was lost at shutdown", i) + } +} + +// blockingNotifier holds every Notify until its ctx ends, tracking calls in +// flight. +type blockingNotifier struct{ inFlight atomic.Int32 } + +func (b *blockingNotifier) Notify(ctx context.Context, _ string) error { + b.inFlight.Add(1) + defer b.inFlight.Add(-1) + <-ctx.Done() + return ctx.Err() +} + +// Watch does not return while its delivery goroutine is still notifying. +func TestWatchReturnsOnlyAfterDeliveryHasStopped(t *testing.T) { + prev := shutdownFlushTimeout + shutdownFlushTimeout = 50 * time.Millisecond + t.Cleanup(func() { shutdownFlushTimeout = prev }) + brain := t.TempDir() + n := &blockingNotifier{} + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { Watch(ctx, brain, 5*time.Millisecond, n); close(done) }() + link := filepath.Join(brain, "x") + require.NoError(t, os.Symlink(t.TempDir(), link)) + waitGone(t, link) + require.Eventually(t, func() bool { return n.inFlight.Load() == 1 }, 2*time.Second, time.Millisecond) + cancel() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("Watch did not return after cancel") + } + assert.Zero(t, n.inFlight.Load(), "Notify still running after Watch returned") +} + +// slowToggleNotifier fails slowly (a hung ntfy) while down, then records. +type slowToggleNotifier struct { + recNotifier + down atomic.Bool + latency time.Duration +} + +func (r *slowToggleNotifier) Notify(ctx context.Context, m string) error { + if r.down.Load() { + select { + case <-time.After(r.latency): + case <-ctx.Done(): + } + return errors.New("ntfy timeout") + } + return r.recNotifier.Notify(ctx, m) +} + +// Review of PR #180: with a slow notifier the scan's handoff overflows, and +// those alerts exist only in the overflow counter. Every removal is still +// delivered or counted in the summary. +func TestWatchCountsHandoffOverflowWithASlowNotifier(t *testing.T) { + brain := t.TempDir() + n := &slowToggleNotifier{latency: 30 * time.Millisecond} + n.down.Store(true) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go Watch(ctx, brain, 2*time.Millisecond, n) + const plants = 40 + for i := 0; i < plants; i++ { + link := filepath.Join(brain, "ov"+strconv.Itoa(i)) + require.NoError(t, os.Symlink(t.TempDir(), link)) + waitGone(t, link) + } + n.down.Store(false) + delivered, counted := 0, 0 + require.Eventually(t, func() bool { + 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") { + var c int + _, _ = fmt.Sscanf(m[i+3:], "%d", &c) + counted += c + } + } + return delivered+counted == plants + }, 5*time.Second, 10*time.Millisecond, "delivered %d + counted %d of %d", delivered, counted, plants) +} + +// An alert still queued when Watch stops (its send failed, the notifier has +// since recovered, the retry tick has not come) is flushed once more before +// Watch returns. +func TestWatchFlushesTheQueueOnceMoreAtShutdown(t *testing.T) { + brain := t.TempDir() + n := &toggleNotifier{} + n.down.Store(true) + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { Watch(ctx, brain, 300*time.Millisecond, n); close(done) }() + link := filepath.Join(brain, "queued") + require.NoError(t, os.Symlink(t.TempDir(), link)) + waitGone(t, link) + time.Sleep(20 * time.Millisecond) // let the immediate send fail + n.down.Store(false) + cancel() + <-done + assert.Equal(t, 1, n.count(), "the queued alert was not flushed at shutdown") }