From 8d58afea6cb87144abedbf586452532eca260534 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 26 Oct 2026 23:28:03 +0100 Subject: [PATCH] Broker tests read a consumer's pending count through pendingOf, which waits up to 10 s, wherever they expect a count after a publish (issue 1126) The server counts a consumer's pending after it acknowledges the publish, so a read at once failed the try of main on a loaded ace (TestAResetOfTheEventsConsumerLeavesTheMergesAndTheirConsumerAlone, 2026-10-26 22:24 UTC). Fixed for the kind, not the one test: all six reads of a positive count go through the package's own pendingOf, whose deadline was 1 s. Research 'The controller, NATS-native': none; this changes tests only, not a part of the controller. --- internal/broker/consumer_from_now_test.go | 16 +++++++++------- .../controller_buckets_integration_test.go | 15 +++++++-------- internal/broker/modstreams_live_test.go | 5 ++--- 3 files changed, 18 insertions(+), 18 deletions(-) diff --git a/internal/broker/consumer_from_now_test.go b/internal/broker/consumer_from_now_test.go index 0b66e105..cd83a209 100644 --- a/internal/broker/consumer_from_now_test.go +++ b/internal/broker/consumer_from_now_test.go @@ -46,9 +46,8 @@ func TestAConsumerMadeFromNowNeverReplaysTheStreamsHistory(t *testing.T) { t.Fatalf("a consumer made from now holds %d of the stream's past (policy %v)", info.NumPending, info.Config.DeliverPolicy) } _, _ = js.js.Publish("history.merged", []byte("new")) - time.Sleep(100 * time.Millisecond) - if info, _ = js.js.ConsumerInfo(stream, "fresh"); info.NumPending != 1 { - t.Fatalf("a new announcement is not pending: %d", info.NumPending) + if got := pendingOf(js, stream, "fresh", 1); got != 1 { + t.Fatalf("a new announcement is not pending: %d", got) } // Asserted again, it keeps where it is. if err := js.EnsureConsumer(fresh); err != nil { @@ -61,8 +60,8 @@ func TestAConsumerMadeFromNowNeverReplaysTheStreamsHistory(t *testing.T) { if err := js.EnsureConsumer(old); err != nil { t.Fatal(err) } - if info, _ = js.js.ConsumerInfo(stream, "old"); info.NumPending != 6 { - t.Fatalf("the default should replay all six: %d", info.NumPending) + if got := pendingOf(js, stream, "old", 6); got != 6 { + t.Fatalf("the default should replay all six: %d", got) } before, after, err := js.ResetConsumer(stream, "old") if err != nil { @@ -89,10 +88,13 @@ func TestAConsumerMadeFromNowNeverReplaysTheStreamsHistory(t *testing.T) { } } -// pendingOf polls a consumer's pending count until it is want or a second has passed, and returns what it saw last. +// pendingOf polls a consumer's pending count until it is want or ten seconds have passed, and returns what it saw +// last. The server counts a consumer's pending after it acknowledges the publish, not before, so a test that reads +// the count at once fails on a loaded build node (a try of main on ace, 2026-10-26): every test that expects a +// count after a publish reads it through here. func pendingOf(js *JetStream, stream, name string, want uint64) uint64 { var got uint64 - for deadline := time.Now().Add(time.Second); ; time.Sleep(20 * time.Millisecond) { + for deadline := time.Now().Add(10 * time.Second); ; time.Sleep(20 * time.Millisecond) { if info, err := js.js.ConsumerInfo(stream, name); err == nil { got = info.NumPending } diff --git a/internal/broker/controller_buckets_integration_test.go b/internal/broker/controller_buckets_integration_test.go index bf8a1ca4..42858eec 100644 --- a/internal/broker/controller_buckets_integration_test.go +++ b/internal/broker/controller_buckets_integration_test.go @@ -73,15 +73,15 @@ func TestAResetOfTheEventsConsumerLeavesTheMergesAndTheirConsumerAlone(t *testin if _, err := js.Context().Publish(ControllerFollows[0], []byte(`{}`)); err != nil { t.Fatal(err) } - merges, err := js.Context().ConsumerInfo(EventsStream, MergesConsumerName) - if err != nil || merges.NumPending != 1 { - t.Fatalf("the merges consumer holds %v (%v); want the merge", merges, err) + if got := pendingOf(js, EventsStream, MergesConsumerName, 1); got != 1 { + t.Fatalf("the merges consumer holds %d pending; want the merge", got) } if _, _, err := js.ResetConsumer(EventsStream, ControllerName); err != nil { t.Fatal(err) } - merges, err = js.Context().ConsumerInfo(EventsStream, MergesConsumerName) - if err != nil || merges.NumPending != 1 || merges.Delivered.Stream != 0 { + pending := pendingOf(js, EventsStream, MergesConsumerName, 1) + merges, err := js.Context().ConsumerInfo(EventsStream, MergesConsumerName) + if err != nil || pending != 1 || merges.Delivered.Stream != 0 { t.Fatalf("after the reset the merges consumer holds %+v (%v); want its merge left alone", merges, err) } } @@ -107,9 +107,8 @@ func TestAMergeTheOldConsumerHadNotAcknowledgedIsTheMergesConsumers(t *testing.T if err := AssertMeshConsumers(js); err != nil { t.Fatal(err) } - merges, err := js.Context().ConsumerInfo(EventsStream, MergesConsumerName) - if err != nil || merges.NumPending != 1 { - t.Fatalf("the merges consumer holds %v (%v); want the merge the old consumer never acknowledged", merges, err) + if got := pendingOf(js, EventsStream, MergesConsumerName, 1); got != 1 { + t.Fatalf("the merges consumer holds %d pending; want the merge the old consumer never acknowledged", got) } } diff --git a/internal/broker/modstreams_live_test.go b/internal/broker/modstreams_live_test.go index 877aba32..92e6f996 100644 --- a/internal/broker/modstreams_live_test.go +++ b/internal/broker/modstreams_live_test.go @@ -52,9 +52,8 @@ func TestAModuleStreamOnARealBus(t *testing.T) { t.Fatalf("publishing %s's mail: %v", account, err) } } - time.Sleep(50 * time.Millisecond) - if info, _ := js.js.ConsumerInfo("MAIL", c.Name); info.NumPending != 1 { - t.Fatalf("the reader granted `mesh` has %d pending, want 1", info.NumPending) + if got := pendingOf(js, "MAIL", c.Name, 1); got != 1 { + t.Fatalf("the reader granted `mesh` has %d pending, want 1", got) } narrowed := c narrowed.Filters = []string{"mesh.stream.mesh-mail.event.mail.mesh.message.changed"} -- 2.54.0