Skip to content

Commit 075f0d5

Browse files
fix(CallBatchLayer): don't batch if single request (#2397)
* Raising Draft pr CallBatchLayer single request * Single Rpc Call * numbing testcase * numbing tc * numbing tc + fmt * fmt * Refactor+ adding tc * fmt * fmt * doc + removing clone * fmt * chore: clean up --------- Co-authored-by: DaniPopes <57450786+DaniPopes@users.noreply.github.com>
1 parent a04aea9 commit 075f0d5

1 file changed

Lines changed: 94 additions & 59 deletions

File tree

crates/provider/src/layers/batch.rs

Lines changed: 94 additions & 59 deletions
Original file line numberDiff line numberDiff line change
@@ -123,15 +123,29 @@ where
123123

124124
type CallBatchMsgTx = TransportResult<IMulticall3::Result>;
125125

126-
struct CallBatchMsg {
127-
call: IMulticall3::Call3,
126+
struct CallBatchMsg<N: Network> {
127+
kind: CallBatchMsgKind<N>,
128128
tx: oneshot::Sender<CallBatchMsgTx>,
129129
}
130130

131-
impl fmt::Debug for CallBatchMsg {
131+
impl<N: Network> Clone for CallBatchMsgKind<N>
132+
where
133+
N::TransactionRequest: Clone,
134+
{
135+
fn clone(&self) -> Self {
136+
match self {
137+
Self::Call(tx) => Self::Call(tx.clone()),
138+
Self::BlockNumber => Self::BlockNumber,
139+
Self::ChainId => Self::ChainId,
140+
Self::Balance(addr) => Self::Balance(*addr),
141+
}
142+
}
143+
}
144+
145+
impl<N: Network> fmt::Debug for CallBatchMsg<N> {
132146
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
133147
f.write_str("BatchProviderMessage(")?;
134-
self.call.fmt(f)?;
148+
self.kind.fmt(f)?;
135149
f.write_str(")")
136150
}
137151
}
@@ -144,18 +158,15 @@ enum CallBatchMsgKind<N: Network = Ethereum> {
144158
Balance(Address),
145159
}
146160

147-
impl CallBatchMsg {
148-
fn new<N: Network>(
149-
kind: CallBatchMsgKind<N>,
150-
m3a: Address,
151-
) -> (Self, oneshot::Receiver<CallBatchMsgTx>) {
161+
impl<N: Network> CallBatchMsg<N> {
162+
fn new(kind: CallBatchMsgKind<N>) -> (Self, oneshot::Receiver<CallBatchMsgTx>) {
152163
let (tx, rx) = oneshot::channel();
153-
(Self { call: kind.into_call3(m3a), tx }, rx)
164+
(Self { kind, tx }, rx)
154165
}
155166
}
156167

157168
impl<N: Network> CallBatchMsgKind<N> {
158-
fn into_call3(self, m3a: Address) -> IMulticall3::Call3 {
169+
fn to_call3(&self, m3a: Address) -> IMulticall3::Call3 {
159170
let m3a_call = |data: Vec<u8>| IMulticall3::Call3 {
160171
target: m3a,
161172
allowFailure: true,
@@ -169,7 +180,7 @@ impl<N: Network> CallBatchMsgKind<N> {
169180
},
170181
Self::BlockNumber => m3a_call(IMulticall3::getBlockNumberCall {}.abi_encode()),
171182
Self::ChainId => m3a_call(IMulticall3::getChainIdCall {}.abi_encode()),
172-
Self::Balance(addr) => m3a_call(IMulticall3::getEthBalanceCall { addr }.abi_encode()),
183+
&Self::Balance(addr) => m3a_call(IMulticall3::getEthBalanceCall { addr }.abi_encode()),
173184
}
174185
}
175186
}
@@ -179,7 +190,7 @@ impl<N: Network> CallBatchMsgKind<N> {
179190
/// See [`CallBatchLayer`] for more information.
180191
pub struct CallBatchProvider<P, N: Network = Ethereum> {
181192
provider: Arc<P>,
182-
inner: CallBatchProviderInner,
193+
inner: CallBatchProviderInner<N>,
183194
_pd: PhantomData<N>,
184195
}
185196

@@ -201,28 +212,23 @@ impl<P: Provider<N> + 'static, N: Network> CallBatchProvider<P, N> {
201212
fn new(inner: P, layer: &CallBatchLayer) -> Self {
202213
let inner = Arc::new(inner);
203214
let tx = CallBatchBackend::spawn(inner.clone(), layer);
204-
Self {
205-
provider: inner,
206-
inner: CallBatchProviderInner { tx, m3a: layer.m3a },
207-
_pd: PhantomData,
208-
}
215+
Self { provider: inner, inner: CallBatchProviderInner { tx }, _pd: PhantomData }
209216
}
210217
}
211218

212219
#[derive(Clone)]
213-
struct CallBatchProviderInner {
214-
tx: mpsc::UnboundedSender<CallBatchMsg>,
215-
m3a: Address,
220+
struct CallBatchProviderInner<N: Network> {
221+
tx: mpsc::UnboundedSender<CallBatchMsg<N>>,
216222
}
217223

218-
impl CallBatchProviderInner {
224+
impl<N: Network> CallBatchProviderInner<N> {
219225
/// We only want to perform a scheduled multicall if:
220226
/// - The request has no block ID or state overrides,
221227
/// - The request has a target address,
222228
/// - The request has no other properties (`nonce`, `gas`, etc cannot be sent with a multicall).
223229
///
224230
/// Ref: <https://github.com/wevm/viem/blob/ba8319f71503af8033fd3c77cfb64c7eb235c6a9/src/actions/public/call.ts#L295>
225-
fn should_batch_call<N: Network>(&self, params: &crate::EthCallParams<N>) -> bool {
231+
fn should_batch_call(&self, params: &crate::EthCallParams<N>) -> bool {
226232
// TODO: block ID is not yet implemented
227233
if params.block().is_some_and(|block| block != BlockId::latest()) {
228234
return false;
@@ -242,25 +248,28 @@ impl CallBatchProviderInner {
242248
true
243249
}
244250

245-
async fn schedule<N: Network>(self, msg: CallBatchMsgKind<N>) -> TransportResult<Bytes> {
246-
let (msg, rx) = CallBatchMsg::new(msg, self.m3a);
251+
async fn schedule(self, msg: CallBatchMsgKind<N>) -> TransportResult<Bytes> {
252+
let (msg, rx) = CallBatchMsg::new(msg);
247253
self.tx.send(msg).map_err(|_| TransportErrorKind::backend_gone())?;
248254

249-
let IMulticall3::Result { success, returnData: data } =
255+
let IMulticall3::Result { success, returnData } =
250256
rx.await.map_err(|_| TransportErrorKind::backend_gone())??;
257+
251258
if !success {
252-
let revert_data = if data.is_empty() { "" } else { &format!(" with data: {data}") };
253-
return Err(TransportErrorKind::custom_str(&format!(
259+
let revert_data = if returnData.is_empty() {
260+
"".to_string()
261+
} else {
262+
format!(" with data: {returnData}")
263+
};
264+
Err(TransportErrorKind::custom_str(&format!(
254265
"multicall batched call reverted{revert_data}"
255-
)));
266+
)))
267+
} else {
268+
Ok(returnData)
256269
}
257-
Ok(data)
258270
}
259271

260-
async fn schedule_and_decode<N: Network, T>(
261-
self,
262-
msg: CallBatchMsgKind<N>,
263-
) -> TransportResult<T>
272+
async fn schedule_and_decode<T>(self, msg: CallBatchMsgKind<N>) -> TransportResult<T>
264273
where
265274
T: SolValue + From<<T::SolType as SolType>::RustType>,
266275
{
@@ -273,13 +282,13 @@ struct CallBatchBackend<P, N: Network = Ethereum> {
273282
inner: Arc<P>,
274283
m3a: Address,
275284
wait: Duration,
276-
rx: mpsc::UnboundedReceiver<CallBatchMsg>,
277-
pending: Vec<CallBatchMsg>,
285+
rx: mpsc::UnboundedReceiver<CallBatchMsg<N>>,
286+
pending: Vec<CallBatchMsg<N>>,
278287
_pd: PhantomData<N>,
279288
}
280289

281290
impl<P: Provider<N> + 'static, N: Network> CallBatchBackend<P, N> {
282-
fn spawn(inner: Arc<P>, layer: &CallBatchLayer) -> mpsc::UnboundedSender<CallBatchMsg> {
291+
fn spawn(inner: Arc<P>, layer: &CallBatchLayer) -> mpsc::UnboundedSender<CallBatchMsg<N>> {
283292
let CallBatchLayer { m3a, wait } = *layer;
284293
let (tx, rx) = mpsc::unbounded_channel();
285294
let this = Self { inner, m3a, wait, rx, pending: Vec::new(), _pd: PhantomData };
@@ -311,13 +320,22 @@ impl<P: Provider<N> + 'static, N: Network> CallBatchBackend<P, N> {
311320
}
312321
}
313322

314-
fn process_msg(&mut self, msg: CallBatchMsg) {
323+
fn process_msg(&mut self, msg: CallBatchMsg<N>) {
315324
self.pending.push(msg);
316325
}
317326

318327
async fn send_batch(&mut self) {
319-
let result = self.send_batch_inner().await;
320328
let pending = std::mem::take(&mut self.pending);
329+
330+
// If there's only a single call, avoid batching and perform the request directly.
331+
if pending.len() == 1 {
332+
let msg = pending.into_iter().next().unwrap();
333+
let result = self.call_one(msg.kind).await;
334+
let _ = msg.tx.send(result);
335+
return;
336+
}
337+
338+
let result = self.send_batch_inner(&pending).await;
321339
match result {
322340
Ok(results) => {
323341
debug_assert_eq!(results.len(), pending.len());
@@ -333,28 +351,45 @@ impl<P: Provider<N> + 'static, N: Network> CallBatchBackend<P, N> {
333351
}
334352
}
335353

336-
async fn send_batch_inner(&mut self) -> TransportResult<Vec<IMulticall3::Result>> {
337-
debug_assert!(!self.pending.is_empty());
338-
debug!(len = self.pending.len(), "sending multicall");
339-
let tx = N::TransactionRequest::default().with_to(self.m3a).with_input(self.make_payload());
354+
async fn call_one(&mut self, msg: CallBatchMsgKind<N>) -> TransportResult<IMulticall3::Result> {
355+
let m3_res =
356+
|success, return_data| IMulticall3::Result { success, returnData: return_data };
357+
match msg {
358+
CallBatchMsgKind::Call(tx) => self.inner.call(tx).await.map(|res| m3_res(true, res)),
359+
CallBatchMsgKind::BlockNumber => {
360+
self.inner.get_block_number().await.map(|res| m3_res(true, res.abi_encode().into()))
361+
}
362+
CallBatchMsgKind::ChainId => {
363+
self.inner.get_chain_id().await.map(|res| m3_res(true, res.abi_encode().into()))
364+
}
365+
CallBatchMsgKind::Balance(addr) => {
366+
self.inner.get_balance(addr).await.map(|res| m3_res(true, res.abi_encode().into()))
367+
}
368+
}
369+
}
370+
371+
async fn send_batch_inner(
372+
&self,
373+
pending: &[CallBatchMsg<N>],
374+
) -> TransportResult<Vec<IMulticall3::Result>> {
375+
let calls: Vec<_> = pending.iter().map(|msg| msg.kind.to_call3(self.m3a)).collect();
376+
377+
let tx = N::TransactionRequest::default()
378+
.with_to(self.m3a)
379+
.with_input(IMulticall3::aggregate3Call { calls }.abi_encode());
380+
340381
let bytes = self.inner.call(tx).await?;
341382
if bytes.is_empty() {
342383
return Err(TransportErrorKind::custom_str(&format!(
343384
"Multicall3 not deployed at {}",
344385
self.m3a
345386
)));
346387
}
388+
347389
let ret = IMulticall3::aggregate3Call::abi_decode_returns(&bytes)
348390
.map_err(TransportErrorKind::custom)?;
349391
Ok(ret)
350392
}
351-
352-
fn make_payload(&self) -> Vec<u8> {
353-
IMulticall3::aggregate3Call {
354-
calls: self.pending.iter().map(|msg| msg.call.clone()).collect(),
355-
}
356-
.abi_encode()
357-
}
358393
}
359394

360395
impl<P: Provider<N> + 'static, N: Network> Provider<N> for CallBatchProvider<P, N> {
@@ -374,7 +409,7 @@ impl<P: Provider<N> + 'static, N: Network> Provider<N> for CallBatchProvider<P,
374409
alloy_primitives::BlockNumber,
375410
> {
376411
crate::ProviderCall::BoxedFuture(Box::pin(
377-
self.inner.clone().schedule_and_decode::<N, u64>(CallBatchMsgKind::BlockNumber),
412+
self.inner.clone().schedule_and_decode::<u64>(CallBatchMsgKind::BlockNumber),
378413
))
379414
}
380415

@@ -386,7 +421,7 @@ impl<P: Provider<N> + 'static, N: Network> Provider<N> for CallBatchProvider<P,
386421
alloy_primitives::ChainId,
387422
> {
388423
crate::ProviderCall::BoxedFuture(Box::pin(
389-
self.inner.clone().schedule_and_decode::<N, u64>(CallBatchMsgKind::ChainId),
424+
self.inner.clone().schedule_and_decode::<u64>(CallBatchMsgKind::ChainId),
390425
))
391426
}
392427

@@ -399,25 +434,25 @@ impl<P: Provider<N> + 'static, N: Network> Provider<N> for CallBatchProvider<P,
399434
ProviderCall::BoxedFuture(Box::pin(
400435
this.inner
401436
.clone()
402-
.schedule_and_decode::<N, U256>(CallBatchMsgKind::Balance(address)),
437+
.schedule_and_decode::<U256>(CallBatchMsgKind::Balance(address)),
403438
))
404439
}
405440
})
406441
}
407442
}
408443

409-
struct CallBatchCaller {
410-
inner: CallBatchProviderInner,
444+
struct CallBatchCaller<N: Network> {
445+
inner: CallBatchProviderInner<N>,
411446
weak: WeakClient,
412447
}
413448

414-
impl CallBatchCaller {
415-
fn new<P: Provider<N>, N: Network>(provider: &CallBatchProvider<P, N>) -> Self {
449+
impl<N: Network> CallBatchCaller<N> {
450+
fn new<P: Provider<N>>(provider: &CallBatchProvider<P, N>) -> Self {
416451
Self { inner: provider.inner.clone(), weak: provider.provider.weak_client() }
417452
}
418453
}
419454

420-
impl<N: Network> Caller<N, Bytes> for CallBatchCaller {
455+
impl<N: Network> Caller<N, Bytes> for CallBatchCaller<N> {
421456
fn call(
422457
&self,
423458
params: crate::EthCallParams<N>,
@@ -427,7 +462,7 @@ impl<N: Network> Caller<N, Bytes> for CallBatchCaller {
427462
}
428463

429464
Ok(crate::ProviderCall::BoxedFuture(Box::pin(
430-
self.inner.clone().schedule::<N>(CallBatchMsgKind::Call(params.into_data())),
465+
self.inner.clone().schedule(CallBatchMsgKind::Call(params.into_data())),
431466
)))
432467
}
433468

0 commit comments

Comments
 (0)