Optimize TaskBroker scanning: larger batches + cursor-based scan

[airhorns]

Summary

• Increase scan_batch from 1024 → 4096 so a single scan pass can refill the buffer from low-watermark to target, eliminating 3x redundant re-reads per refill cycle
• Cache the SlateDB scan iterator across passes: the scanner marches forward through the key range without recreating the iterator or re-reading buffered entries. Stops at future-timestamped tasks and resumes from there on the next pass.
• When a future task is encountered, the consumed entry is held back and re-evaluated on the next pass with a fresh now_ms, so it is never lost from the iterator
• wakeup() invalidates the cached iterator, ensuring newly-created tasks (from enqueue, expedite, restart, concurrency grants, retries) that sort before the old cursor position are picked up via a fresh scan from the beginning
• Iterator is also refreshed after MAX_ITER_AGE (5s) to pick up scheduled tasks that have become ready
• Add 9 tests covering every code path that creates tasks "behind the cursor"
• Add real-shard dequeue benchmark exercising the full TaskBroker pipeline

Test plan

• All 9 scanner ordering tests pass (enqueue at time=0, expedite, restart, concurrency grants, retries, priority ordering, multi-group, bulk drain)
• All 21 floating concurrency tests pass
• Existing dequeue, enqueue, cancel, and metrics tests pass
• Clippy clean
• CI green

🤖 Generated with Claude Code

---

Note

Medium Risk
Changes core dequeue/scanner behavior by introducing a cached iterator and new stop/resume semantics around future tasks and wakeups, which could affect task availability/latency if incorrect. Risk is mitigated by extensive new regression tests covering behind-cursor task creation paths.

Overview
Optimizes TaskBroker scanning to avoid redundant DB re-reads by caching a SlateDB DbIterator across scan passes, advancing a cursor through the task-group key range instead of restarting from the beginning every time.

The scan loop now refreshes the iterator on wakeup(), on iterator exhaustion, or after MAX_ITER_AGE (5s) to pick up newly-ready scheduled work; it also halts at the first future task and carries that consumed entry forward for re-evaluation on the next pass.

Adds a new task_scanner_ordering_tests.rs suite to ensure tasks created “behind the cursor” (e.g., start_at_ms=0, expedite/restart, concurrency grants, retries, priority ordering, multi-group, bulk enqueue) are still eventually dequeued, and updates task_scanner_profile benchmark to measure real sustained dequeue throughput instead of synthetic scan timing.

^{Written by Cursor Bugbot for commit cdc0501. This will update automatically on new commits. Configure here.}

[cursor]

Cursor Bugbot has reviewed your changes and found 1 potential issue.

^{Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, enable autofix in the Cursor dashboard.}