fair RPC batching layer

[MartinquaXD]

Description

When we inspect Solvable_orders::update() - the function responsible for assembling the final auction - with tempo we see that sometimes fetching balances takes a very long time (>100ms). This is very bad since we only issue very few RPC requests fetching new balances because basically all balances we need are already cached.
The reason why this takes so long is that the background task that updates balances whenever we see a new block saturates the RPC buffering layer. What happens is effectively this:

1. new block appears
2. balance cache immediately sends 1000 requests to the batching layer
3. Solvable_orders::update() processes orders
   a. it finds a missing balance - gets enqueued AFTER the 1000 already enqueue requests
4. 1000 requests of balances background task get processed (>100ms)
5. Solvable_orders::update() RPC calls get resolved (~few ms)

That's why simply delaying the balances cache background task for 1 or 2 seconds causes Solvable_orders::update() to be significantly faster and less spikey.

There are a few ways how to handle this:

• simply delay balances queries
  • obviously hacky
• introduce some notification mechanism between auction building and balances
  • slightly less hacky but gets awkward when we deal with chains that have multiple blocks between auctions
• buffer balances queries (e.g. at most n calls in flight at one time)
  • requires tons of changes and only addresses the issue for the balances task

Changes

What I ended up going for was to simply make the buffering layer fairer.
Instead of every caller sending requests to 1 global queue which allows 1 spammy producer to significantly delay other producers we now store requests in 1 queue per producer (identified by tokio task id). Then when it comes time to assemble the batches we send over the wire we simply do round robin across all sub-queues.
That way the problem should be resolved for all producers without actually having to change any callers or introduce additional communication channels.

To facilitate this the batching logic provided in futures (calls.chunks_timeout(config.ethrpc_max_batch_size, config.ethrpc_batch_delay)) needed to be replaced with handrolled logic but the behavior was preserved.

The last change is that we now spawn a separate task per batch instead of doing everything in 1 task. This allows multiple cores to work on those tasks in parallel and was informed by the fact that the task regularly blocks the IO worker for >50ms when we are processing huge requests or responses (instrumented with dial9).

How to test

added basic unit test to verify that the round robin logic works

I temporarily tested 2 variants of this PR in prod:

1. the version of the PR (spawn 1 task per batch)
2. 1 worker task that polls all batch futures concurrently via FuturesUnordered

There is definitely a noticeable improvement but unfortunately not as large as simply delaying the balances cache task. This PR still makes sense to me as it addresses a fundamental issue and whatever change we apply to the balance caching task (if we even do anything) would still cooperate with this PR.

[claude]

Claude finished @MartinquaXD's task in 4m 46s —— View job

---

PR Review: fair RPC batching layer

• Read the changed file and understand the diff vs origin/main
• Verify the round-robin fairness logic and FairQueue invariants
• Check the hand-rolled batching (drain / blocking-wait / batch_delay flush) preserves chunks_timeout behavior
• Check per-batch task spawning, semaphore backpressure, and error/cancel handling
• Post findings

---

Nice change — the approach is clean and the fairness reasoning holds up. I traced the motivating scenario (spammy producer floods 1000 balance calls, solvable_orders::update() enqueues one) through collect_requests → build_fair_batch: the single request lands in the very next assembled batch (2nd round-robin slot) instead of behind the 1000, so the goal is achieved even with the default batch_delay: ZERO. Response routing via per-caller sub-queues and the oneshot senders checks out, and the FairQueue invariant (round_robin membership ⇔ non-empty per-caller queue, len = sum of sub-queue lengths) is maintained across enqueue/pop.

Correctness: I did not find any hard bugs.

• The try_recv drain → blocking-wait → batch_delay timeout sequence in collect_requests faithfully reproduces the old chunks_timeout(max_batch_size, batch_delay) semantics, including resetting the delay window per batch and leftover items flushing promptly (leftover ≥ max_batch_size skips the wait).
• Concurrency is now bounded by the Semaphore (default 10) instead of for_each_concurrent, with the permit moved into the spawned task so it's released on completion/panic — equivalent, and the per-batch spawn is a reasonable way to spread large ser/de work across cores.
• Canceled senders are correctly skipped in both build_fair_batch and process_batch, and duplicate-id handling is unchanged.

One optional (non-blocking) note left inline: the permit is acquired after the batch is assembled, so under sustained overload a built batch is held out of the queue while blocked on acquire_owned(), and requests arriving during that wait only get their fair slot in the next batch. Acquiring first, then collecting/building, would yield fresher/fuller batches in exactly the overload case this PR targets. Marginal — up to you.

Minor observations (not worth acting on): test coverage is limited to the zero-delay round-robin path (the batch_delay timeout and concurrency paths are exercised only in prod, which the PR description acknowledges); and fairness degenerates to a single sub-queue for callers sharing caller = None or spawning a task per request — both already documented in the module doc comment.
• fair-rpc-batching-layer

[claude]

Claude finished @MartinquaXD's task in 4m 2s —— View job

---

PR Review: fair RPC batching layer

• Read crates/ethrpc/src/alloy/buffering.rs and the diff vs origin/main
• Verify FairQueue invariants + round-robin fairness
• Verify hand-rolled collect_requests preserves chunks_timeout semantics
• Verify per-batch task spawning + semaphore backpressure + permit ordering
• Confirm the earlier perf note (acquire-permit-first) was addressed

---

Solid change — the fairness approach is well reasoned and the implementation holds up. I re-traced the motivating scenario (spammy balance producer floods requests, solvable_orders::update() enqueues one) and confirmed the latency-sensitive request lands in the very next assembled batch's round-robin slot instead of behind the backlog.

Correctness: no bugs found.

• FairQueue invariant (round_robin membership ⇔ non-empty per-caller queue; len = Σ sub-queue lengths) is maintained across enqueue/pop, and len -= 1 can't underflow (only reached when round_robin is non-empty).
• collect_requests (drain → block-for-first → batch_delay window, skipping the window when the delay is zero or the queue already has ≥ max_batch_size) faithfully reproduces the old chunks_timeout(max_batch_size, batch_delay) behavior, including leftover items flushing promptly on the next iteration.
• Permit is now acquired before assembling the batch (commit 46893ee07), so batches are built from the freshest queue state and newly-arrived callers get immediate round-robin participation — this resolves the perf note from the prior review. The permit is moved into the spawned task, so it's released on completion/panic; concurrency stays bounded by the Semaphore.
• Canceled senders are skipped in both build_fair_batch and process_batch, and duplicate-id FIFO handling via BatchRequestEntry is unchanged.

One minor, non-blocking observation (edge case, up to you — no change requested): if the calls channel closes while collect_requests has already drained items into the FairQueue (e.g. during the batch_delay select on line 337), the worker returns false and exits without processing those buffered requests, so their senders drop and callers see a receive error. The old chunks_timeout would emit the final partial chunk before the stream ended. This only occurs at shutdown, when the provider (and thus callers) are being torn down anyway, so impact is negligible — flagging only for completeness.

The limited test coverage (zero-delay round-robin only) and the caller = None / task-per-request fairness degeneration are already acknowledged in the PR description and module doc comment, so I won't belabor them.

Nothing here blocks merge. 👍
• fair-rpc-batching-layer

[jmg-duarte]

LGTM, just have a small nit