feat: add wait_for_new_round to Creator trait and remove orchestrator polling

[bagelface]

Summary

• Adds wait_for_new_round(current: u64) -> Result<(Vec<u8>, u64)> to the Creator trait, which blocks until the round advances past current
• Removes executed_rounds: HashSet<u64> from the orchestrator's run loop — the orchestrator now calls wait_for_new_round after a successful execution and is guaranteed to receive a fresh round before broadcasting again
• Removes the 2-second polling sleep and receiver-drain guard that were needed to suppress re-broadcast of already-executed rounds
• Implements wait_for_new_round for CounterCreator (polls provider at 2s intervals), ListeningCounterCreator (waits for next queued task then checks round), CounterCreatorType (dispatches to variant), and MockCreator (increments counter)

Test plan

• cargo fmt --all — clean
• cargo clippy --all-targets --all-features -- -D warnings — clean
• cargo test --all-targets --all-features — all 44 tests pass, including new test_wait_for_new_round_returns_strictly_greater_round
• Existing orchestrator integration tests (test_executor_called_exactly_once_after_threshold, test_metrics_observed_on_quorum_path, test_metrics_round_timeout_counted) all continue to pass

Closes #153

[BreadrichEngels]

Reviewed against #153. The structure matches the issue's intent — wait_for_new_round is added to the Creator trait, and the orchestrator's executed_rounds: HashSet + 2s polling sleep + receiver-drain guard are all gone, which are exactly the deliverables #153 asked for. But the way the orchestrator consumes the new method reintroduces correctness problems on the queue-backed (listening/ingress) creator — i.e. the production-shaped path. A few of these are blocking.

---

🚫 1. The orchestrator discards wait_for_new_round's return and re-fetches — double-consuming the task queue.

#153 prescribed using the returned value:

let (payload, current_round) = self.task_creator.wait_for_new_round(executed_round).await?;
// guaranteed fresh round

The implementation instead throws the result away and breaks, so the outer loop calls get_payload_and_round() again (types.rs):

self.task_creator.wait_for_new_round(msg.round).await.unwrap();
break;   // → outer loop → get_payload_and_round() refetches

For CounterCreator that's just a wasted get_current_round() RPC. For ListeningCounterCreator it's data loss: both wait_for_new_round and get_payload_and_round call wait_for_task(), which does self.queue.pop(). So per executed round the queue is popped twice — once in wait_for_new_round (the task is stored in current_task, then immediately discarded when the refetch overwrites it) and once in get_payload_and_round. Every other submitted task is silently dropped, never broadcast or executed. Worse: if the provider round hasn't advanced yet, wait_for_new_round keeps looping and popping/discarding tasks until get_current_round() ticks — it can drain the whole queue. Fix by using the returned (payload, round) and restructuring the outer loop so it doesn't immediately re-fetch (the issue's design), rather than break-and-refetch.

🚫 2. .unwrap() on wait_for_new_round crashes the orchestrator — including on normal idle.

self.task_creator.wait_for_new_round(msg.round).await.unwrap() panics on any Err. For ListeningCounterCreator, wait_for_task() returns Err("Timeout waiting for task after …ms") when no task arrives within timeout_ms — so an idle queue makes wait_for_new_round return Err, and the orchestrator panics and dies. The old executed_rounds path just slept 2s and continued, so this is a new crash-on-idle. (The pre-existing get_payload_and_round().await.unwrap() at the top of the loop has the same smell, but this PR adds a second, more easily-triggered one.) Handle the error — log and continue/retry — instead of unwrapping.

🚫 3. The P2P receiver is no longer serviced while blocked in wait_for_new_round.

The removed executed_rounds branch deliberately kept draining the receiver during its 2s sleep ("still draining the receiver to prevent buffer build-up" — its own comment). Now, after execution the orchestrator is parked in wait_for_new_round(...).await, which only polls the provider / waits for a task and never reads receiver. For the listening creator that can block indefinitely between tasks. With the bounded P2P backlog (service#181 / the configurable backlog work), late signatures and other channel traffic accumulate with nothing draining them — backpressure/drops. Whatever the new design, it needs to keep servicing receiver during the inter-round wait (e.g. select! the wait against receiver.recv()), as the old code did.

---

Coordination (not blocking on its own, but plan for it):

• wait_for_new_round is a new required trait method with no default impl, so this is a breaking change to the public Creator API. The real production creator lives in gas-killer/service and won't compile against this until it implements the method — and once it does, it inherits bug #1 unless the orchestrator is fixed first. This needs a coordinated service-side PR (same pattern as the #171/#274 pair). Consider whether a provided default is feasible, though for a queue-backed creator a naive default would hit the same consume hazard — fixing the orchestrator wiring is the better lever.
• Per your own audit comment on #153, this loop is also being reworked by #164/#165 (event-driven Start re-broadcast largely subsumes this polling concern). Sequence these so they don't stomp each other.

Minor:

• Test coverage doesn't exercise the risk. MockCreator::wait_for_new_round just delegates to get_payload_and_round (counter++), so it models no queue — the double-consume, idle-timeout panic, and receiver-starvation all pass through green. test_wait_for_new_round_returns_strictly_greater_round only checks the counter increments. Add a queue-backed test asserting no task is dropped across an execute→next-round transition (and that get_payload_and_round isn't called an extra time per round).
• Style nit: CounterCreator uses fully-qualified tokio::time::… while ListeningCounterCreator/its wait_for_task do use tokio::time::{Duration, sleep}; inside the fn — pick one.
• CI: no checks are reporting on the head commit (status came back empty) and the fmt/clippy/test boxes in the test plan are unchecked — confirm green before merge.

Net: the trait addition and the removal of the HashSet/2s-poll machinery are the right direction for #153, but the break-and-refetch wiring needs to become use-the-returned-value, the unwraps need real error handling, and the receiver must stay drained during the wait. Happy to re-review once those are addressed.

[BreadrichEngels]

Re-review — all three blockers from the last pass are addressed, and cleanly. Clearing the change request.

1. Double-consume fixed ✅ The orchestrator now caches wait_for_new_round's result in cached_round and the outer loop consumes it via cached_round.take() before falling back to get_payload_and_round(). So after a successful execution the next round is driven by the value wait_for_new_round returned — no second fetch, exactly one queue pop per round (this is the design #153 prescribed). The get_task_metadata() read also lines up now, since the broadcast uses the task that wait_for_new_round→wait_for_task popped. The new test_get_payload_not_recalled_after_execution (TrackingCreator asserting get_count == 1, wait_count == 1) pins this contract directly — nice, that's a more robust guard than a queue-based test would have been.

2. No more unwrap panic ✅ Both the top-of-loop get_payload_and_round and the post-exec wait_for_new_round now match on Err — the former logs and continues, the latter logs and retries via wait_fut.set(...). test_orchestrator_survives_wait_for_new_round_error confirms the run loop stays alive after a wait error instead of the task finishing on a panic.

3. Receiver stays drained ✅ The post-execution wait is a tokio::select! (biased) between the wait_for_new_round future and receiver.recv(), so P2P traffic is serviced during the inter-round gap — restoring the property the old executed_rounds branch had. Biased polling also means a round advance is picked up promptly.

Style nit from last time also handled (use tokio::time::{Duration, sleep} hoisted, CounterCreator uses sleep(...) consistently).

Two non-blocking notes for follow-up:

• Retry backoff. The post-exec retry loop calls wait_for_new_round again immediately on Err with no delay. The listening creator self-paces (its wait_for_task sleeps internally up to timeout_ms), but a creator that errors immediately — e.g. CounterCreator::wait_for_new_round returns on the first get_current_round().await? with no sleep — would hot-spin and flood warn! logs if the provider is persistently down. A small backoff before re-setting the future would bound that.
• Coordination still stands. wait_for_new_round remains a required trait method with no default, so this is still a breaking change to the public Creator API: the gas-killer/service production creator must implement it (under the same single-consume contract the orchestrator now relies on) before/with the dependency bump — the #171/#274 pattern. And per your own audit note on #153, sequence against #164/#165 since they rework this same loop.

CI still shows no checks reporting on the head commit (status empty) and the test-plan boxes are unchecked — confirm fmt/clippy/test are green before merge, especially that the pin!/select! retry block is clippy-clean. Otherwise this is good to go for #153. LGTM.