Merge pull request 'Broker tests read a consumer's pending count through pendingOf wherever they expect a count after a publish (issue 1126)' (#477) from fix/1126-tests-wait-for-consumer-counts into main

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 got := pendingOf(js, stream, "fresh", 1); got != 1 {
-	if info, _ = js.js.ConsumerInfo(stream, "fresh"); info.NumPending != 1 {
+		t.Fatalf("a new announcement is not pending: %d", got)
-		t.Fatalf("a new announcement is not pending: %d", info.NumPending)
 	}
 	// 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 {
+	if got := pendingOf(js, stream, "old", 6); got != 6 {
-		t.Fatalf("the default should replay all six: %d", info.NumPending)
+		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
 		}

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 got := pendingOf(js, EventsStream, MergesConsumerName, 1); got != 1 {
-	if err != nil || merges.NumPending != 1 {
+		t.Fatalf("the merges consumer holds %d pending; want the merge", got)
-		t.Fatalf("the merges consumer holds %v (%v); want the merge", merges, err)
 	}
 	if _, _, err := js.ResetConsumer(EventsStream, ControllerName); err != nil {
 		t.Fatal(err)
 	}
-	merges, err = js.Context().ConsumerInfo(EventsStream, MergesConsumerName)
+	pending := pendingOf(js, EventsStream, MergesConsumerName, 1)
-	if err != nil || merges.NumPending != 1 || merges.Delivered.Stream != 0 {
+	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 got := pendingOf(js, EventsStream, MergesConsumerName, 1); got != 1 {
-	if err != nil || merges.NumPending != 1 {
+		t.Fatalf("the merges consumer holds %d pending; want the merge the old consumer never acknowledged", got)
-		t.Fatalf("the merges consumer holds %v (%v); want the merge the old consumer never acknowledged", merges, err)
 	}
 }
 

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 got := pendingOf(js, "MAIL", c.Name, 1); got != 1 {
-	if info, _ := js.js.ConsumerInfo("MAIL", c.Name); info.NumPending != 1 {
+		t.Fatalf("the reader granted `mesh` has %d pending, want 1", got)
-		t.Fatalf("the reader granted `mesh` has %d pending, want 1", info.NumPending)
 	}
 	narrowed := c
 	narrowed.Filters = []string{"mesh.stream.mesh-mail.event.mail.mesh.message.changed"}