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 }