From d5a336bc4ee586f05ee791ead696ee089429d45e Mon Sep 17 00:00:00 2001 From: MartinquaXD Date: Sun, 2 Aug 2026 07:56:10 +0000 Subject: [PATCH 1/4] fairer buffering impl --- crates/ethrpc/src/alloy/buffering.rs | 457 ++++++++++++++++++++++----- 1 file changed, 374 insertions(+), 83 deletions(-) diff --git a/crates/ethrpc/src/alloy/buffering.rs b/crates/ethrpc/src/alloy/buffering.rs index dcff520a64..8279d7574f 100644 --- a/crates/ethrpc/src/alloy/buffering.rs +++ b/crates/ethrpc/src/alloy/buffering.rs @@ -10,23 +10,35 @@ //! read from. The background task then does the batching, forwards the requests //! to the next layer, and reports the results of the individual calls back via //! another channel. +//! +//! To prevent a single caller from monopolizing the pipeline (e.g. the balance +//! cache dumping thousands of requests just before the auction assembler needs +//! a handful), each enqueued call is tagged with the tokio task id of the +//! caller. The background worker keeps a queue per caller and assembles +//! batches by round-robin across those queues. Fairness is best-effort: it +//! depends on each subsystem driving its RPC calls from a stable task rather +//! than spawning a fresh task per request. use { crate::Config, alloy_json_rpc::{RequestPacket, Response, ResponsePacket, SerializedRequest}, alloy_transport::{RpcError, TransportError, TransportErrorKind}, futures::{ - channel::{mpsc, oneshot}, + channel::{ + mpsc::{self, TryRecvError}, + oneshot, + }, stream::StreamExt as _, }, std::{ - collections::{HashMap, VecDeque}, + collections::{HashMap, VecDeque, hash_map::Entry}, fmt::Debug, marker::PhantomData, pin::Pin, + sync::Arc, task::{Context, Poll}, + time::Duration, }, - tokio::task::JoinHandle, - tokio_stream::StreamExt, + tokio::{sync::Semaphore, task::JoinHandle}, tower::{Layer, Service}, }; @@ -62,13 +74,23 @@ where #[derive(Debug, Clone)] pub(crate) struct BatchCallProvider { inner: PhantomData, - calls: mpsc::UnboundedSender, + calls: mpsc::UnboundedSender>>, } -type CallContext = ( - oneshot::Sender>, - SerializedRequest, -); +/// Identifies which tokio task enqueued a call. `None` covers contexts +/// without a task id (e.g. calls issued from blocking tasks). +/// The tokio runtime reuses ids under certain conditions. In practice +/// we should not run into those cases but even if we do the worst thing +/// that can happen is that we think the old task sent more requests +/// which is not a huge issue. +type CallerId = Option; + +struct CallContext { + /// tokio task that issued this request + caller: CallerId, + request: REQ, + response_sender: oneshot::Sender, +} type ResponseSender = oneshot::Sender>>; @@ -136,102 +158,279 @@ where &self, request: SerializedRequest, ) -> oneshot::Receiver> { - let (sender, receiver) = oneshot::channel(); + let (response_sender, receiver) = oneshot::channel(); + // Tag with the caller's tokio task id so the worker can interleave + // requests from different subsystems fairly. + let caller = tokio::task::try_id(); // Theoreticallly we could propagate the error to the caller, however // this is a critical error we can't recover from (i.e. we'll not be // able to send any more RPC calls). That's why we panic ASAP to immediately // cause a restart of the pod if this is running in kubernetes. self.calls - .unbounded_send((sender, request)) + .unbounded_send(CallContext { + caller, + request, + response_sender, + }) .expect("worker task unexpectedly dropped"); receiver } /// Start a background worker for batching buffered requests. + /// + /// The worker keeps one queue per caller (keyed by the tokio task id + /// captured at enqueue time). Each batch is assembled by round-robin + /// across those queues, which prevents a single caller from monopolizing + /// the pipeline when many requests are enqueued in a burst. fn background_worker( mut inner: S, config: Config, - calls: mpsc::UnboundedReceiver, + mut calls: mpsc::UnboundedReceiver< + CallContext>, + >, ) -> JoinHandle<()> { - let process_batch = move |batch: Vec<(ResponseSender, SerializedRequest)>| { - // Clones service via [`std::mem::replace`] as recommended by - // - let clone: S = inner.clone(); - let mut inner = std::mem::replace(&mut inner, clone); - - // Map because even with random IDs we might get duplicates, - // (e.g. some ID outgrew another and now they overlap) in that case - // we use the Deque to enforce FIFO and hope the node didn't re-order responses - let mut senders: HashMap<_, BatchRequestEntry> = HashMap::with_capacity(batch.len()); - let mut requests = Vec::with_capacity(batch.len()); - - async move { - for (sender, request) in batch { - if sender.is_canceled() { - tracing::trace!(request_id = %request.id(), "canceled sender"); - continue; - } + let semaphore = Arc::new(Semaphore::new(config.ethrpc_max_concurrent_requests)); + let max_batch_size = config.ethrpc_max_batch_size; + let batch_delay = config.ethrpc_batch_delay; - match senders.entry(request.id().clone()) { - std::collections::hash_map::Entry::Occupied(mut occupied_entry) => { - occupied_entry.get_mut().push_back(sender); - } - std::collections::hash_map::Entry::Vacant(vacant_entry) => { - vacant_entry.insert(BatchRequestEntry::new(sender)); - } - } - requests.push(request); - } + tokio::task::spawn(async move { + let mut queue = + FairQueue::>::default(); - if requests.is_empty() { - tracing::trace!("all callers stopped awaiting their request"); + loop { + if !queue + .collect_requests(&mut calls, max_batch_size, batch_delay) + .await + { + tracing::debug!("rpc batching channel closed"); return; } - let result = inner - .call(RequestPacket::Batch(requests)) + let batch = queue.build_fair_batch(max_batch_size); + + // wait for a concurrency slot to apply backpressure + let permit = semaphore + .clone() + .acquire_owned() .await - .map(|response| match response { - ResponsePacket::Batch(res) => res, - ResponsePacket::Single(res) => { - tracing::warn!("received single response for batch request"); - vec![res] - } - }); - - match result { - Ok(responses) => { - for response in responses { - tracing::trace!(response_id = %response.id, "attempting to remove response"); - let Some(entry) = senders.get_mut(&response.id) else { - tracing::warn!(response_id = %response.id, "missing sender for response"); - continue; - }; - let Some(sender) = entry.pop_front() else { - tracing::warn!(response_id = %response.id, "more responses than senders (may have lost some sender)"); - continue; - }; - tracing::debug!(response_id = %response.id, "sending response"); - let _ = sender.send(Ok(response)); - } - } - Err(err) => { - let err = format!("batch call failed: {err:?}"); - senders - .into_values() - .flat_map(|sender| sender.into_iter()) - .for_each(|sender| { - let _ = sender.send(Err(TransportErrorKind::custom_str(&err))); - }); + .expect("semaphore never closed"); + + // Clone the inner service per batch as recommended in + // . + let clone = inner.clone(); + let this_inner = std::mem::replace(&mut inner, clone); + + tokio::task::spawn(async move { + // move permit into the task so we only return it when + // task is done + let _permit = permit; + process_batch(this_inner, batch).await; + }); + } + }) + } +} + +/// Fair FIFO queue of pending calls partitioned by caller. Items are +/// enqueued into per-caller sub-queues and dequeued round-robin, so a +/// single caller with a large backlog cannot monopolize the pipeline. +/// +/// Invariant: a caller is in `round_robin` iff its entry in `per_caller` +/// is non-empty, and `len` equals the sum of all per-caller queue lengths. +struct FairQueue { + per_caller: HashMap)>>, + round_robin: VecDeque, + len: usize, +} + +impl Default for FairQueue { + fn default() -> Self { + Self { + per_caller: Default::default(), + round_robin: Default::default(), + len: 0, + } + } +} + +impl FairQueue { + /// Enqueues request and adds caller to the round-robin queue if necessary. + fn enqueue(&mut self, call: CallContext) { + let queue = self.per_caller.entry(call.caller).or_default(); + let first = queue.is_empty(); + queue.push_back((call.request, call.response_sender)); + if first { + self.round_robin.push_back(call.caller); + } + self.len += 1; + } + + /// Pop the next call in round-robin order, if any. + fn pop(&mut self) -> Option<(REQ, oneshot::Sender)> { + let caller = self.round_robin.pop_front()?; + let queue = self + .per_caller + .get_mut(&caller) + .expect("caller in round_robin has a non-empty per-caller queue"); + let item = queue + .pop_front() + .expect("caller in round_robin has a non-empty per-caller queue"); + if queue.is_empty() { + self.per_caller.remove(&caller); + } else { + self.round_robin.push_back(caller); + } + self.len -= 1; + Some(item) + } + + fn len(&self) -> usize { + self.len + } + + fn is_empty(&self) -> bool { + self.len == 0 + } + + /// Drains the items from `calls` into the fair queue until we have at least + /// `max_batch_size` items or waited for `batch_delay` after starting to + /// build a batch. + /// 1. drain all immediately available items from `calls` + /// 2. if we still have no items, await the next one + /// 3. if we don't have at least `max_batch_size` items yet, await for + /// `batch_delay` time longer (to avoid tiny batches) + /// + /// Returns `true` if the `calls` channels is still alive and continuing + /// to process requests makes sense. + async fn collect_requests( + &mut self, + calls: &mut mpsc::UnboundedReceiver>, + max_batch_size: usize, + batch_delay: Duration, + ) -> bool { + loop { + match calls.try_recv() { + Ok(call) => self.enqueue(call), + Err(TryRecvError::Closed) => { + return false; + } + Err(TryRecvError::Empty) => break, + } + } + + // wait for the next request to appear. do this outside the select + // loop below to only start the timeout when we actually have an item + // in the pipeline + if self.is_empty() { + let Some(call) = calls.next().await else { + return false; + }; + self.enqueue(call); + } + + if self.len() < max_batch_size && !batch_delay.is_zero() { + let deadline = tokio::time::sleep(batch_delay); + tokio::pin!(deadline); + while self.len() < max_batch_size { + tokio::select! { + _ = &mut deadline => break, + msg = calls.next() => { + let Some(call) = msg else { + return false; + }; + self.enqueue(call); } } } - }; - tokio::task::spawn( - calls - .chunks_timeout(config.ethrpc_max_batch_size, config.ethrpc_batch_delay) - .for_each_concurrent(config.ethrpc_max_concurrent_requests, process_batch), - ) + } + + true + } + + /// Batches at most `max_batch_size` items in a round-robin fashion to + /// prevent individual callers from starving all the others. + fn build_fair_batch(&mut self, max_batch_size: usize) -> Vec<(REQ, oneshot::Sender)> { + let mut batch = Vec::with_capacity(self.len().min(max_batch_size)); + while batch.len() < max_batch_size { + let Some((request, sender)) = self.pop() else { + break; + }; + if !sender.is_canceled() { + // only add to batch if caller is still waiting for response + batch.push((request, sender)); + } + } + batch + } +} + +async fn process_batch(mut inner: S, batch: Vec<(SerializedRequest, ResponseSender)>) +where + S: Service, +{ + // Map because even with random IDs we might get duplicates, + // (e.g. some ID outgrew another and now they overlap) in that case + // we use the Deque to enforce FIFO and hope the node didn't re-order responses + let mut senders: HashMap<_, BatchRequestEntry> = HashMap::with_capacity(batch.len()); + let mut requests = Vec::with_capacity(batch.len()); + + for (request, sender) in batch { + if sender.is_canceled() { + tracing::trace!(request_id = %request.id(), "canceled sender"); + continue; + } + match senders.entry(request.id().clone()) { + Entry::Occupied(mut occupied_entry) => { + occupied_entry.get_mut().push_back(sender); + } + Entry::Vacant(vacant_entry) => { + vacant_entry.insert(BatchRequestEntry::new(sender)); + } + } + requests.push(request); + } + + if requests.is_empty() { + tracing::trace!("all callers stopped awaiting their request"); + return; + } + + let result = inner + .call(RequestPacket::Batch(requests)) + .await + .map(|response| match response { + ResponsePacket::Batch(res) => res, + ResponsePacket::Single(res) => { + tracing::warn!("received single response for batch request"); + vec![res] + } + }); + + match result { + Ok(responses) => { + for response in responses { + tracing::trace!(response_id = %response.id, "attempting to remove response"); + let Some(entry) = senders.get_mut(&response.id) else { + tracing::warn!(response_id = %response.id, "missing sender for response"); + continue; + }; + let Some(sender) = entry.pop_front() else { + tracing::warn!(response_id = %response.id, "more responses than senders (may have lost some sender)"); + continue; + }; + tracing::debug!(response_id = %response.id, "sending response"); + let _ = sender.send(Ok(response)); + } + } + Err(err) => { + let err = format!("batch call failed: {err:?}"); + senders + .into_values() + .flat_map(|sender| sender.into_iter()) + .for_each(|sender| { + let _ = sender.send(Err(TransportErrorKind::custom_str(&err))); + }); + } } } @@ -288,7 +487,7 @@ where #[cfg(test)] mod test { - use {crate::alloy::buffering::BatchRequestEntry, futures::channel::oneshot}; + use {super::*, futures::FutureExt as _}; #[test] fn test_batch_request_entry_pop_twice() { @@ -319,4 +518,96 @@ mod test { let third_pop = entry.pop_front(); assert!(third_pop.is_none()); } + + /// Tests that the fair queue builds batches in a round robin fashion. + /// Also tests that the associated response senders send the data to + /// the correct caller. + #[tokio::test] + async fn batching_does_round_robin() { + let (request_sender, mut receiver) = mpsc::unbounded(); + let mut queue = FairQueue::default(); + + fn call_context(index: u64) -> (CallContext, oneshot::Receiver) { + let (response_sender, receiver) = oneshot::channel(); + let context = CallContext { + caller: tokio::task::try_id(), + request: index, + response_sender, + }; + (context, receiver) + } + + // spammy producer that enques 100 calls before other + // tasks even start + let mut response_receivers: Vec<_> = (0..100) + .map(|id| { + let (context, receiver) = call_context(id); + request_sender.unbounded_send(context).unwrap(); + receiver + }) + .collect(); + + for id in 100..103 { + let request_sender2 = request_sender.clone(); + + // enqueue calls from new separate tasks to give each one + // its own queue to test round robin (keyed by task id) + #[allow(clippy::async_yields_async)] + let receiver = tokio::task::spawn(async move { + let (context, receiver) = call_context(id); + request_sender2.unbounded_send(context).unwrap(); + receiver + }) + .await + .unwrap(); + response_receivers.push(receiver); + } + + let should_continue = queue + .collect_requests(&mut receiver, 5, Default::default()) + .now_or_never() + .expect("if we have enough requests already enqueued this is actually sync"); + assert!(should_continue); + + let batch = queue.build_fair_batch(5); + + // ASSERT THAT BATCH WAS FAIR (ROUND ROBIN) + assert_eq!(batch.len(), 5); + let mut iter = batch.into_iter(); + // first request of spammy producer + let (request, sender) = iter.next().unwrap(); + assert_eq!(request, 0); + sender.send(0).unwrap(); + + // requests of other producers + let (request, sender) = iter.next().unwrap(); + assert_eq!(request, 100); + sender.send(100).unwrap(); + let (request, sender) = iter.next().unwrap(); + assert_eq!(request, 101); + sender.send(101).unwrap(); + let (request, sender) = iter.next().unwrap(); + assert_eq!(request, 102); + sender.send(102).unwrap(); + + // round robin wrapped around so this is the second request of the spammy + // producer + let (request, sender) = iter.next().unwrap(); + assert_eq!(request, 1); + sender.send(1).unwrap(); + + // ASSERT THAT RESPONSES REACHED THE CORRECT CALLERS + let mut responses = response_receivers.into_iter(); + // first 2 calls of the spammy producer resolved + assert_eq!(responses.next().unwrap().now_or_never().unwrap(), Ok(0)); + assert_eq!(responses.next().unwrap().now_or_never().unwrap(), Ok(1)); + // next 98 calls fo the spammy producer did not resolve yet + for _ in 0..98 { + assert!(responses.next().unwrap().now_or_never().is_none()); + } + // requests of other producers resolved + assert_eq!(responses.next().unwrap().now_or_never().unwrap(), Ok(100)); + assert_eq!(responses.next().unwrap().now_or_never().unwrap(), Ok(101)); + assert_eq!(responses.next().unwrap().now_or_never().unwrap(), Ok(102)); + } } From 24341393970ec135fca69154cfc54bdc9694e053 Mon Sep 17 00:00:00 2001 From: MartinquaXD Date: Sun, 2 Aug 2026 17:36:29 +0000 Subject: [PATCH 2/4] fixup comment --- crates/ethrpc/src/alloy/buffering.rs | 12 +++++------- 1 file changed, 5 insertions(+), 7 deletions(-) diff --git a/crates/ethrpc/src/alloy/buffering.rs b/crates/ethrpc/src/alloy/buffering.rs index 8279d7574f..e6d6fef83d 100644 --- a/crates/ethrpc/src/alloy/buffering.rs +++ b/crates/ethrpc/src/alloy/buffering.rs @@ -11,13 +11,11 @@ //! to the next layer, and reports the results of the individual calls back via //! another channel. //! -//! To prevent a single caller from monopolizing the pipeline (e.g. the balance -//! cache dumping thousands of requests just before the auction assembler needs -//! a handful), each enqueued call is tagged with the tokio task id of the -//! caller. The background worker keeps a queue per caller and assembles -//! batches by round-robin across those queues. Fairness is best-effort: it -//! depends on each subsystem driving its RPC calls from a stable task rather -//! than spawning a fresh task per request. +//! To prevent a single caller from monopolizing the pipeline, each enqueued +//! call is tagged with the tokio task id of the caller. The background worker +//! keeps a queue per caller and assembles batches by round-robin across those +//! queues. Fairness is best-effort: it depends on each subsystem driving its +//! RPC calls from a stable task rather than spawning a fresh task per request. use { crate::Config, alloy_json_rpc::{RequestPacket, Response, ResponsePacket, SerializedRequest}, From 46893ee078cd8413b9589fd4c18bd87cc2eb2160 Mon Sep 17 00:00:00 2001 From: MartinquaXD Date: Mon, 3 Aug 2026 06:08:35 +0000 Subject: [PATCH 3/4] acquire permit before assembling batch --- crates/ethrpc/src/alloy/buffering.rs | 16 +++++++++------- 1 file changed, 9 insertions(+), 7 deletions(-) diff --git a/crates/ethrpc/src/alloy/buffering.rs b/crates/ethrpc/src/alloy/buffering.rs index e6d6fef83d..a49e3c6ab1 100644 --- a/crates/ethrpc/src/alloy/buffering.rs +++ b/crates/ethrpc/src/alloy/buffering.rs @@ -196,6 +196,15 @@ where FairQueue::>::default(); loop { + // first wait for a concurrency slot to become available. + // that way we end up with the most up-to-data batch + // possible in the end. + let permit = semaphore + .clone() + .acquire_owned() + .await + .expect("semaphore never closed"); + if !queue .collect_requests(&mut calls, max_batch_size, batch_delay) .await @@ -206,13 +215,6 @@ where let batch = queue.build_fair_batch(max_batch_size); - // wait for a concurrency slot to apply backpressure - let permit = semaphore - .clone() - .acquire_owned() - .await - .expect("semaphore never closed"); - // Clone the inner service per batch as recommended in // . let clone = inner.clone(); From 74892ef29977d54763f8982d135782a992f4d88c Mon Sep 17 00:00:00 2001 From: MartinquaXD Date: Mon, 3 Aug 2026 14:00:34 +0000 Subject: [PATCH 4/4] CamelCase type parameters --- crates/ethrpc/src/alloy/buffering.rs | 22 +++++++++++----------- 1 file changed, 11 insertions(+), 11 deletions(-) diff --git a/crates/ethrpc/src/alloy/buffering.rs b/crates/ethrpc/src/alloy/buffering.rs index a49e3c6ab1..bbd6750422 100644 --- a/crates/ethrpc/src/alloy/buffering.rs +++ b/crates/ethrpc/src/alloy/buffering.rs @@ -83,11 +83,11 @@ pub(crate) struct BatchCallProvider { /// which is not a huge issue. type CallerId = Option; -struct CallContext { +struct CallContext { /// tokio task that issued this request caller: CallerId, - request: REQ, - response_sender: oneshot::Sender, + request: Req, + response_sender: oneshot::Sender, } type ResponseSender = oneshot::Sender>>; @@ -237,13 +237,13 @@ where /// /// Invariant: a caller is in `round_robin` iff its entry in `per_caller` /// is non-empty, and `len` equals the sum of all per-caller queue lengths. -struct FairQueue { - per_caller: HashMap)>>, +struct FairQueue { + per_caller: HashMap)>>, round_robin: VecDeque, len: usize, } -impl Default for FairQueue { +impl Default for FairQueue { fn default() -> Self { Self { per_caller: Default::default(), @@ -253,9 +253,9 @@ impl Default for FairQueue { } } -impl FairQueue { +impl FairQueue { /// Enqueues request and adds caller to the round-robin queue if necessary. - fn enqueue(&mut self, call: CallContext) { + fn enqueue(&mut self, call: CallContext) { let queue = self.per_caller.entry(call.caller).or_default(); let first = queue.is_empty(); queue.push_back((call.request, call.response_sender)); @@ -266,7 +266,7 @@ impl FairQueue { } /// Pop the next call in round-robin order, if any. - fn pop(&mut self) -> Option<(REQ, oneshot::Sender)> { + fn pop(&mut self) -> Option<(Req, oneshot::Sender)> { let caller = self.round_robin.pop_front()?; let queue = self .per_caller @@ -304,7 +304,7 @@ impl FairQueue { /// to process requests makes sense. async fn collect_requests( &mut self, - calls: &mut mpsc::UnboundedReceiver>, + calls: &mut mpsc::UnboundedReceiver>, max_batch_size: usize, batch_delay: Duration, ) -> bool { @@ -349,7 +349,7 @@ impl FairQueue { /// Batches at most `max_batch_size` items in a round-robin fashion to /// prevent individual callers from starving all the others. - fn build_fair_batch(&mut self, max_batch_size: usize) -> Vec<(REQ, oneshot::Sender)> { + fn build_fair_batch(&mut self, max_batch_size: usize) -> Vec<(Req, oneshot::Sender)> { let mut batch = Vec::with_capacity(self.len().min(max_batch_size)); while batch.len() < max_batch_size { let Some((request, sender)) = self.pop() else {