1use std::collections::HashMap;
104use std::sync::{Arc, OnceLock};
105
106use alloy_eips::BlockId;
107use alloy_primitives::{Address, B256, Bytes, U256, hex};
108use alloy_provider::Provider;
109use alloy_provider::network::AnyNetwork;
110use alloy_rpc_types_eth::TransactionRequest;
111use alloy_rpc_types_eth::state::{AccountOverride, StateOverride};
112use alloy_sol_types::SolCall;
113use futures::stream::{self, StreamExt};
114use tracing::{debug, warn};
115
116use crate::cache::{StorageBatchFetchFn, block_in_place_handle};
117use crate::errors::{StorageFetchError, StorageFetchResult};
118use crate::multicall::{IMulticall3, MULTICALL3_ADDRESS};
119
120pub const STORAGE_EXTRACTOR_CODE: &[u8] = &hex!("5f5b80361460135780355481526020016001565b365ff3");
127
128pub const STORAGE_EXTRACTOR_CODE_PRE_SHANGHAI: &[u8] =
131 &hex!("60005b80361460145780355481526020016002565b366000f3");
132
133pub fn multicall3_runtime_code() -> &'static Bytes {
141 static CODE: OnceLock<Bytes> = OnceLock::new();
142 CODE.get_or_init(|| {
143 let raw = include_str!("../fixtures/multicall3_runtime.hex");
144 Bytes::from(hex::decode(raw.trim()).expect("valid multicall3 runtime hex fixture"))
145 })
146}
147
148#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
150pub enum CallDispatch {
151 #[default]
155 PerCall,
156 CallMany,
164}
165
166#[derive(Debug, Clone, Copy, PartialEq, Eq)]
175pub struct BulkCallConfig {
176 pub max_slots_per_call: usize,
180 pub max_targets_per_call: usize,
183 pub max_concurrent_calls: usize,
186 pub point_read_threshold: usize,
191 pub pre_shanghai_extractor: bool,
193 pub dispatch: CallDispatch,
196 pub max_slots_per_request: usize,
202 pub max_request_bytes: usize,
209}
210
211impl Default for BulkCallConfig {
212 fn default() -> Self {
213 Self {
214 max_slots_per_call: 10_000,
215 max_targets_per_call: 250,
216 max_concurrent_calls: 4,
217 point_read_threshold: 2,
218 pre_shanghai_extractor: false,
219 dispatch: CallDispatch::PerCall,
220 max_slots_per_request: 25_000,
221 max_request_bytes: 2_400_000,
222 }
223 }
224}
225
226impl BulkCallConfig {
227 fn normalized(self) -> Self {
228 const JSON_RPC_ENVELOPE_RESERVE: usize = 512;
229 const JSON_HEX_BYTES_PER_SLOT: usize = 64;
230 let byte_limited_slots = self
231 .max_request_bytes
232 .saturating_sub(JSON_RPC_ENVELOPE_RESERVE)
233 .checked_div(JSON_HEX_BYTES_PER_SLOT)
234 .unwrap_or(0)
235 .max(1);
236 Self {
237 max_slots_per_call: self.max_slots_per_call.max(1).min(byte_limited_slots),
238 max_targets_per_call: self.max_targets_per_call.max(1),
239 max_concurrent_calls: self.max_concurrent_calls.max(1),
240 max_slots_per_request: self.max_slots_per_request.max(1).min(byte_limited_slots),
241 max_request_bytes: self
242 .max_request_bytes
243 .max(JSON_RPC_ENVELOPE_RESERVE.saturating_add(JSON_HEX_BYTES_PER_SLOT)),
244 ..self
245 }
246 }
247
248 fn extractor(&self) -> Bytes {
249 if self.pre_shanghai_extractor {
250 Bytes::from_static(STORAGE_EXTRACTOR_CODE_PRE_SHANGHAI)
251 } else {
252 Bytes::from_static(STORAGE_EXTRACTOR_CODE)
253 }
254 }
255}
256
257pub fn pack_slots_calldata(slots: &[U256]) -> Bytes {
260 let mut out = Vec::with_capacity(slots.len() * 32);
261 for slot in slots {
262 out.extend_from_slice(&slot.to_be_bytes::<32>());
263 }
264 out.into()
265}
266
267pub fn decode_packed_values(data: &[u8], expected: usize) -> Option<Vec<U256>> {
273 if data.len() != expected * 32 {
274 return None;
275 }
276 Some(
277 data.as_chunks::<32>()
278 .0
279 .iter()
280 .map(|chunk| U256::from_be_slice(chunk))
281 .collect(),
282 )
283}
284
285pub fn encode_multi_target_calldata(targets: &[(Address, Vec<U256>)]) -> Bytes {
290 let calls: Vec<IMulticall3::Call3> = targets
291 .iter()
292 .map(|(target, slots)| IMulticall3::Call3 {
293 target: *target,
294 allowFailure: true,
295 callData: pack_slots_calldata(slots),
296 })
297 .collect();
298 IMulticall3::aggregate3Call { calls }.abi_encode().into()
299}
300
301pub fn decode_multi_target_response(
304 targets: &[(Address, Vec<U256>)],
305 response: &[u8],
306) -> Vec<(Address, U256, StorageFetchResult<U256>)> {
307 let decoded = match IMulticall3::aggregate3Call::abi_decode_returns(response) {
308 Ok(results) if results.len() == targets.len() => results,
309 Ok(results) => {
310 return per_target_errors(targets, || {
311 StorageFetchError::custom(format!(
312 "aggregate3 returned {} results for {} extraction targets",
313 results.len(),
314 targets.len()
315 ))
316 });
317 }
318 Err(e) => {
319 return per_target_errors(targets, || {
320 StorageFetchError::custom(format!("failed to decode aggregate3 response: {e}"))
321 });
322 }
323 };
324
325 let mut out = Vec::with_capacity(targets.iter().map(|(_, s)| s.len()).sum());
326 for ((target, slots), result) in targets.iter().zip(decoded) {
327 if !result.success {
328 out.extend(slots.iter().map(|slot| {
329 (
330 *target,
331 *slot,
332 Err(StorageFetchError::custom(
333 "extractor subcall failed (allowFailure=true); the target may be a precompile",
334 )),
335 )
336 }));
337 continue;
338 }
339 match decode_packed_values(&result.returnData, slots.len()) {
340 Some(values) => out.extend(
341 slots
342 .iter()
343 .zip(values)
344 .map(|(slot, value)| (*target, *slot, Ok(value))),
345 ),
346 None => out.extend(slots.iter().map(|slot| {
347 (
348 *target,
349 *slot,
350 Err(StorageFetchError::custom(format!(
351 "extractor at {target} returned {} bytes, expected {}",
352 result.returnData.len(),
353 slots.len() * 32
354 ))),
355 )
356 })),
357 }
358 }
359 out
360}
361
362fn per_target_errors(
363 targets: &[(Address, Vec<U256>)],
364 make: impl Fn() -> StorageFetchError,
365) -> Vec<(Address, U256, StorageFetchResult<U256>)> {
366 targets
367 .iter()
368 .flat_map(|(target, slots)| slots.iter().map(|slot| (*target, *slot, Err(make()))))
369 .collect()
370}
371
372#[derive(Debug, Clone, PartialEq, Eq)]
374enum CallPlan {
375 Single { target: Address, slots: Vec<U256> },
377 Multi { targets: Vec<(Address, Vec<U256>)> },
379}
380
381impl CallPlan {
382 fn request_slot_count(&self) -> usize {
383 match self {
384 Self::Single { slots, .. } => slots.len(),
385 Self::Multi { targets } => targets.iter().map(|(_, s)| s.len()).sum(),
386 }
387 }
388}
389
390fn plan_calls(requests: &[(Address, U256)], config: &BulkCallConfig) -> Vec<CallPlan> {
399 let mut order: Vec<Address> = Vec::new();
400 let mut groups: HashMap<Address, Vec<U256>> = HashMap::new();
401 for (address, slot) in requests {
402 groups
403 .entry(*address)
404 .or_insert_with(|| {
405 order.push(*address);
406 Vec::new()
407 })
408 .push(*slot);
409 }
410
411 let mut plans = Vec::new();
412 let mut packable: Vec<(Address, Vec<U256>)> = Vec::new();
413 for address in order {
414 let slots = groups.remove(&address).expect("grouped above");
415 for chunk in slots.chunks(config.max_slots_per_call) {
416 let full = chunk.len() == config.max_slots_per_call;
417 if full || address == MULTICALL3_ADDRESS {
422 plans.push(CallPlan::Single {
423 target: address,
424 slots: chunk.to_vec(),
425 });
426 } else {
427 packable.push((address, chunk.to_vec()));
428 }
429 }
430 }
431
432 let mut current: Vec<(Address, Vec<U256>)> = Vec::new();
434 let mut current_slots = 0usize;
435 let flush =
436 |current: &mut Vec<(Address, Vec<U256>)>, plans: &mut Vec<CallPlan>| match current.len() {
437 0 => {}
438 1 => {
439 let (target, slots) = current.pop().expect("len checked");
440 plans.push(CallPlan::Single { target, slots });
441 }
442 _ => plans.push(CallPlan::Multi {
443 targets: std::mem::take(current),
444 }),
445 };
446 for (address, slots) in packable {
447 let would_overflow = current_slots + slots.len() > config.max_slots_per_call
448 || current.len() >= config.max_targets_per_call;
449 if !current.is_empty() && would_overflow {
450 flush(&mut current, &mut plans);
451 current_slots = 0;
452 }
453 current_slots += slots.len();
454 current.push((address, slots));
455 }
456 flush(&mut current, &mut plans);
457
458 plans
459}
460
461fn overrides_for_plan(plan: &CallPlan, extractor: &Bytes) -> StateOverride {
464 let mut overrides = StateOverride::default();
465 match plan {
466 CallPlan::Single { target, .. } => {
467 overrides.insert(
468 *target,
469 AccountOverride::default().with_code(extractor.clone()),
470 );
471 }
472 CallPlan::Multi { targets } => {
473 overrides.insert(
474 MULTICALL3_ADDRESS,
475 AccountOverride::default().with_code(multicall3_runtime_code().clone()),
476 );
477 for (target, _) in targets {
478 overrides.insert(
479 *target,
480 AccountOverride::default().with_code(extractor.clone()),
481 );
482 }
483 }
484 }
485 overrides
486}
487
488fn plan_call_parts(plan: &CallPlan) -> (Address, Bytes) {
490 match plan {
491 CallPlan::Single { target, slots } => (*target, pack_slots_calldata(slots)),
492 CallPlan::Multi { targets } => (MULTICALL3_ADDRESS, encode_multi_target_calldata(targets)),
493 }
494}
495
496fn decode_plan_response(
498 plan: &CallPlan,
499 bytes: &[u8],
500) -> Vec<(Address, U256, StorageFetchResult<U256>)> {
501 match plan {
502 CallPlan::Single { target, slots } => match decode_packed_values(bytes, slots.len()) {
503 Some(values) => slots
504 .iter()
505 .zip(values)
506 .map(|(slot, value)| (*target, *slot, Ok(value)))
507 .collect(),
508 None => slots
509 .iter()
510 .map(|slot| {
511 (
512 *target,
513 *slot,
514 Err(StorageFetchError::custom(format!(
515 "extractor at {target} returned {} bytes, expected {} — the \
516 provider may not support eth_call state overrides, or the \
517 target is a precompile",
518 bytes.len(),
519 slots.len() * 32
520 ))),
521 )
522 })
523 .collect(),
524 },
525 CallPlan::Multi { targets } => decode_multi_target_response(targets, bytes),
526 }
527}
528
529fn plan_error_results(
531 plan: &CallPlan,
532 err: StorageFetchError,
533) -> Vec<(Address, U256, StorageFetchResult<U256>)> {
534 match plan {
535 CallPlan::Single { target, slots } => slots
536 .iter()
537 .map(|slot| (*target, *slot, Err(err.clone())))
538 .collect(),
539 CallPlan::Multi { targets } => targets
540 .iter()
541 .flat_map(|(target, slots)| {
542 slots.iter().map({
543 let err = err.clone();
544 move |slot| (*target, *slot, Err(err.clone()))
545 })
546 })
547 .collect(),
548 }
549}
550
551async fn execute_plan<P: Provider<AnyNetwork>>(
552 provider: &P,
553 block: BlockId,
554 plan: CallPlan,
555 extractor: &Bytes,
556) -> Vec<(Address, U256, StorageFetchResult<U256>)> {
557 let overrides = overrides_for_plan(&plan, extractor);
558 let (to, data) = plan_call_parts(&plan);
559 let tx = TransactionRequest::default().to(to).input(data.into());
560
561 let response: Result<Bytes, _> = provider
562 .client()
563 .request("eth_call", (tx, block, overrides))
564 .await;
565
566 match response {
567 Ok(bytes) => decode_plan_response(&plan, &bytes),
568 Err(e) => plan_error_results(&plan, StorageFetchError::provider("eth_call", &e)),
569 }
570}
571
572#[derive(Debug, serde::Deserialize)]
575struct CallManyEntry {
576 value: Option<Bytes>,
577 error: Option<serde_json::Value>,
578}
579
580async fn execute_plans_call_many<P: Provider<AnyNetwork>>(
587 provider: &P,
588 number: alloy_eips::BlockNumberOrTag,
589 plans: &[CallPlan],
590 extractor: &Bytes,
591) -> Result<Vec<(Address, U256, StorageFetchResult<U256>)>, StorageFetchError> {
592 let mut overrides = StateOverride::default();
597 let mut transactions = Vec::with_capacity(plans.len());
598 for plan in plans {
599 for (address, account) in overrides_for_plan(plan, extractor) {
600 overrides.insert(address, account);
601 }
602 let (to, data) = plan_call_parts(plan);
603 transactions.push(serde_json::json!({ "to": to, "data": data }));
604 }
605
606 let bundles = serde_json::json!([{ "transactions": transactions }]);
607 let context = serde_json::json!({ "blockNumber": number, "transactionIndex": -1 });
608 let response: Vec<Vec<CallManyEntry>> = provider
609 .client()
610 .request("eth_callMany", (bundles, context, overrides))
611 .await
612 .map_err(|e| StorageFetchError::provider("eth_callMany", &e))?;
613
614 let entries: Vec<CallManyEntry> = response.into_iter().flatten().collect();
615 if entries.len() != plans.len() {
616 return Err(StorageFetchError::custom(format!(
617 "eth_callMany returned {} results for {} bundled calls",
618 entries.len(),
619 plans.len()
620 )));
621 }
622
623 let mut out = Vec::new();
624 for (plan, entry) in plans.iter().zip(entries) {
625 match entry.value {
626 Some(bytes) => out.extend(decode_plan_response(plan, &bytes)),
627 None => {
628 let detail = entry
629 .error
630 .map(|e| e.to_string())
631 .unwrap_or_else(|| "no value returned".to_string());
632 out.extend(plan_error_results(
633 plan,
634 StorageFetchError::custom(format!("eth_callMany transaction failed: {detail}")),
635 ));
636 }
637 }
638 }
639 Ok(out)
640}
641
642fn group_plans_for_call_many(
645 plans: Vec<CallPlan>,
646 max_slots_per_request: usize,
647) -> Vec<Vec<CallPlan>> {
648 let mut requests: Vec<Vec<CallPlan>> = Vec::new();
649 let mut current: Vec<CallPlan> = Vec::new();
650 let mut current_slots = 0usize;
651 for plan in plans {
652 let slots = plan.request_slot_count();
653 if !current.is_empty() && current_slots + slots > max_slots_per_request {
654 requests.push(std::mem::take(&mut current));
655 current_slots = 0;
656 }
657 current_slots += slots;
658 current.push(plan);
659 }
660 if !current.is_empty() {
661 requests.push(current);
662 }
663 requests
664}
665
666pub async fn fetch_slots_bulk<P: Provider<AnyNetwork>>(
678 provider: &P,
679 requests: Vec<(Address, U256)>,
680 block: BlockId,
681 config: BulkCallConfig,
682) -> Vec<(Address, U256, StorageFetchResult<U256>)> {
683 let config = config.normalized();
684 if requests.is_empty() {
685 return Vec::new();
686 }
687 let extractor = config.extractor();
688 let plans = plan_calls(&requests, &config);
689 debug!(
690 slots = requests.len(),
691 calls = plans.len(),
692 dispatch = ?config.dispatch,
693 "bulk storage extraction dispatch"
694 );
695
696 let extractor = &extractor;
697 let call_many_number = match (config.dispatch, block) {
700 (CallDispatch::CallMany, BlockId::Number(number)) => Some(number),
701 _ => None,
702 };
703
704 let Some(number) = call_many_number else {
705 let results: Vec<Vec<_>> = stream::iter(
706 plans
707 .into_iter()
708 .map(|plan| execute_plan(provider, block, plan, extractor)),
709 )
710 .buffer_unordered(config.max_concurrent_calls)
711 .collect()
712 .await;
713 return results.into_iter().flatten().collect();
714 };
715
716 let (conflicting, bundleable): (Vec<_>, Vec<_>) = plans.into_iter().partition(
720 |plan| matches!(plan, CallPlan::Single { target, .. } if *target == MULTICALL3_ADDRESS),
721 );
722 let groups = group_plans_for_call_many(bundleable, config.max_slots_per_request);
723 let group_futs = groups.into_iter().map(|group| async move {
724 match execute_plans_call_many(provider, number, &group, extractor).await {
725 Ok(results) => results,
726 Err(e) => {
727 warn!(
731 error = %e,
732 chunks = group.len(),
733 "eth_callMany dispatch failed; re-dispatching per-call"
734 );
735 let mut results = Vec::new();
736 for plan in group {
737 results.extend(execute_plan(provider, block, plan, extractor).await);
738 }
739 results
740 }
741 }
742 });
743 let mut results: Vec<_> = stream::iter(group_futs)
744 .buffer_unordered(config.max_concurrent_calls)
745 .collect::<Vec<Vec<_>>>()
746 .await
747 .into_iter()
748 .flatten()
749 .collect();
750 for plan in conflicting {
751 results.extend(execute_plan(provider, block, plan, extractor).await);
752 }
753 results
754}
755
756pub fn planned_call_count(requests: &[(Address, U256)], config: &BulkCallConfig) -> usize {
762 plan_calls(requests, &config.normalized()).len()
763}
764
765#[derive(Clone, Debug)]
781pub struct BulkFetcherStatus {
782 consecutive_failures: Arc<std::sync::atomic::AtomicUsize>,
783 latch_threshold: usize,
784}
785
786impl BulkFetcherStatus {
787 pub fn consecutive_override_failures(&self) -> usize {
793 self.consecutive_failures
794 .load(std::sync::atomic::Ordering::Relaxed)
795 }
796
797 pub fn latch_threshold(&self) -> usize {
799 self.latch_threshold
800 }
801
802 pub fn fallback_latched(&self) -> bool {
808 self.consecutive_override_failures() >= self.latch_threshold
809 }
810}
811
812pub fn bulk_call_storage_fetcher<P: Provider<AnyNetwork> + 'static>(
824 provider: Arc<P>,
825 config: BulkCallConfig,
826) -> StorageBatchFetchFn {
827 make_fetcher(provider, config, None).0
828}
829
830pub fn bulk_call_storage_fetcher_with_fallback<P: Provider<AnyNetwork> + 'static>(
840 provider: Arc<P>,
841 config: BulkCallConfig,
842 fallback: StorageBatchFetchFn,
843) -> StorageBatchFetchFn {
844 make_fetcher(provider, config, Some(fallback)).0
845}
846
847pub fn bulk_call_storage_fetcher_with_status<P: Provider<AnyNetwork> + 'static>(
857 provider: Arc<P>,
858 config: BulkCallConfig,
859 fallback: StorageBatchFetchFn,
860) -> (StorageBatchFetchFn, BulkFetcherStatus) {
861 make_fetcher(provider, config, Some(fallback))
862}
863
864fn make_fetcher<P: Provider<AnyNetwork> + 'static>(
865 provider: Arc<P>,
866 config: BulkCallConfig,
867 fallback: Option<StorageBatchFetchFn>,
868) -> (StorageBatchFetchFn, BulkFetcherStatus) {
869 let config = config.normalized();
870 const OVERRIDE_FAILURE_LATCH: usize = 2;
877 let consecutive_failures = Arc::new(std::sync::atomic::AtomicUsize::new(0));
878 let status = BulkFetcherStatus {
879 consecutive_failures: Arc::clone(&consecutive_failures),
880 latch_threshold: OVERRIDE_FAILURE_LATCH,
881 };
882 let fetcher: StorageBatchFetchFn =
883 Arc::new(move |requests: Vec<(Address, U256)>, block: BlockId| {
884 use std::sync::atomic::Ordering;
885 if requests.is_empty() {
886 return Vec::new();
887 }
888 if let Some(fallback) = &fallback
889 && (requests.len() < config.point_read_threshold
890 || consecutive_failures.load(Ordering::Relaxed) >= OVERRIDE_FAILURE_LATCH)
891 {
892 return fallback(requests, block);
893 }
894
895 let bulk_results = match block_in_place_handle() {
902 Ok(handle) => tokio::task::block_in_place(|| {
903 handle.block_on(fetch_slots_bulk(provider.as_ref(), requests, block, config))
904 }),
905 Err(e) => requests
906 .into_iter()
907 .map(|(addr, slot)| (addr, slot, Err(StorageFetchError::Runtime(e.clone()))))
908 .collect(),
909 };
910
911 let Some(fallback) = &fallback else {
912 return bulk_results;
913 };
914
915 if bulk_results.iter().any(|(_, _, r)| r.is_ok()) {
918 consecutive_failures.store(0, Ordering::Relaxed);
919 } else if bulk_results
920 .iter()
921 .any(|(_, _, r)| matches!(r, Err(StorageFetchError::Provider { .. })))
922 {
923 let streak = consecutive_failures.fetch_add(1, Ordering::Relaxed) + 1;
924 if streak == OVERRIDE_FAILURE_LATCH {
925 warn!(
926 streak,
927 "bulk storage extraction failed consecutive batches with provider errors; \
928 latching this fetcher to the point-read fallback (install a fresh fetcher \
929 to retry bulk extraction)"
930 );
931 }
932 }
933
934 let mut repaired = Vec::with_capacity(bulk_results.len());
937 let mut failed: Vec<(Address, U256)> = Vec::new();
938 for (addr, slot, result) in bulk_results {
939 match result {
940 Ok(value) => repaired.push((addr, slot, Ok(value))),
941 Err(_) => failed.push((addr, slot)),
942 }
943 }
944 if !failed.is_empty() {
945 warn!(
946 failed = failed.len(),
947 "bulk storage extraction failed for some slots; repairing via fallback fetcher"
948 );
949 repaired.extend(fallback(failed, block));
950 }
951 repaired
952 });
953 (fetcher, status)
954}
955
956#[derive(Debug, Clone, PartialEq, Eq)]
978pub struct StorageProgram {
979 pub target: Address,
982 pub code: Bytes,
984 pub calldata: Bytes,
986}
987
988pub async fn run_storage_program<P: Provider<AnyNetwork>>(
990 provider: &P,
991 block: BlockId,
992 program: &StorageProgram,
993) -> StorageFetchResult<Bytes> {
994 let mut overrides = StateOverride::default();
995 overrides.insert(
996 program.target,
997 AccountOverride::default().with_code(program.code.clone()),
998 );
999 let tx = TransactionRequest::default()
1000 .to(program.target)
1001 .input(program.calldata.clone().into());
1002 provider
1003 .client()
1004 .request("eth_call", (tx, block, overrides))
1005 .await
1006 .map_err(|e| StorageFetchError::provider("eth_call", &e))
1007}
1008
1009pub async fn run_storage_programs<P: Provider<AnyNetwork>>(
1016 provider: &P,
1017 block: BlockId,
1018 programs: &[StorageProgram],
1019) -> Vec<StorageFetchResult<Bytes>> {
1020 let mut seen = std::collections::HashSet::new();
1021 let mut bundle: Vec<usize> = Vec::new();
1022 let mut individual: Vec<usize> = Vec::new();
1023 for (index, program) in programs.iter().enumerate() {
1024 if program.target != MULTICALL3_ADDRESS && seen.insert(program.target) {
1025 bundle.push(index);
1026 } else {
1027 individual.push(index);
1028 }
1029 }
1030 if bundle.len() == 1 {
1032 individual.append(&mut bundle);
1033 }
1034
1035 let mut out: Vec<Option<StorageFetchResult<Bytes>>> = vec![None; programs.len()];
1036
1037 if !bundle.is_empty() {
1038 let mut overrides = StateOverride::default();
1039 overrides.insert(
1040 MULTICALL3_ADDRESS,
1041 AccountOverride::default().with_code(multicall3_runtime_code().clone()),
1042 );
1043 let calls: Vec<IMulticall3::Call3> = bundle
1044 .iter()
1045 .map(|&index| {
1046 let program = &programs[index];
1047 overrides.insert(
1048 program.target,
1049 AccountOverride::default().with_code(program.code.clone()),
1050 );
1051 IMulticall3::Call3 {
1052 target: program.target,
1053 allowFailure: true,
1054 callData: program.calldata.clone(),
1055 }
1056 })
1057 .collect();
1058 let data: Bytes = IMulticall3::aggregate3Call { calls }.abi_encode().into();
1059 let tx = TransactionRequest::default()
1060 .to(MULTICALL3_ADDRESS)
1061 .input(data.into());
1062 let response: Result<Bytes, _> = provider
1063 .client()
1064 .request("eth_call", (tx, block, overrides))
1065 .await;
1066 match response
1067 .map_err(|e| StorageFetchError::provider("eth_call", &e))
1068 .and_then(|bytes| {
1069 IMulticall3::aggregate3Call::abi_decode_returns(&bytes).map_err(|e| {
1070 StorageFetchError::custom(format!("failed to decode aggregate3 response: {e}"))
1071 })
1072 }) {
1073 Ok(results) if results.len() == bundle.len() => {
1074 for (&index, result) in bundle.iter().zip(results) {
1075 out[index] = Some(if result.success {
1076 Ok(result.returnData)
1077 } else {
1078 Err(StorageFetchError::custom(
1079 "storage program subcall failed (allowFailure=true)",
1080 ))
1081 });
1082 }
1083 }
1084 Ok(results) => {
1085 let err = StorageFetchError::custom(format!(
1086 "aggregate3 returned {} results for {} programs",
1087 results.len(),
1088 bundle.len()
1089 ));
1090 for &index in &bundle {
1091 out[index] = Some(Err(err.clone()));
1092 }
1093 }
1094 Err(err) => {
1095 for &index in &bundle {
1096 out[index] = Some(Err(err.clone()));
1097 }
1098 }
1099 }
1100 }
1101
1102 for index in individual {
1103 out[index] = Some(run_storage_program(provider, block, &programs[index]).await);
1104 }
1105
1106 out.into_iter()
1107 .map(|entry| entry.expect("every program resolved"))
1108 .collect()
1109}
1110
1111pub const ACCOUNT_FIELDS_EXTRACTOR_CODE: &[u8] =
1131 &hex!("5f5b803614602057803580318260011b523f8160011b602001526020016001565b3660011b5ff3");
1132
1133#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1136pub struct AccountFieldsSample {
1137 pub balance: U256,
1139 pub code_hash: B256,
1142}
1143
1144pub async fn fetch_account_fields_bulk<P: Provider<AnyNetwork>>(
1153 provider: &P,
1154 addresses: &[Address],
1155 block: BlockId,
1156) -> StorageFetchResult<Vec<(Address, AccountFieldsSample)>> {
1157 if addresses.is_empty() {
1158 return Ok(Vec::new());
1159 }
1160 let mut calldata = Vec::with_capacity(addresses.len() * 32);
1161 for address in addresses {
1162 calldata.extend_from_slice(&[0u8; 12]);
1163 calldata.extend_from_slice(address.as_slice());
1164 }
1165 let program = StorageProgram {
1166 target: MULTICALL3_ADDRESS,
1167 code: Bytes::from_static(ACCOUNT_FIELDS_EXTRACTOR_CODE),
1168 calldata: calldata.into(),
1169 };
1170 let bytes = run_storage_program(provider, block, &program).await?;
1171 if bytes.len() != addresses.len() * 64 {
1172 return Err(StorageFetchError::custom(format!(
1173 "account-fields extractor returned {} bytes, expected {}",
1174 bytes.len(),
1175 addresses.len() * 64
1176 )));
1177 }
1178 Ok(addresses
1179 .iter()
1180 .enumerate()
1181 .map(|(i, address)| {
1182 (
1183 *address,
1184 AccountFieldsSample {
1185 balance: U256::from_be_slice(&bytes[i * 64..i * 64 + 32]),
1186 code_hash: B256::from_slice(&bytes[i * 64 + 32..i * 64 + 64]),
1187 },
1188 )
1189 })
1190 .collect())
1191}
1192
1193pub const BLOCK_CONTEXT_EXTRACTOR_CODE: &[u8] =
1199 &hex!("435f52426020524860405241606052446080524560a0524660c05260e05ff3");
1200
1201#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1204pub struct BlockContextSample {
1205 pub number: u64,
1207 pub timestamp: u64,
1209 pub basefee: U256,
1211 pub coinbase: Address,
1213 pub prevrandao: B256,
1215 pub gas_limit: u64,
1217 pub chain_id: u64,
1219}
1220
1221pub async fn fetch_block_context<P: Provider<AnyNetwork>>(
1224 provider: &P,
1225 block: BlockId,
1226) -> StorageFetchResult<BlockContextSample> {
1227 let program = StorageProgram {
1228 target: MULTICALL3_ADDRESS,
1229 code: Bytes::from_static(BLOCK_CONTEXT_EXTRACTOR_CODE),
1230 calldata: Bytes::new(),
1231 };
1232 let bytes = run_storage_program(provider, block, &program).await?;
1233 if bytes.len() != 7 * 32 {
1234 return Err(StorageFetchError::custom(format!(
1235 "block-context extractor returned {} bytes, expected 224",
1236 bytes.len()
1237 )));
1238 }
1239 let word = |i: usize| U256::from_be_slice(&bytes[i * 32..(i + 1) * 32]);
1240 let to_u64 = |v: U256| u64::try_from(v).unwrap_or(u64::MAX);
1241 Ok(BlockContextSample {
1242 number: to_u64(word(0)),
1243 timestamp: to_u64(word(1)),
1244 basefee: word(2),
1245 coinbase: Address::from_slice(&bytes[3 * 32 + 12..4 * 32]),
1246 prevrandao: B256::from_slice(&bytes[4 * 32..5 * 32]),
1247 gas_limit: to_u64(word(5)),
1248 chain_id: to_u64(word(6)),
1249 })
1250}
1251
1252#[cfg(test)]
1253mod tests {
1254 use super::*;
1255
1256 fn addr(byte: u8) -> Address {
1257 Address::repeat_byte(byte)
1258 }
1259
1260 fn cfg(max_slots: usize, max_targets: usize) -> BulkCallConfig {
1261 BulkCallConfig {
1262 max_slots_per_call: max_slots,
1263 max_targets_per_call: max_targets,
1264 ..BulkCallConfig::default()
1265 }
1266 }
1267
1268 #[test]
1269 fn pack_and_decode_roundtrip() {
1270 let slots = vec![U256::ZERO, U256::from(1u64), U256::MAX];
1271 let packed = pack_slots_calldata(&slots);
1272 assert_eq!(packed.len(), 96);
1273 assert_eq!(&packed[32..64], &U256::from(1u64).to_be_bytes::<32>());
1274 let decoded = decode_packed_values(&packed, 3).expect("exact length");
1275 assert_eq!(decoded, slots);
1276 assert!(decode_packed_values(&packed, 2).is_none());
1277 assert!(decode_packed_values(&packed[..95], 3).is_none());
1278 }
1279
1280 #[test]
1281 fn extractor_constants_are_wellformed() {
1282 assert_eq!(STORAGE_EXTRACTOR_CODE.len(), 23);
1285 assert_eq!(STORAGE_EXTRACTOR_CODE[0], 0x5f, "PUSH0 entry");
1286 assert_eq!(STORAGE_EXTRACTOR_CODE_PRE_SHANGHAI.len(), 25);
1287 assert!(
1288 !STORAGE_EXTRACTOR_CODE_PRE_SHANGHAI.contains(&0x5f),
1289 "pre-Shanghai variant must not use PUSH0"
1290 );
1291 assert!(multicall3_runtime_code().len() > 1_000);
1292 }
1293
1294 #[test]
1295 fn planning_single_small_group() {
1296 let requests = vec![
1297 (addr(0xaa), U256::from(1u64)),
1298 (addr(0xaa), U256::from(2u64)),
1299 ];
1300 let plans = plan_calls(&requests, &cfg(100, 10));
1301 assert_eq!(
1302 plans,
1303 vec![CallPlan::Single {
1304 target: addr(0xaa),
1305 slots: vec![U256::from(1u64), U256::from(2u64)],
1306 }]
1307 );
1308 }
1309
1310 #[test]
1311 fn planning_splits_oversized_target_and_packs_remainder() {
1312 let mut requests: Vec<_> = (0..7u64).map(|i| (addr(0x01), U256::from(i))).collect();
1315 requests.push((addr(0x02), U256::from(99u64)));
1316 let plans = plan_calls(&requests, &cfg(3, 10));
1317 assert_eq!(plans.len(), 3);
1318 assert_eq!(
1319 plans[0],
1320 CallPlan::Single {
1321 target: addr(0x01),
1322 slots: (0..3u64).map(U256::from).collect(),
1323 }
1324 );
1325 assert_eq!(
1326 plans[1],
1327 CallPlan::Single {
1328 target: addr(0x01),
1329 slots: (3..6u64).map(U256::from).collect(),
1330 }
1331 );
1332 assert_eq!(
1333 plans[2],
1334 CallPlan::Multi {
1335 targets: vec![
1336 (addr(0x01), vec![U256::from(6u64)]),
1337 (addr(0x02), vec![U256::from(99u64)]),
1338 ],
1339 }
1340 );
1341 let planned: usize = plans.iter().map(CallPlan::request_slot_count).sum();
1342 assert_eq!(planned, requests.len());
1343 }
1344
1345 #[test]
1346 fn planning_respects_target_budget() {
1347 let requests: Vec<_> = (0..5u8)
1348 .map(|i| (addr(i + 1), U256::from(i as u64)))
1349 .collect();
1350 let plans = plan_calls(&requests, &cfg(100, 2));
1351 assert_eq!(plans.len(), 3);
1353 assert!(matches!(&plans[0], CallPlan::Multi { targets } if targets.len() == 2));
1354 assert!(matches!(&plans[1], CallPlan::Multi { targets } if targets.len() == 2));
1355 assert!(matches!(&plans[2], CallPlan::Single { .. }));
1356 }
1357
1358 #[test]
1359 fn planning_lone_remainder_degrades_to_single_call() {
1360 let requests: Vec<_> = (0..4u64).map(|i| (addr(0x01), U256::from(i))).collect();
1361 let plans = plan_calls(&requests, &cfg(3, 10));
1362 assert_eq!(plans.len(), 2);
1363 assert!(matches!(&plans[1], CallPlan::Single { slots, .. } if slots.len() == 1));
1364 }
1365
1366 #[test]
1367 fn planning_isolates_dispatcher_address_collision() {
1368 let requests = vec![
1369 (MULTICALL3_ADDRESS, U256::from(1u64)),
1370 (addr(0x02), U256::from(2u64)),
1371 (addr(0x03), U256::from(3u64)),
1372 ];
1373 let plans = plan_calls(&requests, &cfg(100, 10));
1374 assert_eq!(
1375 plans[0],
1376 CallPlan::Single {
1377 target: MULTICALL3_ADDRESS,
1378 slots: vec![U256::from(1u64)],
1379 }
1380 );
1381 assert!(matches!(&plans[1], CallPlan::Multi { targets } if targets.len() == 2));
1382 }
1383
1384 #[test]
1385 fn multi_target_overrides_include_dispatcher_and_extractors() {
1386 let plan = CallPlan::Multi {
1387 targets: vec![
1388 (addr(0x02), vec![U256::from(1u64)]),
1389 (addr(0x03), vec![U256::from(2u64)]),
1390 ],
1391 };
1392 let extractor = Bytes::from_static(STORAGE_EXTRACTOR_CODE);
1393 let overrides = overrides_for_plan(&plan, &extractor);
1394 assert_eq!(overrides.len(), 3);
1395 assert_eq!(
1396 overrides[&MULTICALL3_ADDRESS].code.as_ref(),
1397 Some(multicall3_runtime_code())
1398 );
1399 assert_eq!(overrides[&addr(0x02)].code.as_ref(), Some(&extractor));
1400 assert_eq!(overrides[&addr(0x03)].code.as_ref(), Some(&extractor));
1401 }
1402
1403 #[test]
1404 fn call_many_grouping_respects_request_budget() {
1405 let plans = vec![
1406 CallPlan::Single {
1407 target: addr(0x01),
1408 slots: (0..6u64).map(U256::from).collect(),
1409 },
1410 CallPlan::Single {
1411 target: addr(0x02),
1412 slots: (0..6u64).map(U256::from).collect(),
1413 },
1414 CallPlan::Single {
1415 target: addr(0x03),
1416 slots: (0..2u64).map(U256::from).collect(),
1417 },
1418 ];
1419 let groups = group_plans_for_call_many(plans, 10);
1420 assert_eq!(groups.len(), 2);
1422 assert_eq!(groups[0].len(), 1);
1423 assert_eq!(groups[1].len(), 2);
1424 let total: usize = groups
1425 .iter()
1426 .flatten()
1427 .map(CallPlan::request_slot_count)
1428 .sum();
1429 assert_eq!(total, 14);
1430 }
1431
1432 #[test]
1433 fn request_byte_budget_clamps_slot_limits() {
1434 let config = BulkCallConfig {
1435 max_slots_per_call: 25_000,
1436 max_slots_per_request: 25_000,
1437 max_request_bytes: 640_512,
1438 ..BulkCallConfig::default()
1439 }
1440 .normalized();
1441
1442 assert_eq!(config.max_slots_per_call, 10_000);
1443 assert_eq!(config.max_slots_per_request, 10_000);
1444 }
1445
1446 #[test]
1447 fn default_call_many_plan_fits_measured_request_budget_at_target_limit() {
1448 let config = BulkCallConfig {
1449 dispatch: CallDispatch::CallMany,
1450 ..BulkCallConfig::default()
1451 }
1452 .normalized();
1453 let requests: Vec<_> = (1..=config.max_targets_per_call as u64)
1454 .flat_map(|target| {
1455 let address = Address::from_word(U256::from(target).into());
1456 (0..100_u64).map(move |slot| (address, U256::from(slot)))
1457 })
1458 .collect();
1459 let plans = plan_calls(&requests, &config);
1460 let groups = group_plans_for_call_many(plans, config.max_slots_per_request);
1461 assert_eq!(groups.len(), 1);
1462
1463 let extractor = config.extractor();
1464 let mut overrides = StateOverride::default();
1465 let mut transactions = Vec::new();
1466 for plan in &groups[0] {
1467 overrides.extend(overrides_for_plan(plan, &extractor));
1468 let (to, data) = plan_call_parts(plan);
1469 transactions.push(serde_json::json!({ "to": to, "data": data }));
1470 }
1471 let bundles = serde_json::json!([{ "transactions": transactions }]);
1472 let context =
1473 serde_json::json!({ "blockNumber": "0xffffffffffffffff", "transactionIndex": -1 });
1474 let request = serde_json::json!({
1475 "jsonrpc": "2.0",
1476 "id": u64::MAX,
1477 "method": "eth_callMany",
1478 "params": [bundles, context, overrides],
1479 });
1480 let serialized = serde_json::to_vec(&request).unwrap();
1481
1482 assert!(
1483 serialized.len() <= config.max_request_bytes,
1484 "default worst-case request was {} bytes, over the {} byte planning budget",
1485 serialized.len(),
1486 config.max_request_bytes
1487 );
1488 }
1489
1490 #[test]
1491 fn multi_target_response_decodes_per_target_failures() {
1492 let targets = vec![
1493 (addr(0x02), vec![U256::from(1u64), U256::from(2u64)]),
1494 (addr(0x03), vec![U256::from(3u64)]),
1495 ];
1496 let response = IMulticall3::aggregate3Call::abi_encode_returns(&vec![
1497 IMulticall3::Result {
1498 success: true,
1499 returnData: pack_slots_calldata(&[U256::from(11u64), U256::from(22u64)]),
1500 },
1501 IMulticall3::Result {
1502 success: false,
1503 returnData: Bytes::new(),
1504 },
1505 ]);
1506 let results = decode_multi_target_response(&targets, &response);
1507 assert_eq!(results.len(), 3);
1508 assert!(matches!(results[0], (_, _, Ok(v)) if v == U256::from(11u64)));
1509 assert!(matches!(results[1], (_, _, Ok(v)) if v == U256::from(22u64)));
1510 assert!(results[2].2.is_err());
1511 }
1512}