Deliver traffic whose timestamp has stopped advancing

[SimonHeybrock]

Motivation

A producer whose timestamp stops advancing can never satisfy a slot gate (zero diffs are filtered, so no rate converges) and never moves the high-water mark past the timeout threshold. If it is the only traffic in a service, nothing ever closes the window. The wall-clock stall backstop saw the anchor inside the window, took the stall for quietness and re-armed, so the buffer grew at the full ingest rate while nothing was delivered. Held-back messages in the future queue were invisible to the stall check altogether, which is the variant the review of #1271 found.

Symptom of #1275, which tracks the underlying missing bound on active-window residency. This PR bounds only the stuck-clock case.

What this does

Quietness (a stopped stream's trailing partial batch) and a stuck clock look identical at a single stall check and differ in one observable: whether the buffer has grown since the previous check. When it has, the backstop fires the timeout close, which delivers the traffic and advances the data clock by one batch length per stall interval. That is the one case where wall time moves the data clock, and it does so because the data clock itself has stopped. The buffer is bounded at a couple of stall intervals' worth of traffic, and every message is delivered.

The stall check now counts held-back traffic as buffered, so a future-queue-only stall recovers like a bucket stall.

The trailing-partial-batch case is unchanged: a buffer that does not grow still re-arms silently, and the existing no-thrash test pins that.

Test plan

• test_stuck_clock_is_delivered_and_bounded, parametrized over the bucket and future-queue variants: everything fed is delivered or still held, and the held depth stays under three stall intervals
• tests/core tests/services tests/kafka: 971 passed, 1 skipped, 4 xfailed (pre-existing)