Count flush trigger reseeded from the ETA backlog caps batch size on countdown queues

[dmcwhorter]

Summary

On a task whose messages carry an ETA/countdown, the count-based flush trigger is reseeded from the number of parked requests:

if len(ready_requests) > 0:
    logger.debug("Batches: Ready buffer complete: %s", len(ready_requests))
    self.flush(ready_requests)
    self._count = count(self._pending.qsize() + 1)   # celery_batches/__init__.py#L345

_pending holds requests whose ETA has not elapsed. On a delayed queue that backlog is large and roughly constant, so the next count flush fires after flush_every - len(_pending) further arrivals instead of flush_every. The trigger no longer tracks how full _buffer is, and mean batch size plateaus near half of flush_every at any arrival rate.

Queues with no ETA are unaffected: their _pending is always empty, so count(_pending.qsize() + 1) and count(1) are the same.

Reproduction

Discrete-event simulation driving the real Strategy() handler and _do_flush against a virtual clock. Poisson arrivals, flush_every=16, flush_interval=5, a 120 s countdown on every message, 5 seeds × 1800 s. Only monotonic, to_timestamp and the request factory are faked; the trigger logic is unmodified.

arrival (msg/s)	mean batch, current	mean batch, count(1)
1	4.0	5.0
3	6.3	11.2
5	6.7	12.2
13.9	7.6	14.4
40	8.0	15.4

Current behaviour plateaus at 8.0/16 however high arrival goes; reseeding from 1 reaches 15.4/16.

The invariant break is visible directly: at 13.9 msg/s, 212 of 212 count-triggered flushes carried 12 requests against flush_every=16. A count-triggered flush should mean flush_every messages were buffered.

Why the current seed backfires

Seeding with _pending.qsize() + 1 makes the next flush fire sooner when a backlog exists, which reads as "there is parked work, so check more often". But parked work is not gated on checking — only on its ETA elapsing. The extra wake-ups cannot release anything early; they only chop _buffer into smaller batches, which is the opposite of what the count trigger is for.

Suggested fix

-            self._count = count(self._pending.qsize() + 1)
+            self._count = count(1)

The counter counts arrivals into _buffer. _pending is a separate population the count trigger does not gate, so folding it into the seed conflates the two.

Happy to open a PR with a regression test if you'd like the change in this shape — and equally happy to be told the current seed is deliberate for a case I've missed.

[dmcwhorter]

Following up with the two questions I'd expect this to raise — whether the _pending.qsize() term is doing useful work for backlogs. Both checked against the real _do_flush.

Does reseeding from 1 slow backlog drain? No. With 500 parked requests and zero new arrivals, current main and count(1) both drain all 500 in the same 20 flushes, finishing at the same instant — exactly when the last ETA elapses. _do_flush re-drains _pending on every call (#L310), and the interval timer stays armed as long as anything is parked, since the teardown at #L347 requires not ready_requests and self._pending.qsize() == 0. The count trigger only ever adds flushes on top of the timer's cadence; it cannot make a backlog drain faster, only make batches smaller.

Would a backlog-aware seed be better? No — every variant degenerates, because parked requests are exactly the ones a flush cannot emit. With 500 parked (none due) and 16 ready arrivals:

seed	batches emitted	still parked
count(self._pending.qsize() + 1) (current)	[12]	500
count(min(P, flush_every))	[1] × 16	500
count(max(P, flush_every))	[1] × 16	500
count(1)	[16]	500

min/max land on a seed ≡ 0 (mod flush_every), so they fire on every delivery and shred the batch without releasing anything. Nothing drains the parked 500 in any variant — only time does.

There's also no monotonicity in the current seed: deliveries-to-trigger is flush_every - (P mod flush_every), so P=16 and P=1600 both give flush_every, while P=500 gives 12. That 12 is exactly the batch size observed in the simulation run above.

To be clear about scope: #95 fixed a genuine bug and the reset itself is right — this is only about the seed, and it is inert for anyone not using ETAs, since _pending stays empty and count(P+1) is then literally count(1).