Merge pull request 'The controller's check stops waiting on fixtures and fixed intervals (issue 591)' (#279) from fix/591-the-controller-check-runs-in-minutes into main

internal/link/queue.go

@@ -321,11 +321,23 @@ func ReadBuildEvents(ctx context.Context, js *broker.JetStream, seat string, sin
 		return heard, fmt.Errorf("cannot read %s from the bus: %w", subject, err)
 	}
 	defer func() { _ = sub.Unsubscribe() }()
-	for {
-		batch, err := sub.Fetch(500, nats.MaxWait(2*time.Second))
+	// **Read what is there, then stop** (novox/hq issue 591): the consumer says how many events it holds
+	// from the moment asked, and the reading ends once they are read. A fetch for a full batch waits out
+	// its whole wait when fewer are there, so every reading of the queue took two seconds and more.
+	info, err := sub.ConsumerInfo()
+	if err != nil {
+		return heard, fmt.Errorf("cannot read %s from the bus: %w", subject, err)
+	}
+	for remaining := info.NumPending; remaining > 0; {
+		batch, err := sub.Fetch(int(min(remaining, 500)), nats.MaxWait(2*time.Second))
 		if err != nil && !errors.Is(err, nats.ErrTimeout) && !errors.Is(err, context.DeadlineExceeded) {
 			return heard, fmt.Errorf("reading %s: %w", subject, err)
 		}
+		// None in the whole wait: what was counted is gone from the stream since (its limits), not coming.
+		if len(batch) == 0 {
+			break
+		}
+		remaining -= uint64(len(batch))
 		for _, msg := range batch {
 			switch msg.Subject {
 			case BuildStartedOf(seat):
@@ -342,9 +354,6 @@ func ReadBuildEvents(ctx context.Context, js *broker.JetStream, seat string, sin
 				}
 			}
 		}
-		if len(batch) < 500 {
-			break
-		}
 		if ctx.Err() != nil {
 			return heard, ctx.Err()
 		}