From cafb4a991364622116b918b1997d050b6335b5ff Mon Sep 17 00:00:00 2001
From: Geno Roupsky <185580+groupsky@users.noreply.github.com>
Date: Thu, 15 Oct 2026 23:36:59 +0300
Subject: [PATCH] test(broker): wait for the foreign-tap worker before counting
 entries

TestForeignTap counted the audit entries right after the "dropped
foreign_tap entries" log line, before the worker had written the queued
taps. So it failed now and then (48 of 5,040 runs on 4 CPUs).

It now waits until the worker holds one tap at the gate, so exactly
tapQueue taps are queued and the rest are dropped. It waits for every
entry, checks the dropped count in the log, closes the service, and only
then checks the upper limit. ForeignTap runs with a deadline, so a
ForeignTap that blocks fails the test instead of hanging it.
---
 internal/broker/helpers_test.go |  9 +++--
 internal/broker/service_test.go | 61 +++++++++++++++++++++++++++------
 2 files changed, 57 insertions(+), 13 deletions(-)

diff --git a/internal/broker/helpers_test.go b/internal/broker/helpers_test.go
index f8a1ee4..c4d7f6e 100644
--- a/internal/broker/helpers_test.go
+++ b/internal/broker/helpers_test.go
@@ -148,12 +148,13 @@ type fakeAudit struct {
 	failOn  func(audit.Event) error // if set, Append fails with its error
 	broken  error
 	gate    chan struct{} // if set, Append waits until it is closed
+	gated   chan struct{} // gets a value when an Append starts to wait at the gate
 	panicOn func(audit.Event) bool
 	changed chan struct{}
 }
 
 func newFakeAudit() *fakeAudit {
-	return &fakeAudit{changed: make(chan struct{})}
+	return &fakeAudit{changed: make(chan struct{}), gated: make(chan struct{}, 64)}
 }
 
 func (f *fakeAudit) Append(ev audit.Event) (audit.Anchor, error) {
@@ -161,6 +162,10 @@ func (f *fakeAudit) Append(ev audit.Event) (audit.Anchor, error) {
 	gate := f.gate
 	f.mu.Unlock()
 	if gate != nil {
+		select {
+		case f.gated <- struct{}{}:
+		default:
+		}
 		<-gate
 	}
 	f.mu.Lock()
@@ -218,7 +223,7 @@ func (f *fakeAudit) waitFor(t testing.TB, n int) []audit.Event {
 		select {
 		case <-changed:
 		case <-deadline.C:
-			t.Fatalf("the audit log has %d entries, want %d: %s", len(events), n, eventNames(events))
+			t.Fatalf("the audit log has %d entries, want at least %d: %s", len(events), n, eventNames(events))
 			return nil
 		}
 	}
diff --git a/internal/broker/service_test.go b/internal/broker/service_test.go
index a6c773d..cccd035 100644
--- a/internal/broker/service_test.go
+++ b/internal/broker/service_test.go
@@ -11,6 +11,7 @@ import (
 	"net/http"
 	"net/http/httptest"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -32,6 +33,16 @@ func (f *fakeAudit) setGate(gate chan struct{}) {
 	f.gate = gate
 }
 
+// waitGated waits until an Append waits at the gate of setGate.
+func (f *fakeAudit) waitGated(t testing.TB) {
+	t.Helper()
+	select {
+	case <-f.gated:
+	case <-time.After(waitLimit):
+		t.Fatal("no Append waits at the gate")
+	}
+}
+
 // waitLog waits until the log holds want, and returns the log.
 func waitLog(t testing.TB, logs *syncBuffer, want string) string {
 	t.Helper()
@@ -490,22 +501,50 @@ func TestForeignTap(t *testing.T) {
 	if got := e.audit.waitFor(t, 1); got[0] != audit.Event(tap) {
 		t.Errorf("the entry is %+v", got[0])
 	}
+	// A slow audit log: Append waits at the gate. ForeignTap must not
+	// wait for it. sendTaps calls ForeignTap n times in a goroutine, so a
+	// ForeignTap that waits fails the test and does not hang it.
 	gate := make(chan struct{})
 	e.audit.setGate(gate)
-	start := time.Now()
-	for range 3 * tapQueue {
-		e.svc.ForeignTap(tap)
-	}
-	if d := time.Since(start); d > 5*time.Second {
-		t.Errorf("ForeignTap waited %v for a slow audit log", d)
+	openGate := sync.OnceFunc(func() { close(gate) })
+	defer openGate() // also when the test fails, so that the worker and Close can end
+	sendTaps := func(n int) {
+		t.Helper()
+		sent := make(chan struct{})
+		go func() {
+			defer close(sent)
+			for range n {
+				e.svc.ForeignTap(tap)
+			}
+		}()
+		select {
+		case <-sent:
+		case <-time.After(5 * time.Second):
+			t.Fatal("ForeignTap waited more than 5s for a slow audit log")
+		}
 	}
+	// The worker takes the next tap, and its Append waits at the gate.
+	// Now the worker holds one tap, and the queue is empty.
+	sendTaps(1)
+	e.audit.waitGated(t)
+	// 3 times the queue: tapQueue taps fit, the others are dropped.
+	sendTaps(3 * tapQueue)
+	// The audit log is fast again. The worker writes the tap that it
+	// held, then the queue. Before the first entry of the queue, it logs
+	// how many taps it dropped.
 	e.audit.setGate(nil)
-	close(gate)
-	waitLog(t, e.logs, "dropped foreign_tap entries")
-	if n := len(e.audit.all()); n < 2 || n > 2+tapQueue {
-		t.Errorf("%d entries, want at most the queue", n)
-	}
+	openGate()
+	e.audit.waitFor(t, 2+tapQueue)
+	logs := waitLog(t, e.logs, "dropped foreign_tap entries")
+	if want := fmt.Sprintf("dropped=%d\n", 2*tapQueue); !strings.Contains(logs, want) {
+		t.Errorf("the log does not count the %d dropped taps:\n%s", 2*tapQueue, logs)
+	}
+	// Close waits until the worker has written every queued entry, so no
+	// entry can come later.
 	e.svc.Close()
+	if n := len(e.audit.all()); n > 2+tapQueue {
+		t.Errorf("%d entries, want at most %d: the queue keeps more than %d taps", n, 2+tapQueue, tapQueue)
+	}
 	e.svc.ForeignTap(tap) // no panic, no wait
 }