test(broker): wait for the foreign-tap worker before counting entries

internal/broker/helpers_test.go

@@ -148,19 +148,24 @@ 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) {
 	f.mu.Lock()
 	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
 		}
 	}

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
 }