test(symlinkguard): synchronise the alert-queue tests on observable completion (#193)

Three Watch tests flaked under -race and CPU load (3/8 of 8 parallel
x10 runs on main):
- QueueIsBounded slept 100ms before looking for the summary; it now
  waits for the summary, then checks the bound.
- DeliversEveryQueuedAlert asserted FIFO order, but the outage ends
  asynchronously, possibly mid-flush, so a later item can send first. It
  now asserts each removal is delivered exactly once (order is not what
  it is named for).
- AccountsForEveryRemoval slept 50ms after the first summary, counted
  only "removed 1" and took the last summary's count. It now sums every
  "removed N" and every suppression count and waits until they account
  for all plants.

Stress: 0/8 failures in three rounds of 8 parallel x -test.count=10 under
-race. Flips still red: unbounded queue, dropped retries, suppressions
not counted.

Fixes #193.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

ingestion/internal/symlinkguard/symlinkguard_test.go

@@ -381,18 +381,19 @@ func TestWatchQueueIsBoundedDuringAnOutage(t *testing.T) {
 		require.Eventually(t, func() bool { _, err := os.Lstat(link); return os.IsNotExist(err) }, 2*time.Second, 2*time.Millisecond)
 	}
 	n.down.Store(false)
-	require.Eventually(t, func() bool { return n.count() >= 1 }, 2*time.Second, 5*time.Millisecond)
-	time.Sleep(100 * time.Millisecond)
-	assert.LessOrEqual(t, n.count(), maxPending+1, "queued alerts plus at most one suppression summary")
-	n.mu.Lock()
-	defer n.mu.Unlock()
-	found := false
-	for _, m := range n.msgs {
-		if strings.Contains(m, "suppressed") {
-			found = true
+	// Synchronise on the summary being delivered (hyperguild#193: a fixed
+	// sleep flaked under load); once it is, the queue has been flushed.
+	require.Eventually(t, func() bool {
+		n.mu.Lock()
+		defer n.mu.Unlock()
+		for _, m := range n.msgs {
+			if strings.Contains(m, "suppressed") {
+				return true
+			}
 		}
-	}
-	assert.True(t, found, "suppressed alerts are counted in a summary")
+		return false
+	}, 10*time.Second, 5*time.Millisecond, "suppressed alerts are counted in a summary")
+	assert.LessOrEqual(t, n.count(), maxPending+1, "queued alerts plus at most one suppression summary")
 }
 
 // Within the bound, every alert queued during an outage is delivered after
@@ -411,11 +412,21 @@ func TestWatchDeliversEveryQueuedAlertAfterAShortOutage(t *testing.T) {
 		require.Eventually(t, func() bool { _, err := os.Lstat(link); return os.IsNotExist(err) }, 2*time.Second, 2*time.Millisecond)
 	}
 	n.down.Store(false)
-	require.Eventually(t, func() bool { return n.count() == 3 }, 2*time.Second, 5*time.Millisecond)
+	require.Eventually(t, func() bool { return n.count() >= 3 }, 10*time.Second, 5*time.Millisecond)
 	n.mu.Lock()
 	defer n.mu.Unlock()
-	for i, name := range []string{"qa", "qb", "qc"} {
-		assert.Contains(t, n.msgs[i], strconv.Quote(name))
+	// Each removal delivered exactly once, in any order: the outage ends
+	// asynchronously, possibly mid-flush, so a later item can send before an
+	// earlier one (hyperguild#193). Order is not what this test is about.
+	assert.Len(t, n.msgs, 3)
+	for _, name := range []string{"qa", "qb", "qc"} {
+		hits := 0
+		for _, m := range n.msgs {
+			if strings.Contains(m, strconv.Quote(name)) {
+				hits++
+			}
+		}
+		assert.Equal(t, 1, hits, "%s delivered exactly once", name)
 	}
 }
 
@@ -457,29 +468,30 @@ func TestWatchAccountsForEveryRemovalDuringAnOutage(t *testing.T) {
 		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 {
+	// Synchronise on the record being complete, not on a fixed sleep after
+	// the first summary (hyperguild#193): sum every "removed N" alert and
+	// every suppression count until they account for all plants.
+	account := func() (delivered, counted int) {
 		n.mu.Lock()
 		defer n.mu.Unlock()
 		for _, m := range n.msgs {
-			if strings.Contains(m, "suppressed") {
-				return true
+			var k int
+			if i := strings.Index(m, "removed "); i >= 0 && strings.Contains(m, "symlink(s)") {
+				if _, err := fmt.Sscanf(m[i+len("removed "):], "%d", &k); err == nil {
+					delivered += k
+				}
+			}
+			if i := strings.Index(m, "): "); i >= 0 && strings.Contains(m, "suppressed") {
+				if _, err := fmt.Sscanf(m[i+3:], "%d", &k); err == nil {
+					counted += k
+				}
 			}
 		}
-		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)
-		}
+		return delivered, counted
 	}
-	assert.Equal(t, plants, delivered+counted, "delivered %d + counted %d must equal %d removals", delivered, counted, plants)
+	require.Eventually(t, func() bool { d, c := account(); return d+c >= plants }, 10*time.Second, 5*time.Millisecond)
+	d, c := account()
+	assert.Equal(t, plants, d+c, "delivered %d + counted %d must equal %d removals", d, c, plants)
 }
 
 // A poison message is still dropped (others send while it fails), and the